From f2e6e08986516388c1235e877a2bbce7e353cf14 Mon Sep 17 00:00:00 2001 From: tornquist Date: Thu, 17 Sep 2026 09:49:04 +0000 Subject: [PATCH 01/11] feat(buzz-db): add community and replica fence writer lock foundations Signed-off-by: tornquist Co-authored-by: Codex --- ARCHITECTURE.md | 28 ++++ crates/buzz-db/src/runtime/mod.rs | 82 +++++++++ crates/buzz-db/src/runtime/replica_fence.rs | 44 +++-- crates/buzz-db/src/runtime/tests.rs | 154 +++++++++++++++++ crates/buzz-db/src/store/deletion.rs | 155 ++++++++++++++++++ .../2026-09-17-triggerless-write-contract.md | 53 ++++++ scripts/reconcile-schema-after-pgschema.sql | 10 ++ 7 files changed, 513 insertions(+), 13 deletions(-) create mode 100644 docs/plans/2026-09-17-triggerless-write-contract.md diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index bf435c270a3..664190f24a5 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -433,6 +433,34 @@ All database access. Uses `sqlx::query()` (runtime, not compile-time macros) — - Approval tokens: `create_approval` receives the raw token and hashes it internally with SHA-256. - DDL injection protection in partition manager: allowlist of table names + strict suffix/date validators. +### Supported Writer Contract (transition state) + +The repository is moving toward a **supported-writer contract** where relay-owned +write APIs provide the only supported mutation path for tenant data. The current +shape is an explicit transition, not a flag day: + +- **Trajectory:** converge from trigger/function/FK-enforced fencing toward + application-owned transaction protocols, while keeping existing DB backstops + in place until coverage and fleet gates are proven. +- **Community fence:** supported writers acquire shared community deletion locks; + deletion lifecycle transitions acquire the matching exclusive lock. Multi- + community writers acquire locks in stable UUID order. +- **Replica floor:** compliant channel-event writers take the shared replica-floor + lock and validate against the active `buzz.created_at_floor` + + `clock_timestamp()` contract; the writer probe handshake takes the exclusive + counterpart before `S`/activity/token. +- **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. +- **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 + 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/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 00bb0dc3259..d49c8e9b043 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -562,6 +562,13 @@ pub const DEFAULT_LOCK_TIMEOUT_MS: u64 = 5_000; /// Default writer `idle_in_transaction_session_timeout` in milliseconds. pub const DEFAULT_IDLE_TXN_TIMEOUT_MS: u64 = 60_000; +fn channel_event_is_below_replica_floor( + event_created_at: DateTime, + floor_cutoff: DateTime, +) -> bool { + event_created_at < floor_cutoff +} + impl DbConfig { /// Overlay writer session timeouts from the shared `BUZZ_DB_*_TIMEOUT_MS` /// environment variables. Missing or invalid values retain the existing @@ -1143,6 +1150,81 @@ impl Db { .map_err(Into::into) } + /// Begin an event-write transaction and take the shared community + /// deletion/admission lock before any tenant mutation. + pub async fn begin_community_write_transaction( + &self, + community: CommunityId, + ) -> Result> { + let mut tx = self.begin_event_write_transaction().await?; + self.deletion_store() + .guard_transaction(&mut tx, community) + .await?; + Ok(tx) + } + + /// Begin one event-write transaction and guard multiple communities in + /// deterministic UUID order under shared deletion/admission locks. + /// + /// The sorted lock order prevents opposite-order callers from deadlocking + /// when they need overlapping communities in one transaction. + pub async fn begin_community_write_transaction_batch( + &self, + communities: &[CommunityId], + ) -> Result> { + if communities.is_empty() { + return Err(DbError::InvalidData( + "community write transaction batch requires at least one community".to_string(), + )); + } + let mut tx = self.begin_event_write_transaction().await?; + let mut ordered = communities.to_vec(); + ordered.sort_unstable(); + ordered.dedup(); + let store = self.deletion_store(); + for community in ordered { + store.guard_transaction(&mut tx, community).await?; + } + Ok(tx) + } + + /// Begin a channel-event write transaction that takes the shared + /// application-owned replica-floor lock and pre-validates `created_at` + /// against the active floor from the writer-session GUC. + /// + /// The commit-time trigger remains authoritative while this foundation + /// exists; this pre-check is a dual-enforcement guardrail for obvious + /// below-floor inputs before commit. + pub async fn begin_replica_floor_channel_write_transaction( + &self, + event_created_at: DateTime, + ) -> Result> { + let mut tx = self.begin_event_write_transaction().await?; + sqlx::query("SELECT pg_advisory_xact_lock_shared($1)") + .bind(replica_fence::REPLICA_FLOOR_LOCK_KEY) + .execute(&mut *tx) + .await?; + let (floor_secs, floor_cutoff): (Option, DateTime) = sqlx::query_as( + r#"SELECT nullif(current_setting('buzz.created_at_floor', true), '')::double precision, + clock_timestamp() - make_interval( + secs => COALESCE( + nullif(current_setting('buzz.created_at_floor', true), '')::double precision, + 0.0 + ) + )"#, + ) + .fetch_one(&mut *tx) + .await?; + if floor_secs.is_some_and(|secs| { + secs > 0.0 && channel_event_is_below_replica_floor(event_created_at, floor_cutoff) + }) { + return Err(DbError::InvalidData(format!( + "channel event created_at {event_created_at} is below active replica floor {floor_cutoff}" + ))); + } + Ok(tx) + } + /// Begin an event-write transaction through the pre-operation API name. /// /// New callers should use [`Self::begin_event_write_transaction`] so the diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 044dc3a58c6..72447c1ac40 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -10,10 +10,12 @@ //! is armed per session via the `buzz.created_at_floor` GUC, which the //! relay's writer pool sets on every connection. //! 2. **Ordered heartbeat handshake** (this module): on one pinned writer -//! connection, separately-awaited statements sample -//! `S = clock_timestamp()`, then scan `pg_stat_activity` for the oldest -//! open transaction, then — **last** — commit heartbeat token `M` via a -//! single-row `UPDATE replica_heartbeat ... RETURNING token, epoch` +//! connection, one transaction first takes the exclusive replica-floor +//! advisory lock (draining compliant shared-lock channel writers), then +//! separately-awaited statements sample `S = clock_timestamp()`, scan +//! `pg_stat_activity` for the oldest open transaction, and — **last** — +//! commit heartbeat token `M` via a single-row +//! `UPDATE replica_heartbeat ... RETURNING token, epoch` //! (migration 0026). Because the single-row UPDATE serializes all pods' //! probes, tokens are globally commit-ordered. A reader **session** that //! observes `token >= M` on its own connection has, by WAL/storage replay @@ -73,6 +75,14 @@ use buzz_datastore_tracing::datastore_span; /// and the fence subtracts it; the two uses must never diverge. pub const CREATED_AT_FLOOR_SECS: i64 = 960; +/// Deployment-global advisory-lock key ordering compliant channel-event write +/// 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. +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 @@ -550,22 +560,28 @@ pub enum ProbeError { HeartbeatRowMissing, } -/// Take one ordered writer sample: S, then activity scan, then commit the -/// heartbeat token **last**. +/// Take one ordered writer sample: under the exclusive replica-floor lock, +/// sample S, then activity scan, then commit the heartbeat token **last**. /// -/// The statements are separately awaited on a single pinned connection; +/// The statements are separately awaited on one pinned writer transaction; /// a single SELECT would not guarantee evaluation order across the /// subexpressions, reopening the race this ordering exists to close. async fn sample_writer(writer: &PgPool) -> Result { - let mut conn = crate::observability::acquire_writer( + let connection = crate::observability::acquire_writer( writer, crate::observability::WriterOperation::Maintenance, ) .await?; + let mut tx = sqlx::Transaction::begin(connection, None).await?; + + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(REPLICA_FLOOR_LOCK_KEY) + .execute(&mut *tx) + .await?; // 1. S first. let sampled_at: DateTime = sqlx::query_scalar("SELECT clock_timestamp()") - .fetch_one(&mut *conn) + .fetch_one(&mut *tx) .await?; // 2. Activity scan. Classification (fail closed on anything unknown): @@ -612,7 +628,7 @@ async fn sample_writer(writer: &PgPool) -> Result { ) AS masked "#, ) - .fetch_one(&mut *conn) + .fetch_one(&mut *tx) .await?; let masked: i64 = row.get("masked"); if masked > 0 { @@ -629,17 +645,19 @@ async fn sample_writer(writer: &PgPool) -> Result { let row = sqlx::query( "UPDATE replica_heartbeat SET token = token + 1 WHERE id = 1 RETURNING token, epoch", ) - .fetch_optional(&mut *conn) + .fetch_optional(&mut *tx) .await? .ok_or(ProbeError::HeartbeatRowMissing)?; - Ok(WriterSample { + let sample = WriterSample { sampled_at, oldest_xact_start, token: row.get("token"), epoch: row.get("epoch"), committed_at, - }) + }; + tx.commit().await?; + Ok(sample) } /// The fence wall proved by one handshake: diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index d97ebf0ac54..2a90e4da58f 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2825,6 +2825,160 @@ async fn armed_pool_rejects_old_channel_inserts_through_public_api() { db.pool.close().await; } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn replica_floor_writer_transaction_holds_shared_lock_and_validates_timestamp() { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (seed_pool, name) = create_scratch_db(&admin, "floor_writer_foundation").await; + + 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 now_secs = chrono::Utc::now().timestamp() as u64; + let floor = crate::replica_fence::CREATED_AT_FLOOR_SECS as u64; + let stale_ts = chrono::DateTime::from_timestamp((now_secs - floor - 120) as i64, 0) + .expect("valid stale timestamp"); + let error = db + .begin_replica_floor_channel_write_transaction(stale_ts) + .await + .expect_err("below-floor timestamp must reject"); + assert!( + matches!(&error, DbError::InvalidData(message) if message.contains("below active replica floor")), + "expected floor rejection, got: {error:#}" + ); + + let fresh_ts = chrono::DateTime::from_timestamp(now_secs as i64, 0).expect("valid timestamp"); + let writer = db + .begin_replica_floor_channel_write_transaction(fresh_ts) + .await + .expect("open compliant floor-guarded writer tx"); + + let mut contender = db.pool.begin().await.expect("begin exclusive contender"); + let exclusive_taken: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)") + .bind(crate::replica_fence::REPLICA_FLOOR_LOCK_KEY) + .fetch_one(&mut *contender) + .await + .expect("probe exclusive floor lock"); + assert!( + !exclusive_taken, + "compliant writer must hold the shared replica-floor advisory lock" + ); + + contender + .rollback() + .await + .expect("rollback exclusive contender"); + writer.rollback().await.expect("rollback writer tx"); + db.pool.close().await; + drop_scratch_db(&admin, seed_pool, &name).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn replica_floor_probe_waits_for_shared_writer_and_records_after_release() { + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (seed_pool, name) = create_scratch_db(&admin, "floor_probe_foundation").await; + + 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 token_before: i64 = sqlx::query_scalar("SELECT token FROM replica_heartbeat WHERE id = 1") + .fetch_one(&db.pool) + .await + .expect("read token before probe"); + + let writer = db + .begin_replica_floor_channel_write_transaction( + chrono::Utc::now() + chrono::Duration::seconds(60), + ) + .await + .expect("open compliant floor-guarded writer tx"); + + let probe_pool = db.pool.clone(); + let probe_fence = std::sync::Arc::clone(db.fence()); + let probing = tokio::spawn(async move { + crate::replica_fence::probe_once(&probe_pool, probe_fence.as_ref()).await + }); + let mut probing = probing; + assert!( + tokio::time::timeout(std::time::Duration::from_millis(100), &mut probing) + .await + .is_err(), + "probe must wait for the exclusive floor lock while compliant writer is open" + ); + + writer + .rollback() + .await + .expect("release shared floor writer"); + let entry = tokio::time::timeout(std::time::Duration::from_secs(5), probing) + .await + .expect("probe must complete after writer release") + .expect("probe task") + .expect("probe succeeds"); + + assert_eq!( + entry.token, + token_before + 1, + "existing handshake must publish one token via probe_once" + ); + assert_eq!( + db.fence().verified_through(), + Some(entry.fence_wall), + "probe entry must be retained in the in-memory fence ring" + ); + + let error = db + .begin_replica_floor_channel_write_transaction( + entry.fence_wall - chrono::Duration::microseconds(1), + ) + .await + .expect_err("later below-wall timestamp must reject"); + assert!( + matches!(&error, DbError::InvalidData(message) if message.contains("below active replica floor")), + "expected below-floor rejection, got: {error:#}" + ); + + db.pool.close().await; + drop_scratch_db(&admin, seed_pool, &name).await; +} + +#[test] +fn replica_floor_boundary_matches_trigger_contract() { + let boundary = chrono::DateTime::from_timestamp(1_700_000_000, 0).expect("valid boundary ts"); + assert!( + !channel_event_is_below_replica_floor(boundary, boundary), + "exact floor boundary must be admitted (trigger rejects only strict below-floor rows)" + ); + assert!( + channel_event_is_below_replica_floor( + boundary - chrono::Duration::microseconds(1), + boundary + ), + "rows strictly older than floor boundary must reject" + ); +} + /// `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 diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 0e184e00d88..09170f45350 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -4075,6 +4075,161 @@ mod postgres_tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn compliant_writer_shared_xact_lock_blocks_deletion_exclusive_xact_lock() { + let (db, _) = store().await; + let community = db + .ensure_configured_community(&format!( + "community-lock-contract-{}.example", + Uuid::new_v4().simple() + )) + .await + .expect("create community") + .id; + + let writer = db + .begin_community_write_transaction(community) + .await + .expect("open writer transaction with community lock"); + + let mut deleter = db.pool.begin().await.expect("begin deletion contender"); + let exclusive_taken: bool = + sqlx::query_scalar("SELECT pg_try_advisory_xact_lock(community_deletion_lock_key($1))") + .bind(community.as_uuid()) + .fetch_one(&mut *deleter) + .await + .expect("try deletion exclusive lock"); + assert!( + !exclusive_taken, + "compliant writer must hold shared community lock that blocks deletion exclusive lock" + ); + deleter + .rollback() + .await + .expect("rollback deletion contender"); + writer + .rollback() + .await + .expect("rollback writer transaction"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn community_fence_rejects_fresh_begin_transaction_after_fence() { + let (db, store) = store().await; + let (request, _) = inventoried_request(&db, &store).await; + store + .approve(request.id, "approver", None) + .await + .expect("approve"); + let claim = store + .claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION) + .await + .expect("claim") + .expect("won claim"); + + store.begin_quiescing(&claim.lease).await.expect("quiesce"); + store.fence(&claim.lease).await.expect("fence"); + + let error = db + .begin_community_write_transaction(request.community_id) + .await + .expect_err("fenced community must reject fresh write admission"); + assert!( + matches!(&error, DbError::AccessDenied(message) if message.contains("write-fenced")), + "expected write-fenced access denial, got: {error:#}" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn community_fence_batch_begin_rejects_empty_input() { + let (db, _) = store().await; + let error = db + .begin_community_write_transaction_batch(&[]) + .await + .expect_err("empty batch must fail closed"); + assert!( + matches!(&error, DbError::InvalidData(message) if message.contains("at least one community")), + "expected invalid-data empty-input rejection, got: {error:#}" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn community_fence_batch_begin_uses_stable_uuid_order() { + let (db, _) = store().await; + let community_a = db + .ensure_configured_community(&format!( + "community-fence-batch-a-{}.example", + Uuid::new_v4().simple() + )) + .await + .expect("create first community") + .id; + let community_b = db + .ensure_configured_community(&format!( + "community-fence-batch-b-{}.example", + Uuid::new_v4().simple() + )) + .await + .expect("create second community") + .id; + + let (first, second) = if community_a <= community_b { + (community_a, community_b) + } else { + (community_b, community_a) + }; + + let mut gate = db.pool.begin().await.expect("begin order gate"); + sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))") + .bind(first.as_uuid()) + .execute(&mut *gate) + .await + .expect("hold first community exclusively"); + + let db_for_batch = db.clone(); + let batching = tokio::spawn(async move { + let tx = db_for_batch + .begin_community_write_transaction_batch(&[second, first]) + .await?; + tx.rollback().await?; + Result::<()>::Ok(()) + }); + let mut batching = batching; + assert!( + tokio::time::timeout(Duration::from_millis(100), &mut batching) + .await + .is_err(), + "batch admission should block on the first sorted community lock" + ); + + let mut second_probe = db.pool.begin().await.expect("begin second-lock probe"); + let second_exclusive_taken: bool = + sqlx::query_scalar("SELECT pg_try_advisory_xact_lock(community_deletion_lock_key($1))") + .bind(second.as_uuid()) + .fetch_one(&mut *second_probe) + .await + .expect("probe second community exclusive lock"); + assert!( + second_exclusive_taken, + "stable ordering must block on the lower UUID lock before taking higher UUID lock" + ); + second_probe + .rollback() + .await + .expect("rollback second-lock probe"); + + gate.rollback().await.expect("release order gate"); + tokio::time::timeout(Duration::from_secs(5), batching) + .await + .expect("batch lock acquisition must not deadlock") + .expect("batch task") + .expect("batch transaction completes"); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn fence_waits_for_open_write_and_rejects_it_after_transition() { diff --git a/docs/plans/2026-09-17-triggerless-write-contract.md b/docs/plans/2026-09-17-triggerless-write-contract.md new file mode 100644 index 00000000000..31f02b9886e --- /dev/null +++ b/docs/plans/2026-09-17-triggerless-write-contract.md @@ -0,0 +1,53 @@ +# Implementation Plan — Triggerless Write Contract Foundation (2026-09-17) + +## Status + +Implemented on this branch for the first approved foundation slice. + +## Objective + +Establish application-owned transaction protocols for community fencing and +replica-floor ordering while preserving current trigger/function enforcement. + +## Scope in this change + +1. **Community-fence foundations** + - Added `Db::begin_community_write_transaction(community)`. + - Added `Db::begin_community_write_transaction_batch(communities)` with + stable UUID lock ordering and empty-input fail-closed behavior. + - Added tests proving: + - shared writer lock blocks deletion exclusive lock, + - post-fence fresh admission is rejected, + - batch ordering is deterministic and avoids opposite-order lock hazards, + - empty batch input is rejected. + +2. **Replica-floor foundations (no second publication concept)** + - Added `REPLICA_FLOOR_LOCK_KEY` and retained the existing + `probe_once/sample_writer` heartbeat publication model as the sole + token/fence-wall publication mechanism. + - Added `Db::begin_replica_floor_channel_write_transaction(created_at)`: + shared floor lock + GUC/clock pre-validation (`buzz.created_at_floor` + + `clock_timestamp()`-derived cutoff) under current trigger backstop. + - Updated the existing probe handshake to acquire the exclusive floor lock + **before** sampling `S`, activity scan, and heartbeat token commit. + - Added tests proving: + - compliant shared writer blocks probe progress, + - probe resumes and records normal token/fence-wall after writer release, + - below-wall later input rejects, + - boundary behavior follows strict `<` semantics. + +## Explicitly not in this change + +- Trigger/function removal. +- FK-based replacement for advisory lock ordering. +- Direct-owner SQL execution paths outside reviewed writer protocols. +- Any second replica-floor publication API outside `probe_once`. +- Persisting an active floor cutoff in schema. + +## Validation performed + +- Targeted RED→GREEN for each added behavior. +- Aggregated sweeps: + - `community_fence_` tests, + - `replica_floor_` tests, + - `probe_` tests. diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 7c3a7be871a..5a3e1653437 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -5,6 +5,16 @@ -- partition children as standalone CREATE TABLE statements. Every pgschema -- apply caller must run this idempotent script so fresh bootstraps converge on -- the same live database contract as migration-managed databases. +-- +-- Supported-writer operational contract (current transition): +-- * Direct owner SQL mutation is unsupported unless explicitly reviewed. +-- * Reviewed backfills/reconciliation must use relay-owned transaction/lock +-- protocols where available (community shared/exclusive fences, stable lock +-- order across communities, shared/exclusive replica-floor ordering), or +-- run while replica routing fences are intentionally held closed. +-- * Role separation is deliberate: schema reconciliation lives here; serving +-- admission/fencing remains in runtime/store code paths with their metrics +-- and startup verification gates. DO $$ BEGIN From d356c65e69ca00d41c59eb00e9d74ee7114e8943 Mon Sep 17 00:00:00 2001 From: tornquist Date: Thu, 17 Sep 2026 09:56:43 +0000 Subject: [PATCH 02/11] refactor(buzz-db): narrow lock foundations and clarify transition contract Signed-off-by: tornquist Co-authored-by: Codex --- ARCHITECTURE.md | 15 ++--- crates/buzz-db/src/runtime/mod.rs | 58 +++++-------------- crates/buzz-db/src/runtime/replica_fence.rs | 50 ++++++++++++++++ crates/buzz-db/src/runtime/tests.rs | 49 +--------------- crates/buzz-db/src/store/deletion.rs | 37 ++++++++++++ .../2026-09-17-triggerless-write-contract.md | 10 ++-- scripts/cutover/README.md | 15 +++++ scripts/reconcile-schema-after-pgschema.sql | 11 ---- 8 files changed, 132 insertions(+), 113 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 664190f24a5..aad5ddf3305 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -442,13 +442,14 @@ shape is an explicit transition, not a flag day: - **Trajectory:** converge from trigger/function/FK-enforced fencing toward application-owned transaction protocols, while keeping existing DB backstops in place until coverage and fleet gates are proven. -- **Community fence:** supported writers acquire shared community deletion locks; - deletion lifecycle transitions acquire the matching exclusive lock. Multi- - community writers acquire locks in stable UUID order. -- **Replica floor:** compliant channel-event writers take the shared replica-floor - lock and validate against the active `buzz.created_at_floor` + - `clock_timestamp()` contract; the writer probe handshake takes the exclusive - counterpart before `S`/activity/token. +- **Community fence (transition foundation):** runtime admission APIs now offer + shared community deletion locks, while deletion lifecycle transitions take the + matching exclusive lock. Multi-community batches lock in stable UUID order; + 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 `S`/activity/token. + 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. diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index d49c8e9b043..f32c55ca3e2 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -562,13 +562,6 @@ pub const DEFAULT_LOCK_TIMEOUT_MS: u64 = 5_000; /// Default writer `idle_in_transaction_session_timeout` in milliseconds. pub const DEFAULT_IDLE_TXN_TIMEOUT_MS: u64 = 60_000; -fn channel_event_is_below_replica_floor( - event_created_at: DateTime, - floor_cutoff: DateTime, -) -> bool { - event_created_at < floor_cutoff -} - impl DbConfig { /// Overlay writer session timeouts from the shared `BUZZ_DB_*_TIMEOUT_MS` /// environment variables. Missing or invalid values retain the existing @@ -1150,24 +1143,22 @@ impl Db { .map_err(Into::into) } - /// Begin an event-write transaction and take the shared community - /// deletion/admission lock before any tenant mutation. + /// Begin an event-write transaction and guard one community through the + /// stable multi-community admission path. pub async fn begin_community_write_transaction( &self, community: CommunityId, ) -> Result> { - let mut tx = self.begin_event_write_transaction().await?; - self.deletion_store() - .guard_transaction(&mut tx, community) - .await?; - Ok(tx) + self.begin_community_write_transaction_batch(&[community]) + .await } - /// Begin one event-write transaction and guard multiple communities in - /// deterministic UUID order under shared deletion/admission locks. + /// Begin one event-write transaction and guard communities in deterministic + /// UUID order under shared deletion/admission locks. /// - /// The sorted lock order prevents opposite-order callers from deadlocking - /// when they need overlapping communities in one transaction. + /// The batch must include every community the transaction may mutate. The + /// sorted lock order prevents opposite-order callers from deadlocking when + /// overlapping community sets are locked in one transaction. pub async fn begin_community_write_transaction_batch( &self, communities: &[CommunityId], @@ -1188,40 +1179,19 @@ impl Db { Ok(tx) } - /// Begin a channel-event write transaction that takes the shared - /// application-owned replica-floor lock and pre-validates `created_at` - /// against the active floor from the writer-session GUC. + /// Begin an event-write transaction that takes the shared replica-floor + /// advisory lock. /// - /// The commit-time trigger remains authoritative while this foundation - /// exists; this pre-check is a dual-enforcement guardrail for obvious - /// below-floor inputs before commit. - pub async fn begin_replica_floor_channel_write_transaction( + /// This is a lock-ordering foundation only. Floor correctness remains + /// authoritative at commit time via the existing trigger/GUC contract. + pub async fn begin_replica_floor_locked_event_write_transaction( &self, - event_created_at: DateTime, ) -> Result> { let mut tx = self.begin_event_write_transaction().await?; sqlx::query("SELECT pg_advisory_xact_lock_shared($1)") .bind(replica_fence::REPLICA_FLOOR_LOCK_KEY) .execute(&mut *tx) .await?; - let (floor_secs, floor_cutoff): (Option, DateTime) = sqlx::query_as( - r#"SELECT nullif(current_setting('buzz.created_at_floor', true), '')::double precision, - clock_timestamp() - make_interval( - secs => COALESCE( - nullif(current_setting('buzz.created_at_floor', true), '')::double precision, - 0.0 - ) - )"#, - ) - .fetch_one(&mut *tx) - .await?; - if floor_secs.is_some_and(|secs| { - secs > 0.0 && channel_event_is_below_replica_floor(event_created_at, floor_cutoff) - }) { - return Err(DbError::InvalidData(format!( - "channel event created_at {event_created_at} is below active replica floor {floor_cutoff}" - ))); - } Ok(tx) } diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 72447c1ac40..63c638e4ac9 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -1046,10 +1046,60 @@ mod postgres_tests { tx.rollback().await.expect("rollback"); } + /// The sample must not capture `S` while a compliant shared writer lock is + /// still held. It must block until release, then sample. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn sample_writer_waits_for_shared_floor_lock_before_sampling_time() { + let (admin, pool, name) = scratch_db().await; + + let mut blocker = pool.begin().await.expect("begin shared-lock blocker"); + sqlx::query("SELECT pg_advisory_xact_lock_shared($1)") + .bind(REPLICA_FLOOR_LOCK_KEY) + .execute(&mut *blocker) + .await + .expect("hold shared floor lock"); + + let sample_pool = pool.clone(); + let sampling = tokio::spawn(async move { sample_writer(&sample_pool).await }); + let mut sampling = sampling; + assert!( + tokio::time::timeout(Duration::from_millis(100), &mut sampling) + .await + .is_err(), + "sample_writer should block while a shared floor lock is held" + ); + + let marker_before_release: DateTime = sqlx::query_scalar("SELECT clock_timestamp()") + .fetch_one(&pool) + .await + .expect("capture marker while lock held"); + + blocker + .rollback() + .await + .expect("release shared-lock blocker"); + + let sample = tokio::time::timeout(Duration::from_secs(5), sampling) + .await + .expect("sample should complete after shared lock release") + .expect("sampling task") + .expect("sample writer"); + assert!( + sample.sampled_at >= marker_before_release, + "sample time {:?} must be at or after marker {:?}", + sample.sampled_at, + marker_before_release + ); + + drop_scratch_db(&admin, pool, &name).await; + } + /// An unprivileged probe role sees NULL `state`/`xact_start` for other /// sessions' rows in `pg_stat_activity`. The oldest-xact term is then /// untrustworthy and the sample must fail closed (`MaskedActivity`) — /// never silently `MIN()` the hidden row away. + #[tokio::test] #[ignore = "requires Postgres"] async fn cluster_global_sample_writer_fails_closed_when_activity_is_masked() { diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 2a90e4da58f..8c187a3c59a 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2827,7 +2827,7 @@ async fn armed_pool_rejects_old_channel_inserts_through_public_api() { #[tokio::test] #[ignore = "requires Postgres"] -async fn replica_floor_writer_transaction_holds_shared_lock_and_validates_timestamp() { +async fn replica_floor_writer_transaction_holds_shared_lock() { let admin = PgPool::connect(&admin_url().await) .await .expect("connect admin"); @@ -2844,22 +2844,8 @@ async fn replica_floor_writer_transaction_holds_shared_lock_and_validates_timest .await .expect("connect armed Db"); - let now_secs = chrono::Utc::now().timestamp() as u64; - let floor = crate::replica_fence::CREATED_AT_FLOOR_SECS as u64; - let stale_ts = chrono::DateTime::from_timestamp((now_secs - floor - 120) as i64, 0) - .expect("valid stale timestamp"); - let error = db - .begin_replica_floor_channel_write_transaction(stale_ts) - .await - .expect_err("below-floor timestamp must reject"); - assert!( - matches!(&error, DbError::InvalidData(message) if message.contains("below active replica floor")), - "expected floor rejection, got: {error:#}" - ); - - let fresh_ts = chrono::DateTime::from_timestamp(now_secs as i64, 0).expect("valid timestamp"); let writer = db - .begin_replica_floor_channel_write_transaction(fresh_ts) + .begin_replica_floor_locked_event_write_transaction() .await .expect("open compliant floor-guarded writer tx"); @@ -2908,9 +2894,7 @@ async fn replica_floor_probe_waits_for_shared_writer_and_records_after_release() .expect("read token before probe"); let writer = db - .begin_replica_floor_channel_write_transaction( - chrono::Utc::now() + chrono::Duration::seconds(60), - ) + .begin_replica_floor_locked_event_write_transaction() .await .expect("open compliant floor-guarded writer tx"); @@ -2948,37 +2932,10 @@ async fn replica_floor_probe_waits_for_shared_writer_and_records_after_release() "probe entry must be retained in the in-memory fence ring" ); - let error = db - .begin_replica_floor_channel_write_transaction( - entry.fence_wall - chrono::Duration::microseconds(1), - ) - .await - .expect_err("later below-wall timestamp must reject"); - assert!( - matches!(&error, DbError::InvalidData(message) if message.contains("below active replica floor")), - "expected below-floor rejection, got: {error:#}" - ); - db.pool.close().await; drop_scratch_db(&admin, seed_pool, &name).await; } -#[test] -fn replica_floor_boundary_matches_trigger_contract() { - let boundary = chrono::DateTime::from_timestamp(1_700_000_000, 0).expect("valid boundary ts"); - assert!( - !channel_event_is_below_replica_floor(boundary, boundary), - "exact floor boundary must be admitted (trigger rejects only strict below-floor rows)" - ); - assert!( - channel_event_is_below_replica_floor( - boundary - chrono::Duration::microseconds(1), - boundary - ), - "rows strictly older than floor boundary must reject" - ); -} - /// `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 diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 09170f45350..a9cb4a3e11c 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -4142,6 +4142,43 @@ mod postgres_tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn community_fence_singleton_and_batch_share_fenced_admission_contract() { + let (db, store) = store().await; + let (request, _) = inventoried_request(&db, &store).await; + store + .approve(request.id, "approver", None) + .await + .expect("approve"); + let claim = store + .claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION) + .await + .expect("claim") + .expect("won claim"); + + store.begin_quiescing(&claim.lease).await.expect("quiesce"); + store.fence(&claim.lease).await.expect("fence"); + + let singleton_error = db + .begin_community_write_transaction(request.community_id) + .await + .expect_err("fenced community must reject singleton admission"); + assert!( + matches!(&singleton_error, DbError::AccessDenied(message) if message.contains("write-fenced")), + "expected singleton write-fenced denial, got: {singleton_error:#}" + ); + + let batch_error = db + .begin_community_write_transaction_batch(&[request.community_id]) + .await + .expect_err("fenced community must reject singleton batch admission"); + assert!( + matches!(&batch_error, DbError::AccessDenied(message) if message.contains("write-fenced")), + "expected batch write-fenced denial, got: {batch_error:#}" + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn community_fence_batch_begin_rejects_empty_input() { diff --git a/docs/plans/2026-09-17-triggerless-write-contract.md b/docs/plans/2026-09-17-triggerless-write-contract.md index 31f02b9886e..b5855addcfc 100644 --- a/docs/plans/2026-09-17-triggerless-write-contract.md +++ b/docs/plans/2026-09-17-triggerless-write-contract.md @@ -25,16 +25,16 @@ replica-floor ordering while preserving current trigger/function enforcement. - Added `REPLICA_FLOOR_LOCK_KEY` and retained the existing `probe_once/sample_writer` heartbeat publication model as the sole token/fence-wall publication mechanism. - - Added `Db::begin_replica_floor_channel_write_transaction(created_at)`: - shared floor lock + GUC/clock pre-validation (`buzz.created_at_floor` + - `clock_timestamp()`-derived cutoff) under current trigger backstop. + - Added `Db::begin_replica_floor_locked_event_write_transaction()`: + shared replica-floor transaction lock only (no caller timestamp preflight), + while retaining the existing commit-time trigger/GUC backstop. - Updated the existing probe handshake to acquire the exclusive floor lock **before** sampling `S`, activity scan, and heartbeat token commit. - Added tests proving: - compliant shared writer blocks probe progress, - probe resumes and records normal token/fence-wall after writer release, - - below-wall later input rejects, - - boundary behavior follows strict `<` semantics. + - `sample_writer` cannot sample `S` until a held shared lock releases + (timestamp marker proof with bounded synchronization). ## Explicitly not in this change diff --git a/scripts/cutover/README.md b/scripts/cutover/README.md index 27388bdaa2b..e3f6f9c9d01 100644 --- a/scripts/cutover/README.md +++ b/scripts/cutover/README.md @@ -19,6 +19,21 @@ Only when upgrading a Postgres that already holds **pre-1321 single-community data** to 1321. A brand-new deployment does **not** run this — it provisions from `migrations/0001_initial_schema.sql` (or `schema/schema.sql`) directly. +## Supported writer contract during transition + +Outside this one-off cutover, direct owner SQL mutation is not a supported +steady-state path unless explicitly reviewed. + +- Reviewed backfills and reconciliations should use relay-owned transaction and + lock protocols where they exist (community shared/exclusive fence ordering, + stable multi-community lock order, and replica-floor shared/exclusive lock + ordering). +- If a workflow cannot use those protocols yet, execute it under an explicitly + reviewed procedure that keeps replica routing fences closed for the run. +- Role separation is deliberate: schema reconciliation scripts carry schema + convergence only; serving admission/fencing semantics remain in runtime/store + code paths with startup verification and metrics. + ## Preconditions - The DB is pristine pre-1321: no `communities` table, no `community_id` diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 5a3e1653437..c44c1e7384b 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -5,17 +5,6 @@ -- partition children as standalone CREATE TABLE statements. Every pgschema -- apply caller must run this idempotent script so fresh bootstraps converge on -- the same live database contract as migration-managed databases. --- --- Supported-writer operational contract (current transition): --- * Direct owner SQL mutation is unsupported unless explicitly reviewed. --- * Reviewed backfills/reconciliation must use relay-owned transaction/lock --- protocols where available (community shared/exclusive fences, stable lock --- order across communities, shared/exclusive replica-floor ordering), or --- run while replica routing fences are intentionally held closed. --- * Role separation is deliberate: schema reconciliation lives here; serving --- admission/fencing remains in runtime/store code paths with their metrics --- and startup verification gates. - DO $$ BEGIN IF NOT EXISTS ( From 151d4cef1e6b59c5e0c9602ed020f5392091e3d0 Mon Sep 17 00:00:00 2001 From: tornquist Date: Tue, 22 Sep 2026 19:19:17 +0000 Subject: [PATCH 03/11] refactor(buzz-db): keep community write locks tenant-local Signed-off-by: tornquist Co-authored-by: Codex --- ARCHITECTURE.md | 7 +- crates/buzz-db/src/runtime/mod.rs | 33 ++----- crates/buzz-db/src/store/deletion.rs | 125 --------------------------- scripts/cutover/README.md | 5 +- 4 files changed, 11 insertions(+), 159 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index aad5ddf3305..1e26db8678d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -443,9 +443,10 @@ shape is an explicit transition, not a flag day: application-owned transaction protocols, while keeping existing DB backstops in place until coverage and fleet gates are proven. - **Community fence (transition foundation):** runtime admission APIs now offer - shared community deletion locks, while deletion lifecycle transitions take the - matching exclusive lock. Multi-community batches lock in stable UUID order; - unmigrated paths still rely on trigger/function backstops. + one shared community deletion lock per transaction, while deletion lifecycle + 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 `S`/activity/token. diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index f32c55ca3e2..e7596ebdd59 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -1143,39 +1143,16 @@ impl Db { .map_err(Into::into) } - /// Begin an event-write transaction and guard one community through the - /// stable multi-community admission path. + /// Begin an event-write transaction and guard its community against + /// concurrent deletion. pub async fn begin_community_write_transaction( &self, community: CommunityId, ) -> Result> { - self.begin_community_write_transaction_batch(&[community]) - .await - } - - /// Begin one event-write transaction and guard communities in deterministic - /// UUID order under shared deletion/admission locks. - /// - /// The batch must include every community the transaction may mutate. The - /// sorted lock order prevents opposite-order callers from deadlocking when - /// overlapping community sets are locked in one transaction. - pub async fn begin_community_write_transaction_batch( - &self, - communities: &[CommunityId], - ) -> Result> { - if communities.is_empty() { - return Err(DbError::InvalidData( - "community write transaction batch requires at least one community".to_string(), - )); - } let mut tx = self.begin_event_write_transaction().await?; - let mut ordered = communities.to_vec(); - ordered.sort_unstable(); - ordered.dedup(); - let store = self.deletion_store(); - for community in ordered { - store.guard_transaction(&mut tx, community).await?; - } + self.deletion_store() + .guard_transaction(&mut tx, community) + .await?; Ok(tx) } diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index a9cb4a3e11c..c8d1e4a3800 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -4142,131 +4142,6 @@ mod postgres_tests { ); } - #[tokio::test] - #[ignore = "requires Postgres"] - async fn community_fence_singleton_and_batch_share_fenced_admission_contract() { - let (db, store) = store().await; - let (request, _) = inventoried_request(&db, &store).await; - store - .approve(request.id, "approver", None) - .await - .expect("approve"); - let claim = store - .claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION) - .await - .expect("claim") - .expect("won claim"); - - store.begin_quiescing(&claim.lease).await.expect("quiesce"); - store.fence(&claim.lease).await.expect("fence"); - - let singleton_error = db - .begin_community_write_transaction(request.community_id) - .await - .expect_err("fenced community must reject singleton admission"); - assert!( - matches!(&singleton_error, DbError::AccessDenied(message) if message.contains("write-fenced")), - "expected singleton write-fenced denial, got: {singleton_error:#}" - ); - - let batch_error = db - .begin_community_write_transaction_batch(&[request.community_id]) - .await - .expect_err("fenced community must reject singleton batch admission"); - assert!( - matches!(&batch_error, DbError::AccessDenied(message) if message.contains("write-fenced")), - "expected batch write-fenced denial, got: {batch_error:#}" - ); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn community_fence_batch_begin_rejects_empty_input() { - let (db, _) = store().await; - let error = db - .begin_community_write_transaction_batch(&[]) - .await - .expect_err("empty batch must fail closed"); - assert!( - matches!(&error, DbError::InvalidData(message) if message.contains("at least one community")), - "expected invalid-data empty-input rejection, got: {error:#}" - ); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn community_fence_batch_begin_uses_stable_uuid_order() { - let (db, _) = store().await; - let community_a = db - .ensure_configured_community(&format!( - "community-fence-batch-a-{}.example", - Uuid::new_v4().simple() - )) - .await - .expect("create first community") - .id; - let community_b = db - .ensure_configured_community(&format!( - "community-fence-batch-b-{}.example", - Uuid::new_v4().simple() - )) - .await - .expect("create second community") - .id; - - let (first, second) = if community_a <= community_b { - (community_a, community_b) - } else { - (community_b, community_a) - }; - - let mut gate = db.pool.begin().await.expect("begin order gate"); - sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))") - .bind(first.as_uuid()) - .execute(&mut *gate) - .await - .expect("hold first community exclusively"); - - let db_for_batch = db.clone(); - let batching = tokio::spawn(async move { - let tx = db_for_batch - .begin_community_write_transaction_batch(&[second, first]) - .await?; - tx.rollback().await?; - Result::<()>::Ok(()) - }); - let mut batching = batching; - assert!( - tokio::time::timeout(Duration::from_millis(100), &mut batching) - .await - .is_err(), - "batch admission should block on the first sorted community lock" - ); - - let mut second_probe = db.pool.begin().await.expect("begin second-lock probe"); - let second_exclusive_taken: bool = - sqlx::query_scalar("SELECT pg_try_advisory_xact_lock(community_deletion_lock_key($1))") - .bind(second.as_uuid()) - .fetch_one(&mut *second_probe) - .await - .expect("probe second community exclusive lock"); - assert!( - second_exclusive_taken, - "stable ordering must block on the lower UUID lock before taking higher UUID lock" - ); - second_probe - .rollback() - .await - .expect("rollback second-lock probe"); - - gate.rollback().await.expect("release order gate"); - tokio::time::timeout(Duration::from_secs(5), batching) - .await - .expect("batch lock acquisition must not deadlock") - .expect("batch task") - .expect("batch transaction completes"); - } - #[tokio::test] #[ignore = "requires Postgres"] async fn fence_waits_for_open_write_and_rejects_it_after_transition() { diff --git a/scripts/cutover/README.md b/scripts/cutover/README.md index e3f6f9c9d01..be6f63c8999 100644 --- a/scripts/cutover/README.md +++ b/scripts/cutover/README.md @@ -25,9 +25,8 @@ Outside this one-off cutover, direct owner SQL mutation is not a supported steady-state path unless explicitly reviewed. - Reviewed backfills and reconciliations should use relay-owned transaction and - lock protocols where they exist (community shared/exclusive fence ordering, - stable multi-community lock order, and replica-floor shared/exclusive lock - ordering). + lock protocols where they exist (one community fence per transaction and + replica-floor shared/exclusive lock ordering). - If a workflow cannot use those protocols yet, execute it under an explicitly reviewed procedure that keeps replica routing fences closed for the run. - Role separation is deliberate: schema reconciliation scripts carry schema From d9e0b3aea49a54c2caa57bf1b5a539838eecc54c Mon Sep 17 00:00:00 2001 From: tornquist Date: Tue, 22 Sep 2026 19:19:44 +0000 Subject: [PATCH 04/11] docs: move writer contract into maintained references Signed-off-by: tornquist Co-authored-by: Codex --- ARCHITECTURE.md | 10 ++-- .../2026-09-17-triggerless-write-contract.md | 53 ------------------- scripts/cutover/README.md | 3 ++ 3 files changed, 10 insertions(+), 56 deletions(-) delete mode 100644 docs/plans/2026-09-17-triggerless-write-contract.md diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 1e26db8678d..673efa621d8 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -449,11 +449,15 @@ shape is an explicit transition, not a flag day: 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 `S`/activity/token. - Commit-time trigger+GUC enforcement remains authoritative for all paths. + 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. + remain authoritative during this phase. Removing those backstops is a separate + gated migration, not part of the lock foundation. - **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. diff --git a/docs/plans/2026-09-17-triggerless-write-contract.md b/docs/plans/2026-09-17-triggerless-write-contract.md deleted file mode 100644 index b5855addcfc..00000000000 --- a/docs/plans/2026-09-17-triggerless-write-contract.md +++ /dev/null @@ -1,53 +0,0 @@ -# Implementation Plan — Triggerless Write Contract Foundation (2026-09-17) - -## Status - -Implemented on this branch for the first approved foundation slice. - -## Objective - -Establish application-owned transaction protocols for community fencing and -replica-floor ordering while preserving current trigger/function enforcement. - -## Scope in this change - -1. **Community-fence foundations** - - Added `Db::begin_community_write_transaction(community)`. - - Added `Db::begin_community_write_transaction_batch(communities)` with - stable UUID lock ordering and empty-input fail-closed behavior. - - Added tests proving: - - shared writer lock blocks deletion exclusive lock, - - post-fence fresh admission is rejected, - - batch ordering is deterministic and avoids opposite-order lock hazards, - - empty batch input is rejected. - -2. **Replica-floor foundations (no second publication concept)** - - Added `REPLICA_FLOOR_LOCK_KEY` and retained the existing - `probe_once/sample_writer` heartbeat publication model as the sole - token/fence-wall publication mechanism. - - Added `Db::begin_replica_floor_locked_event_write_transaction()`: - shared replica-floor transaction lock only (no caller timestamp preflight), - while retaining the existing commit-time trigger/GUC backstop. - - Updated the existing probe handshake to acquire the exclusive floor lock - **before** sampling `S`, activity scan, and heartbeat token commit. - - Added tests proving: - - compliant shared writer blocks probe progress, - - probe resumes and records normal token/fence-wall after writer release, - - `sample_writer` cannot sample `S` until a held shared lock releases - (timestamp marker proof with bounded synchronization). - -## Explicitly not in this change - -- Trigger/function removal. -- FK-based replacement for advisory lock ordering. -- Direct-owner SQL execution paths outside reviewed writer protocols. -- Any second replica-floor publication API outside `probe_once`. -- Persisting an active floor cutoff in schema. - -## Validation performed - -- Targeted RED→GREEN for each added behavior. -- Aggregated sweeps: - - `community_fence_` tests, - - `replica_floor_` tests, - - `probe_` tests. diff --git a/scripts/cutover/README.md b/scripts/cutover/README.md index be6f63c8999..51c6b836b8e 100644 --- a/scripts/cutover/README.md +++ b/scripts/cutover/README.md @@ -29,6 +29,9 @@ steady-state path unless explicitly reviewed. replica-floor shared/exclusive lock ordering). - If a workflow cannot use those protocols yet, execute it under an explicitly reviewed procedure that keeps replica routing fences closed for the run. +- Existing trigger/function backstops remain the commit-time authority during + this transition. Their removal requires separate coverage and fleet gates; + this cutover does not authorize it. - Role separation is deliberate: schema reconciliation scripts carry schema convergence only; serving admission/fencing semantics remain in runtime/store code paths with startup verification and metrics. From 4a897b254bf8e02cf84a2c380759e59527658bf3 Mon Sep 17 00:00:00 2001 From: tornquist Date: Thu, 24 Sep 2026 18:18:43 +0000 Subject: [PATCH 05/11] fix(buzz-db): keep community write locks shared Signed-off-by: tornquist Co-authored-by: Codex --- crates/buzz-db/src/runtime/tests.rs | 17 ++++++++++++++++- crates/buzz-db/src/store/deletion.rs | 19 ++++++++++++++++++- 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 8c187a3c59a..eb137771273 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2838,7 +2838,7 @@ async fn replica_floor_writer_transaction_holds_shared_lock() { let scratch_url = format!("{}/{}", &base[..idx], name); let db = Db::new(&DbConfig { database_url: scratch_url, - max_connections: 2, + max_connections: 3, ..DbConfig::default() }) .await @@ -2849,6 +2849,17 @@ async fn replica_floor_writer_transaction_holds_shared_lock() { .await .expect("open compliant floor-guarded writer tx"); + let mut shared_contender = db.pool.begin().await.expect("begin shared contender"); + let shared_taken: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock_shared($1)") + .bind(crate::replica_fence::REPLICA_FLOOR_LOCK_KEY) + .fetch_one(&mut *shared_contender) + .await + .expect("probe shared floor lock"); + assert!( + shared_taken, + "compliant writer must allow another shared replica-floor lock holder" + ); + let mut contender = db.pool.begin().await.expect("begin exclusive contender"); let exclusive_taken: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)") .bind(crate::replica_fence::REPLICA_FLOOR_LOCK_KEY) @@ -2864,6 +2875,10 @@ async fn replica_floor_writer_transaction_holds_shared_lock() { .rollback() .await .expect("rollback exclusive contender"); + shared_contender + .rollback() + .await + .expect("rollback shared contender"); writer.rollback().await.expect("rollback writer tx"); db.pool.close().await; drop_scratch_db(&admin, seed_pool, &name).await; diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index c8d1e4a3800..cb12b3d657f 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -2666,7 +2666,7 @@ async fn lock_community_deletion( ) -> Result<()> { crate::observability::observe_advisory_lock( crate::observability::LockType::Deletion, - sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))") + sqlx::query("SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1))") .bind(community.as_uuid()) .execute(&mut **tx), ) @@ -4093,6 +4093,19 @@ mod postgres_tests { .await .expect("open writer transaction with community lock"); + let mut shared_contender = db.pool.begin().await.expect("begin shared contender"); + let shared_taken: bool = sqlx::query_scalar( + "SELECT pg_try_advisory_xact_lock_shared(community_deletion_lock_key($1))", + ) + .bind(community.as_uuid()) + .fetch_one(&mut *shared_contender) + .await + .expect("try shared deletion lock"); + assert!( + shared_taken, + "compliant writer must allow another shared community deletion lock holder" + ); + let mut deleter = db.pool.begin().await.expect("begin deletion contender"); let exclusive_taken: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock(community_deletion_lock_key($1))") @@ -4108,6 +4121,10 @@ mod postgres_tests { .rollback() .await .expect("rollback deletion contender"); + shared_contender + .rollback() + .await + .expect("rollback shared contender"); writer .rollback() .await From 9f88eea2d79cfca3db398732d292d853b1790138 Mon Sep 17 00:00:00 2001 From: tornquist Date: Thu, 24 Sep 2026 18:18:48 +0000 Subject: [PATCH 06/11] test(buzz-db): pin the exact fenced write denial Signed-off-by: tornquist Co-authored-by: Codex --- crates/buzz-db/src/store/deletion.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index cb12b3d657f..4dd10eec791 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -4154,7 +4154,12 @@ mod postgres_tests { .await .expect_err("fenced community must reject fresh write admission"); assert!( - matches!(&error, DbError::AccessDenied(message) if message.contains("write-fenced")), + matches!( + &error, + DbError::AccessDenied(message) + if message + == &format!("community {} is write-fenced (fenced)", request.community_id) + ), "expected write-fenced access denial, got: {error:#}" ); } From 9e178ab098b44f12e55bdd1d624b11acbf4176db Mon Sep 17 00:00:00 2001 From: tornquist Date: Thu, 24 Sep 2026 19:22:38 +0000 Subject: [PATCH 07/11] fix(buzz-db): serialize abort and quiescing Co-authored-by: Codex Signed-off-by: tornquist --- crates/buzz-db/src/store/deletion.rs | 82 ++++++++++++++++++++++------ 1 file changed, 66 insertions(+), 16 deletions(-) diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 4dd10eec791..ecbbfdfaf79 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -1284,6 +1284,7 @@ impl DeletionStore { .bind(token.community_id.as_uuid()) .fetch_one(&mut *tx) .await?; + verify_lease(&mut tx, token, DeletionStage::Approved).await?; sqlx::query( "UPDATE community_deletion_requests SET pre_quiesce_archived_at = $2, \ quiescing_started_at = now(), updated_at = now() \ @@ -2033,9 +2034,16 @@ impl DeletionStore { .await? .map(CommunityId::from_uuid) .ok_or_else(|| DbError::NotFound(format!("community deletion {request_id}")))?; - // Every lifecycle transition takes the community lock before any row lock. - // Inverting this order lets abort and the executor deadlock each other. + // Every lifecycle transition takes the community advisory lock before any + // row lock, then the community row before the request row. Inverting + // either order lets abort and the executor deadlock each other. lock_community_deletion(&mut tx, community_id).await?; + let (old_generation, current_archived_at): (i64, Option>) = sqlx::query_as( + "SELECT deletion_fence_generation, archived_at FROM communities WHERE id = $1 FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .fetch_one(&mut *tx) + .await?; let row = sqlx::query("SELECT * FROM community_deletion_requests WHERE id = $1 FOR UPDATE") .bind(request_id) .fetch_optional(&mut *tx) @@ -2068,12 +2076,6 @@ impl DeletionStore { "deletion {request_id} cannot abort while {active_writes} serving write lease(s) remain active" ))); } - let (old_generation, current_archived_at): (i64, Option>) = sqlx::query_as( - "SELECT deletion_fence_generation, archived_at FROM communities WHERE id = $1 FOR UPDATE", - ) - .bind(request.community_id.as_uuid()) - .fetch_one(&mut *tx) - .await?; let new_generation = old_generation.checked_add(1).ok_or_else(|| { DbError::DeletionSafety("community deletion fence generation overflow".to_string()) })?; @@ -3755,6 +3757,12 @@ mod postgres_tests { .execute(&mut *gate) .await .expect("hold community lock"); + let mut request_gate = db.pool.begin().await.expect("begin request gate"); + sqlx::query("SELECT id FROM community_deletion_requests WHERE id = $1 FOR UPDATE") + .bind(request.id) + .execute(&mut *request_gate) + .await + .expect("hold request row gate"); let abort_store = store.clone(); let aborting = tokio::spawn(async move { @@ -3776,25 +3784,67 @@ mod postgres_tests { "forward transition must queue on the same lock" ); gate.commit().await.expect("release lock gate"); + tokio::time::sleep(Duration::from_millis(50)).await; + assert!( + !aborting.is_finished(), + "abort must still be queued while the request row stays locked" + ); + assert!( + !forwarding.is_finished(), + "quiescing must still be queued while the request row stays locked" + ); + request_gate + .commit() + .await + .expect("release request row gate"); let aborted = tokio::time::timeout(Duration::from_secs(5), aborting) .await .expect("abort must not deadlock") - .expect("abort task") - .expect("abort wins lock queue"); - assert_eq!(aborted.stage, DeletionStage::Aborted); - let forward_error = tokio::time::timeout(Duration::from_secs(5), forwarding) + .expect("abort task"); + let forwarding = tokio::time::timeout(Duration::from_secs(5), forwarding) .await .expect("forward transition must not deadlock") - .expect("forward task") - .expect_err("post-lock lease verification rejects aborted request"); + .expect("forward task"); + assert!( !matches!( - &forward_error, - DbError::Sqlx(sqlx::Error::Database(error)) if error.code().as_deref() == Some("40P01") + &aborted, + Err(DbError::Sqlx(sqlx::Error::Database(error))) + if error.code().as_deref() == Some("40P01") + ), + "abort must not report a PostgreSQL deadlock" + ); + assert!( + !matches!( + &forwarding, + Err(DbError::Sqlx(sqlx::Error::Database(error))) + if error.code().as_deref() == Some("40P01") ), "serialization must not report a PostgreSQL deadlock" ); + + match (aborted, forwarding) { + (Ok(aborted), Ok(())) => { + assert_eq!(aborted.stage, DeletionStage::Aborted); + let reloaded = store.get(request.id).await.expect("reload request"); + assert_eq!(reloaded.stage, DeletionStage::Aborted); + assert!( + reloaded.quiescing_started_at.is_some(), + "successful quiescing should persist its durable intent before abort" + ); + } + (Ok(aborted), Err(DbError::AccessDenied(message))) => { + assert_eq!(aborted.stage, DeletionStage::Aborted); + assert!( + message.contains("stale deletion lease"), + "post-lock quiescing must reject a stale aborted lease: {message}" + ); + } + (aborted, forwarding) => panic!( + "unexpected abort/quiesce outcome: abort={aborted:?}, forward={forwarding:?}" + ), + } } #[tokio::test] From 5bec227b9c2155844c45698e7c8ad56553e5edd3 Mon Sep 17 00:00:00 2001 From: tornquist Date: Thu, 24 Sep 2026 20:51:53 +0000 Subject: [PATCH 08/11] fix(buzz-db): keep lifecycle deletion locks exclusive Co-authored-by: Codex Signed-off-by: tornquist --- crates/buzz-db/src/store/deletion.rs | 62 +++++++++++++++++++++++++++- 1 file changed, 61 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index ecbbfdfaf79..ef45908d675 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -2668,7 +2668,7 @@ async fn lock_community_deletion( ) -> Result<()> { crate::observability::observe_advisory_lock( crate::observability::LockType::Deletion, - sqlx::query("SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1))") + sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))") .bind(community.as_uuid()) .execute(&mut **tx), ) @@ -4214,6 +4214,66 @@ mod postgres_tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn begin_quiescing_waits_for_open_admitted_writer_note_update() { + let (db, store) = store().await; + let (request, _) = inventoried_request(&db, &store).await; + store + .approve(request.id, "approver", None) + .await + .expect("approve"); + let claim = store + .claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION) + .await + .expect("claim") + .expect("won claim"); + + let pubkey = [7_u8; 32]; + let added_by = [8_u8; 32]; + assert!( + db.add_to_allowlist(request.community_id, &pubkey, &added_by, Some("before")) + .await + .expect("seed allowlist row"), + "seeded allowlist row must insert" + ); + + let mut admitted_writer = db + .begin_community_write_transaction(request.community_id) + .await + .expect("open admitted writer"); + let updated = sqlx::query( + "UPDATE pubkey_allowlist SET note = $3 WHERE community_id = $1 AND pubkey = $2", + ) + .bind(request.community_id.as_uuid()) + .bind(pubkey.to_vec()) + .bind("during") + .execute(&mut *admitted_writer) + .await + .expect("update allowlist note") + .rows_affected(); + assert_eq!(updated, 1, "note-only update must touch the seeded row"); + + let store_for_quiesce = store.clone(); + let lease = claim.lease.clone(); + let quiescing = + tokio::spawn(async move { store_for_quiesce.begin_quiescing(&lease).await }); + tokio::time::sleep(Duration::from_millis(50)).await; + assert!( + !quiescing.is_finished(), + "begin_quiescing must wait for an admitted writer holding the shared deletion lock" + ); + + admitted_writer + .commit() + .await + .expect("release admitted writer"); + quiescing + .await + .expect("quiesce task") + .expect("quiesce after writer release"); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn fence_waits_for_open_write_and_rejects_it_after_transition() { From dd91a7598c618ad6bae931bafad41ffb9b09a0fd Mon Sep 17 00:00:00 2001 From: tornquist Date: Fri, 25 Sep 2026 20:19:50 +0000 Subject: [PATCH 09/11] test(buzz-db): isolate lock mode witnesses Signed-off-by: tornquist Co-authored-by: Codex --- crates/buzz-db/src/runtime/tests.rs | 8 ++++---- crates/buzz-db/src/store/deletion.rs | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index ec69d1f83ed..5201110f5e3 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -3132,6 +3132,10 @@ async fn replica_floor_writer_transaction_holds_shared_lock() { shared_taken, "compliant writer must allow another shared replica-floor lock holder" ); + shared_contender + .rollback() + .await + .expect("rollback shared contender"); let mut contender = db.pool.begin().await.expect("begin exclusive contender"); let exclusive_taken: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)") @@ -3148,10 +3152,6 @@ async fn replica_floor_writer_transaction_holds_shared_lock() { .rollback() .await .expect("rollback exclusive contender"); - shared_contender - .rollback() - .await - .expect("rollback shared contender"); writer.rollback().await.expect("rollback writer tx"); db.pool.close().await; drop_scratch_db(&admin, seed_pool, &name).await; diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index ef45908d675..dace2a3cb83 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -4155,6 +4155,10 @@ mod postgres_tests { shared_taken, "compliant writer must allow another shared community deletion lock holder" ); + shared_contender + .rollback() + .await + .expect("rollback shared contender"); let mut deleter = db.pool.begin().await.expect("begin deletion contender"); let exclusive_taken: bool = @@ -4171,10 +4175,6 @@ mod postgres_tests { .rollback() .await .expect("rollback deletion contender"); - shared_contender - .rollback() - .await - .expect("rollback shared contender"); writer .rollback() .await From 04befdfb5b470410d1a4220afb2314f08edcd216 Mon Sep 17 00:00:00 2001 From: tornquist Date: Fri, 25 Sep 2026 20:20:05 +0000 Subject: [PATCH 10/11] test(buzz-db): clean up replica fence tests Signed-off-by: tornquist Co-authored-by: Codex --- crates/buzz-db/src/runtime/replica_fence.rs | 4 +--- crates/buzz-db/src/runtime/tests.rs | 3 +-- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 63c638e4ac9..520ff9e8698 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -1061,8 +1061,7 @@ mod postgres_tests { .expect("hold shared floor lock"); let sample_pool = pool.clone(); - let sampling = tokio::spawn(async move { sample_writer(&sample_pool).await }); - let mut sampling = sampling; + let mut sampling = tokio::spawn(async move { sample_writer(&sample_pool).await }); assert!( tokio::time::timeout(Duration::from_millis(100), &mut sampling) .await @@ -1099,7 +1098,6 @@ mod postgres_tests { /// sessions' rows in `pg_stat_activity`. The oldest-xact term is then /// untrustworthy and the sample must fail closed (`MaskedActivity`) — /// never silently `MIN()` the hidden row away. - #[tokio::test] #[ignore = "requires Postgres"] async fn cluster_global_sample_writer_fails_closed_when_activity_is_masked() { diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 5201110f5e3..4a7d64e7a25 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -3188,10 +3188,9 @@ async fn replica_floor_probe_waits_for_shared_writer_and_records_after_release() let probe_pool = db.pool.clone(); let probe_fence = std::sync::Arc::clone(db.fence()); - let probing = tokio::spawn(async move { + let mut probing = tokio::spawn(async move { crate::replica_fence::probe_once(&probe_pool, probe_fence.as_ref()).await }); - let mut probing = probing; assert!( tokio::time::timeout(std::time::Duration::from_millis(100), &mut probing) .await From 12a45379251adc9d4d57300cd2854aac621f10b7 Mon Sep 17 00:00:00 2001 From: tornquist Date: Fri, 25 Sep 2026 20:20:16 +0000 Subject: [PATCH 11/11] chore: drop unrelated schema script diff Signed-off-by: tornquist Co-authored-by: Codex --- scripts/reconcile-schema-after-pgschema.sql | 1 + 1 file changed, 1 insertion(+) diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index c44c1e7384b..7c3a7be871a 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -5,6 +5,7 @@ -- partition children as standalone CREATE TABLE statements. Every pgschema -- apply caller must run this idempotent script so fresh bootstraps converge on -- the same live database contract as migration-managed databases. + DO $$ BEGIN IF NOT EXISTS (