From b44e79b7fa26b946016bcab53cb6731be2148078 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 10:12:57 -0400 Subject: [PATCH 1/8] buzz-db: enforce five single-table guards in app writers Phase 1 of BUZZ-318. The five single-table guard triggers stay for mixed-version pods; their invariants now also hold in the application writers, so a later PR can drop them. - communities lifecycle: begin_quiescing, fence, purge_postgres and abort go through one transition_community state machine over a locked snapshot (deleted_at one-way, generation monotonic, deleted rows rejected). set_community_icon only updates an active, undeleted row. - community_deletion_requests: both initial inventory freezes require an unfrozen, unblocked submitted row in the UPDATE predicate; the destructive-manifest freeze refuses a frozen row with missing JSON. - community_deletion_manifest_keys: append, pre-freeze clear and verified cleanup check the request row via guard_manifest_chunk_write. - channels.community_id and community_deletion_approvals hold by construction; a Rust source policy in observability_source.rs forbids retenanting, lifecycle writes outside the transition helper, community DELETE, request retargeting, approval UPDATE/DELETE and unguarded manifest writers. Five PG tests each disable only their own trigger and exercise the app path's rejected and permitted mutations. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/store/community.rs | 12 +- crates/buzz-db/src/store/deletion.rs | 731 +++++++++++++++---- crates/buzz-db/tests/observability_source.rs | 430 +++++++++-- 3 files changed, 1006 insertions(+), 167 deletions(-) diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index e084d8ed176..dcaabdf1bb7 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -426,17 +426,23 @@ impl Db { crate::observability::WriterOperation::EventWrite, ) .await?; - sqlx::query( + let affected = sqlx::query( r#" UPDATE communities SET icon = $2 - WHERE id = $1 + WHERE id = $1 AND deletion_state = 'active' AND deleted_at IS NULL "#, ) .bind(community_id.as_uuid()) .bind(icon) .execute(&mut *connection) - .await?; + .await? + .rows_affected(); + if affected != 1 { + return Err(DbError::AccessDenied(format!( + "community {community_id} is missing or write-fenced" + ))); + } Ok(()) } diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index d673c5d8d7e..83757e548f5 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -1278,6 +1278,7 @@ impl DeletionStore { inventory_frozen_at = now(), updated_at = now(), last_error = NULL, last_error_at = NULL WHERE id = $1 AND stage = 'submitted' AND blocked_at IS NULL + AND inventory_frozen_at IS NULL RETURNING * "#, ) @@ -1290,7 +1291,7 @@ impl DeletionStore { .await? .ok_or_else(|| { DbError::DeletionSafety(format!( - "deletion {request_id} is not an unblocked submitted request" + "deletion {request_id} is not an unfrozen, unblocked submitted request" )) })?; row_to_request(row) @@ -1586,11 +1587,12 @@ impl DeletionStore { token.request_id )) })?; - sqlx::query( + let affected = sqlx::query( "UPDATE community_deletion_requests SET stage = 'inventoried', \ schema_manifest = $2, storage_manifest = $3, inventory_manifest = $4, \ inventory_digest = $5, inventory_frozen_at = now(), updated_at = now() \ - WHERE id = $1", + WHERE id = $1 AND stage = 'submitted' AND blocked_at IS NULL \ + AND inventory_frozen_at IS NULL", ) .bind(token.request_id) .bind(schema) @@ -1598,7 +1600,14 @@ impl DeletionStore { .bind(frozen) .bind(&digest) .execute(&mut *tx) - .await?; + .await? + .rows_affected(); + if affected != 1 { + return Err(DbError::DeletionSafety(format!( + "deletion {} is not an unfrozen, unblocked submitted request", + token.request_id + ))); + } sqlx::query( "INSERT INTO community_deletion_approvals \ (request_id, community_id, inventory_digest, approved_by, approval_origin) \ @@ -1802,52 +1811,37 @@ impl DeletionStore { .await?; transaction_timer .observe(async { - verify_lease(&mut tx, token, DeletionStage::Approved).await?; - lock_community_deletion(&mut tx, token.community_id).await?; - verify_lease(&mut tx, token, DeletionStage::Approved).await?; - let (generation, archived_at): (i64, Option>) = sqlx::query_as( - "SELECT deletion_fence_generation, archived_at FROM communities WHERE id = $1 FOR UPDATE", - ) - .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, \ + verify_lease(&mut tx, token, DeletionStage::Approved).await?; + lock_community_deletion(&mut tx, token.community_id).await?; + verify_lease(&mut tx, token, DeletionStage::Approved).await?; + let community = lock_community_lifecycle(&mut tx, token.community_id).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() \ WHERE id = $1 AND quiescing_started_at IS NULL", - ) - .bind(token.request_id) - .bind(archived_at) - .execute(&mut *tx) - .await?; - set_executor_gucs(&mut tx, token.community_id, generation).await?; - let affected = sqlx::query( - "UPDATE communities SET deletion_state = 'quiescing', \ - archived_at = COALESCE(archived_at, now()) \ - WHERE id = $1 AND deletion_state IN ('active', 'quiescing') \ - AND deleted_at IS NULL", - ) - .bind(token.community_id.as_uuid()) - .execute(&mut *tx) - .await? - .rows_affected(); - if affected != 1 { - return Err(DbError::DeletionSafety(format!( - "community {} cannot enter quiescing", - token.community_id - ))); - } - checkpoint_completed_tx( - &mut tx, - token, - DeletionStage::Approved, - "quiesce_serving_writes", - serde_json::json!({"community_state": "quiescing"}), - ) - .await?; - tx.commit().await?; - Ok(()) + ) + .bind(token.request_id) + .bind(community.archived_at) + .execute(&mut *tx) + .await?; + transition_community( + &mut tx, + token.community_id, + community, + CommunityTransition::Quiesce, + ) + .await?; + checkpoint_completed_tx( + &mut tx, + token, + DeletionStage::Approved, + "quiesce_serving_writes", + serde_json::json!({"community_state": "quiescing"}), + ) + .await?; + tx.commit().await?; + Ok(()) }) .await } @@ -1881,32 +1875,13 @@ impl DeletionStore { operations: active_serving_writes.try_get("operations")?, }); } - let current_generation: i64 = sqlx::query_scalar( - "SELECT deletion_fence_generation FROM communities WHERE id = $1 FOR UPDATE", - ) - .bind(token.community_id.as_uuid()) - .fetch_one(&mut *tx) - .await?; - let generation = current_generation.checked_add(1).ok_or_else(|| { - DbError::DeletionSafety("community deletion fence generation overflow".to_string()) - })?; - set_executor_gucs(&mut tx, token.community_id, generation).await?; - let affected = sqlx::query( - "UPDATE communities SET deletion_state = 'fenced', \ - deletion_fence_generation = $2, archived_at = COALESCE(archived_at, now()) \ - WHERE id = $1 AND deletion_state = 'quiescing'", - ) - .bind(token.community_id.as_uuid()) - .bind(generation) - .execute(&mut *tx) - .await? - .rows_affected(); - if affected != 1 { - return Err(DbError::DeletionSafety(format!( - "community {} is no longer quiescing while fencing", - token.community_id - ))); - } + let community = lock_community_lifecycle(&mut tx, token.community_id).await?; + let generation = transition_community( + &mut tx, + token.community_id, + community, + CommunityTransition::Fence, + ).await?; advance_request_tx( &mut tx, token, @@ -1972,6 +1947,8 @@ impl DeletionStore { AND stage = 'fenced' \ AND (destructive_storage_manifest IS NULL \ OR destructive_storage_manifest = $4) \ + AND (destructive_storage_frozen_at IS NULL \ + OR destructive_storage_manifest = $4) \ RETURNING id", ) .bind(token.request_id) @@ -1999,6 +1976,7 @@ impl DeletionStore { let generation = require_fence_generation(token)?; let mut tx = self.pool.begin().await?; verify_lease_and_fence(&mut tx, token, DeletionStage::Fenced, generation).await?; + guard_manifest_chunk_write(&mut tx, token, ManifestChunkWrite::Unfrozen).await?; sqlx::query("DELETE FROM community_deletion_manifest_keys WHERE request_id = $1") .bind(token.request_id) .execute(&mut *tx) @@ -2023,6 +2001,7 @@ impl DeletionStore { let generation = require_fence_generation(token)?; let mut tx = self.pool.begin().await?; verify_lease_and_fence(&mut tx, token, DeletionStage::Fenced, generation).await?; + guard_manifest_chunk_write(&mut tx, token, ManifestChunkWrite::Unfrozen).await?; sqlx::query( "INSERT INTO community_deletion_manifest_keys \ (request_id, chunk_no, prefix, keys) VALUES ($1, $2, $3, $4)", @@ -2319,25 +2298,14 @@ impl DeletionStore { .await?; } - let affected = sqlx::query( - "UPDATE communities SET deletion_state = 'tombstone', \ - deleted_at = COALESCE(deleted_at, now()), \ - archived_at = COALESCE(archived_at, now()), \ - signing_key = NULL, icon = NULL \ - WHERE id = $1 AND deletion_state = 'fenced' \ - AND deletion_fence_generation = $2", + let community = lock_community_lifecycle(&mut tx, token.community_id).await?; + transition_community( + &mut tx, + token.community_id, + community, + CommunityTransition::Tombstone { generation }, ) - .bind(token.community_id.as_uuid()) - .bind(generation) - .execute(&mut *tx) - .await? - .rows_affected(); - if affected != 1 { - return Err(DbError::DeletionSafety(format!( - "community {} tombstone update affected {affected} rows", - token.community_id - ))); - } + .await?; advance_request_tx( &mut tx, token, @@ -2458,6 +2426,7 @@ impl DeletionStore { detail, ) .await?; + guard_manifest_chunk_write(&mut tx, token, ManifestChunkWrite::VerifiedCleanup).await?; sqlx::query("DELETE FROM community_deletion_manifest_keys WHERE request_id = $1") .bind(token.request_id) .execute(&mut *tx) @@ -2674,12 +2643,7 @@ impl DeletionStore { // 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 community = lock_community_lifecycle(&mut tx, community_id).await?; let row = sqlx::query("SELECT * FROM community_deletion_requests WHERE id = $1 FOR UPDATE") .bind(request_id) .fetch_optional(&mut *tx) @@ -2715,32 +2679,21 @@ impl DeletionStore { "deletion {request_id} cannot abort while {active_writes} serving write lease(s) remain active" ))); } - let new_generation = old_generation.checked_add(1).ok_or_else(|| { - DbError::DeletionSafety("community deletion fence generation overflow".to_string()) - })?; let restored_archived_at = if request.quiescing_started_at.is_some() { request.pre_quiesce_archived_at } else { - current_archived_at + community.archived_at }; - set_executor_gucs(&mut tx, request.community_id, new_generation).await?; - let restored = sqlx::query( - "UPDATE communities SET deletion_state = 'active', deletion_fence_generation = $2, \ - archived_at = $3 WHERE id = $1 AND deletion_state IN ('active', 'quiescing', 'fenced') \ - AND deleted_at IS NULL", + let old_generation = community.generation; + let new_generation = transition_community( + &mut tx, + request.community_id, + community, + CommunityTransition::Abort { + restored_archived_at, + }, ) - .bind(request.community_id.as_uuid()) - .bind(new_generation) - .bind(restored_archived_at) - .execute(&mut *tx) - .await? - .rows_affected(); - if restored != 1 { - return Err(DbError::DeletionSafety(format!( - "community {} cannot be restored during abort", - request.community_id - ))); - } + .await?; sqlx::query( "INSERT INTO community_deletion_checkpoints \ (request_id, stage, unit_key, status, lease_generation, detail, completed_at) \ @@ -3315,6 +3268,43 @@ impl DeletionStore { } } +enum ManifestChunkWrite { + Unfrozen, + VerifiedCleanup, +} + +/// Serialize chunk writes with manifest freeze on the request row. Both chunk +/// insertion and pre-freeze replacement use this check; terminal cleanup is +/// admitted only after the logical-verification stage is persisted in the same +/// transaction. +async fn guard_manifest_chunk_write( + tx: &mut Transaction<'_, Postgres>, + token: &LeaseToken, + operation: ManifestChunkWrite, +) -> Result<()> { + let row: Option<(String, Option>)> = sqlx::query_as( + "SELECT stage, destructive_storage_frozen_at FROM community_deletion_requests \ + WHERE id = $1 FOR UPDATE", + ) + .bind(token.request_id) + .fetch_optional(&mut **tx) + .await?; + let allowed = match (operation, row) { + (ManifestChunkWrite::Unfrozen, Some((stage, None))) => stage == "fenced", + (ManifestChunkWrite::VerifiedCleanup, Some((stage, _))) => { + matches!(stage.as_str(), "logically_verified" | "retention_pending") + } + _ => false, + }; + if !allowed { + return Err(DbError::DeletionSafety(format!( + "manifest key chunks are frozen or in the wrong stage for deletion {}", + token.request_id + ))); + } + Ok(()) +} + async fn lock_community_deletion( tx: &mut Transaction<'_, Postgres>, community: CommunityId, @@ -3329,6 +3319,123 @@ async fn lock_community_deletion( Ok(()) } +/// The row snapshot is taken under `FOR UPDATE` after the community deletion +/// advisory lock. All lifecycle writers use it in the same transaction. +struct CommunityLifecycle { + state: String, + generation: i64, + deleted_at: Option>, + archived_at: Option>, +} + +enum CommunityTransition { + Quiesce, + Fence, + Tombstone { + generation: i64, + }, + Abort { + restored_archived_at: Option>, + }, +} + +async fn lock_community_lifecycle( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, +) -> Result { + let row = sqlx::query( + "SELECT deletion_state, deletion_fence_generation, deleted_at, archived_at \ + FROM communities WHERE id = $1 FOR UPDATE", + ) + .bind(community.as_uuid()) + .fetch_optional(&mut **tx) + .await? + .ok_or_else(|| DbError::DeletionSafety(format!("community {community} is missing")))?; + Ok(CommunityLifecycle { + state: row.try_get("deletion_state")?, + generation: row.try_get("deletion_fence_generation")?, + deleted_at: row.try_get("deleted_at")?, + archived_at: row.try_get("archived_at")?, + }) +} + +/// The sole application writer of community deletion state. The executor GUCs +/// remain for mixed-version database guards, while this state machine rejects +/// invalid transitions before issuing the UPDATE. +async fn transition_community( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + old: CommunityLifecycle, + transition: CommunityTransition, +) -> Result { + if old.deleted_at.is_some() { + return Err(DbError::DeletionSafety(format!( + "community {community} is permanently tombstoned" + ))); + } + let (new_state, generation, restore_archive, tombstone) = match transition { + CommunityTransition::Quiesce if matches!(old.state.as_str(), "active" | "quiescing") => { + ("quiescing", old.generation, None, false) + } + CommunityTransition::Fence if old.state == "quiescing" => ( + "fenced", + old.generation.checked_add(1).ok_or_else(|| { + DbError::DeletionSafety("community deletion fence generation overflow".to_string()) + })?, + None, + false, + ), + CommunityTransition::Tombstone { generation } + if old.state == "fenced" && old.generation == generation => + { + ("tombstone", generation, None, true) + } + CommunityTransition::Abort { + restored_archived_at, + } if matches!(old.state.as_str(), "active" | "quiescing" | "fenced") => ( + "active", + old.generation.checked_add(1).ok_or_else(|| { + DbError::DeletionSafety("community deletion fence generation overflow".to_string()) + })?, + restored_archived_at, + false, + ), + _ => { + return Err(DbError::DeletionSafety(format!( + "community {community} cannot make this deletion transition from {} generation {}", + old.state, old.generation + ))); + } + }; + set_executor_gucs(tx, community, generation).await?; + let affected = sqlx::query( + "UPDATE communities SET deletion_state = $2, deletion_fence_generation = $3, \ + archived_at = CASE WHEN $4 THEN COALESCE(archived_at, now()) ELSE $5 END, \ + deleted_at = CASE WHEN $6 THEN now() ELSE deleted_at END, \ + signing_key = CASE WHEN $6 THEN NULL ELSE signing_key END, \ + icon = CASE WHEN $6 THEN NULL ELSE icon END \ + WHERE id = $1 AND deletion_state = $7 AND deletion_fence_generation = $8 \ + AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(new_state) + .bind(generation) + .bind(!matches!(transition, CommunityTransition::Abort { .. })) + .bind(restore_archive) + .bind(tombstone) + .bind(&old.state) + .bind(old.generation) + .execute(&mut **tx) + .await? + .rows_affected(); + if affected != 1 { + return Err(DbError::DeletionSafety(format!( + "community {community} changed during deletion transition" + ))); + } + Ok(generation) +} + /// Admit a community write inside an existing transaction. /// /// Takes the shared community deletion lock, then requires the community to @@ -4363,11 +4470,8 @@ mod postgres_tests { use std::str::FromStr; async fn store() -> (Db, DeletionStore) { - let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") - .or_else(|_| std::env::var("DATABASE_URL")) - .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); // sadscan:disable np.postgres.1 -- local test-only credentials let db = Db::new(&DbConfig { - database_url, + database_url: crate::test_support::database_url(), max_connections: 5, min_connections: 0, ..DbConfig::default() @@ -4446,6 +4550,387 @@ mod postgres_tests { (host, owner, created.id) } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn single_table_guard_channels_remain_tenant_bound_without_trigger() { + let (db, _) = store().await; + let first = db + .ensure_configured_community(&format!("channel-a-{}.example", Uuid::new_v4())) + .await + .expect("first community"); + let second = db + .ensure_configured_community(&format!("channel-b-{}.example", Uuid::new_v4())) + .await + .expect("second community"); + let channel = db + .create_channel( + first.id, + "original", + crate::channel::ChannelType::Stream, + crate::channel::ChannelVisibility::Open, + None, + &[1; 32], + None, + ) + .await + .expect("create first-community channel"); + sqlx::query("ALTER TABLE channels DISABLE TRIGGER trg_channels_community_id_immutable") + .execute(&db.pool) + .await + .expect("disable only channel immutability trigger"); + assert!(matches!( + db.update_channel( + second.id, + channel.id, + crate::channel::ChannelUpdate { + name: Some("wrong-community".to_string()), + ..Default::default() + }, + ) + .await, + Err(DbError::ChannelNotFound(_)) + )); + db.update_channel( + first.id, + channel.id, + crate::channel::ChannelUpdate { + name: Some("same-community".to_string()), + ..Default::default() + }, + ) + .await + .expect("same-community metadata update"); + let owner: Uuid = sqlx::query_scalar("SELECT community_id FROM channels WHERE id = $1") + .bind(channel.id) + .fetch_one(&db.pool) + .await + .expect("channel owner"); + assert_eq!(owner, *first.id.as_uuid()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn single_table_guard_community_lifecycle_rejects_deleted_row_without_trigger() { + let (db, store) = store().await; + let (request, inventory) = inventoried_request(&db, &store).await; + store + .approve(request.id, "approver", None) + .await + .expect("approve"); + db.set_community_icon(request.community_id, Some("active-icon")) + .await + .expect("active community icon update"); + 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"); + sqlx::query("ALTER TABLE communities DISABLE TRIGGER communities_deletion_tombstone") + .execute(&db.pool) + .await + .expect("disable only tombstone trigger"); + assert!(matches!( + db.set_community_icon(request.community_id, Some("quiescing-icon")) + .await, + Err(DbError::AccessDenied(_)) + )); + sqlx::query("UPDATE communities SET deleted_at = now() WHERE id = $1") + .bind(request.community_id.as_uuid()) + .execute(&db.pool) + .await + .expect("simulate malformed deleted row"); + assert!(matches!( + store.fence(&claim.lease).await, + Err(DbError::DeletionSafety(message)) if message.contains("permanently tombstoned") + )); + sqlx::query("UPDATE communities SET deleted_at = NULL WHERE id = $1") + .bind(request.community_id.as_uuid()) + .execute(&db.pool) + .await + .expect("repair test fixture"); + let generation = store.fence(&claim.lease).await.expect("valid fence"); + let token = LeaseToken { + fence_generation: Some(generation), + ..claim.lease + }; + store + .freeze_destructive_storage_manifest(&token, &inventory.storage) + .await + .expect("freeze empty manifest"); + store.mark_drained(&token).await.expect("drain"); + store + .mark_bindings_removed(&token, serde_json::json!({"keys": 0})) + .await + .expect("bindings removed"); + store.purge_postgres(&token).await.expect("tombstone"); + let (state, stored_generation, deleted_at): (String, i64, Option>) = + sqlx::query_as( + "SELECT deletion_state, deletion_fence_generation, deleted_at FROM communities WHERE id = $1", + ) + .bind(request.community_id.as_uuid()) + .fetch_one(&db.pool) + .await + .expect("tombstone row"); + assert_eq!(state, "tombstone"); + assert_eq!(stored_generation, generation); + assert!(deleted_at.is_some()); + assert!(store + .abort(request.id, "operator", "too late") + .await + .is_err()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn single_table_guard_request_inventory_freeze_is_one_way_without_trigger() { + let (db, store) = store().await; + let host = format!("request-{}.example", Uuid::new_v4()); + let community = db + .ensure_configured_community(&host) + .await + .expect("community"); + let request = store.submit(&host, "operator", None).await.expect("submit"); + let inventory = FrozenInventory { + schema: store + .inventory_schema(community.id) + .await + .expect("schema inventory"), + storage: empty_storage_manifest(community.id), + }; + sqlx::query( + "ALTER TABLE community_deletion_requests DISABLE TRIGGER community_deletion_request_retargeting_guard", + ) + .execute(&db.pool) + .await + .expect("disable only request guard"); + sqlx::query( + "UPDATE community_deletion_requests SET inventory_frozen_at = now(), \ + inventory_digest = decode(repeat('00', 32), 'hex') WHERE id = $1", + ) + .bind(request.id) + .execute(&db.pool) + .await + .expect("simulate earlier inventory freeze"); + assert!(matches!( + store.freeze_inventory(request.id, &inventory).await, + Err(DbError::DeletionSafety(message)) if message.contains("unfrozen") + )); + sqlx::query( + "UPDATE community_deletion_requests SET inventory_frozen_at = NULL, \ + inventory_digest = NULL WHERE id = $1", + ) + .bind(request.id) + .execute(&db.pool) + .await + .expect("repair test fixture"); + store + .freeze_inventory(request.id, &inventory) + .await + .expect("first freeze"); + assert_eq!( + store.get(request.id).await.expect("request").community_id, + community.id + ); + let (owner_host, owner, owner_community) = archived_owned_community(&db).await; + let (other_host, other_owner, _) = archived_owned_community(&db).await; + let operator = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + let owner_request_id = Uuid::new_v4(); + assert!(matches!( + store + .admit_owner_request(&owner_host, &owner, operator, 1, owner_request_id, None) + .await + .expect("owner request"), + OwnerDeletionAdmission::Accepted(_) + )); + assert_eq!( + store + .admit_owner_request( + &other_host, + &other_owner, + operator, + 1, + owner_request_id, + None + ) + .await + .expect("retarget attempt"), + OwnerDeletionAdmission::RequestConflict + ); + + let owner_claim = store + .claim_specific_owner_submission(owner_request_id, "preparer", DEFAULT_LEASE_DURATION) + .await + .expect("claim owner preparation") + .expect("owner request is preparable"); + let owner_inventory = FrozenInventory { + schema: store + .inventory_schema(owner_community) + .await + .expect("owner schema inventory"), + storage: empty_storage_manifest(owner_community), + }; + sqlx::query( + "UPDATE community_deletion_requests SET inventory_frozen_at = now(), \ + inventory_digest = decode(repeat('00', 32), 'hex') WHERE id = $1", + ) + .bind(owner_request_id) + .execute(&db.pool) + .await + .expect("simulate earlier owner inventory freeze"); + assert!(matches!( + store.complete_owner_preparation(&owner_claim.lease, &owner_inventory).await, + Err(DbError::DeletionSafety(message)) if message.contains("unfrozen") + )); + let owner_request = store + .get(owner_request_id) + .await + .expect("unchanged owner request"); + assert_eq!(owner_request.stage, DeletionStage::Submitted); + assert_eq!(owner_request.inventory_digest, Some("00".repeat(32))); + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM community_deletion_approvals WHERE request_id = $1", + ) + .bind(owner_request_id) + .fetch_one(&db.pool) + .await + .expect("no approval after rejected owner freeze"), + 0 + ); + + 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"); + let generation = store.fence(&claim.lease).await.expect("fence"); + let token = LeaseToken { + fence_generation: Some(generation), + ..claim.lease + }; + sqlx::query( + "UPDATE community_deletion_requests SET destructive_storage_frozen_at = now() \ + WHERE id = $1", + ) + .bind(request.id) + .execute(&db.pool) + .await + .expect("simulate frozen manifest with missing JSON"); + assert!(matches!( + store + .freeze_destructive_storage_manifest(&token, &inventory.storage) + .await, + Err(DbError::DeletionSafety(message)) if message.contains("manifest changed") + )); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn single_table_guard_approval_evidence_is_append_only_without_trigger() { + let (db, store) = store().await; + let (request, _) = inventoried_request(&db, &store).await; + sqlx::query( + "ALTER TABLE community_deletion_approvals DISABLE TRIGGER community_deletion_approval_removal_guard", + ) + .execute(&db.pool) + .await + .expect("disable only approval guard"); + store + .approve(request.id, "first", None) + .await + .expect("first approval"); + assert!(matches!( + store.approve(request.id, "second", None).await, + Err(DbError::DeletionSafety(message)) + if message == format!("deletion {} is not an unblocked inventoried request", request.id) + )); + let approved_by: String = sqlx::query_scalar( + "SELECT approved_by FROM community_deletion_approvals WHERE request_id = $1", + ) + .bind(request.id) + .fetch_one(&db.pool) + .await + .expect("retained approval"); + assert_eq!(approved_by, "first"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn single_table_guard_manifest_chunks_stop_at_freeze_without_trigger() { + let (db, store) = store().await; + let (request, inventory) = 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"); + let generation = store.fence(&claim.lease).await.expect("fence"); + let token = LeaseToken { + fence_generation: Some(generation), + ..claim.lease + }; + let prefix = format!("_meta/{}/", request.community_id); + let key = format!("{prefix}{}.json", "a".repeat(64)); + store + .append_manifest_key_chunk(&token, 0, &prefix, std::slice::from_ref(&key)) + .await + .expect("pre-freeze append"); + let mut digest = KeyStreamDigest::new(); + digest.fold(&key).expect("fold key"); + let (hash, count) = digest.finish(); + let mut manifest = inventory.storage; + manifest.prefixes[0].object_count = count; + manifest.prefixes[0].keys_digest = hash; + store + .freeze_destructive_storage_manifest(&token, &manifest) + .await + .expect("freeze matching manifest"); + sqlx::query( + "ALTER TABLE community_deletion_manifest_keys DISABLE TRIGGER community_deletion_manifest_keys_guard", + ) + .execute(&db.pool) + .await + .expect("disable only manifest key guard"); + assert!(matches!( + store.append_manifest_key_chunk(&token, 1, &prefix, std::slice::from_ref(&key)).await, + Err(DbError::DeletionSafety(message)) if message.contains("frozen") + )); + assert!(matches!( + store.clear_manifest_key_chunks(&token).await, + Err(DbError::DeletionSafety(message)) if message.contains("frozen") + )); + assert_eq!( + store + .manifest_chunk_progress(request.id) + .await + .expect("progress"), + (1, 0) + ); + store.mark_drained(&token).await.expect("drained"); + store + .mark_manifest_chunk_deleted(&token, 0, serde_json::json!({"deleted": 1})) + .await + .expect("one-way deleted stamp"); + assert_eq!( + store + .manifest_chunk_progress(request.id) + .await + .expect("progress"), + (1, 1) + ); + } + struct AdmissionInsertGate { connection: sqlx::pool::PoolConnection, trigger_name: String, diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 4d4edcf8e98..e334b31aded 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1336,10 +1336,23 @@ fn should_scan_guarded_write_source_file(path: &std::path::Path) -> bool { // Whole files loaded only via `#[cfg(test)] #[path = "..."] mod ...;` are // entirely test code but carry no internal `#[cfg(test)]` marker of their // own to slice against. Standalone `*_tests.rs` modules share that shape. + && file_name != "test_support.rs" && file_name != "tests.rs" && !file_name.ends_with("_tests.rs") } +fn collect_production_rs_files(dir: &std::path::Path, out: &mut Vec) { + for entry in std::fs::read_dir(dir).expect("read source directory") { + let entry = entry.expect("read directory entry"); + let path = entry.path(); + if path.is_dir() { + collect_production_rs_files(&path, out); + } else if should_scan_guarded_write_source_file(&path) { + out.push(path); + } + } +} + #[test] fn serving_table_policy_skips_standalone_test_modules() { use std::path::Path; @@ -1358,6 +1371,10 @@ fn serving_table_policy_skips_standalone_test_modules() { !should_scan_guarded_write_source_file(Path::new("src/store/foo_tests.rs")), "the suffix-based rule must cover other standalone test-only modules" ); + assert!( + !should_scan_guarded_write_source_file(Path::new("src/test_support.rs")), + "test_support.rs is loaded only under cfg(test)" + ); assert!( should_scan_guarded_write_source_file(Path::new("src/store/event.rs")), "production source must remain in scope" @@ -1366,19 +1383,7 @@ fn serving_table_policy_skips_standalone_test_modules() { #[test] fn serving_table_writes_expose_syntactic_chokepoint_or_guarded_tx_routes() { - use std::path::{Path, PathBuf}; - - fn collect_rs_files(dir: &Path, out: &mut Vec) { - for entry in std::fs::read_dir(dir).expect("read source directory") { - let entry = entry.expect("read directory entry"); - let path = entry.path(); - if path.is_dir() { - collect_rs_files(&path, out); - } else if should_scan_guarded_write_source_file(&path) { - out.push(path); - } - } - } + use std::path::Path; // `buzz-relay` is scanned too: relay code reaches fenced tables only through // `buzz-db` helpers, and a direct relay write must meet the same rule. @@ -1389,7 +1394,7 @@ fn serving_table_writes_expose_syntactic_chokepoint_or_guarded_tx_routes() { ("buzz-relay", manifest.join("../buzz-relay/src")), ] { let mut files = Vec::new(); - collect_rs_files(&src_root, &mut files); + collect_production_rs_files(&src_root, &mut files); assert!( !files.is_empty(), "guarded-table scan must see {crate_name} production source files" @@ -1631,20 +1636,7 @@ fn masker_blanks_every_literal_form_and_keeps_code() { /// continuations, `\n`/`\t`/`\r` escapes, and with or without a `public.` /// schema prefix. fn inserts_events_row(source: &str) -> bool { - let mut normalized = String::with_capacity(source.len()); - let mut chars = source.chars().peekable(); - while let Some(ch) = chars.next() { - if ch.is_whitespace() || ch == '\\' { - if ch == '\\' && matches!(chars.peek(), Some('n' | 't' | 'r')) { - chars.next(); - } - if !normalized.ends_with(' ') { - normalized.push(' '); - } - } else { - normalized.push(ch.to_ascii_lowercase()); - } - } + let normalized = normalize_sql_source(source); ["insert into events", "insert into public.events"] .iter() .any(|marker| { @@ -1879,18 +1871,7 @@ mod tests { #[test] fn every_production_event_insert_runs_the_follow_up_hook() { - use std::path::{Path, PathBuf}; - - fn collect_rs_files(dir: &Path, out: &mut Vec) { - for entry in std::fs::read_dir(dir).expect("read source directory") { - let path = entry.expect("read directory entry").path(); - if path.is_dir() { - collect_rs_files(&path, out); - } else if should_scan_guarded_write_source_file(&path) { - out.push(path); - } - } - } + use std::path::Path; let crates_root = Path::new(env!("CARGO_MANIFEST_DIR")) .parent() @@ -1899,7 +1880,7 @@ fn every_production_event_insert_runs_the_follow_up_hook() { for entry in std::fs::read_dir(crates_root).expect("read crates directory") { let src = entry.expect("read crates entry").path().join("src"); if src.is_dir() { - collect_rs_files(&src, &mut files); + collect_production_rs_files(&src, &mut files); } } @@ -1935,3 +1916,370 @@ fn every_production_event_insert_runs_the_follow_up_hook() { "expected at least seven hooked event writers, found {hooked_writers}" ); } + +fn normalize_sql_source(source: &str) -> String { + let mut normalized = String::with_capacity(source.len()); + let mut chars = source.chars().peekable(); + while let Some(ch) = chars.next() { + if ch.is_whitespace() || ch == '\\' { + if ch == '\\' && matches!(chars.peek(), Some('n' | 't' | 'r')) { + chars.next(); + } + if !normalized.ends_with(' ') { + normalized.push(' '); + } + } else { + normalized.push(ch.to_ascii_lowercase()); + } + } + normalized +} + +// This policy covers the supported Rust writers while the retiring database +// guards remain in place for mixed-version pods. SQL is read from the original +// source; calls and struct fields use the existing masked view. +const SINGLE_TABLE_GUARD_TABLES: [&str; 5] = [ + "channels", + "communities", + "community_deletion_requests", + "community_deletion_approvals", + "community_deletion_manifest_keys", +]; +const DELETION_STORE_PATH: &str = "buzz-db/src/store/deletion.rs"; + +fn sql_table_suffixes<'a>(source: &'a str, verb: &str, table: &str) -> Vec<&'a str> { + [format!("{verb} {table}"), format!("{verb} public.{table}")] + .iter() + .flat_map(|marker| { + source.match_indices(marker).filter_map(|(at, found)| { + let before = &source[..at]; + let rest = &source[at + found.len()..]; + let is_ident = |ch: char| ch.is_ascii_alphanumeric() || ch == '_'; + (!before.chars().next_back().is_some_and(is_ident) + && !rest.chars().next().is_some_and(is_ident)) + .then_some(rest) + }) + }) + .collect() +} + +fn sql_update_sets<'a>(source: &'a str, table: &str) -> Vec<&'a str> { + sql_table_suffixes(source, "update", table) + .into_iter() + .filter_map(|rest| { + let rest = rest.trim_start(); + let set = rest.strip_prefix("set ").or_else(|| { + // UPDATE table [AS] alias SET ... + let (alias, set) = rest.split_once(" set ")?; + (alias.split_whitespace().count() <= 2).then_some(set) + })?; + Some(set.split_once(" where ").map_or(set, |(set, _)| set)) + }) + .collect() +} + +fn sql_assigned_columns(set: &str) -> Vec<&str> { + set.match_indices('=') + .filter_map(|(at, _)| { + set[..at] + .trim_end() + .rsplit(|ch: char| !(ch.is_ascii_alphanumeric() || ch == '_')) + .next() + .filter(|column| !column.is_empty()) + }) + .collect() +} + +fn single_table_guard_violations(path: &str, production: &str) -> Vec { + let mut violations = Vec::new(); + let code = mask_comments_and_literals(production); + if let Some((_, fields)) = code.split_once("struct ChannelUpdate") { + let fields = fields + .split_once('{') + .expect("struct body") + .1 + .split('}') + .next() + .expect("struct fields"); + if fields.split(',').any(|field| { + field.split_once(':').is_some_and(|(name, _)| { + name.split_whitespace() + .last() + .is_some_and(|name| name.starts_with("community")) + }) + }) { + violations.push("ChannelUpdate must not expose a community field".to_owned()); + } + } + for function in function_slices(production) { + let header = function_header(function); + let sql = normalize_sql_source(function); + let code = mask_comments_and_literals(function); + let mut reject = |reason: &str| violations.push(format!("{header}: {reason}")); + let channel_sets = sql_update_sets(&sql, "channels"); + if channel_sets + .iter() + .any(|set| sql_assigned_columns(set).contains(&"community_id")) + || (!channel_sets.is_empty() + && sql.split("set_parts.push(").skip(1).any(|part| { + let assignment = part.split(';').next().expect("dynamic SET fragment"); + sql_assigned_columns(assignment).contains(&"community_id") + })) + { + reject("channels.community_id UPDATE"); + } + if sql_update_sets(&sql, "communities").iter().any(|set| { + sql_assigned_columns(set).iter().any(|column| { + ["deletion_state", "deletion_fence_generation", "deleted_at"].contains(column) + }) + }) && !(path == DELETION_STORE_PATH + && header.starts_with("async fn transition_community(")) + { + reject("community lifecycle UPDATE outside transition_community"); + } + if sql_table_suffixes(&sql, "insert into", "communities") + .iter() + .any(|rest| { + rest.trim_start().strip_prefix('(').is_some_and(|columns| { + columns + .split(')') + .next() + .expect("INSERT columns") + .split(',') + .any(|column| { + ["deletion_state", "deletion_fence_generation", "deleted_at"] + .contains(&column.trim()) + }) + }) + }) + { + reject("community creation sets deletion lifecycle columns"); + } + if !sql_table_suffixes(&sql, "delete from", "communities").is_empty() { + reject("production community DELETE"); + } + if sql_update_sets(&sql, "community_deletion_requests") + .iter() + .any(|set| { + sql_assigned_columns(set).iter().any(|column| { + [ + "community_id", + "community_host", + "request_origin", + "owner_pubkey", + "mediating_operator_pubkey", + "acknowledgement_version", + ] + .contains(column) + }) + }) + { + reject("deletion request identity/provenance UPDATE"); + } + if ["update", "delete from"] + .iter() + .any(|verb| !sql_table_suffixes(&sql, verb, "community_deletion_approvals").is_empty()) + { + reject("approval UPDATE or DELETE"); + } + let manifest_writes: usize = ["insert into", "update", "delete from"] + .iter() + .map(|verb| sql_table_suffixes(&sql, verb, "community_deletion_manifest_keys").len()) + .sum(); + if manifest_writes > 0 { + let sets = sql_update_sets(&sql, "community_deletion_manifest_keys"); + let deletion_stamp = header.starts_with("pub async fn mark_manifest_chunk_deleted(") + && manifest_writes == 1 + && sets.len() == 1 + && sql_assigned_columns(sets[0]) == ["deleted_at"]; + if path != DELETION_STORE_PATH + || !(code.contains("guard_manifest_chunk_write(") || deletion_stamp) + { + reject("manifest chunk writer must use guard_manifest_chunk_write or the deleted_at-only stamp"); + } + } + } + violations +} + +#[test] +fn single_table_policy_rejects_channel_retenanting() { + for source in [ + r#"async fn retenant() { query("UPDATE channels SET community_id = $1 WHERE id = $2"); }"#, + r#"async fn retenant() { query("update public.channels as c set name = $1, community_id=$2 where id=$3"); }"#, + r#"async fn retenant() { set_parts.push(format!("community_id = $1")); query("UPDATE channels SET {} WHERE id = $2"); }"#, + "pub struct ChannelUpdate { pub community_id: Option, }", + "pub struct ChannelUpdate { pub community: CommunityId, }", + ] { + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, source).len(), + 1, + "{source}" + ); + } +} + +#[test] +fn single_table_policy_rejects_community_lifecycle_bypasses() { + for column in ["deletion_state", "deletion_fence_generation", "deleted_at"] { + let source = format!("async fn bypass() {{ query(\"UPDATE communities SET {column} = $1 WHERE id = $2\"); }}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + 1, + "{column}" + ); + let transition = source.replace("bypass", "transition_community"); + assert!(single_table_guard_violations(DELETION_STORE_PATH, &transition).is_empty()); + assert_eq!( + single_table_guard_violations("buzz-relay/src/bypass.rs", &transition).len(), + 1 + ); + } + for column in ["deletion_state", "deletion_fence_generation", "deleted_at"] { + let source = format!("async fn bypass() {{ query(\"INSERT INTO communities (id, {column}) VALUES ($1, $2)\"); }}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + 1, + "{column}" + ); + } + let deletion = r#"async fn bypass() { query("DELETE FROM communities WHERE id = $1"); }"#; + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, deletion).len(), + 1 + ); +} + +#[test] +fn single_table_policy_rejects_request_retargeting() { + for column in [ + "community_id", + "community_host", + "request_origin", + "owner_pubkey", + "mediating_operator_pubkey", + "acknowledgement_version", + ] { + let source = format!("async fn bypass() {{ query(\"UPDATE community_deletion_requests SET {column} = $1 WHERE id = $2\"); }}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + 1, + "{column}" + ); + } +} + +#[test] +fn single_table_policy_rejects_approval_mutations() { + for sql in [ + "UPDATE community_deletion_approvals SET approved_by = $1 WHERE request_id = $2", + "DELETE FROM community_deletion_approvals WHERE request_id = $1", + ] { + let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + 1, + "{sql}" + ); + } +} + +#[test] +fn single_table_policy_rejects_unguarded_manifest_writers() { + for sql in [ + "INSERT INTO community_deletion_manifest_keys (keys) VALUES ($1)", + "UPDATE community_deletion_manifest_keys SET keys = $1 WHERE request_id = $2", + "DELETE FROM community_deletion_manifest_keys WHERE request_id = $1", + ] { + let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + 1, + "{sql}" + ); + // A guard in another function or a literal does not admit this writer. + let mixed = format!("async fn guarded() {{ guard_manifest_chunk_write(); }}\n{source}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &mixed).len(), + 1 + ); + let fake = source.replace("query(", "let _ = \"guard_manifest_chunk_write(\"; query("); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &fake).len(), + 1 + ); + let guarded = source.replace("query(", "guard_manifest_chunk_write(); query("); + assert!(single_table_guard_violations(DELETION_STORE_PATH, &guarded).is_empty()); + assert_eq!( + single_table_guard_violations("buzz-relay/src/bypass.rs", &guarded).len(), + 1 + ); + } + let stamp = r#"pub async fn mark_manifest_chunk_deleted() { + query("UPDATE community_deletion_manifest_keys SET deleted_at = now() WHERE request_id = $1"); + }"#; + assert!(single_table_guard_violations(DELETION_STORE_PATH, stamp).is_empty()); + let mutating_stamp = stamp.replace("deleted_at = now()", "deleted_at = now(), keys = $2"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &mutating_stamp).len(), + 1 + ); +} + +#[test] +fn every_production_single_table_guard_writer_preserves_its_invariants() { + use std::path::Path; + + let crates_root = Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .expect("crates root"); + let mut files = Vec::new(); + for entry in std::fs::read_dir(crates_root).expect("read crates directory") { + let src = entry.expect("read crate entry").path().join("src"); + if src.is_dir() { + collect_production_rs_files(&src, &mut files); + } + } + let mut writers = [0usize; 5]; + let mut violations = Vec::new(); + for path in files { + let source = std::fs::read_to_string(&path).expect("read production source"); + let production = strip_cfg_test_items(&source); + for function in function_slices(&production) { + let sql = normalize_sql_source(function); + for (index, table) in SINGLE_TABLE_GUARD_TABLES.iter().enumerate() { + if ["insert into", "update", "delete from"] + .iter() + .any(|verb| !sql_table_suffixes(&sql, verb, table).is_empty()) + { + writers[index] += 1; + } + } + } + let relative = path + .strip_prefix(crates_root) + .expect("crate source") + .to_string_lossy() + .replace('\\', "/"); + violations.extend( + single_table_guard_violations(&relative, &production) + .into_iter() + .map(|violation| format!("{relative}: {violation}")), + ); + } + assert!( + violations.is_empty(), + "single-table guard policy: {violations:#?}" + ); + // Lower bounds pin discovery without fixing the exact number of writers. + for ((table, seen), floor) in SINGLE_TABLE_GUARD_TABLES + .iter() + .zip(writers) + .zip([7, 4, 15, 2, 4]) + { + assert!( + seen >= floor, + "{table}: expected at least {floor} writers, found {seen}" + ); + } +} From d736917ae1189c5735d0a7811845a45299c9b3c8 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 11:01:20 -0400 Subject: [PATCH 2/8] buzz-db: pin the remaining single-table guard halves in source policy Address review on the single-table guard writers: - Require deletion_state = 'active' on every non-lifecycle communities UPDATE and conflict upsert; add it to archive/unarchive UPDATEs, which already lock an active row. - Require the inventory and destructive-storage one-way freeze predicates on every request UPDATE that assigns those columns. - Pin abort's generation increment with a trigger-disabled PG test covering quiescing/fenced and archived/unarchived communities. - Parse ON CONFLICT DO UPDATE SET, tuple SET, and column-less community inserts; require AND deleted_at IS NULL on the manifest stamp. - Assert exact production rule-hit sets so a parser that stops matching fails. - Correct the lifecycle lock doc comment and format the fence body. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/store/community.rs | 12 +- crates/buzz-db/src/store/deletion.rs | 176 ++++++-- crates/buzz-db/tests/observability_source.rs | 451 ++++++++++++++++--- 3 files changed, 521 insertions(+), 118 deletions(-) diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index dcaabdf1bb7..f99fbb97e4d 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -667,7 +667,7 @@ impl Db { } let row = sqlx::query( "UPDATE communities SET archived_at = COALESCE(archived_at, now()) \ - WHERE id = $1 RETURNING id, host, archived_at", + WHERE id = $1 AND deletion_state = 'active' RETURNING id, host, archived_at", ) .bind(community_id) .fetch_one(&mut *tx) @@ -734,10 +734,12 @@ impl Db { tx.rollback().await?; return Ok(UnarchiveCommunityResult::DeletionPending); } - sqlx::query("UPDATE communities SET archived_at = NULL WHERE id = $1") - .bind(community_id) - .execute(&mut *tx) - .await?; + sqlx::query( + "UPDATE communities SET archived_at = NULL WHERE id = $1 AND deletion_state = 'active'", + ) + .bind(community_id) + .execute(&mut *tx) + .await?; tx.commit().await?; Ok(UnarchiveCommunityResult::Unarchived( UnarchivedCommunityRecord { diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 83757e548f5..e588fcd436a 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -1855,51 +1855,52 @@ impl DeletionStore { .await?; transaction_timer .observe(async { - verify_lease(&mut tx, token, DeletionStage::Approved).await?; - lock_community_deletion(&mut tx, token.community_id).await?; - verify_lease(&mut tx, token, DeletionStage::Approved).await?; - let active_serving_writes = sqlx::query( - "SELECT count(*)::BIGINT AS active_count, \ - COALESCE(array_agg(DISTINCT operation ORDER BY operation), ARRAY[]::TEXT[]) AS operations \ - FROM community_serving_write_leases \ - WHERE community_id = $1 AND lease_until >= now()", - ) - .bind(token.community_id.as_uuid()) - .fetch_one(&mut *tx) - .await?; - let active_count: i64 = active_serving_writes.try_get("active_count")?; - if active_count > 0 { - return Err(DbError::ServingWritesNotDrained { - community_id: *token.community_id.as_uuid(), - active_count, - operations: active_serving_writes.try_get("operations")?, - }); - } - let community = lock_community_lifecycle(&mut tx, token.community_id).await?; - let generation = transition_community( - &mut tx, - token.community_id, - community, - CommunityTransition::Fence, - ).await?; - advance_request_tx( - &mut tx, - token, - DeletionStage::Approved, - DeletionStage::Fenced, - Some(generation), - ) - .await?; - checkpoint_completed_tx( - &mut tx, - token, - DeletionStage::Approved, - "activate_fence", - serde_json::json!({"fence_generation": generation}), - ) - .await?; - tx.commit().await?; - Ok(generation) + verify_lease(&mut tx, token, DeletionStage::Approved).await?; + lock_community_deletion(&mut tx, token.community_id).await?; + verify_lease(&mut tx, token, DeletionStage::Approved).await?; + let active_serving_writes = sqlx::query( + "SELECT count(*)::BIGINT AS active_count, \ + COALESCE(array_agg(DISTINCT operation ORDER BY operation), ARRAY[]::TEXT[]) AS operations \ + FROM community_serving_write_leases \ + WHERE community_id = $1 AND lease_until >= now()", + ) + .bind(token.community_id.as_uuid()) + .fetch_one(&mut *tx) + .await?; + let active_count: i64 = active_serving_writes.try_get("active_count")?; + if active_count > 0 { + return Err(DbError::ServingWritesNotDrained { + community_id: *token.community_id.as_uuid(), + active_count, + operations: active_serving_writes.try_get("operations")?, + }); + } + let community = lock_community_lifecycle(&mut tx, token.community_id).await?; + let generation = transition_community( + &mut tx, + token.community_id, + community, + CommunityTransition::Fence, + ) + .await?; + advance_request_tx( + &mut tx, + token, + DeletionStage::Approved, + DeletionStage::Fenced, + Some(generation), + ) + .await?; + checkpoint_completed_tx( + &mut tx, + token, + DeletionStage::Approved, + "activate_fence", + serde_json::json!({"fence_generation": generation}), + ) + .await?; + tx.commit().await?; + Ok(generation) }) .await } @@ -2639,9 +2640,9 @@ impl DeletionStore { .await? .map(CommunityId::from_uuid) .ok_or_else(|| DbError::NotFound(format!("community deletion {request_id}")))?; - // Every lifecycle transition takes the community advisory lock before any + // Quiesce, fence, and abort take 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. + // either order lets abort and those executor transitions deadlock. lock_community_deletion(&mut tx, community_id).await?; let community = lock_community_lifecycle(&mut tx, community_id).await?; let row = sqlx::query("SELECT * FROM community_deletion_requests WHERE id = $1 FOR UPDATE") @@ -3319,8 +3320,10 @@ async fn lock_community_deletion( Ok(()) } -/// The row snapshot is taken under `FOR UPDATE` after the community deletion -/// advisory lock. All lifecycle writers use it in the same transaction. +/// The lifecycle row is locked `FOR UPDATE` in the caller's transaction. +/// Quiesce, fence, and abort also hold the community deletion advisory lock. +/// `purge_postgres` does not; its tombstone transition fails if a concurrent +/// abort won the row lock and changed the lifecycle state. struct CommunityLifecycle { state: String, generation: i64, @@ -4681,6 +4684,81 @@ mod postgres_tests { .is_err()); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn single_table_guard_abort_invalidates_generation_without_trigger() { + let (db, store) = store().await; + sqlx::query("ALTER TABLE communities DISABLE TRIGGER communities_deletion_tombstone") + .execute(&db.pool) + .await + .expect("disable only tombstone trigger"); + for pre_archived in [false, true] { + for fenced in [false, true] { + let (request, _) = inventoried_request(&db, &store).await; + if pre_archived { + sqlx::query("UPDATE communities SET archived_at = now() WHERE id = $1") + .bind(request.community_id.as_uuid()) + .execute(&db.pool) + .await + .expect("pre-archive fixture"); + } + let pre_quiesce_archived_at: Option> = + sqlx::query_scalar("SELECT archived_at FROM communities WHERE id = $1") + .bind(request.community_id.as_uuid()) + .fetch_one(&db.pool) + .await + .expect("original archive state"); + assert_eq!(pre_quiesce_archived_at.is_some(), pre_archived); + 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"); + if fenced { + store.fence(&claim.lease).await.expect("fence"); + } + let (state, fenced_gen): (String, i64) = sqlx::query_as( + "SELECT deletion_state, deletion_fence_generation FROM communities WHERE id = $1", + ) + .bind(request.community_id.as_uuid()) + .fetch_one(&db.pool) + .await + .expect("pre-abort lifecycle"); + assert_eq!(state, if fenced { "fenced" } else { "quiescing" }); + assert_eq!(fenced_gen, if fenced { 1 } else { 0 }); + let aborted = store + .abort(request.id, "operator", "generation regression") + .await + .expect("abort through app path"); + assert_eq!(aborted.stage, DeletionStage::Aborted); + assert_eq!(aborted.pre_quiesce_archived_at, pre_quiesce_archived_at); + assert_eq!(aborted.fence_generation, Some(fenced_gen + 1)); + let (state, generation, archived_at, deleted_at): ( + String, + i64, + Option>, + Option>, + ) = sqlx::query_as( + "SELECT deletion_state, deletion_fence_generation, archived_at, deleted_at \ + FROM communities WHERE id = $1", + ) + .bind(request.community_id.as_uuid()) + .fetch_one(&db.pool) + .await + .expect("post-abort lifecycle"); + assert_eq!(state, "active"); + assert_eq!(generation, fenced_gen + 1); + assert_eq!(archived_at, pre_quiesce_archived_at); + assert_eq!(deleted_at, None); + } + } + } + #[tokio::test] #[ignore = "requires Postgres"] async fn single_table_guard_request_inventory_freeze_is_one_way_without_trigger() { diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index e334b31aded..e74ec4dfa61 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1963,34 +1963,69 @@ fn sql_table_suffixes<'a>(source: &'a str, verb: &str, table: &str) -> Vec<&'a s .collect() } -fn sql_update_sets<'a>(source: &'a str, table: &str) -> Vec<&'a str> { - sql_table_suffixes(source, "update", table) +// Keep SET and WHERE from the same statement, including an INSERT's conflict UPDATE. +fn sql_update_sets<'a>(source: &'a str, table: &str) -> Vec<(&'a str, &'a str)> { + let updates = sql_table_suffixes(source, "update", table) .into_iter() .filter_map(|rest| { - let rest = rest.trim_start(); - let set = rest.strip_prefix("set ").or_else(|| { + let rest = rest.split(['"', ';']).next()?.trim_start(); + rest.strip_prefix("set ").or_else(|| { // UPDATE table [AS] alias SET ... let (alias, set) = rest.split_once(" set ")?; (alias.split_whitespace().count() <= 2).then_some(set) - })?; - Some(set.split_once(" where ").map_or(set, |(set, _)| set)) - }) + }) + }); + let conflict_updates = sql_table_suffixes(source, "insert into", table) + .into_iter() + .filter_map(|rest| { + let statement = rest.split(['"', ';']).next()?; + let (_, conflict) = statement.split_once(" on conflict ")?; + conflict.split_once(" do update set ").map(|(_, set)| set) + }); + updates + .chain(conflict_updates) + .map(|set| set.split_once(" where ").unwrap_or((set, ""))) .collect() } fn sql_assigned_columns(set: &str) -> Vec<&str> { set.match_indices('=') - .filter_map(|(at, _)| { - set[..at] - .trim_end() - .rsplit(|ch: char| !(ch.is_ascii_alphanumeric() || ch == '_')) + .flat_map(|(at, _)| -> Vec<&str> { + let left = set[..at].trim_end(); + if left.ends_with(')') { + // Tuple assignment: SET (a, b) = (...). + let mut depth = 0; + for (at, ch) in left.char_indices().rev() { + match ch { + ')' => depth += 1, + '(' => { + depth -= 1; + if depth == 0 { + return left[at + 1..left.len() - 1] + .split(',') + .map(str::trim) + .collect(); + } + } + _ => {} + } + } + } + left.rsplit(|ch: char| !(ch.is_ascii_alphanumeric() || ch == '_')) .next() .filter(|column| !column.is_empty()) + .into_iter() + .collect() }) .collect() } -fn single_table_guard_violations(path: &str, production: &str) -> Vec { +fn single_table_guard_violations( + path: &str, + production: &str, + // Lifecycle exemption, stamp exemption, chunk guard, active predicate, request freezes. + rule_hits: &mut [Vec; 5], +) -> Vec { let mut violations = Vec::new(); let code = mask_comments_and_literals(production); if let Some((_, fields)) = code.split_once("struct ChannelUpdate") { @@ -2016,10 +2051,19 @@ fn single_table_guard_violations(path: &str, production: &str) -> Vec { let sql = normalize_sql_source(function); let code = mask_comments_and_literals(function); let mut reject = |reason: &str| violations.push(format!("{header}: {reason}")); + let name = header + .split_once("fn ") + .expect("function name") + .1 + .split('(') + .next() + .expect("function arguments"); + let lifecycle_writer = + path == DELETION_STORE_PATH && header.starts_with("async fn transition_community("); let channel_sets = sql_update_sets(&sql, "channels"); if channel_sets .iter() - .any(|set| sql_assigned_columns(set).contains(&"community_id")) + .any(|(set, _)| sql_assigned_columns(set).contains(&"community_id")) || (!channel_sets.is_empty() && sql.split("set_parts.push(").skip(1).any(|part| { let assignment = part.split(';').next().expect("dynamic SET fragment"); @@ -2028,54 +2072,103 @@ fn single_table_guard_violations(path: &str, production: &str) -> Vec { { reject("channels.community_id UPDATE"); } - if sql_update_sets(&sql, "communities").iter().any(|set| { + let community_sets = sql_update_sets(&sql, "communities"); + if community_sets.iter().any(|(set, _)| { sql_assigned_columns(set).iter().any(|column| { ["deletion_state", "deletion_fence_generation", "deleted_at"].contains(column) }) - }) && !(path == DELETION_STORE_PATH - && header.starts_with("async fn transition_community(")) - { - reject("community lifecycle UPDATE outside transition_community"); + }) { + if lifecycle_writer { + rule_hits[0].push(name.to_owned()); + } else { + reject("community lifecycle UPDATE outside transition_community"); + } + } + if !community_sets.is_empty() && !lifecycle_writer { + rule_hits[3].push(name.to_owned()); + if community_sets + .iter() + .any(|(_, predicate)| !predicate.contains("deletion_state = 'active'")) + { + reject("community UPDATE requires an active predicate in its statement"); + } } if sql_table_suffixes(&sql, "insert into", "communities") .iter() .any(|rest| { - rest.trim_start().strip_prefix('(').is_some_and(|columns| { - columns - .split(')') - .next() - .expect("INSERT columns") - .split(',') - .any(|column| { - ["deletion_state", "deletion_fence_generation", "deleted_at"] - .contains(&column.trim()) - }) - }) + let Some(columns) = rest.trim_start().strip_prefix('(') else { + return true; + }; + columns + .split(')') + .next() + .expect("INSERT columns") + .split(',') + .any(|column| { + ["deletion_state", "deletion_fence_generation", "deleted_at"] + .contains(&column.trim()) + }) }) { - reject("community creation sets deletion lifecycle columns"); + reject( + "community creation requires explicit columns without deletion lifecycle fields", + ); } if !sql_table_suffixes(&sql, "delete from", "communities").is_empty() { reject("production community DELETE"); } - if sql_update_sets(&sql, "community_deletion_requests") - .iter() - .any(|set| { - sql_assigned_columns(set).iter().any(|column| { - [ - "community_id", - "community_host", - "request_origin", - "owner_pubkey", - "mediating_operator_pubkey", - "acknowledgement_version", - ] - .contains(column) - }) + let request_sets = sql_update_sets(&sql, "community_deletion_requests"); + if request_sets.iter().any(|(set, _)| { + sql_assigned_columns(set).iter().any(|column| { + [ + "community_id", + "community_host", + "request_origin", + "owner_pubkey", + "mediating_operator_pubkey", + "acknowledgement_version", + ] + .contains(column) }) - { + }) { reject("deletion request identity/provenance UPDATE"); } + let mut freezes_inventory = false; + let mut freezes_destructive_storage = false; + for (set, predicate) in &request_sets { + let columns = sql_assigned_columns(set); + if columns.iter().any(|column| { + [ + "schema_manifest", + "storage_manifest", + "inventory_manifest", + "inventory_digest", + "inventory_frozen_at", + ] + .contains(column) + }) { + freezes_inventory = true; + if !predicate.contains("inventory_frozen_at is null") { + reject( + "request inventory UPDATE requires inventory_frozen_at IS NULL in WHERE", + ); + } + } + if columns + .iter() + .any(|column| column.starts_with("destructive_storage_")) + { + freezes_destructive_storage = true; + if !predicate.contains( + "destructive_storage_frozen_at is null or destructive_storage_manifest =", + ) { + reject("request destructive storage UPDATE requires the frozen-or-equal guard in WHERE"); + } + } + } + if freezes_inventory || freezes_destructive_storage { + rule_hits[4].push(name.to_owned()); + } if ["update", "delete from"] .iter() .any(|verb| !sql_table_suffixes(&sql, verb, "community_deletion_approvals").is_empty()) @@ -2091,7 +2184,13 @@ fn single_table_guard_violations(path: &str, production: &str) -> Vec { let deletion_stamp = header.starts_with("pub async fn mark_manifest_chunk_deleted(") && manifest_writes == 1 && sets.len() == 1 - && sql_assigned_columns(sets[0]) == ["deleted_at"]; + && sql_assigned_columns(sets[0].0) == ["deleted_at"] + && sets[0].1.contains("and deleted_at is null"); + if deletion_stamp { + rule_hits[1].push(name.to_owned()); + } else if code.contains("guard_manifest_chunk_write(") { + rule_hits[2].push(name.to_owned()); + } if path != DELETION_STORE_PATH || !(code.contains("guard_manifest_chunk_write(") || deletion_stamp) { @@ -2112,40 +2211,206 @@ fn single_table_policy_rejects_channel_retenanting() { "pub struct ChannelUpdate { pub community: CommunityId, }", ] { assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, source).len(), + single_table_guard_violations(DELETION_STORE_PATH, source, &mut Default::default()) + .len(), 1, "{source}" ); } } +#[test] +fn single_table_policy_requires_active_community_updates() { + let source = r#"async fn bypass() { + query("UPDATE communities SET icon = NULL WHERE id = $1"); + }"#; + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, source, &mut Default::default()).len(), + 1 + ); + let guarded = source.replace( + "WHERE id = $1", + "WHERE id = $1 AND deletion_state = 'active'", + ); + assert!( + single_table_guard_violations(DELETION_STORE_PATH, &guarded, &mut Default::default()) + .is_empty() + ); + // A predicate in a SELECT or another UPDATE does not admit this statement. + let separate = source.replace( + "query(", + "query(\"SELECT id FROM communities WHERE deletion_state = 'active'\"); query(", + ); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &separate, &mut Default::default()) + .len(), + 1 + ); + let two_updates = r#"async fn bypass() { + query("UPDATE communities SET icon = NULL WHERE id = $1 AND deletion_state = 'active'"); + query("UPDATE communities SET icon = NULL WHERE id = $2"); + }"#; + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, two_updates, &mut Default::default()) + .len(), + 1 + ); +} + +#[test] +fn single_table_policy_requires_inventory_freeze_predicate() { + for column in [ + "schema_manifest", + "storage_manifest", + "inventory_manifest", + "inventory_digest", + "inventory_frozen_at", + ] { + let source = format!("async fn bypass() {{ query(\"UPDATE community_deletion_requests SET {column} = NULL WHERE id = $1\"); }}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), + 1, + "{column}" + ); + let guarded = source.replace( + "WHERE id = $1", + "WHERE id = $1 AND inventory_frozen_at IS NULL", + ); + assert!( + single_table_guard_violations(DELETION_STORE_PATH, &guarded, &mut Default::default()) + .is_empty(), + "{column}" + ); + } +} + +#[test] +fn single_table_policy_requires_destructive_freeze_predicate() { + for column in [ + "destructive_storage_manifest", + "destructive_storage_frozen_at", + ] { + let source = format!("async fn bypass() {{ query(\"UPDATE community_deletion_requests SET {column} = NULL WHERE id = $1\"); }}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), + 1, + "{column}" + ); + let guarded = source.replace("WHERE id = $1", "WHERE id = $1 AND (destructive_storage_frozen_at IS NULL OR destructive_storage_manifest = $2)"); + assert!( + single_table_guard_violations(DELETION_STORE_PATH, &guarded, &mut Default::default()) + .is_empty(), + "{column}" + ); + } +} + +#[test] +fn single_table_policy_checks_conflict_updates() { + for (sql, reason) in [ + ("INSERT INTO communities (host) VALUES ($1) ON CONFLICT (host) DO UPDATE SET icon = NULL", "requires an active predicate"), + ("INSERT INTO communities (host) VALUES ($1) ON CONFLICT (host) DO UPDATE SET deleted_at = now() WHERE communities.deletion_state = 'active'", "lifecycle UPDATE"), + ("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET community_id = $2", "identity/provenance UPDATE"), + ("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET inventory_digest = NULL", "inventory_frozen_at IS NULL"), + ("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET destructive_storage_manifest = NULL", "frozen-or-equal guard"), + ] { + let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); + let violations = single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()); + assert!(violations.iter().any(|violation| violation.contains(reason)), "{sql}: {violations:?}"); + } + let source = r#"async fn guarded() { + query("INSERT INTO communities (host) VALUES ($1) ON CONFLICT (host) DO UPDATE SET host = communities.host WHERE communities.deletion_state = 'active'"); + query("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET inventory_digest = $2 WHERE inventory_frozen_at IS NULL"); + query("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET destructive_storage_manifest = $2 WHERE destructive_storage_frozen_at IS NULL OR destructive_storage_manifest = $2"); + }"#; + assert!( + single_table_guard_violations(DELETION_STORE_PATH, source, &mut Default::default()) + .is_empty() + ); +} + +#[test] +fn single_table_policy_rejects_implicit_community_insert_columns() { + for sql in [ + "INSERT INTO communities VALUES ($1, $2)", + "INSERT INTO communities SELECT * FROM staging", + ] { + let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), + 1, + "{sql}" + ); + } + let explicit = r#"async fn create() { query("INSERT INTO communities (host) SELECT host FROM staging"); }"#; + assert!( + single_table_guard_violations(DELETION_STORE_PATH, explicit, &mut Default::default()) + .is_empty() + ); +} + +#[test] +fn single_table_policy_checks_tuple_assignments() { + for (sql, reason) in [ + ("UPDATE channels SET (name, community_id) = ($1, $2) WHERE id = $3", "channels.community_id UPDATE"), + ("UPDATE communities SET (icon, deletion_state) = (NULL, 'active') WHERE id = $1 AND deletion_state = 'active'", "lifecycle UPDATE"), + ("UPDATE community_deletion_requests SET (stage, community_id) = ($1, $2) WHERE id = $3", "identity/provenance UPDATE"), + ("UPDATE community_deletion_requests SET (inventory_digest, inventory_manifest) = (NULL, NULL) WHERE id = $1", "inventory_frozen_at IS NULL"), + ("UPDATE community_deletion_requests SET (destructive_storage_manifest, destructive_storage_frozen_at) = (NULL, NULL) WHERE id = $1", "frozen-or-equal guard"), + ] { + let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); + let violations = single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()); + assert!(violations.iter().any(|violation| violation.contains(reason)), "{sql}: {violations:?}"); + } + let guarded = r#"async fn freeze() { query("UPDATE community_deletion_requests SET (inventory_manifest, inventory_digest) = ($2, $3) WHERE id = $1 AND inventory_frozen_at IS NULL"); }"#; + assert!( + single_table_guard_violations(DELETION_STORE_PATH, guarded, &mut Default::default()) + .is_empty() + ); +} + #[test] fn single_table_policy_rejects_community_lifecycle_bypasses() { for column in ["deletion_state", "deletion_fence_generation", "deleted_at"] { - let source = format!("async fn bypass() {{ query(\"UPDATE communities SET {column} = $1 WHERE id = $2\"); }}"); + let source = format!("async fn bypass() {{ query(\"UPDATE communities SET {column} = $1 WHERE id = $2 AND deletion_state = 'active'\"); }}"); assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), 1, "{column}" ); let transition = source.replace("bypass", "transition_community"); - assert!(single_table_guard_violations(DELETION_STORE_PATH, &transition).is_empty()); + assert!(single_table_guard_violations( + DELETION_STORE_PATH, + &transition, + &mut Default::default() + ) + .is_empty()); assert_eq!( - single_table_guard_violations("buzz-relay/src/bypass.rs", &transition).len(), + single_table_guard_violations( + "buzz-relay/src/bypass.rs", + &transition, + &mut Default::default() + ) + .len(), 1 ); } for column in ["deletion_state", "deletion_fence_generation", "deleted_at"] { let source = format!("async fn bypass() {{ query(\"INSERT INTO communities (id, {column}) VALUES ($1, $2)\"); }}"); assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), 1, "{column}" ); } let deletion = r#"async fn bypass() { query("DELETE FROM communities WHERE id = $1"); }"#; assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, deletion).len(), + single_table_guard_violations(DELETION_STORE_PATH, deletion, &mut Default::default()).len(), 1 ); } @@ -2162,7 +2427,8 @@ fn single_table_policy_rejects_request_retargeting() { ] { let source = format!("async fn bypass() {{ query(\"UPDATE community_deletion_requests SET {column} = $1 WHERE id = $2\"); }}"); assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), 1, "{column}" ); @@ -2177,7 +2443,8 @@ fn single_table_policy_rejects_approval_mutations() { ] { let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), 1, "{sql}" ); @@ -2193,35 +2460,66 @@ fn single_table_policy_rejects_unguarded_manifest_writers() { ] { let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, &source).len(), + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), 1, "{sql}" ); // A guard in another function or a literal does not admit this writer. let mixed = format!("async fn guarded() {{ guard_manifest_chunk_write(); }}\n{source}"); assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, &mixed).len(), + single_table_guard_violations(DELETION_STORE_PATH, &mixed, &mut Default::default()) + .len(), 1 ); let fake = source.replace("query(", "let _ = \"guard_manifest_chunk_write(\"; query("); assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, &fake).len(), + single_table_guard_violations(DELETION_STORE_PATH, &fake, &mut Default::default()) + .len(), 1 ); let guarded = source.replace("query(", "guard_manifest_chunk_write(); query("); - assert!(single_table_guard_violations(DELETION_STORE_PATH, &guarded).is_empty()); + assert!(single_table_guard_violations( + DELETION_STORE_PATH, + &guarded, + &mut Default::default() + ) + .is_empty()); assert_eq!( - single_table_guard_violations("buzz-relay/src/bypass.rs", &guarded).len(), + single_table_guard_violations( + "buzz-relay/src/bypass.rs", + &guarded, + &mut Default::default() + ) + .len(), 1 ); } let stamp = r#"pub async fn mark_manifest_chunk_deleted() { - query("UPDATE community_deletion_manifest_keys SET deleted_at = now() WHERE request_id = $1"); + query("UPDATE community_deletion_manifest_keys SET deleted_at = now() WHERE request_id = $1 AND deleted_at IS NULL"); }"#; - assert!(single_table_guard_violations(DELETION_STORE_PATH, stamp).is_empty()); + assert!( + single_table_guard_violations(DELETION_STORE_PATH, stamp, &mut Default::default()) + .is_empty() + ); + let repeatable_stamp = stamp.replace(" AND deleted_at IS NULL", ""); + assert_eq!( + single_table_guard_violations( + DELETION_STORE_PATH, + &repeatable_stamp, + &mut Default::default() + ) + .len(), + 1 + ); let mutating_stamp = stamp.replace("deleted_at = now()", "deleted_at = now(), keys = $2"); assert_eq!( - single_table_guard_violations(DELETION_STORE_PATH, &mutating_stamp).len(), + single_table_guard_violations( + DELETION_STORE_PATH, + &mutating_stamp, + &mut Default::default() + ) + .len(), 1 ); } @@ -2241,6 +2539,7 @@ fn every_production_single_table_guard_writer_preserves_its_invariants() { } } let mut writers = [0usize; 5]; + let mut rule_hits: [Vec; 5] = Default::default(); let mut violations = Vec::new(); for path in files { let source = std::fs::read_to_string(&path).expect("read production source"); @@ -2262,7 +2561,7 @@ fn every_production_single_table_guard_writer_preserves_its_invariants() { .to_string_lossy() .replace('\\', "/"); violations.extend( - single_table_guard_violations(&relative, &production) + single_table_guard_violations(&relative, &production, &mut rule_hits) .into_iter() .map(|violation| format!("{relative}: {violation}")), ); @@ -2271,6 +2570,30 @@ fn every_production_single_table_guard_writer_preserves_its_invariants() { violations.is_empty(), "single-table guard policy: {violations:#?}" ); + // Exact rule hits pin the parser to the known production seams. + for (hits, expected) in rule_hits.iter_mut().zip([ + vec!["transition_community"], + vec!["mark_manifest_chunk_deleted"], + vec![ + "append_manifest_key_chunk", + "clear_manifest_key_chunks", + "mark_logically_verified", + ], + vec![ + "archive_community_owned_by", + "ensure_configured_community_with_operation", + "set_community_icon", + "unarchive_community_owned_by", + ], + vec![ + "complete_owner_preparation", + "freeze_destructive_storage_manifest", + "freeze_inventory", + ], + ]) { + hits.sort(); + assert_eq!(*hits, expected, "production single-table rule hits"); + } // Lower bounds pin discovery without fixing the exact number of writers. for ((table, seen), floor) in SINGLE_TABLE_GUARD_TABLES .iter() From 3f7fba66d4bb4eb3e2f5acff1c3a819622f0e14d Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 11:21:56 -0400 Subject: [PATCH 3/8] buzz-db: reject weakened single-table guards in source policy A guard now admits a statement only as a plain conjunct of its WHERE (no OR outside the frozen-or-equal groups, no subquery), and every destructive_storage_* assignment must preserve a frozen value with COALESCE. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/tests/observability_source.rs | 59 +++++++++++++++++--- 1 file changed, 52 insertions(+), 7 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index e74ec4dfa61..b4655a8a81a 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1988,6 +1988,19 @@ fn sql_update_sets<'a>(source: &'a str, table: &str) -> Vec<(&'a str, &'a str)> .collect() } +const DESTRUCTIVE_FROZEN_OR_EQUAL: &str = + "destructive_storage_frozen_at is null or destructive_storage_manifest ="; + +/// A guard admits a statement only as a plain conjunct of its own WHERE: no +/// `OR` outside the known frozen-or-equal groups, and no subquery. +fn plain_guard(predicate: &str, guard: &str) -> bool { + let rest = predicate.replace(DESTRUCTIVE_FROZEN_OR_EQUAL, "").replace( + "destructive_storage_manifest is null or destructive_storage_manifest =", + "", + ); + predicate.contains(guard) && !rest.contains(" or ") && !rest.contains("select ") +} + fn sql_assigned_columns(set: &str) -> Vec<&str> { set.match_indices('=') .flat_map(|(at, _)| -> Vec<&str> { @@ -2088,7 +2101,7 @@ fn single_table_guard_violations( rule_hits[3].push(name.to_owned()); if community_sets .iter() - .any(|(_, predicate)| !predicate.contains("deletion_state = 'active'")) + .any(|(_, predicate)| !plain_guard(predicate, "deletion_state = 'active'")) { reject("community UPDATE requires an active predicate in its statement"); } @@ -2148,7 +2161,7 @@ fn single_table_guard_violations( .contains(column) }) { freezes_inventory = true; - if !predicate.contains("inventory_frozen_at is null") { + if !plain_guard(predicate, "inventory_frozen_at is null") { reject( "request inventory UPDATE requires inventory_frozen_at IS NULL in WHERE", ); @@ -2159,11 +2172,16 @@ fn single_table_guard_violations( .any(|column| column.starts_with("destructive_storage_")) { freezes_destructive_storage = true; - if !predicate.contains( - "destructive_storage_frozen_at is null or destructive_storage_manifest =", - ) { + if !plain_guard(predicate, DESTRUCTIVE_FROZEN_OR_EQUAL) { reject("request destructive storage UPDATE requires the frozen-or-equal guard in WHERE"); } + if columns + .iter() + .filter(|column| column.starts_with("destructive_storage_")) + .any(|column| !set.contains(&format!("{column} = coalesce({column},"))) + { + reject("request destructive storage UPDATE must preserve a frozen value with COALESCE"); + } } } if freezes_inventory || freezes_destructive_storage { @@ -2257,6 +2275,25 @@ fn single_table_policy_requires_active_community_updates() { ); } +#[test] +fn single_table_policy_rejects_weakened_guards() { + for sql in [ + "UPDATE communities SET icon = NULL WHERE id = $1 AND (deletion_state = 'active' OR true)", + "UPDATE communities SET icon = NULL WHERE id = $1 OR deletion_state = 'active'", + "UPDATE communities SET icon = NULL WHERE id IN (SELECT id FROM communities WHERE deletion_state = 'active')", + "UPDATE community_deletion_requests SET inventory_digest = NULL WHERE id = $1 AND (inventory_frozen_at IS NULL OR true)", + "UPDATE community_deletion_requests SET destructive_storage_manifest = COALESCE(destructive_storage_manifest, $2) WHERE id = $1 AND (destructive_storage_frozen_at IS NULL OR destructive_storage_manifest = $2 OR true)", + ] { + let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), + 1, + "{sql}" + ); + } +} + #[test] fn single_table_policy_requires_inventory_freeze_predicate() { for column in [ @@ -2291,7 +2328,7 @@ fn single_table_policy_requires_destructive_freeze_predicate() { "destructive_storage_manifest", "destructive_storage_frozen_at", ] { - let source = format!("async fn bypass() {{ query(\"UPDATE community_deletion_requests SET {column} = NULL WHERE id = $1\"); }}"); + let source = format!("async fn bypass() {{ query(\"UPDATE community_deletion_requests SET {column} = COALESCE({column}, $2) WHERE id = $1\"); }}"); assert_eq!( single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) .len(), @@ -2304,6 +2341,14 @@ fn single_table_policy_requires_destructive_freeze_predicate() { .is_empty(), "{column}" ); + // The guard keeps an equal manifest admissible, so SET must not overwrite it. + let overwrite = guarded.replace(&format!("COALESCE({column}, $2)"), "NULL"); + let violations = + single_table_guard_violations(DELETION_STORE_PATH, &overwrite, &mut Default::default()); + assert!( + violations.len() == 1 && violations[0].contains("preserve a frozen value"), + "{column}: {violations:?}" + ); } } @@ -2323,7 +2368,7 @@ fn single_table_policy_checks_conflict_updates() { let source = r#"async fn guarded() { query("INSERT INTO communities (host) VALUES ($1) ON CONFLICT (host) DO UPDATE SET host = communities.host WHERE communities.deletion_state = 'active'"); query("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET inventory_digest = $2 WHERE inventory_frozen_at IS NULL"); - query("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET destructive_storage_manifest = $2 WHERE destructive_storage_frozen_at IS NULL OR destructive_storage_manifest = $2"); + query("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET destructive_storage_manifest = COALESCE(destructive_storage_manifest, $2) WHERE destructive_storage_frozen_at IS NULL OR destructive_storage_manifest = $2"); }"#; assert!( single_table_guard_violations(DELETION_STORE_PATH, source, &mut Default::default()) From 1e2c3f8f6966bdfbf5ba1ec257b1cdf0acdf5f24 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 11:37:36 -0400 Subject: [PATCH 4/8] buzz-db: require single-table guards as whole-word conjuncts A guard must follow the WHERE start or an AND (optionally table-qualified), so AND NOT no longer satisfies it, and OR/SELECT are matched as whole words so )OR( is caught. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/tests/observability_source.rs | 22 +++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index b4655a8a81a..dbb545773a4 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1991,14 +1991,27 @@ fn sql_update_sets<'a>(source: &'a str, table: &str) -> Vec<(&'a str, &'a str)> const DESTRUCTIVE_FROZEN_OR_EQUAL: &str = "destructive_storage_frozen_at is null or destructive_storage_manifest ="; -/// A guard admits a statement only as a plain conjunct of its own WHERE: no -/// `OR` outside the known frozen-or-equal groups, and no subquery. +/// A guard admits a statement only as a plain conjunct of its own WHERE: it +/// follows the WHERE start or an `AND` (never `NOT`), and there is no `OR` +/// word outside the known frozen-or-equal groups, and no subquery. fn plain_guard(predicate: &str, guard: &str) -> bool { + let conjunct = predicate.match_indices(guard).any(|(at, _)| { + let before = &predicate[..at]; + // Allow a table qualifier: `communities.deletion_state = ...`. + let before = before + .strip_suffix('.') + .map_or(before, |qualified| { + qualified.trim_end_matches(|ch: char| ch.is_ascii_alphanumeric() || ch == '_') + }) + .trim_end_matches([' ', '(']); + before.is_empty() || before.ends_with(" and") + }); let rest = predicate.replace(DESTRUCTIVE_FROZEN_OR_EQUAL, "").replace( "destructive_storage_manifest is null or destructive_storage_manifest =", "", ); - predicate.contains(guard) && !rest.contains(" or ") && !rest.contains("select ") + let words = || rest.split(|ch: char| !(ch.is_ascii_alphanumeric() || ch == '_')); + conjunct && !words().any(|word| word == "or" || word == "select") } fn sql_assigned_columns(set: &str) -> Vec<&str> { @@ -2281,6 +2294,9 @@ fn single_table_policy_rejects_weakened_guards() { "UPDATE communities SET icon = NULL WHERE id = $1 AND (deletion_state = 'active' OR true)", "UPDATE communities SET icon = NULL WHERE id = $1 OR deletion_state = 'active'", "UPDATE communities SET icon = NULL WHERE id IN (SELECT id FROM communities WHERE deletion_state = 'active')", + "UPDATE communities SET icon = NULL WHERE id = $1 AND NOT deletion_state = 'active'", + "UPDATE communities SET icon = NULL WHERE id = $1 AND NOT communities.deletion_state = 'active'", + "UPDATE communities SET icon = NULL WHERE id = $1 AND (deletion_state = 'active')OR(true)", "UPDATE community_deletion_requests SET inventory_digest = NULL WHERE id = $1 AND (inventory_frozen_at IS NULL OR true)", "UPDATE community_deletion_requests SET destructive_storage_manifest = COALESCE(destructive_storage_manifest, $2) WHERE id = $1 AND (destructive_storage_frozen_at IS NULL OR destructive_storage_manifest = $2 OR true)", ] { From 8aeda9a914544d42153d44c1d907a937b3104569 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 09:57:14 -0400 Subject: [PATCH 5/8] buzz-db: close approval upsert and manifest UPDATE gaps in guard policy Reject ON CONFLICT DO UPDATE on approvals, reject every manifest chunk UPDATE other than the one-way deleted_at stamp even when the chunk guard is present, and hold that stamp to the plain-conjunct rule. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/tests/observability_source.rs | 108 +++++++++++++------ 1 file changed, 78 insertions(+), 30 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index dbb545773a4..466597ec094 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -2203,29 +2203,36 @@ fn single_table_guard_violations( if ["update", "delete from"] .iter() .any(|verb| !sql_table_suffixes(&sql, verb, "community_deletion_approvals").is_empty()) + || !sql_update_sets(&sql, "community_deletion_approvals").is_empty() { - reject("approval UPDATE or DELETE"); + reject("approval UPDATE, upsert or DELETE"); } - let manifest_writes: usize = ["insert into", "update", "delete from"] - .iter() - .map(|verb| sql_table_suffixes(&sql, verb, "community_deletion_manifest_keys").len()) - .sum(); + let manifest = "community_deletion_manifest_keys"; + let manifest_updates = sql_table_suffixes(&sql, "update", manifest).len(); + let manifest_writes = manifest_updates + + sql_table_suffixes(&sql, "insert into", manifest).len() + + sql_table_suffixes(&sql, "delete from", manifest).len(); if manifest_writes > 0 { - let sets = sql_update_sets(&sql, "community_deletion_manifest_keys"); - let deletion_stamp = header.starts_with("pub async fn mark_manifest_chunk_deleted(") + // Chunk content never changes. The only UPDATE is the one-way + // `deleted_at` stamp, held to the same plain-conjunct rule as guards. + let sets = sql_update_sets(&sql, manifest); + let deletion_stamp = path == DELETION_STORE_PATH + && header.starts_with("pub async fn mark_manifest_chunk_deleted(") && manifest_writes == 1 + && manifest_updates == 1 && sets.len() == 1 - && sql_assigned_columns(sets[0].0) == ["deleted_at"] - && sets[0].1.contains("and deleted_at is null"); + && sets[0].0.trim() == "deleted_at = now()" + && plain_guard(sets[0].1, "request_id = $1") + && plain_guard(sets[0].1, "deleted_at is null"); + let guarded = code.contains("guard_manifest_chunk_write("); if deletion_stamp { rule_hits[1].push(name.to_owned()); - } else if code.contains("guard_manifest_chunk_write(") { + } else if manifest_updates > 0 || !sets.is_empty() { + reject("manifest chunk UPDATE other than the one-way deleted_at stamp"); + } else if path == DELETION_STORE_PATH && guarded { rule_hits[2].push(name.to_owned()); - } - if path != DELETION_STORE_PATH - || !(code.contains("guard_manifest_chunk_write(") || deletion_stamp) - { - reject("manifest chunk writer must use guard_manifest_chunk_write or the deleted_at-only stamp"); + } else { + reject("manifest chunk INSERT or DELETE must use guard_manifest_chunk_write"); } } } @@ -2501,6 +2508,7 @@ fn single_table_policy_rejects_approval_mutations() { for sql in [ "UPDATE community_deletion_approvals SET approved_by = $1 WHERE request_id = $2", "DELETE FROM community_deletion_approvals WHERE request_id = $1", + "INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2) ON CONFLICT (request_id) DO UPDATE SET note = EXCLUDED.note", ] { let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); assert_eq!( @@ -2510,13 +2518,20 @@ fn single_table_policy_rejects_approval_mutations() { "{sql}" ); } + let append = r#"async fn approve() { + query("INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2)"); + query("INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2) ON CONFLICT (request_id) DO NOTHING"); + }"#; + assert!( + single_table_guard_violations(DELETION_STORE_PATH, append, &mut Default::default()) + .is_empty() + ); } #[test] fn single_table_policy_rejects_unguarded_manifest_writers() { for sql in [ "INSERT INTO community_deletion_manifest_keys (keys) VALUES ($1)", - "UPDATE community_deletion_manifest_keys SET keys = $1 WHERE request_id = $2", "DELETE FROM community_deletion_manifest_keys WHERE request_id = $1", ] { let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); @@ -2556,31 +2571,64 @@ fn single_table_policy_rejects_unguarded_manifest_writers() { 1 ); } +} + +#[test] +fn single_table_policy_rejects_manifest_content_updates() { + // The chunk guard admits INSERT and DELETE only: content never changes. + for sql in [ + "UPDATE community_deletion_manifest_keys SET keys = $1 WHERE request_id = $2", + "INSERT INTO community_deletion_manifest_keys (request_id, chunk_no, keys) VALUES ($1, $2, $3) ON CONFLICT (request_id, chunk_no) DO UPDATE SET keys = EXCLUDED.keys", + ] { + let source = format!("async fn append() {{ guard_manifest_chunk_write(); query(\"{sql}\"); }}"); + let violations = + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()); + assert_eq!(violations.len(), 1, "{sql}: {violations:?}"); + assert!(violations[0].contains("other than the one-way deleted_at stamp")); + } +} + +#[test] +fn single_table_policy_holds_the_manifest_stamp_to_plain_conjuncts() { let stamp = r#"pub async fn mark_manifest_chunk_deleted() { - query("UPDATE community_deletion_manifest_keys SET deleted_at = now() WHERE request_id = $1 AND deleted_at IS NULL"); + query("UPDATE community_deletion_manifest_keys SET deleted_at = now() WHERE request_id = $1 AND chunk_no = $2 AND deleted_at IS NULL"); }"#; - assert!( - single_table_guard_violations(DELETION_STORE_PATH, stamp, &mut Default::default()) - .is_empty() - ); - let repeatable_stamp = stamp.replace(" AND deleted_at IS NULL", ""); + let mut hits: [Vec; 5] = Default::default(); + assert!(single_table_guard_violations(DELETION_STORE_PATH, stamp, &mut hits).is_empty()); + assert_eq!(hits[1], ["mark_manifest_chunk_deleted"]); + for (from, to) in [ + (" AND deleted_at IS NULL", ""), + ("deleted_at = now()", "deleted_at = now(), keys = $3"), + ("deleted_at = now()", "deleted_at = NULL"), + ("deleted_at IS NULL", "deleted_at IS NULL OR true"), + ( + "chunk_no = $2 AND deleted_at IS NULL", + "chunk_no IN (SELECT chunk_no FROM chunks WHERE request_id = $1 AND deleted_at IS NULL)", + ), + ("request_id = $1 AND ", ""), + ] { + let weakened = stamp.replace(from, to); + assert_ne!(weakened, stamp); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &weakened, &mut Default::default()) + .len(), + 1, + "{weakened}" + ); + } + let guarded_elsewhere = stamp.replace("mark_manifest_chunk_deleted", "restamp"); assert_eq!( single_table_guard_violations( DELETION_STORE_PATH, - &repeatable_stamp, + &guarded_elsewhere, &mut Default::default() ) .len(), 1 ); - let mutating_stamp = stamp.replace("deleted_at = now()", "deleted_at = now(), keys = $2"); assert_eq!( - single_table_guard_violations( - DELETION_STORE_PATH, - &mutating_stamp, - &mut Default::default() - ) - .len(), + single_table_guard_violations("buzz-relay/src/bypass.rs", stamp, &mut Default::default()) + .len(), 1 ); } From 88c6a3231022c1de19fd6cbcca8becbbc16a3341 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 10:05:57 -0400 Subject: [PATCH 6/8] buzz-db: pin every condition of the manifest stamp exemption Add negatives for a second chunk write and a conflict-upsert stamp, and require chunk_no = $2 so one chunk's purge cannot stamp the whole request. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/tests/observability_source.rs | 30 +++++++++++++------- 1 file changed, 19 insertions(+), 11 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 466597ec094..813880c6c65 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -2220,10 +2220,11 @@ fn single_table_guard_violations( && header.starts_with("pub async fn mark_manifest_chunk_deleted(") && manifest_writes == 1 && manifest_updates == 1 - && sets.len() == 1 - && sets[0].0.trim() == "deleted_at = now()" - && plain_guard(sets[0].1, "request_id = $1") - && plain_guard(sets[0].1, "deleted_at is null"); + && matches!(sets.as_slice(), [(set, predicate)] + if set.trim() == "deleted_at = now()" + && ["request_id = $1", "chunk_no = $2", "deleted_at is null"] + .iter() + .all(|guard| plain_guard(predicate, guard))); let guarded = code.contains("guard_manifest_chunk_write("); if deletion_stamp { rule_hits[1].push(name.to_owned()); @@ -2606,6 +2607,17 @@ fn single_table_policy_holds_the_manifest_stamp_to_plain_conjuncts() { "chunk_no IN (SELECT chunk_no FROM chunks WHERE request_id = $1 AND deleted_at IS NULL)", ), ("request_id = $1 AND ", ""), + ("chunk_no = $2 AND ", ""), + // A second write in the stamp function does not share its exemption. + ( + "deleted_at IS NULL\");", + "deleted_at IS NULL\"); query(\"DELETE FROM community_deletion_manifest_keys WHERE request_id = $1\");", + ), + // The stamp must be a plain UPDATE, not a conflict upsert. + ( + "UPDATE community_deletion_manifest_keys SET", + "INSERT INTO community_deletion_manifest_keys (request_id, chunk_no) VALUES ($1, $2) ON CONFLICT (request_id, chunk_no) DO UPDATE SET", + ), ] { let weakened = stamp.replace(from, to); assert_ne!(weakened, stamp); @@ -2616,14 +2628,10 @@ fn single_table_policy_holds_the_manifest_stamp_to_plain_conjuncts() { "{weakened}" ); } - let guarded_elsewhere = stamp.replace("mark_manifest_chunk_deleted", "restamp"); + let renamed_stamp = stamp.replace("mark_manifest_chunk_deleted", "restamp"); assert_eq!( - single_table_guard_violations( - DELETION_STORE_PATH, - &guarded_elsewhere, - &mut Default::default() - ) - .len(), + single_table_guard_violations(DELETION_STORE_PATH, &renamed_stamp, &mut Default::default()) + .len(), 1 ); assert_eq!( From 592e97158855182be01ca77a30bc2452f29fd0e5 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 15:44:51 -0400 Subject: [PATCH 7/8] test(buzz-db): detect ON CONFLICT( upserts at a token boundary sql_update_sets() matched only the spaced ' on conflict ' spelling, so ON CONFLICT(request_id) DO UPDATE ... slipped past the approval append-only rule and the manifest content-UPDATE rule. Match on conflict and do update set at identifier boundaries and add no-space negative fixtures for both tables plus a no-space DO NOTHING accept fixture. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/tests/observability_source.rs | 35 +++++++++++++------- 1 file changed, 23 insertions(+), 12 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 813880c6c65..18dc409cfd3 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1947,19 +1947,25 @@ const SINGLE_TABLE_GUARD_TABLES: [&str; 5] = [ ]; const DELETION_STORE_PATH: &str = "buzz-db/src/store/deletion.rs"; +/// The text after each occurrence of `marker` that sits on identifier +/// boundaries, so `on conflict` matches before a space or `(` alike. +fn sql_token_suffixes<'a>(source: &'a str, marker: &str) -> Vec<&'a str> { + let is_ident = |ch: char| ch.is_ascii_alphanumeric() || ch == '_'; + source + .match_indices(marker) + .filter_map(|(at, found)| { + let rest = &source[at + found.len()..]; + (!source[..at].chars().next_back().is_some_and(is_ident) + && !rest.chars().next().is_some_and(is_ident)) + .then_some(rest) + }) + .collect() +} + fn sql_table_suffixes<'a>(source: &'a str, verb: &str, table: &str) -> Vec<&'a str> { [format!("{verb} {table}"), format!("{verb} public.{table}")] .iter() - .flat_map(|marker| { - source.match_indices(marker).filter_map(|(at, found)| { - let before = &source[..at]; - let rest = &source[at + found.len()..]; - let is_ident = |ch: char| ch.is_ascii_alphanumeric() || ch == '_'; - (!before.chars().next_back().is_some_and(is_ident) - && !rest.chars().next().is_some_and(is_ident)) - .then_some(rest) - }) - }) + .flat_map(|marker| sql_token_suffixes(source, marker)) .collect() } @@ -1979,8 +1985,10 @@ fn sql_update_sets<'a>(source: &'a str, table: &str) -> Vec<(&'a str, &'a str)> .into_iter() .filter_map(|rest| { let statement = rest.split(['"', ';']).next()?; - let (_, conflict) = statement.split_once(" on conflict ")?; - conflict.split_once(" do update set ").map(|(_, set)| set) + let conflict = *sql_token_suffixes(statement, "on conflict").first()?; + sql_token_suffixes(conflict, "do update set") + .first() + .map(|set| set.trim_start()) }); updates .chain(conflict_updates) @@ -2510,6 +2518,7 @@ fn single_table_policy_rejects_approval_mutations() { "UPDATE community_deletion_approvals SET approved_by = $1 WHERE request_id = $2", "DELETE FROM community_deletion_approvals WHERE request_id = $1", "INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2) ON CONFLICT (request_id) DO UPDATE SET note = EXCLUDED.note", + "INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2) ON CONFLICT(request_id) DO UPDATE SET note = EXCLUDED.note", ] { let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); assert_eq!( @@ -2522,6 +2531,7 @@ fn single_table_policy_rejects_approval_mutations() { let append = r#"async fn approve() { query("INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2)"); query("INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2) ON CONFLICT (request_id) DO NOTHING"); + query("INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2) ON CONFLICT(request_id) DO NOTHING"); }"#; assert!( single_table_guard_violations(DELETION_STORE_PATH, append, &mut Default::default()) @@ -2580,6 +2590,7 @@ fn single_table_policy_rejects_manifest_content_updates() { for sql in [ "UPDATE community_deletion_manifest_keys SET keys = $1 WHERE request_id = $2", "INSERT INTO community_deletion_manifest_keys (request_id, chunk_no, keys) VALUES ($1, $2, $3) ON CONFLICT (request_id, chunk_no) DO UPDATE SET keys = EXCLUDED.keys", + "INSERT INTO community_deletion_manifest_keys (request_id, chunk_no, keys) VALUES ($1, $2, $3) ON CONFLICT(request_id, chunk_no) DO UPDATE SET keys = EXCLUDED.keys", ] { let source = format!("async fn append() {{ guard_manifest_chunk_write(); query(\"{sql}\"); }}"); let violations = From 11fb45ba1e171f3c215ac0ddf19b97d2914c0b29 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 15:50:55 -0400 Subject: [PATCH 8/8] test(buzz-db): detect SET( at a token boundary in UPDATE and DO UPDATE Match the SET keyword at identifier boundaries in plain UPDATEs and in ON CONFLICT ... DO UPDATE, so the multi-column SET(a, b) = ... spelling without a space is parsed like SET (a, b). Add no-space fixtures for a channels tuple UPDATE and an approval DO UPDATE SET(note) upsert. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/tests/observability_source.rs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 18dc409cfd3..d0941335e1f 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1975,11 +1975,10 @@ fn sql_update_sets<'a>(source: &'a str, table: &str) -> Vec<(&'a str, &'a str)> .into_iter() .filter_map(|rest| { let rest = rest.split(['"', ';']).next()?.trim_start(); - rest.strip_prefix("set ").or_else(|| { - // UPDATE table [AS] alias SET ... - let (alias, set) = rest.split_once(" set ")?; - (alias.split_whitespace().count() <= 2).then_some(set) - }) + // UPDATE table [[AS] alias] SET ..., including SET(a, b) = ... + let set = *sql_token_suffixes(rest, "set").first()?; + let alias = &rest[..rest.len() - set.len() - "set".len()]; + (alias.split_whitespace().count() <= 2).then_some(set.trim_start()) }); let conflict_updates = sql_table_suffixes(source, "insert into", table) .into_iter() @@ -2433,6 +2432,7 @@ fn single_table_policy_rejects_implicit_community_insert_columns() { fn single_table_policy_checks_tuple_assignments() { for (sql, reason) in [ ("UPDATE channels SET (name, community_id) = ($1, $2) WHERE id = $3", "channels.community_id UPDATE"), + ("UPDATE channels SET(name, community_id) = ($1, $2) WHERE id = $3", "channels.community_id UPDATE"), ("UPDATE communities SET (icon, deletion_state) = (NULL, 'active') WHERE id = $1 AND deletion_state = 'active'", "lifecycle UPDATE"), ("UPDATE community_deletion_requests SET (stage, community_id) = ($1, $2) WHERE id = $3", "identity/provenance UPDATE"), ("UPDATE community_deletion_requests SET (inventory_digest, inventory_manifest) = (NULL, NULL) WHERE id = $1", "inventory_frozen_at IS NULL"), @@ -2519,6 +2519,7 @@ fn single_table_policy_rejects_approval_mutations() { "DELETE FROM community_deletion_approvals WHERE request_id = $1", "INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2) ON CONFLICT (request_id) DO UPDATE SET note = EXCLUDED.note", "INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2) ON CONFLICT(request_id) DO UPDATE SET note = EXCLUDED.note", + "INSERT INTO community_deletion_approvals (request_id, note) VALUES ($1, $2) ON CONFLICT(request_id) DO UPDATE SET(note) = ROW(EXCLUDED.note)", ] { let source = format!("async fn bypass() {{ query(\"{sql}\"); }}"); assert_eq!(