diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index e0875618dd..101542f0ea 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -503,24 +503,29 @@ shape is an explicit transition, not a flag day: transitions take the matching exclusive lock. A supported serving transaction never spans communities; unmigrated paths still rely on trigger/function backstops. -- **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, not part of the lock foundation. +- **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 0062 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. Those backstops remain in place until their own separately gated + retirement. - **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. **Does NOT:** cache queries, implement connection pooling logic (delegated to sqlx), or make network calls outside Postgres. diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 2f306c35d0..3d091bc4e8 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -45,15 +45,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 a1ab04022f..710d3e04a5 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 b6ead54e06..a0759d99c7 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -50,14 +50,6 @@ pub(crate) async fn run_migrations_through(pool: &PgPool, target: i64) -> Result async fn run_migrations_locked(conn: &mut PgConnection) -> Result<()> { reject_legacy_nip_rs_cardinality_ambiguity(conn).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?; crate::channel::verify_channel_roster_fence_catalog(&mut *conn).await?; Ok(()) } @@ -705,10 +697,11 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 60); + assert_eq!(migrations.len(), 61); assert_eq!(migrations[58].version, 59); assert!(migrations[58].sql.as_str().contains("buzz.push_enabled")); assert_eq!(migrations[59].version, 60); + assert_eq!(migrations[60].version, 62); assert!(migrations[59] .sql .as_str() @@ -1044,12 +1037,6 @@ mod postgres_tests { .as_str() .contains("CREATE TABLE join_policy_acceptances")); - // Replica-fence commit-time floor guard on channel-bearing events. - assert_eq!(migrations[20].version, 21); - assert!(migrations[20] - .sql - .as_str() - .contains("events_created_at_floor_guard")); assert!(!migrations[0] .sql .as_str() diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 266003f5a0..0e7178ae87 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -356,8 +356,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: @@ -415,18 +415,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 @@ -708,10 +708,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 @@ -742,11 +738,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`]. pub async fn connect_writer_pool(config: &DbConfig) -> Result { use connection_observability::{ @@ -771,22 +766,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, @@ -893,9 +872,8 @@ impl Db { /// are dialed only on first acquire; the ~10-minute reaper never tops /// the pool back up, which is fine — routed reads re-fill it on demand. /// - /// 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. fn connect_read_pool(config: &DbConfig, url: &str, max_connections: u32) -> Result { Ok(PgPoolOptions::new() .max_connections(max_connections) @@ -1015,43 +993,19 @@ 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?; tokio::spawn(replica_fence::run_probe( self.pool.clone(), std::sync::Arc::clone(&self.fence), )); - 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, @@ -1613,3 +1567,7 @@ mod pool_role_tests { #[cfg(test)] #[path = "tests.rs"] mod postgres_tests; + +#[cfg(test)] +#[path = "tests/floor_retirement.rs"] +mod floor_retirement_postgres_tests; diff --git a/crates/buzz-db/src/runtime/observability.rs b/crates/buzz-db/src/runtime/observability.rs index d487b9a362..d6fdd3a408 100644 --- a/crates/buzz-db/src/runtime/observability.rs +++ b/crates/buzz-db/src/runtime/observability.rs @@ -1478,11 +1478,14 @@ mod tests { let test_scope = CommunityId::from_uuid(Uuid::new_v4()); let query = crate::EventQuery::for_community(test_scope); - routed_db.read_pool_boot_ping_for_tests().await; - routed_db - .verify_replica_fence_at_boot() + writer_db + .ensure_configured_community_for_bootstrap(&format!( + "pool-operation-bootstrap-{}.invalid", + Uuid::new_v4().simple() + )) .await - .expect("real startup fence verification succeeds"); + .expect("real community bootstrap succeeds"); + routed_db.read_pool_boot_ping_for_tests().await; let _ = crate::replica_fence::probe_once(&writer_pool, routed_db.fence()).await; routed_db.fence().force_open_for_tests(Utc::now()); assert_eq!( diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index b07a233990..f1fb75bf75 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -5,10 +5,9 @@ //! //! 1. **Insert-time floor**: supported channel-event writers check `created_at` //! against database `clock_timestamp()` on their admitted transaction for -//! each new row. Duplicate INSERTs need no floor check. This is the primary -//! guard. Migration 0021's deferred trigger and the writer-pool -//! `buzz.created_at_floor` GUC remain the mixed-version backstop until phase 2; -//! they also reject rows that age below the floor before COMMIT. +//! each new row. Duplicate INSERTs need no floor check. This is the only +//! guard: migration 0062 retired 0021's deferred trigger and the +//! writer-pool `buzz.created_at_floor` GUC. //! 2. **Ordered heartbeat handshake** (this module): on one pinned writer //! connection, one transaction first takes the exclusive replica-floor //! advisory lock (draining compliant shared-lock channel writers), then @@ -67,12 +66,12 @@ use uuid::Uuid; use buzz_datastore_tracing::datastore_span; -/// Seconds of `created_at` history the app and mixed-version trigger tolerate. +/// Seconds of `created_at` history the application floor check tolerates. /// /// Must exceed the relay's ingest envelope (±900 s) by enough slack that a -/// legitimately accepted event still commits within the floor even under -/// slow validation/lock waits. The writer pool arms the guard with this value -/// and the fence subtracts it; the two uses must never diverge. +/// legitimately accepted event still passes the check after slow +/// validation/lock waits. Application checks and the fence wall use this +/// same value; the two uses must never diverge. pub const CREATED_AT_FLOOR_SECS: i64 = 960; /// Enforce the channel-event floor on the admitted INSERT connection. @@ -110,13 +109,13 @@ pub(crate) async fn guard_event_created_at_floor( /// transactions against the writer probe handshake. /// /// Shared holders are channel-event write transactions validating `created_at` -/// under the armed floor contract; the probe takes the exclusive lock before -/// sampling and publishing its heartbeat token. +/// under the application floor contract; the probe takes the exclusive lock +/// before sampling and publishing its heartbeat token. pub const REPLICA_FLOOR_LOCK_KEY: i64 = 0x62757a7a666c6f72; /// Safety margin subtracted from the fence on top of the floor. /// -/// All proof timestamps (`clock_timestamp()`, `xact_start`, the guard's +/// All proof timestamps (`clock_timestamp()`, `xact_start`, the floor check's /// clock) come from the writer host, so this only needs to absorb /// `created_at` second-truncation and scheduling noise, not clock skew /// between machines. @@ -350,208 +349,6 @@ impl ReplicaFence { } } -/// Catalog-level verification that the commit-time floor guard (migration -/// 0021) is present and correctly shaped on the `events` parent AND every -/// partition: right function, `DEFERRABLE INITIALLY DEFERRED`, row-level, -/// AFTER, firing on both INSERT and UPDATE (an UPDATE can move an exempt -/// channel-NULL row into the guarded set, or rewrite `created_at` downward). -/// -/// This is a name-and-shape check only; it cannot detect a sabotaged -/// function body. [`verify_floor_guard_behavior`] proves the semantics. -/// -/// Generic over the executor so the migration path can run it on the -/// lock-holding connection while the startup probe keeps using the pool. -#[datastore_span(name = "replica_fence_verify_catalog", system = "postgresql")] -pub async fn verify_floor_guard_catalog<'e>( - 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 { @@ -632,17 +429,14 @@ async fn sample_writer(writer: &PgPool) -> Result { // conservative direction (a long autovacuum merely holds the fence // back), and it keeps the classification simple. // Scope is every other backend regardless of role or application - // name: an admin psql transaction writes under the same trigger and - // must be representable in the oldest-xact term. + // name: every open transaction must be representable in the + // oldest-xact term. Out-of-band writes require a closed breaker. // // Prepared transactions (2PC) are a bucket of their own: while // prepared they have left `pg_stat_activity` but can still commit - // after the token. Their deferred floor guard already ran at PREPARE, - // so `pg_prepared_xacts.prepared` bounds their rows exactly like - // `xact_start`; fold it into the same minimum. Only the retained - // deferred trigger provides that PREPARE-time check: the insert-time - // app guard runs earlier, and no app writer issues PREPARE TRANSACTION. - // Dropping the trigger must keep this case covered. + // after the token. Their floor check must precede PREPARE, so + // `pg_prepared_xacts.prepared` bounds their rows exactly like + // `xact_start`; fold it into the same minimum. let row = sqlx::query( r#" SELECT diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index d3f429baee..9c66f7fe44 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -1933,7 +1933,7 @@ async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() { format!("{}/{}", &base[..idx], wname) }; - // `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 (min_connections=0, 150ms // acquire timeout). Reader is sized 1 so holding one connection // saturates it. @@ -2241,9 +2241,8 @@ async fn routed_reads_are_confined_to_the_requested_community() { } /// D4: a LAZY reader pool (connect_lazy, min_connections=0, 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() { @@ -2258,11 +2257,7 @@ async fn lazy_reader_pool_still_spawns_fence_probe() { let idx = base.rfind('/').expect("db url has a path segment"); format!("{}/{}", &base[..idx], wname) }; - // `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 `create_scratch_db`'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), @@ -2270,12 +2265,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"); drop_scratch_db(&admin, db.pool.clone(), &wname).await; @@ -2568,100 +2560,6 @@ async fn thread_full_replica_page_above_fence_is_reverified_on_writer() { drop_scratch_db(&admin, writer, &wname).await; } -/// 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 admin = PgPool::connect(&admin_url().await) - .await - .expect("connect admin"); - let (pool, name) = create_scratch_db(&admin, "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; - - drop_scratch_db(&admin, pool, &name).await; -} - #[test] fn writer_pool_safety_hook_is_single_and_composed() { let source = include_str!("mod.rs"); @@ -2675,13 +2573,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 @@ -2738,7 +2633,6 @@ async fn writer_pool_metrics_record_every_initial_connection_ready() { 2 ); for step in [ - DbConnectionStep::CreatedAtFloor, DbConnectionStep::SessionTimeouts, DbConnectionStep::Isolation, ] { @@ -3090,106 +2984,6 @@ async fn session_timeouts_install_through_db_new_and_bound_lock_waits() { drop_scratch_db(&admin, db.pool.clone(), &name).await; } -/// 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 admin = PgPool::connect(&admin_url().await) - .await - .expect("connect admin"); - let (seed_pool, name) = create_scratch_db(&admin, "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 base = admin_url().await; - let idx = base.rfind('/').expect("db url has a path segment"); - let scratch_url = format!("{}/{}", &base[..idx], name); - 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"); - - drop_scratch_db(&admin, seed_pool, &name).await; - // db pool still holds connections to the dropped DB; close it. - 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> { @@ -3197,7 +2991,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) @@ -3325,201 +3119,6 @@ async fn replica_floor_probe_waits_for_shared_writer_and_records_after_release() drop_scratch_db(&admin, seed_pool, &name).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 admin = PgPool::connect(&admin_url().await) - .await - .expect("connect admin"); - let (seed_pool, wname) = create_scratch_db(&admin, "fence_gate_w").await; - let (replica_pool, rname) = create_scratch_db(&admin, "fence_gate_r").await; - seed_pool.close().await; - replica_pool.close().await; - - let base = admin_url().await; - let idx = base.rfind('/').expect("db url has a path segment"); - let writer_url = format!("{}/{}", &base[..idx], wname); - let replica_url = format!("{}/{}", &base[..idx], rname); - - // 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" - ); - - db_healthy.pool.close().await; - if let Some(rp) = &db_healthy.read_pool { - rp.close().await; - } - db.pool.close().await; - if let Some(rp) = &db.read_pool { - rp.close().await; - } - let _ = sqlx::query(sqlx::AssertSqlSafe(format!( - "DROP DATABASE IF EXISTS {wname} WITH (FORCE)" - ))) - .execute(&admin) - .await; - let _ = sqlx::query(sqlx::AssertSqlSafe(format!( - "DROP DATABASE IF EXISTS {rname} WITH (FORCE)" - ))) - .execute(&admin) - .await; -} - -/// 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 admin = PgPool::connect(&admin_url().await) - .await - .expect("connect admin"); - let (pool, name) = create_scratch_db(&admin, "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}" - ); - - drop_scratch_db(&admin, pool, &name).await; -} - // ---- 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 0000000000..6fe582a23e --- /dev/null +++ b/crates/buzz-db/src/runtime/tests/floor_retirement.rs @@ -0,0 +1,161 @@ +//! Contract-migration coverage; application insert coverage belongs to phase 1. + +use super::*; + +async fn floor_objects(pool: &PgPool) -> (i64, i64) { + sqlx::query_as( + "SELECT (SELECT count(*) FROM pg_trigger WHERE tgname = 'events_created_at_floor'), \ + (SELECT count(*) FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace \ + WHERE n.nspname = 'public' AND p.proname = 'events_created_at_floor_guard')", + ) + .fetch_one(pool) + .await + .expect("read floor catalog") +} + +async fn retained_guards(pool: &PgPool) -> Vec<(String, String)> { + sqlx::query_as( + "SELECT c.relname, pg_get_triggerdef(t.oid) FROM pg_trigger t \ + JOIN pg_class c ON c.oid = t.tgrelid \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE n.nspname = 'public' AND t.tgname <> 'events_created_at_floor' \ + UNION ALL \ + SELECT c.relname, pg_get_constraintdef(k.oid) FROM pg_constraint k \ + JOIN pg_class c ON c.oid = k.conrelid \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE n.nspname = 'public' AND k.contype = 'f' \ + ORDER BY 1, 2", + ) + .fetch_all(pool) + .await + .expect("read retained triggers and foreign keys") +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_floor_retirement_drops_parent_and_partition_clones() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect isolated migration database"); + crate::migration::run_migrations_through(&pool, 60) + .await + .expect("apply pre-contract schema"); + let partitions: i64 = + sqlx::query_scalar("SELECT count(*) FROM pg_partition_tree('events'::regclass)") + .fetch_one(&pool) + .await + .expect("count parent and partitions"); + assert_eq!(floor_objects(&pool).await, (partitions, 1)); + let before = retained_guards(&pool).await; + crate::migration::run_migrations(&pool) + .await + .expect("apply contract migration and retained startup checks"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + assert_eq!(retained_guards(&pool).await, before); + pool.close().await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_floor_retirement_cleans_child_copies_idempotently() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect isolated migration database"); + crate::migration::run_migrations_through(&pool, 60) + .await + .expect("apply pre-contract schema"); + // Model independently attached copies after the parent guard was removed. + sqlx::raw_sql( + "DROP TRIGGER events_created_at_floor ON events; \ + CREATE CONSTRAINT TRIGGER events_created_at_floor \ + AFTER INSERT OR UPDATE OF created_at, channel_id ON events_p_past \ + DEFERRABLE INITIALLY DEFERRED FOR EACH ROW \ + EXECUTE FUNCTION events_created_at_floor_guard();", + ) + .execute(&pool) + .await + .expect("install independent child copy"); + assert_eq!(floor_objects(&pool).await, (1, 1)); + let before = retained_guards(&pool).await; + crate::migration::run_migrations(&pool) + .await + .expect("retire independent copy"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + let mut tx = pool.begin().await.expect("begin idempotence check"); + sqlx::raw_sql(include_str!( + "../../../../../migrations/0062_retire_created_at_floor.sql" + )) + .execute(&mut *tx) + .await + .expect("repeat contract SQL after retirement"); + tx.commit().await.expect("commit repeated retirement"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + assert_eq!(retained_guards(&pool).await, before); + pool.close().await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn floor_retirement_desired_and_migrated_catalogs_match() { + // The nextest wrapper bootstraps this database through real pgschema apply + // and reconciliation. Check the retired floor catalog against the real + // migration boot path; unrelated trigger/FK preservation is checked above. + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect desired-state database"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + sqlx::raw_sql("DROP SCHEMA public CASCADE; CREATE SCHEMA public;") + .execute(&pool) + .await + .expect("reset isolated database for migration boot"); + crate::migration::run_migrations(&pool) + .await + .expect("boot migration schema"); + assert_eq!(floor_objects(&pool).await, (0, 0)); + pool.close().await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn floor_retirement_writer_and_probe_accept_legacy_and_retired_schema() { + let url = crate::test_support::database_url(); + let db = Db::new(&DbConfig { + database_url: url.clone(), + read_database_url: Some(url), + max_connections: 2, + min_connections: 1, + ..DbConfig::default() + }) + .await + .expect("connect production writer with retired schema"); + let armed: Option = + sqlx::query_scalar("SELECT nullif(current_setting('buzz.created_at_floor', true), '')") + .fetch_one(&db.pool) + .await + .expect("inspect writer session setting"); + assert_eq!(armed, None, "writer must no longer arm the retired guard"); + assert_eq!(floor_objects(&db.pool).await, (0, 0)); + assert!(db.spawn_fence_probe()); + crate::replica_fence::probe_once(&db.pool, db.fence()) + .await + .expect("retired schema still supports the ordered heartbeat"); + + // This binary can also run before migration, without changing legacy SQL. + sqlx::raw_sql(include_str!( + "../../../../../migrations/0021_created_at_fence_floor.sql" + )) + .execute(&db.pool) + .await + .expect("restore actual legacy guard"); + assert!(floor_objects(&db.pool).await.0 > 1); + assert!(db.spawn_fence_probe()); + crate::replica_fence::probe_once(&db.pool, db.fence()) + .await + .expect("legacy schema still supports ordered heartbeat"); + db.pool.close().await; + db.read_pool + .as_ref() + .expect("reader configured") + .close() + .await; +} diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 1fe5a9b5df..341a9a8da8 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -2322,7 +2322,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 741e3b7d52..bf7451763f 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 60632e75df..1aab895598 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -444,22 +444,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")); assert!(replica_fence_production.contains("WriterOperation::Maintenance")); assert!(!replica_fence_production.contains("pool.begin().await")); assert!(!replica_fence_production.contains("writer.acquire().await")); @@ -915,8 +904,7 @@ 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; 3] = [ - "pub async fn verify_floor_guard_behavior(", +const GUARDED_WRITE_FUNCTION_EXCEPTIONS: [&str; 2] = [ "pub async fn verify_channel_roster_fence_behavior(", "pub async fn backfill_d_tags(&self) -> Result {", ]; @@ -1422,10 +1410,8 @@ fn serving_table_writes_expose_syntactic_chokepoint_or_guarded_tx_routes() { // Rolled-back verification probes write `events` rows that never commit, so // they owe no push job or TTL refresh. -const EVENT_INSERT_FOLLOW_UP_EXCEPTIONS: [&str; 2] = [ - "pub async fn verify_floor_guard_behavior(", - "pub async fn verify_channel_roster_fence_behavior(", -]; +const EVENT_INSERT_FOLLOW_UP_EXCEPTIONS: [&str; 1] = + ["pub async fn verify_channel_roster_fence_behavior("]; /// `source` with every `#[cfg(test)]` item removed, wherever it sits in the /// file. Each item runs to the `;` that ends it or to the brace that closes @@ -1743,7 +1729,7 @@ pub(crate) async fn mention_writer(tx: &mut AdmittedTx) { .await .expect("write"); } -pub async fn verify_floor_guard_behavior(pool: &PgPool) { +pub async fn verify_channel_roster_fence_behavior(pool: &PgPool) { sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") .execute(pool) .await @@ -1957,13 +1943,11 @@ fn every_production_event_insert_runs_the_follow_up_hook() { ); } -// These startup probes always roll back; they deliberately exercise the DB -// backstops rather than the app check. The two serving writers are exempt only +// This startup probe always rolls back; it deliberately exercises the roster +// DB backstop rather than the app check. The two serving writers are exempt only // while their INSERT remains structurally channel-NULL (checked below). -const EVENT_FLOOR_PROBE_EXCEPTIONS: [&str; 2] = [ - "pub async fn verify_floor_guard_behavior(", - "pub async fn verify_channel_roster_fence_behavior(", -]; +const EVENT_FLOOR_PROBE_EXCEPTIONS: [&str; 1] = + ["pub async fn verify_channel_roster_fence_behavior("]; fn is_channel_null_event_writer(function: &str) -> bool { let header = function_header(function); diff --git a/crates/buzz-relay/src/api/admin/mod.rs b/crates/buzz-relay/src/api/admin/mod.rs index b5cfa1ef49..c9e071b6b9 100644 --- a/crates/buzz-relay/src/api/admin/mod.rs +++ b/crates/buzz-relay/src/api/admin/mod.rs @@ -9029,18 +9029,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]; @@ -9144,43 +9132,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"); + .expect("install isolated insert failure"); + let fresh_state = state_from_pool(pool.clone()).await; - // 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; - - // 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 @@ -9193,14 +9160,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 02ef106cda..3f39f2e410 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 4ed127dfff..031ef8e8de 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -551,26 +551,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 3980b6b86a..ee248e6395 100644 --- a/crates/buzz-relay/src/startup_steps.rs +++ b/crates/buzz-relay/src/startup_steps.rs @@ -47,7 +47,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 5704a7622b..4cc1901ea4 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -354,7 +354,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 @@ -368,8 +368,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/0062_retire_created_at_floor.sql b/migrations/0062_retire_created_at_floor.sql new file mode 100644 index 0000000000..fac3f74364 --- /dev/null +++ b/migrations/0062_retire_created_at_floor.sql @@ -0,0 +1,26 @@ +SET LOCAL lock_timeout = '5s'; + +-- DESTRUCTIVE CONTRACT: deploy phase 1's application floor check to every +-- writer pod/job first, with no rollback expected. Stop out-of-band writers +-- and hold the replica breaker closed through this drop until replay catches +-- up. Old pods that already started do not recheck the guard on each probe. +-- Keep the 960-second application floor and the ordered heartbeat handshake. +-- No other trigger, function, FK, or tenant fence is retired here. +DROP TRIGGER IF EXISTS events_created_at_floor ON events; + +-- Dropping the parent removes its inherited clones (including partitions +-- created since 0021). Also remove any independently attached child copy. +DO $$ +DECLARE + partition_table regclass; +BEGIN + FOR partition_table IN + SELECT relid FROM pg_partition_tree('events'::regclass) + WHERE relid <> 'events'::regclass + LOOP + EXECUTE format('DROP TRIGGER IF EXISTS events_created_at_floor ON %s', partition_table); + END LOOP; +END +$$; + +DROP FUNCTION IF EXISTS events_created_at_floor_guard(); diff --git a/schema/functions/events_created_at_floor_guard.sql b/schema/functions/events_created_at_floor_guard.sql deleted file mode 100644 index 263886f83f..0000000000 --- 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 5ec8bd5829..30670abb82 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -67,7 +67,6 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i functions/enqueue_push_match_job.sql \i functions/refresh_channel_ttl_after_event_insert.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 5491d1631d..aea4facc0e 100644 --- a/schema/triggers/events.sql +++ b/schema/triggers/events.sql @@ -10,15 +10,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 7eef8c4bbe..131c1085bd 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -18,7 +18,6 @@ BEGIN -- triggers while attaching and rejects same-named child triggers. DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p_past; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p_past; - 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 @@ -32,7 +31,6 @@ BEGIN ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_01; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_01; - 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 @@ -46,7 +44,6 @@ BEGIN ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_02; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_02; - 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 @@ -60,7 +57,6 @@ BEGIN ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_03; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_03; - 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 @@ -74,7 +70,6 @@ BEGIN ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_04; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_04; - 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 @@ -88,7 +83,6 @@ BEGIN ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_05; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_05; - 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 @@ -102,7 +96,6 @@ BEGIN ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_06; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_06; - 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 @@ -116,7 +109,6 @@ BEGIN ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p_future; DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p_future; - 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