# 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** ```python 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: ```bash .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: ```text cryptography==43.0.3 PyMySQL==1.1.1 ``` - [ ] **Step 4: Add bounded environment parsing** Add to `app/config/config.py`: ```python 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: ```python 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: ```text 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: ```bash .venv/bin/python -m pytest tests/core/data_source/test_config.py -q ``` Expected: PASS. - [ ] **Step 7: Commit** ```bash 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** ```python 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** ```bash .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: ```sql 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** ```bash 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** ```bash 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** ```python 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** ```bash .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** ```python @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: ```python 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: ```python 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** ```bash .venv/bin/python -m pytest \ tests/core/data_source/test_credentials.py -q ``` Expected: PASS. - [ ] **Step 7: Commit** ```bash 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** ```python 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** ```bash .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: ```python 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: ```python logger.debug("保存数据源请求数据: %s", json.dumps(data)) ``` with: ```python logger.debug("保存数据源请求: %s", redact_mapping(data or {})) ``` The list response must strip `username`, `password`, `credential_ref`, `encrypted_payload` and `nonce`; expose only: ```python item["credential_configured"] = bool(item.pop("credential_version", None)) ``` - [ ] **Step 5: Run security tests and source scan** ```bash .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** ```bash 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** ```python 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** ```bash .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: ```python 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: ```cypher CREATE CONSTRAINT datasource_uid_unique IF NOT EXISTS FOR (node:DataSource) REQUIRE node.uid IS UNIQUE ``` - [ ] **Step 5: Run tests** ```bash .venv/bin/python -m pytest \ tests/core/data_source/test_definitions.py \ tests/test_governance_identifiers.py -q ``` Expected: PASS. - [ ] **Step 6: Commit** ```bash 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** ```python @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** ```bash .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** ```python 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: ```python { "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: ```python { "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 8. Map a database read-only violation to `DATASOURCE_READ_ONLY_VIOLATION`. Never retry or replay write transactions. - [ ] **Step 5: Implement temporary connection testing** ```python 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** ```bash .venv/bin/python -m pytest \ tests/core/data_source/test_adapters.py -q ``` Expected: PASS. - [ ] **Step 7: Commit** ```bash 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** ```python 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** ```bash .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** ```bash .venv/bin/python -m pytest \ tests/core/data_source/test_circuit_breaker.py -q ``` Expected: PASS, including concurrent half-open callers. - [ ] **Step 5: Commit** ```bash 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** ```python 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** ```bash .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** ```python @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: ```python 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: ```python 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** ```bash .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** ```bash 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** ```python 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** ```bash .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** ```python @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: ```python 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** ```bash .venv/bin/python -m pytest \ tests/core/data_source/test_manager.py -q ``` Expected: PASS. - [ ] **Step 6: Commit** ```bash 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** ```python 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** ```bash .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: 1. Validate only `postgresql` and `mysql`. 2. Test using Adapter + `NullPool`. 3. Write an encrypted immutable credential version in a PostgreSQL transaction. 4. Enqueue `datasource.credential_version_created` with safe definition fields. 5. Save the secret-free Neo4j definition. 6. Invalidate the old pool after the definition points to the new version. 7. 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: ```python 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** ```bash .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** ```bash .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** ```bash 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** ```python 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** ```bash .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. ```bash .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** ```bash 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** ```python 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** ```bash .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** ```text 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** ```bash .venv/bin/python -m pytest \ tests/test_datasource_pool_diagnostics.py \ tests/test_outbox.py -q ``` Expected: PASS. - [ ] **Step 6: Commit** ```bash 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** ```python 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** ```bash .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: ```text 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: ```bash python -m app.commands.migrate_datasource_credentials \ --uid \ --confirm-encrypt-and-remove-plaintext ``` Without the confirmation flag, exit non-zero without mutation. For each source: 1. Backfill UUIDv7 when missing. 2. Encrypt username/password. 3. Decrypt and compare in memory. 4. Update credential reference. 5. Remove plaintext Neo4j fields. 6. Record safe audit event. - [ ] **Step 5: Implement reconciliation** Default is report-only. Compare Neo4j references to PostgreSQL credentials and report: ```text orphaned_credential missing_credential stale_credential_version revoked_credential_referenced plaintext_remaining ``` Repairs require `--repair --uid `. Never delete source data or credentials in a broad repair run. - [ ] **Step 6: Run tests** ```bash .venv/bin/python -m pytest \ tests/core/data_source/test_credential_migration.py -q ``` Expected: PASS. - [ ] **Step 7: Commit** ```bash 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** ```python 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** ```bash .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: ```yaml 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: ```sql 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** ```bash 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** ```bash 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: ```python 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** ```bash .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** ```bash 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** ```bash 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** ```bash .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** ```bash 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** ```bash 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** ```bash 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** ```bash 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** ```bash 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.