From d06bab6b156679e46aaa2307436af54837a9b393 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 10:39:13 -0400 Subject: [PATCH 1/4] buzz-db: tolerate absent universal write-fence objects Expand step for retiring the 32 community_write_fence_* triggers and their SQL functions (BUZZ-315). Relay pods auto-migrate at boot, so the drop migration can only ship once every pod no longer depends on those objects. - Serving startup/readiness and the deletion catalog check no longer require the fence functions or triggers; tenant-table equality, the deletion lock key and the community tombstone are still required. - Live scoped-table discovery uses a Rust exclusion list instead of community_write_fence_excluded_table(), pinned equal to the SQL list while both exist. - Serving-lease admission uses guard_community_write instead of assert_community_write_allowed; fenced/missing still map to AccessDenied. - guard_community_write rejects non-READ-COMMITTED transactions, reading the isolation level from the existing advisory-lock statement (no extra round trip). - The deletion executor stops comparing the legacy fenced_tables snapshot, but inventories still record live fences so older executors keep matching during this rollout. No schema or migration change. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- ARCHITECTURE.md | 8 +- crates/buzz-db/src/runtime/admitted_tx.rs | 5 +- crates/buzz-db/src/store/deletion.rs | 278 ++++++++++++++++------ crates/buzz-deletion/src/lib.rs | 50 +++- 4 files changed, 251 insertions(+), 90 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index e0875618dd0..127a5492f50 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -501,8 +501,9 @@ shape is an explicit transition, not a flag day: - **Community fence (transition foundation):** runtime admission APIs now offer 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. + never spans communities. Serving leases use the same application admission. + Startup and deletion catalog validation tolerate the universal fence objects + being absent; inventories still record live fences for older executors. - **Replica floor (transition foundation):** runtime now provides a shared replica-floor lock helper for compliant channel-event writers, and the writer probe handshake takes the exclusive counterpart before sampling `S`, scanning @@ -513,7 +514,8 @@ shape is an explicit transition, not a flag day: - **Dual enforcement (current):** application-owned lock+precheck paths run in front of the existing trigger/function enforcement; commit-time trigger checks remain authoritative during this phase. Removing those backstops is a separate - gated migration, not part of the lock foundation. + gated migration, after every pod tolerates their absence and no rollback below + that version is expected. - **Role separation:** relay/runtime code owns admission and lock protocols; operator maintenance/backfill workflows must either use those protocols or run under explicit reviewed procedures that keep routing fences closed. diff --git a/crates/buzz-db/src/runtime/admitted_tx.rs b/crates/buzz-db/src/runtime/admitted_tx.rs index eb4607be257..d1e266d3fe9 100644 --- a/crates/buzz-db/src/runtime/admitted_tx.rs +++ b/crates/buzz-db/src/runtime/admitted_tx.rs @@ -24,8 +24,9 @@ use crate::Result; /// are the only ways it can end the transaction. Inside the crate, the /// `conn()` accessor hands statements the connection; the type can't stop /// crate code from issuing a raw `COMMIT` through it, so the database -/// community write fences remain the backstop. Dropping the value without -/// committing rolls back, as with [`sqlx::Transaction`]. +/// community write fences remain the backstop while deployed. Internal writers +/// must preserve this boundary and use `commit()` for staged follow-up work. +/// Dropping the value without committing rolls back, as with [`sqlx::Transaction`]. /// /// These are live regressions for the guarantee. Event-write helpers accept an /// admitted transaction: diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index d673c5d8d7e..85ce17f4798 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -110,6 +110,20 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "workflows", ]; +/// Control-plane tables with community provenance that the tenant purge does +/// not own. Keep this list separate from the expected tenant table manifest so +/// catalog validation still detects newly introduced tenant tables. +const EXCLUDED_SCOPED_TABLES: &[&str] = &[ + "community_deletion_requests", + "community_deletion_approvals", + "community_deletion_checkpoints", + "community_serving_write_leases", + "community_deletion_executor_heartbeats", + "product_feedback", + "rate_limit_violations", + "operator_listener_outbox", +]; + /// Foreign-key-safe child-before-parent order for the PostgreSQL purge. pub const PURGE_SCOPED_TABLES: &[&str] = &[ "personal_read_frontiers", @@ -370,7 +384,9 @@ pub struct SchemaManifest { pub scoped_tables: Vec, /// Per-table row counts for the target. pub row_counts: BTreeMap, - /// Sorted tables with the universal write-fence trigger. + /// Legacy fence snapshot retained for existing frozen inventory digests. + /// Populated from the live catalog while mixed-version executors need it. + #[serde(default)] pub fenced_tables: Vec, } @@ -1173,30 +1189,8 @@ impl DeletionStore { ))); } - let required_fences = EXPECTED_SCOPED_TABLES - .iter() - .copied() - .map(str::to_owned) - .collect::>(); - let live_fences = live_fenced_tables_on(&mut *conn).await?; - let missing_fences = required_fences - .difference(&live_fences) - .cloned() - .collect::>(); - if !missing_fences.is_empty() { - return Err(DbError::DeletionSafety(format!( - "community serving write fences missing: {}", - missing_fences.join(",") - ))); - } - let required_objects_present: bool = sqlx::query_scalar( "SELECT to_regprocedure('community_deletion_lock_key(uuid)') IS NOT NULL \ - AND to_regprocedure('community_write_allowed(uuid)') IS NOT NULL \ - AND (SELECT provolatile = 'v' FROM pg_proc \ - WHERE oid = 'community_write_allowed(uuid)'::regprocedure) \ - AND to_regprocedure('assert_community_write_allowed(uuid)') IS NOT NULL \ - AND to_regprocedure('enforce_community_write_fence()') IS NOT NULL \ AND EXISTS (SELECT 1 FROM pg_trigger t \ JOIN pg_class c ON c.oid = t.tgrelid \ JOIN pg_proc p ON p.oid = t.tgfoid \ @@ -1208,15 +1202,15 @@ impl DeletionStore { .await?; if !required_objects_present { return Err(DbError::DeletionSafety( - "community serving fence functions or tombstone trigger are missing".to_string(), + "community deletion lock function or tombstone trigger is missing".to_string(), )); } Ok(()) } - /// Validate the exact live scoped-table and write-fence catalog for destruction. + /// Validate the exact live scoped-table catalog for destruction. /// - /// Exact table and fence equality rejects unknown tenant data even + /// Exact table equality rejects unknown tenant data even /// while unrelated SQLx migrations continue to advance. This pool-based /// check is an early rejection only; destructive transactions revalidate /// on their own connection under [`SCHEMA_DESTRUCTION_LOCK_KEY`]. @@ -3062,24 +3056,10 @@ impl DeletionStore { let lease_seconds = i64::try_from(lease_duration.as_secs()).unwrap_or(i64::MAX); self.bounded_serving_lease_sql(SERVING_WRITE_LEASE_SQL_TIMEOUT, async |connection| { let mut tx = serving_lease_transaction(connection).await?; - // The assertion owns both the shared ordering lock and the supported - // READ COMMITTED check. The lease table is trigger-excluded, so this - // explicit admission is its database-enforced write fence. - if let Err(error) = sqlx::query("SELECT assert_community_write_allowed($1)") - .bind(community.as_uuid()) - .execute(&mut *tx) - .await - { - if error.as_database_error().is_some_and(|database_error| { - database_error.code().as_deref() == Some("55000") - && database_error.message().starts_with("community write") - }) { - return Err(DbError::AccessDenied(format!( - "community {community} is write-fenced or missing" - ))); - } - return Err(error.into()); - } + // Admit under the same shared lock as ordinary serving writes. + // The lease INSERT remains in this transaction and rechecks active + // state after admission, so a fence cannot interleave. + guard_community_write(&mut tx, community).await?; let row = sqlx::query( "INSERT INTO community_serving_write_leases \ (community_id, operation, owner, fence_generation, lease_until) \ @@ -3341,7 +3321,15 @@ pub async fn guard_community_write( tx: &mut Transaction<'_, Postgres>, community: CommunityId, ) -> Result<()> { - lock_community_deletion_shared(tx, community).await?; + let isolation = lock_community_deletion_shared(tx, community).await?; + // The lock statement may pin a snapshot in REPEATABLE READ/SERIALIZABLE. + // Refuse it before reading state; READ COMMITTED takes a fresh snapshot + // for the state query after any wait for the deletion lock. + if isolation != "read committed" { + return Err(DbError::AccessDenied(format!( + "community write admission requires read committed isolation, got {isolation}" + ))); + } let state: Option = sqlx::query_scalar( "SELECT deletion_state FROM communities WHERE id = $1 AND deleted_at IS NULL", ) @@ -3362,15 +3350,18 @@ pub async fn guard_community_write( pub(crate) async fn lock_community_deletion_shared( tx: &mut Transaction<'_, Postgres>, community: CommunityId, -) -> Result<()> { - crate::observability::observe_advisory_lock( +) -> Result { + let (_, isolation): ((), String) = crate::observability::observe_advisory_lock( crate::observability::LockType::Deletion, - sqlx::query("SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1))") - .bind(community.as_uuid()) - .execute(&mut **tx), + sqlx::query_as( + "SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1)), \ + current_setting('transaction_isolation')", + ) + .bind(community.as_uuid()) + .fetch_one(&mut **tx), ) .await?; - Ok(()) + Ok(isolation) } /// Take the shared schema/destruction advisory lock for the current @@ -3419,22 +3410,6 @@ async fn validate_catalog_on(conn: &mut PgConnection) -> Result<()> { ))); } - let fenced_tables = live_fenced_tables_on(conn).await?; - if fenced_tables != expected { - let missing = expected - .difference(&fenced_tables) - .cloned() - .collect::>(); - let unknown = fenced_tables - .difference(&expected) - .cloned() - .collect::>(); - return Err(DbError::DeletionSafety(format!( - "community deletion write-fence drift (missing={}, unknown={})", - missing.join(","), - unknown.join(",") - ))); - } Ok(()) } @@ -3450,10 +3425,11 @@ async fn live_scoped_tables_on(conn: &mut PgConnection) -> Result String { mod tests { use super::*; + #[test] + fn scoped_table_exclusions_match_sql_during_transition() { + let sql = + include_str!("../../../../schema/functions/community_write_fence_excluded_table.sql"); + let sql_names = sql + .split("ARRAY[") + .nth(1) + .expect("SQL exclusion array") + .split(']') + .next() + .expect("SQL exclusion names") + .split(',') + .map(|name| name.trim().trim_matches('\'')) + .collect::>(); + assert_eq!(sql_names.len(), 8); + assert_eq!(sql_names, EXCLUDED_SCOPED_TABLES.iter().copied().collect()); + } + fn empty_prefix(prefix: &str) -> PrefixManifest { PrefixManifest { prefix: prefix.to_string(), @@ -4381,6 +4375,140 @@ mod postgres_tests { (db, store) } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn migration_schema_fence_objects_are_optional_for_serving_and_deletion() { + let db = Db::new(&DbConfig { + database_url: crate::test_support::database_url(), + max_connections: 5, + min_connections: 0, + ..DbConfig::default() + }) + .await + .expect("connect isolated migration DB"); + crate::migration::run_migrations_through(&db.pool, 60) + .await + .expect("migrate with fence objects present"); + let store = db.deletion_store(); + let community = db + .ensure_configured_community("optional-fences.example") + .await + .expect("community") + .id; + for objects_present in [true, false] { + if !objects_present { + // Inline the contract migration: B1 ships no schema changes. + sqlx::raw_sql(r#" +DROP TRIGGER IF EXISTS community_write_fence_api_tokens ON api_tokens; +DROP TRIGGER IF EXISTS community_write_fence_archived_identities ON archived_identities; +DROP TRIGGER IF EXISTS community_write_fence_artifact_heads ON artifact_heads; +DROP TRIGGER IF EXISTS community_write_fence_artifact_revisions ON artifact_revisions; +DROP TRIGGER IF EXISTS community_write_fence_audit_log ON audit_log; +DROP TRIGGER IF EXISTS community_write_fence_channel_members ON channel_members; +DROP TRIGGER IF EXISTS community_write_fence_channels ON channels; +DROP TRIGGER IF EXISTS community_write_fence_community_bans ON community_bans; +DROP TRIGGER IF EXISTS community_write_fence_delivery_log ON delivery_log; +DROP TRIGGER IF EXISTS community_write_fence_event_mentions ON event_mentions; +DROP TRIGGER IF EXISTS community_write_fence_events ON events; +DROP TRIGGER IF EXISTS community_write_fence_git_repo_names ON git_repo_names; +DROP TRIGGER IF EXISTS community_write_fence_join_policy_acceptances ON join_policy_acceptances; +DROP TRIGGER IF EXISTS community_write_fence_moderation_actions ON moderation_actions; +DROP TRIGGER IF EXISTS community_write_fence_moderation_reports ON moderation_reports; +DROP TRIGGER IF EXISTS community_write_fence_parameterized_event_watermarks ON parameterized_event_watermarks; +DROP TRIGGER IF EXISTS community_write_fence_personal_read_accounts ON personal_read_accounts; +DROP TRIGGER IF EXISTS community_write_fence_personal_read_frontiers ON personal_read_frontiers; +DROP TRIGGER IF EXISTS community_write_fence_pubkey_allowlist ON pubkey_allowlist; +DROP TRIGGER IF EXISTS community_write_fence_push_leases ON push_leases; +DROP TRIGGER IF EXISTS community_write_fence_push_match_queue ON push_match_queue; +DROP TRIGGER IF EXISTS community_write_fence_push_wake_outbox ON push_wake_outbox; +DROP TRIGGER IF EXISTS community_write_fence_reactions ON reactions; +DROP TRIGGER IF EXISTS community_write_fence_relay_invites ON relay_invites; +DROP TRIGGER IF EXISTS community_write_fence_relay_members ON relay_members; +DROP TRIGGER IF EXISTS community_write_fence_scheduled_workflow_fires ON scheduled_workflow_fires; +DROP TRIGGER IF EXISTS community_write_fence_subscriptions ON subscriptions; +DROP TRIGGER IF EXISTS community_write_fence_thread_metadata ON thread_metadata; +DROP TRIGGER IF EXISTS community_write_fence_users ON users; +DROP TRIGGER IF EXISTS community_write_fence_workflow_approvals ON workflow_approvals; +DROP TRIGGER IF EXISTS community_write_fence_workflow_runs ON workflow_runs; +DROP TRIGGER IF EXISTS community_write_fence_workflows ON workflows; + +DROP FUNCTION attach_community_write_fence(REGCLASS); +DROP FUNCTION enforce_community_write_fence(); +DROP FUNCTION assert_community_write_allowed(UUID); +DROP FUNCTION community_write_allowed(UUID); +DROP FUNCTION community_write_fence_excluded_table(NAME); + + "#) + .execute(&db.pool) + .await + .expect("retire only universal fence objects"); + } + store + .validate_serving_catalog() + .await + .expect("serving startup"); + db.validate_deletion_serving_catalog_for_readiness( + tokio::time::Instant::now() + Duration::from_secs(5), + ) + .await + .expect("readiness"); + store.validate_catalog().await.expect("deletion catalog"); + let inventory = store.inventory_schema(community).await.expect("inventory"); + assert_eq!(inventory.scoped_tables.len(), EXPECTED_SCOPED_TABLES.len()); + assert_eq!( + inventory.fenced_tables.len(), + if objects_present { + EXPECTED_SCOPED_TABLES.len() + } else { + 0 + } + ); + let lease = store + .acquire_serving_write_lease( + community, + "optional_fences", + "test-owner", + DEFAULT_LEASE_DURATION, + ) + .await + .expect("active serving lease"); + store + .release_serving_write_lease(&lease) + .await + .expect("release lease"); + for state in ["quiescing", "fenced", "tombstone"] { + crate::test_support::set_deletion_state(&db.pool, *community.as_uuid(), state) + .await; + assert!( + matches!( + store + .acquire_serving_write_lease( + community, + "denied", + "test-owner", + DEFAULT_LEASE_DURATION + ) + .await, + Err(DbError::AccessDenied(_)) + ), + "{state} must map to AccessDenied with objects_present={objects_present}" + ); + } + assert!(matches!( + store + .acquire_serving_write_lease( + CommunityId::from_uuid(Uuid::new_v4()), + "missing", + "test-owner", + DEFAULT_LEASE_DURATION + ) + .await, + Err(DbError::AccessDenied(_)) + )); + crate::test_support::set_deletion_state(&db.pool, *community.as_uuid(), "active").await; + } + } + fn empty_prefix_manifest(prefix: String) -> PrefixManifest { PrefixManifest { prefix, @@ -7080,7 +7208,7 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] - async fn write_assertion_rejects_pinned_snapshot_isolation_before_authorization() { + async fn app_admission_rejects_pinned_snapshot_isolation_before_authorization() { let (db, _) = store().await; let community = db .ensure_configured_community(&format!( @@ -7107,17 +7235,11 @@ mod postgres_tests { .execute(&mut *tx) .await .expect("forge executor authorization"); - let error = sqlx::query("SELECT assert_community_write_allowed($1)") - .bind(community.as_uuid()) - .execute(&mut *tx) + let error = guard_community_write(&mut tx, community) .await .expect_err("pinned snapshot isolation must fail before authorization"); - assert_eq!( - error - .as_database_error() - .and_then(|error| error.code()) - .as_deref(), - Some("25000") + assert!( + matches!(error, DbError::AccessDenied(message) if message.contains("requires read committed")) ); } } diff --git a/crates/buzz-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index 4e226755ce0..c469a64c84d 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -1409,6 +1409,18 @@ where Ok(output) } +fn validate_approved_schema( + live: &buzz_db::deletion::SchemaManifest, + frozen: &FrozenInventory, +) -> Result<()> { + if live.scoped_tables != frozen.schema.scoped_tables { + return Err(permanent( + "approved structural catalog drifted before fencing", + )); + } + Ok(()) +} + async fn execute_stage( services: &Services, claim: &ClaimedDeletion, @@ -1467,13 +1479,7 @@ async fn execute_stage( .inventory_schema(request.community_id) .await?; let frozen = validate_frozen_inventory(request)?; - if live_schema.scoped_tables != frozen.schema.scoped_tables - || live_schema.fenced_tables != frozen.schema.fenced_tables - { - return Err(permanent( - "approved structural catalog drifted before fencing", - )); - } + validate_approved_schema(&live_schema, &frozen)?; services.store.begin_quiescing(&token).await?; match services.store.fence(&token).await { Ok(_) => {} @@ -2417,6 +2423,36 @@ mod postgres_tests { assert_eq!(fenced.stage, DeletionStage::Fenced); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn approved_legacy_fence_inventory_accepts_current_catalog() { + let (_, services, mut claim) = claimed_test_deletion("deletion-legacy-fences").await; + let mut frozen = validate_frozen_inventory(&claim.request).expect("frozen inventory"); + frozen.schema.fenced_tables = buzz_db::deletion::EXPECTED_SCOPED_TABLES + .iter() + .map(|table| (*table).to_owned()) + .collect(); + claim.request.inventory_digest = Some(hex::encode(frozen.digest().expect("legacy digest"))); + claim.request.inventory_manifest = + Some(serde_json::to_value(&frozen).expect("encode legacy inventory")); + + let decoded = + validate_frozen_inventory(&claim.request).expect("legacy digest remains valid"); + let mut live = services + .store + .inventory_schema(claim.request.community_id) + .await + .expect("live catalog"); + // Simulate the retired catalog while retaining the legacy digest above. + // B1 itself must still report live fences for old executors during rollout. + live.fenced_tables.clear(); + validate_approved_schema(&live, &decoded) + .expect("legacy fence list is no longer structural"); + + live.scoped_tables.push("unknown_tenant_table".to_string()); + assert!(validate_approved_schema(&live, &decoded).is_err()); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn frozen_inventory_digest_and_storage_ownership_fail_closed() { From 797181cf19255d1946568cad782f4a1fccb91412 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 11:44:46 -0400 Subject: [PATCH 2/4] test(deletion): pin lease admission to the deletion lock; log isolation refusals Add a regression proving serving-lease acquisition waits for the exclusive community deletion lock and is refused once the fence commits. Assert the quiescing refusal is AccessDenied, and log wrong-isolation refusals at error level so a configuration fault is distinguishable from a fenced community. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/store/deletion.rs | 76 ++++++++++++++++++++++++++++ crates/buzz-deletion/src/lib.rs | 23 +++++---- 2 files changed, 88 insertions(+), 11 deletions(-) diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 85ce17f4798..58ef3d9d4e0 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -3326,6 +3326,8 @@ pub async fn guard_community_write( // Refuse it before reading state; READ COMMITTED takes a fresh snapshot // for the state query after any wait for the deletion lock. if isolation != "read committed" { + tracing::error!(%isolation, %community, "community write admission requires read committed isolation"); + // AccessDenied deliberately refuses writes; the log distinguishes config errors from fences. return Err(DbError::AccessDenied(format!( "community write admission requires read committed isolation, got {isolation}" ))); @@ -7322,6 +7324,80 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn serving_lease_acquire_waits_for_fence_lock_then_refuses() { + let db = Db::new(&DbConfig { + database_url: crate::test_support::database_url(), + max_connections: 5, + min_connections: 0, + ..DbConfig::default() + }) + .await + .expect("connect isolated test DB"); + let community = db + .ensure_configured_community("lease-fence-lock.example") + .await + .expect("active community") + .id; + let mut fence_tx = db.pool.begin().await.expect("begin fence transaction"); + sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))") + .bind(community.as_uuid()) + .execute(&mut *fence_tx) + .await + .expect("hold exclusive deletion lock"); + + let store = db.deletion_store(); + let mut acquisition = tokio::spawn(async move { + store + .acquire_serving_write_lease( + community, + "fence_lock_test", + "test-owner", + DEFAULT_LEASE_DURATION, + ) + .await + }); + assert!( + tokio::time::timeout(Duration::from_millis(250), &mut acquisition) + .await + .is_err(), + "lease acquisition must wait for the exclusive deletion lock" + ); + + sqlx::query( + "SELECT set_config('buzz.deletion_executor_community', $1, true), \ + set_config('buzz.deletion_fence_generation', \ + (SELECT deletion_fence_generation::text FROM communities WHERE id = $2), true)", + ) + .bind(community.to_string()) + .bind(community.as_uuid()) + .execute(&mut *fence_tx) + .await + .expect("authorize fence transition"); + sqlx::query("UPDATE communities SET deletion_state = 'fenced' WHERE id = $1") + .bind(community.as_uuid()) + .execute(&mut *fence_tx) + .await + .expect("fence community while holding the lock"); + fence_tx.commit().await.expect("commit fence"); + + let error = tokio::time::timeout(Duration::from_secs(5), acquisition) + .await + .expect("acquisition completes after fence commit") + .expect("acquisition task") + .expect_err("fenced community refuses a new lease"); + assert!(matches!(error, DbError::AccessDenied(_)), "{error:?}"); + let leases: i64 = sqlx::query_scalar( + "SELECT count(*) FROM community_serving_write_leases WHERE community_id = $1", + ) + .bind(community.as_uuid()) + .fetch_one(&db.pool) + .await + .expect("count serving leases"); + assert_eq!(leases, 0, "refused acquisition must not persist a lease"); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn quiescing_rejects_new_leases_but_renews_admitted_lease_until_release() { diff --git a/crates/buzz-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index c469a64c84d..5fd9d31e2c3 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -2822,18 +2822,19 @@ mod postgres_tests { .begin_quiescing(&claim.lease) .await .expect("quiesce"); + let error = services + .store + .acquire_serving_write_lease( + community, + "late_external", + "late-owner", + DEFAULT_LEASE_DURATION, + ) + .await + .expect_err("quiescing must reject newly admitted work"); assert!( - services - .store - .acquire_serving_write_lease( - community, - "late_external", - "late-owner", - DEFAULT_LEASE_DURATION, - ) - .await - .is_err(), - "quiescing must reject newly admitted work" + matches!(error, buzz_db::DbError::AccessDenied(_)), + "{error:?}" ); let completed = Arc::new(AtomicBool::new(false)); From 6e69c12abe60e151a07af31f78049917450f755e Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 09:57:30 -0400 Subject: [PATCH 3/4] fix(deletion): refuse pinned snapshots in every lock-then-read admission Move the READ COMMITTED check from guard_community_write into lock_community_deletion_shared, so serving-lease finalization, lease renewal/verification, owner preparation and owner-mutation admission get the same refusal. Today the fence trigger backs those paths up; after the drop only the pool default would. The isolation test now covers guard_transaction_with_serving_lease and drops executor-GUC setup that guard_community_write no longer reads. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/store/deletion.rs | 93 +++++++++++++++++----------- crates/buzz-deletion/src/lib.rs | 3 +- 2 files changed, 59 insertions(+), 37 deletions(-) diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 58ef3d9d4e0..a81a58ceefe 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -3321,17 +3321,7 @@ pub async fn guard_community_write( tx: &mut Transaction<'_, Postgres>, community: CommunityId, ) -> Result<()> { - let isolation = lock_community_deletion_shared(tx, community).await?; - // The lock statement may pin a snapshot in REPEATABLE READ/SERIALIZABLE. - // Refuse it before reading state; READ COMMITTED takes a fresh snapshot - // for the state query after any wait for the deletion lock. - if isolation != "read committed" { - tracing::error!(%isolation, %community, "community write admission requires read committed isolation"); - // AccessDenied deliberately refuses writes; the log distinguishes config errors from fences. - return Err(DbError::AccessDenied(format!( - "community write admission requires read committed isolation, got {isolation}" - ))); - } + lock_community_deletion_shared(tx, community).await?; let state: Option = sqlx::query_scalar( "SELECT deletion_state FROM communities WHERE id = $1 AND deleted_at IS NULL", ) @@ -3349,10 +3339,18 @@ pub async fn guard_community_write( } } +/// Take the shared community deletion lock before an admission read. +/// +/// Every caller reads lifecycle state after this returns, so the transaction +/// must be READ COMMITTED: the lock statement may pin a snapshot under +/// REPEATABLE READ/SERIALIZABLE, and a later state read would then miss a +/// fence committed while this transaction waited for the lock. The isolation +/// level is read in the same statement as the lock and refused as +/// [`DbError::AccessDenied`]. pub(crate) async fn lock_community_deletion_shared( tx: &mut Transaction<'_, Postgres>, community: CommunityId, -) -> Result { +) -> Result<()> { let (_, isolation): ((), String) = crate::observability::observe_advisory_lock( crate::observability::LockType::Deletion, sqlx::query_as( @@ -3363,7 +3361,14 @@ pub(crate) async fn lock_community_deletion_shared( .fetch_one(&mut **tx), ) .await?; - Ok(isolation) + if isolation != "read committed" { + tracing::error!(%isolation, %community, "community write admission requires read committed isolation"); + // AccessDenied deliberately refuses writes; the log distinguishes config errors from fences. + return Err(DbError::AccessDenied(format!( + "community write admission requires read committed isolation, got {isolation}" + ))); + } + Ok(()) } /// Take the shared schema/destruction advisory lock for the current @@ -4399,7 +4404,8 @@ mod postgres_tests { .id; for objects_present in [true, false] { if !objects_present { - // Inline the contract migration: B1 ships no schema changes. + // Drop the fence objects by hand, as the retirement migration + // will; this change itself ships no schema changes. sqlx::raw_sql(r#" DROP TRIGGER IF EXISTS community_write_fence_api_tokens ON api_tokens; DROP TRIGGER IF EXISTS community_write_fence_archived_identities ON archived_identities; @@ -7210,8 +7216,8 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); #[tokio::test] #[ignore = "requires Postgres"] - async fn app_admission_rejects_pinned_snapshot_isolation_before_authorization() { - let (db, _) = store().await; + async fn deletion_lock_admission_rejects_pinned_snapshot_isolation() { + let (db, store) = store().await; let community = db .ensure_configured_community(&format!( "isolation-guard-{}.example", @@ -7220,29 +7226,44 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .await .expect("create community") .id; + // Never acquired: under READ COMMITTED this lease is refused as stale, + // so a stale-lease error here means the isolation check did not run. + let lease = ServingWriteLease { + id: Uuid::new_v4(), + community_id: community, + operation: "isolation-probe".to_owned(), + owner: "isolation-probe".to_owned(), + generation: 0, + fence_generation: 0, + lease_until: Utc::now(), + }; for isolation in ["REPEATABLE READ", "SERIALIZABLE"] { - let mut tx = db.pool.begin().await.expect("begin isolation probe"); - sqlx::query(sqlx::AssertSqlSafe(format!( - "SET TRANSACTION ISOLATION LEVEL {isolation}" - ))) - .execute(&mut *tx) - .await - .expect("set transaction isolation"); - sqlx::query( - "SELECT set_config('buzz.deletion_executor_community', $1, true), \ - set_config('buzz.deletion_fence_generation', '0', true)", - ) - .bind(community.to_string()) - .execute(&mut *tx) - .await - .expect("forge executor authorization"); - let error = guard_community_write(&mut tx, community) + for path in [ + "guard_community_write", + "guard_transaction_with_serving_lease", + ] { + let mut tx = db.pool.begin().await.expect("begin isolation probe"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "SET TRANSACTION ISOLATION LEVEL {isolation}" + ))) + .execute(&mut *tx) .await - .expect_err("pinned snapshot isolation must fail before authorization"); - assert!( - matches!(error, DbError::AccessDenied(message) if message.contains("requires read committed")) - ); + .expect("set transaction isolation"); + let error = match path { + "guard_community_write" => guard_community_write(&mut tx, community).await, + _ => { + store + .guard_transaction_with_serving_lease(&mut tx, &lease) + .await + } + } + .expect_err("pinned snapshot isolation must be refused"); + assert!( + matches!(&error, DbError::AccessDenied(message) if message.contains("requires read committed")), + "{path} under {isolation}: {error:?}" + ); + } } } diff --git a/crates/buzz-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index 5fd9d31e2c3..0e63f09b020 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -2444,7 +2444,8 @@ mod postgres_tests { .await .expect("live catalog"); // Simulate the retired catalog while retaining the legacy digest above. - // B1 itself must still report live fences for old executors during rollout. + // While the fence objects exist, inventory still reports them so + // executors that predate fence-optional catalogs keep validating. live.fenced_tables.clear(); validate_approved_schema(&live, &decoded) .expect("legacy fence list is no longer structural"); From 0c7681a25cf89d335dfba7a680becf054cee6637 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 10:21:23 -0400 Subject: [PATCH 4/4] buzz-db: name deletion-lock admission in the isolation refusal Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/store/deletion.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index a81a58ceefe..387015cea7d 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -3362,10 +3362,10 @@ pub(crate) async fn lock_community_deletion_shared( ) .await?; if isolation != "read committed" { - tracing::error!(%isolation, %community, "community write admission requires read committed isolation"); + tracing::error!(%isolation, %community, "deletion-lock admission requires read committed isolation"); // AccessDenied deliberately refuses writes; the log distinguishes config errors from fences. return Err(DbError::AccessDenied(format!( - "community write admission requires read committed isolation, got {isolation}" + "deletion-lock admission requires read committed isolation, got {isolation}" ))); } Ok(())