diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index bf435c270a3..aad5ddf3305 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -433,6 +433,35 @@ 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 (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. +- **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..f32c55ca3e2 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -1143,6 +1143,58 @@ impl Db { .map_err(Into::into) } + /// 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> { + 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?; + } + Ok(tx) + } + + /// Begin an event-write transaction that takes the shared replica-floor + /// advisory lock. + /// + /// 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, + ) -> 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?; + 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..63c638e4ac9 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: @@ -1028,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 d97ebf0ac54..8c187a3c59a 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2825,6 +2825,117 @@ 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() { + 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 writer = db + .begin_replica_floor_locked_event_write_transaction() + .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_locked_event_write_transaction() + .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" + ); + + db.pool.close().await; + 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 diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 0e184e00d88..a9cb4a3e11c 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -4075,6 +4075,198 @@ 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_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/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..b5855addcfc --- /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_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 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 7c3a7be871a..c44c1e7384b 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -5,7 +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. - DO $$ BEGIN IF NOT EXISTS (