fix(log): panel round 1 for #764 — run-lock statements in the log, pool hooks decided, operation values documented

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.

docs/pages/guide/connections.md

@@ -242,12 +242,28 @@ A statement the database rejects ends its line with `failed: <error>`, 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

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<Vec<EngineRow>, 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<Vec<EngineRow>, 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<u64, sqlx::Error>) {

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<bool, sqlx::Error> {
+    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::<bool, _>(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;
                 }
             }

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)