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..743dda04b25 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() @@ -4133,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!( 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 { @@ -673,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, @@ -698,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", } @@ -860,26 +670,27 @@ 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. 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. 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 - 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() @@ -896,6 +707,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)) }) @@ -1671,6 +1486,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, &mut ProbeTimings::default()).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, &mut ProbeTimings::default()) + .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/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 6fa4a720a76..97f459240c7 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"); } @@ -2622,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" ); @@ -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..df30d44b270 --- /dev/null +++ b/crates/buzz-db/src/runtime/tests/floor_retirement.rs @@ -0,0 +1,145 @@ +//! 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, 86) + .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_through(&pool, 87) + .await + .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; +} + +#[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, 86) + .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_through(&pool, 87) + .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_state_has_no_floor_objects() { + // The nextest wrapper bootstraps this database through real pgschema apply + // 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)); + pool.close().await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +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(), + 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"); + + 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..36e21752dc3 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. @@ -873,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(", @@ -954,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 @@ -1002,9 +982,7 @@ fn syntactic_guarded_write_route_violations(production_source: &str) -> Vec 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; @@ -1481,6 +1494,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) @@ -1502,6 +1516,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; @@ -1509,18 +1524,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`, @@ -1649,12 +1670,7 @@ fn event_insert_follow_up_violations(production_source: &str) -> 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() @@ -1707,12 +1723,6 @@ pub(crate) async fn mention_writer(tx: &mut AdmittedTx) { .await .expect("write"); } -pub async fn verify_floor_guard_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)") @@ -1974,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 { @@ -2036,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() { @@ -2082,16 +2086,13 @@ 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}" ); } -// 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 { @@ -2105,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() @@ -2146,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() { @@ -2205,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}" ); } @@ -2242,13 +2236,8 @@ 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 -// 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(", -]; +// 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); @@ -2264,61 +2253,422 @@ fn is_channel_null_event_writer(function: &str) -> bool { false } -fn updates_event_floor_columns(function: &str) -> bool { - let normalized = normalized_sql(function); - normalized.match_indices("update ").any(|(index, marker)| { - let Some((target, assignments)) = normalized[index + marker.len()..].split_once(" set ") - else { +/// 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`. +/// 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| { + 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(',') + .map(str::trim) + .any(|column| { + matches!(column, "created_at" | "channel_id") || column.starts_with('{') + }) + }) + }) +} + +/// 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 = set_list(assignments); + 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 { + 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. +/// 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)| { + 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, but only from a readable SET list. + return unreadable; + } + let Some(update) = before.rfind(" update ") else { + // A bare SET (Redis `SET`, prose) is not SQL unless it assigns. + return !unreadable; }; - 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; + let (Some(table), [] | [_] | ["as", _]) = (table, alias.as_slice()) else { + // 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('{') + }) +} + +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)| { + 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 +} + +/// 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"); } - 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), - _ => {} + } + 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 { + 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); } - 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")) - }) - }) + } + _ => {} + } + } + None +} + +/// `text` as a plain Rust string literal. +fn plain_literal(text: &str) -> String { + 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) }) } -fn event_floor_policy_violations(production: &str) -> Vec { +/// 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, + }) +} + +/// 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 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)] { + // 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, open)) = macro_calls(&masked, macro_name).first() else { + break; + }; + 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 joined = String::new(); + 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) + .and_then(|span| plain_string_body(&span, arg)) + { + Some(body) => joined.push_str(&body), + None => joined.push_str("{opaque}"), + } + } + _ => {} + } + } + 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) +} + +/// 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() - .filter(|function| { - let header = function_header(function); - if EVENT_FLOOR_PROBE_EXCEPTIONS - .iter() - .any(|exception| header.starts_with(exception)) - { - return false; + .map(|function| &function[..function_item_len(&mask_comments_and_literals(function))]) + .collect() +} + +/// 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) + .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) = plain_string_body(&span, production) { + literals.push_str(&format!("\"{body}\"\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)); + // 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. + 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 +} + +fn event_floor_policy_violations(production: &str) -> Vec { + function_items(production) + .into_iter() + .filter(|function| { updates_event_floor_columns(function) || (inserts_events_row(function) && !is_channel_null_event_writer(function) @@ -2379,6 +2729,37 @@ 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_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"); +} +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"); +} +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), @@ -2396,6 +2777,10 @@ 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) {", + "async fn guarded_upsert_rehome_returning(tx: &mut AdmittedTx) {", ] ); } @@ -2419,6 +2804,194 @@ 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); + + // 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]); 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. + ( + 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], + ), + // 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"], + ), + ( + 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"#, + 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) {", + ] + ); + 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"], + "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(); @@ -2432,8 +3005,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) @@ -2442,7 +3016,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())), ); @@ -2470,7 +3044,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/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..057171e9752 --- /dev/null +++ b/migrations/0087_retire_created_at_floor.sql @@ -0,0 +1,29 @@ +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. +-- 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; + +-- 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