Statement log: the ferro.sql logger and the Rust emit path, proven on migration DDL #764

Parent

#700

What to build

The statement log exists and Ferro's loggers reach the application. After this ticket a developer who sets ferro.sql to DEBUG sees every DDL statement the reconciliation pass or a migration run sends, as it finishes:

logging.getLogger("ferro.sql").setLevel(logging.DEBUG)
report = await ferro.migrate()
# ferro.sql DEBUG CREATE TABLE "post" (...) params=() 1.2ms
# ferro.sql DEBUG ALTER TABLE "author" ADD COLUMN "bio" varchar params=() 0.8ms

Each record is a logging.LogRecord on ferro.sql with the message <sql> params=<params> <duration>ms (plus failed: <error> when the statement raised) and structured attributes sql, params (tuple), dialect, duration_ms, operation, subject, connection, rows, error. DDL from the pass and from a run carries operation="migrate"; the pass's subject is the table or enum type (the same string PassReport records), a run's is NNNN_name/<step file>.

The log costs nothing when it is off: Rust asks Python once per FFI entry whether ferro.sql is enabled for DEBUG (it holds the GIL at entry) and opens a task-local statement context only when it is; the engine's execute/fetch layer, the one place every statement passes through, measures and emits only inside such a context, taking the GIL once per statement. Rust formats nothing: it hands the record's data to one Python callable that builds the message and the LogRecord.

The ferro logger stops installing a StreamHandler with propagate = False at import; it installs a NullHandler and propagates. The ferro migrate CLI installs its own stream handler on ferro so a data step's ctx.log.info(...) lines print as they do today. The free-text ferro migrate: <file>: <sql> and ferro auto-migrate: <subject>: <sql> debug lines are removed in favour of the records; PassReport is unchanged. The auto-migrate guide's sentence about those debug lines now points readers at the ferro.sql records.

The emitter is the primitive the next tickets use: the operation ticket opens the same context at every operation entry with its own operation/connection and a set of secret values; the rendering ticket compares .sql() against these records. This ticket's records never mask anything (DDL has no parameters) but the record carries params and the emitter accepts a secret-value set so the operation ticket adds no second emit path.

Acceptance criteria

• ferro.migrate() and ferro.create_tables() on SQLite and Postgres leave one ferro.sql record per statement in report.statements, in order, with sql equal to the report's, operation == "migrate", subject equal to the report's, params == (), duration_ms >= 0, error is None.
• A migration run (the migration test harness applying one migration) leaves one record per DDL/SQL-step statement with subject == "NNNN_name/<step file>"; the lock-timeout test asserts on records, not on ferro text lines.
• A statement the database rejects leaves a record with error set and the message ending failed: <error>, and the exception still propagates.
• With ferro.sql not enabled for DEBUG, the Python emit callable is never called during a pass (asserted with a spy on it).
• pytest's caplog at DEBUG captures ferro.sql records and ferro records with no handler setup; after import ferro, logging.getLogger("ferro").handlers is exactly one NullHandler and propagate is True.
• ferro migrate up on a project whose data step calls ctx.log.info("backfilled 3 rows") prints that line, as today.
• No ferro migrate: / ferro auto-migrate: statement line is logged anywhere; grep -rn "ferro migrate: \|ferro auto-migrate: " src finds only the error/refusal texts that are not statement logs.
• The ferro.sql logger, the record attributes and the ferro-logger change are documented in the Connections guide's logging section (one short section; the full guide page is the docs ticket) and the auto-migrate page's sentence is updated.

Files owned

• Create: src/sql_log.rs, src/ferro/_sql_log.py, tests/test_statement_log.py
• Modify: src/lib.rs, src/backend.rs, src/ddl_exec.rs, src/run.rs, src/migrate.rs, src/migrations_ffi.rs, src/ferro/__init__.py, src/ferro/cli/__init__.py, tests/conftest.py, docs/pages/guide/schema/auto-migrate.md, docs/pages/guide/connections.md
• Test: tests/test_ddl_lock_timeout.py (its ferro_log fixture and the ferro migrate: 0003_… line assertions become record assertions)

Interfaces

• Consumes: PassReport.statements (subject, sql, role) and the DDL executor's Executed recording, unchanged (ADR-0049).
• Consumes: ferro.migrations.testing harness and the backend_matrix / postgres_only markers for the tests.
• Produces (Rust, src/sql_log.rs): enabled(py: Python) -> PyResult<bool> (reads logging.getLogger("ferro.sql").isEnabledFor(DEBUG) through a cached logger); StatementContext { operation: &'static str, connection: String, subject: Option<String>, secret_values: Vec<EngineBindValue> }; scope<F>(ctx: Option<StatementContext>, fut: F) -> impl Future (tokio task-local, None when disabled); observe(sql: &str, binds: &[EngineBindValue], dialect: Dialect, started: Instant, outcome: Result<Option<u64>, &sqlx::Error>) called by every EngineHandle / EngineConnection execute and fetch method after the statement finishes, a no-op outside a scope; with_subject(subject: &str, fut) to override the subject for one statement group (a run's step). Used by the operation ticket.
• Produces (Python, src/ferro/_sql_log.py): LOGGER = logging.getLogger("ferro.sql"); REDACTED (a singleton whose repr and str are ***); emit(record: dict[str, object]) -> None (builds the message and calls LOGGER.debug(msg, extra=...); the params entry arrives as a tuple with None in masked positions and mask: tuple[bool, ...], and emit substitutes REDACTED). Used by the operation and rendering tickets.
• Produces (tests, tests/conftest.py): fixture statement_records yielding the list of LogRecords emitted on ferro.sql during the test, with ferro.sql set to DEBUG for its duration. Used by every later ticket's tests.
• Produces: ferro logger with a NullHandler, propagating; ferro.cli.main installs a StreamHandler on ferro at INFO before dispatching a verb.

Test scenarios

• Happy path: ferro.create_tables() after dropping a table → one record per report.statements entry, same order, sql/subject equal, operation == "migrate", params == (), dialect matches the backend.
• Happy path: the migration harness applies a migration with a DDL step and a SQL step → records with subject == "0001_<name>/01_<step>.up.sql" (the step file name), one per statement.
• Happy path: caplog.set_level(logging.DEBUG) alone captures the records; logging.getLogger("ferro") has one NullHandler and propagate is True.
• Happy path: ferro migrate up through the CLI entry point with a data step calling ctx.log.info("backfilled 3 rows") → the line appears on stderr.
• Edge case: ferro.sql at WARNING → a spy on ferro._sql_log.emit is never called during ferro.migrate().
• Edge case: duration is measured around the send only: a record's duration_ms is a non-negative float.
• Error path: a migration SQL step with invalid SQL → a record with error naming the database error and the message ending failed: …; the run's failure propagates as today.
• Integration: tests/test_ddl_lock_timeout.py's executed-statement assertions read statement_records and still pass on Postgres.

Verify

uv run maturin develop
uv run --no-sync pytest tests/test_statement_log.py tests/test_ddl_lock_timeout.py tests/test_migrations_testing_harness.py -q --no-cov -p no:cacheprovider
uv run --no-sync pytest tests -q --no-cov -p no:cacheprovider -k "cli or auto_migrate or create_tables"
cargo test --no-default-features --features testing
uv run --no-sync ruff check src/ferro tests

Set FERRO_POSTGRES_URL and pass --db-backends=sqlite,postgres so the Postgres cases run rather than skip.

Blocked by

None (can start immediately).