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

buzz-db: tolerate absent universal write-fence objects

Browse Source 
Expand step for retiring the 32 community_write_fence_* triggers and their
SQL functions (BUZZ-315). Relay pods auto-migrate at boot, so the drop
migration can only ship once every pod no longer depends on those objects.

- Serving startup/readiness and the deletion catalog check no longer require
  the fence functions or triggers; tenant-table equality, the deletion lock
  key and the community tombstone are still required.
- Live scoped-table discovery uses a Rust exclusion list instead of
  community_write_fence_excluded_table(), pinned equal to the SQL list while
  both exist.
- Serving-lease admission uses guard_community_write instead of
  assert_community_write_allowed; fenced/missing still map to AccessDenied.
- guard_community_write rejects non-READ-COMMITTED transactions, reading the
  isolation level from the existing advisory-lock statement (no extra round
  trip).
- The deletion executor stops comparing the legacy fenced_tables snapshot,
  but inventories still record live fences so older executors keep matching
  during this rollout.

No schema or migration change.

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
This commit is contained in:
Elrond
2026-10-14 10:39:13 -04:00
parent 67c4fe5e17
commit d06bab6b15
4 changed files with 251 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
ARCHITECTURE.md
View File

@@ -501,8 +501,9 @@ shape is an explicit transition, not a flag day:
- **Community fence (transition foundation):** runtime admission APIs now offer
one shared community deletion lock per transaction, while deletion lifecycle
transitions take the matching exclusive lock. A supported serving transaction
never spans communities; unmigrated paths still rely on trigger/function
backstops.
never spans communities. Serving leases use the same application admission.
Startup and deletion catalog validation tolerate the universal fence objects
being absent; inventories still record live fences for older executors.
- **Replica floor (transition foundation):** runtime now provides a shared
replica-floor lock helper for compliant channel-event writers, and the writer
probe handshake takes the exclusive counterpart before sampling `S`, scanning
@@ -513,7 +514,8 @@ shape is an explicit transition, not a flag day:
- **Dual enforcement (current):** application-owned lock+precheck paths run in
front of the existing trigger/function enforcement; commit-time trigger checks
remain authoritative during this phase. Removing those backstops is a separate
gated migration, not part of the lock foundation.
gated migration, after every pod tolerates their absence and no rollback below
that version is expected.
- **Role separation:** relay/runtime code owns admission and lock protocols;
operator maintenance/backfill workflows must either use those protocols or run
under explicit reviewed procedures that keep routing fences closed.

5
crates/buzz-db/src/runtime/admitted_tx.rs
View File

@@ -24,8 +24,9 @@ use crate::Result;
/// are the only ways it can end the transaction. Inside the crate, the
/// `conn()` accessor hands statements the connection; the type can't stop
/// crate code from issuing a raw `COMMIT` through it, so the database
/// community write fences remain the backstop. Dropping the value without
/// committing rolls back, as with [`sqlx::Transaction`].
/// community write fences remain the backstop while deployed. Internal writers
/// must preserve this boundary and use `commit()` for staged follow-up work.
/// Dropping the value without committing rolls back, as with [`sqlx::Transaction`].
///
/// These are live regressions for the guarantee. Event-write helpers accept an
/// admitted transaction:

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

@@ -110,6 +110,20 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
"workflows",
];
/// Control-plane tables with community provenance that the tenant purge does
/// not own. Keep this list separate from the expected tenant table manifest so
/// catalog validation still detects newly introduced tenant tables.
const EXCLUDED_SCOPED_TABLES: &[&str] = &[
"community_deletion_requests",
"community_deletion_approvals",
"community_deletion_checkpoints",
"community_serving_write_leases",
"community_deletion_executor_heartbeats",
"product_feedback",
"rate_limit_violations",
"operator_listener_outbox",
];
/// Foreign-key-safe child-before-parent order for the PostgreSQL purge.
pub const PURGE_SCOPED_TABLES: &[&str] = &[
"personal_read_frontiers",
@@ -370,7 +384,9 @@ pub struct SchemaManifest {
pub scoped_tables: Vec<String>,
/// Per-table row counts for the target.
pub row_counts: BTreeMap<String, i64>,
/// Sorted tables with the universal write-fence trigger.
/// Legacy fence snapshot retained for existing frozen inventory digests.
/// Populated from the live catalog while mixed-version executors need it.
#[serde(default)]
pub fenced_tables: Vec<String>,
}
@@ -1173,30 +1189,8 @@ impl DeletionStore {
)));
}
let required_fences = EXPECTED_SCOPED_TABLES
.iter()
.copied()
.map(str::to_owned)
.collect::<BTreeSet<_>>();
let live_fences = live_fenced_tables_on(&mut *conn).await?;
let missing_fences = required_fences
.difference(&live_fences)
.cloned()
.collect::<Vec<_>>();
if !missing_fences.is_empty() {
return Err(DbError::DeletionSafety(format!(
"community serving write fences missing: {}",
missing_fences.join(",")
)));
}
let required_objects_present: bool = sqlx::query_scalar(
"SELECT to_regprocedure('community_deletion_lock_key(uuid)') IS NOT NULL \
AND to_regprocedure('community_write_allowed(uuid)') IS NOT NULL \
AND (SELECT provolatile = 'v' FROM pg_proc \
WHERE oid = 'community_write_allowed(uuid)'::regprocedure) \
AND to_regprocedure('assert_community_write_allowed(uuid)') IS NOT NULL \
AND to_regprocedure('enforce_community_write_fence()') IS NOT NULL \
AND EXISTS (SELECT 1 FROM pg_trigger t \
JOIN pg_class c ON c.oid = t.tgrelid \
JOIN pg_proc p ON p.oid = t.tgfoid \
@@ -1208,15 +1202,15 @@ impl DeletionStore {
.await?;
if !required_objects_present {
return Err(DbError::DeletionSafety(
"community serving fence functions or tombstone trigger are missing".to_string(),
"community deletion lock function or tombstone trigger is missing".to_string(),
));
}
Ok(())
}
/// Validate the exact live scoped-table and write-fence catalog for destruction.
/// Validate the exact live scoped-table catalog for destruction.
///
/// Exact table and fence equality rejects unknown tenant data even
/// Exact table equality rejects unknown tenant data even
/// while unrelated SQLx migrations continue to advance. This pool-based
/// check is an early rejection only; destructive transactions revalidate
/// on their own connection under [`SCHEMA_DESTRUCTION_LOCK_KEY`].
@@ -3062,24 +3056,10 @@ impl DeletionStore {
let lease_seconds = i64::try_from(lease_duration.as_secs()).unwrap_or(i64::MAX);
self.bounded_serving_lease_sql(SERVING_WRITE_LEASE_SQL_TIMEOUT, async |connection| {
let mut tx = serving_lease_transaction(connection).await?;
// The assertion owns both the shared ordering lock and the supported
// READ COMMITTED check. The lease table is trigger-excluded, so this
// explicit admission is its database-enforced write fence.
if let Err(error) = sqlx::query("SELECT assert_community_write_allowed($1)")
.bind(community.as_uuid())
.execute(&mut *tx)
.await
{
if error.as_database_error().is_some_and(|database_error| {
database_error.code().as_deref() == Some("55000")
&& database_error.message().starts_with("community write")
}) {
return Err(DbError::AccessDenied(format!(
"community {community} is write-fenced or missing"
)));
}
return Err(error.into());
}
// Admit under the same shared lock as ordinary serving writes.
// The lease INSERT remains in this transaction and rechecks active
// state after admission, so a fence cannot interleave.
guard_community_write(&mut tx, community).await?;
let row = sqlx::query(
"INSERT INTO community_serving_write_leases \
(community_id, operation, owner, fence_generation, lease_until) \
@@ -3341,7 +3321,15 @@ pub async fn guard_community_write(
tx: &mut Transaction<'_, Postgres>,
community: CommunityId,
) -> Result<()> {
lock_community_deletion_shared(tx, community).await?;
let isolation = lock_community_deletion_shared(tx, community).await?;
// The lock statement may pin a snapshot in REPEATABLE READ/SERIALIZABLE.
// Refuse it before reading state; READ COMMITTED takes a fresh snapshot
// for the state query after any wait for the deletion lock.
if isolation != "read committed" {
return Err(DbError::AccessDenied(format!(
"community write admission requires read committed isolation, got {isolation}"
)));
}
let state: Option<String> = sqlx::query_scalar(
"SELECT deletion_state FROM communities WHERE id = $1 AND deleted_at IS NULL",
)
@@ -3362,15 +3350,18 @@ pub async fn guard_community_write(
pub(crate) async fn lock_community_deletion_shared(
tx: &mut Transaction<'_, Postgres>,
community: CommunityId,
) -> Result<()> {
crate::observability::observe_advisory_lock(
) -> Result<String> {
let (_, isolation): ((), String) = crate::observability::observe_advisory_lock(
crate::observability::LockType::Deletion,
sqlx::query("SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1))")
.bind(community.as_uuid())
.execute(&mut **tx),
sqlx::query_as(
"SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1)), \
current_setting('transaction_isolation')",
)
.bind(community.as_uuid())
.fetch_one(&mut **tx),
)
.await?;
Ok(())
Ok(isolation)
}
/// Take the shared schema/destruction advisory lock for the current
@@ -3419,22 +3410,6 @@ async fn validate_catalog_on(conn: &mut PgConnection) -> Result<()> {
)));
}
let fenced_tables = live_fenced_tables_on(conn).await?;
if fenced_tables != expected {
let missing = expected
.difference(&fenced_tables)
.cloned()
.collect::<Vec<_>>();
let unknown = fenced_tables
.difference(&expected)
.cloned()
.collect::<Vec<_>>();
return Err(DbError::DeletionSafety(format!(
"community deletion write-fence drift (missing={}, unknown={})",
missing.join(","),
unknown.join(",")
)));
}
Ok(())
}
@@ -3450,10 +3425,11 @@ async fn live_scoped_tables_on(conn: &mut PgConnection) -> Result<BTreeSet<Strin
AND NOT c.relispartition
AND a.attname = 'community_id'
AND NOT a.attisdropped
AND NOT community_write_fence_excluded_table(c.relname)
AND c.relname != ALL($1)
ORDER BY c.relname
"#,
)
.bind(EXCLUDED_SCOPED_TABLES)
.fetch_all(conn)
.await?;
Ok(rows.into_iter().collect())
@@ -3912,6 +3888,24 @@ fn bound_text(input: &str, max: usize) -> String {
mod tests {
use super::*;
#[test]
fn scoped_table_exclusions_match_sql_during_transition() {
let sql =
include_str!("../../../../schema/functions/community_write_fence_excluded_table.sql");
let sql_names = sql
.split("ARRAY[")
.nth(1)
.expect("SQL exclusion array")
.split(']')
.next()
.expect("SQL exclusion names")
.split(',')
.map(|name| name.trim().trim_matches('\''))
.collect::<BTreeSet<_>>();
assert_eq!(sql_names.len(), 8);
assert_eq!(sql_names, EXCLUDED_SCOPED_TABLES.iter().copied().collect());
}
fn empty_prefix(prefix: &str) -> PrefixManifest {
PrefixManifest {
prefix: prefix.to_string(),
@@ -4381,6 +4375,140 @@ mod postgres_tests {
(db, store)
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn migration_schema_fence_objects_are_optional_for_serving_and_deletion() {
let db = Db::new(&DbConfig {
database_url: crate::test_support::database_url(),
max_connections: 5,
min_connections: 0,
..DbConfig::default()
})
.await
.expect("connect isolated migration DB");
crate::migration::run_migrations_through(&db.pool, 60)
.await
.expect("migrate with fence objects present");
let store = db.deletion_store();
let community = db
.ensure_configured_community("optional-fences.example")
.await
.expect("community")
.id;
for objects_present in [true, false] {
if !objects_present {
// Inline the contract migration: B1 ships no schema changes.
sqlx::raw_sql(r#"
DROP TRIGGER IF EXISTS community_write_fence_api_tokens ON api_tokens;
DROP TRIGGER IF EXISTS community_write_fence_archived_identities ON archived_identities;
DROP TRIGGER IF EXISTS community_write_fence_artifact_heads ON artifact_heads;
DROP TRIGGER IF EXISTS community_write_fence_artifact_revisions ON artifact_revisions;
DROP TRIGGER IF EXISTS community_write_fence_audit_log ON audit_log;
DROP TRIGGER IF EXISTS community_write_fence_channel_members ON channel_members;
DROP TRIGGER IF EXISTS community_write_fence_channels ON channels;
DROP TRIGGER IF EXISTS community_write_fence_community_bans ON community_bans;
DROP TRIGGER IF EXISTS community_write_fence_delivery_log ON delivery_log;
DROP TRIGGER IF EXISTS community_write_fence_event_mentions ON event_mentions;
DROP TRIGGER IF EXISTS community_write_fence_events ON events;
DROP TRIGGER IF EXISTS community_write_fence_git_repo_names ON git_repo_names;
DROP TRIGGER IF EXISTS community_write_fence_join_policy_acceptances ON join_policy_acceptances;
DROP TRIGGER IF EXISTS community_write_fence_moderation_actions ON moderation_actions;
DROP TRIGGER IF EXISTS community_write_fence_moderation_reports ON moderation_reports;
DROP TRIGGER IF EXISTS community_write_fence_parameterized_event_watermarks ON parameterized_event_watermarks;
DROP TRIGGER IF EXISTS community_write_fence_personal_read_accounts ON personal_read_accounts;
DROP TRIGGER IF EXISTS community_write_fence_personal_read_frontiers ON personal_read_frontiers;
DROP TRIGGER IF EXISTS community_write_fence_pubkey_allowlist ON pubkey_allowlist;
DROP TRIGGER IF EXISTS community_write_fence_push_leases ON push_leases;
DROP TRIGGER IF EXISTS community_write_fence_push_match_queue ON push_match_queue;
DROP TRIGGER IF EXISTS community_write_fence_push_wake_outbox ON push_wake_outbox;
DROP TRIGGER IF EXISTS community_write_fence_reactions ON reactions;
DROP TRIGGER IF EXISTS community_write_fence_relay_invites ON relay_invites;
DROP TRIGGER IF EXISTS community_write_fence_relay_members ON relay_members;
DROP TRIGGER IF EXISTS community_write_fence_scheduled_workflow_fires ON scheduled_workflow_fires;
DROP TRIGGER IF EXISTS community_write_fence_subscriptions ON subscriptions;
DROP TRIGGER IF EXISTS community_write_fence_thread_metadata ON thread_metadata;
DROP TRIGGER IF EXISTS community_write_fence_users ON users;
DROP TRIGGER IF EXISTS community_write_fence_workflow_approvals ON workflow_approvals;
DROP TRIGGER IF EXISTS community_write_fence_workflow_runs ON workflow_runs;
DROP TRIGGER IF EXISTS community_write_fence_workflows ON workflows;
DROP FUNCTION attach_community_write_fence(REGCLASS);
DROP FUNCTION enforce_community_write_fence();
DROP FUNCTION assert_community_write_allowed(UUID);
DROP FUNCTION community_write_allowed(UUID);
DROP FUNCTION community_write_fence_excluded_table(NAME);
"#)
.execute(&db.pool)
.await
.expect("retire only universal fence objects");
}
store
.validate_serving_catalog()
.await
.expect("serving startup");
db.validate_deletion_serving_catalog_for_readiness(
tokio::time::Instant::now() + Duration::from_secs(5),
)
.await
.expect("readiness");
store.validate_catalog().await.expect("deletion catalog");
let inventory = store.inventory_schema(community).await.expect("inventory");
assert_eq!(inventory.scoped_tables.len(), EXPECTED_SCOPED_TABLES.len());
assert_eq!(
inventory.fenced_tables.len(),
if objects_present {
EXPECTED_SCOPED_TABLES.len()
} else {
0
}
);
let lease = store
.acquire_serving_write_lease(
community,
"optional_fences",
"test-owner",
DEFAULT_LEASE_DURATION,
)
.await
.expect("active serving lease");
store
.release_serving_write_lease(&lease)
.await
.expect("release lease");
for state in ["quiescing", "fenced", "tombstone"] {
crate::test_support::set_deletion_state(&db.pool, *community.as_uuid(), state)
.await;
assert!(
matches!(
store
.acquire_serving_write_lease(
community,
"denied",
"test-owner",
DEFAULT_LEASE_DURATION
)
.await,
Err(DbError::AccessDenied(_))
),
"{state} must map to AccessDenied with objects_present={objects_present}"
);
}
assert!(matches!(
store
.acquire_serving_write_lease(
CommunityId::from_uuid(Uuid::new_v4()),
"missing",
"test-owner",
DEFAULT_LEASE_DURATION
)
.await,
Err(DbError::AccessDenied(_))
));
crate::test_support::set_deletion_state(&db.pool, *community.as_uuid(), "active").await;
}
}
fn empty_prefix_manifest(prefix: String) -> PrefixManifest {
PrefixManifest {
prefix,
@@ -7080,7 +7208,7 @@ mod postgres_tests {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn write_assertion_rejects_pinned_snapshot_isolation_before_authorization() {
async fn app_admission_rejects_pinned_snapshot_isolation_before_authorization() {
let (db, _) = store().await;
let community = db
.ensure_configured_community(&format!(
@@ -7107,17 +7235,11 @@ mod postgres_tests {
.execute(&mut *tx)
.await
.expect("forge executor authorization");
let error = sqlx::query("SELECT assert_community_write_allowed($1)")
.bind(community.as_uuid())
.execute(&mut *tx)
let error = guard_community_write(&mut tx, community)
.await
.expect_err("pinned snapshot isolation must fail before authorization");
assert_eq!(
error
.as_database_error()
.and_then(|error| error.code())
.as_deref(),
Some("25000")
assert!(
matches!(error, DbError::AccessDenied(message) if message.contains("requires read committed"))
);
}
}

50
crates/buzz-deletion/src/lib.rs
View File

@@ -1409,6 +1409,18 @@ where
Ok(output)
}
fn validate_approved_schema(
live: &buzz_db::deletion::SchemaManifest,
frozen: &FrozenInventory,
) -> Result<()> {
if live.scoped_tables != frozen.schema.scoped_tables {
return Err(permanent(
"approved structural catalog drifted before fencing",
));
}
Ok(())
}
async fn execute_stage(
services: &Services,
claim: &ClaimedDeletion,
@@ -1467,13 +1479,7 @@ async fn execute_stage(
.inventory_schema(request.community_id)
.await?;
let frozen = validate_frozen_inventory(request)?;
if live_schema.scoped_tables != frozen.schema.scoped_tables
|| live_schema.fenced_tables != frozen.schema.fenced_tables
{
return Err(permanent(
"approved structural catalog drifted before fencing",
));
}
validate_approved_schema(&live_schema, &frozen)?;
services.store.begin_quiescing(&token).await?;
match services.store.fence(&token).await {
Ok(_) => {}
@@ -2417,6 +2423,36 @@ mod postgres_tests {
assert_eq!(fenced.stage, DeletionStage::Fenced);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn approved_legacy_fence_inventory_accepts_current_catalog() {
let (_, services, mut claim) = claimed_test_deletion("deletion-legacy-fences").await;
let mut frozen = validate_frozen_inventory(&claim.request).expect("frozen inventory");
frozen.schema.fenced_tables = buzz_db::deletion::EXPECTED_SCOPED_TABLES
.iter()
.map(|table| (*table).to_owned())
.collect();
claim.request.inventory_digest = Some(hex::encode(frozen.digest().expect("legacy digest")));
claim.request.inventory_manifest =
Some(serde_json::to_value(&frozen).expect("encode legacy inventory"));
let decoded =
validate_frozen_inventory(&claim.request).expect("legacy digest remains valid");
let mut live = services
.store
.inventory_schema(claim.request.community_id)
.await
.expect("live catalog");
// Simulate the retired catalog while retaining the legacy digest above.
// B1 itself must still report live fences for old executors during rollout.
live.fenced_tables.clear();
validate_approved_schema(&live, &decoded)
.expect("legacy fence list is no longer structural");
live.scoped_tables.push("unknown_tenant_table".to_string());
assert!(validate_approved_schema(&live, &decoded).is_err());
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn frozen_inventory_digest_and_storage_ownership_fail_closed() {