diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index 50d1b20f45..c483bed88a 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -486,17 +486,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 3a3d4e6e4f..f63f5a4dbe 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -1337,6 +1337,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 * "#, ) @@ -1349,7 +1350,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) @@ -1645,11 +1646,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) @@ -1657,7 +1659,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) \ @@ -1861,52 +1870,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 } @@ -1940,32 +1934,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, @@ -2031,6 +2006,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) @@ -2058,6 +2035,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) @@ -2082,6 +2060,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)", @@ -2379,25 +2358,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, @@ -2521,6 +2489,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) @@ -2737,12 +2706,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) @@ -2778,32 +2742,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) \ @@ -3364,6 +3317,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, @@ -3378,6 +3368,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 @@ -4562,11 +4669,8 @@ mod postgres_tests { } 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() @@ -4788,6 +4892,390 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); (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(Some(&host), "operator", None, 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 f8969b0f65..3d1d8dc7b9 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1291,10 +1291,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; @@ -1313,6 +1326,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" @@ -1321,19 +1338,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. @@ -1344,7 +1349,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" @@ -1846,18 +1851,7 @@ mod tests { } fn production_event_writer_source_files() -> Vec { - 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() @@ -1866,7 +1860,7 @@ fn production_event_writer_source_files() -> Vec { 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); } } @@ -2229,3 +2223,352 @@ fn every_search_hydrator_preserves_the_exact_hit_timestamp() { ); } } + +// 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 = normalized_sql(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 = normalized_sql(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}" + ); + } +}