buzz-db: enforce five single-table guards in app writers

Phase 1 of BUZZ-318. The five single-table guard triggers stay for
mixed-version pods; their invariants now also hold in the application
writers, so a later PR can drop them.

- communities lifecycle: begin_quiescing, fence, purge_postgres and
  abort go through one transition_community state machine over a locked
  snapshot (deleted_at one-way, generation monotonic, deleted rows
  rejected). set_community_icon only updates an active, undeleted row.
- community_deletion_requests: both initial inventory freezes require
  an unfrozen, unblocked submitted row in the UPDATE predicate; the
  destructive-manifest freeze refuses a frozen row with missing JSON.
- community_deletion_manifest_keys: append, pre-freeze clear and
  verified cleanup check the request row via guard_manifest_chunk_write.
- channels.community_id and community_deletion_approvals hold by
  construction; a Rust source policy in observability_source.rs forbids
  retenanting, lifecycle writes outside the transition helper, community
  DELETE, request retargeting, approval UPDATE/DELETE and unguarded
  manifest writers.

Five PG tests each disable only their own trigger and exercise the app
path's rejected and permitted mutations.

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
This commit is contained in:
Elrond
2026-10-14 10:12:57 -04:00
committed by Luke Tornquist
parent e2dc54e22f
commit cb1bd16805
3 changed files with 990 additions and 153 deletions

View File

@@ -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(())
}

View File

@@ -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<DateTime<Utc>>) = 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<DateTime<Utc>>) = 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<DateTime<Utc>>)> = 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<DateTime<Utc>>,
archived_at: Option<DateTime<Utc>>,
}
enum CommunityTransition {
Quiesce,
Fence,
Tombstone {
generation: i64,
},
Abort {
restored_archived_at: Option<DateTime<Utc>>,
},
}
async fn lock_community_lifecycle(
tx: &mut Transaction<'_, Postgres>,
community: CommunityId,
) -> Result<CommunityLifecycle> {
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<i64> {
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<DateTime<Utc>>) =
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<Postgres>,
trigger_name: String,

View File

@@ -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<std::path::PathBuf>) {
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<PathBuf>) {
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<std::path::PathBuf> {
use std::path::{Path, PathBuf};
fn collect_rs_files(dir: &Path, out: &mut Vec<PathBuf>) {
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<std::path::PathBuf> {
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<String> {
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<CommunityId>, }",
"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}"
);
}
}