Data Source Connection Pool Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build a secure, lazy, per-data-source connection pool framework for PostgreSQL and MySQL while keeping the platform PostgreSQL connection path separate.

Architecture: Flask-SQLAlchemy continues to own only the platform control database. External DataSource definitions are resolved through a DataSourceConnectionManager, which loads encrypted credentials from PostgreSQL, selects an allowlisted adapter, and acquires a connection from a worker-local lazy pool registry. Credential/configuration version changes invalidate old pools; per-source circuit breakers and diagnostics isolate failures.

Tech Stack: Flask 2.3, Flask-SQLAlchemy 3.1, SQLAlchemy 2.0 QueuePool/NullPool, PostgreSQL, MySQL 8, Neo4j, AES-256-GCM from cryptography, psycopg2, PyMySQL, Alembic, pytest, Docker Compose.

File and responsibility map

File	Responsibility
`app/core/data_source/errors.py`	Stable data-source error classes and safe API codes
`app/core/data_source/redaction.py`	Recursive request and exception redaction
`app/core/data_source/models.py`	Immutable definitions, credentials and pool-status value objects
`app/core/data_source/credentials.py`	AES-GCM codec and PostgreSQL credential repository
`app/core/data_source/definitions.py`	Neo4j `DataSource` definition repository without secrets
`app/core/data_source/adapters/base.py`	Adapter protocol and allowlisted common options
`app/core/data_source/adapters/postgresql.py`	PostgreSQL Engine and connection probe
`app/core/data_source/adapters/mysql.py`	MySQL Engine and connection probe
`app/core/data_source/circuit_breaker.py`	Per-source closed/open/half-open state machine
`app/core/data_source/pool_registry.py`	Worker-local Engine registry, leases, draining and idle eviction
`app/core/data_source/manager.py`	Public `connect(uid, purpose)` context manager
`app/core/data_source/runtime.py`	One manager instance per Gunicorn worker
`app/core/data_source/service.py`	Save/update/delete/test orchestration across PostgreSQL and Neo4j
`app/api/data_source/routes.py`	HTTP boundary; no direct Engine or credential access
`app/commands/report_datasource_credentials.py`	Read-only legacy credential audit
`app/commands/migrate_datasource_credentials.py`	Explicit encryption/backfill and plaintext removal
`app/commands/reconcile_datasource_credentials.py`	Cross-store orphan/stale reference report and repair
`migrations/versions/20260718_50_datasource_credentials.py`	Encrypted credential and audit schema
`tests/core/data_source/`	Unit and contract tests
`tests/integration/test_datasource_pools.py`	PostgreSQL/MySQL pool integration
`tests/integration/test_datasource_pool_failures.py`	Outage, recovery and invalidation acceptance
`frontend/src/views/dataGovernance/dataSource/`	Secret-free PostgreSQL/MySQL management UI

Task 1: Add dependencies and environment contract

Files:

Modify: requirements.txt
Modify: app/config/config.py
Modify: deploy/docker/.env.example
Modify: deployment/dataops.env
Test: tests/core/data_source/test_config.py
[ ] Step 1: Write the failing configuration test

def test_datasource_pool_defaults_are_bounded(monkeypatch):
    monkeypatch.delenv("DATASOURCE_POOL_SIZE", raising=False)
    monkeypatch.delenv("DATASOURCE_MAX_OVERFLOW", raising=False)
    from app.config.config import BaseConfig

    assert BaseConfig.DATASOURCE_POOL_SIZE == 2
    assert BaseConfig.DATASOURCE_MAX_OVERFLOW == 3
    assert BaseConfig.DATASOURCE_POOL_TIMEOUT == 10
    assert BaseConfig.DATASOURCE_POOL_RECYCLE == 1800
    assert BaseConfig.DATASOURCE_POOL_IDLE_TTL == 900
    assert BaseConfig.DATASOURCE_MAX_IDLE_POOLS == 20
    assert BaseConfig.DATASOURCE_QUERY_TIMEOUT == 30


def test_pool_limits_reject_unsafe_values(monkeypatch):
    monkeypatch.setenv("DATASOURCE_POOL_SIZE", "100")
    from app.config.config import datasource_pool_settings

    with pytest.raises(ValueError, match="DATASOURCE_POOL_SIZE"):
        datasource_pool_settings()


def test_datasource_override_is_clamped_to_server_limits():
    settings = datasource_pool_settings(
        {"pool_size": 4, "max_overflow": 5}
    )
    assert settings["pool_size"] == 4
    assert settings["max_overflow"] == 5

    with pytest.raises(ValueError, match="pool_size"):
        datasource_pool_settings({"pool_size": 11})

Step 2: Run the test and confirm the red state

Run:

.venv/bin/python -m pytest tests/core/data_source/test_config.py -q

Expected: FAIL because the pool settings do not exist.

Step 3: Pin the required drivers

Add:

cryptography==43.0.3
PyMySQL==1.1.1

Step 4: Add bounded environment parsing

Add to app/config/config.py:

def _bounded_int(name: str, default: int, minimum: int, maximum: int) -> int:
    value = int(os.environ.get(name, str(default)))
    if value < minimum or value > maximum:
        raise ValueError(f"{name} must be between {minimum} and {maximum}")
    return value


def datasource_pool_settings(overrides: Mapping[str, int] | None = None) -> dict:
    settings = {
        "pool_size": _bounded_int("DATASOURCE_POOL_SIZE", 2, 1, 10),
        "max_overflow": _bounded_int("DATASOURCE_MAX_OVERFLOW", 3, 0, 10),
        "pool_timeout": _bounded_int("DATASOURCE_POOL_TIMEOUT", 10, 1, 60),
        "pool_recycle": _bounded_int("DATASOURCE_POOL_RECYCLE", 1800, 60, 86400),
        "idle_ttl": _bounded_int("DATASOURCE_POOL_IDLE_TTL", 900, 60, 86400),
        "max_idle_pools": _bounded_int("DATASOURCE_MAX_IDLE_POOLS", 20, 1, 100),
        "query_timeout": _bounded_int("DATASOURCE_QUERY_TIMEOUT", 30, 1, 300),
    }
    return apply_allowlisted_pool_overrides(settings, overrides or {})

Expose the same values on BaseConfig, plus:

DATASOURCE_CREDENTIAL_MASTER_KEY = os.environ.get(
    "DATASOURCE_CREDENTIAL_MASTER_KEY", ""
)
DATASOURCE_CREDENTIAL_KEY_VERSION = os.environ.get(
    "DATASOURCE_CREDENTIAL_KEY_VERSION", "v1"
)
DATASOURCE_CERT_DIR = os.environ.get(
    "DATASOURCE_CERT_DIR", "/etc/dataops-platform/datasource-certs"
)

Only pool_size and max_overflow may be overridden per data source in phase one. Reject unknown keys and values outside the same server-side bounds; never accept pool_timeout, driver names or raw connect_args from an API request.

Step 5: Document environment variables without real secrets

Add template values:

DATASOURCE_CREDENTIAL_MASTER_KEY=replace-with-base64-32-byte-key
DATASOURCE_CREDENTIAL_KEY_VERSION=v1
DATASOURCE_POOL_SIZE=2
DATASOURCE_MAX_OVERFLOW=3
DATASOURCE_POOL_TIMEOUT=10
DATASOURCE_POOL_RECYCLE=1800
DATASOURCE_POOL_IDLE_TTL=900
DATASOURCE_MAX_IDLE_POOLS=20
DATASOURCE_QUERY_TIMEOUT=30
DATASOURCE_CERT_DIR=/etc/dataops-platform/datasource-certs

Step 6: Run tests

Run:

.venv/bin/python -m pytest tests/core/data_source/test_config.py -q

Expected: PASS.

Step 7: Commit

git add requirements.txt app/config/config.py deploy/docker/.env.example \
  deployment/dataops.env tests/core/data_source/test_config.py
git commit -m "feat: define external datasource pool settings"

Task 2: Create the encrypted credential schema

Files:

Create: migrations/versions/20260718_50_datasource_credentials.py
Create: tests/core/data_source/test_credential_schema.py
Modify: tests/test_database_migrations.py
[ ] Step 1: Write the failing migration contract test

def test_datasource_credential_migration_is_encrypted_and_versioned():
    source = Path(
        "migrations/versions/20260718_50_datasource_credentials.py"
    ).read_text(encoding="utf-8")

    assert "datasource_credentials" in source
    assert "encrypted_payload" in source
    assert "nonce" in source
    assert "key_version" in source
    assert "credential_version" in source
    assert "UNIQUE (data_source_uid, credential_version)" in source
    assert "password VARCHAR" not in source

Step 2: Run the test and confirm failure

.venv/bin/python -m pytest \
  tests/core/data_source/test_credential_schema.py -q

Expected: FAIL because the migration does not exist.

Step 3: Add the non-destructive migration

Use down_revision = "20260716_40" and create:

CREATE TABLE IF NOT EXISTS public.datasource_credentials (
    id UUID PRIMARY KEY,
    data_source_uid UUID NOT NULL,
    credential_version INTEGER NOT NULL CHECK (credential_version > 0),
    encrypted_payload BYTEA NOT NULL,
    nonce BYTEA NOT NULL CHECK (octet_length(nonce) = 12),
    key_version VARCHAR(40) NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'active'
        CHECK (status IN ('active', 'retired', 'revoked')),
    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    retired_at TIMESTAMPTZ,
    UNIQUE (data_source_uid, credential_version)
);

CREATE UNIQUE INDEX IF NOT EXISTS uq_datasource_credential_active
ON public.datasource_credentials(data_source_uid)
WHERE status = 'active';

Also create datasource_credential_audit_events with event type, UID, version, actor UID, success flag, safe detail and timestamp. Do not create password, connection URL or raw request columns.

Keep downgrade() data-preserving, matching the current migration policy.

Step 4: Extend disposable-database migration verification

Add datasource_credentials and datasource_credential_audit_events to the post-upgrade expected table set, not to the original baseline migration set.

Step 5: Run schema tests

TEST_POSTGRES_ADMIN_URL=postgresql://dataops:dataops-test-password@localhost:15432/postgres \
  .venv/bin/python -m pytest \
  tests/core/data_source/test_credential_schema.py \
  tests/test_database_migrations.py -q

Expected: all tests PASS; temporary database is dropped.

Step 6: Commit

git add migrations/versions/20260718_50_datasource_credentials.py \
  tests/core/data_source/test_credential_schema.py \
  tests/test_database_migrations.py
git commit -m "feat: add encrypted datasource credentials schema"

Task 3: Implement AES-GCM credential handling

Files:

Create: app/core/data_source/__init__.py
Create: app/core/data_source/errors.py
Create: app/core/data_source/models.py
Create: app/core/data_source/credentials.py
Test: tests/core/data_source/test_credentials.py
[ ] Step 1: Write failing codec tests

def test_credential_round_trip_and_tamper_detection():
    codec = CredentialCodec(
        master_key=b"k" * 32,
        key_version="v1",
    )
    sealed = codec.encrypt(
        data_source_uid="01900000-0000-7000-8000-000000000010",
        credential_version=1,
        credential=DataSourceCredential("reader", "secret"),
    )

    assert b"secret" not in sealed.encrypted_payload
    assert codec.decrypt(sealed) == DataSourceCredential("reader", "secret")

    tampered = dataclasses.replace(
        sealed,
        encrypted_payload=sealed.encrypted_payload[:-1] + b"x",
    )
    with pytest.raises(DataSourceCredentialUnavailable):
        codec.decrypt(tampered)


def test_credential_cannot_be_replayed_for_another_datasource():
    sealed = codec.encrypt(DATA_SOURCE_A, 1, credential)
    replayed = dataclasses.replace(sealed, data_source_uid=DATA_SOURCE_B)

    with pytest.raises(DataSourceCredentialUnavailable):
        codec.decrypt(replayed)

Step 2: Verify tests fail

.venv/bin/python -m pytest \
  tests/core/data_source/test_credentials.py -q

Expected: FAIL with missing modules.

Step 3: Define immutable value objects and stable errors

@dataclass(frozen=True)
class DataSourceCredential:
    username: str
    password: str
    options: Mapping[str, str] = field(default_factory=dict)


@dataclass(frozen=True)
class SealedCredential:
    id: str
    data_source_uid: str
    credential_version: int
    encrypted_payload: bytes
    nonce: bytes
    key_version: str
    status: str = "active"

Define DataSourceError subclasses for every stable phase-one code:

class DataSourceCredentialUnavailable(DataSourceError):
    code = "DATASOURCE_CREDENTIAL_UNAVAILABLE"
    http_status = 503

The complete set is DATASOURCE_NOT_FOUND, DATASOURCE_TYPE_UNSUPPORTED, DATASOURCE_CONFIGURATION_INVALID, DATASOURCE_CREDENTIAL_UNAVAILABLE, DATASOURCE_CONNECTION_FAILED, DATASOURCE_POOL_TIMEOUT, DATASOURCE_CIRCUIT_OPEN, DATASOURCE_QUERY_TIMEOUT, and DATASOURCE_READ_ONLY_VIOLATION. Safe public messages must not include DBAPI exception text.

Step 4: Implement AES-256-GCM

Use:

aad = f"{data_source_uid}:{credential_version}".encode("utf-8")
nonce = os.urandom(12)
payload = json.dumps(
    {
        "username": credential.username,
        "password": credential.password,
        "options": dict(credential.options),
    },
    sort_keys=True,
    separators=(",", ":"),
).encode("utf-8")
ciphertext = AESGCM(master_key).encrypt(nonce, payload, aad)

Decode the environment key with base64.urlsafe_b64decode; require exactly 32 bytes. Convert InvalidTag, malformed JSON and key mismatch to DataSourceCredentialUnavailable("credential cannot be decrypted").

Step 5: Add repository create/get/retire/revoke methods

Repository methods accept a SQLAlchemy Session and use parameterized text(). create_version() takes an advisory transaction lock on data_source_uid, increments the version and retires the prior active row in the same transaction.

Step 6: Run tests

.venv/bin/python -m pytest \
  tests/core/data_source/test_credentials.py -q

Expected: PASS.

Step 7: Commit

git add app/core/data_source tests/core/data_source/test_credentials.py
git commit -m "feat: encrypt external datasource credentials"

Task 4: Remove secrets from logs and API serialization

Files:

Create: app/core/data_source/redaction.py
Modify: app/api/data_source/routes.py
Test: tests/core/data_source/test_redaction.py
Test: tests/test_datasource_api_security.py
[ ] Step 1: Write failing redaction tests

def test_recursive_redaction_covers_connection_secrets():
    value = {
        "name": "warehouse",
        "password": "plain",
        "options": {
            "api_key": "key",
            "url": "postgresql://user:secret@db:5432/app",
        },
    }

    assert redact_mapping(value) == {
        "name": "warehouse",
        "password": "[redacted]",
        "options": {
            "api_key": "[redacted]",
            "url": "postgresql://[redacted]@db:5432/app",
        },
    }


def test_datasource_list_never_serializes_credentials(client, viewer_token):
    response = client.post(
        "/api/datasource/list",
        json={},
        headers=bearer(viewer_token),
    )
    text = response.get_data(as_text=True).lower()

    assert "password" not in text
    assert "encrypted_payload" not in text
    assert "credential_configured" in text

Step 2: Verify tests fail

.venv/bin/python -m pytest \
  tests/core/data_source/test_redaction.py \
  tests/test_datasource_api_security.py -q

Expected: FAIL because request logging and serialization expose sensitive keys.

Step 3: Implement recursive redaction

Redact case-insensitive keys:

SENSITIVE_KEYS = {
    "password",
    "passwd",
    "credential",
    "credentials",
    "api_key",
    "token",
    "authorization",
    "encrypted_payload",
    "nonce",
    "conn_str",
    "connection_string",
}

Redact URL userinfo using urllib.parse.urlsplit() and return a URL containing [redacted]@host. Truncate sanitized exception messages to 1000 characters.

Step 4: Remove raw request logging immediately

Replace every pattern like:

logger.debug("保存数据源请求数据: %s", json.dumps(data))

with:

logger.debug("保存数据源请求: %s", redact_mapping(data or {}))

The list response must strip username, password, credential_ref, encrypted_payload and nonce; expose only:

item["credential_configured"] = bool(item.pop("credential_version", None))

Step 5: Run security tests and source scan

.venv/bin/python -m pytest \
  tests/core/data_source/test_redaction.py \
  tests/test_datasource_api_security.py -q
rg -n "json\\.dumps\\(data|log_data.*password|RETURN n.*password" \
  app/api/data_source app/core/data_source

Expected: tests PASS and rg returns no unsafe logging pattern.

Step 6: Commit

git add app/core/data_source/redaction.py app/api/data_source/routes.py \
  tests/core/data_source/test_redaction.py \
  tests/test_datasource_api_security.py
git commit -m "fix: prevent datasource credential disclosure"

Task 5: Implement the Neo4j definition repository

Files:

Create: app/core/data_source/definitions.py
Test: tests/core/data_source/test_definitions.py
[ ] Step 1: Write failing repository tests

def test_definition_repository_assigns_uid_and_excludes_secrets(fake_graph):
    repository = DataSourceDefinitionRepository(fake_graph)
    saved = repository.save(
        DataSourceDefinition(
            uid=None,
            name_en="warehouse",
            database_type="postgresql",
            host="source-postgres",
            port=5432,
            database="analytics",
            schema="public",
            credential_ref="01900000-0000-7000-8000-000000000011",
            credential_version=1,
        )
    )

    assert uuid.UUID(saved.uid).version == 7
    assert "password" not in fake_graph.last_parameters["properties"]
    assert "username" not in fake_graph.last_parameters["properties"]

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/core/data_source/test_definitions.py -q

Expected: FAIL because the repository does not exist.

Step 3: Implement a parameterized repository

Support:

get(uid: str) -> DataSourceDefinition | None
list(filters: DataSourceFilters) -> list[DataSourceDefinition]
save(definition: DataSourceDefinition) -> DataSourceDefinition
delete(uid: str) -> bool
credential_references() -> dict[str, tuple[str, int]]
remove_legacy_secret_fields(uid: str) -> None

Use uid, not Neo4j internal numeric IDs, for new writes and updates. Keep a temporary numeric-ID lookup only inside the migration command.

Step 4: Add uniqueness constraint reconciliation

Extend app/commands/reconcile_governance_uids.py with:

CREATE CONSTRAINT datasource_uid_unique IF NOT EXISTS
FOR (node:DataSource) REQUIRE node.uid IS UNIQUE

Step 5: Run tests

.venv/bin/python -m pytest \
  tests/core/data_source/test_definitions.py \
  tests/test_governance_identifiers.py -q

Expected: PASS.

Step 6: Commit

git add app/core/data_source/definitions.py \
  app/commands/reconcile_governance_uids.py \
  tests/core/data_source/test_definitions.py \
  tests/test_governance_identifiers.py
git commit -m "feat: add secret-free datasource definitions"

Task 6: Implement PostgreSQL and MySQL adapters

Files:

Create: app/core/data_source/adapters/__init__.py
Create: app/core/data_source/adapters/base.py
Create: app/core/data_source/adapters/postgresql.py
Create: app/core/data_source/adapters/mysql.py
Test: tests/core/data_source/test_adapters.py
[ ] Step 1: Write failing adapter tests

@pytest.mark.parametrize(
    ("database_type", "drivername"),
    [
        ("postgresql", "postgresql+psycopg2"),
        ("mysql", "mysql+pymysql"),
    ],
)
def test_adapter_uses_allowlisted_driver(database_type, drivername):
    adapter = adapter_for(database_type)
    url = adapter.build_url(definition, credential)

    assert url.drivername == drivername
    assert url.password == "secret"


def test_unknown_driver_is_rejected():
    with pytest.raises(DataSourceTypeUnsupported):
        adapter_for("sqlite")


def test_adapter_rejects_arbitrary_connection_options():
    with pytest.raises(DataSourceConfigurationInvalid):
        adapter.validate_options({"driver": "custom", "connect_args": {}})


def test_read_purpose_configures_database_read_only_transaction():
    adapter.configure_transaction(connection, purpose="metadata_preview")
    assert connection.executed == ["SET TRANSACTION READ ONLY"]


def test_connection_test_uses_null_pool_and_disposes(monkeypatch):
    engine = FakeEngine()
    monkeypatch.setattr(base, "create_engine", lambda *a, **kw: engine)

    adapter.test_connection(definition, credential)

    assert engine.pool_class is NullPool
    assert engine.disposed is True

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/core/data_source/test_adapters.py -q

Expected: FAIL with missing adapter modules.

Step 3: Define the adapter protocol

class DataSourceAdapter(Protocol):
    database_type: str

    def build_url(
        self,
        definition: DataSourceDefinition,
        credential: DataSourceCredential,
    ) -> URL: ...

    def connect_args(
        self,
        definition: DataSourceDefinition,
        credential: DataSourceCredential,
    ) -> dict[str, object]: ...

    def validate_options(self, options: Mapping[str, object]) -> None: ...

    def configure_transaction(
        self,
        connection: Connection,
        purpose: str,
    ) -> None: ...

    def probe(self, connection: Connection) -> None: ...

Step 4: Implement the allowlisted adapters

PostgreSQL returns:

{
    "connect_timeout": 5,
    "keepalives": 1,
    "keepalives_idle": 30,
    "keepalives_interval": 10,
    "keepalives_count": 3,
    "application_name": "dataops-platform-datasource",
    "options": "-c statement_timeout=30000",
}

MySQL returns:

{
    "connect_timeout": 5,
    "read_timeout": 30,
    "write_timeout": 30,
    "charset": "utf8mb4",
}

Both execute text("SELECT 1") in probe().

Build the PostgreSQL statement_timeout value from the validated query_timeout; keep it platform-controlled. MySQL uses the same bounded value for read_timeout and write_timeout.

Allow only documented TLS settings: PostgreSQL sslmode, sslrootcert, sslcert, and sslkey; MySQL ssl_ca, ssl_cert, and ssl_key. File paths must resolve under the configured certificate directory. Reject unknown option keys, URL fragments, frontend driver names, and raw connect_args.

For every purpose except dataflow_write, configure the database transaction as read-only before yielding the connection. PostgreSQL executes SET TRANSACTION READ ONLY; MySQL executes the equivalent supported by MySQL

Map a database read-only violation to DATASOURCE_READ_ONLY_VIOLATION. Never retry or replay write transactions.

Step 5: Implement temporary connection testing

engine = create_engine(
    adapter.build_url(definition, credential),
    poolclass=NullPool,
    connect_args=adapter.connect_args(definition, credential),
)
try:
    with engine.connect() as connection:
        adapter.probe(connection)
finally:
    engine.dispose()

Step 6: Run adapter tests

.venv/bin/python -m pytest \
  tests/core/data_source/test_adapters.py -q

Expected: PASS.

Step 7: Commit

git add app/core/data_source/adapters \
  tests/core/data_source/test_adapters.py requirements.txt
git commit -m "feat: add PostgreSQL and MySQL datasource adapters"

Task 7: Build the per-source circuit breaker

Files:

Create: app/core/data_source/circuit_breaker.py
Test: tests/core/data_source/test_circuit_breaker.py
[ ] Step 1: Write failing state-machine tests

def test_breaker_opens_after_three_failures_and_recovers():
    clock = FakeClock()
    breaker = CircuitBreaker(
        failure_threshold=3,
        recovery_seconds=30,
        clock=clock,
    )

    breaker.record_failure()
    breaker.record_failure()
    breaker.record_failure()
    assert breaker.state == "open"

    with pytest.raises(DataSourceCircuitOpen):
        with breaker.probe_permission():
            pass

    clock.advance(30)
    with breaker.probe_permission():
        breaker.record_success()

    assert breaker.state == "closed"

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/core/data_source/test_circuit_breaker.py -q

Expected: FAIL because CircuitBreaker is undefined.

Step 3: Implement closed/open/half-open

Use a threading.Lock. Only one caller may own the half-open probe. A normal query failure does not open the breaker; only connection creation, checkout disconnect and probe failures count.

Step 4: Run tests

.venv/bin/python -m pytest \
  tests/core/data_source/test_circuit_breaker.py -q

Expected: PASS, including concurrent half-open callers.

Step 5: Commit

git add app/core/data_source/circuit_breaker.py \
  tests/core/data_source/test_circuit_breaker.py
git commit -m "feat: isolate datasource failures with circuit breakers"

Task 8: Build the worker-local pool registry

Files:

Create: app/core/data_source/pool_registry.py
Test: tests/core/data_source/test_pool_registry.py
[ ] Step 1: Write failing reuse, drain and eviction tests

def test_registry_reuses_same_engine_and_invalidates_new_version():
    registry = PoolRegistry(settings)
    first = PoolKey(DATA_SOURCE_UID, 1, "hash-a")
    second = PoolKey(DATA_SOURCE_UID, 2, "hash-b")

    with registry.lease(first, lambda: engine_factory(first)) as engine_a:
        with registry.lease(first, lambda: engine_factory(first)) as engine_b:
            assert engine_a is engine_b

    with registry.lease(second, lambda: engine_factory(second)) as engine_c:
        assert engine_c is not engine_a

    assert engine_a.disposed is True


def test_idle_pool_is_evicted_only_after_last_lease_returns():
    with registry.lease(key, lambda: engine):
        clock.advance(901)
        registry.evict_idle()
        assert engine.disposed is False

    registry.evict_idle()
    assert engine.disposed is True

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/core/data_source/test_pool_registry.py -q

Expected: FAIL because the registry does not exist.

Step 3: Implement pool entries

@dataclass
class PoolEntry:
    key: PoolKey
    engine: Engine
    breaker: CircuitBreaker
    created_at: float
    last_used_at: float
    leases: int = 0
    draining: bool = False
    checkout_wait_ms: float = 0.0

Protect the entry map with threading.RLock. Do not hold the registry lock while executing SQL. Create one Engine per key using:

create_engine(
    url,
    pool_size=settings.pool_size,
    max_overflow=settings.max_overflow,
    pool_timeout=settings.pool_timeout,
    pool_recycle=settings.pool_recycle,
    pool_pre_ping=True,
    pool_use_lifo=True,
    pool_reset_on_return="rollback",
    connect_args=connect_args,
)

Resolve settings from the server defaults plus the validated per-source pool_size and max_overflow override. A lease for a new key must mark all older keys for the same UID as draining.

Step 4: Add invalidation and bounded idle cache

Implement:

invalidate(data_source_uid: str, reason: str) -> None
expire_draining(now: float | None = None) -> int
evict_idle(now: float | None = None) -> int
close_all() -> None
snapshot() -> list[PoolStatus]

When the idle-entry count exceeds max_idle_pools, dispose least-recently-used idle entries first. invalidate() immediately blocks new leases for the old key, lets existing leases drain for a bounded grace period, then expire_draining() force-disposes the Engine. Returning a lease after forced disposal must not re-register the old Engine.

Step 5: Add SQLAlchemy event metrics

Track connect, checkout, checkin and invalidate. Store counts only; never inspect or log DBAPI connection arguments.

Step 6: Run pool tests

.venv/bin/python -m pytest \
  tests/core/data_source/test_pool_registry.py -q

Expected: PASS, including a 50-thread lease/release test with no duplicate Engine for the same key and no leaked leases.

Step 7: Commit

git add app/core/data_source/pool_registry.py \
  tests/core/data_source/test_pool_registry.py
git commit -m "feat: add lazy worker-local datasource pool registry"

Task 9: Implement the public connection manager

Files:

Create: app/core/data_source/manager.py
Create: app/core/data_source/runtime.py
Test: tests/core/data_source/test_manager.py
Modify: gunicorn_config.py
[ ] Step 1: Write failing manager tests

def test_manager_resolves_definition_credential_and_adapter():
    with manager.connect(DATA_SOURCE_UID, purpose="metadata_preview") as connection:
        connection.execute(text("SELECT 1"))

    assert definitions.requested_uid == DATA_SOURCE_UID
    assert credentials.requested_version == 4
    assert adapters.requested_type == "postgresql"
    assert registry.last_key.credential_version == 4


def test_write_requires_explicit_write_purpose(manager, fake_engine):
    with manager.connect(DATA_SOURCE_UID, purpose="metadata_preview"):
        pass
    read_transaction = fake_engine.transactions[-1]
    assert read_transaction.rolled_back is True
    assert read_transaction.committed is False

    with manager.connect(DATA_SOURCE_UID, purpose="dataflow_write"):
        pass
    write_transaction = fake_engine.transactions[-1]
    assert write_transaction.committed is True

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/core/data_source/test_manager.py -q

Expected: FAIL because DataSourceConnectionManager does not exist.

Step 3: Implement the context manager

@contextmanager
def connect(self, data_source_uid: str, purpose: str):
    definition = self.definitions.get(data_source_uid)
    if definition is None:
        raise DataSourceNotFound(data_source_uid)
    credential = self.credentials.get_active(
        data_source_uid,
        definition.credential_version,
    )
    adapter = adapter_for(definition.database_type)
    key = PoolKey(
        data_source_uid=data_source_uid,
        credential_version=definition.credential_version,
        config_fingerprint=definition.connection_fingerprint(),
    )
    with self.registry.lease(
        key,
        lambda: adapter.create_pooled_engine(definition, credential, self.settings),
    ) as engine:
        with engine.connect() as connection:
            transaction = connection.begin()
            try:
                adapter.configure_transaction(connection, purpose)
                yield ManagedConnection(connection, purpose)
                if purpose == "dataflow_write":
                    transaction.commit()
                else:
                    transaction.rollback()
            except Exception:
                transaction.rollback()
                raise

ManagedConnection.execute() records query duration and translates known driver timeout and read-only exceptions to the stable safe errors. Measure checkout wait around engine.connect(). It must not retry statements or transactions.

Step 4: Create one manager per Worker

runtime.py lazily constructs one manager in a process-local variable guarded by a lock. Add:

def close_data_source_runtime() -> None:
    manager = _manager
    if manager is not None:
        manager.close()

Call it from Gunicorn worker_exit, not from every Flask request.

Step 5: Run manager tests

.venv/bin/python -m pytest \
  tests/core/data_source/test_manager.py -q

Expected: PASS.

Step 6: Commit

git add app/core/data_source/manager.py app/core/data_source/runtime.py \
  tests/core/data_source/test_manager.py gunicorn_config.py
git commit -m "feat: expose managed datasource connections"

Task 10: Orchestrate secure data-source lifecycle

Files:

Create: app/core/data_source/service.py
Modify: app/api/data_source/routes.py
Modify: app/core/system/permissions.py
Test: tests/test_datasource_lifecycle_api.py
[ ] Step 1: Write failing API lifecycle tests

def test_create_stores_secret_only_in_credential_store(
    client, editor_token, fake_graph, credential_repository
):
    response = client.post(
        "/api/datasource/save",
        json=POSTGRES_SOURCE_PAYLOAD,
        headers=bearer(editor_token),
    )

    assert response.status_code == 201
    graph_properties = fake_graph.saved_properties
    assert "password" not in graph_properties
    assert "username" not in graph_properties
    assert graph_properties["credential_version"] == 1
    assert credential_repository.decrypt_latest().password == "source-pass"


def test_update_invalidates_old_pool(client, editor_token, pool_registry):
    response = client.post(
        "/api/datasource/save",
        json={**UPDATED_PAYLOAD, "uid": DATA_SOURCE_UID},
        headers=bearer(editor_token),
    )

    assert response.status_code == 200
    assert pool_registry.invalidations == [
        (DATA_SOURCE_UID, "configuration_changed")
    ]

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/test_datasource_lifecycle_api.py -q

Expected: FAIL because routes still write request properties directly to Neo4j.

Step 3: Implement save/update orchestration

DataSourceService.save() must:

Validate only postgresql and mysql.
Test using Adapter + NullPool.
Write an encrypted immutable credential version in a PostgreSQL transaction.
Enqueue datasource.credential_version_created with safe definition fields.
Save the secret-free Neo4j definition.
Invalidate the old pool after the definition points to the new version.
Return the sanitized definition.

On Neo4j failure, retain the prior definition and record the newly created credential as an orphan candidate for reconciliation.

For an existing source, omitted username/password means “reuse the active credential internally”: decrypt it only inside the service, combine any explicit replacement fields, and write the next immutable credential version. Never send the old credential back to the frontend. Per-source pool overrides and TLS fields pass through the server allowlists before connection testing.

Step 4: Implement delete orchestration

Order:

registry.invalidate(uid, reason="datasource_deleted")
definitions.delete(uid)
credentials.revoke_all(uid, actor_id)
enqueue_outbox(..., event_type="datasource.deleted", payload={"uid": uid})

If PostgreSQL finalization fails, the reconciliation command must detect a deleted graph definition with active credentials.

Step 5: Refactor HTTP routes

Routes call DataSourceService only. Remove imports of create_engine and URL from app/api/data_source/routes.py. Use real HTTP status codes.

Keep the existing governance permissions for normal data-source operations:

viewer: READ_GOVERNANCE for list, detail and approved read use;
editor: viewer permissions plus EDIT_GOVERNANCE for save, update, delete, connection test and dataflow use;
admin: editor permissions plus the new DATASOURCE_POOL_MANAGE = "datasources:pools:manage" for pool diagnostics and forced invalidation.

Do not grant DATASOURCE_POOL_MANAGE to editor or viewer.

Step 6: Run API tests

.venv/bin/python -m pytest \
  tests/test_datasource_lifecycle_api.py \
  tests/test_permission_matrix.py \
  tests/test_datasource_api_security.py -q

Expected: PASS.

Step 7: Regenerate OpenAPI

.venv/bin/python scripts/generate_openapi.py
.venv/bin/python -m pytest tests/test_architecture_artifacts.py -q

Expected: generated contract matches route source.

Step 8: Commit

git add app/core/data_source/service.py app/api/data_source/routes.py \
  app/core/system/permissions.py tests/test_datasource_lifecycle_api.py \
  tests/test_permission_matrix.py tests/test_datasource_api_security.py \
  docs/architecture/OPENAPI.yaml
git commit -m "feat: secure datasource lifecycle APIs"

Task 11: Make the data-source UI secret-free

Files:

Modify: frontend/src/views/dataGovernance/dataSource/index.vue
Modify: frontend/src/views/dataGovernance/dataSource/components/edit.vue
Modify: frontend/src/api/dataOrigin.js
Create: tests/test_datasource_frontend_contract.py
[ ] Step 1: Write the failing frontend contract test

def test_datasource_ui_does_not_render_or_replay_stored_credentials():
    listing = Path(
        "frontend/src/views/dataGovernance/dataSource/index.vue"
    ).read_text(encoding="utf-8")
    editor = Path(
        "frontend/src/views/dataGovernance/dataSource/components/edit.vue"
    ).read_text(encoding="utf-8")

    assert "value: 'username'" not in listing
    assert "value: 'param'" not in listing
    assert "credential_configured" in listing
    assert "Object.assign(this.formValues, this.itemData)" not in editor
    assert "datasourceParse" not in editor
    assert "key: 'param'" not in editor
    assert "postgresql" in editor
    assert "mysql" in editor

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/test_datasource_frontend_contract.py -q

Expected: FAIL because the existing UI lists username and arbitrary connection parameters and copies the full list item into the credential form.

Step 3: Refactor list and identity handling

Use uid, never the Neo4j numeric id, for edit and delete. Remove username and free-form connection-parameter columns. Show a credential_configured status chip instead. Keep host, port, database and schema visible because they are non-secret definition fields.

Step 4: Refactor create/edit/test payloads

Use a fixed database-type selector containing only PostgreSQL and MySQL. Remove quick connection-string parsing and the free-form param field. On edit, copy only safe definition fields into formValues; initialize username and password as empty and show “凭据已配置，留空则沿用当前凭据”. Omit blank credential fields from the request.

For a new source, require username and password. For an existing source, send its UID so the backend may combine safe field edits with the current credential internally and create the next immutable credential version. Connection testing follows the same rule: new source uses entered credentials; existing source may use UID plus any explicitly entered replacement credentials.

Step 5: Remove obsolete API helpers and run verification

Remove datasourceParse() when no other import remains. Keep the backend compatibility endpoint out of navigation until its later removal is separately verified.

.venv/bin/python -m pytest \
  tests/test_datasource_frontend_contract.py \
  tests/test_datasource_api_security.py -q
cd frontend
npm run build

Expected: tests PASS and the production frontend build exits 0.

Step 6: Commit

git add frontend/src/views/dataGovernance/dataSource/index.vue \
  frontend/src/views/dataGovernance/dataSource/components/edit.vue \
  frontend/src/api/dataOrigin.js \
  tests/test_datasource_frontend_contract.py
git commit -m "fix: make datasource management UI secret-free"

Task 12: Add diagnostics and health aggregation

Files:

Modify: app/api/data_source/routes.py
Modify: app/core/system/health.py
Test: tests/test_datasource_pool_diagnostics.py
[ ] Step 1: Write failing diagnostics tests

def test_admin_pool_status_is_safe(client, admin_token, seeded_pool):
    response = client.get(
        f"/api/datasource/{DATA_SOURCE_UID}/pool",
        headers=bearer(admin_token),
    )
    body = response.get_json()["data"]

    assert body["data_source_uid"] == DATA_SOURCE_UID
    assert body["pool_state"] == "healthy"
    assert "host" not in body
    assert "username" not in body
    assert "password" not in response.get_data(as_text=True).lower()


def test_platform_health_contains_only_pool_aggregates(client):
    body = client.get("/api/system/health").get_json()["data"]["components"]
    assert body["datasource_pools"] == {
        "active_pool_count": 1,
        "degraded_pool_count": 0,
        "open_circuit_count": 0,
        "checked_out_total": 0,
        "pool_timeout_total": 0,
    }

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/test_datasource_pool_diagnostics.py -q

Expected: FAIL because diagnostics are not registered.

Step 3: Add admin-only endpoints

GET  /api/datasource/pools
GET  /api/datasource/{data_source_uid}/pool
POST /api/datasource/{data_source_uid}/pool/invalidate

All require datasources:pools:manage; invalidation accepts only a fixed reason enum and records an audit event.

Step 4: Add aggregate health

If the runtime has not created a registry, report zeros without creating any data-source connections. Do not perform a network probe from platform health.

Step 5: Run tests

.venv/bin/python -m pytest \
  tests/test_datasource_pool_diagnostics.py \
  tests/test_outbox.py -q

Expected: PASS.

Step 6: Commit

git add app/api/data_source/routes.py app/core/system/health.py \
  tests/test_datasource_pool_diagnostics.py
git commit -m "feat: expose safe datasource pool diagnostics"

Task 13: Add legacy credential report, migration and reconciliation

Files:

Create: app/commands/report_datasource_credentials.py
Create: app/commands/migrate_datasource_credentials.py
Create: app/commands/reconcile_datasource_credentials.py
Test: tests/core/data_source/test_credential_migration.py
[ ] Step 1: Write failing migration tests

def test_report_never_returns_plaintext(fake_graph):
    report = inspect_legacy_datasources(fake_graph)

    assert report == [
        {
            "data_source_uid": DATA_SOURCE_UID,
            "issues": ["plaintext_credentials"],
        }
    ]
    assert "legacy-password" not in json.dumps(report)


def test_migration_removes_plaintext_only_after_round_trip_verification(
    fake_graph, credential_repository
):
    result = migrate_one(DATA_SOURCE_UID, fake_graph, credential_repository)

    assert result.status == "migrated"
    assert credential_repository.decrypt_latest().password == "legacy-password"
    assert "password" not in fake_graph.node
    assert "username" not in fake_graph.node

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/core/data_source/test_credential_migration.py -q

Expected: FAIL because commands do not exist.

Step 3: Implement read-only reporting

Report issue categories only:

missing_uid
plaintext_credentials
missing_credential_reference
unsupported_database_type
incomplete_connection_definition

Default output is JSON Lines and excludes host, username and password.

Step 4: Implement explicit migration

CLI contract:

python -m app.commands.migrate_datasource_credentials \
  --uid <uuid> \
  --confirm-encrypt-and-remove-plaintext

Without the confirmation flag, exit non-zero without mutation. For each source:

Backfill UUIDv7 when missing.
Encrypt username/password.
Decrypt and compare in memory.
Update credential reference.
Remove plaintext Neo4j fields.
Record safe audit event.

Step 5: Implement reconciliation

Default is report-only. Compare Neo4j references to PostgreSQL credentials and report:

orphaned_credential
missing_credential
stale_credential_version
revoked_credential_referenced
plaintext_remaining

Repairs require --repair --uid <uuid>. Never delete source data or credentials in a broad repair run.

Step 6: Run tests

.venv/bin/python -m pytest \
  tests/core/data_source/test_credential_migration.py -q

Expected: PASS.

Step 7: Commit

git add app/commands/report_datasource_credentials.py \
  app/commands/migrate_datasource_credentials.py \
  app/commands/reconcile_datasource_credentials.py \
  tests/core/data_source/test_credential_migration.py
git commit -m "feat: migrate legacy datasource credentials safely"

Task 14: Add isolated PostgreSQL and MySQL acceptance services

Files:

Modify: deploy/docker/docker-compose.yml
Create: deploy/docker/datasources/postgres/init.sql
Create: deploy/docker/datasources/mysql/init.sql
Modify: deploy/docker/.env.example
Modify: tests/test_local_docker_contract.py
Create: tests/integration/test_datasource_pools.py
[ ] Step 1: Write failing Compose contract tests

def test_compose_has_isolated_external_datasources():
    source = COMPOSE.read_text(encoding="utf-8")

    assert "source-postgres:" in source
    assert "source-mysql:" in source
    assert "dataops-test-source-postgres:" in source
    assert "dataops-test-source-mysql:" in source
    assert "DATASOURCE_CREDENTIAL_MASTER_KEY:" in source

Step 2: Verify failure

.venv/bin/python -m pytest \
  tests/test_local_docker_contract.py -q

Expected: FAIL because the services are absent.

Step 3: Add isolated services

Use pinned major/minor image tags:

source-postgres:
  image: postgres:16-alpine
  environment:
    POSTGRES_DB: source_db
    POSTGRES_USER: source_reader
    POSTGRES_PASSWORD: source-postgres-test-password
  ports: ["25432:5432"]
  volumes:
    - dataops-test-source-postgres:/var/lib/postgresql/data
    - ./datasources/postgres/init.sql:/docker-entrypoint-initdb.d/init.sql:ro

source-mysql:
  image: mysql:8.4
  environment:
    MYSQL_DATABASE: source_db
    MYSQL_USER: source_reader
    MYSQL_PASSWORD: source-mysql-test-password
    MYSQL_ROOT_PASSWORD: source-mysql-root-test-password
  ports: ["23306:3306"]
  volumes:
    - dataops-test-source-mysql:/var/lib/mysql
    - ./datasources/mysql/init.sql:/docker-entrypoint-initdb.d/init.sql:ro

Both join dataops-test-net and define real health checks. Add test-only 32-byte base64 master key to the backend environment.

Step 4: Seed equivalent acceptance tables

Both databases create:

CREATE TABLE acceptance_customers (
    customer_id INTEGER PRIMARY KEY,
    customer_name VARCHAR(100) NOT NULL
);
INSERT INTO acceptance_customers VALUES
    (1, 'Alpha'),
    (2, 'Beta');

Step 5: Add integration tests

Test:

create both definitions through the API;
run SELECT COUNT(*) through DataSourceConnectionManager;
run 100 concurrent read operations;
assert every result is 2;
assert no entry exceeds pool_size + max_overflow;
assert all leases return to zero;
advance the clock past 15 minutes and prove idle pools are disposed;
close and recreate the Worker runtime and prove pools are rebuilt lazily;
delete a source and prove no new connection can be acquired.
[ ] Step 6: Start services and run integration

docker compose -f deploy/docker/docker-compose.yml up -d \
  source-postgres source-mysql backend

TEST_DATABASE_URL=postgresql://dataops:dataops-test-password@localhost:15432/dataops \
  TEST_SOURCE_POSTGRES_URL=postgresql://source_reader:source-postgres-test-password@localhost:25432/source_db \
  TEST_SOURCE_MYSQL_URL=mysql+pymysql://source_reader:source-mysql-test-password@localhost:23306/source_db \
  .venv/bin/python -m pytest \
  tests/integration/test_datasource_pools.py -q

Expected: PASS; Compose reports both sources healthy.

Step 7: Commit

git add deploy/docker/docker-compose.yml deploy/docker/datasources \
  deploy/docker/.env.example tests/test_local_docker_contract.py \
  tests/integration/test_datasource_pools.py
git commit -m "test: add isolated datasource pool stack"

Task 15: Verify outage, recovery and credential invalidation

Files:

Create: tests/integration/test_datasource_pool_failures.py
Modify: deploy/docker/README.md
[ ] Step 1: Add outage acceptance test

The test helper must execute these externally controlled phases:

assert query(POSTGRES_UID).scalar_one() == 2
assert query(MYSQL_UID).scalar_one() == 2

    compose_stop("source-mysql")
    wait_for_error(MYSQL_UID, "DATASOURCE_CONNECTION_FAILED", attempts=3)
    assert query_error(MYSQL_UID).code == "DATASOURCE_CIRCUIT_OPEN"
    assert query(POSTGRES_UID).scalar_one() == 2
    assert platform_health()["database"] is True

    compose_start_and_wait_healthy("source-mysql")
    clock.advance(30)
    assert query(MYSQL_UID).scalar_one() == 2
    assert pool_status(MYSQL_UID)["pool_state"] == "healthy"

Step 2: Make the outage test explicitly opt-in and self-restoring

The test may run Docker Compose commands only when RUN_DATASOURCE_OUTAGE_TEST=1. Inject a fake monotonic clock into the breaker so recovery does not require a 30-second sleep. Wrap the outage in try/finally; the finally block always starts source-mysql and waits for its Compose health condition. A failed assertion must never leave the service stopped.

Step 3: Verify the default skip

.venv/bin/python -m pytest \
  tests/integration/test_datasource_pool_failures.py -q

Expected: SKIP unless RUN_DATASOURCE_OUTAGE_TEST=1.

Step 4: Run outage, recovery and credential-rotation acceptance

RUN_DATASOURCE_OUTAGE_TEST=1 \
  .venv/bin/python -m pytest \
  tests/integration/test_datasource_pool_failures.py -q

Within the same test process: establish both pools, stop MySQL, open only the MySQL circuit after three failures, prove PostgreSQL and the platform control database remain healthy, restart MySQL, advance the injected clock, and prove the half-open probe closes the circuit. Then update a test credential version, assert the old Engine is disposed, and prove the next query uses the new version. Poll the Compose health condition rather than sleeping a fixed duration.

Step 5: Document operations

Add:

pool defaults and environment variables;
report/migrate/reconcile commands;
admin diagnostics;
safe invalidation;
outage recovery;
explicit warning that platform PostgreSQL is not managed by this registry.
[ ] Step 6: Commit

git add tests/integration/test_datasource_pool_failures.py \
  deploy/docker/README.md
git commit -m "test: verify datasource outage and recovery"

Task 16: Final contract, release sync and complete verification

Files:

Modify: docs/architecture/ARCHITECTURE_OVERVIEW.md
Modify: docs/architecture/DATA_MODEL.md
Modify: docs/architecture/OPENAPI.yaml
Modify: deployment/sync_release.sh only if new source directories are not already copied
Test: existing full test suites
[ ] Step 1: Update architecture artifacts

Document:

control-plane PostgreSQL versus external data-source pools;
credential table and Neo4j reference;
Worker-local pool multiplication;
supported adapters;
failure and security boundaries.
[ ] Step 2: Regenerate OpenAPI and sync release source

.venv/bin/python scripts/generate_openapi.py
bash deployment/sync_release.sh
diff -qr app deployment/app
diff -qr migrations deployment/migrations
cmp requirements.txt deployment/requirements.txt

Expected: all diff/cmp commands exit 0.

Step 3: Run full Python verification

TEST_POSTGRES_ADMIN_URL=postgresql://dataops:dataops-test-password@localhost:15432/postgres \
TEST_DATABASE_URL=postgresql://dataops:dataops-test-password@localhost:15432/dataops \
TEST_ADMIN_PASSWORD=AdminPass123 \
  .venv/bin/python -m pytest -q

Expected: zero failures and zero errors.

Step 4: Run frontend build

cd frontend
npm run build

Expected: build exits 0; pre-existing Vue 2 dependency warnings may remain, but no new ESLint errors are allowed.

Step 5: Rebuild and smoke-test the complete stack

docker compose -f deploy/docker/docker-compose.yml up -d --build
docker compose -f deploy/docker/docker-compose.yml ps
curl -fsS http://localhost:18183/ >/dev/null
curl -fsS http://localhost:15500/api/system/health

Expected: platform services, source-postgres and source-mysql are healthy.

Step 6: Scan for plaintext secrets

rg -n \
  "password.*logger|logger.*password|postgresql://[^[:space:]]+:[^[:space:]@]+@" \
  app deployment/app docs/architecture

Expected: no data-source credential disclosure. Test-only Compose credentials must remain limited to deploy/docker.

Step 7: Commit final artifacts

git add -- \
  docs/architecture/ARCHITECTURE_OVERVIEW.md \
  docs/architecture/DATA_MODEL.md \
  docs/architecture/OPENAPI.yaml \
  deployment/app/api/data_source/routes.py \
  deployment/app/commands/migrate_datasource_credentials.py \
  deployment/app/commands/reconcile_datasource_credentials.py \
  deployment/app/commands/reconcile_governance_uids.py \
  deployment/app/commands/report_datasource_credentials.py \
  deployment/app/config/config.py \
  deployment/app/core/data_source \
  deployment/app/core/system/health.py \
  deployment/app/core/system/permissions.py \
  deployment/gunicorn_config.py \
  deployment/migrations/versions/20260718_50_datasource_credentials.py \
  deployment/requirements.txt
git commit -m "docs: finalize datasource pool operations"

Before staging, inspect git diff --name-only -- deployment/ and confirm every changed release-copy file is the generated counterpart of a source file from Tasks 1–15. Do not stage unrelated pre-existing workspace changes.

Step 8: Request code review

Use the requesting-code-review skill. Review specifically for:

secret leakage;
cross-store partial failure;
connection leaks;
pool multiplication across Workers;
unsafe driver or connect-argument injection;
write-transaction replay.

2026-07-18-datasource-connection-pool.md 54 KB Historia Raaka

Data Source Connection Pool Implementation Plan

File and responsibility map

Task 1: Add dependencies and environment contract

Task 2: Create the encrypted credential schema

Task 3: Implement AES-GCM credential handling

Task 4: Remove secrets from logs and API serialization

Task 5: Implement the Neo4j definition repository

Task 6: Implement PostgreSQL and MySQL adapters

Task 7: Build the per-source circuit breaker

Task 8: Build the worker-local pool registry

Task 9: Implement the public connection manager

Task 10: Orchestrate secure data-source lifecycle

Task 11: Make the data-source UI secret-free

Task 12: Add diagnostics and health aggregation

Task 13: Add legacy credential report, migration and reconciliation

Task 14: Add isolated PostgreSQL and MySQL acceptance services

Task 15: Verify outage, recovery and credential invalidation

Task 16: Final contract, release sync and complete verification

2026-07-18-datasource-connection-pool.md 54 KB

Historia Raaka