diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index e084d8ed176..f99fbb97e4d 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(()) } @@ -661,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) @@ -728,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 d673c5d8d7e..e588fcd436a 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 } @@ -1861,70 +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 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 - ))); - } - 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 } @@ -1972,6 +1948,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 +1977,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 +2002,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 +2299,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 +2427,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) @@ -2670,16 +2640,11 @@ 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 (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 +2680,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 +3269,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 +3320,125 @@ async fn lock_community_deletion( Ok(()) } +/// 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, + 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 +4473,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 +4553,462 @@ 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_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() { + 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..d0941335e1f 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,822 @@ 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"; + +/// 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| sql_token_suffixes(source, marker)) + .collect() +} + +// 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.split(['"', ';']).next()?.trim_start(); + // 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() + .filter_map(|rest| { + let statement = rest.split(['"', ';']).next()?; + 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) + .map(|set| set.split_once(" where ").unwrap_or((set, ""))) + .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: 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 =", + "", + ); + 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> { + set.match_indices('=') + .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, + // 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") { + 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 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")) + || (!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"); + } + 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) + }) + }) { + 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)| !plain_guard(predicate, "deletion_state = 'active'")) + { + reject("community UPDATE requires an active predicate in its statement"); + } + } + if sql_table_suffixes(&sql, "insert into", "communities") + .iter() + .any(|rest| { + 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 requires explicit columns without deletion lifecycle fields", + ); + } + if !sql_table_suffixes(&sql, "delete from", "communities").is_empty() { + reject("production community DELETE"); + } + 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 !plain_guard(predicate, "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 !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 { + rule_hits[4].push(name.to_owned()); + } + 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, upsert or DELETE"); + } + 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 { + // 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 + && 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()); + } 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()); + } else { + reject("manifest chunk INSERT or DELETE must use guard_manifest_chunk_write"); + } + } + } + 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, &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_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 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)", + ] { + 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 [ + "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} = COALESCE({column}, $2) 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}" + ); + // 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:?}" + ); + } +} + +#[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 = 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()) + .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 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 AND deletion_state = 'active'\"); }}"); + assert_eq!( + 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, + &mut Default::default() + ) + .is_empty()); + assert_eq!( + 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, &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, &mut Default::default()).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, &mut Default::default()) + .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", + "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!( + single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default()) + .len(), + 1, + "{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"); + 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)", + "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, &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, &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, &mut Default::default()) + .len(), + 1 + ); + let guarded = source.replace("query(", "guard_manifest_chunk_write(); query("); + 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, + &mut Default::default() + ) + .len(), + 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", + "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 chunk_no = $2 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 ", ""), + ("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); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &weakened, &mut Default::default()) + .len(), + 1, + "{weakened}" + ); + } + let renamed_stamp = stamp.replace("mark_manifest_chunk_deleted", "restamp"); + assert_eq!( + single_table_guard_violations(DELETION_STORE_PATH, &renamed_stamp, &mut Default::default()) + .len(), + 1 + ); + assert_eq!( + single_table_guard_violations("buzz-relay/src/bypass.rs", stamp, &mut Default::default()) + .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 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"); + 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, &mut rule_hits) + .into_iter() + .map(|violation| format!("{relative}: {violation}")), + ); + } + assert!( + 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() + .zip(writers) + .zip([7, 4, 15, 2, 4]) + { + assert!( + seen >= floor, + "{table}: expected at least {floor} writers, found {seen}" + ); + } +}