diff --git a/docs/adr/0057-sql-is-seen-through-the-statement-log-and-a-terminals-sql.md b/docs/adr/0057-sql-is-seen-through-the-statement-log-and-a-terminals-sql.md index 7509713f..d487b0c3 100644 --- a/docs/adr/0057-sql-is-seen-through-the-statement-log-and-a-terminals-sql.md +++ b/docs/adr/0057-sql-is-seen-through-the-statement-log-and-a-terminals-sql.md @@ -36,3 +36,7 @@ On Postgres the SQL text depends on the live table: ferro casts bound values (`$ A bound value shows as `***`, a sentinel whose `repr` is `***` so a test can tell it from the string, when it is a `SecretStr`/`SecretBytes` or when it is bound to a column whose field is declared one. Either rule alone leaks: a plain `"hunter2"` compared against a secret column, or a `SecretStr` passed to `raw.execute` where no column is known. The mask is decided in Python, from the model's declared secret fields and the value's type, and travels to Rust beside the binds; Rust never re-derives it, the one-decision pattern I-1 uses. `.sql()`'s `params` are redacted the same way: one rule, applied in one place, and `print(rendered)` in a bug report is the common use. `SET LOCAL` takes no bind parameters on Postgres, so a session setting's value appears inline, unredacted; settings are tenancy scope, not secrets by declaration. Depends on #760, which makes a secret field store its raw value and hydrate as a secret type. A pin holds the two doors together: for every terminal on both dialects, `.sql()` equals the record the executed call wrote. + +## Amendment (2026-10-21): pool hooks are not in the statement log + +A Postgres pool sends two statements of its own, outside any operation: the `after_connect` hook's `SET search_path TO …` on each new connection, and, with `settings_delivery="connection"`, the `after_release` guard's `SELECT coalesce(current_setting($1, true), '')` on each returned connection while a session holds a pinned one. They are not in the statement log. A record needs an operation's scope, and these belong to no operation: sqlx runs the release hook on a task it spawns, and opens connections for `min_connections` in its background maintenance, so a scope could reach at most the connects an operation happens to wait on. Logging them would mean asking the logger once per connection opened and once per connection returned, a GIL take on each, which is the per-pool-event cost the one switch exists to avoid. The run lock's statements are not pool hooks: the run sends them on its own session inside its `migrate` scope, so they are records like the DDL around them. diff --git a/docs/pages/guide/connections.md b/docs/pages/guide/connections.md index 0f3fcf2d..8ae2bebe 100644 --- a/docs/pages/guide/connections.md +++ b/docs/pages/guide/connections.md @@ -242,12 +242,28 @@ A statement the database rejects ends its line with `failed: `, and the e | `params` | The bound parameters, as a tuple | | `dialect` | `"sqlite"` or `"postgres"` | | `duration_ms` | Time the database took, in milliseconds | -| `operation` | What sent it: `"migrate"` for the reconciliation pass and `ferro migrate` | +| `operation` | What sent it: one of the values below | | `subject` | The table or enum type a pass statement changes, or the step file (`0003_add_slug/01_add_slug.up.sql`) a migration statement comes from; `None` otherwise | | `connection` | The connection's name | | `rows` | Rows affected, or `None` for a read | | `error` | The database's error text, or `None` | +| `operation` | Sent by | +| --- | --- | +| `"fetch"` | `all()`, `first()`, `get()` and relation loads | +| `"count"` / `"exists"` | `count()` / `exists()` | +| `"save"` | `save()`, insert or update | +| `"bulk_create"` | `bulk_create()`, one record per chunk | +| `"update"` / `"delete"` | A query's `update()` / `delete()`; `delete()` on an instance | +| `"m2m_add"` / `"m2m_remove"` / `"m2m_clear"` | A many-to-many relation's `add()` / `remove()` / `clear()` | +| `"raw"` | `ferro.raw.execute()`, `fetch_all()` and `fetch_one()` | +| `"transaction"` | A `transaction()` block's `BEGIN`, `SAVEPOINT`, `RELEASE`, `ROLLBACK TO`, `COMMIT` and `ROLLBACK` | +| `"settings"` | A session's settings batch (`SET LOCAL`, `set_config`) and its reset | +| `"catalog"` | The Postgres catalog read the first time an operation touches a table | +| `"migrate"` | The reconciliation pass and `ferro migrate`, the run lock's statements included | + +Statements a Postgres pool sends on its own, outside any operation, are not in the log: the `SET search_path` it runs on each new connection, and, with `settings_delivery="connection"`, the check it runs on each returned connection while a session has one pinned. + The logger is the only switch: while `ferro.sql` is not enabled for `DEBUG`, Ferro collects nothing. Ferro installs only a `NullHandler` on its `ferro` logger, which propagates, so your logging configuration (and pytest's `caplog`) decides where the records go. The `ferro` command line installs its own handler, so a data step's `ctx.log.info(...)` lines still print there. ## See Also diff --git a/src/backend.rs b/src/backend.rs index 13cec57c..f70a8092 100644 --- a/src/backend.rs +++ b/src/backend.rs @@ -877,16 +877,7 @@ impl EngineConnection { .await .map(|rows| rows.iter().map(materialize_engine_row).collect()) } - EngineConnection::Postgres(conn) => { - let mut query = sqlx::query(sql).persistent(false); - for value in values { - query = bind_engine_value(query, value); - } - query - .fetch_all(&mut **conn) - .await - .map(|rows| rows.iter().map(materialize_pg_row).collect()) - } + EngineConnection::Postgres(conn) => fetch_all_pg_unprepared(conn, sql, values).await, }; observe_fetch(sql, values, dialect, started, &result); result @@ -1022,6 +1013,42 @@ impl EngineConnection { } } +/// Fetch on a Postgres connection the pool no longer owns, without entering +/// its prepared-statement cache, and write the statement to the statement +/// log: the run lock's own session (`crate::run::RunLock`) sends its +/// `pg_try_advisory_lock`, its held-by-this-session check and its +/// `pg_advisory_unlock` through here, so they are records like any other. +/// +/// # Errors +/// The `sqlx::Error` the database returned; it is in the record too. +pub async fn fetch_all_detached_pg_unprepared( + conn: &mut sqlx::PgConnection, + sql: &str, + values: &[EngineBindValue], +) -> Result, sqlx::Error> { + let started = Instant::now(); + let result = fetch_all_pg_unprepared(conn, sql, values).await; + observe_fetch(sql, values, Dialect::Postgres, started, &result); + result +} + +/// The Postgres fetch both an [`EngineConnection`] and a detached connection +/// send; the caller writes the record. +async fn fetch_all_pg_unprepared( + conn: &mut sqlx::PgConnection, + sql: &str, + values: &[EngineBindValue], +) -> Result, sqlx::Error> { + let mut query = sqlx::query(sql).persistent(false); + for value in values { + query = bind_engine_value(query, value); + } + query + .fetch_all(conn) + .await + .map(|rows| rows.iter().map(materialize_pg_row).collect()) +} + /// The statement log's record of an unbound execute that returned `result`'s /// rows affected. fn observe_rows(sql: &str, dialect: Dialect, started: Instant, result: &Result) { diff --git a/src/run.rs b/src/run.rs index 33435216..7c9b4316 100644 --- a/src/run.rs +++ b/src/run.rs @@ -46,7 +46,7 @@ use ferro_migrate::snapshot::encode_checksum; use ferro_schema_ir::SchemaModel; use once_cell::sync::Lazy; use pyo3::prelude::*; -use sqlx::{Connection, Row}; +use sqlx::Connection; use std::collections::{BTreeMap, HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -650,6 +650,28 @@ const HELD_BY_ANY_SESSION: &str = "SELECT EXISTS (SELECT 1 FROM pg_locks \ AND database = (SELECT oid FROM pg_database WHERE datname = current_database()) \ AND classid::bigint = $1 AND objid::bigint = $2)"; +/// Send one of the run lock's statements on its own session and read the +/// boolean it answers. The statement goes through the engine's fetch layer, +/// so it is a record in the statement log under the run's scope. +/// +/// # Errors +/// The database's error; `RowNotFound` or a decode error when the answer is +/// not one boolean. +async fn lock_answer( + conn: &mut sqlx::PgConnection, + sql: &str, + binds: &[EngineBindValue], +) -> Result { + let rows = crate::backend::fetch_all_detached_pg_unprepared(conn, sql, binds).await?; + match rows.first().and_then(|row| column(row, 0)) { + Some(EngineValue::Bool(answer)) => Ok(*answer), + Some(other) => Err(sqlx::Error::Decode( + format!("the run lock's `{sql}` answered {other:?}, not a boolean").into(), + )), + None => Err(sqlx::Error::RowNotFound), + } +} + impl RunLock { /// Take the run lock for the governed `schema` (the connection's current /// schema when `None`), waiting up to `timeout`. `on_wait` is called once, @@ -711,13 +733,13 @@ impl RunLock { .map_err(|e| db_error("opening the run lock's connection", e))? .detach(); loop { - let taken: bool = sqlx::query("SELECT pg_try_advisory_lock($1)") - .bind(key) - .persistent(false) - .fetch_one(&mut conn) - .await - .and_then(|row| row.try_get(0)) - .map_err(|e| db_error("taking the run lock", e))?; + let taken = lock_answer( + &mut conn, + "SELECT pg_try_advisory_lock($1)", + &[EngineBindValue::I64(key)], + ) + .await + .map_err(|e| db_error("taking the run lock", e))?; if taken { break; } @@ -820,13 +842,12 @@ impl RunLock { None => false, Some(conn) => { let (classid, objid) = lock_ids(*key); - let probe = sqlx::query(HELD_BY_THIS_SESSION) - .bind(classid) - .bind(objid) - .persistent(false) - .fetch_one(conn) - .await - .and_then(|row| row.try_get::(0)); + let probe = lock_answer( + conn, + HELD_BY_THIS_SESSION, + &[EngineBindValue::I64(classid), EngineBindValue::I64(objid)], + ) + .await; match probe { Ok(held) => held, Err(err) if !first && probe_lost_the_session(&err) => false, @@ -848,11 +869,12 @@ impl RunLock { match &mut self.state { LockState::Postgres { conn, key, .. } => { if let Some(mut conn) = conn.take() { - let _ = sqlx::query("SELECT pg_advisory_unlock($1)") - .bind(*key) - .persistent(false) - .execute(&mut conn) - .await; + let _ = lock_answer( + &mut conn, + "SELECT pg_advisory_unlock($1)", + &[EngineBindValue::I64(*key)], + ) + .await; let _ = conn.close().await; } } diff --git a/tests/test_statement_log_run.py b/tests/test_statement_log_run.py new file mode 100644 index 00000000..dc77adbe --- /dev/null +++ b/tests/test_statement_log_run.py @@ -0,0 +1,120 @@ +# ruff: noqa: F811 - pytest fixtures imported from a sibling module are redefined as arguments +"""The run lock's statements are in the statement log (#764, ADR-0057). + +On Postgres a run holds its lock as a session advisory lock on a connection +of its own. Taking it, checking it and letting it go are statements like any +other, so a run's log shows them among its DDL: + +```text +ferro.sql DEBUG SELECT pg_try_advisory_lock($1) params=(4242...,) 0.4ms +ferro.sql DEBUG SELECT EXISTS (SELECT 1 FROM pg_locks ...) params=(...) 0.6ms +ferro.sql DEBUG CREATE TABLE "slog_author" (...) params=() 1.2ms +ferro.sql DEBUG SELECT pg_advisory_unlock($1) params=(4242...,) 0.3ms +``` + +SQLite's run lock is a file lock and sends no statement. +""" + +import logging +from typing import Annotated + +import pytest + +import ferro +from ferro import Model +from ferro.base import FerroField +from ferro.migrations.testing import harness +from tests.test_migrate_new import ( # noqa: F401 - fixtures + pkg, + project, + write_models, +) +from tests.test_migrate_up import ( # noqa: F401 - fixtures + configure, + db, + new, +) + +pytestmark = [pytest.mark.postgres_only, pytest.mark.asyncio] + +TRY_LOCK = "SELECT pg_try_advisory_lock($1)" +UNLOCK = "SELECT pg_advisory_unlock($1)" +PROBE_PREFIX = "SELECT EXISTS (SELECT 1 FROM pg_locks" + + +def _is_probe(sql: str) -> bool: + """The check that this run's own session holds the lock (``status``'s + probe of any session's lock reads the same view without the pid).""" + return sql.startswith(PROBE_PREFIX) and "pg_backend_pid()" in sql + + +AUTHOR = """ +class Author(Model): + __ferro_table__ = "slog_author" + id: Annotated[int | None, FerroField(primary_key=True)] = None + name: str +""" + + +def _lock_records(records: list[logging.LogRecord]) -> list[logging.LogRecord]: + return [r for r in records if r.sql in (TRY_LOCK, UNLOCK) or _is_probe(r.sql)] + + +def _assert_lock_lifecycle(records: list[logging.LogRecord]) -> None: + """The lock is taken, checked on its own session, then released, each one + record under ``operation="migrate"``, around every other statement of the + run.""" + lock = _lock_records(records) + assert [r.sql for r in lock][:1] == [TRY_LOCK] + assert _is_probe(lock[1].sql) + assert lock[-1].sql == UNLOCK + key = lock[0].params + assert len(key) == 1 and isinstance(key[0], int) + assert lock[-1].params == key + assert len(lock[1].params) == 2 and all(isinstance(p, int) for p in lock[1].params) + for record in lock: + assert record.operation == "migrate" + assert record.subject is None + assert record.connection == "default" + assert record.dialect == "postgres" + assert record.error is None + assert record.duration_ms >= 0 + first, last = records.index(lock[0]), records.index(lock[-1]) + ddl = [r for r in records if r.sql.startswith("CREATE TABLE")] + assert ddl and all(first < records.index(r) < last for r in ddl) + + +@pytest.mark.usefixtures("isolated_imports", "clean_registry") +async def test_a_run_logs_taking_checking_and_releasing_its_lock( + project, pkg, db, statement_records +): + configure(project, pkg, db.backend) + write_models(project, pkg, AUTHOR) + new("create_author") + await ferro.connect(db.url) + try: + await harness().apply("0001") + finally: + ferro.reset_engine() + + _assert_lock_lifecycle(statement_records) + + +@pytest.fixture +def _fresh(clean_registry): + yield + ferro.reset_engine() + + +@pytest.mark.usefixtures("_fresh") +async def test_the_reconciliation_pass_logs_its_lock(db_url, statement_records): + class Author(Model): + __ferro_table__ = "slog_pass_author" + id: Annotated[int | None, FerroField(primary_key=True)] = None + name: str + + await ferro.connect(db_url) + + await ferro.migrate() + + _assert_lock_lifecycle(statement_records)