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

Merge #8318 (fence sweeps tenant-admitted) into the fence-retire stack base

Browse Source 
* origin/elrond/fence-sweeps-tenant-admitted:
  fix(db): pin sweep admission with race tests; continue past failing tenants
  fix(db): admit fleet sweeps per tenant instead of in-SQL fence predicate
This commit is contained in:
Elrond
2026-10-16 09:58:47 -04:00
parent 2179773391 2c15121789
commit 0b09dc28bb
11 changed files with 1015 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
crates/buzz-db/src/lib.rs
View File

@@ -60,9 +60,9 @@ pub const DB_CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionO
pub const DB_CONNECTION_RAW_SERIES_PER_POD: usize = runtime::CONNECTION_RAW_SERIES_PER_POD;
pub(crate) use runtime::{
begin_community_event_write_transaction,
begin_community_event_write_transaction_with_legacy_metrics, begin_community_write_transaction,
insert_mentions_in_transaction, observability, route_proof, ReadSessionInner, RouteDecision,
RoutePredicate,
begin_community_event_write_transaction_with_legacy_metrics, begin_community_sweep_transaction,
begin_community_write_transaction, insert_mentions_in_transaction, observability, route_proof,
ReadSessionInner, RouteDecision, RoutePredicate,
};
pub use store::{
admin_moderation, allowlist, api_token, archived_identities, artifact, channel,

14
crates/buzz-db/src/runtime/mod.rs
View File

@@ -202,6 +202,20 @@ pub(crate) async fn begin_community_write_transaction(
.await
}
/// Begin a fleet sweep's tenant-local mutation, skipping communities whose
/// deletion lifecycle no longer admits writes. All other failures propagate.
pub(crate) async fn begin_community_sweep_transaction(
pool: &PgPool,
community: CommunityId,
operation: observability::WriterOperation,
) -> Result<Option<AdmittedTx>> {
match begin_community_write_transaction(pool, community, operation).await {
Ok(tx) => Ok(Some(tx)),
Err(DbError::AccessDenied(_)) => Ok(None),
Err(error) => Err(error),
}
}
/// Database handle. Clone is cheap (Arc-backed pool).
#[derive(Clone, Debug)]
pub struct Db {

67
crates/buzz-db/src/runtime/push_enqueue.rs
View File

@@ -206,7 +206,6 @@ async fn enqueue(
"INSERT INTO push_match_queue (community_id, event_id) \
SELECT $1, $2 FROM events e \
WHERE e.community_id = $1 AND e.id = $2 AND e.deleted_at IS NULL \
AND community_write_allowed($1) \
AND EXISTS (SELECT 1 FROM push_leases l \
WHERE l.community_id = $1 AND l.active AND l.endpoint_enabled \
AND l.updated_at <= e.received_at \
@@ -244,4 +243,70 @@ mod tests {
assert_eq!(receiver.len(), CAPACITY - 1);
producer.join().await;
}
mod postgres_tests {
use super::*;
#[tokio::test]
#[ignore = "requires Postgres"]
async fn enqueue_admits_active_tenant_and_rejects_quiescing_tenant() {
let (db, communities) = crate::test_support::admission_fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
for (index, community) in communities[..2].iter().enumerate() {
sqlx::query(
"INSERT INTO events \
(community_id, id, pubkey, created_at, kind, tags, content, sig) \
VALUES ($1, $2, $3, to_timestamp(1), 9, '[]', '', $4)",
)
.bind(community.as_uuid())
.bind(vec![index as u8; 32])
.bind(vec![1_u8; 32])
.bind(vec![2_u8; 64])
.execute(&db.pool)
.await
.expect("seed event");
sqlx::query(
"INSERT INTO push_leases (community_id, author, installation_id, \
source_event_id, source_created_at, generation, active, endpoint_hash, \
endpoint_grant, max_class, subscriptions, expires_at, updated_at) \
VALUES ($1, $2, 'test', $3, 1, 1, true, $4, 'grant', 'default', '[]', \
EXTRACT(EPOCH FROM now())::bigint + 3600, now() - interval '1 hour')",
)
.bind(community.as_uuid())
.bind(vec![3_u8; 32])
.bind(vec![4_u8; 32])
.bind(vec![5_u8; 32])
.execute(&db.pool)
.await
.expect("seed lease");
}
let active = Job {
community: communities[0],
event_id: vec![0_u8; 32],
barrier: None,
};
let quiescing = Job {
community: communities[1],
event_id: vec![1_u8; 32],
barrier: None,
};
let err = enqueue(&db.pool, &quiescing, None)
.await
.expect_err("quiescing enqueue");
assert!(crate::test_support::is_admission_rejection(&err));
enqueue(&db.pool, &active, None)
.await
.expect("admitted enqueue");
for (community, expected) in [(communities[0], 1), (communities[1], 0)] {
let count: i64 = sqlx::query_scalar(
"SELECT count(*) FROM push_match_queue WHERE community_id = $1",
)
.bind(community.as_uuid())
.fetch_one(&db.pool)
.await
.expect("read enqueue state");
assert_eq!(count, expected, "queue state for {community}");
}
}
}
}

563
crates/buzz-db/src/store/admission_postgres_tests.rs
View File

@@ -1,55 +1,11 @@
//! Live writer-admission regressions. A one-connection pool runs with ordinary
//! PostgreSQL triggers disabled, so a missing application check is observable.
use crate::{Db, DbConfig, DbError};
use crate::{Db, DbError};
use buzz_core::CommunityId;
use uuid::Uuid;
async fn fixture() -> (Db, [CommunityId; 3]) {
let config = DbConfig {
database_url: crate::test_support::database_url(),
max_connections: 1,
min_connections: 1,
..DbConfig::default()
};
let db = Db::new(&config).await.expect("connect test database");
let pool = &db.pool;
let mut communities = [CommunityId::from_uuid(Uuid::nil()); 3];
for community in &mut communities {
let id = Uuid::new_v4();
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(id)
.bind(format!("admission-{id}.example"))
.execute(pool)
.await
.expect("create community");
*community = CommunityId::from_uuid(id);
}
crate::test_support::set_deletion_state(pool, *communities[1].as_uuid(), "quiescing").await;
crate::test_support::set_deletion_state(pool, *communities[2].as_uuid(), "fenced").await;
// One physical session serves every test API call. PostgreSQL's replica
// role skips the 32 community-write triggers as well as other ordinary
// triggers; the application precheck remains fully active.
sqlx::query("SET session_replication_role = replica")
.execute(pool)
.await
.expect("disable ordinary triggers on the test session");
let role: String = sqlx::query_scalar("SHOW session_replication_role")
.fetch_one(pool)
.await
.expect("read replication role");
assert_eq!(role, "replica");
// Witness that the database backstop is truly bypassed: this raw fenced
// insert would be rejected with the community-write trigger enabled.
sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)")
.bind(communities[2].as_uuid())
.bind(vec![0xfe_u8; 32])
.execute(pool)
.await
.expect("raw fenced insert succeeds with triggers disabled");
(db, communities)
}
use crate::test_support::admission_fixture as fixture;
fn assert_fenced(error: DbError) {
assert!(
@@ -799,3 +755,518 @@ async fn owner_mutations_report_pending_deletion_when_admission_refuses() {
assert_eq!(count_relay_members(&db, quiescing).await, 0);
assert_eq!(count_relay_members(&db, fenced).await, 0);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn invite_sweep_admits_active_tenant_and_skips_quiescing_tenant() {
let (db, communities) = fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
let cutoff = chrono::Utc::now();
for (index, community) in communities[..2].iter().enumerate() {
sqlx::query(
"INSERT INTO relay_invites (community_id, token_hash, expires_at, created_by) \
VALUES ($1, $2, $3, 'test')",
)
.bind(community.as_uuid())
.bind(vec![index as u8; 32])
.bind(cutoff - chrono::Duration::minutes(1))
.execute(&db.pool)
.await
.expect("seed invite");
}
assert_eq!(
super::relay_invite::reap_expired_relay_invites(&db.pool, cutoff)
.await
.unwrap(),
1
);
for (community, expected) in [(communities[0], 0), (communities[1], 1)] {
let count: i64 =
sqlx::query_scalar("SELECT count(*) FROM relay_invites WHERE community_id = $1")
.bind(community.as_uuid())
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(count, expected, "invite state for {community}");
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn workflow_prune_admits_active_tenant_and_skips_quiescing_tenant() {
let (db, communities) = fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
let cutoff = chrono::Utc::now();
for community in &communities[..2] {
sqlx::query(
"INSERT INTO scheduled_workflow_fires \
(community_id, workflow_id, scheduled_for, claimed_at) VALUES ($1, $2, $3, $4)",
)
.bind(community.as_uuid())
.bind(Uuid::new_v4())
.bind(cutoff - chrono::Duration::hours(2))
.bind(cutoff - chrono::Duration::hours(1))
.execute(&db.pool)
.await
.expect("seed fire");
}
assert_eq!(
super::workflow::prune_scheduled_workflow_fires_before(&db.pool, cutoff)
.await
.unwrap(),
1
);
for (community, expected) in [(communities[0], 0), (communities[1], 1)] {
let count: i64 = sqlx::query_scalar(
"SELECT count(*) FROM scheduled_workflow_fires WHERE community_id = $1",
)
.bind(community.as_uuid())
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(count, expected, "fire state for {community}");
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn channel_reap_admits_active_tenant_and_skips_quiescing_tenant() {
let (db, communities) = fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
for community in &communities[..2] {
sqlx::query(
"INSERT INTO channels (community_id, name, created_by, ttl_seconds, ttl_deadline) \
VALUES ($1, 'expired', $2, 60, now() - interval '1 minute')",
)
.bind(community.as_uuid())
.bind(vec![1_u8; 32])
.execute(&db.pool)
.await
.expect("seed channel");
}
let reaped = super::channel::reap_expired_ephemeral_channels(&db.pool)
.await
.unwrap();
assert_eq!(reaped.len(), 1);
assert_eq!(reaped[0].community_id, communities[0]);
for (community, expected) in [(communities[0], true), (communities[1], false)] {
let archived: bool = sqlx::query_scalar(
"SELECT archived_at IS NOT NULL FROM channels WHERE community_id = $1",
)
.bind(community.as_uuid())
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(archived, expected, "archive state for {community}");
}
}
/// Keep every failure-injection object, including the ALWAYS trigger's table,
/// in this call's shadow schema so a shared-database run cannot leak it.
async fn fail_tenant_mutations(pool: &sqlx::PgPool, schema: &str, table: &str, id: Uuid) {
// Dynamic identifiers come only from our UUID schema and literal table names.
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.{table} (LIKE public.{table} INCLUDING ALL)"
)))
.execute(pool)
.await
.expect("create test-local serving table");
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.fail_tenant (id uuid PRIMARY KEY)"
)))
.execute(pool)
.await
.expect("create failure target");
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.fail_tenant (id) VALUES ($1)"
)))
.bind(id)
.execute(pool)
.await
.expect("select failing tenant");
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE FUNCTION {schema}.fail_tenant_mutation() RETURNS trigger \
LANGUAGE plpgsql AS $$ BEGIN \
IF EXISTS (SELECT 1 FROM {schema}.fail_tenant \
WHERE id = COALESCE(NEW.community_id, OLD.community_id)) \
THEN RAISE EXCEPTION 'tenant mutation failed'; END IF; \
IF TG_OP = 'DELETE' THEN RETURN OLD; END IF; RETURN NEW; END $$",
)))
.execute(pool)
.await
.expect("create failure trigger function");
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TRIGGER fail_tenant_mutation BEFORE UPDATE OR DELETE ON {schema}.{table} \
FOR EACH ROW EXECUTE FUNCTION {schema}.fail_tenant_mutation()",
)))
.execute(pool)
.await
.expect("create test-local failure trigger");
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.{table} ENABLE ALWAYS TRIGGER fail_tenant_mutation"
)))
.execute(pool)
.await
.expect("enable trigger with replica role");
}
async fn two_active_tenants(db: &Db, first: CommunityId) -> [Uuid; 2] {
let extra = Uuid::new_v4();
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(extra)
.bind(format!("admission-extra-{extra}.example"))
.execute(&db.pool)
.await
.expect("seed second active tenant");
let mut ids = [*first.as_uuid(), extra];
ids.sort();
ids
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn channel_reap_returns_committed_rows_after_earlier_tenant_error() {
let (db, communities) = fixture().await;
let schema = crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
let ids = two_active_tenants(&db, communities[0]).await;
fail_tenant_mutations(&db.pool, &schema, "channels", ids[0]).await;
for id in ids {
sqlx::query(
"INSERT INTO channels (community_id, name, created_by, ttl_seconds, ttl_deadline) \
VALUES ($1, 'expired', $2, 60, now() - interval '1 minute')",
)
.bind(id)
.bind(vec![1_u8; 32])
.execute(&db.pool)
.await
.expect("seed expired channel");
}
let reaped = super::channel::reap_expired_ephemeral_channels(&db.pool)
.await
.expect("tenant failure must not stop the fleet sweep");
assert_eq!(reaped.len(), 1);
assert_eq!(*reaped[0].community_id.as_uuid(), ids[1]);
for (id, expected) in [(ids[0], false), (ids[1], true)] {
let archived: bool = sqlx::query_scalar(
"SELECT archived_at IS NOT NULL FROM channels WHERE community_id = $1",
)
.bind(id)
.fetch_one(&db.pool)
.await
.expect("read archive state");
assert_eq!(archived, expected, "archive state for {id}");
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.fail_tenant"
)))
.execute(&db.pool)
.await
.unwrap();
let retried = super::channel::reap_expired_ephemeral_channels(&db.pool)
.await
.unwrap();
assert_eq!(retried.len(), 1);
assert_eq!(*retried[0].community_id.as_uuid(), ids[0]);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn d_tag_backfill_admits_active_tenant_and_skips_quiescing_tenant() {
let (db, communities) = fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
for (index, community) in communities[..2].iter().enumerate() {
sqlx::query(
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig) \
VALUES ($1, $2, $3, to_timestamp(1), 30023, '[[\"d\",\"tag\"]]'::jsonb, '', $4)",
)
.bind(community.as_uuid())
.bind(vec![index as u8; 32])
.bind(vec![2_u8; 32])
.bind(vec![3_u8; 64])
.execute(&db.pool)
.await
.expect("seed addressable event");
}
assert_eq!(db.backfill_d_tags().await.unwrap(), 1);
let active: Option<String> =
sqlx::query_scalar("SELECT d_tag FROM events WHERE community_id = $1")
.bind(communities[0].as_uuid())
.fetch_one(&db.pool)
.await
.unwrap();
let quiescing: Option<String> =
sqlx::query_scalar("SELECT d_tag FROM events WHERE community_id = $1")
.bind(communities[1].as_uuid())
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(active.as_deref(), Some("tag"));
assert_eq!(quiescing, None);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn match_claim_admits_active_tenant_and_skips_quiescing_tenant() {
let (db, communities) = fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
for (index, community) in communities[..2].iter().enumerate() {
sqlx::query(
"INSERT INTO push_match_queue (community_id, event_id, next_attempt_at) \
VALUES ($1, $2, now() - make_interval(mins => $3))",
)
.bind(community.as_uuid())
.bind(vec![index as u8; 32])
.bind(if index == 0 { 1 } else { 2 })
.execute(&db.pool)
.await
.expect("seed match job");
}
// The production claim is exercised; absent source events are its normal
// terminal cleanup case, so it returns None after claiming and pruning.
assert!(super::push::claim_due_match_batch(
&db.pool,
8,
chrono::Utc::now() + chrono::Duration::minutes(1)
)
.await
.unwrap()
.is_none());
for (community, expected) in [(communities[0], 0), (communities[1], 1)] {
let count: i64 =
sqlx::query_scalar("SELECT count(*) FROM push_match_queue WHERE community_id = $1")
.bind(community.as_uuid())
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(count, expected, "match job for {community}");
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn match_reap_admits_active_tenant_and_skips_quiescing_tenant() {
let (db, communities) = fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
for (index, community) in communities[..2].iter().enumerate() {
sqlx::query(
"INSERT INTO push_match_queue (community_id, event_id, attempts) VALUES ($1, $2, $3)",
)
.bind(community.as_uuid())
.bind(vec![index as u8; 32])
.bind(super::push::MAX_MATCH_ATTEMPTS)
.execute(&db.pool)
.await
.expect("seed exhausted match job");
}
assert_eq!(
super::push::reap_exhausted_matches(&db.pool).await.unwrap(),
1
);
for (community, expected) in [(communities[0], 0), (communities[1], 1)] {
let count: i64 =
sqlx::query_scalar("SELECT count(*) FROM push_match_queue WHERE community_id = $1")
.bind(community.as_uuid())
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(count, expected, "exhausted job for {community}");
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn invite_sweep_rechecks_admission_after_candidate_read() {
let (db, communities) = fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
let community = communities[0];
let cutoff = chrono::Utc::now();
sqlx::query(
"INSERT INTO relay_invites (community_id, token_hash, expires_at, created_by) \
VALUES ($1, $2, $3, 'test')",
)
.bind(community.as_uuid())
.bind(vec![0_u8; 32])
.bind(cutoff - chrono::Duration::minutes(1))
.execute(&db.pool)
.await
.unwrap();
let pool = db.pool.clone();
let reaped = crate::test_support::fence_while_sweep_waits(community, async move {
super::relay_invite::reap_expired_relay_invites(&pool, cutoff).await
})
.await
.unwrap();
assert_eq!(reaped, 0);
let count: i64 =
sqlx::query_scalar("SELECT count(*) FROM relay_invites WHERE community_id = $1")
.bind(community.as_uuid())
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(count, 1, "newly fenced tenant's invite must remain");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn channel_reap_rechecks_admission_after_candidate_read() {
let (db, communities) = fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
let community = communities[0];
sqlx::query(
"INSERT INTO channels (community_id, name, created_by, ttl_seconds, ttl_deadline) \
VALUES ($1, 'expired', $2, 60, now() - interval '1 minute')",
)
.bind(community.as_uuid())
.bind(vec![1_u8; 32])
.execute(&db.pool)
.await
.unwrap();
let pool = db.pool.clone();
let reaped = crate::test_support::fence_while_sweep_waits(community, async move {
super::channel::reap_expired_ephemeral_channels(&pool).await
})
.await
.unwrap();
assert!(reaped.is_empty());
let archived: bool =
sqlx::query_scalar("SELECT archived_at IS NOT NULL FROM channels WHERE community_id = $1")
.bind(community.as_uuid())
.fetch_one(&db.pool)
.await
.unwrap();
assert!(
!archived,
"newly fenced tenant's channel must remain unarchived"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn match_claim_rechecks_admission_and_continues_after_candidate_read() {
let (db, communities) = fixture().await;
crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
let ids = two_active_tenants(&db, communities[0]).await;
for (index, id) in ids.iter().enumerate() {
sqlx::query(
"INSERT INTO push_match_queue (community_id, event_id, next_attempt_at) \
VALUES ($1, $2, now() - make_interval(mins => $3))",
)
.bind(id)
.bind(vec![index as u8; 32])
.bind(if index == 0 { 2 } else { 1 })
.execute(&db.pool)
.await
.unwrap();
}
let pool = db.pool.clone();
// Missing source events take the production claim's terminal cleanup path.
crate::test_support::fence_while_sweep_waits(CommunityId::from_uuid(ids[0]), async move {
super::push::claim_due_match_batch(
&pool,
8,
chrono::Utc::now() + chrono::Duration::minutes(1),
)
.await
})
.await
.unwrap();
for (id, expected) in [(ids[0], 1), (ids[1], 0)] {
let state: Vec<(String, i32)> =
sqlx::query_as("SELECT state, attempts FROM push_match_queue WHERE community_id = $1")
.bind(id)
.fetch_all(&db.pool)
.await
.unwrap();
assert_eq!(state.len(), expected, "queue state for {id}");
if expected == 1 {
assert_eq!(state[0], ("pending".to_owned(), 0));
}
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn periodic_reaps_continue_after_tenant_error_and_retry_retained_rows() {
for table in [
"relay_invites",
"scheduled_workflow_fires",
"push_match_queue",
] {
let (db, communities) = fixture().await;
let schema = crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await;
let ids = two_active_tenants(&db, communities[0]).await;
fail_tenant_mutations(&db.pool, &schema, table, ids[0]).await;
let cutoff = chrono::Utc::now();
for id in ids {
match table {
"relay_invites" => {
sqlx::query(
"INSERT INTO relay_invites (community_id, token_hash, expires_at, created_by) \
VALUES ($1, $2, $3, 'test')",
)
.bind(id).bind(vec![1_u8; 32])
.bind(cutoff - chrono::Duration::minutes(if id == ids[0] { 2 } else { 1 }))
.execute(&db.pool).await.unwrap();
}
"scheduled_workflow_fires" => {
sqlx::query(
"INSERT INTO scheduled_workflow_fires \
(community_id, workflow_id, scheduled_for, claimed_at) \
VALUES ($1, $2, $3, $3)",
)
.bind(id)
.bind(Uuid::new_v4())
.bind(cutoff - chrono::Duration::hours(1))
.execute(&db.pool)
.await
.unwrap();
}
"push_match_queue" => {
sqlx::query(
"INSERT INTO push_match_queue (community_id, event_id, attempts) \
VALUES ($1, $2, $3)",
)
.bind(id)
.bind(vec![1_u8; 32])
.bind(super::push::MAX_MATCH_ATTEMPTS)
.execute(&db.pool)
.await
.unwrap();
}
_ => unreachable!(),
}
}
for retry in [false, true] {
if retry {
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.fail_tenant"
)))
.execute(&db.pool)
.await
.unwrap();
}
let count = match table {
"relay_invites" => {
super::relay_invite::reap_expired_relay_invites(&db.pool, cutoff).await
}
"scheduled_workflow_fires" => {
super::workflow::prune_scheduled_workflow_fires_before(&db.pool, cutoff).await
}
"push_match_queue" => super::push::reap_exhausted_matches(&db.pool).await,
_ => unreachable!(),
}
.expect("per-tenant failures must not stop recurring sweeps");
assert_eq!(count, 1, "{table}, retry={retry}");
for (id, expected) in [(ids[0], i64::from(!retry)), (ids[1], 0)] {
let remaining: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT count(*) FROM {table} WHERE community_id = $1",
)))
.bind(id)
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(remaining, expected, "{table} state for {id}, retry={retry}");
}
}
}
}

71
crates/buzz-db/src/store/channel.rs
View File

@@ -846,33 +846,64 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result<Vec<Reaped
crate::observability::WriterOperation::Maintenance,
)
.await?;
let rows = sqlx::query(
"UPDATE channels AS ch SET archived_at = NOW() \
FROM communities AS c \
WHERE ch.community_id = c.id \
AND ch.ttl_seconds IS NOT NULL \
let communities: Vec<Uuid> = sqlx::query_scalar(
"SELECT DISTINCT ch.community_id FROM channels AS ch \
JOIN communities AS c ON ch.community_id = c.id \
WHERE ch.ttl_seconds IS NOT NULL \
AND ch.ttl_deadline < NOW() \
AND ch.archived_at IS NULL \
AND ch.deleted_at IS NULL \
AND c.archived_at IS NULL \
AND community_write_allowed(ch.community_id) \
RETURNING ch.community_id, c.host, ch.id",
ORDER BY ch.community_id",
)
.fetch_all(&mut *connection)
.await?;
rows.into_iter()
.map(|row| {
let community_id: Uuid = row.try_get("community_id")?;
let host: String = row.try_get("host")?;
let channel_id: Uuid = row.try_get("id")?;
Ok(ReapedEphemeralChannel {
community_id: CommunityId::from_uuid(community_id),
host,
channel_id,
})
})
.collect()
drop(connection);
let mut reaped = Vec::new();
for id in communities {
let batch: Result<Vec<ReapedEphemeralChannel>> = async {
let Some(mut tx) = crate::begin_community_sweep_transaction(
pool,
CommunityId::from_uuid(id),
crate::observability::WriterOperation::Maintenance,
)
.await?
else {
return Ok(Vec::new());
};
let rows = sqlx::query(
"UPDATE channels AS ch SET archived_at = NOW() \
FROM communities AS c \
WHERE ch.community_id = $1 AND ch.community_id = c.id \
AND ch.ttl_seconds IS NOT NULL AND ch.ttl_deadline < NOW() \
AND ch.archived_at IS NULL AND ch.deleted_at IS NULL \
AND c.archived_at IS NULL \
RETURNING ch.community_id, c.host, ch.id",
)
.bind(id)
.fetch_all(tx.conn())
.await?;
let parsed = rows
.into_iter()
.map(|row| {
Ok(ReapedEphemeralChannel {
community_id: CommunityId::from_uuid(row.try_get("community_id")?),
host: row.try_get("host")?,
channel_id: row.try_get("id")?,
})
})
.collect::<Result<Vec<_>>>()?;
tx.commit().await?;
Ok(parsed)
}
.await;
match batch {
Ok(rows) => reaped.extend(rows),
Err(error) => tracing::warn!(community_id = %id, error = %error,
"Ephemeral channel reap failed; retained rows will retry next tick"),
}
}
Ok(reaped)
}
impl Db {

39
crates/buzz-db/src/store/event.rs
View File

@@ -2717,7 +2717,7 @@ impl Db {
/// Backfill `d_tag` for existing NIP-33 events (kind 30000–39999) that have `d_tag IS NULL`.
///
/// Idempotent — safe to call on every startup. No-ops when all rows are already populated.
/// Runs a single UPDATE touching only NIP-33 rows with NULL d_tag.
/// Admits each community before updating its NIP-33 rows with NULL d_tag.
#[datastore_span(name = "backfill_d_tags", system = "postgresql")]
pub async fn backfill_d_tags(&self) -> Result<u64> {
let mut connection = crate::observability::acquire_writer(
@@ -2725,19 +2725,40 @@ impl Db {
crate::observability::WriterOperation::Bootstrap,
)
.await?;
let result = sqlx::query(
"UPDATE events \
let communities: Vec<Uuid> = sqlx::query_scalar(
"SELECT DISTINCT community_id FROM events \
WHERE kind BETWEEN 30000 AND 39999 AND d_tag IS NULL",
)
.fetch_all(&mut *connection)
.await?;
drop(connection);
let mut updated = 0;
for id in communities {
let Some(mut tx) = crate::begin_community_sweep_transaction(
&self.pool,
CommunityId::from_uuid(id),
crate::observability::WriterOperation::Bootstrap,
)
.await?
else {
continue;
};
let result = sqlx::query(
"UPDATE events \
SET d_tag = COALESCE( \
(SELECT elem->>1 FROM jsonb_array_elements(tags) AS elem \
WHERE elem->>0 = 'd' LIMIT 1), \
'' \
) \
WHERE kind BETWEEN 30000 AND 39999 AND d_tag IS NULL \
AND community_write_allowed(community_id)",
)
.execute(&mut *connection)
.await?;
Ok(result.rows_affected())
WHERE community_id = $1 AND kind BETWEEN 30000 AND 39999 AND d_tag IS NULL",
)
.bind(id)
.execute(tx.conn())
.await?;
tx.commit().await?;
updated += result.rows_affected();
}
Ok(updated)
}
/// Soft-delete NIP-29 discovery events for a channel created by a specific relay pubkey.

139
crates/buzz-db/src/store/push.rs
View File

@@ -48,6 +48,10 @@ async fn acquire_push_gate_lock(tx: &mut sqlx::PgConnection, community: Communit
/// Maximum claims for a malformed matcher job before it is discarded.
pub const MAX_MATCH_ATTEMPTS: i32 = 8;
/// Bound tenant admission attempts when a due community fences after the
/// candidate read. A later tick can continue beyond this page if needed.
const MATCH_CLAIM_COMMUNITY_CANDIDATES: i64 = 16;
/// Common signed-event ordering fields for a lease replacement.
#[derive(Debug, Clone, Copy)]
pub struct LeaseVersion<'a> {
@@ -861,26 +865,47 @@ where
Fut: std::future::Future<Output = Result<Vec<buzz_core::StoredEvent>>>,
{
let claim_id = Uuid::new_v4();
let mut connection =
acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance)
.await?;
let rows = sqlx::query(
r#"
WITH target AS (
SELECT community_id
FROM push_match_queue
WHERE attempts < $3
AND next_attempt_at <= now()
AND (state = 'pending' OR (state = 'matching' AND lease_until < now()))
AND community_write_allowed(community_id)
ORDER BY next_attempt_at, created_at
LIMIT 1
),
candidates AS (
let mut refused = Vec::<Uuid>::new();
let mut claimed = None;
for _ in 0..MATCH_CLAIM_COMMUNITY_CANDIDATES {
let mut connection =
acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance)
.await?;
// Exclude fenced tenants so they cannot occupy the queue head.
let candidate: Option<Uuid> = sqlx::query_scalar(
"SELECT q.community_id FROM push_match_queue q \
JOIN communities c ON c.id = q.community_id \
WHERE q.attempts < $1 AND q.next_attempt_at <= now() \
AND (q.state = 'pending' OR (q.state = 'matching' AND q.lease_until < now())) \
AND c.deletion_state = 'active' AND c.deleted_at IS NULL \
AND q.community_id <> ALL($2) \
ORDER BY q.next_attempt_at, q.created_at LIMIT 1",
)
.bind(MAX_MATCH_ATTEMPTS)
.bind(&refused)
.fetch_optional(&mut *connection)
.await?;
drop(connection);
let Some(id) = candidate else {
break;
};
let community = CommunityId::from_uuid(id);
let Some(mut tx) = crate::begin_community_sweep_transaction(
pool,
community,
crate::observability::WriterOperation::Maintenance,
)
.await?
else {
refused.push(id);
continue;
};
let rows = sqlx::query(
r#"
WITH candidates AS (
SELECT q.community_id, q.event_id
FROM push_match_queue q
JOIN target t ON q.community_id = t.community_id
WHERE q.attempts < $3
WHERE q.community_id = $5 AND q.attempts < $3
AND q.next_attempt_at <= now()
AND (q.state = 'pending' OR (q.state = 'matching' AND q.lease_until < now()))
ORDER BY q.next_attempt_at, q.created_at
@@ -893,22 +918,26 @@ where
WHERE q.community_id=c.community_id AND q.event_id=c.event_id
RETURNING q.community_id, q.event_id, q.attempts
"#,
)
.bind(claim_id)
.bind(lease_until)
.bind(MAX_MATCH_ATTEMPTS)
.bind(limit)
.fetch_all(&mut *connection)
.await?;
)
.bind(claim_id)
.bind(lease_until)
.bind(MAX_MATCH_ATTEMPTS)
.bind(limit)
.bind(community.as_uuid())
.fetch_all(tx.conn())
.await?;
tx.commit().await?;
claimed = Some((community, rows));
break;
}
let Some((community, rows)) = claimed else {
return Ok(None);
};
if rows.is_empty() {
return Ok(None);
}
// The claim query is a single autocommitted statement. Release its pool
// slot before loading source events, because the production loader owns a
// separately attributed acquisition. Holding this connection across the
// load would self-starve a supported size-one writer pool.
drop(connection);
let community = CommunityId::from_uuid(rows[0].try_get("community_id")?);
// The admitted transaction committed above, so no connection is held
// across the loader (size-one pools must work).
let mut attempts = std::collections::HashMap::with_capacity(rows.len());
for row in &rows {
let event_id: Vec<u8> = row.try_get("event_id")?;
@@ -967,15 +996,45 @@ pub async fn reap_exhausted_matches(pool: &PgPool) -> Result<u64> {
let mut connection =
acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance)
.await?;
Ok(sqlx::query(
"DELETE FROM push_match_queue WHERE attempts >= $1 \
AND (state='pending' OR (state='matching' AND lease_until < now())) \
AND community_write_allowed(community_id)",
let communities: Vec<Uuid> = sqlx::query_scalar(
"SELECT DISTINCT community_id FROM push_match_queue WHERE attempts >= $1 \
AND (state='pending' OR (state='matching' AND lease_until < now()))",
)
.bind(MAX_MATCH_ATTEMPTS)
.execute(&mut *connection)
.await?
.rows_affected())
.fetch_all(&mut *connection)
.await?;
drop(connection);
let mut deleted = 0;
for id in communities {
let batch: Result<u64> = async {
let Some(mut tx) = crate::begin_community_sweep_transaction(
pool,
CommunityId::from_uuid(id),
crate::observability::WriterOperation::Maintenance,
)
.await?
else {
return Ok(0);
};
let result = sqlx::query(
"DELETE FROM push_match_queue WHERE community_id = $1 AND attempts >= $2 \
AND (state='pending' OR (state='matching' AND lease_until < now()))",
)
.bind(id)
.bind(MAX_MATCH_ATTEMPTS)
.execute(tx.conn())
.await?;
tx.commit().await?;
Ok(result.rows_affected())
}
.await;
match batch {
Ok(count) => deleted += count,
Err(error) => tracing::warn!(community_id = %id, error = %error,
"Exhausted push-match reap failed; retained rows will retry next tick"),
}
}
Ok(deleted)
}
/// Load active endpoint-enabled leases for one tenant.
@@ -2601,7 +2660,7 @@ mod postgres_tests {
#[ignore = "requires Postgres"]
async fn matcher_claim_skips_quiescing_tenant_while_active_bystanders_progress() {
let pool = setup_pool().await;
sqlx::query("DELETE FROM push_match_queue WHERE community_write_allowed(community_id)")
sqlx::query("DELETE FROM push_match_queue")
.execute(&pool)
.await
.expect("drain active matcher queue");
@@ -2631,7 +2690,7 @@ mod postgres_tests {
#[ignore = "requires Postgres"]
async fn exhausted_match_reaper_skips_quiescing_tenant_and_reaps_active_bystanders() {
let pool = setup_pool().await;
sqlx::query("DELETE FROM push_match_queue WHERE community_write_allowed(community_id)")
sqlx::query("DELETE FROM push_match_queue")
.execute(&pool)
.await
.expect("drain active matcher queue");

66
crates/buzz-db/src/store/relay_invite.rs
View File

@@ -24,6 +24,7 @@ use buzz_core::invite::{
use buzz_datastore_tracing::datastore_span;
use chrono::{DateTime, Utc};
use sqlx::{PgPool, Row as _};
use uuid::Uuid;
use crate::error::Result;
use crate::{CommunityId, Db};
@@ -179,22 +180,65 @@ pub async fn reap_expired_relay_invites(pool: &PgPool, cutoff: DateTime<Utc>) ->
crate::observability::WriterOperation::Maintenance,
)
.await?;
let result = sqlx::query(
"DELETE FROM relay_invites \
WHERE (community_id, id) IN (\
SELECT community_id, id FROM relay_invites \
WHERE expires_at < $1 \
AND community_write_allowed(community_id) \
ORDER BY expires_at \
LIMIT $2\
)",
// Keep the original global expiry order and 1,000-row ceiling. The state
// read only narrows candidates; admission below is the write authority.
// Exclude fenced tenants so they cannot fill the LIMIT page.
let candidates = sqlx::query(
"SELECT invite.community_id, invite.id FROM relay_invites invite \
JOIN communities community ON community.id = invite.community_id \
WHERE invite.expires_at < $1 AND community.deletion_state = 'active' \
AND community.deleted_at IS NULL \
ORDER BY invite.expires_at LIMIT $2",
)
.bind(cutoff)
.bind(RETENTION_SWEEP_BATCH_SIZE)
.execute(&mut *connection)
.fetch_all(&mut *connection)
.await?;
drop(connection);
Ok(result.rows_affected())
let mut ordered_communities = Vec::new();
let mut ids_by_community = std::collections::HashMap::<Uuid, Vec<Uuid>>::new();
for row in candidates {
let community: Uuid = row.try_get("community_id")?;
let id: Uuid = row.try_get("id")?;
if !ids_by_community.contains_key(&community) {
ordered_communities.push(community);
}
ids_by_community.entry(community).or_default().push(id);
}
let mut deleted = 0;
for id in ordered_communities {
let batch: Result<u64> = async {
let community = CommunityId::from_uuid(id);
let Some(mut tx) = crate::begin_community_sweep_transaction(
pool,
community,
crate::observability::WriterOperation::Maintenance,
)
.await?
else {
return Ok(0);
};
let result = sqlx::query(
"DELETE FROM relay_invites WHERE community_id = $1 AND id = ANY($2) AND expires_at < $3",
)
.bind(id)
.bind(&ids_by_community[&id])
.bind(cutoff)
.execute(tx.conn())
.await?;
tx.commit().await?;
Ok(result.rows_affected())
}
.await;
match batch {
Ok(count) => deleted += count,
Err(error) => tracing::warn!(community_id = %id, error = %error,
"Relay invite reap failed; retained rows will retry next tick"),
}
}
Ok(deleted)
}
/// Atomically claim a v2 relay invite.

42
crates/buzz-db/src/store/workflow.rs
View File

@@ -634,18 +634,42 @@ pub async fn prune_scheduled_workflow_fires_before(
pool: &PgPool,
older_than: DateTime<Utc>,
) -> Result<u64> {
let result = sqlx::query(
r#"
DELETE FROM scheduled_workflow_fires
WHERE claimed_at < $1
AND community_write_allowed(community_id)
"#,
let communities: Vec<Uuid> = sqlx::query_scalar(
"SELECT DISTINCT community_id FROM scheduled_workflow_fires WHERE claimed_at < $1",
)
.bind(older_than)
.execute(pool)
.fetch_all(pool)
.await?;
Ok(result.rows_affected())
let mut deleted = 0;
for id in communities {
let batch: Result<u64> = async {
let Some(mut tx) = crate::begin_community_sweep_transaction(
pool,
CommunityId::from_uuid(id),
crate::observability::WriterOperation::Maintenance,
)
.await?
else {
return Ok(0);
};
let result = sqlx::query(
"DELETE FROM scheduled_workflow_fires WHERE community_id = $1 AND claimed_at < $2",
)
.bind(id)
.bind(older_than)
.execute(tx.conn())
.await?;
tx.commit().await?;
Ok(result.rows_affected())
}
.await;
match batch {
Ok(count) => deleted += count,
Err(error) => tracing::warn!(community_id = %id, error = %error,
"Workflow fire prune failed; retained rows will retry next tick"),
}
}
Ok(deleted)
}
/// Update a workflow's name, definition, and definition_hash.

148
crates/buzz-db/src/test_support.rs
View File

@@ -8,12 +8,155 @@ pub(crate) fn database_url() -> String {
.unwrap_or_else(|_| DEFAULT_DATABASE_URL.to_owned())
}
/// One-session fixture that bypasses ordinary write-fence triggers while
/// retaining application admission. Communities are active, quiescing, fenced.
pub(crate) async fn admission_fixture() -> (crate::Db, [buzz_core::CommunityId; 3]) {
use buzz_core::CommunityId;
use uuid::Uuid;
let db = crate::Db::new(&crate::DbConfig {
database_url: database_url(),
max_connections: 1,
min_connections: 1,
..crate::DbConfig::default()
})
.await
.expect("connect test database");
let pool = &db.pool;
let mut communities = [CommunityId::from_uuid(Uuid::nil()); 3];
for community in &mut communities {
let id = Uuid::new_v4();
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(id)
.bind(format!("admission-{id}.example"))
.execute(pool)
.await
.expect("create community");
*community = CommunityId::from_uuid(id);
}
set_deletion_state(pool, *communities[1].as_uuid(), "quiescing").await;
set_deletion_state(pool, *communities[2].as_uuid(), "fenced").await;
// The one physical session serves every API call in this fixture.
sqlx::query("SET session_replication_role = replica")
.execute(pool)
.await
.expect("disable ordinary triggers");
let role: String = sqlx::query_scalar("SHOW session_replication_role")
.fetch_one(pool)
.await
.expect("read replication role");
assert_eq!(role, "replica");
// A raw write to the fenced tenant proves the DB backstop is bypassed.
sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)")
.bind(communities[2].as_uuid())
.bind(vec![0xfe_u8; 32])
.execute(pool)
.await
.expect("raw fenced insert succeeds with triggers disabled");
(db, communities)
}
/// In one per-test database, make any return to the retired SQL predicate
/// observable without changing the production function in the public schema.
pub(crate) async fn reject_legacy_sql_fence_predicate(pool: &sqlx::PgPool) -> String {
let schema = format!("admission_test_{}", uuid::Uuid::new_v4().simple());
sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {schema}")))
.execute(pool)
.await
.expect("create test-only function schema");
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE FUNCTION {schema}.community_write_allowed(uuid) RETURNS boolean \
LANGUAGE plpgsql VOLATILE AS $$ BEGIN \
RAISE EXCEPTION 'legacy SQL fence predicate called'; END $$",
)))
.execute(pool)
.await
.expect("shadow legacy SQL predicate");
sqlx::query(sqlx::AssertSqlSafe(format!(
"SET search_path = {schema}, public"
)))
.execute(pool)
.await
.expect("resolve test-local predicate first");
let shadowed: bool = sqlx::query_scalar(
"SELECT to_regprocedure('community_write_allowed(uuid)') = to_regprocedure($1)",
)
.bind(format!("{schema}.community_write_allowed(uuid)"))
.fetch_one(pool)
.await
.expect("resolve shadowed predicate");
assert!(shadowed);
schema
}
/// Hold the executor's exclusive lock until a sweep reaches shared admission,
/// then quiesce in that same transaction before releasing it. The ordinary
/// second pool keeps lifecycle triggers enabled; the sweep fixture bypasses them.
pub(crate) async fn fence_while_sweep_waits<F, T>(community: buzz_core::CommunityId, sweep: F) -> T
where
F: std::future::Future<Output = T> + Send + 'static,
T: Send + 'static,
{
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(1)
.connect(&database_url())
.await
.expect("connect deletion executor pool");
let mut tx = pool.begin().await.expect("begin deletion executor");
sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))")
.bind(community.as_uuid())
.execute(&mut *tx)
.await
.expect("hold exclusive deletion lock");
let mut task = tokio::spawn(sweep);
let waiting = tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let blocked: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM pg_locks \
WHERE locktype = 'advisory' AND NOT granted \
AND database = (SELECT oid FROM pg_database WHERE datname = current_database()) \
AND pg_backend_pid() = ANY(pg_blocking_pids(pid)))",
)
.fetch_one(&mut *tx)
.await
.expect("inspect blocked advisory admission");
if blocked {
break;
}
tokio::task::yield_now().await;
}
})
.await;
if waiting.is_err() {
task.abort();
panic!("sweep did not block in admission before deadline");
}
set_deletion_state_in_transaction(&mut tx, *community.as_uuid(), "quiescing").await;
tx.commit().await.expect("commit deletion fence");
let result = tokio::time::timeout(std::time::Duration::from_secs(5), &mut task).await;
if result.is_err() {
task.abort();
panic!("sweep did not finish after deletion fence committed");
}
pool.close().await;
result.unwrap().expect("sweep task")
}
/// Move a community fixture into a deletion lifecycle state the way the
/// executor does: the tombstone trigger only accepts lifecycle changes from a
/// transaction carrying the executor GUCs for that community, including its
/// current fence generation.
pub(crate) async fn set_deletion_state(pool: &sqlx::PgPool, id: uuid::Uuid, state: &str) {
let mut tx = pool.begin().await.expect("begin lifecycle fixture");
set_deletion_state_in_transaction(&mut tx, id, state).await;
tx.commit().await.expect("commit lifecycle fixture");
}
async fn set_deletion_state_in_transaction(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
id: uuid::Uuid,
state: &str,
) {
sqlx::query(
"SELECT set_config('buzz.deletion_executor_community', $1, true), \
set_config('buzz.deletion_fence_generation', \
@@ -21,7 +164,7 @@ pub(crate) async fn set_deletion_state(pool: &sqlx::PgPool, id: uuid::Uuid, stat
)
.bind(id.to_string())
.bind(id)
.execute(&mut *tx)
.execute(&mut **tx)
.await
.expect("authorize lifecycle fixture");
sqlx::query(
@@ -31,10 +174,9 @@ pub(crate) async fn set_deletion_state(pool: &sqlx::PgPool, id: uuid::Uuid, stat
)
.bind(id)
.bind(state)
.execute(&mut *tx)
.execute(&mut **tx)
.await
.expect("set lifecycle state");
tx.commit().await.expect("commit lifecycle fixture");
}
/// Move a disposable test community into `quiescing`, the state in which

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

@@ -164,7 +164,6 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() {
.expect("d-tag backfill must precede discovery deletion")
.0;
assert!(backfill_d_tags.contains("WriterOperation::Bootstrap"));
assert!(backfill_d_tags.contains("execute(&mut *connection)"));
let soft_delete_discovery = event
.split_once("pub async fn soft_delete_discovery_events")
.expect("event store must expose discovery-event deletion")
@@ -908,6 +907,7 @@ const COMMUNITY_CHOKEPOINT_LEGACY_MARKER: &str =
fn has_any_tenant_local_chokepoint(source: &str) -> bool {
source.contains(COMMUNITY_CHOKEPOINT_MARKER)
|| source.contains("begin_community_sweep_transaction(")
|| source.contains(GENERAL_COMMUNITY_CHOKEPOINT_MARKER)
|| source.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER)
}
@@ -915,10 +915,9 @@ fn has_any_tenant_local_chokepoint(source: &str) -> bool {
const ADMITTED_TX_SIGNATURE_MARKERS: [&str; 2] = ["&mut AdmittedTx", "&mut crate::AdmittedTx"];
// Narrow reviewed exceptions for non-serving verification probes only.
const GUARDED_WRITE_FUNCTION_EXCEPTIONS: [&str; 3] = [
const GUARDED_WRITE_FUNCTION_EXCEPTIONS: [&str; 2] = [
"pub async fn verify_floor_guard_behavior(",
"pub async fn verify_channel_roster_fence_behavior(",
"pub async fn backfill_d_tags(&self) -> Result<u64> {",
];
const GUARDED_TX_ADAPTER_METHOD_PINS: [(&str, &str); 1] = [(