Absorb turn lifecycle hardening into the live stack

crates/persistence/tests/search_path_postgres.rs

@@ -8,18 +8,19 @@
 //! needed. The assertion derives the reachable set from the dependency
 //! catalogue — the functions that check constraints and indexes record in
 //! `pg_depend`, the implementation functions of any operators they record,
-//! plus one hop of body references from those functions — rather than
-//! matching rendered definition text, so a function reached only through a
-//! user-defined operator is still found, and a migration that adds an
+//! plus body references followed transitively from those functions — rather
+//! than matching rendered definition text, so a function reached only through
+//! a user-defined operator is still found, and a migration that adds an
 //! unpinned reachable function fails here instead of failing the next
 //! restore. Each pin must carry the canonical value — the migration-selected
 //! schema, then `pg_catalog`, then `pg_temp` — because a pin that omits the
-//! working schema fails restore exactly like a missing pin. One
-//! body-reference hop is a deliberate limit: deeper call chains have no
-//! mechanical catalogue representation, and the schema's current chains are
-//! one deep. The test also fails when discovery returns nothing: the schema's
-//! check constraints do reach functions, so an empty set means the discovery
-//! query broke, not that nothing needs pinning.
+//! working schema fails restore exactly like a missing pin. Body references
+//! close transitively to a fixed point: `pg_depend` has no body-level
+//! representation, so the closure follows `prosrc` name references until no
+//! new function appears, and a chain of unqualified calls is followed to its
+//! end rather than one hop deep. The test also fails when discovery returns
+//! nothing: the schema's check constraints do reach functions, so an empty
+//! set means the discovery query broke, not that nothing needs pinning.
 
 #![allow(
     clippy::expect_used,
@@ -46,7 +47,7 @@ const DATABASE_USER: &str = "signalbox";
 const DATABASE_PASSWORD: &str = "signalbox-test-only";
 
 const RESTORE_REACHABLE_FUNCTIONS: &str = "
-    WITH restore_dependency AS (
+    WITH RECURSIVE restore_dependency AS (
         SELECT d.refclassid, d.refobjid
           FROM pg_depend AS d
          WHERE (
@@ -84,11 +85,12 @@ const RESTORE_REACHABLE_FUNCTIONS: &str = "
            )
     ),
     covered AS (
-        SELECT oid, proname, proconfig FROM reachable
+        SELECT oid, proname, proconfig, prosrc, pronamespace FROM reachable
         UNION
-        SELECT callee.oid, callee.proname, callee.proconfig
+        SELECT callee.oid, callee.proname, callee.proconfig, callee.prosrc,
+               callee.pronamespace
           FROM pg_proc AS callee
-          JOIN reachable AS caller
+          JOIN covered AS caller
             ON callee.pronamespace = caller.pronamespace
            AND callee.oid <> caller.oid
            AND caller.prosrc ~ ('\\m' || callee.proname || '\\M')
@@ -106,6 +108,45 @@ const RESTORE_REACHABLE_FUNCTIONS: &str = "
      ORDER BY proname
 ";
 
+const RESTORE_PROBE_HEAD: &str = "restore_probe_head";
+const RESTORE_PROBE_MIDDLE: &str = "restore_probe_middle";
+const RESTORE_PROBE_TAIL: &str = "restore_probe_tail";
+
+/// DDL for a three-deep call chain behind a check constraint, rendered from
+/// the probe-name constants so the assertion can never drift from the fixture.
+/// The fields are named because the statements must run in tail-to-table
+/// dependency order; a positional collection would let a fixture edit
+/// silently reorder them.
+struct SyntheticTransitiveChain {
+    create_tail: String,
+    create_middle: String,
+    create_head: String,
+    create_probe_table: String,
+}
+
+fn synthetic_transitive_chain() -> SyntheticTransitiveChain {
+    SyntheticTransitiveChain {
+        create_tail: format!(
+            "CREATE FUNCTION {RESTORE_PROBE_TAIL}() RETURNS boolean
+                LANGUAGE sql IMMUTABLE AS 'SELECT true'"
+        ),
+        create_middle: format!(
+            "CREATE FUNCTION {RESTORE_PROBE_MIDDLE}() RETURNS boolean
+                LANGUAGE sql IMMUTABLE AS 'SELECT {RESTORE_PROBE_TAIL}()'"
+        ),
+        create_head: format!(
+            "CREATE FUNCTION {RESTORE_PROBE_HEAD}(value text) RETURNS boolean
+                LANGUAGE sql IMMUTABLE AS 'SELECT {RESTORE_PROBE_MIDDLE}()'"
+        ),
+        create_probe_table: format!(
+            "CREATE TABLE restore_probe (
+                value text,
+                CONSTRAINT restore_probe_reaches_functions CHECK ({RESTORE_PROBE_HEAD}(value))
+            )"
+        ),
+    }
+}
+
 /// Names of covered functions that lack the canonical pin.
 fn unpinned_names(covered: &[(String, bool)]) -> Vec<&str> {
     covered
@@ -156,3 +197,54 @@ async fn every_restore_reachable_function_pins_its_search_path() -> Result<(), B
     );
     Ok(())
 }
+
+/// INV-070: body-reference discovery closes transitively — a check constraint
+/// whose function calls through an intermediate body still surfaces the
+/// unpinned function at the end of the chain.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore = "requires ephemeral PostgreSQL"]
+async fn transitive_body_references_close_to_a_fixed_point() -> Result<(), Box<dyn Error>> {
+    let container = Postgres::default()
+        .with_db_name(DATABASE_NAME)
+        .with_user(DATABASE_USER)
+        .with_password(DATABASE_PASSWORD)
+        .with_cmd(disposable_postgres_server_args())
+        .with_mount(disposable_postgres_state_tmpfs_from_example()?)
+        .with_tag(POSTGRES_IMAGE_TAG)
+        .with_labels(disposable_test_container_labels())
+        .start()
+        .await?;
+    let host = container.get_host().await?;
+    let port = container.get_host_port_ipv4(5432).await?;
+    let database_url =
+        format!("postgres://{DATABASE_USER}:{DATABASE_PASSWORD}@{host}:{port}/{DATABASE_NAME}");
+    let pool = PgPoolOptions::new()
+        .max_connections(2)
+        .connect_with(local_test_connection_options(&database_url)?)
+        .await?;
+    migrate(&pool).await?;
+    let chain = synthetic_transitive_chain();
+    sqlx::query(sqlx::AssertSqlSafe(chain.create_tail.as_str()))
+        .execute(&pool)
+        .await?;
+    sqlx::query(sqlx::AssertSqlSafe(chain.create_middle.as_str()))
+        .execute(&pool)
+        .await?;
+    sqlx::query(sqlx::AssertSqlSafe(chain.create_head.as_str()))
+        .execute(&pool)
+        .await?;
+    sqlx::query(sqlx::AssertSqlSafe(chain.create_probe_table.as_str()))
+        .execute(&pool)
+        .await?;
+
+    let covered: Vec<(String, bool)> = sqlx::query_as(RESTORE_REACHABLE_FUNCTIONS)
+        .fetch_all(&pool)
+        .await?;
+    assert_eq!(
+        unpinned_names(&covered),
+        [RESTORE_PROBE_HEAD, RESTORE_PROBE_MIDDLE, RESTORE_PROBE_TAIL],
+        "the probe chain must surface: head directly, middle one body hop deep, \
+         and tail two hops deep, which only a transitive closure reaches"
+    );
+    Ok(())
+}

tooling/test_sweep_test_containers.py

@@ -184,7 +184,7 @@ def rust_string_constant(name: str) -> str:
 
 # Re-verified against the head; a scan that silently matched nothing would
 # otherwise satisfy the marking test with no evidence at all.
-CONTAINER_START_SITES = 38
+CONTAINER_START_SITES = 39
 
 
 def container_start_sites() -> tuple[list[str], list[str]]: