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] 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 e0875618dd..127a5492f5 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 eb4607be25..d1e266d3fe 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 d673c5d8d7..85ce17f479 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 4e226755ce..c469a64c84 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() {