jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-11 20:01:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

buzz-db: pin the remaining single-table guard halves in source policy

Browse Source 
Address review on the single-table guard writers:

- Require deletion_state = 'active' on every non-lifecycle communities
  UPDATE and conflict upsert; add it to archive/unarchive UPDATEs, which
  already lock an active row.
- Require the inventory and destructive-storage one-way freeze
  predicates on every request UPDATE that assigns those columns.
- Pin abort's generation increment with a trigger-disabled PG test
  covering quiescing/fenced and archived/unarchived communities.
- Parse ON CONFLICT DO UPDATE SET, tuple SET, and column-less community
  inserts; require AND deleted_at IS NULL on the manifest stamp.
- Assert exact production rule-hit sets so a parser that stops matching
  fails.
- Correct the lifecycle lock doc comment and format the fence body.

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
This commit is contained in:
Elrond
2026-10-14 11:01:20 -04:00
committed by Luke Tornquist
parent cb1bd16805
commit 52645266b9
3 changed files with 521 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
crates/buzz-db/src/store/community.rs
View File

@@ -824,7 +824,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)
@@ -891,10 +891,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 {

176
crates/buzz-db/src/store/deletion.rs
View File

@@ -1914,51 +1914,52 @@ impl DeletionStore {
.await?;
transaction_timer
.observe(async {
verify_lease(&mut tx, token, DeletionStage::Approved).await?;
lock_community_deletion(&mut tx, token.community_id).await?;
verify_lease(&mut tx, token, DeletionStage::Approved).await?;
let active_serving_writes = sqlx::query(
"SELECT count(*)::BIGINT AS active_count, \
COALESCE(array_agg(DISTINCT operation ORDER BY operation), ARRAY[]::TEXT[]) AS operations \
FROM community_serving_write_leases \
WHERE community_id = $1 AND lease_until >= now()",
)
.bind(token.community_id.as_uuid())
.fetch_one(&mut *tx)
.await?;
let active_count: i64 = active_serving_writes.try_get("active_count")?;
if active_count > 0 {
return Err(DbError::ServingWritesNotDrained {
community_id: *token.community_id.as_uuid(),
active_count,
operations: active_serving_writes.try_get("operations")?,
});
}
let community = lock_community_lifecycle(&mut tx, token.community_id).await?;
let generation = transition_community(
&mut tx,
token.community_id,
community,
CommunityTransition::Fence,
).await?;
advance_request_tx(
&mut tx,
token,
DeletionStage::Approved,
DeletionStage::Fenced,
Some(generation),
)
.await?;
checkpoint_completed_tx(
&mut tx,
token,
DeletionStage::Approved,
"activate_fence",
serde_json::json!({"fence_generation": generation}),
)
.await?;
tx.commit().await?;
Ok(generation)
verify_lease(&mut tx, token, DeletionStage::Approved).await?;
lock_community_deletion(&mut tx, token.community_id).await?;
verify_lease(&mut tx, token, DeletionStage::Approved).await?;
let active_serving_writes = sqlx::query(
"SELECT count(*)::BIGINT AS active_count, \
COALESCE(array_agg(DISTINCT operation ORDER BY operation), ARRAY[]::TEXT[]) AS operations \
FROM community_serving_write_leases \
WHERE community_id = $1 AND lease_until >= now()",
)
.bind(token.community_id.as_uuid())
.fetch_one(&mut *tx)
.await?;
let active_count: i64 = active_serving_writes.try_get("active_count")?;
if active_count > 0 {
return Err(DbError::ServingWritesNotDrained {
community_id: *token.community_id.as_uuid(),
active_count,
operations: active_serving_writes.try_get("operations")?,
});
}
let community = lock_community_lifecycle(&mut tx, token.community_id).await?;
let generation = transition_community(
&mut tx,
token.community_id,
community,
CommunityTransition::Fence,
)
.await?;
advance_request_tx(
&mut tx,
token,
DeletionStage::Approved,
DeletionStage::Fenced,
Some(generation),
)
.await?;
checkpoint_completed_tx(
&mut tx,
token,
DeletionStage::Approved,
"activate_fence",
serde_json::json!({"fence_generation": generation}),
)
.await?;
tx.commit().await?;
Ok(generation)
})
.await
}
@@ -2702,9 +2703,9 @@ impl DeletionStore {
.await?
.map(CommunityId::from_uuid)
.ok_or_else(|| DbError::NotFound(format!("community deletion {request_id}")))?;
// Every lifecycle transition takes the community advisory lock before any
// Quiesce, fence, and abort take the community advisory lock before any
// row lock, then the community row before the request row. Inverting
// either order lets abort and the executor deadlock each other.
// either order lets abort and those executor transitions deadlock.
lock_community_deletion(&mut tx, community_id).await?;
let community = lock_community_lifecycle(&mut tx, community_id).await?;
let row = sqlx::query("SELECT * FROM community_deletion_requests WHERE id = $1 FOR UPDATE")
@@ -3368,8 +3369,10 @@ async fn lock_community_deletion(
Ok(())
}
/// The row snapshot is taken under `FOR UPDATE` after the community deletion
/// advisory lock. All lifecycle writers use it in the same transaction.
/// The lifecycle row is locked `FOR UPDATE` in the caller's transaction.
/// Quiesce, fence, and abort also hold the community deletion advisory lock.
/// `purge_postgres` does not; its tombstone transition fails if a concurrent
/// abort won the row lock and changed the lifecycle state.
struct CommunityLifecycle {
state: String,
generation: i64,
@@ -5023,6 +5026,81 @@ DROP FUNCTION community_write_fence_excluded_table(NAME);
.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<DateTime<Utc>> =
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<DateTime<Utc>>,
Option<DateTime<Utc>>,
) = 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() {

451
crates/buzz-db/tests/observability_source.rs
View File

@@ -2252,34 +2252,69 @@ fn sql_table_suffixes<'a>(source: &'a str, verb: &str, table: &str) -> Vec<&'a s
.collect()
}
fn sql_update_sets<'a>(source: &'a str, table: &str) -> Vec<&'a str> {
sql_table_suffixes(source, "update", table)
// Keep SET and WHERE from the same statement, including an INSERT's conflict UPDATE.
fn sql_update_sets<'a>(source: &'a str, table: &str) -> Vec<(&'a str, &'a str)> {
let updates = sql_table_suffixes(source, "update", table)
.into_iter()
.filter_map(|rest| {
let rest = rest.trim_start();
let set = rest.strip_prefix("set ").or_else(|| {
let rest = rest.split(['"', ';']).next()?.trim_start();
rest.strip_prefix("set ").or_else(|| {
// UPDATE table [AS] alias SET ...
let (alias, set) = rest.split_once(" set ")?;
(alias.split_whitespace().count() <= 2).then_some(set)
})?;
Some(set.split_once(" where ").map_or(set, |(set, _)| set))
})
})
});
let conflict_updates = sql_table_suffixes(source, "insert into", table)
.into_iter()
.filter_map(|rest| {
let statement = rest.split(['"', ';']).next()?;
let (_, conflict) = statement.split_once(" on conflict ")?;
conflict.split_once(" do update set ").map(|(_, set)| set)
});
updates
.chain(conflict_updates)
.map(|set| set.split_once(" where ").unwrap_or((set, "")))
.collect()
}
fn sql_assigned_columns(set: &str) -> Vec<&str> {
set.match_indices('=')
.filter_map(|(at, _)| {
set[..at]
.trim_end()
.rsplit(|ch: char| !(ch.is_ascii_alphanumeric() || ch == '_'))
.flat_map(|(at, _)| -> Vec<&str> {
let left = set[..at].trim_end();
if left.ends_with(')') {
// Tuple assignment: SET (a, b) = (...).
let mut depth = 0;
for (at, ch) in left.char_indices().rev() {
match ch {
')' => depth += 1,
'(' => {
depth -= 1;
if depth == 0 {
return left[at + 1..left.len() - 1]
.split(',')
.map(str::trim)
.collect();
}
}
_ => {}
}
}
}
left.rsplit(|ch: char| !(ch.is_ascii_alphanumeric() || ch == '_'))
.next()
.filter(|column| !column.is_empty())
.into_iter()
.collect()
})
.collect()
}
fn single_table_guard_violations(path: &str, production: &str) -> Vec<String> {
fn single_table_guard_violations(
path: &str,
production: &str,
// Lifecycle exemption, stamp exemption, chunk guard, active predicate, request freezes.
rule_hits: &mut [Vec<String>; 5],
) -> Vec<String> {
let mut violations = Vec::new();
let code = mask_comments_and_literals(production);
if let Some((_, fields)) = code.split_once("struct ChannelUpdate") {
@@ -2305,10 +2340,19 @@ fn single_table_guard_violations(path: &str, production: &str) -> Vec<String> {
let sql = normalized_sql(function);
let code = mask_comments_and_literals(function);
let mut reject = |reason: &str| violations.push(format!("{header}: {reason}"));
let name = header
.split_once("fn ")
.expect("function name")
.1
.split('(')
.next()
.expect("function arguments");
let lifecycle_writer =
path == DELETION_STORE_PATH && header.starts_with("async fn transition_community(");
let channel_sets = sql_update_sets(&sql, "channels");
if channel_sets
.iter()
.any(|set| sql_assigned_columns(set).contains(&"community_id"))
.any(|(set, _)| sql_assigned_columns(set).contains(&"community_id"))
|| (!channel_sets.is_empty()
&& sql.split("set_parts.push(").skip(1).any(|part| {
let assignment = part.split(';').next().expect("dynamic SET fragment");
@@ -2317,54 +2361,103 @@ fn single_table_guard_violations(path: &str, production: &str) -> Vec<String> {
{
reject("channels.community_id UPDATE");
}
if sql_update_sets(&sql, "communities").iter().any(|set| {
let community_sets = sql_update_sets(&sql, "communities");
if community_sets.iter().any(|(set, _)| {
sql_assigned_columns(set).iter().any(|column| {
["deletion_state", "deletion_fence_generation", "deleted_at"].contains(column)
})
}) && !(path == DELETION_STORE_PATH
&& header.starts_with("async fn transition_community("))
{
reject("community lifecycle UPDATE outside transition_community");
}) {
if lifecycle_writer {
rule_hits[0].push(name.to_owned());
} else {
reject("community lifecycle UPDATE outside transition_community");
}
}
if !community_sets.is_empty() && !lifecycle_writer {
rule_hits[3].push(name.to_owned());
if community_sets
.iter()
.any(|(_, predicate)| !predicate.contains("deletion_state = 'active'"))
{
reject("community UPDATE requires an active predicate in its statement");
}
}
if sql_table_suffixes(&sql, "insert into", "communities")
.iter()
.any(|rest| {
rest.trim_start().strip_prefix('(').is_some_and(|columns| {
columns
.split(')')
.next()
.expect("INSERT columns")
.split(',')
.any(|column| {
["deletion_state", "deletion_fence_generation", "deleted_at"]
.contains(&column.trim())
})
})
let Some(columns) = rest.trim_start().strip_prefix('(') else {
return true;
};
columns
.split(')')
.next()
.expect("INSERT columns")
.split(',')
.any(|column| {
["deletion_state", "deletion_fence_generation", "deleted_at"]
.contains(&column.trim())
})
})
{
reject("community creation sets deletion lifecycle columns");
reject(
"community creation requires explicit columns without deletion lifecycle fields",
);
}
if !sql_table_suffixes(&sql, "delete from", "communities").is_empty() {
reject("production community DELETE");
}
if sql_update_sets(&sql, "community_deletion_requests")
.iter()
.any(|set| {
sql_assigned_columns(set).iter().any(|column| {
[
"community_id",
"community_host",
"request_origin",
"owner_pubkey",
"mediating_operator_pubkey",
"acknowledgement_version",
]
.contains(column)
})
let request_sets = sql_update_sets(&sql, "community_deletion_requests");
if request_sets.iter().any(|(set, _)| {
sql_assigned_columns(set).iter().any(|column| {
[
"community_id",
"community_host",
"request_origin",
"owner_pubkey",
"mediating_operator_pubkey",
"acknowledgement_version",
]
.contains(column)
})
{
}) {
reject("deletion request identity/provenance UPDATE");
}
let mut freezes_inventory = false;
let mut freezes_destructive_storage = false;
for (set, predicate) in &request_sets {
let columns = sql_assigned_columns(set);
if columns.iter().any(|column| {
[
"schema_manifest",
"storage_manifest",
"inventory_manifest",
"inventory_digest",
"inventory_frozen_at",
]
.contains(column)
}) {
freezes_inventory = true;
if !predicate.contains("inventory_frozen_at is null") {
reject(
"request inventory UPDATE requires inventory_frozen_at IS NULL in WHERE",
);
}
}
if columns
.iter()
.any(|column| column.starts_with("destructive_storage_"))
{
freezes_destructive_storage = true;
if !predicate.contains(
"destructive_storage_frozen_at is null or destructive_storage_manifest =",
) {
reject("request destructive storage UPDATE requires the frozen-or-equal guard in WHERE");
}
}
}
if freezes_inventory || freezes_destructive_storage {
rule_hits[4].push(name.to_owned());
}
if ["update", "delete from"]
.iter()
.any(|verb| !sql_table_suffixes(&sql, verb, "community_deletion_approvals").is_empty())
@@ -2380,7 +2473,13 @@ fn single_table_guard_violations(path: &str, production: &str) -> Vec<String> {
let deletion_stamp = header.starts_with("pub async fn mark_manifest_chunk_deleted(")
&& manifest_writes == 1
&& sets.len() == 1
&& sql_assigned_columns(sets[0]) == ["deleted_at"];
&& sql_assigned_columns(sets[0].0) == ["deleted_at"]
&& sets[0].1.contains("and deleted_at is null");
if deletion_stamp {
rule_hits[1].push(name.to_owned());
} else if code.contains("guard_manifest_chunk_write(") {
rule_hits[2].push(name.to_owned());
}
if path != DELETION_STORE_PATH
|| !(code.contains("guard_manifest_chunk_write(") || deletion_stamp)
{
@@ -2401,40 +2500,206 @@ fn single_table_policy_rejects_channel_retenanting() {
"pub struct ChannelUpdate { pub community: CommunityId, }",
] {
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, source).len(),
single_table_guard_violations(DELETION_STORE_PATH, source, &mut Default::default())
.len(),
1,
"{source}"
);
}
}
#[test]
fn single_table_policy_requires_active_community_updates() {
let source = r#"async fn bypass() {
query("UPDATE communities SET icon = NULL WHERE id = $1");
}"#;
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, source, &mut Default::default()).len(),
1
);
let guarded = source.replace(
"WHERE id = $1",
"WHERE id = $1 AND deletion_state = 'active'",
);
assert!(
single_table_guard_violations(DELETION_STORE_PATH, &guarded, &mut Default::default())
.is_empty()
);
// A predicate in a SELECT or another UPDATE does not admit this statement.
let separate = source.replace(
"query(",
"query(\"SELECT id FROM communities WHERE deletion_state = 'active'\"); query(",
);
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &separate, &mut Default::default())
.len(),
1
);
let two_updates = r#"async fn bypass() {
query("UPDATE communities SET icon = NULL WHERE id = $1 AND deletion_state = 'active'");
query("UPDATE communities SET icon = NULL WHERE id = $2");
}"#;
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, two_updates, &mut Default::default())
.len(),
1
);
}
#[test]
fn single_table_policy_requires_inventory_freeze_predicate() {
for column in [
"schema_manifest",
"storage_manifest",
"inventory_manifest",
"inventory_digest",
"inventory_frozen_at",
] {
let source = format!("async fn bypass() {{ query(\"UPDATE community_deletion_requests SET {column} = NULL WHERE id = $1\"); }}");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default())
.len(),
1,
"{column}"
);
let guarded = source.replace(
"WHERE id = $1",
"WHERE id = $1 AND inventory_frozen_at IS NULL",
);
assert!(
single_table_guard_violations(DELETION_STORE_PATH, &guarded, &mut Default::default())
.is_empty(),
"{column}"
);
}
}
#[test]
fn single_table_policy_requires_destructive_freeze_predicate() {
for column in [
"destructive_storage_manifest",
"destructive_storage_frozen_at",
] {
let source = format!("async fn bypass() {{ query(\"UPDATE community_deletion_requests SET {column} = NULL WHERE id = $1\"); }}");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default())
.len(),
1,
"{column}"
);
let guarded = source.replace("WHERE id = $1", "WHERE id = $1 AND (destructive_storage_frozen_at IS NULL OR destructive_storage_manifest = $2)");
assert!(
single_table_guard_violations(DELETION_STORE_PATH, &guarded, &mut Default::default())
.is_empty(),
"{column}"
);
}
}
#[test]
fn single_table_policy_checks_conflict_updates() {
for (sql, reason) in [
("INSERT INTO communities (host) VALUES ($1) ON CONFLICT (host) DO UPDATE SET icon = NULL", "requires an active predicate"),
("INSERT INTO communities (host) VALUES ($1) ON CONFLICT (host) DO UPDATE SET deleted_at = now() WHERE communities.deletion_state = 'active'", "lifecycle UPDATE"),
("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET community_id = $2", "identity/provenance UPDATE"),
("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET inventory_digest = NULL", "inventory_frozen_at IS NULL"),
("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET destructive_storage_manifest = NULL", "frozen-or-equal guard"),
] {
let source = format!("async fn bypass() {{ query(\"{sql}\"); }}");
let violations = single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default());
assert!(violations.iter().any(|violation| violation.contains(reason)), "{sql}: {violations:?}");
}
let source = r#"async fn guarded() {
query("INSERT INTO communities (host) VALUES ($1) ON CONFLICT (host) DO UPDATE SET host = communities.host WHERE communities.deletion_state = 'active'");
query("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET inventory_digest = $2 WHERE inventory_frozen_at IS NULL");
query("INSERT INTO community_deletion_requests (id) VALUES ($1) ON CONFLICT (id) DO UPDATE SET destructive_storage_manifest = $2 WHERE destructive_storage_frozen_at IS NULL OR destructive_storage_manifest = $2");
}"#;
assert!(
single_table_guard_violations(DELETION_STORE_PATH, source, &mut Default::default())
.is_empty()
);
}
#[test]
fn single_table_policy_rejects_implicit_community_insert_columns() {
for sql in [
"INSERT INTO communities VALUES ($1, $2)",
"INSERT INTO communities SELECT * FROM staging",
] {
let source = format!("async fn bypass() {{ query(\"{sql}\"); }}");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default())
.len(),
1,
"{sql}"
);
}
let explicit = r#"async fn create() { query("INSERT INTO communities (host) SELECT host FROM staging"); }"#;
assert!(
single_table_guard_violations(DELETION_STORE_PATH, explicit, &mut Default::default())
.is_empty()
);
}
#[test]
fn single_table_policy_checks_tuple_assignments() {
for (sql, reason) in [
("UPDATE channels SET (name, community_id) = ($1, $2) WHERE id = $3", "channels.community_id UPDATE"),
("UPDATE communities SET (icon, deletion_state) = (NULL, 'active') WHERE id = $1 AND deletion_state = 'active'", "lifecycle UPDATE"),
("UPDATE community_deletion_requests SET (stage, community_id) = ($1, $2) WHERE id = $3", "identity/provenance UPDATE"),
("UPDATE community_deletion_requests SET (inventory_digest, inventory_manifest) = (NULL, NULL) WHERE id = $1", "inventory_frozen_at IS NULL"),
("UPDATE community_deletion_requests SET (destructive_storage_manifest, destructive_storage_frozen_at) = (NULL, NULL) WHERE id = $1", "frozen-or-equal guard"),
] {
let source = format!("async fn bypass() {{ query(\"{sql}\"); }}");
let violations = single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default());
assert!(violations.iter().any(|violation| violation.contains(reason)), "{sql}: {violations:?}");
}
let guarded = r#"async fn freeze() { query("UPDATE community_deletion_requests SET (inventory_manifest, inventory_digest) = ($2, $3) WHERE id = $1 AND inventory_frozen_at IS NULL"); }"#;
assert!(
single_table_guard_violations(DELETION_STORE_PATH, guarded, &mut Default::default())
.is_empty()
);
}
#[test]
fn single_table_policy_rejects_community_lifecycle_bypasses() {
for column in ["deletion_state", "deletion_fence_generation", "deleted_at"] {
let source = format!("async fn bypass() {{ query(\"UPDATE communities SET {column} = $1 WHERE id = $2\"); }}");
let source = format!("async fn bypass() {{ query(\"UPDATE communities SET {column} = $1 WHERE id = $2 AND deletion_state = 'active'\"); }}");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &source).len(),
single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default())
.len(),
1,
"{column}"
);
let transition = source.replace("bypass", "transition_community");
assert!(single_table_guard_violations(DELETION_STORE_PATH, &transition).is_empty());
assert!(single_table_guard_violations(
DELETION_STORE_PATH,
&transition,
&mut Default::default()
)
.is_empty());
assert_eq!(
single_table_guard_violations("buzz-relay/src/bypass.rs", &transition).len(),
single_table_guard_violations(
"buzz-relay/src/bypass.rs",
&transition,
&mut Default::default()
)
.len(),
1
);
}
for column in ["deletion_state", "deletion_fence_generation", "deleted_at"] {
let source = format!("async fn bypass() {{ query(\"INSERT INTO communities (id, {column}) VALUES ($1, $2)\"); }}");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &source).len(),
single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default())
.len(),
1,
"{column}"
);
}
let deletion = r#"async fn bypass() { query("DELETE FROM communities WHERE id = $1"); }"#;
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, deletion).len(),
single_table_guard_violations(DELETION_STORE_PATH, deletion, &mut Default::default()).len(),
1
);
}
@@ -2451,7 +2716,8 @@ fn single_table_policy_rejects_request_retargeting() {
] {
let source = format!("async fn bypass() {{ query(\"UPDATE community_deletion_requests SET {column} = $1 WHERE id = $2\"); }}");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &source).len(),
single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default())
.len(),
1,
"{column}"
);
@@ -2466,7 +2732,8 @@ fn single_table_policy_rejects_approval_mutations() {
] {
let source = format!("async fn bypass() {{ query(\"{sql}\"); }}");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &source).len(),
single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default())
.len(),
1,
"{sql}"
);
@@ -2482,35 +2749,66 @@ fn single_table_policy_rejects_unguarded_manifest_writers() {
] {
let source = format!("async fn bypass() {{ query(\"{sql}\"); }}");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &source).len(),
single_table_guard_violations(DELETION_STORE_PATH, &source, &mut Default::default())
.len(),
1,
"{sql}"
);
// A guard in another function or a literal does not admit this writer.
let mixed = format!("async fn guarded() {{ guard_manifest_chunk_write(); }}\n{source}");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &mixed).len(),
single_table_guard_violations(DELETION_STORE_PATH, &mixed, &mut Default::default())
.len(),
1
);
let fake = source.replace("query(", "let _ = \"guard_manifest_chunk_write(\"; query(");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &fake).len(),
single_table_guard_violations(DELETION_STORE_PATH, &fake, &mut Default::default())
.len(),
1
);
let guarded = source.replace("query(", "guard_manifest_chunk_write(); query(");
assert!(single_table_guard_violations(DELETION_STORE_PATH, &guarded).is_empty());
assert!(single_table_guard_violations(
DELETION_STORE_PATH,
&guarded,
&mut Default::default()
)
.is_empty());
assert_eq!(
single_table_guard_violations("buzz-relay/src/bypass.rs", &guarded).len(),
single_table_guard_violations(
"buzz-relay/src/bypass.rs",
&guarded,
&mut Default::default()
)
.len(),
1
);
}
let stamp = r#"pub async fn mark_manifest_chunk_deleted() {
query("UPDATE community_deletion_manifest_keys SET deleted_at = now() WHERE request_id = $1");
query("UPDATE community_deletion_manifest_keys SET deleted_at = now() WHERE request_id = $1 AND deleted_at IS NULL");
}"#;
assert!(single_table_guard_violations(DELETION_STORE_PATH, stamp).is_empty());
assert!(
single_table_guard_violations(DELETION_STORE_PATH, stamp, &mut Default::default())
.is_empty()
);
let repeatable_stamp = stamp.replace(" AND deleted_at IS NULL", "");
assert_eq!(
single_table_guard_violations(
DELETION_STORE_PATH,
&repeatable_stamp,
&mut Default::default()
)
.len(),
1
);
let mutating_stamp = stamp.replace("deleted_at = now()", "deleted_at = now(), keys = $2");
assert_eq!(
single_table_guard_violations(DELETION_STORE_PATH, &mutating_stamp).len(),
single_table_guard_violations(
DELETION_STORE_PATH,
&mutating_stamp,
&mut Default::default()
)
.len(),
1
);
}
@@ -2530,6 +2828,7 @@ fn every_production_single_table_guard_writer_preserves_its_invariants() {
}
}
let mut writers = [0usize; 5];
let mut rule_hits: [Vec<String>; 5] = Default::default();
let mut violations = Vec::new();
for path in files {
let source = std::fs::read_to_string(&path).expect("read production source");
@@ -2551,7 +2850,7 @@ fn every_production_single_table_guard_writer_preserves_its_invariants() {
.to_string_lossy()
.replace('\\', "/");
violations.extend(
single_table_guard_violations(&relative, &production)
single_table_guard_violations(&relative, &production, &mut rule_hits)
.into_iter()
.map(|violation| format!("{relative}: {violation}")),
);
@@ -2560,6 +2859,30 @@ fn every_production_single_table_guard_writer_preserves_its_invariants() {
violations.is_empty(),
"single-table guard policy: {violations:#?}"
);
// Exact rule hits pin the parser to the known production seams.
for (hits, expected) in rule_hits.iter_mut().zip([
vec!["transition_community"],
vec!["mark_manifest_chunk_deleted"],
vec![
"append_manifest_key_chunk",
"clear_manifest_key_chunks",
"mark_logically_verified",
],
vec![
"archive_community_owned_by",
"ensure_configured_community_with_operation",
"set_community_icon",
"unarchive_community_owned_by",
],
vec![
"complete_owner_preparation",
"freeze_destructive_storage_manifest",
"freeze_inventory",
],
]) {
hits.sort();
assert_eq!(*hits, expected, "production single-table rule hits");
}
// Lower bounds pin discovery without fixing the exact number of writers.
for ((table, seen), floor) in SINGLE_TABLE_GUARD_TABLES
.iter()