From 1bce2bbffe8ca042e62fd4f6b46b80cd37e3358a Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 14:10:13 -0400 Subject: [PATCH 01/12] buzz-db: drop the events_created_at_floor trigger (phase 2) Phase 2 of BUZZ-316. The insert-time application floor check (phase 1) is now the only created_at floor guard, so drop the deferred constraint trigger, its function, the writer-pool buzz.created_at_floor GUC, and the startup catalog/behavior probes that required them. Migration 0062 drops the trigger on events and every partition plus the function, all IF EXISTS. Desired-state schema, pgschema reconciliation and migration asserts drop the same objects. spawn_fence_probe becomes synchronous; the ordered heartbeat probe still fails closed. DESTRUCTIVE: deploy only after phase 1 runs on every writer, with no rollback expected, out-of-band writers stopped, and the replica breaker held closed through the drop until replay catches up. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- ARCHITECTURE.md | 33 +- crates/buzz-db/src/lib.rs | 6 +- .../src/runtime/connection_observability.rs | 39 +- crates/buzz-db/src/runtime/migration.rs | 21 +- crates/buzz-db/src/runtime/mod.rs | 84 +--- crates/buzz-db/src/runtime/observability.rs | 11 +- crates/buzz-db/src/runtime/replica_fence.rs | 236 +---------- crates/buzz-db/src/runtime/tests.rs | 375 +----------------- .../src/runtime/tests/floor_retirement.rs | 161 ++++++++ crates/buzz-db/src/store/event.rs | 2 +- .../src/store/event_floor_postgres_tests.rs | 13 +- crates/buzz-db/tests/observability_source.rs | 23 +- crates/buzz-relay/src/api/admin/mod.rs | 71 +--- .../bridge/thread_window/postgres_tests.rs | 2 +- crates/buzz-relay/src/main.rs | 25 +- crates/buzz-relay/src/startup_steps.rs | 2 +- deploy/charts/buzz/README.md | 6 +- migrations/0087_retire_created_at_floor.sql | 26 ++ .../events_created_at_floor_guard.sql | 29 -- schema/schema.sql | 1 - schema/triggers/events.sql | 11 - scripts/reconcile-schema-after-pgschema.sql | 8 - 22 files changed, 315 insertions(+), 870 deletions(-) create mode 100644 crates/buzz-db/src/runtime/tests/floor_retirement.rs create mode 100644 migrations/0087_retire_created_at_floor.sql delete mode 100644 schema/functions/events_created_at_floor_guard.sql diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index dfb7302abf9..98bf240a229 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -506,25 +506,30 @@ shape is an explicit transition, not a flag day: being absent; inventories still record live fences for older executors. Desired-state bootstraps install only explicitly declared legacy fences. New tenant-owned tables rely on application admission, without a fence trigger. -- **Replica floor (transition foundation):** runtime now provides a shared - replica-floor lock helper for compliant channel-event writers, and the writer - probe handshake takes the exclusive counterpart before sampling `S`, scanning - activity, and committing the heartbeat token last. `probe_once` remains the - sole token/fence-wall publication path; there is no separately persisted active - floor cutoff. Commit-time trigger+GUC enforcement remains authoritative for all - paths. -- **Dual enforcement (current):** application-owned lock+precheck paths run in - front of the existing trigger/function enforcement; commit-time trigger checks - remain authoritative during this phase. Removing those backstops is a separate - gated migration, after every pod tolerates their absence and no rollback below - that version is expected. +- **Replica floor:** supported channel-event insertions check `created_at` + against `clock_timestamp()` on their admitted transaction using the shared + 960-second floor. The ordered heartbeat handshake and activity scan retain + the same fence-wall calculation. Migration 0087 retires only the deferred + floor trigger/function; writer connections no longer arm its GUC and probe + startup no longer requires those objects. +- **Floor retirement deploy gate:** phase 1 must be deployed to every writer + pod/job, with no rollback expected, before applying the destructive contract. + Drain old connections/in-flight transactions, stop out-of-band writers, and + hold the replica breaker closed through the drop until replay catches up. + Already-running old pods do not reverify the trigger on each heartbeat. +- **Dual enforcement (community fence transition):** application-owned + lock+precheck paths run in front of the existing community trigger/function + enforcement. Removing those backstops is a separate gated migration, after + every pod tolerates their absence and no rollback below that version is + expected. - **Role separation:** relay/runtime code owns admission and lock protocols; operator maintenance/backfill workflows must either use those protocols or run under explicit reviewed procedures that keep routing fences closed. - **Unsupported shape:** direct owner SQL that bypasses supported transaction and lock protocols is not a supported write path. -- **Fleet gates and reconciliation:** startup/fence probes verify guard catalog + - behavior, pgschema bootstraps run reconciliation, and lock/pool metrics remain +- **Fleet gates and reconciliation:** retained guards keep their catalog and + behavior checks; heartbeat/epoch and activity checks still fail closed. + pgschema bootstraps run reconciliation, and lock/pool metrics remain the operational evidence path for rollout safety. The database end state (a declarative schema source, applied out of band at diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 92eeb42f882..8a7f296474b 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -70,15 +70,15 @@ pub const DB_POOL_ACQUIRE_RAW_SERIES_PER_POD: usize = runtime::observability::POOL_ACQUIRE_RAW_SERIES_PER_POD; /// Valid database connection role/step pairs with a start counter. -pub const DB_CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 4] = +pub const DB_CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 3] = runtime::CONNECTION_STARTED_STEPS; /// Valid database connection role/step pairs with a duration histogram. -pub const DB_CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 4] = +pub const DB_CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 3] = runtime::CONNECTION_DURATION_STEPS; /// Valid database connection role/step/outcome terminal combinations. -pub const DB_CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 15] = +pub const DB_CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 12] = runtime::CONNECTION_TERMINALS; /// Raw Prometheus series ceiling per pod for connection-step telemetry. diff --git a/crates/buzz-db/src/runtime/connection_observability.rs b/crates/buzz-db/src/runtime/connection_observability.rs index a1ab04022fd..710d3e04a50 100644 --- a/crates/buzz-db/src/runtime/connection_observability.rs +++ b/crates/buzz-db/src/runtime/connection_observability.rs @@ -17,8 +17,6 @@ pub enum DbConnectionStep { WriterPool, /// A physical connection has completed DNS/network/TLS/authentication. PhysicalConnect, - /// Install the created-at replica-fence floor. - CreatedAtFloor, /// Install lock, idle-transaction, and statement timeouts. SessionTimeouts, /// Verify READ COMMITTED transaction isolation. @@ -29,10 +27,9 @@ pub enum DbConnectionStep { impl DbConnectionStep { /// Complete metric-label vocabulary. - pub const ALL: [Self; 6] = [ + pub const ALL: [Self; 5] = [ Self::WriterPool, Self::PhysicalConnect, - Self::CreatedAtFloor, Self::SessionTimeouts, Self::Isolation, Self::Ready, @@ -43,7 +40,6 @@ impl DbConnectionStep { match self { Self::WriterPool => "writer_pool", Self::PhysicalConnect => "physical_connect", - Self::CreatedAtFloor => "created_at_floor", Self::SessionTimeouts => "session_timeouts", Self::Isolation => "isolation", Self::Ready => "ready", @@ -77,18 +73,17 @@ impl DbConnectionOutcome { } /// Valid role/step pairs with an explicit start counter. -pub const CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 4] = [ +pub const CONNECTION_STARTED_STEPS: [(DbPoolRole, DbConnectionStep); 3] = [ (DbPoolRole::Writer, DbConnectionStep::WriterPool), - (DbPoolRole::Writer, DbConnectionStep::CreatedAtFloor), (DbPoolRole::Writer, DbConnectionStep::SessionTimeouts), (DbPoolRole::Writer, DbConnectionStep::Isolation), ]; /// Valid role/step pairs with a duration histogram. -pub const CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 4] = CONNECTION_STARTED_STEPS; +pub const CONNECTION_DURATION_STEPS: [(DbPoolRole, DbConnectionStep); 3] = CONNECTION_STARTED_STEPS; /// Valid role/step/outcome terminal combinations. -pub const CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 15] = [ +pub const CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutcome); 12] = [ ( DbPoolRole::Writer, DbConnectionStep::WriterPool, @@ -114,21 +109,6 @@ pub const CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutc DbConnectionStep::PhysicalConnect, DbConnectionOutcome::Succeeded, ), - ( - DbPoolRole::Writer, - DbConnectionStep::CreatedAtFloor, - DbConnectionOutcome::Succeeded, - ), - ( - DbPoolRole::Writer, - DbConnectionStep::CreatedAtFloor, - DbConnectionOutcome::Failed, - ), - ( - DbPoolRole::Writer, - DbConnectionStep::CreatedAtFloor, - DbConnectionOutcome::Cancelled, - ), ( DbPoolRole::Writer, DbConnectionStep::SessionTimeouts, @@ -166,8 +146,8 @@ pub const CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionOutc ), ]; -/// Four start counters + four 13-series histograms + fifteen terminal counters. -pub const CONNECTION_RAW_SERIES_PER_POD: usize = 4 + (4 * 13) + 15; +/// Three start counters + three 13-series histograms + twelve terminal counters. +pub const CONNECTION_RAW_SERIES_PER_POD: usize = 3 + (3 * 13) + 12; pub(crate) struct DbConnectionStepAttempt { pool_role: DbPoolRole, @@ -282,13 +262,12 @@ mod tests { [ "writer_pool", "physical_connect", - "created_at_floor", "session_timeouts", "isolation", "ready", ] ); - assert_eq!(CONNECTION_RAW_SERIES_PER_POD, 71); + assert_eq!(CONNECTION_RAW_SERIES_PER_POD, 54); } #[test] @@ -297,7 +276,7 @@ mod tests { let snapshotter = recorder.snapshotter(); let _guard = metrics::set_default_local_recorder(&recorder); let attempt = - DbConnectionStepAttempt::start(DbPoolRole::Writer, DbConnectionStep::CreatedAtFloor); + DbConnectionStepAttempt::start(DbPoolRole::Writer, DbConnectionStep::SessionTimeouts); drop(attempt); let metrics = snapshotter.snapshot().into_vec(); @@ -309,7 +288,7 @@ mod tests { .map(|label| (label.key(), label.value())) .collect::>(); assert_eq!(labels.get("pool_role"), Some(&"writer")); - assert_eq!(labels.get("step"), Some(&"created_at_floor")); + assert_eq!(labels.get("step"), Some(&"session_timeouts")); assert!(!labels.contains_key("reason")); assert!(!labels.contains_key("connection_ordinal")); match value { diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index f7f0888e96f..81ad9af8c89 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -58,14 +58,6 @@ async fn run_migrations_locked(conn: &mut PgConnection) -> Result<()> { build_indexes_before_migration(conn, *version, doc, indexes).await?; } MIGRATOR.run(&mut *conn).await?; - // The replica-fence proof (see `replica_fence`) requires the commit-time - // `created_at` floor trigger from migration 0021 — correctly shaped — on - // the `events` parent and every partition. `CREATE TABLE .. PARTITION OF` - // clones parent triggers, but a partition attached with `ATTACH - // PARTITION` or created by an older code path would silently escape the - // guard, so migration fails closed if any is missing. (The fence probe - // re-runs this same check at startup on non-migrating relays.) - crate::replica_fence::verify_floor_guard_catalog(&mut *conn).await?; Ok(()) } @@ -902,7 +894,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 86); + assert_eq!(migrations.len(), 87); assert!(migrations[83] .sql .as_str() @@ -915,6 +907,11 @@ mod postgres_tests { .sql .as_str() .contains("CREATE TABLE media_uploaders")); + // 0087 retires the deferred created_at floor trigger and its function. + assert_eq!(migrations[86].version, 87); + let retire_floor = strip_sql_comments(migrations[86].sql.as_str()); + assert!(retire_floor.contains("DROP TRIGGER IF EXISTS events_created_at_floor ON events;")); + assert!(retire_floor.contains("DROP FUNCTION IF EXISTS events_created_at_floor_guard();")); for (index, migration) in migrations.iter().enumerate() { assert_eq!(migration.version, (index + 1) as i64); } @@ -1330,12 +1327,6 @@ mod postgres_tests { .as_str() .contains("CREATE TABLE join_policy_acceptances")); - // Replica-fence commit-time floor guard on channel-bearing events. - assert_eq!(migrations[20].version, 21); - assert!(migrations[20] - .sql - .as_str() - .contains("events_created_at_floor_guard")); assert!(!migrations[0] .sql .as_str() diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 5158954db5e..3f7e7b1bf16 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -423,8 +423,8 @@ pub(crate) mod route_proof { use uuid::Uuid; /// Proof that a query/page can only return rows with - /// `channel_id IS NOT NULL` — the domain of the commit-time floor guard - /// (migration 0021). `channel_ids` (retains channel-NULL rows) and + /// `channel_id IS NOT NULL` — the domain of the application floor check + /// (see [`replica_fence`]). `channel_ids` (retains channel-NULL rows) and /// `global_only = false` are explicitly NOT proofs. /// /// Each constructor keys off *how* its path proves channel-bearing-ness: @@ -482,18 +482,18 @@ pub(crate) enum RoutePredicate { /// read budget `B` (default off). Bounds TIME — the page misses at most /// the freshest `B` of writes. Sound for ANY query shape, including /// global (channel-NULL) rows: it relies only on heartbeat commit order, - /// not the floor guard. + /// not the floor check. Bounded, /// Completeness: the proved wall must cover the page's upper bound. /// Bounds CONTENT — every row at/below `upper` is present, meaningful /// even when the cursor is hours old, where `B`-freshness says nothing. - /// Sound ONLY on the floor guard's domain (channel-bearing rows), hence + /// Sound ONLY on the application floor's domain (channel-bearing rows), hence /// the proof token. `upper` is non-optional: the no-upper-bound /// post-verifying case is [`RoutePredicate::CoveredPostVerified`]. /// /// Bounds INSERT-completeness only — "no missing rows", not "no extra /// rows". Soft deletes are `UPDATE .. SET deleted_at` commits outside - /// the floor guard and never touch `created_at`, so a covered page can + /// the floor check and never touch `created_at`, so a covered page can /// briefly serve a row the writer already excludes; deletion visibility /// is bounded by replication lag under `FENCE_STALENESS` (30s), not by /// `upper` or `B`. Do not extend the covered arm to a surface that @@ -816,10 +816,6 @@ impl Db { /// When `config.read_database_url` is set, a second pool with the same /// sizing is connected to it for lag-tolerant reads (see [`Db::read`]). /// - /// The writer pool arms the commit-time `created_at` floor guard - /// (migration 0021) on every connection by setting the - /// `buzz.created_at_floor` GUC — this is what makes the replica fence - /// proof hold for every insert path that goes through this pool. pub async fn new(config: &DbConfig) -> Result { let pool = Self::connect_writer_pool(config).await?; let read_max_connections = config @@ -863,11 +859,10 @@ impl Db { /// Connect the writer pool with all session-level safety premises. /// - /// SQLx stores one `after_connect` hook, so the floor guard and transaction - /// isolation assertion must remain in this single closure. Registering a - /// second hook replaces the first and silently disarms the floor trigger. + /// SQLx stores one `after_connect` hook, so session timeouts and the + /// transaction isolation assertion must remain in this single closure. /// Additional writer pools, including the relay audit pool, must use this - /// constructor so they inherit the timeout, floor-guard, and isolation + /// constructor so they inherit the timeout and isolation /// policy installed by [`Db::new`]. /// /// Two single-connection side pools deliberately do not, so serving-pool @@ -913,22 +908,6 @@ impl Db { // point at which DNS/network/TLS/authentication succeeded. record_milestone(DbPoolRole::Writer, DbConnectionStep::PhysicalConnect); - // `SET` cannot take bind parameters; `set_config` can. - let floor = DbConnectionStepAttempt::start( - DbPoolRole::Writer, - DbConnectionStep::CreatedAtFloor, - ); - if let Err(error) = - sqlx::query("SELECT set_config('buzz.created_at_floor', $1, false)") - .bind(replica_fence::CREATED_AT_FLOOR_SECS.to_string()) - .execute(&mut *conn) - .await - { - floor.fail(); - return Err(error); - } - floor.succeed(); - // `lock_timeout` fails the waiting statement; it does not // cancel the holder. `idle_in_transaction_session_timeout` // reaps only holders idling inside an open transaction, @@ -1072,9 +1051,8 @@ impl Db { /// and every released connection still passes the bounded release check /// in [`Db::read_pool_options`]. /// - /// No floor guard or writer-isolation assertion: replica sessions are - /// read-only, so the commit-time trigger from migration 0021 never fires - /// here and the write fence that depends on READ COMMITTED is never reached. + /// No writer-isolation assertion: replica sessions are read-only, so the + /// write fence that depends on READ COMMITTED is never reached here. /// /// Every replica session's `statement_timeout` is lowered to the /// event-read deadline ([`crate::event::READ_STATEMENT_TIMEOUT`]) unless @@ -1462,27 +1440,14 @@ impl Db { &self.fence } - /// Verify the floor guard end-to-end, then spawn the background fence - /// probe. Returns `Ok(false)` when no replica is configured. - /// - /// Ordering matters (Perci, PR #2084 review): this must run **after** - /// the migration decision. On a relay with `BUZZ_AUTO_MIGRATE` off, the - /// writer pool arms the GUC regardless, but if migration 0021 has not - /// been applied there is no trigger enforcing it — and a heartbeat probe - /// would open the fence over an unenforced floor. So the probe is gated - /// on an unconditional two-part verification against the live schema: - /// catalog shape ([`replica_fence::verify_floor_guard_catalog`]) and - /// observed semantics through this exact pool - /// ([`replica_fence::verify_floor_guard_behavior`]). - /// - /// On any verification failure the probe is never spawned and the fence - /// stays closed: every cursor page routes to the writer. The relay keeps - /// serving — degraded capacity, never holes. - pub async fn spawn_fence_probe(&self) -> Result { + /// Spawn the background fence probe after the migration decision. + /// Returns `false` when no replica is configured. Probe failures keep + /// the fence closed and cursor reads on the writer. Application floor + /// coverage is enforced by the supported-writer contract and fleet gate. + pub fn spawn_fence_probe(&self) -> bool { if self.read_pool.is_none() { - return Ok(false); + return false; } - self.verify_replica_fence_at_boot().await?; // The probe gets its own connection so serving-pool saturation // cannot delay the beat (see `replica_fence::connect_probe_pool`). tokio::spawn(replica_fence::run_probe( @@ -1490,18 +1455,7 @@ impl Db { std::sync::Arc::clone(&self.fence), self.fence_probe_interval, )); - Ok(true) - } - - /// Verify replica-fence catalog shape and behavior through attributed - /// writer/bootstrap acquisitions without starting the recurring probe. - pub(crate) async fn verify_replica_fence_at_boot(&self) -> Result<()> { - let mut connection = - observability::acquire_writer(&self.pool, observability::WriterOperation::Bootstrap) - .await?; - replica_fence::verify_floor_guard_catalog(&mut *connection).await?; - drop(connection); - replica_fence::verify_floor_guard_behavior(&self.pool).await + true } /// The pool for lag-tolerant reads: the read replica when configured, @@ -2259,3 +2213,7 @@ fn media_authorization_slots(pool: &PgPool) -> std::sync::Arc( - executor: impl sqlx::PgExecutor<'e>, -) -> crate::Result<()> { - // tgtype bits: 1 = ROW, 2 = BEFORE, 4 = INSERT, 16 = UPDATE, 64 = INSTEAD. - // Required: ROW + INSERT + UPDATE set, BEFORE + INSTEAD clear. - let missing: Vec = sqlx::query_scalar( - r#" - SELECT c.relname::text - FROM ( - SELECT 'events'::regclass AS oid - UNION ALL - SELECT inhrelid FROM pg_inherits WHERE inhparent = 'events'::regclass - ) rels - JOIN pg_class c ON c.oid = rels.oid - WHERE NOT EXISTS ( - SELECT 1 FROM pg_trigger t - WHERE t.tgrelid = rels.oid - AND t.tgname = 'events_created_at_floor' - AND t.tgfoid = 'events_created_at_floor_guard'::regproc - AND t.tgdeferrable - AND t.tginitdeferred - AND t.tgtype & 1 = 1 -- row-level - AND t.tgtype & 2 = 0 -- AFTER, not BEFORE - AND t.tgtype & 64 = 0 -- not INSTEAD OF - AND t.tgtype & 4 = 4 -- fires on INSERT - AND t.tgtype & 16 = 16 -- fires on UPDATE - ) - "#, - ) - .fetch_all(executor) - .await?; - if !missing.is_empty() { - return Err(crate::error::DbError::InvalidData(format!( - "created_at floor guard trigger missing or mis-shaped on: {} \ - (replica fence must stay closed)", - missing.join(", ") - ))); - } - Ok(()) -} - -/// Behavioral verification of the floor guard, end-to-end through the armed -/// pool. A catalog check cannot detect a no-op function body or an unarmed -/// pool; this proves the semantics the fence proof cites, inside one -/// rolled-back transaction: -/// -/// 1. the pool's session GUC equals [`CREATED_AT_FLOOR_SECS`] (arming); -/// 2. an old channel-bearing INSERT raises `check_violation` (23514); -/// 3. a fresh channel-bearing INSERT commits; -/// 4. rewriting a fresh row's `created_at` below the floor raises; -/// 5. an old channel-NULL INSERT is exempt, but flipping its `channel_id` -/// on raises (the `UPDATE OF` arm). -/// -/// `SET CONSTRAINTS ALL IMMEDIATE` makes the deferred trigger fire per -/// statement so each adversary is observable under a savepoint; deferral to -/// COMMIT is separately pinned by the held-transaction fixture. -#[datastore_span(name = "replica_fence_verify_behavior", system = "postgresql")] -pub async fn verify_floor_guard_behavior(pool: &PgPool) -> crate::Result<()> { - use crate::error::DbError; - - let expect_violation = |res: Result, - what: &str| - -> crate::Result<()> { - match res { - Err(sqlx::Error::Database(e)) if e.code().as_deref() == Some("23514") => Ok(()), - Ok(_) => Err(DbError::InvalidData(format!( - "floor guard is inert: {what} was accepted (replica fence must stay closed)" - ))), - Err(e) => Err(DbError::InvalidData(format!( - "floor guard verification failed unexpectedly on {what}: {e}" - ))), - } - }; - - let connection = crate::observability::acquire_writer( - pool, - crate::observability::WriterOperation::Bootstrap, - ) - .await?; - let mut tx = sqlx::Transaction::begin(connection, None).await?; - - // 1. Pool arming (Perci: assert the effective value, not the intent). - let armed: String = sqlx::query_scalar("SHOW buzz.created_at_floor") - .fetch_one(&mut *tx) - .await - .map_err(|e| { - DbError::InvalidData(format!( - "buzz.created_at_floor GUC not set on this pool: {e}" - )) - })?; - if armed != CREATED_AT_FLOOR_SECS.to_string() { - return Err(DbError::InvalidData(format!( - "buzz.created_at_floor is '{armed}', expected '{CREATED_AT_FLOOR_SECS}': \ - pool is not armed" - ))); - } - - sqlx::query("SET CONSTRAINTS ALL IMMEDIATE") - .execute(&mut *tx) - .await?; - - // Scratch community satisfying the FK; the whole transaction rolls back. - let community = uuid::Uuid::new_v4(); - sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") - .bind(community) - .bind(format!("fence-verify-{}.invalid", community.simple())) - .execute(&mut *tx) - .await?; - let channel = uuid::Uuid::new_v4(); - - let insert = |tx_id: [u8; 32], age_secs: i64, ch: Option| { - sqlx::query( - "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, \ - content, sig, received_at, channel_id) \ - VALUES ($1, $2, $3, clock_timestamp() - make_interval(secs => $4::double precision), \ - 9, '[]', 'fence-verify', $5, NOW(), $6)", - ) - .bind(community) - .bind(tx_id.to_vec()) - .bind(vec![0u8; 32]) - .bind(age_secs as f64) - .bind(vec![0u8; 64]) - .bind(ch) - }; - let old_age = CREATED_AT_FLOOR_SECS + 60; - - // 2. Old channel-bearing insert must raise. - sqlx::query("SAVEPOINT floor_probe") - .execute(&mut *tx) - .await?; - let res = insert(rand_id(), old_age, Some(channel)) - .execute(&mut *tx) - .await; - expect_violation(res, "an old channel-bearing INSERT")?; - sqlx::query("ROLLBACK TO SAVEPOINT floor_probe") - .execute(&mut *tx) - .await?; - - // 3. Fresh channel-bearing insert must pass. - let fresh_id = rand_id(); - insert(fresh_id, 0, Some(channel)).execute(&mut *tx).await?; - - // 4. Rewriting created_at below the floor must raise (in- or - // cross-partition, either arm of the guard catches the NEW row). - sqlx::query("SAVEPOINT floor_probe") - .execute(&mut *tx) - .await?; - let res = sqlx::query( - "UPDATE events SET created_at = clock_timestamp() - make_interval(secs => $1::double precision) \ - WHERE community_id = $2 AND id = $3", - ) - .bind(old_age as f64) - .bind(community) - .bind(fresh_id.to_vec()) - .execute(&mut *tx) - .await; - expect_violation(res, "rewriting created_at below the floor")?; - sqlx::query("ROLLBACK TO SAVEPOINT floor_probe") - .execute(&mut *tx) - .await?; - - // 5. Old channel-NULL insert is exempt; flipping channel_id on must raise. - let null_id = rand_id(); - insert(null_id, old_age, None).execute(&mut *tx).await?; - sqlx::query("SAVEPOINT floor_probe") - .execute(&mut *tx) - .await?; - let res = sqlx::query("UPDATE events SET channel_id = $1 WHERE community_id = $2 AND id = $3") - .bind(channel) - .bind(community) - .bind(null_id.to_vec()) - .execute(&mut *tx) - .await; - expect_violation(res, "moving an old channel-NULL row into a channel")?; - sqlx::query("ROLLBACK TO SAVEPOINT floor_probe") - .execute(&mut *tx) - .await?; - - tx.rollback().await?; - Ok(()) -} - -fn rand_id() -> [u8; 32] { - let mut id = [0u8; 32]; - for chunk in id.chunks_mut(16) { - chunk.copy_from_slice(&uuid::Uuid::new_v4().into_bytes()[..chunk.len()]); - } - id -} - /// One writer-side sample of the ordered handshake. #[derive(Debug)] struct WriterSample { @@ -860,17 +657,14 @@ async fn sample_writer( // conservative direction (a long autovacuum merely holds the fence // back), and it keeps the classification simple. // Scope is every other backend regardless of role or application - // name: an admin psql transaction writes under the same trigger and - // must be representable in the oldest-xact term. + // name: every open transaction must be representable in the + // oldest-xact term. Out-of-band writes require a closed breaker. // // Prepared transactions (2PC) are a bucket of their own: while // prepared they have left `pg_stat_activity` but can still commit - // after the token. Their deferred floor guard already ran at PREPARE, - // so `pg_prepared_xacts.prepared` bounds their rows exactly like - // `xact_start`; fold it into the same minimum. Only the retained - // deferred trigger provides that PREPARE-time check: the insert-time - // app guard runs earlier, and no app writer issues PREPARE TRANSACTION. - // Dropping the trigger must keep this case covered. + // after the token. Their floor check must precede PREPARE, so + // `pg_prepared_xacts.prepared` bounds their rows exactly like + // `xact_start`; fold it into the same minimum. let row = sqlx::query( r#" SELECT diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 6fa4a720a76..7af3b287b33 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2235,7 +2235,7 @@ async fn saturated_reader_spends_one_acquire_budget_and_rejects_history_reads() seed.close().await; let scratch_url = scratch.url().to_owned(); - // `Db::new` so the writer arms the floor guard and the reader is the + // `Db::new` so the writer uses production session policy and the reader is the // real lazy `connect_read_pool` pool (warm floor of 1, 150ms // acquire timeout). Reader is sized 1 so holding one connection // saturates it. @@ -2569,9 +2569,8 @@ async fn routed_reads_are_confined_to_the_requested_community() { } /// D4: a LAZY reader pool (connect_lazy, never yet -/// used) must still let [`Db::spawn_fence_probe`] verify the writer's -/// floor guard and spawn — reader-down or reader-idle at boot must not -/// disable fence probing. +/// used) must still let [`Db::spawn_fence_probe`] spawn — reader-down or +/// reader-idle at boot must not disable fence probing. #[tokio::test] #[ignore = "requires Postgres"] async fn lazy_reader_pool_still_spawns_fence_probe() { @@ -2579,11 +2578,7 @@ async fn lazy_reader_pool_still_spawns_fence_probe() { seed.close().await; let writer_url = scratch.url().to_owned(); - // `Db::new` (not `from_pools`) so the WRITER pool arms the - // `buzz.created_at_floor` GUC — `spawn_fence_probe` verifies the - // floor guard on a writer connection, and `ScratchDb::migrated`'s - // plain `PgPool::connect` never arms it. The reader is still the - // lazy `connect_read_pool` pool this test is about. + // Use the production writer policy and lazy `connect_read_pool`. let db = Db::new(&DbConfig { database_url: writer_url.clone(), read_database_url: Some(writer_url), @@ -2591,12 +2586,9 @@ async fn lazy_reader_pool_still_spawns_fence_probe() { ..DbConfig::default() }) .await - .expect("connect armed Db with lazy reader"); + .expect("connect Db with lazy reader"); - let spawned = db - .spawn_fence_probe() - .await - .expect("floor-guard verification must pass on the migrated writer"); + let spawned = db.spawn_fence_probe(); assert!(spawned, "a configured (lazy) reader must spawn the probe"); } @@ -2929,95 +2921,6 @@ async fn thread_full_replica_page_above_fence_is_reverified_on_writer() { ); } -/// Commit-time floor guard (migration 0021), exact held-transaction -/// adversary: a channel-bearing row whose `created_at` is older than the -/// floor at COMMIT time must abort the transaction — the guard runs -/// inside commit processing with `clock_timestamp()`, so holding the -/// transaction open cannot outrun it. channel_id-NULL rows are -/// structurally exempt, and sessions without the GUC are unaffected. -#[tokio::test] -#[ignore = "requires Postgres"] -async fn created_at_floor_guard_aborts_old_channel_rows_at_commit() { - let (pool, _scratch) = ScratchDb::migrated(&admin_url().await, "floor_guard").await; - - let author = nostr::Keys::generate(); - let community = Uuid::new_v4(); - let channel = Uuid::new_v4(); - seed_community_channel(&pool, community, channel, &author).await; - - let insert_raw = |ev: nostr::Event, channel_id: Option| { - let pool = pool.clone(); - async move { - let mut tx = pool.begin().await.expect("begin"); - // Arm the guard for this transaction only (the relay's - // writer pool arms it per connection; tests are explicit). - sqlx::query("SELECT set_config('buzz.created_at_floor', $1, true)") - .bind(crate::replica_fence::CREATED_AT_FLOOR_SECS.to_string()) - .execute(&mut *tx) - .await - .expect("arm guard"); - sqlx::query( - "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, \ - content, sig, received_at, channel_id) \ - VALUES ($1, $2, $3, to_timestamp($4), 9, '[]', $5, $6, NOW(), $7)", - ) - .bind(community) - .bind(ev.id.as_bytes().as_slice()) - .bind(ev.pubkey.to_bytes().as_slice()) - .bind(ev.created_at.as_secs() as f64) - .bind(&ev.content) - .bind(ev.sig.serialize().as_slice()) - .bind(channel_id) - .execute(&mut *tx) - .await - .expect("insert inside tx (guard is deferred to commit)"); - // Hold the transaction "open" past the insert, then commit — - // the deferred guard must still see the stale created_at. - sqlx::query("SELECT pg_sleep(0.05)") - .execute(&mut *tx) - .await - .expect("hold tx"); - tx.commit().await - } - }; - - let now_secs = chrono::Utc::now().timestamp() as u64; - let floor = crate::replica_fence::CREATED_AT_FLOOR_SECS as u64; - - // Old channel-bearing row → COMMIT aborts with check_violation. - let old = signed_event_at(&author, "old-held-tx", now_secs - floor - 60); - let err = insert_raw(old, Some(channel)) - .await - .expect_err("below-floor channel row must abort at COMMIT"); - let code = match &err { - sqlx::Error::Database(db_err) => db_err.code().map(|c| c.to_string()), - other => panic!("expected database error, got {other:?}"), - }; - assert_eq!( - code.as_deref(), - Some("23514"), - "guard raises check_violation" - ); - - // Fresh channel-bearing row → commits. - let fresh = signed_event_at(&author, "fresh", now_secs); - insert_raw(fresh, Some(channel)) - .await - .expect("fresh row commits under the armed guard"); - - // Old row WITHOUT a channel (push lease / profile shapes) → - // structurally exempt, commits. - let old_global = signed_event_at(&author, "old-global", now_secs - floor - 60); - insert_raw(old_global, None) - .await - .expect("channel_id-NULL rows are exempt from the floor"); - - // Unarmed session (no GUC) → guard inert; backfills stay possible - // (and must hold the fence closed, per the migration header). - let old_backfill = signed_event_at(&author, "old-backfill", now_secs - floor - 60); - insert_top_level(&pool, community, channel, &old_backfill).await; -} - #[test] fn writer_pool_safety_hook_is_single_and_composed() { let source = include_str!("mod.rs"); @@ -3031,13 +2934,10 @@ fn writer_pool_safety_hook_is_single_and_composed() { 1, "SQLx replaces after_connect hooks; writer safety must use exactly one" ); - assert!(connect_pool.contains("buzz.created_at_floor")); assert!(connect_pool.contains("SHOW transaction_isolation")); assert!(connect_pool.contains("'lock_timeout'")); assert!(connect_pool.contains("'idle_in_transaction_session_timeout'")); assert!(connect_pool.contains("'statement_timeout'")); - assert!(!connect_pool.contains("arm_floor_guard")); - assert!(!connect_pool.contains("_arm_floor_guard")); assert!(!connect_pool.contains("allow(unused_variables)")); let reader_doc = source @@ -3094,7 +2994,6 @@ async fn writer_pool_metrics_record_every_initial_connection_ready() { 2 ); for step in [ - DbConnectionStep::CreatedAtFloor, DbConnectionStep::SessionTimeouts, DbConnectionStep::Isolation, ] { @@ -3529,100 +3428,6 @@ async fn session_timeouts_install_through_db_new_and_bound_lock_waits() { drop(holder); } -/// The armed writer pool (`Db::new`) must enforce the floor end-to-end -/// through the public insert APIs, and the session GUC must be verifiably -/// set on pooled connections. -#[tokio::test] -#[ignore = "requires Postgres"] -async fn armed_pool_rejects_old_channel_inserts_through_public_api() { - let (seed_pool, scratch) = ScratchDb::migrated(&admin_url().await, "floor_pool").await; - - let author = nostr::Keys::generate(); - let community = Uuid::new_v4(); - let channel = Uuid::new_v4(); - seed_community_channel(&seed_pool, community, channel, &author).await; - - // Connect a Db the production way: after_connect arms the guard. - let scratch_url = scratch.url().to_owned(); - let db = Db::new(&DbConfig { - database_url: scratch_url, - max_connections: 2, - ..DbConfig::default() - }) - .await - .expect("connect armed Db"); - let cid = CommunityId::from_uuid(community); - - // Perci nit: assert the effective session value, not the intent. - let effective: String = sqlx::query_scalar("SHOW buzz.created_at_floor") - .fetch_one(&db.pool) - .await - .expect("SHOW guard GUC"); - assert_eq!( - effective, - crate::replica_fence::CREATED_AT_FLOOR_SECS.to_string(), - "writer pool must arm the floor guard on every connection" - ); - let isolation: String = sqlx::query_scalar("SHOW transaction_isolation") - .fetch_one(&db.pool) - .await - .expect("SHOW writer isolation"); - assert_eq!( - isolation, "read committed", - "the same writer after_connect hook must enforce the isolation premise" - ); - - let now_secs = chrono::Utc::now().timestamp() as u64; - let floor = crate::replica_fence::CREATED_AT_FLOOR_SECS as u64; - - // insert_event (single INSERT, autocommit): old channel row rejected. - let old = signed_event_at(&author, "old-direct", now_secs - floor - 60); - let err = event::insert_event(&db.pool, cid, &old, Some(channel)) - .await - .expect_err("armed pool must reject below-floor channel inserts"); - assert!( - err.to_string().contains("below the replica-fence floor"), - "unexpected error: {err}" - ); - - // insert_event_with_thread_metadata (multi-statement tx): same. - let old2 = signed_event_at(&author, "old-thread-meta", now_secs - floor - 90); - let ts = - chrono::DateTime::from_timestamp(old2.created_at.as_secs() as i64, 0).expect("valid ts"); - let err = event::insert_event_with_thread_metadata( - &db.pool, - cid, - &old2, - Some(channel), - Some(event::ThreadMetadataParams { - event_id: old2.id.as_bytes(), - event_created_at: ts, - channel_id: channel, - parent_event_id: None, - parent_event_created_at: None, - root_event_id: None, - root_event_created_at: None, - depth: 0, - broadcast: true, - }), - ) - .await - .expect_err("armed pool must reject below-floor thread-metadata inserts"); - assert!( - err.to_string().contains("below the replica-fence floor"), - "unexpected error: {err}" - ); - - // Fresh events pass through both APIs. - let fresh = signed_event_at(&author, "fresh-direct", now_secs); - event::insert_event(&db.pool, cid, &fresh, Some(channel)) - .await - .expect("fresh insert passes the armed guard"); - - // Close before the `scratch` guard force-drops the database on return. - db.pool.close().await; -} - /// A writer transaction holding the shared replica-floor advisory lock, the /// shape a floor-compliant writer takes before the exclusive probe can run. async fn begin_replica_floor_locked_writer(db: &Db) -> sqlx::Transaction<'static, sqlx::Postgres> { @@ -3630,7 +3435,7 @@ async fn begin_replica_floor_locked_writer(db: &Db) -> sqlx::Transaction<'static .pool .begin() .await - .expect("begin floor-guarded writer tx"); + .expect("begin floor-compliant writer tx"); sqlx::query("SELECT pg_advisory_xact_lock_shared($1)") .bind(crate::replica_fence::REPLICA_FLOOR_LOCK_KEY) .execute(&mut *tx) @@ -3748,172 +3553,6 @@ async fn cluster_global_replica_floor_probe_waits_for_shared_writer_and_records_ db.pool.close().await; } -/// `spawn_fence_probe` must verify the floor guard before letting the -/// probe run — catalog shape AND observed behavior — and refuse on -/// sabotage. This is the production gate for a relay running with -/// `BUZZ_AUTO_MIGRATE` off: an armed GUC with no enforcing trigger must -/// never yield an open fence. -#[tokio::test] -#[ignore = "requires Postgres"] -async fn fence_probe_refuses_to_start_without_verified_floor_guard() { - let (seed_pool, writer_db) = ScratchDb::migrated(&admin_url().await, "fence_gate_w").await; - let (replica_pool, replica_db) = ScratchDb::migrated(&admin_url().await, "fence_gate_r").await; - seed_pool.close().await; - replica_pool.close().await; - - let writer_url = writer_db.url().to_owned(); - let replica_url = replica_db.url().to_owned(); - - // Healthy schema: verification passes, probe starts. A SEPARATE Db - // instance, because its background probe legitimately opens its own - // fence (the heartbeat probe is writer-side only) — the refusal - // assertions below must run against a fence whose spawns were all - // refused. - let db_healthy = Db::new(&DbConfig { - database_url: writer_url.clone(), - read_database_url: Some(replica_url.clone()), - max_connections: 2, - ..DbConfig::default() - }) - .await - .expect("connect armed Db with replica"); - assert!( - db_healthy - .spawn_fence_probe() - .await - .expect("verification passes"), - "probe must start on a verified schema" - ); - - let db = Db::new(&DbConfig { - database_url: writer_url, - read_database_url: Some(replica_url), - max_connections: 2, - ..DbConfig::default() - }) - .await - .expect("connect armed Db with replica"); - - // Sabotage A: catalog-shaped no-op — same trigger, gutted function - // body. Catalog check alone would pass; behavior check must refuse. - sqlx::query( - "CREATE OR REPLACE FUNCTION events_created_at_floor_guard() RETURNS trigger \ - LANGUAGE plpgsql AS $$ BEGIN RETURN NULL; END $$", - ) - .execute(&db.pool) - .await - .expect("gut the guard function"); - let err = db - .spawn_fence_probe() - .await - .expect_err("inert guard body must refuse the probe"); - assert!( - err.to_string().contains("floor guard is inert"), - "unexpected error: {err}" - ); - - // Sabotage B: trigger dropped entirely (the BUZZ_AUTO_MIGRATE=off / - // 0021-unapplied shape). Catalog check must refuse. - sqlx::query("DROP TRIGGER events_created_at_floor ON events") - .execute(&db.pool) - .await - .expect("drop the guard trigger"); - let err = db - .spawn_fence_probe() - .await - .expect_err("missing trigger must refuse the probe"); - assert!( - err.to_string().contains("missing or mis-shaped"), - "unexpected error: {err}" - ); - - // In both refusal states the fence never opened. - assert!( - db.fence().verified_through().is_none(), - "fence must remain closed when verification refuses the probe" - ); -} - -/// The `UPDATE OF` arm of the floor guard (Perci's second structural -/// hole): an old row legitimately admitted with `channel_id` NULL must -/// not be movable into keyset windows, and a channel row's `created_at` -/// must not be movable below the fence — through raw SQL, at COMMIT. -#[tokio::test] -#[ignore = "requires Postgres"] -async fn floor_guard_blocks_updates_that_move_rows_below_the_fence() { - let (pool, _scratch) = ScratchDb::migrated(&admin_url().await, "floor_upd").await; - - let author = nostr::Keys::generate(); - let community = Uuid::new_v4(); - let channel = Uuid::new_v4(); - seed_community_channel(&pool, community, channel, &author).await; - - let now_secs = chrono::Utc::now().timestamp() as u64; - let floor = crate::replica_fence::CREATED_AT_FLOOR_SECS as u64; - - // Seed via unarmed session: one old channel-NULL row, one fresh - // channel row. - let old_null = signed_event_at(&author, "old-null", now_secs - floor - 120); - insert_top_level(&pool, community, channel, &old_null).await; - sqlx::query("UPDATE events SET channel_id = NULL WHERE community_id = $1 AND id = $2") - .bind(community) - .bind(old_null.id.as_bytes().as_slice()) - .execute(&pool) - .await - .expect("detach channel (unarmed seed)"); - let fresh = signed_event_at(&author, "fresh-row", now_secs); - insert_top_level(&pool, community, channel, &fresh).await; - - // Armed transaction, deferred to COMMIT (the production shape). - let run_armed_update = |sql: &'static str, id: Vec, age: Option| { - let pool = pool.clone(); - async move { - let mut tx = pool.begin().await.expect("begin"); - sqlx::query("SELECT set_config('buzz.created_at_floor', $1, true)") - .bind(crate::replica_fence::CREATED_AT_FLOOR_SECS.to_string()) - .execute(&mut *tx) - .await - .expect("arm guard"); - let q = sqlx::query(sql).bind(community).bind(id); - let q = match age { - Some(a) => q.bind(a as f64), - None => q, - }; - q.execute(&mut *tx) - .await - .expect("update inside tx (deferred)"); - tx.commit().await - } - }; - - // channel-NULL → channel-bearing on an old row: COMMIT must abort. - let err = run_armed_update( - "UPDATE events SET channel_id = community_id WHERE community_id = $1 AND id = $2", - old_null.id.as_bytes().to_vec(), - None, - ) - .await - .expect_err("moving an old channel-NULL row into a channel must abort at COMMIT"); - assert!( - matches!(&err, sqlx::Error::Database(e) if e.code().as_deref() == Some("23514")), - "unexpected error: {err}" - ); - - // created_at rewrite below the floor on a channel row: COMMIT must abort. - let err = run_armed_update( - "UPDATE events SET created_at = clock_timestamp() - make_interval(secs => $3::double precision) \ - WHERE community_id = $1 AND id = $2", - fresh.id.as_bytes().to_vec(), - Some(floor + 120), - ) - .await - .expect_err("rewriting created_at below the floor must abort at COMMIT"); - assert!( - matches!(&err, sqlx::Error::Database(e) if e.code().as_deref() == Some("23514")), - "unexpected error: {err}" - ); -} - // ---- Statement-deadline cancellation vs. replica fallback ---------------- // // A replica read cancelled by `statement_timeout` (SQLSTATE 57014) must be diff --git a/crates/buzz-db/src/runtime/tests/floor_retirement.rs b/crates/buzz-db/src/runtime/tests/floor_retirement.rs new file mode 100644 index 00000000000..cf938cb7d10 --- /dev/null +++ b/crates/buzz-db/src/runtime/tests/floor_retirement.rs @@ -0,0 +1,161 @@ +//! Contract-migration coverage; application insert coverage belongs to phase 1. + +use super::*; + +async fn floor_objects(pool: &PgPool) -> (i64, i64) { + sqlx::query_as( + "SELECT (SELECT count(*) FROM pg_trigger WHERE tgname = 'events_created_at_floor'), \ + (SELECT count(*) FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace \ + WHERE n.nspname = 'public' AND p.proname = 'events_created_at_floor_guard')", + ) + .fetch_one(pool) + .await + .expect("read floor catalog") +} + +async fn retained_guards(pool: &PgPool) -> Vec<(String, String)> { + sqlx::query_as( + "SELECT c.relname, pg_get_triggerdef(t.oid) FROM pg_trigger t \ + JOIN pg_class c ON c.oid = t.tgrelid \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE n.nspname = 'public' AND t.tgname <> 'events_created_at_floor' \ + UNION ALL \ + SELECT c.relname, pg_get_constraintdef(k.oid) FROM pg_constraint k \ + JOIN pg_class c ON c.oid = k.conrelid \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE n.nspname = 'public' AND k.contype = 'f' \ + ORDER BY 1, 2", + ) + .fetch_all(pool) + .await + .expect("read retained triggers and foreign keys") +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_floor_retirement_drops_parent_and_partition_clones() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect isolated migration database"); + crate::migration::run_migrations_through(&pool, 60) + .await + .expect("apply pre-contract schema"); + let partitions: i64 = + sqlx::query_scalar("SELECT count(*) FROM pg_partition_tree('events'::regclass)") + .fetch_one(&pool) + .await + .expect("count parent and partitions"); + assert_eq!(floor_objects(&pool).await, (partitions, 1)); + let before = retained_guards(&pool).await; + crate::migration::run_migrations(&pool) + .await + .expect("apply contract migration and retained startup checks"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + assert_eq!(retained_guards(&pool).await, before); + pool.close().await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_floor_retirement_cleans_child_copies_idempotently() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect isolated migration database"); + crate::migration::run_migrations_through(&pool, 60) + .await + .expect("apply pre-contract schema"); + // Model independently attached copies after the parent guard was removed. + sqlx::raw_sql( + "DROP TRIGGER events_created_at_floor ON events; \ + CREATE CONSTRAINT TRIGGER events_created_at_floor \ + AFTER INSERT OR UPDATE OF created_at, channel_id ON events_p_past \ + DEFERRABLE INITIALLY DEFERRED FOR EACH ROW \ + EXECUTE FUNCTION events_created_at_floor_guard();", + ) + .execute(&pool) + .await + .expect("install independent child copy"); + assert_eq!(floor_objects(&pool).await, (1, 1)); + let before = retained_guards(&pool).await; + crate::migration::run_migrations(&pool) + .await + .expect("retire independent copy"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + let mut tx = pool.begin().await.expect("begin idempotence check"); + sqlx::raw_sql(include_str!( + "../../../../../migrations/0087_retire_created_at_floor.sql" + )) + .execute(&mut *tx) + .await + .expect("repeat contract SQL after retirement"); + tx.commit().await.expect("commit repeated retirement"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + assert_eq!(retained_guards(&pool).await, before); + pool.close().await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn floor_retirement_desired_and_migrated_catalogs_match() { + // The nextest wrapper bootstraps this database through real pgschema apply + // and reconciliation. Check the retired floor catalog against the real + // migration boot path; unrelated trigger/FK preservation is checked above. + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect desired-state database"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + sqlx::raw_sql("DROP SCHEMA public CASCADE; CREATE SCHEMA public;") + .execute(&pool) + .await + .expect("reset isolated database for migration boot"); + crate::migration::run_migrations(&pool) + .await + .expect("boot migration schema"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + pool.close().await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn floor_retirement_writer_and_probe_accept_legacy_and_retired_schema() { + let url = crate::test_support::database_url(); + let db = Db::new(&DbConfig { + database_url: url.clone(), + read_database_url: Some(url), + max_connections: 2, + min_connections: 1, + ..DbConfig::default() + }) + .await + .expect("connect production writer with retired schema"); + let armed: Option = + sqlx::query_scalar("SELECT nullif(current_setting('buzz.created_at_floor', true), '')") + .fetch_one(&db.pool) + .await + .expect("inspect writer session setting"); + assert_eq!(armed, None, "writer must no longer arm the retired guard"); + assert_eq!(floor_objects(&db.pool).await, (0, 0)); + assert!(db.spawn_fence_probe()); + crate::replica_fence::probe_once(&db.pool, db.fence()) + .await + .expect("retired schema still supports the ordered heartbeat"); + + // This binary can also run before migration, without changing legacy SQL. + sqlx::raw_sql(include_str!( + "../../../../../migrations/0021_created_at_fence_floor.sql" + )) + .execute(&db.pool) + .await + .expect("restore actual legacy guard"); + assert!(floor_objects(&db.pool).await.0 > 1); + assert!(db.spawn_fence_probe()); + crate::replica_fence::probe_once(&db.pool, db.fence()) + .await + .expect("legacy schema still supports ordered heartbeat"); + db.pool.close().await; + db.read_pool + .as_ref() + .expect("reader configured") + .close() + .await; +} diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 98124926a9c..62889e3befc 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -2573,7 +2573,7 @@ impl Db { /// /// Counts route on the BOUNDED arm only, never covered: the covered /// arm bounds insert-completeness but not deletion visibility (soft - /// deletes are UPDATEs outside the floor guard), and a count has no + /// deletes are UPDATEs outside the floor check), and a count has no /// downstream per-row re-filter to absorb extra rows — a silently /// inflated number for up to `FENCE_STALENESS` is a different product /// statement than a page briefly showing a deleted row. `Bounded` ties diff --git a/crates/buzz-db/src/store/event_floor_postgres_tests.rs b/crates/buzz-db/src/store/event_floor_postgres_tests.rs index 24898fa7a1d..ba5e1b7aab5 100644 --- a/crates/buzz-db/src/store/event_floor_postgres_tests.rs +++ b/crates/buzz-db/src/store/event_floor_postgres_tests.rs @@ -1,4 +1,4 @@ -//! App-floor regressions with every floor-trigger clone disabled in the +//! App-floor regressions with no enabled floor-trigger clone in the //! nextest-provided per-test database, so the DB backstop cannot mask a bypass. use buzz_core::CommunityId; @@ -18,19 +18,14 @@ async fn fixture() -> (Db, CommunityId, Uuid, Keys) { .connect(&crate::test_support::database_url()) .await .expect("isolated test database"); - sqlx::query("ALTER TABLE events DISABLE TRIGGER events_created_at_floor") - .execute(&pool) - .await - .expect("disable floor trigger recursively on the partitioned table"); - let (total, enabled): (i64, i64) = sqlx::query_as( - "SELECT count(*), count(*) FILTER (WHERE tgenabled <> 'D') FROM pg_trigger \ - WHERE tgname = 'events_created_at_floor' \ + let enabled: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_trigger \ + WHERE tgname = 'events_created_at_floor' AND tgenabled <> 'D' \ AND tgrelid IN (SELECT relid FROM pg_partition_tree('events'))", ) .fetch_one(&pool) .await .expect("check every trigger clone"); - assert!(total > 0, "the test must disable an existing backstop"); assert_eq!(enabled, 0, "no partition may mask a missing app check"); let community = CommunityId::from_uuid(Uuid::new_v4()); sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 80dbcb8f048..286f52fa151 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -438,22 +438,11 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() { let runtime = include_str!("../src/runtime/mod.rs"); assert!(runtime.contains("observability::refresh_pool_waiters(self.read_pool.is_some())")); - assert!(runtime.contains("self.verify_replica_fence_at_boot().await?")); - let fence_boot = runtime - .split_once("pub(crate) async fn verify_replica_fence_at_boot") - .expect("runtime must expose attributed boot fence verification") - .1 - .split_once("/// The pool for lag-tolerant reads") - .expect("boot fence verification must precede routed-read plumbing") - .0; - assert!(fence_boot.contains("WriterOperation::Bootstrap")); - let replica_fence = include_str!("../src/runtime/replica_fence.rs"); let replica_fence_production = replica_fence .split("\n#[cfg(test)]") .next() .expect("replica-fence production source"); - assert!(replica_fence_production.contains("WriterOperation::Bootstrap")); // The heartbeat probe runs on its own pool, never a serving writer // checkout: queued behind request traffic, its beat goes stale exactly // when the writer is saturated. @@ -1707,7 +1696,7 @@ pub(crate) async fn mention_writer(tx: &mut AdmittedTx) { .await .expect("write"); } -pub async fn verify_floor_guard_behavior(pool: &PgPool) { +pub async fn verify_channel_roster_fence_behavior(pool: &PgPool) { sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") .execute(pool) .await @@ -2242,13 +2231,11 @@ fn every_search_hydrator_preserves_the_exact_hit_timestamp() { } } -// These startup probes always roll back; they deliberately exercise the DB -// backstops rather than the app check. The two serving writers are exempt only +// This startup probe always rolls back; it deliberately exercises the roster +// DB backstop rather than the app check. The two serving writers are exempt only // while their INSERT remains structurally channel-NULL (checked below). -const EVENT_FLOOR_PROBE_EXCEPTIONS: [&str; 2] = [ - "pub async fn verify_floor_guard_behavior(", - "pub async fn verify_channel_roster_fence_behavior(", -]; +const EVENT_FLOOR_PROBE_EXCEPTIONS: [&str; 1] = + ["pub async fn verify_channel_roster_fence_behavior("]; fn is_channel_null_event_writer(function: &str) -> bool { let header = function_header(function); diff --git a/crates/buzz-relay/src/api/admin/mod.rs b/crates/buzz-relay/src/api/admin/mod.rs index 8c0ed1b78f0..c29797185b6 100644 --- a/crates/buzz-relay/src/api/admin/mod.rs +++ b/crates/buzz-relay/src/api/admin/mod.rs @@ -9194,18 +9194,6 @@ mod postgres_tests { // `fail_outbox_row` (not `mark_outbox_delivered`) when the insert itself // fails — so nothing is durably persisted, and the row is NOT marked delivered. // - // The failure is induced AFTER tenant resolution, inside `emit_system_message`'s - // `insert_event` call, by: - // 1. Building a dedicated test pool whose `after_connect` sets the - // `buzz.created_at_floor` GUC session-locally (not database-globally). - // Every connection from that pool inherits the floor; no other pool or - // test is affected, and there is no cleanup race on panic. - // 2. Backdating the outbox row's `created_at` beyond that floor. - // `emit_system_message` derives the Nostr event's `created_at` from - // `row.created_at` (the idempotency timestamp). With the floor active, the - // deferrable trigger fires on INSERT and raises a check_violation, which - // `insert_event` propagates as `Err`. The `?` in `emit_system_message` then - // propagates it up through `deliver_tombstone → deliver_one → fail_outbox_row`. let pool = e2e_pool().await; let (community_id, _host) = e2e_community(&pool, "race-c3").await; let target = vec![24u8; 32]; @@ -9309,43 +9297,22 @@ mod postgres_tests { .await .expect("insert fail-outbox row"); - // Backdate the outbox row's created_at so emit_system_message uses an old - // idempotency_ts. The events_created_at_floor trigger will reject the INSERT - // once we arm the GUC below. - sqlx::query( - "UPDATE relay_admin_outbox SET created_at = now() - interval '10 seconds' WHERE id = $1", + // Inject a durable insert failure only in this per-test database. This + // exercises error propagation and retry state independently of the floor. + sqlx::raw_sql( + "CREATE FUNCTION reject_c3_system_message() RETURNS trigger LANGUAGE plpgsql AS $$ \ + BEGIN IF NEW.kind = 40099 THEN \ + RAISE EXCEPTION 'injected system-message insert failure' USING ERRCODE = '23514'; \ + END IF; RETURN NEW; END $$; \ + CREATE TRIGGER reject_c3_system_message BEFORE INSERT ON events \ + FOR EACH ROW EXECUTE FUNCTION reject_c3_system_message();", ) - .bind(fail_outbox_id) .execute(&pool) .await - .expect("backdate outbox created_at"); - - // Build a dedicated pool whose after_connect sets buzz.created_at_floor = 5 - // session-locally on each connection (set_config 3rd arg false = session scope). - // A floor of 5 s means any event with created_at > 5 s ago is rejected. - // Our outbox row's created_at is ~10 s ago → trigger fires on insert_event. - // This pool is fully isolated: no other pool or test is affected, and there - // is no cleanup dependence (dropping the pool closes all its connections). - let db_url = database_url(); - let floor_pool = sqlx::postgres::PgPoolOptions::new() - .max_connections(4) - .after_connect(|conn, _meta| { - Box::pin(async move { - sqlx::query("SELECT set_config('buzz.created_at_floor', '5', false)") - .execute(conn) - .await?; - Ok(()) - }) - }) - .connect(&db_url) - .await - .expect("connect floor pool"); - - // Build an AppState around the floor pool so deliver_one's insert_event call - // runs on a connection where the deferrable trigger is active. - let fresh_state = state_from_pool(floor_pool.clone()).await; + .expect("install isolated insert failure"); + let fresh_state = state_from_pool(pool.clone()).await; - // Claim the row via the floor pool so deliver_one has a real claim token. + // Claim the row so deliver_one has a real claim token. let lease_until = chrono::Utc::now() + chrono::Duration::seconds(30); let mut batch = fresh_state .db @@ -9358,14 +9325,18 @@ mod postgres_tests { .expect("fail_outbox_id must be in batch"); let row = batch.remove(row_idx); - // deliver_one fails inside emit_system_message at insert_event (deferrable - // floor-guard trigger → check_violation) and must call fail_outbox_row — + // deliver_one fails inside emit_system_message at insert_event and + // must call fail_outbox_row — // NOT mark_outbox_delivered. crate::handlers::admin_outbox_worker::deliver_one(&fresh_state, &row).await; - // Drop the floor pool — all its connections close, GUC vanishes with them. - // No ALTER DATABASE, no global state, no reset required. - drop(floor_pool); + sqlx::raw_sql( + "DROP TRIGGER reject_c3_system_message ON events; \ + DROP FUNCTION reject_c3_system_message();", + ) + .execute(&pool) + .await + .expect("remove isolated insert failure"); // No tombstone event was persisted — the failure was inside insert_event. let post_event_count: i64 = sqlx::query_scalar( diff --git a/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs b/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs index 35318b0c614..7430d7d7a99 100644 --- a/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs +++ b/crates/buzz-relay/src/api/bridge/thread_window/postgres_tests.rs @@ -30,7 +30,7 @@ impl Fixture { .await .unwrap(); let mut state = (*state).clone(); - // Use production after_connect policy (floor guard, isolation and + // Use production after_connect policy (isolation and // session timeouts), not raw SQLx pools that mask deployed failures. state.db = production_db(buzz_db::DbConfig::default()).await; Arc::make_mut(&mut state.config).require_auth_token = true; diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 532207a3dbc..0f8f049e327 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -410,26 +410,11 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { info!("Community deletion serving fences verified"); step.finish(); - // Freshness fence probe: cursor pages route to the replica only for - // history the probe has verified as fully replayed. Deliberately AFTER - // the migration decision: spawn_fence_probe first verifies the - // commit-time floor guard (catalog shape + observed behavior through the - // armed pool) against the live schema, so a relay running with - // BUZZ_AUTO_MIGRATE off and migration 0021 unapplied can never open the - // fence over an unenforced floor. Verification failure is loud but - // non-fatal: the fence stays closed and every cursor page routes to the - // writer. - let mut step = StepTimer::start(StartupStep::ReplicaFenceProbe); - match db.spawn_fence_probe().await { - Ok(true) => info!("Replica fence probe started (floor guard verified)"), - Ok(false) => {} - Err(e) => { - step.degrade(); - error!( - "Replica fence disabled — floor guard verification failed: {e}. \ - All cursor reads stay on the writer." - ); - } + // The ordered heartbeat probe opens replica routing only after a + // successful sample. Errors keep cursor reads on the writer. + let step = StepTimer::start(StartupStep::ReplicaFenceProbe); + if db.spawn_fence_probe() { + info!("Replica fence probe started"); } step.finish(); diff --git a/crates/buzz-relay/src/startup_steps.rs b/crates/buzz-relay/src/startup_steps.rs index bdda346a616..98776459882 100644 --- a/crates/buzz-relay/src/startup_steps.rs +++ b/crates/buzz-relay/src/startup_steps.rs @@ -51,7 +51,7 @@ pub enum StartupStep { PartitionEnsure, /// Verify the community-deletion serving fences. DeletionFenceVerify, - /// Verify the replica floor guard and start the fence probe. + /// Start the replica freshness fence probe. ReplicaFenceProbe, /// Ensure the deployment community, backfill the allowlist, and bootstrap the owner. MembershipBootstrap, diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index 328307ec613..606b5895296 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -362,7 +362,7 @@ can give it to a caller. | `buzz_db_connection_step_duration_seconds` | histogram | `pool_role`, `step` | | `buzz_db_connection_step_attempts_total` | counter | `pool_role`, `step`, `outcome` | -The fixed steps are `writer_pool`, `physical_connect`, `created_at_floor`, +The fixed steps are `writer_pool`, `physical_connect`, `session_timeouts`, `isolation`, and `ready`. Measurable phases emit start, duration, and terminal evidence. `physical_connect` and `ready` are success milestones: SQLx 0.9 exposes `after_connect` only after DNS, network, TLS, and @@ -376,8 +376,8 @@ start counter to derive the number of attempts currently in progress. The session phases run in order, so a later phase start also proves the earlier phases succeeded for that attempt. -Four start counters, four 13-series histograms, and fifteen terminal counters -create a hard ceiling of 71 raw Prometheus series per pod. Connection ordinals, +Three start counters, three 13-series histograms, and twelve terminal counters +create a hard ceiling of 54 raw Prometheus series per pod. Connection ordinals, database URLs, hosts, usernames, SQL, and raw errors are forbidden as metric labels. diff --git a/migrations/0087_retire_created_at_floor.sql b/migrations/0087_retire_created_at_floor.sql new file mode 100644 index 00000000000..fac3f74364d --- /dev/null +++ b/migrations/0087_retire_created_at_floor.sql @@ -0,0 +1,26 @@ +SET LOCAL lock_timeout = '5s'; + +-- DESTRUCTIVE CONTRACT: deploy phase 1's application floor check to every +-- writer pod/job first, with no rollback expected. Stop out-of-band writers +-- and hold the replica breaker closed through this drop until replay catches +-- up. Old pods that already started do not recheck the guard on each probe. +-- Keep the 960-second application floor and the ordered heartbeat handshake. +-- No other trigger, function, FK, or tenant fence is retired here. +DROP TRIGGER IF EXISTS events_created_at_floor ON events; + +-- Dropping the parent removes its inherited clones (including partitions +-- created since 0021). Also remove any independently attached child copy. +DO $$ +DECLARE + partition_table regclass; +BEGIN + FOR partition_table IN + SELECT relid FROM pg_partition_tree('events'::regclass) + WHERE relid <> 'events'::regclass + LOOP + EXECUTE format('DROP TRIGGER IF EXISTS events_created_at_floor ON %s', partition_table); + END LOOP; +END +$$; + +DROP FUNCTION IF EXISTS events_created_at_floor_guard(); diff --git a/schema/functions/events_created_at_floor_guard.sql b/schema/functions/events_created_at_floor_guard.sql deleted file mode 100644 index 263886f83f9..00000000000 --- a/schema/functions/events_created_at_floor_guard.sql +++ /dev/null @@ -1,29 +0,0 @@ --- Replica-fence floor guard (keep in sync with migrations/0021). A deferred --- constraint trigger re-checks, inside COMMIT processing, that channel-bearing --- event rows are no older than `buzz.created_at_floor` seconds before commit --- time (clock_timestamp(), NOT the transaction-frozen now()). This turns the --- relay's ingest-time created_at envelope into a commit-time storage --- invariant, which is what lets keyset-cursor pages below the replica fence --- be served by a read replica without holes. Enforcement is armed per session --- via the GUC (set by the relay's writer pool on connect); sessions without --- the GUC (pg_restore, manual backfills) bypass it and must hold the replica --- fence closed for their duration. The only structural exemption is --- channel_id IS NULL: those rows never appear in keyset-paged windows. -CREATE FUNCTION events_created_at_floor_guard() RETURNS trigger -LANGUAGE plpgsql AS $$ -DECLARE - floor_secs numeric := nullif(current_setting('buzz.created_at_floor', true), '')::numeric; -BEGIN - IF floor_secs IS NOT NULL - AND floor_secs > 0 - AND NEW.channel_id IS NOT NULL - AND NEW.created_at < clock_timestamp() - make_interval(secs => floor_secs) - THEN - RAISE EXCEPTION - 'events.created_at % is more than % s before commit time %; below the replica-fence floor', - NEW.created_at, floor_secs, clock_timestamp() - USING ERRCODE = 'check_violation'; - END IF; - RETURN NULL; -END -$$; diff --git a/schema/schema.sql b/schema/schema.sql index f804fb0bc4a..70ae2264304 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -65,7 +65,6 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; -- functions \i functions/channels_community_id_immutable.sql \i functions/guard_channel_roster_snapshot.sql -\i functions/events_created_at_floor_guard.sql \i functions/prevent_community_deletion_request_retargeting.sql \i functions/prevent_community_deletion_approval_removal.sql \i functions/protect_community_deletion_manifest_keys.sql diff --git a/schema/triggers/events.sql b/schema/triggers/events.sql index 64d9bbdd5d3..7de86addbd2 100644 --- a/schema/triggers/events.sql +++ b/schema/triggers/events.sql @@ -2,15 +2,4 @@ CREATE TRIGGER trg_events_guard_channel_roster_snapshot BEFORE INSERT ON events FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot(); --- INSERT OR UPDATE OF: an UPDATE can move a previously exempt row into the --- guarded set (channel_id NULL -> NOT NULL) or move a channel row's --- created_at below the fence, so both mutation paths re-run the guard on the --- NEW row. A created_at rewrite that crosses partition bounds runs as --- DELETE + INSERT and hits the cloned AFTER INSERT guard on the destination --- partition; an in-partition rewrite fires the UPDATE OF arm. -CREATE CONSTRAINT TRIGGER events_created_at_floor - AFTER INSERT OR UPDATE OF created_at, channel_id ON events - DEFERRABLE INITIALLY DEFERRED - FOR EACH ROW - EXECUTE FUNCTION events_created_at_floor_guard(); SELECT attach_community_write_fence('events'); diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 0c84398aa3d..ee6b0125460 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -16,7 +16,6 @@ BEGIN -- pgschema may copy parent triggers onto standalone children. Drop -- those copies before ATTACH; PostgreSQL recreates inherited parent -- triggers while attaching and rejects same-named child triggers. - DROP TRIGGER IF EXISTS events_created_at_floor ON events_p_past; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p_past; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p_past; ALTER TABLE events ATTACH PARTITION events_p_past @@ -28,7 +27,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_01'::regclass ) THEN - DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_01; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_01; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_01; ALTER TABLE events ATTACH PARTITION events_p2026_01 @@ -40,7 +38,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_02'::regclass ) THEN - DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_02; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_02; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_02; ALTER TABLE events ATTACH PARTITION events_p2026_02 @@ -52,7 +49,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_03'::regclass ) THEN - DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_03; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_03; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_03; ALTER TABLE events ATTACH PARTITION events_p2026_03 @@ -64,7 +60,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_04'::regclass ) THEN - DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_04; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_04; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_04; ALTER TABLE events ATTACH PARTITION events_p2026_04 @@ -76,7 +71,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_05'::regclass ) THEN - DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_05; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_05; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_05; ALTER TABLE events ATTACH PARTITION events_p2026_05 @@ -88,7 +82,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p2026_06'::regclass ) THEN - DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_06; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p2026_06; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p2026_06; ALTER TABLE events ATTACH PARTITION events_p2026_06 @@ -100,7 +93,6 @@ BEGIN WHERE inhparent = 'events'::regclass AND inhrelid = 'events_p_future'::regclass ) THEN - DROP TRIGGER IF EXISTS events_created_at_floor ON events_p_future; DROP TRIGGER IF EXISTS community_write_fence_events ON events_p_future; DROP TRIGGER IF EXISTS trg_events_guard_channel_roster_snapshot ON events_p_future; ALTER TABLE events ATTACH PARTITION events_p_future From ce659adccb5417fe783e9b28863013a13565fbb1 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 14:23:09 -0400 Subject: [PATCH 02/12] buzz-db: trim retirement tests and fix stale floor prose Address review of the phase-2 floor retirement: - the workspace-wide floor source policy now comes from phase 1's shared crates/*/src walk (which pins buzz-relay coverage); this commit only drops the GUC from its constant-parity message - drop the legacy-0021 half of the writer/probe test, which could not fail, and rename the desired-state test to what it checks - fix stale GUC prose and document the ACCESS EXCLUSIVE lock in 0062 Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/runtime/replica_fence.rs | 6 ++-- .../src/runtime/tests/floor_retirement.rs | 28 +++---------------- crates/buzz-db/tests/observability_source.rs | 2 +- migrations/0087_retire_created_at_floor.sql | 3 ++ 4 files changed, 11 insertions(+), 28 deletions(-) diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 51f6e8d4345..806cf063f7a 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -56,9 +56,9 @@ //! keeps failing, the newest entry ages past [`FENCE_STALENESS`] and the //! fence closes. //! -//! Raw SQL without the armed GUC and `session_replication_role = replica` -//! restores bypass the supported app writers. These operational bypasses are -//! outside the proof and require holding the fence closed; see `migrations/0021`. +//! Raw SQL and `session_replication_role = replica` restores bypass the +//! application floor check. These operational bypasses are outside the proof +//! and require holding the fence closed. use std::collections::VecDeque; use std::sync::{Arc, Mutex}; diff --git a/crates/buzz-db/src/runtime/tests/floor_retirement.rs b/crates/buzz-db/src/runtime/tests/floor_retirement.rs index cf938cb7d10..d82b7eb19c7 100644 --- a/crates/buzz-db/src/runtime/tests/floor_retirement.rs +++ b/crates/buzz-db/src/runtime/tests/floor_retirement.rs @@ -96,28 +96,20 @@ async fn migration_schema_floor_retirement_cleans_child_copies_idempotently() { #[tokio::test] #[ignore = "requires Postgres"] -async fn floor_retirement_desired_and_migrated_catalogs_match() { +async fn floor_retirement_desired_state_has_no_floor_objects() { // The nextest wrapper bootstraps this database through real pgschema apply - // and reconciliation. Check the retired floor catalog against the real - // migration boot path; unrelated trigger/FK preservation is checked above. + // and reconciliation; desired-state/migration parity is checked by + // `admin_schema_parity_between_desired_state_and_migrations`. let pool = PgPool::connect(&crate::test_support::database_url()) .await .expect("connect desired-state database"); assert_eq!(floor_objects(&pool).await, (0, 0)); - sqlx::raw_sql("DROP SCHEMA public CASCADE; CREATE SCHEMA public;") - .execute(&pool) - .await - .expect("reset isolated database for migration boot"); - crate::migration::run_migrations(&pool) - .await - .expect("boot migration schema"); - assert_eq!(floor_objects(&pool).await, (0, 0)); pool.close().await; } #[tokio::test] #[ignore = "requires Postgres"] -async fn floor_retirement_writer_and_probe_accept_legacy_and_retired_schema() { +async fn floor_retirement_writer_and_probe_run_on_retired_schema() { let url = crate::test_support::database_url(); let db = Db::new(&DbConfig { database_url: url.clone(), @@ -140,18 +132,6 @@ async fn floor_retirement_writer_and_probe_accept_legacy_and_retired_schema() { .await .expect("retired schema still supports the ordered heartbeat"); - // This binary can also run before migration, without changing legacy SQL. - sqlx::raw_sql(include_str!( - "../../../../../migrations/0021_created_at_fence_floor.sql" - )) - .execute(&db.pool) - .await - .expect("restore actual legacy guard"); - assert!(floor_objects(&db.pool).await.0 > 1); - assert!(db.spawn_fence_probe()); - crate::replica_fence::probe_once(&db.pool, db.fence()) - .await - .expect("legacy schema still supports ordered heartbeat"); db.pool.close().await; db.read_pool .as_ref() diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 286f52fa151..c535f54bef1 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -2457,7 +2457,7 @@ fn event_floor_guard_uses_the_database_statement_clock_and_shared_floor() { ); assert!( guard.contains(".bind(CREATED_AT_FLOOR_SECS as f64)"), - "app check, GUC and fence wall must share the same floor constant" + "app check and fence wall must share the same floor constant" ); } diff --git a/migrations/0087_retire_created_at_floor.sql b/migrations/0087_retire_created_at_floor.sql index fac3f74364d..057171e9752 100644 --- a/migrations/0087_retire_created_at_floor.sql +++ b/migrations/0087_retire_created_at_floor.sql @@ -5,6 +5,9 @@ SET LOCAL lock_timeout = '5s'; -- and hold the replica breaker closed through this drop until replay catches -- up. Old pods that already started do not recheck the guard on each probe. -- Keep the 960-second application floor and the ordered heartbeat handshake. +-- DROP TRIGGER takes ACCESS EXCLUSIVE on events and every partition, so it +-- can stall events reads/writes for up to lock_timeout. Run off-peak; a lock +-- timeout aborts the whole migration and is safe to retry. -- No other trigger, function, FK, or tenant fence is retired here. DROP TRIGGER IF EXISTS events_created_at_floor ON events; From 945f73aecb31738eb28e3a3827e59e6c93ea5109 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 15:33:42 -0400 Subject: [PATCH 03/12] buzz-db: catch SET( and ON CONFLICT DO UPDATE in the floor policy With the floor trigger retired, the source policy is the only guard against rewriting created_at/channel_id on existing event rows. Close two matcher gaps found in review: - `UPDATE events SET(created_at, ...)` with no space before the parenthesis slipped past the " set " split - `INSERT INTO events ... ON CONFLICT ... DO UPDATE SET channel_id = ...` names no table, so it passed even though it can re-home an existing row below the floor; bind each DO UPDATE to the INSERT it belongs to and treat it as an events update when that INSERT targets events Share the SET-list check between both paths, and pin each gap with a negative fixture plus controls: a non-floor-column events upsert, DO NOTHING, and a guarded events insert next to a watermark upsert that sets created_at (the shape of the replaceable-event writer). Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/tests/observability_source.rs | 96 +++++++++++++++----- 1 file changed, 73 insertions(+), 23 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index c535f54bef1..353bd0c03c9 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -2251,9 +2251,36 @@ fn is_channel_null_event_writer(function: &str) -> bool { false } +/// Whether a `SET` list assigns `created_at` or `channel_id`, including the +/// row form `(a, b) = (...)`. Stops at the first `WHERE`. +fn assigns_event_floor_column(assignments: &str) -> bool { + let assignments = assignments.split(" where ").next().expect("first part"); + let mut depth = 0usize; + assignments + .split(|ch| { + match ch { + '(' => depth += 1, + ')' => depth = depth.saturating_sub(1), + _ => {} + } + ch == ',' && depth == 0 + }) + .any(|assignment| { + assignment.split_once('=').is_some_and(|(columns, _)| { + columns + .trim() + .trim_start_matches('(') + .trim_end_matches(')') + .split(',') + .any(|column| matches!(column.trim(), "created_at" | "channel_id")) + }) + }) +} + fn updates_event_floor_columns(function: &str) -> bool { - let normalized = normalized_sql(function); - normalized.match_indices("update ").any(|(index, marker)| { + // `SET(` with no space is still a row assignment. + let normalized = normalized_sql(function).replace(" set(", " set ("); + let direct_update = normalized.match_indices("update ").any(|(index, marker)| { let Some((target, assignments)) = normalized[index + marker.len()..].split_once(" set ") else { return false; @@ -2271,28 +2298,24 @@ fn updates_event_floor_columns(function: &str) -> bool { if !matches!(alias.as_slice(), [] | [_] | ["as", _]) { return false; } - let assignments = assignments.split(" where ").next().expect("first part"); - let mut depth = 0usize; - assignments - .split(|ch| { - match ch { - '(' => depth += 1, - ')' => depth = depth.saturating_sub(1), - _ => {} - } - ch == ',' && depth == 0 - }) - .any(|assignment| { - assignment.split_once('=').is_some_and(|(columns, _)| { - columns - .trim() - .trim_start_matches('(') - .trim_end_matches(')') - .split(',') - .any(|column| matches!(column.trim(), "created_at" | "channel_id")) + assigns_event_floor_column(assignments) + }); + // `ON CONFLICT ... DO UPDATE SET` names no table: it updates the target of + // the INSERT it belongs to, so an events upsert can re-home an existing row. + let events_upsert = normalized + .match_indices("do update set ") + .any(|(index, marker)| { + normalized[..index] + .rfind("insert into ") + .is_some_and(|insert| { + let table = normalized[insert + "insert into ".len()..] + .split(|ch: char| !(ch.is_ascii_alphanumeric() || ch == '_' || ch == '.')) + .next(); + matches!(table, Some("events" | "public.events")) }) - }) - }) + && assigns_event_floor_column(&normalized[index + marker.len()..]) + }); + direct_update || events_upsert } fn event_floor_policy_violations(production: &str) -> Vec { @@ -2366,6 +2389,30 @@ async fn quoted_schema_update(tx: &mut AdmittedTx) { async fn delete_only(tx: &mut AdmittedTx) { sqlx::query("UPDATE events SET deleted_at=now() WHERE channel_id=$1 AND created_at<$2"); } +async fn row_time_no_space(tx: &mut AdmittedTx) { + sqlx::query("UPDATE events SET(created_at,kind)=($1,1) WHERE id=$2"); +} +async fn guarded_upsert_rehome(tx: &mut AdmittedTx) { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; + sqlx::query("INSERT INTO events (id, channel_id) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET channel_id = EXCLUDED.channel_id"); +} +async fn guarded_upsert_row_time(tx: &mut AdmittedTx) { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; + sqlx::query("INSERT INTO events (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET(kind, created_at) = (1, $2)"); +} +async fn guarded_upsert_other_column(tx: &mut AdmittedTx) { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; + sqlx::query("INSERT INTO events (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET tags = EXCLUDED.tags"); +} +async fn guarded_upsert_do_nothing(tx: &mut AdmittedTx) { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; + sqlx::query("INSERT INTO events (id) VALUES ($1) ON CONFLICT (id) DO NOTHING"); +} +async fn guarded_insert_then_other_table_upsert(tx: &mut AdmittedTx) { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; + sqlx::query("INSERT INTO events (id) VALUES ($1) ON CONFLICT DO NOTHING"); + sqlx::query("INSERT INTO parameterized_event_watermarks (d_tag, created_at) VALUES ($1, $2) ON CONFLICT (d_tag) DO UPDATE SET created_at = EXCLUDED.created_at"); +} "#; assert_eq!( event_floor_policy_violations(source), @@ -2383,6 +2430,9 @@ async fn delete_only(tx: &mut AdmittedTx) { "async fn quoted_schema_insert(tx: &mut AdmittedTx) {", "async fn quoted_update(tx: &mut AdmittedTx) {", "async fn quoted_schema_update(tx: &mut AdmittedTx) {", + "async fn row_time_no_space(tx: &mut AdmittedTx) {", + "async fn guarded_upsert_rehome(tx: &mut AdmittedTx) {", + "async fn guarded_upsert_row_time(tx: &mut AdmittedTx) {", ] ); } From 7ceb57b251a29eae7e34cf3d1680268bacb8e845 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 10:04:31 -0400 Subject: [PATCH 04/12] buzz-db: fail the floor policy closed on SQL it cannot attribute With the trigger gone the source policy is the only guard. Expose SQL the per-function scan could not see: include_str! is expanded from the file it loads and concat! literals are joined, so both reach the existing insert/update checks. Fail closed on what stays opaque: a non-literal include path, a concat! that spells SQL around a non-literal argument, event-write SQL in a const, static or macro body outside any function, and PREPARE TRANSACTION in any production writer. Cut each function slice to its own item so trailing consts are judged at item level. The replica fence no longer folds pg_prepared_xacts.prepared into the wall: only the retired trigger checked prepared rows at PREPARE, and the insert-time guard's clock is not exposed there. A prepared transaction now fails the sample closed (PreparedTransactions). Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/runtime/replica_fence.rs | 93 ++++- crates/buzz-db/tests/observability_source.rs | 390 ++++++++++++++++++- 2 files changed, 462 insertions(+), 21 deletions(-) diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 806cf063f7a..8b859c938dd 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -23,7 +23,8 @@ //! (a) finished before the activity scan — its commit precedes `M`'s //! commit, so the replica session has replayed it; //! (b) open at the activity scan — represented by `xact_start`, so it is -//! bounded by the `oldest_xact_start` term; +//! bounded by the `oldest_xact_start` term (a prepared transaction has +//! no `xact_start` to read, so one existing fails the sample closed); //! (c) started after the activity scan — its insert-time app guard runs //! after `S`, so it cannot commit a row with //! `created_at < S - floor`. @@ -470,6 +471,17 @@ pub enum ProbeError { /// Number of other client backends with masked/unknown state. masked: i64, }, + /// A prepared (2PC) transaction exists. Its rows were floor-checked at + /// insert time, which `pg_prepared_xacts` cannot bound, and no supported + /// writer prepares transactions — treat it as out-of-band. + #[error( + "{prepared} prepared transaction(s) in pg_prepared_xacts — out-of-band writer; \ + replica floor cannot bound their rows" + )] + PreparedTransactions { + /// Number of prepared transactions in the cluster. + prepared: i64, + }, /// The single heartbeat row (migration 0026) is missing on the writer. #[error("replica_heartbeat row missing on the writer — migration 0026 not applied?")] HeartbeatRowMissing, @@ -662,18 +674,19 @@ async fn sample_writer( // // Prepared transactions (2PC) are a bucket of their own: while // prepared they have left `pg_stat_activity` but can still commit - // after the token. Their floor check must precede PREPARE, so - // `pg_prepared_xacts.prepared` bounds their rows exactly like - // `xact_start`; fold it into the same minimum. + // after the token. Only the retired deferred trigger checked their + // rows at PREPARE; the insert-time app guard runs earlier, at a clock + // `pg_prepared_xacts` does not expose, so `prepared` no longer bounds + // their rows. No supported writer issues PREPARE TRANSACTION (the + // source policy enforces it) and `max_prepared_transactions` defaults + // to 0, so any prepared transaction is out-of-band: fail closed. let row = sqlx::query( r#" SELECT - least( - (SELECT min(xact_start) - FROM pg_stat_activity - WHERE pid <> pg_backend_pid()), - (SELECT min(prepared) FROM pg_prepared_xacts) - ) AS oldest_xact_start, + (SELECT min(xact_start) + FROM pg_stat_activity + WHERE pid <> pg_backend_pid()) AS oldest_xact_start, + (SELECT count(*) FROM pg_prepared_xacts) AS prepared, (SELECT count(*) FROM pg_stat_activity WHERE pid <> pg_backend_pid() @@ -690,6 +703,10 @@ async fn sample_writer( if masked > 0 { return Err(ProbeError::MaskedActivity { masked }); } + let prepared: i64 = row.get("prepared"); + if prepared > 0 { + return Err(ProbeError::PreparedTransactions { prepared }); + } let oldest_xact_start: Option> = row.get("oldest_xact_start"); Ok((sampled_at, oldest_xact_start)) }) @@ -1465,6 +1482,62 @@ mod postgres_tests { unpriv.close().await; } + /// With the floor trigger retired, nothing bounds a prepared (2PC) + /// transaction's rows by `pg_prepared_xacts.prepared`, so the sample must + /// fail closed while one exists. On a server with the default + /// `max_prepared_transactions = 0` (CI, compose, chart) the test instead + /// pins that 2PC is unavailable, which is the supported-writer exclusion. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn cluster_global_sample_writer_fails_closed_on_prepared_transaction() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect"); + let max_prepared: String = sqlx::query_scalar("SHOW max_prepared_transactions") + .fetch_one(&pool) + .await + .expect("show max_prepared_transactions"); + + let gid = format!("fence_probe_{}", uuid::Uuid::new_v4().simple()); + let mut conn = pool.acquire().await.expect("acquire"); + sqlx::query("BEGIN") + .execute(&mut *conn) + .await + .expect("begin"); + sqlx::query("SELECT 1") + .execute(&mut *conn) + .await + .expect("touch tx"); + let prepare = sqlx::query(sqlx::AssertSqlSafe(format!("PREPARE TRANSACTION '{gid}'"))) + .execute(&mut *conn) + .await; + + if max_prepared == "0" { + prepare.expect_err("PREPARE TRANSACTION must fail when 2PC is disabled"); + sqlx::query("ROLLBACK") + .execute(&mut *conn) + .await + .expect("rollback"); + return; + } + + prepare.expect("prepare transaction"); + drop(conn); + let result = sample_writer(&pool).await; + sqlx::query(sqlx::AssertSqlSafe(format!("ROLLBACK PREPARED '{gid}'"))) + .execute(&pool) + .await + .expect("rollback prepared"); + let err = result.expect_err("a prepared transaction must fail the sample closed"); + assert!( + matches!(err, ProbeError::PreparedTransactions { prepared } if prepared >= 1), + "expected PreparedTransactions, got {err:?}" + ); + sample_writer(&pool) + .await + .expect("sample succeeds once no transaction is prepared"); + } + /// The Aurora PostgreSQL identity function name is exact — the /// MySQL-family near-miss (`aurora_server_id`) would make the /// capability probe cache a permanent false on real Aurora and diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 353bd0c03c9..31da972f573 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1438,21 +1438,60 @@ fn raw_literal_prefix_starts_token(bytes: &[u8], index: usize) -> bool { fn mask_comments_and_literals(source: &str) -> String { let bytes = source.as_bytes(); let mut masked = bytes.to_vec(); - let mut blank = |from: usize, to: usize| { - for byte in &mut masked[from..to.min(bytes.len())] { + for span in source_spans(source) { + for byte in &mut masked[span.start..span.end] { if *byte != b'\n' { *byte = b' '; } } - }; + } + String::from_utf8(masked).expect("masking replaces whole ASCII-delimited spans") +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum SpanKind { + Comment, + /// A plain (`"…"`) string literal; `start` is its opening quote. + Str, + /// A raw string literal; `start` is its `r`, followed by `hashes` `#`s. + RawStr { + hashes: usize, + }, + Char, +} + +struct SourceSpan { + start: usize, + end: usize, + kind: SpanKind, +} + +impl SourceSpan { + /// The text between a string literal's delimiters, escapes left as written. + fn string_body<'a>(&self, source: &'a str) -> Option<&'a str> { + match self.kind { + SpanKind::Str => source.get(self.start + 1..self.end.saturating_sub(1)), + SpanKind::RawStr { hashes } => { + source.get(self.start + 2 + hashes..self.end.saturating_sub(1 + hashes)) + } + SpanKind::Comment | SpanKind::Char => None, + } + } +} + +/// Every comment, string, raw string, and char literal in `source`, in order. +fn source_spans(source: &str) -> Vec { + let bytes = source.as_bytes(); + let mut spans = Vec::new(); let mut index = 0usize; while index < bytes.len() { let start = index; - match bytes[index] { + let kind = match bytes[index] { b'/' if bytes.get(index + 1) == Some(&b'/') => { while index < bytes.len() && bytes[index] != b'\n' { index += 1; } + SpanKind::Comment } b'/' if bytes.get(index + 1) == Some(&b'*') => { let mut depth = 0usize; @@ -1470,6 +1509,7 @@ fn mask_comments_and_literals(source: &str) -> String { index += 1; } } + SpanKind::Comment } b'r' if matches!(bytes.get(index + 1), Some(b'"' | b'#')) && raw_literal_prefix_starts_token(bytes, index) @@ -1491,6 +1531,7 @@ fn mask_comments_and_literals(source: &str) -> String { index = source[body..] .find(&close) .map_or(bytes.len(), |offset| body + offset + close.len()); + SpanKind::RawStr { hashes } } b'"' => { index += 1; @@ -1498,18 +1539,24 @@ fn mask_comments_and_literals(source: &str) -> String { index += if bytes[index] == b'\\' { 2 } else { 1 }; } index += 1; + SpanKind::Str } b'\'' if char_literal_end(source, index).is_some() => { index = char_literal_end(source, index).expect("guard matched"); + SpanKind::Char } _ => { index += 1; continue; } - } - blank(start, index); + }; + spans.push(SourceSpan { + start, + end: index.min(bytes.len()), + kind, + }); } - String::from_utf8(masked).expect("masking replaces whole ASCII-delimited spans") + spans } /// The end (exclusive) of the char literal whose opening quote is at `start`, @@ -2318,8 +2365,247 @@ fn updates_event_floor_columns(function: &str) -> bool { direct_update || events_upsert } -fn event_floor_policy_violations(production: &str) -> Vec { +/// The offset of the delimiter that closes the bracket opened at `open` in +/// `masked` (comments and literals already blanked). +fn matching_close(masked: &str, open: usize) -> Option { + let mut depth = 0usize; + for (index, byte) in masked.bytes().enumerate().skip(open) { + match byte { + b'(' | b'[' | b'{' => depth += 1, + b')' | b']' | b'}' => { + depth = depth.checked_sub(1)?; + if depth == 0 { + return Some(index); + } + } + _ => {} + } + } + None +} + +/// `text` as a plain Rust string literal. +fn plain_literal(text: &str) -> String { + format!("\"{}\"", text.replace('\\', "\\\\").replace('"', "\\\"")) +} + +/// The string literal that is the whole of `arg`, if it is one. +fn sole_string_literal(arg: &str) -> Option { + let trimmed = arg.trim(); + let offset = arg.len() - arg.trim_start().len(); + let mut spans = source_spans(trimmed).into_iter(); + let span = spans.next()?; + (spans.next().is_none() + && span.start == 0 + && span.end == trimmed.len() + && matches!(span.kind, SpanKind::Str | SpanKind::RawStr { .. })) + .then_some(SourceSpan { + start: span.start + offset, + end: span.end + offset, + kind: span.kind, + }) +} + +/// Whether literal text spells event-write SQL the floor policy must see. +fn spells_event_write_sql(text: &str) -> bool { + let normalized = normalized_sql_source(text); + normalized.contains("insert into") || normalized.contains("update ") +} + +/// Bring SQL that per-function literal scans cannot see into view, so the +/// floor policy fails closed on it now that no trigger backs it up: each +/// `include_str!("…")` becomes one literal holding the included file +/// (resolved against `path`), and the literal arguments of each `concat!` are +/// joined into one literal. Forms that stay opaque are violations: an +/// `include_str!` whose path is not one literal or cannot be read, and a +/// `concat!` that spells SQL around a non-literal argument. +fn sql_visible_source(path: &std::path::Path, source: &str) -> (String, Vec) { + let mut violations = Vec::new(); + let mut text = source.to_owned(); + for (macro_name, expand_include) in [("include_str!(", true), ("concat!(", false)] { + // Last occurrence first: a rewrite never moves an earlier offset, and + // an inner `concat!` is joined before the one around it. + loop { + let masked = mask_comments_and_literals(&text); + let Some(at) = masked + .match_indices(macro_name) + .map(|(at, _)| at) + .filter(|at| { + !masked[..*at] + .chars() + .next_back() + .is_some_and(|ch| ch.is_ascii_alphanumeric() || ch == '_') + }) + .last() + else { + break; + }; + let open = at + macro_name.len() - 1; + let Some(close) = matching_close(&masked, open) else { + violations.push(format!("{macro_name}…) is unterminated")); + break; + }; + let args = &text[open + 1..close]; + let replacement = if expand_include { + let included = sole_string_literal(args) + .and_then(|span| span.string_body(args).map(str::to_owned)) + .and_then(|relative| { + let resolved = path.parent()?.join(&relative); + std::fs::read_to_string(resolved).ok() + }); + match included { + Some(contents) => plain_literal(&contents), + None => { + violations.push(format!("include_str!({}) is opaque", args.trim())); + plain_literal("") + } + } + } else { + let masked_args = &masked[open + 1..close]; + let mut pieces = Vec::new(); + let mut joined = String::new(); + let mut opaque = false; + let mut depth = 0usize; + let mut arg_start = 0usize; + for (index, byte) in masked_args.bytes().chain(std::iter::once(b',')).enumerate() { + match byte { + b'(' | b'[' | b'{' => depth += 1, + b')' | b']' | b'}' => depth = depth.saturating_sub(1), + b',' if depth == 0 => { + let arg = &args[arg_start..index.min(args.len())]; + arg_start = index + 1; + if arg.trim().is_empty() { + continue; + } + match sole_string_literal(arg) { + Some(span) => { + let body = span.string_body(arg).unwrap_or_default(); + let body = if matches!(span.kind, SpanKind::Str) { + body.to_owned() + } else { + body.replace('\\', "\\\\").replace('"', "\\\"") + }; + joined.push_str(&body); + } + None => { + opaque = true; + if !joined.is_empty() { + pieces.push(format!("\"{joined}\"")); + } + joined.clear(); + pieces.push(arg.trim().to_owned()); + } + } + } + _ => {} + } + } + if !joined.is_empty() { + pieces.push(format!("\"{joined}\"")); + } + if opaque && spells_event_write_sql(args) { + violations.push(format!( + "concat!({}) spells SQL around a non-literal argument", + args.trim() + )); + } + if opaque { + // Keep the opaque form visible to the per-function scan but + // drop the macro name so this loop terminates. + format!("concat_opaque!({})", pieces.join(", ")) + } else { + pieces.pop().unwrap_or_else(|| plain_literal("")) + } + }; + text.replace_range(at..=close, &replacement); + } + } + (text, violations) +} + +/// Length of the function item that starts `masked_function` (comments and +/// literals blanked): through the `}` closing its body, or through the `;` of +/// a bodiless declaration. Unlike [`cfg_test_item_len`], a `;` inside the +/// signature's parentheses or brackets (`[u8; 4]`) does not end it. Without a +/// body the whole input counts, so an unrecognized shape never hides SQL. +fn function_item_len(masked_function: &str) -> usize { + let mut nesting = 0usize; + for (index, byte) in masked_function.bytes().enumerate() { + match byte { + b'(' | b'[' => nesting += 1, + b')' | b']' => nesting = nesting.saturating_sub(1), + b';' if nesting == 0 => return index + 1, + b'{' if nesting == 0 => { + return matching_close(masked_function, index) + .map_or(masked_function.len(), |close| close + 1); + } + _ => {} + } + } + masked_function.len() +} + +/// Each function slice cut to the function's own item, so a const or static +/// after its body is not attributed to it (see [`item_level_literals`]). +fn function_items(production: &str) -> Vec<&str> { function_slices(production) + .into_iter() + .map(|function| &function[..function_item_len(&mask_comments_and_literals(function))]) + .collect() +} + +/// String literal bodies outside every function body: consts, statics and +/// macro bodies. A function's floor guard cannot be attributed to SQL there. +fn item_level_literals(production: &str) -> String { + let base = production.as_ptr() as usize; + let functions: Vec<(usize, usize)> = function_items(production) + .into_iter() + .map(|function| { + let start = function.as_ptr() as usize - base; + (start, start + function.len()) + }) + .collect(); + let mut literals = String::new(); + for span in source_spans(production) { + if functions + .iter() + .any(|(start, end)| *start <= span.start && span.end <= *end) + { + continue; + } + if let Some(body) = span.string_body(production) { + literals.push_str(body); + literals.push('\n'); + } + } + literals +} + +/// Every floor-policy violation in one production file, after +/// [`sql_visible_source`] has exposed included and concatenated SQL. +fn file_event_floor_policy_violations(path: &std::path::Path, source: &str) -> Vec { + let (production, mut violations) = sql_visible_source(path, &strip_cfg_test_items(source)); + let item_level = item_level_literals(&production); + if inserts_events_row(&item_level) || updates_event_floor_columns(&item_level) { + violations.push("event-write SQL outside any function body".to_owned()); + } + // The fence cannot bound a prepared transaction's rows without the + // retired trigger (see `replica_fence::sample_writer`), so no supported + // writer may prepare one. + let literals: String = source_spans(&production) + .iter() + .filter_map(|span| span.string_body(&production)) + .collect::>() + .join("\n"); + if normalized_sql_source(&literals).contains("prepare transaction") { + violations.push("PREPARE TRANSACTION in a production writer".to_owned()); + } + violations.extend(event_floor_policy_violations(&production)); + violations +} + +fn event_floor_policy_violations(production: &str) -> Vec { + function_items(production) .into_iter() .filter(|function| { let header = function_header(function); @@ -2456,6 +2742,87 @@ pub(crate) async fn write_nip43_membership_snapshot(tx: &mut AdmittedTx) { ); } +#[test] +fn event_floor_policy_fails_closed_on_sql_it_cannot_attribute() { + let dir = std::env::temp_dir().join(format!("buzz-floor-policy-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).expect("fixture dir"); + std::fs::write( + dir.join("insert_event.sql"), + "INSERT INTO \"events\" (id) VALUES ($1)", + ) + .expect("sql fixture"); + std::fs::write(dir.join("notes.md"), "Prose that quotes nothing.").expect("md fixture"); + let path = dir.join("writer.rs"); + let source = r##" +const LEADING_INSERT: &str = "INSERT INTO events (id) VALUES ($1)"; +const USER_AGENT: &str = concat!("buzz-agent/", env!("CARGO_PKG_VERSION")); +static NOTES: &str = include_str!("notes.md"); +async fn joined_insert(tx: &mut AdmittedTx) { + sqlx::query(concat!("INSERT INTO ", "events (id) VALUES ($1)")); +} +async fn guarded_joined_insert(tx: &mut AdmittedTx) { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; + sqlx::query(concat!("INSERT INTO ", r#"events"#, " (id) VALUES ($1)")); +} +const TRAILING_REWRITE: &str = "UPDATE events SET created_at = $1 WHERE id = $2"; +async fn joined_rewrite(tx: &mut AdmittedTx) { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; + sqlx::query(concat!("UPDATE events ", "SET created_at = $1 WHERE id = $2")); +} +async fn included_insert(tx: &mut AdmittedTx) { + sqlx::query(include_str!("insert_event.sql")); +} +async fn guarded_included_insert(tx: &mut AdmittedTx) { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; + sqlx::query(include_str!("insert_event.sql")); +} +async fn opaque_include(tx: &mut AdmittedTx) { + sqlx::query(include_str!(concat!(env!("OUT_DIR"), "/insert.sql"))); +} +async fn opaque_concat(tx: &mut AdmittedTx) { + sqlx::query(concat!("INSERT INTO ", env!("EVENTS_TABLE"))); +} +async fn two_phase(tx: &mut AdmittedTx) { + sqlx::query("PREPARE TRANSACTION 'fence'"); +} +"##; + let violations = file_event_floor_policy_violations(&path, source); + std::fs::remove_dir_all(&dir).expect("remove fixture dir"); + assert_eq!( + violations, + [ + r#"include_str!(concat!(env!("OUT_DIR"), "/insert.sql")) is opaque"#, + r#"concat!("INSERT INTO ", env!("EVENTS_TABLE")) spells SQL around a non-literal argument"#, + "event-write SQL outside any function body", + "PREPARE TRANSACTION in a production writer", + "async fn joined_insert(tx: &mut AdmittedTx) {", + "async fn joined_rewrite(tx: &mut AdmittedTx) {", + "async fn included_insert(tx: &mut AdmittedTx) {", + ] + ); + + // Each item-level form is caught on its own, wherever it sits. + for item in [ + "const LEADING: &str = \"INSERT INTO events (id) VALUES ($1)\";\nfn f() {}\n", + "fn f() {}\nconst TRAILING: &str = \"UPDATE events SET channel_id = $1\";\nfn g() {}\n", + "fn f() {}\nstatic INCLUDED: &str = include_str!(\"insert_event.sql\");\n", + ] { + std::fs::create_dir_all(&dir).expect("fixture dir"); + std::fs::write( + dir.join("insert_event.sql"), + "insert into events (id) values ($1)", + ) + .expect("sql fixture"); + let found = file_event_floor_policy_violations(&path, item); + std::fs::remove_dir_all(&dir).expect("remove fixture dir"); + assert_eq!( + found, + ["event-write SQL outside any function body"], + "item-level SQL in `{item}`" + ); + } +} + #[test] fn every_production_channel_event_insert_checks_the_floor_and_never_rewrites_it() { let files = production_event_writer_source_files(); @@ -2469,8 +2836,9 @@ fn every_production_channel_event_insert_checks_the_floor_and_never_rewrites_it( let mut guarded_writers = 0; let mut violations = Vec::new(); for path in files { - let production = strip_cfg_test_items(&std::fs::read_to_string(&path).expect("source")); - guarded_writers += function_slices(&production) + let source = std::fs::read_to_string(&path).expect("source"); + let (production, _) = sql_visible_source(&path, &strip_cfg_test_items(&source)); + guarded_writers += function_items(&production) .into_iter() .filter(|function| { inserts_events_row(function) @@ -2479,7 +2847,7 @@ fn every_production_channel_event_insert_checks_the_floor_and_never_rewrites_it( }) .count(); violations.extend( - event_floor_policy_violations(&production) + file_event_floor_policy_violations(&path, &source) .into_iter() .map(|header| format!("{}: {header}", path.display())), ); From c0a3baa100061cf260f8f7174047ef0ad3b11dfd Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 10:26:30 -0400 Subject: [PATCH 05/12] buzz-db: check floor SQL per literal, with placeholders for what it cannot read Review of 8752ce124 found the opaque-concat rule red on production: the two retention deletes concat a predicate macro into their WHERE, and any "update " around a non-literal argument was a violation. Replace that heuristic with one detector. concat! now becomes a single literal with an {opaque} placeholder per non-literal argument, and every SQL literal is checked on its own: a placeholder (from concat! or format!) in a write target or SET column, a floor-column SET list whose UPDATE is in another literal, an INSERT INTO with no table after it or an events INSERT with a placeholder tail, events written by MERGE INTO or COPY, and a dangling PREPARE are violations. A placeholder in a WHERE or a value is allowed, so the retention deletes and the tenant-severing UPDATE {table} SET community_id = NULL pass. Also recognize concat!/include_str! with [] and {} delimiters, and ban sqlx query_file*! macros, which load SQL the scan cannot see. Document that the prepared-transaction count is cluster-wide on purpose, like the activity scan. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/runtime/replica_fence.rs | 5 +- crates/buzz-db/tests/observability_source.rs | 381 +++++++++++++------ 2 files changed, 276 insertions(+), 110 deletions(-) diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 8b859c938dd..7ee8149ce7d 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -679,7 +679,10 @@ async fn sample_writer( // `pg_prepared_xacts` does not expose, so `prepared` no longer bounds // their rows. No supported writer issues PREPARE TRANSACTION (the // source policy enforces it) and `max_prepared_transactions` defaults - // to 0, so any prepared transaction is out-of-band: fail closed. + // to 0, so any prepared transaction is out-of-band: fail closed. The + // count is cluster-wide, like the activity scan: a prepared + // transaction in any database on the writer is outside the supported + // deploy, so it closes the fence rather than being scoped away. let row = sqlx::query( r#" SELECT diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 31da972f573..c52e416c10a 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -2299,7 +2299,8 @@ fn is_channel_null_event_writer(function: &str) -> bool { } /// Whether a `SET` list assigns `created_at` or `channel_id`, including the -/// row form `(a, b) = (...)`. Stops at the first `WHERE`. +/// row form `(a, b) = (...)`, or a column it cannot read (a `{…}` +/// placeholder). Stops at the first `WHERE`. fn assigns_event_floor_column(assignments: &str) -> bool { let assignments = assignments.split(" where ").next().expect("first part"); let mut depth = 0usize; @@ -2319,36 +2320,62 @@ fn assigns_event_floor_column(assignments: &str) -> bool { .trim_start_matches('(') .trim_end_matches(')') .split(',') - .any(|column| matches!(column.trim(), "created_at" | "channel_id")) + .map(str::trim) + .any(|column| { + matches!(column, "created_at" | "channel_id") || column.starts_with('{') + }) }) }) } -fn updates_event_floor_columns(function: &str) -> bool { +/// The table a write targets, without its schema (a literal or a `{…}` +/// placeholder; any schema may hold an `events` table). +fn unqualified_table(target: &str) -> &str { + target.rsplit_once('.').map_or(target, |(_, table)| table) +} + +/// Whether one SQL literal (normalized) assigns a floor column through a +/// direct `UPDATE` of `events`, or of a target it does not spell: a +/// placeholder, or a `SET` list whose `UPDATE` lies in another literal. +fn literal_updates_event_floor_columns(sql: &str) -> bool { // `SET(` with no space is still a row assignment. - let normalized = normalized_sql(function).replace(" set(", " set ("); - let direct_update = normalized.match_indices("update ").any(|(index, marker)| { - let Some((target, assignments)) = normalized[index + marker.len()..].split_once(" set ") - else { + let sql = format!(" {} ", sql.trim()).replace(" set(", " set ("); + sql.match_indices(" set ").any(|(index, marker)| { + if !assigns_event_floor_column(&sql[index + marker.len()..]) { return false; + } + let before = &sql[..index]; + if before.ends_with(" do update") { + // The upsert check resolves the INSERT, which may be in another + // literal of the same builder. + return false; + } + let Some(update) = before.rfind(" update ") else { + return true; }; - let mut target = target.split_whitespace(); + let mut target = before[update + " update ".len()..].split_whitespace(); let mut table = target.next(); if table == Some("only") { table = target.next(); } - if !matches!(table, Some("events" | "public.events")) { - return false; - } // The target may have an alias, with or without AS. let alias = target.collect::>(); - if !matches!(alias.as_slice(), [] | [_] | ["as", _]) { - return false; - } - assigns_event_floor_column(assignments) - }); + let (Some(table), [] | [_] | ["as", _]) = (table, alias.as_slice()) else { + return true; + }; + let table = unqualified_table(table); + table == "events" || table.contains('{') + }) +} + +fn updates_event_floor_columns(function: &str) -> bool { + let direct_update = source_spans(function) + .iter() + .filter_map(|span| span.string_body(function)) + .any(|body| literal_updates_event_floor_columns(&normalized_sql(body))); // `ON CONFLICT ... DO UPDATE SET` names no table: it updates the target of // the INSERT it belongs to, so an events upsert can re-home an existing row. + let normalized = normalized_sql(function).replace(" set(", " set ("); let events_upsert = normalized .match_indices("do update set ") .any(|(index, marker)| { @@ -2365,6 +2392,43 @@ fn updates_event_floor_columns(function: &str) -> bool { direct_update || events_upsert } +/// Writes one SQL literal (normalized) hides from the per-function checks: +/// an INSERT whose target or tail is not literal, `events` written by a +/// statement other than INSERT, or a PREPARE. +fn literal_sql_violations(sql: &str) -> Vec<&'static str> { + let sql = format!(" {} ", sql.trim()); + let mut violations = Vec::new(); + for (index, _) in sql.match_indices(" into ") { + let rest = &sql[index + " into ".len()..]; + let target = rest.split([' ', '(']).next().expect("first part"); + let table = unqualified_table(target); + if sql[..index].ends_with(" insert") { + if target.is_empty() || table.contains('{') { + violations.push("INSERT target is not a literal table"); + } else if table == "events" && rest.contains('{') { + violations.push("events INSERT with a non-literal part"); + } + } else if table == "events" { + violations.push("events written by a statement other than INSERT"); + } + } + if sql.match_indices(" copy ").any(|(index, marker)| { + let target = sql[index + marker.len()..].split([' ', '(']).next(); + target.is_some_and(|target| unqualified_table(target) == "events") + }) { + violations.push("events written by a statement other than INSERT"); + } + let words = sql.split_whitespace().collect::>(); + if words + .windows(2) + .any(|pair| pair[0] == "prepare" && (pair[1] == "transaction" || pair[1].starts_with('{'))) + || words.last() == Some(&"prepare") + { + violations.push("PREPARE TRANSACTION in a production writer"); + } + violations +} + /// The offset of the delimiter that closes the bracket opened at `open` in /// `masked` (comments and literals already blanked). fn matching_close(masked: &str, open: usize) -> Option { @@ -2386,7 +2450,22 @@ fn matching_close(masked: &str, open: usize) -> Option { /// `text` as a plain Rust string literal. fn plain_literal(text: &str) -> String { - format!("\"{}\"", text.replace('\\', "\\\\").replace('"', "\\\"")) + format!("\"{}\"", plain_body(text)) +} + +/// Raw `text` escaped as the body of a plain string literal. +fn plain_body(text: &str) -> String { + text.replace('\\', "\\\\").replace('"', "\\\"") +} + +/// A string literal's body, as written in a plain (escaped) literal. +fn plain_string_body(span: &SourceSpan, source: &str) -> Option { + let body = span.string_body(source)?; + Some(if matches!(span.kind, SpanKind::Str) { + body.to_owned() + } else { + plain_body(body) + }) } /// The string literal that is the whole of `arg`, if it is one. @@ -2406,43 +2485,48 @@ fn sole_string_literal(arg: &str) -> Option { }) } -/// Whether literal text spells event-write SQL the floor policy must see. -fn spells_event_write_sql(text: &str) -> bool { - let normalized = normalized_sql_source(text); - normalized.contains("insert into") || normalized.contains("update ") +/// The opening delimiter of each `name!` macro call in `masked` (comments +/// and literals blanked), last first, with any path prefix (`::core::`) and +/// any of `(`, `[` or `{`: `(start of name, delimiter offset)`. +fn macro_calls(masked: &str, name: &str) -> Vec<(usize, usize)> { + let mut calls = masked + .match_indices(name) + .filter_map(|(at, _)| { + let starts_token = !masked[..at] + .chars() + .next_back() + .is_some_and(|ch| ch.is_ascii_alphanumeric() || ch == '_'); + let after = masked[at + name.len()..].strip_prefix('!')?; + let open = at + name.len() + 1 + (after.len() - after.trim_start().len()); + (starts_token && matches!(masked.as_bytes().get(open), Some(b'(' | b'[' | b'{'))) + .then_some((at, open)) + }) + .collect::>(); + calls.reverse(); + calls } /// Bring SQL that per-function literal scans cannot see into view, so the /// floor policy fails closed on it now that no trigger backs it up: each /// `include_str!("…")` becomes one literal holding the included file -/// (resolved against `path`), and the literal arguments of each `concat!` are -/// joined into one literal. Forms that stay opaque are violations: an -/// `include_str!` whose path is not one literal or cannot be read, and a -/// `concat!` that spells SQL around a non-literal argument. +/// (resolved against `path`), and each `concat!` becomes one literal of its +/// arguments, with `{opaque}` standing in for any non-literal argument. The +/// SQL checks treat a placeholder in a write target or a `SET` column as +/// unattributable. An `include_str!` whose path is not one literal or +/// cannot be read is a violation, as is any `query_file` macro. fn sql_visible_source(path: &std::path::Path, source: &str) -> (String, Vec) { let mut violations = Vec::new(); let mut text = source.to_owned(); - for (macro_name, expand_include) in [("include_str!(", true), ("concat!(", false)] { + for (macro_name, expand_include) in [("include_str", true), ("concat", false)] { // Last occurrence first: a rewrite never moves an earlier offset, and // an inner `concat!` is joined before the one around it. loop { let masked = mask_comments_and_literals(&text); - let Some(at) = masked - .match_indices(macro_name) - .map(|(at, _)| at) - .filter(|at| { - !masked[..*at] - .chars() - .next_back() - .is_some_and(|ch| ch.is_ascii_alphanumeric() || ch == '_') - }) - .last() - else { + let Some(&(at, open)) = macro_calls(&masked, macro_name).first() else { break; }; - let open = at + macro_name.len() - 1; let Some(close) = matching_close(&masked, open) else { - violations.push(format!("{macro_name}…) is unterminated")); + violations.push(format!("{macro_name}!…) is unterminated")); break; }; let args = &text[open + 1..close]; @@ -2462,9 +2546,7 @@ fn sql_visible_source(path: &std::path::Path, source: &str) -> (String, Vec (String, Vec { - let body = span.string_body(arg).unwrap_or_default(); - let body = if matches!(span.kind, SpanKind::Str) { - body.to_owned() - } else { - body.replace('\\', "\\\\").replace('"', "\\\"") - }; - joined.push_str(&body); - } - None => { - opaque = true; - if !joined.is_empty() { - pieces.push(format!("\"{joined}\"")); - } - joined.clear(); - pieces.push(arg.trim().to_owned()); - } + match sole_string_literal(arg) + .and_then(|span| plain_string_body(&span, arg)) + { + Some(body) => joined.push_str(&body), + None => joined.push_str("{opaque}"), } } _ => {} } } - if !joined.is_empty() { - pieces.push(format!("\"{joined}\"")); - } - if opaque && spells_event_write_sql(args) { - violations.push(format!( - "concat!({}) spells SQL around a non-literal argument", - args.trim() - )); - } - if opaque { - // Keep the opaque form visible to the per-function scan but - // drop the macro name so this loop terminates. - format!("concat_opaque!({})", pieces.join(", ")) - } else { - pieces.pop().unwrap_or_else(|| plain_literal("")) - } + format!("\"{joined}\"") }; text.replace_range(at..=close, &replacement); } } + // sqlx's `query_file*!` macros load SQL from a file at compile time; + // nothing uses them, so ban them rather than resolve them. + let masked = mask_comments_and_literals(&text); + for (at, _) in masked.match_indices("query_file") { + let starts_token = !masked[..at] + .chars() + .next_back() + .is_some_and(|ch| ch.is_ascii_alphanumeric() || ch == '_'); + let name_len = masked[at..] + .find(|ch: char| !(ch.is_ascii_alphanumeric() || ch == '_')) + .unwrap_or(masked.len() - at); + if starts_token && masked[at + name_len..].trim_start().starts_with('!') { + violations.push(format!( + "{}! loads SQL from a file", + &masked[at..at + name_len] + )); + } + } (text, violations) } @@ -2554,8 +2626,9 @@ fn function_items(production: &str) -> Vec<&str> { .collect() } -/// String literal bodies outside every function body: consts, statics and -/// macro bodies. A function's floor guard cannot be attributed to SQL there. +/// String literals outside every function body (consts, statics and macro +/// bodies), as plain literals one per line. A function's floor guard cannot +/// be attributed to SQL there. fn item_level_literals(production: &str) -> String { let base = production.as_ptr() as usize; let functions: Vec<(usize, usize)> = function_items(production) @@ -2573,9 +2646,8 @@ fn item_level_literals(production: &str) -> String { { continue; } - if let Some(body) = span.string_body(production) { - literals.push_str(body); - literals.push('\n'); + if let Some(body) = plain_string_body(&span, production) { + literals.push_str(&format!("\"{body}\"\n")); } } literals @@ -2585,20 +2657,21 @@ fn item_level_literals(production: &str) -> String { /// [`sql_visible_source`] has exposed included and concatenated SQL. fn file_event_floor_policy_violations(path: &std::path::Path, source: &str) -> Vec { let (production, mut violations) = sql_visible_source(path, &strip_cfg_test_items(source)); - let item_level = item_level_literals(&production); - if inserts_events_row(&item_level) || updates_event_floor_columns(&item_level) { - violations.push("event-write SQL outside any function body".to_owned()); - } // The fence cannot bound a prepared transaction's rows without the // retired trigger (see `replica_fence::sample_writer`), so no supported // writer may prepare one. - let literals: String = source_spans(&production) - .iter() - .filter_map(|span| span.string_body(&production)) - .collect::>() - .join("\n"); - if normalized_sql_source(&literals).contains("prepare transaction") { - violations.push("PREPARE TRANSACTION in a production writer".to_owned()); + for span in source_spans(&production) { + if let Some(body) = span.string_body(&production) { + for violation in literal_sql_violations(&normalized_sql(body)) { + if !violations.iter().any(|seen| seen == violation) { + violations.push(violation.to_owned()); + } + } + } + } + let item_level = item_level_literals(&production); + if inserts_events_row(&item_level) || updates_event_floor_columns(&item_level) { + violations.push("event-write SQL outside any function body".to_owned()); } violations.extend(event_floor_policy_violations(&production)); violations @@ -2787,37 +2860,127 @@ async fn two_phase(tx: &mut AdmittedTx) { } "##; let violations = file_event_floor_policy_violations(&path, source); + + // One writer per shape: the SQL is split, assembled, loaded or spelled so + // the per-literal checks cannot read it whole. + const WRITER: &str = "async fn writer(tx: &mut AdmittedTx) {"; + const OPAQUE_INSERT: &str = "INSERT target is not a literal table"; + let shapes: [(&str, &[&str]); 19] = [ + // Controls: an opaque WHERE predicate (the retention deletes), a + // placeholder table with no floor column (tenant severing), and a + // builder whose upsert continues in another literal. + ( + r#"sqlx::query(concat!("UPDATE events SET deleted_at = NOW() WHERE NOT ", pred!(), ""))"#, + &[], + ), + ( + r#"format!("UPDATE {table} SET community_id = NULL WHERE community_id = $1")"#, + &[], + ), + ( + r#"QueryBuilder::new("INSERT INTO listeners (a) ").push(" ON CONFLICT (a) DO UPDATE SET created_at = EXCLUDED.created_at")"#, + &[], + ), + // A non-literal `concat!` argument as the target or a SET column. + ( + r#"concat!("UPDATE ", table!(), " SET created_at = $1")"#, + &[WRITER], + ), + ( + r#"concat!("UPDATE events SET ", cols!(), " = $1")"#, + &[WRITER], + ), + // Every macro delimiter. + ( + r#"concat!["INSERT INTO ", "events (id) VALUES ($1)"]"#, + &[WRITER], + ), + ( + r#"::core::concat!{"INSERT INTO ", "events (id) VALUES ($1)"}"#, + &[WRITER], + ), + (r#"include_str!["insert_event.sql"]"#, &[WRITER]), + (r#"include_str! {"insert_event.sql"}"#, &[WRITER]), + // Compile-time file SQL is banned outright. + ( + r#"sqlx::query_file!("insert_event.sql")"#, + &["query_file! loads SQL from a file"], + ), + ( + r#"sqlx::query_file_as!(Row, "insert_event.sql")"#, + &["query_file_as! loads SQL from a file"], + ), + // SQL assembled at runtime. + ( + r#"format!("{}{}", "INSERT INTO ", "events (id) VALUES ($1)")"#, + &[OPAQUE_INSERT], + ), + ( + r#"format!("INSERT INTO {t} (id) VALUES ($1)")"#, + &[OPAQUE_INSERT], + ), + ( + r#"["INSERT INTO ", "events (id) VALUES ($1)"].concat()"#, + &[OPAQUE_INSERT], + ), + ( + r#"format!("INSERT INTO events (id) VALUES ($1) {suffix}")"#, + &["events INSERT with a non-literal part", WRITER], + ), + ( + r#"format!("{}{}", "UPDATE ", "events SET created_at = $1")"#, + &[WRITER], + ), + ( + r#"format!("UPDATE public.events SET {column} = $1")"#, + &[WRITER], + ), + ( + r#"sqlx::query("MERGE INTO events e USING src ON e.id = src.id WHEN MATCHED THEN DO NOTHING")"#, + &["events written by a statement other than INSERT"], + ), + ( + r#"format!("{} TRANSACTION 'x'", "PREPARE")"#, + &["PREPARE TRANSACTION in a production writer"], + ), + ]; + let shape_violations = shapes.map(|(shape, _)| { + file_event_floor_policy_violations(&path, &format!("{WRITER}\n {shape};\n}}\n")) + }); + + // Each item-level form is caught on its own, wherever it sits. + std::fs::write( + dir.join("insert_event.sql"), + "insert into events (id) values ($1)", + ) + .expect("sql fixture"); + let items = [ + "const LEADING: &str = \"INSERT INTO events (id) VALUES ($1)\";\nfn f() {}\n", + "fn f() {}\nconst TRAILING: &str = \"UPDATE events SET channel_id = $1\";\nfn g() {}\n", + "fn f() {}\nstatic INCLUDED: &str = include_str!(\"insert_event.sql\");\n", + ]; + let item_violations = items.map(|item| file_event_floor_policy_violations(&path, item)); std::fs::remove_dir_all(&dir).expect("remove fixture dir"); + assert_eq!( violations, [ r#"include_str!(concat!(env!("OUT_DIR"), "/insert.sql")) is opaque"#, - r#"concat!("INSERT INTO ", env!("EVENTS_TABLE")) spells SQL around a non-literal argument"#, - "event-write SQL outside any function body", + OPAQUE_INSERT, "PREPARE TRANSACTION in a production writer", + "event-write SQL outside any function body", "async fn joined_insert(tx: &mut AdmittedTx) {", "async fn joined_rewrite(tx: &mut AdmittedTx) {", "async fn included_insert(tx: &mut AdmittedTx) {", ] ); - - // Each item-level form is caught on its own, wherever it sits. - for item in [ - "const LEADING: &str = \"INSERT INTO events (id) VALUES ($1)\";\nfn f() {}\n", - "fn f() {}\nconst TRAILING: &str = \"UPDATE events SET channel_id = $1\";\nfn g() {}\n", - "fn f() {}\nstatic INCLUDED: &str = include_str!(\"insert_event.sql\");\n", - ] { - std::fs::create_dir_all(&dir).expect("fixture dir"); - std::fs::write( - dir.join("insert_event.sql"), - "insert into events (id) values ($1)", - ) - .expect("sql fixture"); - let found = file_event_floor_policy_violations(&path, item); - std::fs::remove_dir_all(&dir).expect("remove fixture dir"); + for ((shape, expected), found) in shapes.iter().zip(&shape_violations) { + assert_eq!(found, expected, "shape `{shape}`"); + } + for (item, found) in items.iter().zip(&item_violations) { assert_eq!( found, - ["event-write SQL outside any function body"], + &["event-write SQL outside any function body"], "item-level SQL in `{item}`" ); } From e2f763cb42a3886d5e28b9ec51da02b98e814c47 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 10:35:14 -0400 Subject: [PATCH 06/12] buzz-db: treat a SET list split across literals as a floor write The per-literal UPDATE check only saw the SET list inside the literal that holds SET, so format!("UPDATE events SET {}", ..), a split [..].concat(), and QueryBuilder pushes after SET slipped through. A SET list that is empty or holds an item with no '=' is now unattributable when its target is events, a placeholder, or DO UPDATE. Prose that merely contains update/set is still ignored. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/tests/observability_source.rs | 55 +++++++++++++++++--- 1 file changed, 49 insertions(+), 6 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index c52e416c10a..691c30199c7 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -2328,6 +2328,25 @@ fn assigns_event_floor_column(assignments: &str) -> bool { }) } +/// Whether a `SET` list (up to the first `WHERE`) is missing from its +/// literal: empty, or holding an item with no `=`, such as a `{…}` +/// placeholder or a column whose value follows in another literal. +fn set_list_is_unreadable(assignments: &str) -> bool { + let assignments = assignments.split(" where ").next().expect("first part"); + let mut depth = 0usize; + assignments.trim().is_empty() + || assignments + .split(|ch| { + match ch { + '(' => depth += 1, + ')' => depth = depth.saturating_sub(1), + _ => {} + } + ch == ',' && depth == 0 + }) + .any(|assignment| !assignment.contains('=')) +} + /// The table a write targets, without its schema (a literal or a `{…}` /// placeholder; any schema may hold an `events` table). fn unqualified_table(target: &str) -> &str { @@ -2337,21 +2356,26 @@ fn unqualified_table(target: &str) -> &str { /// Whether one SQL literal (normalized) assigns a floor column through a /// direct `UPDATE` of `events`, or of a target it does not spell: a /// placeholder, or a `SET` list whose `UPDATE` lies in another literal. +/// A `SET` list it cannot read counts as assigning one. fn literal_updates_event_floor_columns(sql: &str) -> bool { // `SET(` with no space is still a row assignment. let sql = format!(" {} ", sql.trim()).replace(" set(", " set ("); sql.match_indices(" set ").any(|(index, marker)| { - if !assigns_event_floor_column(&sql[index + marker.len()..]) { + let assignments = &sql[index + marker.len()..]; + // A SET list continued in another literal or argument cannot be read. + let unreadable = set_list_is_unreadable(assignments); + if !unreadable && !assigns_event_floor_column(assignments) { return false; } let before = &sql[..index]; if before.ends_with(" do update") { // The upsert check resolves the INSERT, which may be in another - // literal of the same builder. - return false; + // literal of the same builder, but only from a readable SET list. + return unreadable; } let Some(update) = before.rfind(" update ") else { - return true; + // A bare SET (Redis `SET`, prose) is not SQL unless it assigns. + return !unreadable; }; let mut target = before[update + " update ".len()..].split_whitespace(); let mut table = target.next(); @@ -2361,7 +2385,9 @@ fn literal_updates_event_floor_columns(sql: &str) -> bool { // The target may have an alias, with or without AS. let alias = target.collect::>(); let (Some(table), [] | [_] | ["as", _]) = (table, alias.as_slice()) else { - return true; + // An unreadable target is a violation only when it assigns a + // floor column; prose ("update … is set") assigns nothing. + return !unreadable; }; let table = unqualified_table(table); table == "events" || table.contains('{') @@ -2865,7 +2891,7 @@ async fn two_phase(tx: &mut AdmittedTx) { // the per-literal checks cannot read it whole. const WRITER: &str = "async fn writer(tx: &mut AdmittedTx) {"; const OPAQUE_INSERT: &str = "INSERT target is not a literal table"; - let shapes: [(&str, &[&str]); 19] = [ + let shapes: [(&str, &[&str]); 23] = [ // Controls: an opaque WHERE predicate (the retention deletes), a // placeholder table with no floor column (tenant severing), and a // builder whose upsert continues in another literal. @@ -2935,6 +2961,23 @@ async fn two_phase(tx: &mut AdmittedTx) { r#"format!("UPDATE public.events SET {column} = $1")"#, &[WRITER], ), + // A SET list continued in another literal or argument. + ( + r#"format!("UPDATE events SET {}", "created_at = $1")"#, + &[WRITER], + ), + ( + r#"["UPDATE events SET ", "channel_id = $1"].concat()"#, + &[WRITER], + ), + ( + r#"QueryBuilder::new("UPDATE events SET ").push("created_at = ")"#, + &[WRITER], + ), + ( + r#"QueryBuilder::new("INSERT INTO events (id) ").push(" ON CONFLICT (id) DO UPDATE SET ").push("channel_id = EXCLUDED.channel_id")"#, + &[WRITER], + ), ( r#"sqlx::query("MERGE INTO events e USING src ON e.id = src.id WHEN MATCHED THEN DO NOTHING")"#, &["events written by a statement other than INSERT"], From 17dae4e4928551521566aa56d17f8fde921678c3 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 10 Nov 2026 08:04:55 -0800 Subject: [PATCH 07/12] test(buzz-db): drop policy exemptions for the retired floor probes The startup floor/roster probes are deleted with the triggers, so their exemption lists and fixtures only kept dead names alive in the policy test. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/tests/observability_source.rs | 66 +++----------------- 1 file changed, 7 insertions(+), 59 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 691c30199c7..085ff3d8134 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -862,9 +862,6 @@ fn has_any_tenant_local_chokepoint(source: &str) -> bool { const ADMITTED_TX_SIGNATURE_MARKERS: [&str; 2] = ["&mut AdmittedTx", "&mut crate::AdmittedTx"]; -// Narrow reviewed exceptions for non-serving verification probes only. -const GUARDED_WRITE_FUNCTION_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("]; - const GUARDED_TX_ADAPTER_METHOD_PINS: [(&str, &str); 1] = [( "pub async fn replace_member_event(", "pub async fn lock_member_snapshot(", @@ -943,12 +940,6 @@ fn function_header(function_source: &str) -> &str { .trim() } -fn function_is_guarded_write_exception(function_header: &str) -> bool { - GUARDED_WRITE_FUNCTION_EXCEPTIONS - .iter() - .any(|exception| function_header.starts_with(exception)) -} - fn function_accepts_admitted_transaction(function_source: &str) -> bool { let signature = function_source.split('{').next().unwrap_or(function_source); ADMITTED_TX_SIGNATURE_MARKERS @@ -991,9 +982,7 @@ fn syntactic_guarded_write_route_violations(production_source: &str) -> Vec Vec { function_slices(production_source) .into_iter() .filter(|function_source| { - let header = function_header(function_source); - inserts_events_row(function_source) - && !EVENT_INSERT_FOLLOW_UP_EXCEPTIONS - .iter() - .any(|exception| header.starts_with(exception)) - && !runs_event_follow_ups(function_source) + inserts_events_row(function_source) && !runs_event_follow_ups(function_source) }) .map(|function_source| function_header(function_source).to_owned()) .collect() @@ -1743,12 +1723,6 @@ pub(crate) async fn mention_writer(tx: &mut AdmittedTx) { .await .expect("write"); } -pub async fn verify_channel_roster_fence_behavior(pool: &PgPool) { - sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") - .execute(pool) - .await - .expect("probe"); -} pub(crate) async fn lowercase_writer(tx: &mut AdmittedTx) { sqlx::query("insert into public.events (community_id, id) VALUES ($1, $2)") @@ -2010,17 +1984,14 @@ fn event_insert_hook_inventory_carries_the_insert_timestamp() { assert_eq!(actual, expected); } -// Fixed channel-less kinds cannot be channel rosters, and the replica-floor -// startup probe rolls back all its writes. +// Fixed channel-less kinds cannot be channel rosters. // Exemptions match by function header, so a kind 39002 insert added inside one // of these functions would pass the policy; keep them single-purpose. -const EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS: [&str; 3] = [ +const EVENT_INSERT_ROSTER_GUARD_EXCEPTIONS: [&str; 2] = [ // NIP-43 membership is kind 13534 with channel_id NULL. "pub(crate) async fn write_nip43_membership_snapshot(", // Push leases are kind 30350 with channel_id NULL. "pub async fn accept_lease_event(", - // Rolled-back replica-floor startup probe. - "pub async fn verify_floor_guard_behavior(", ]; fn event_insert_roster_guard_violations(production_source: &str) -> Vec { @@ -2072,9 +2043,6 @@ pub(crate) async fn write_nip43_membership_snapshot(tx: &mut AdmittedTx) { pub async fn accept_lease_event(tx: &mut AdmittedTx) { sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); } -pub async fn verify_floor_guard_behavior(pool: &PgPool) { - sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)"); -} #[cfg(test)] mod tests { fn fixture() { @@ -2125,9 +2093,6 @@ fn every_production_event_insert_runs_the_roster_guard() { ); } -// The rolled-back replica-floor probe writes a kind 9 row, which has no `d`. -const EVENT_INSERT_D_TAG_EXCEPTIONS: [&str; 1] = ["pub async fn verify_floor_guard_behavior("]; - /// Whether an `INSERT INTO events` tail names `d_tag` in an explicit column /// list. An insert without a column list does not count. fn events_insert_names_d_tag(tail: &str) -> bool { @@ -2141,13 +2106,9 @@ fn event_insert_d_tag_violations(production_source: &str) -> Vec { function_slices(production_source) .into_iter() .filter(|function_source| { - let header = function_header(function_source); - !EVENT_INSERT_D_TAG_EXCEPTIONS + !events_insert_tails(function_source) .iter() - .any(|exception| header.starts_with(exception)) - && !events_insert_tails(function_source) - .iter() - .all(|tail| events_insert_names_d_tag(tail)) + .all(|tail| events_insert_names_d_tag(tail)) }) .map(|function_source| function_header(function_source).to_owned()) .collect() @@ -2182,9 +2143,6 @@ pub(crate) async fn second_insert_without_d_tag(tx: &mut AdmittedTx) { pub(crate) async fn mention_writer(tx: &mut AdmittedTx) { sqlx::query("INSERT INTO event_mentions (community_id, event_id) VALUES ($1, $2)"); } -pub async fn verify_floor_guard_behavior(pool: &PgPool) { - sqlx::query("INSERT INTO events (community_id, id, kind) VALUES ($1, $2, 9)"); -} #[cfg(test)] mod tests { fn fixture() { @@ -2278,11 +2236,8 @@ fn every_search_hydrator_preserves_the_exact_hit_timestamp() { } } -// This startup probe always rolls back; it deliberately exercises the roster -// DB backstop rather than the app check. The two serving writers are exempt only -// while their INSERT remains structurally channel-NULL (checked below). -const EVENT_FLOOR_PROBE_EXCEPTIONS: [&str; 1] = - ["pub async fn verify_channel_roster_fence_behavior("]; +// The two serving channel-NULL writers are exempt only while their INSERT +// remains structurally channel-NULL (checked below). fn is_channel_null_event_writer(function: &str) -> bool { let header = function_header(function); @@ -2707,13 +2662,6 @@ fn event_floor_policy_violations(production: &str) -> Vec { function_items(production) .into_iter() .filter(|function| { - let header = function_header(function); - if EVENT_FLOOR_PROBE_EXCEPTIONS - .iter() - .any(|exception| header.starts_with(exception)) - { - return false; - } updates_event_floor_columns(function) || (inserts_events_row(function) && !is_channel_null_event_writer(function) From 68156c50901f3da3bf5b9ca62bce19659d461197 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 10 Nov 2026 08:12:25 -0800 Subject: [PATCH 08/12] buzz-db: fix the floor-retirement port against current main Main grew a nip_fi_issuers argument on community bootstrap, ProbeTimings on sample_writer, and a sync spawn_fence_probe; the rebased commits still used the old shapes and the new PreparedTransactions variant had no span kind. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/src/runtime/observability.rs | 11 +++++++---- crates/buzz-db/src/runtime/replica_fence.rs | 5 +++-- crates/buzz-db/src/runtime/tests.rs | 2 +- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/crates/buzz-db/src/runtime/observability.rs b/crates/buzz-db/src/runtime/observability.rs index ce160e54903..1a1b80f2ec4 100644 --- a/crates/buzz-db/src/runtime/observability.rs +++ b/crates/buzz-db/src/runtime/observability.rs @@ -1457,10 +1457,13 @@ mod tests { let query = crate::EventQuery::for_community(test_scope); writer_db - .ensure_configured_community_for_bootstrap(&format!( - "pool-operation-bootstrap-{}.invalid", - Uuid::new_v4().simple() - )) + .ensure_configured_community_for_bootstrap( + &format!( + "pool-operation-bootstrap-{}.invalid", + Uuid::new_v4().simple() + ), + &[], + ) .await .expect("real community bootstrap succeeds"); routed_db.read_pool_boot_ping_for_tests().await; diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 7ee8149ce7d..96eaaca586e 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -507,6 +507,7 @@ impl ProbeError { Self::Writer(sqlx::Error::Io(_)) => "io", Self::Writer(_) => "writer", Self::MaskedActivity { .. } => "masked_activity", + Self::PreparedTransactions { .. } => "prepared_transactions", Self::HeartbeatRowMissing => "heartbeat_row_missing", Self::Deadline(_) => "deadline", } @@ -1526,7 +1527,7 @@ mod postgres_tests { prepare.expect("prepare transaction"); drop(conn); - let result = sample_writer(&pool).await; + let result = sample_writer(&pool, &mut ProbeTimings::default()).await; sqlx::query(sqlx::AssertSqlSafe(format!("ROLLBACK PREPARED '{gid}'"))) .execute(&pool) .await @@ -1536,7 +1537,7 @@ mod postgres_tests { matches!(err, ProbeError::PreparedTransactions { prepared } if prepared >= 1), "expected PreparedTransactions, got {err:?}" ); - sample_writer(&pool) + sample_writer(&pool, &mut ProbeTimings::default()) .await .expect("sample succeeds once no transaction is prepared"); } diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 7af3b287b33..97f459240c7 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2614,7 +2614,7 @@ async fn fence_probe_beats_while_serving_writer_pool_is_exhausted() { .await .expect("connect armed Db with one serving writer connection"); assert!( - db.spawn_fence_probe().await.expect("verification passes"), + db.spawn_fence_probe(), "a configured reader must spawn the probe" ); From 637ceebe8962b436755e30e1957427b846fc5cc5 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 10 Nov 2026 08:16:43 -0800 Subject: [PATCH 09/12] test(buzz-db): end a SET list at RETURNING in the floor policy Main's put_members_on upsert returns columns after its SET list, which the policy read as an unreadable assignment and flagged. Also drop the writer count floor to seven now that the startup probe is gone. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/tests/observability_source.rs | 29 +++++++++++++++----- 1 file changed, 22 insertions(+), 7 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 085ff3d8134..36e21752dc3 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -2086,10 +2086,10 @@ fn every_production_event_insert_runs_the_roster_guard() { violations.is_empty(), "production `INSERT INTO events` writers must run the app-side roster guard: {violations:?}" ); - // Five guarded writers plus the three explicitly exempt writers/probes. + // Five guarded writers plus the two explicitly exempt writers. assert!( - checked_writers >= 8, - "expected at least eight event writers, found {checked_writers}" + checked_writers >= 7, + "expected at least seven event writers, found {checked_writers}" ); } @@ -2199,8 +2199,8 @@ fn every_production_event_insert_names_d_tag() { // The same writer set as the roster-guard scan. A lower count means the // scan stopped seeing a writer, not that one was removed safely. assert!( - checked_writers >= 8, - "expected at least eight event writers, found {checked_writers}" + checked_writers >= 7, + "expected at least seven event writers, found {checked_writers}" ); } @@ -2256,8 +2256,15 @@ fn is_channel_null_event_writer(function: &str) -> bool { /// Whether a `SET` list assigns `created_at` or `channel_id`, including the /// row form `(a, b) = (...)`, or a column it cannot read (a `{…}` /// placeholder). Stops at the first `WHERE`. -fn assigns_event_floor_column(assignments: &str) -> bool { +/// The assignments of a `SET` list: it ends at `WHERE` or, for an upsert +/// with no `WHERE`, at `RETURNING`. +fn set_list(assignments: &str) -> &str { let assignments = assignments.split(" where ").next().expect("first part"); + assignments.split(" returning ").next().expect("first part") +} + +fn assigns_event_floor_column(assignments: &str) -> bool { + let assignments = set_list(assignments); let mut depth = 0usize; assignments .split(|ch| { @@ -2287,7 +2294,7 @@ fn assigns_event_floor_column(assignments: &str) -> bool { /// literal: empty, or holding an item with no `=`, such as a `{…}` /// placeholder or a column whose value follows in another literal. fn set_list_is_unreadable(assignments: &str) -> bool { - let assignments = assignments.split(" where ").next().expect("first part"); + let assignments = set_list(assignments); let mut depth = 0usize; assignments.trim().is_empty() || assignments @@ -2733,6 +2740,10 @@ async fn guarded_upsert_row_time(tx: &mut AdmittedTx) { crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; sqlx::query("INSERT INTO events (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET(kind, created_at) = (1, $2)"); } +async fn guarded_upsert_rehome_returning(tx: &mut AdmittedTx) { + crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; + sqlx::query("INSERT INTO events (id, channel_id) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET channel_id = EXCLUDED.channel_id RETURNING id, kind"); +} async fn guarded_upsert_other_column(tx: &mut AdmittedTx) { crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?; sqlx::query("INSERT INTO events (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET tags = EXCLUDED.tags"); @@ -2746,6 +2757,9 @@ async fn guarded_insert_then_other_table_upsert(tx: &mut AdmittedTx) { sqlx::query("INSERT INTO events (id) VALUES ($1) ON CONFLICT DO NOTHING"); sqlx::query("INSERT INTO parameterized_event_watermarks (d_tag, created_at) VALUES ($1, $2) ON CONFLICT (d_tag) DO UPDATE SET created_at = EXCLUDED.created_at"); } +async fn other_table_upsert_returning(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO channel_members (pubkey, role) VALUES ($1, $2) ON CONFLICT (pubkey) DO UPDATE SET removed_at = NULL, role = EXCLUDED.role RETURNING pubkey, role, removed_at"); +} "#; assert_eq!( event_floor_policy_violations(source), @@ -2766,6 +2780,7 @@ async fn guarded_insert_then_other_table_upsert(tx: &mut AdmittedTx) { "async fn row_time_no_space(tx: &mut AdmittedTx) {", "async fn guarded_upsert_rehome(tx: &mut AdmittedTx) {", "async fn guarded_upsert_row_time(tx: &mut AdmittedTx) {", + "async fn guarded_upsert_rehome_returning(tx: &mut AdmittedTx) {", ] ); } From 138da61aa7f35e36bf7e533ab65d2178a694cb16 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 10 Nov 2026 08:37:41 -0800 Subject: [PATCH 10/12] test(buzz-db): stop the held-writer media prebuild check at 0086 The prebuilt case holds ROW EXCLUSIVE on events while it migrates. The trigger retirements after 0086 need ACCESS EXCLUSIVE on events, so they hit lock_timeout under that lock. Migrate through 0086 under the writer, then apply the rest once it rolls back. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/src/runtime/migration.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 81ad9af8c89..743dda04b25 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -4124,10 +4124,15 @@ mod postgres_tests { .await .unwrap(); if prebuilt { - run_migrations(&pool) + // Stop at 0086: later trigger retirements take ACCESS + // EXCLUSIVE on events, which this held writer lock blocks. + run_migrations_through(&pool, 86) .await .expect("verified prebuild does not request an index-build lock"); writer.rollback().await.unwrap(); + run_migrations(&pool) + .await + .expect("later migrations apply once the writer is gone"); } else { let error = run_migrations(&pool).await.unwrap_err(); assert!( From 42d73193ff8d42bbb9e2d62c925bc23587e928e7 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 10 Nov 2026 10:16:11 -0800 Subject: [PATCH 11/12] test(buzz-db): snapshot the floor retirement baseline at 0086 The pre-contract step still stopped at 0060 from the original numbering, so the retained-guard snapshot was taken before main's 0061-0086 and the comparison after the full migrator saw their trigger and FK changes. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/src/runtime/tests/floor_retirement.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/runtime/tests/floor_retirement.rs b/crates/buzz-db/src/runtime/tests/floor_retirement.rs index d82b7eb19c7..0f14a8f8eb5 100644 --- a/crates/buzz-db/src/runtime/tests/floor_retirement.rs +++ b/crates/buzz-db/src/runtime/tests/floor_retirement.rs @@ -37,7 +37,7 @@ async fn migration_schema_floor_retirement_drops_parent_and_partition_clones() { let pool = PgPool::connect(&crate::test_support::database_url()) .await .expect("connect isolated migration database"); - crate::migration::run_migrations_through(&pool, 60) + crate::migration::run_migrations_through(&pool, 86) .await .expect("apply pre-contract schema"); let partitions: i64 = @@ -61,7 +61,7 @@ async fn migration_schema_floor_retirement_cleans_child_copies_idempotently() { let pool = PgPool::connect(&crate::test_support::database_url()) .await .expect("connect isolated migration database"); - crate::migration::run_migrations_through(&pool, 60) + crate::migration::run_migrations_through(&pool, 86) .await .expect("apply pre-contract schema"); // Model independently attached copies after the parent guard was removed. From 6ef31ef11372d1f93d3223173fc06bee78c4d892 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 10 Nov 2026 10:18:04 -0800 Subject: [PATCH 12/12] test(buzz-db): measure 0087 alone in the floor retirement tests A 0086 baseline compared against the full migrator still fails once 0088 and 0089 sit on top, because they drop more triggers. Stop at 0087 for the retained-guard comparison, then run the rest and recheck that the floor objects stay gone. Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Luke Tornquist --- crates/buzz-db/src/runtime/tests/floor_retirement.rs | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/crates/buzz-db/src/runtime/tests/floor_retirement.rs b/crates/buzz-db/src/runtime/tests/floor_retirement.rs index 0f14a8f8eb5..df30d44b270 100644 --- a/crates/buzz-db/src/runtime/tests/floor_retirement.rs +++ b/crates/buzz-db/src/runtime/tests/floor_retirement.rs @@ -47,11 +47,15 @@ async fn migration_schema_floor_retirement_drops_parent_and_partition_clones() { .expect("count parent and partitions"); assert_eq!(floor_objects(&pool).await, (partitions, 1)); let before = retained_guards(&pool).await; - crate::migration::run_migrations(&pool) + crate::migration::run_migrations_through(&pool, 87) .await - .expect("apply contract migration and retained startup checks"); + .expect("apply contract migration"); assert_eq!(floor_objects(&pool).await, (0, 0)); assert_eq!(retained_guards(&pool).await, before); + crate::migration::run_migrations(&pool) + .await + .expect("apply later migrations and retained startup checks"); + assert_eq!(floor_objects(&pool).await, (0, 0)); pool.close().await; } @@ -77,7 +81,7 @@ async fn migration_schema_floor_retirement_cleans_child_copies_idempotently() { .expect("install independent child copy"); assert_eq!(floor_objects(&pool).await, (1, 1)); let before = retained_guards(&pool).await; - crate::migration::run_migrations(&pool) + crate::migration::run_migrations_through(&pool, 87) .await .expect("retire independent copy"); assert_eq!(floor_objects(&pool).await, (0, 0));