From 8f0dd967dc8723cb8d3c0b9d6676b663e17b58c8 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 09:39:56 -0400 Subject: [PATCH 1/5] fix(db): admit fleet sweeps per tenant instead of in-SQL fence predicate Remove every production application call to community_write_allowed(). Fleet sweeps (relay invites, scheduled workflow fires, ephemeral channels, exhausted push matches), the due push-match claim and the startup d_tag backfill now read candidate communities, open one AdmittedTx per tenant, and run the original mutation scoped to that community. Refused tenants are skipped; batching, ordering and FOR UPDATE SKIP LOCKED are preserved. The redundant predicate in the post-commit push enqueue is deleted (the transaction is already guarded). The ephemeral channel sweep returns committed rows on a later-tenant failure so relay follow-ups still run. Triggers and SQL fence functions are unchanged; a later migration drops them (BUZZ-315 part B). Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/error.rs | 11 + crates/buzz-db/src/lib.rs | 6 +- crates/buzz-db/src/runtime/mod.rs | 14 + crates/buzz-db/src/runtime/push_enqueue.rs | 61 +++- .../src/store/admission_postgres_tests.rs | 336 +++++++++++++++--- crates/buzz-db/src/store/channel.rs | 77 ++-- crates/buzz-db/src/store/event.rs | 39 +- crates/buzz-db/src/store/push.rs | 130 ++++--- crates/buzz-db/src/store/relay_invite.rs | 57 ++- crates/buzz-db/src/store/workflow.rs | 34 +- crates/buzz-db/src/test_support.rs | 77 ++++ crates/buzz-db/tests/observability_source.rs | 3 +- crates/buzz-relay/src/main.rs | 4 + 13 files changed, 709 insertions(+), 140 deletions(-) diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 8c051a7a793..db697ba730d 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -37,6 +37,17 @@ pub enum DbError { #[error("access denied: {0}")] AccessDenied(String), + /// A fleet channel sweep failed after earlier tenant transactions committed. + /// The caller must still process their returned follow-up work. + #[error("ephemeral channel sweep failed after earlier tenant commits: {source}")] + EphemeralReapPartial { + /// Channels already archived and committed by this sweep. + reaped: Vec, + /// Failure from the later tenant transaction. + #[source] + source: Box, + }, + /// JSON serialization or deserialization failed. #[error("serialization error: {0}")] Serde(#[from] serde_json::Error), diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 2f306c35d08..477723e4443 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -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, diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 266003f5a09..ee15f0c3aa5 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -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> { + 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 { diff --git a/crates/buzz-db/src/runtime/push_enqueue.rs b/crates/buzz-db/src/runtime/push_enqueue.rs index ac9c6e2f5d5..4663fb3b068 100644 --- a/crates/buzz-db/src/runtime/push_enqueue.rs +++ b/crates/buzz-db/src/runtime/push_enqueue.rs @@ -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,64 @@ 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, + }; + assert!(enqueue(&db.pool, &quiescing).await.is_err()); + enqueue(&db.pool, &active).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}"); + } + } + } } diff --git a/crates/buzz-db/src/store/admission_postgres_tests.rs b/crates/buzz-db/src/store/admission_postgres_tests.rs index 2c65beecb48..5491deba635 100644 --- a/crates/buzz-db/src/store/admission_postgres_tests.rs +++ b/crates/buzz-db/src/store/admission_postgres_tests.rs @@ -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,291 @@ 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}"); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn channel_reap_postgres_tests_returns_committed_rows_after_later_tenant_error() { + let (db, communities) = fixture().await; + crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await; + 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 = [*communities[0].as_uuid(), extra]; + ids.sort(); + 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"); + } + sqlx::query("CREATE TABLE admission_test.fail_tenant (id uuid PRIMARY KEY)") + .execute(&db.pool) + .await + .expect("create failure target"); + sqlx::query("INSERT INTO admission_test.fail_tenant (id) VALUES ($1)") + .bind(ids[1]) + .execute(&db.pool) + .await + .expect("select later tenant"); + sqlx::query( + "CREATE FUNCTION admission_test.fail_later_archive() RETURNS trigger \ + LANGUAGE plpgsql AS $$ BEGIN \ + IF EXISTS (SELECT 1 FROM admission_test.fail_tenant WHERE id = NEW.community_id) \ + THEN RAISE EXCEPTION 'later tenant archive failed'; END IF; \ + RETURN NEW; END $$", + ) + .execute(&db.pool) + .await + .expect("create failure trigger function"); + sqlx::query( + "CREATE TRIGGER fail_later_archive BEFORE UPDATE ON channels \ + FOR EACH ROW EXECUTE FUNCTION admission_test.fail_later_archive()", + ) + .execute(&db.pool) + .await + .expect("create failure trigger"); + sqlx::query("ALTER TABLE channels ENABLE ALWAYS TRIGGER fail_later_archive") + .execute(&db.pool) + .await + .expect("enable trigger with replica role"); + + let error = super::channel::reap_expired_ephemeral_channels(&db.pool) + .await + .expect_err("second tenant fails after first commits"); + match error { + DbError::EphemeralReapPartial { reaped, source } => { + assert_eq!(reaped.len(), 1); + assert_eq!(*reaped[0].community_id.as_uuid(), ids[0]); + assert!(source.to_string().contains("later tenant archive failed")); + } + other => panic!("expected committed rows with later failure, got {other:?}"), + } + for (id, expected) in [(ids[0], true), (ids[1], false)] { + 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}"); + } +} + +#[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 = + 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 = + 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}"); + } +} diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index ffc60694eee..91b07d2a43e 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -847,33 +847,70 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result = 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", + AND c.deletion_state = 'active' AND c.deleted_at IS NULL \ + 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> = 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::>>()?; + tx.commit().await?; + Ok(parsed) + } + .await; + match batch { + Ok(rows) => reaped.extend(rows), + Err(error) if reaped.is_empty() => return Err(error), + Err(error) => { + return Err(DbError::EphemeralReapPartial { + reaped, + source: Box::new(error), + }); + } + } + } + Ok(reaped) } impl Db { diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 3e76d751900..9474623da9a 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -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 { 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 = 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. diff --git a/crates/buzz-db/src/store/push.rs b/crates/buzz-db/src/store/push.rs index 26eaee65e4d..18dfefa4a3a 100644 --- a/crates/buzz-db/src/store/push.rs +++ b/crates/buzz-db/src/store/push.rs @@ -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,46 @@ where Fut: std::future::Future>>, { 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::::new(); + let mut claimed = None; + for _ in 0..MATCH_CLAIM_COMMUNITY_CANDIDATES { + let mut connection = + acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) + .await?; + let candidate: Option = 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 +917,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")?); + // Release the admitted connection before loading source events: the + // production loader acquires separately and size-one pools must work. let mut attempts = std::collections::HashMap::with_capacity(rows.len()); for row in &rows { let event_id: Vec = row.try_get("event_id")?; @@ -967,15 +995,37 @@ pub async fn reap_exhausted_matches(pool: &PgPool) -> Result { 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 = 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 Some(mut tx) = crate::begin_community_sweep_transaction( + pool, + CommunityId::from_uuid(id), + crate::observability::WriterOperation::Maintenance, + ) + .await? + else { + continue; + }; + 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?; + deleted += result.rows_affected(); + } + Ok(deleted) } /// Load active endpoint-enabled leases for one tenant. @@ -2601,7 +2651,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 +2681,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"); diff --git a/crates/buzz-db/src/store/relay_invite.rs b/crates/buzz-db/src/store/relay_invite.rs index 64876961511..a4fc89b4bc8 100644 --- a/crates/buzz-db/src/store/relay_invite.rs +++ b/crates/buzz-db/src/store/relay_invite.rs @@ -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,56 @@ pub async fn reap_expired_relay_invites(pool: &PgPool, cutoff: DateTime) -> 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. + 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); + + let mut ordered_communities = Vec::new(); + let mut ids_by_community = std::collections::HashMap::>::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); + } - Ok(result.rows_affected()) + let mut deleted = 0; + for id in ordered_communities { + let community = CommunityId::from_uuid(id); + let Some(mut tx) = crate::begin_community_sweep_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await? + else { + continue; + }; + 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?; + deleted += result.rows_affected(); + } + Ok(deleted) } /// Atomically claim a v2 relay invite. diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 8f63d85555e..f5f10ed2e91 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -634,18 +634,34 @@ pub async fn prune_scheduled_workflow_fires_before( pool: &PgPool, older_than: DateTime, ) -> Result { - let result = sqlx::query( - r#" - DELETE FROM scheduled_workflow_fires - WHERE claimed_at < $1 - AND community_write_allowed(community_id) - "#, + let communities: Vec = 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 Some(mut tx) = crate::begin_community_sweep_transaction( + pool, + CommunityId::from_uuid(id), + crate::observability::WriterOperation::Maintenance, + ) + .await? + else { + continue; + }; + 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?; + deleted += result.rows_affected(); + } + Ok(deleted) } /// Update a workflow's name, definition, and definition_hash. diff --git a/crates/buzz-db/src/test_support.rs b/crates/buzz-db/src/test_support.rs index e94b8b14491..4e0143261e8 100644 --- a/crates/buzz-db/src/test_support.rs +++ b/crates/buzz-db/src/test_support.rs @@ -8,6 +8,83 @@ 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) { + sqlx::query("CREATE SCHEMA admission_test") + .execute(pool) + .await + .expect("create test-only function schema"); + sqlx::query( + "CREATE FUNCTION admission_test.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("SET search_path = admission_test, 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('admission_test.community_write_allowed(uuid)')", + ) + .fetch_one(pool) + .await + .expect("resolve shadowed predicate"); + assert!(shadowed); +} + /// 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 diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 06aa82c8e60..f6d51fbe7e2 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -164,7 +164,8 @@ 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)")); + assert!(backfill_d_tags.contains("begin_community_sweep_transaction(")); + assert!(backfill_d_tags.contains("execute(tx.conn())")); let soft_delete_discovery = event .split_once("pub async fn soft_delete_discovery_events") .expect("event store must expose discovery-event deletion") diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 7c1b6078408..680e2432dc8 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1191,6 +1191,10 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { let expired = match reaper_state.db.reap_expired_ephemeral_channels().await { Ok(ids) => ids, + Err(buzz_db::DbError::EphemeralReapPartial { reaped, source }) => { + error!("Ephemeral reaper tick failed after partial progress: {source}"); + reaped + } Err(e) => { error!("Ephemeral reaper tick failed: {e}"); continue; From f97d7d172f147ee8b8b961c9426738365255c6ed Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Wed, 14 Oct 2026 10:24:10 -0400 Subject: [PATCH 2/5] fix(db): pin sweep admission with race tests; continue past failing tenants Address review on the tenant-admitted fleet sweeps: - Add race tests for invite reap, ephemeral channel reap and push-match claim that fence a tenant while the sweep is blocked in admission, so admission (not the candidate read) is proven to authorize the write. - Drop the unbounded channel-reap candidate pre-filter; keep the LIMIT-bounded invite and claim pre-filters with a comment. - Periodic sweeps (invite reap, workflow fire prune, channel reap, exhausted match reap) warn and continue past a failing tenant; its rows stay eligible for the next tick. Removes DbError::EphemeralReapPartial and its relay arm. Claim, enqueue and startup backfill still propagate. - Unique per-call admission_test_ shadow schema; fault-injection objects live inside it. - Recognise begin_community_sweep_transaction as a tenant-local chokepoint and drop the backfill_d_tags source-policy exception. - Tighten the enqueue rejection assertion, fix a stale comment, rename a test. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/error.rs | 11 - crates/buzz-db/src/runtime/push_enqueue.rs | 34 +- .../src/store/admission_postgres_tests.rs | 323 +++++++++++++++--- crates/buzz-db/src/store/channel.rs | 10 +- crates/buzz-db/src/store/push.rs | 51 +-- crates/buzz-db/src/store/relay_invite.rs | 49 +-- crates/buzz-db/src/store/workflow.rs | 44 ++- crates/buzz-db/src/test_support.rs | 93 ++++- crates/buzz-db/tests/observability_source.rs | 6 +- crates/buzz-relay/src/main.rs | 4 - 10 files changed, 462 insertions(+), 163 deletions(-) diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index db697ba730d..8c051a7a793 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -37,17 +37,6 @@ pub enum DbError { #[error("access denied: {0}")] AccessDenied(String), - /// A fleet channel sweep failed after earlier tenant transactions committed. - /// The caller must still process their returned follow-up work. - #[error("ephemeral channel sweep failed after earlier tenant commits: {source}")] - EphemeralReapPartial { - /// Channels already archived and committed by this sweep. - reaped: Vec, - /// Failure from the later tenant transaction. - #[source] - source: Box, - }, - /// JSON serialization or deserialization failed. #[error("serialization error: {0}")] Serde(#[from] serde_json::Error), diff --git a/crates/buzz-db/src/runtime/push_enqueue.rs b/crates/buzz-db/src/runtime/push_enqueue.rs index 4663fb3b068..6377e9a82c3 100644 --- a/crates/buzz-db/src/runtime/push_enqueue.rs +++ b/crates/buzz-db/src/runtime/push_enqueue.rs @@ -254,22 +254,23 @@ mod tests { 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"); + "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')", + 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]) @@ -289,7 +290,10 @@ mod tests { event_id: vec![1_u8; 32], barrier: None, }; - assert!(enqueue(&db.pool, &quiescing).await.is_err()); + let err = enqueue(&db.pool, &quiescing) + .await + .expect_err("quiescing enqueue"); + assert!(crate::test_support::is_admission_rejection(&err)); enqueue(&db.pool, &active).await.expect("admitted enqueue"); for (community, expected) in [(communities[0], 1), (communities[1], 0)] { let count: i64 = sqlx::query_scalar( diff --git a/crates/buzz-db/src/store/admission_postgres_tests.rs b/crates/buzz-db/src/store/admission_postgres_tests.rs index 5491deba635..af505529c71 100644 --- a/crates/buzz-db/src/store/admission_postgres_tests.rs +++ b/crates/buzz-db/src/store/admission_postgres_tests.rs @@ -861,11 +861,56 @@ async fn channel_reap_admits_active_tenant_and_skips_quiescing_tenant() { } } -#[tokio::test] -#[ignore = "requires Postgres"] -async fn channel_reap_postgres_tests_returns_committed_rows_after_later_tenant_error() { - let (db, communities) = fixture().await; - crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await; +/// 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) @@ -873,8 +918,18 @@ async fn channel_reap_postgres_tests_returns_committed_rows_after_later_tenant_e .execute(&db.pool) .await .expect("seed second active tenant"); - let mut ids = [*communities[0].as_uuid(), extra]; + 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) \ @@ -886,49 +941,12 @@ async fn channel_reap_postgres_tests_returns_committed_rows_after_later_tenant_e .await .expect("seed expired channel"); } - sqlx::query("CREATE TABLE admission_test.fail_tenant (id uuid PRIMARY KEY)") - .execute(&db.pool) - .await - .expect("create failure target"); - sqlx::query("INSERT INTO admission_test.fail_tenant (id) VALUES ($1)") - .bind(ids[1]) - .execute(&db.pool) - .await - .expect("select later tenant"); - sqlx::query( - "CREATE FUNCTION admission_test.fail_later_archive() RETURNS trigger \ - LANGUAGE plpgsql AS $$ BEGIN \ - IF EXISTS (SELECT 1 FROM admission_test.fail_tenant WHERE id = NEW.community_id) \ - THEN RAISE EXCEPTION 'later tenant archive failed'; END IF; \ - RETURN NEW; END $$", - ) - .execute(&db.pool) - .await - .expect("create failure trigger function"); - sqlx::query( - "CREATE TRIGGER fail_later_archive BEFORE UPDATE ON channels \ - FOR EACH ROW EXECUTE FUNCTION admission_test.fail_later_archive()", - ) - .execute(&db.pool) - .await - .expect("create failure trigger"); - sqlx::query("ALTER TABLE channels ENABLE ALWAYS TRIGGER fail_later_archive") - .execute(&db.pool) - .await - .expect("enable trigger with replica role"); - - let error = super::channel::reap_expired_ephemeral_channels(&db.pool) + let reaped = super::channel::reap_expired_ephemeral_channels(&db.pool) .await - .expect_err("second tenant fails after first commits"); - match error { - DbError::EphemeralReapPartial { reaped, source } => { - assert_eq!(reaped.len(), 1); - assert_eq!(*reaped[0].community_id.as_uuid(), ids[0]); - assert!(source.to_string().contains("later tenant archive failed")); - } - other => panic!("expected committed rows with later failure, got {other:?}"), - } - for (id, expected) in [(ids[0], true), (ids[1], false)] { + .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", ) @@ -938,6 +956,17 @@ async fn channel_reap_postgres_tests_returns_committed_rows_after_later_tenant_e .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] @@ -1043,3 +1072,201 @@ async fn match_reap_admits_active_tenant_and_skips_quiescing_tenant() { 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}"); + } + } + } +} diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 91b07d2a43e..9b2c18dc879 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -855,7 +855,6 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result Result reaped.extend(rows), - Err(error) if reaped.is_empty() => return Err(error), - Err(error) => { - return Err(DbError::EphemeralReapPartial { - reaped, - source: Box::new(error), - }); - } + Err(error) => tracing::warn!(community_id = %id, error = %error, + "Ephemeral channel reap failed; retained rows will retry next tick"), } } Ok(reaped) diff --git a/crates/buzz-db/src/store/push.rs b/crates/buzz-db/src/store/push.rs index 18dfefa4a3a..88637ea4fe1 100644 --- a/crates/buzz-db/src/store/push.rs +++ b/crates/buzz-db/src/store/push.rs @@ -871,6 +871,7 @@ where 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 = sqlx::query_scalar( "SELECT q.community_id FROM push_match_queue q \ JOIN communities c ON c.id = q.community_id \ @@ -935,8 +936,8 @@ where if rows.is_empty() { return Ok(None); } - // Release the admitted connection before loading source events: the - // production loader acquires separately and size-one pools must work. + // 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 = row.try_get("event_id")?; @@ -1005,25 +1006,33 @@ pub async fn reap_exhausted_matches(pool: &PgPool) -> Result { drop(connection); let mut deleted = 0; for id in communities { - let Some(mut tx) = crate::begin_community_sweep_transaction( - pool, - CommunityId::from_uuid(id), - crate::observability::WriterOperation::Maintenance, - ) - .await? - else { - continue; - }; - 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?; - deleted += result.rows_affected(); + let batch: Result = 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) } diff --git a/crates/buzz-db/src/store/relay_invite.rs b/crates/buzz-db/src/store/relay_invite.rs index a4fc89b4bc8..252f313f940 100644 --- a/crates/buzz-db/src/store/relay_invite.rs +++ b/crates/buzz-db/src/store/relay_invite.rs @@ -182,6 +182,7 @@ pub async fn reap_expired_relay_invites(pool: &PgPool, cutoff: DateTime) -> .await?; // 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 \ @@ -208,26 +209,34 @@ pub async fn reap_expired_relay_invites(pool: &PgPool, cutoff: DateTime) -> let mut deleted = 0; for id in ordered_communities { - let community = CommunityId::from_uuid(id); - let Some(mut tx) = crate::begin_community_sweep_transaction( - pool, - community, - crate::observability::WriterOperation::Maintenance, - ) - .await? - else { - continue; - }; - 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?; - deleted += result.rows_affected(); + let batch: Result = 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) } diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index f5f10ed2e91..662ceb015d5 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -642,24 +642,32 @@ pub async fn prune_scheduled_workflow_fires_before( .await?; let mut deleted = 0; for id in communities { - let Some(mut tx) = crate::begin_community_sweep_transaction( - pool, - CommunityId::from_uuid(id), - crate::observability::WriterOperation::Maintenance, - ) - .await? - else { - continue; - }; - 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?; - deleted += result.rows_affected(); + let batch: Result = 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) } diff --git a/crates/buzz-db/src/test_support.rs b/crates/buzz-db/src/test_support.rs index 4e0143261e8..4159e5d83f6 100644 --- a/crates/buzz-db/src/test_support.rs +++ b/crates/buzz-db/src/test_support.rs @@ -58,31 +58,88 @@ pub(crate) async fn admission_fixture() -> (crate::Db, [buzz_core::CommunityId; /// 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) { - sqlx::query("CREATE SCHEMA admission_test") +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( - "CREATE FUNCTION admission_test.community_write_allowed(uuid) RETURNS boolean \ + 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("SET search_path = admission_test, public") - .execute(pool) - .await - .expect("resolve test-local predicate first"); + 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('admission_test.community_write_allowed(uuid)')", + "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(community: buzz_core::CommunityId, sweep: F) -> T +where + F: std::future::Future + 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 @@ -91,6 +148,15 @@ pub(crate) async fn reject_legacy_sql_fence_predicate(pool: &sqlx::PgPool) { /// 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', \ @@ -98,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( @@ -108,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 diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index f6d51fbe7e2..671a3d4d959 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -164,8 +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("begin_community_sweep_transaction(")); - assert!(backfill_d_tags.contains("execute(tx.conn())")); let soft_delete_discovery = event .split_once("pub async fn soft_delete_discovery_events") .expect("event store must expose discovery-event deletion") @@ -909,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) } @@ -916,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 {", ]; const GUARDED_TX_ADAPTER_METHOD_PINS: [(&str, &str); 1] = [( diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 680e2432dc8..7c1b6078408 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1191,10 +1191,6 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { let expired = match reaper_state.db.reap_expired_ephemeral_channels().await { Ok(ids) => ids, - Err(buzz_db::DbError::EphemeralReapPartial { reaped, source }) => { - error!("Ephemeral reaper tick failed after partial progress: {source}"); - reaped - } Err(e) => { error!("Ephemeral reaper tick failed: {e}"); continue; From 783ee13974f486bd0a884c3df79697c360a3d33a Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 10:32:55 -0400 Subject: [PATCH 3/5] fix(db): stop fleet sweeps when the database is unreachable The four periodic sweeps (ephemeral channel reap, exhausted match reap, relay invite reap, workflow fire prune) continue past a failing tenant so one tenant's error cannot hold back the others. A pool timeout, closed pool, or socket error is not tenant-local, though: continuing paid one acquire timeout and one warning per remaining tenant in the page. Route all four through one sweep_admitted_communities helper that skips fenced tenants, continues past tenant-local errors, and stops at the first connection-level error (DbError::is_connection_unavailable). Rows already committed are still returned, so reaped channels keep their system events. Also name the sweep chokepoint marker in observability_source (review P2). Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/error.rs | 35 +++++++++++++++++ crates/buzz-db/src/lib.rs | 2 +- crates/buzz-db/src/runtime/mod.rs | 40 ++++++++++++++++++++ crates/buzz-db/src/runtime/tests.rs | 27 +++++++++++++ crates/buzz-db/src/store/channel.rs | 31 +++++---------- crates/buzz-db/src/store/push.rs | 31 +++++---------- crates/buzz-db/src/store/relay_invite.rs | 33 ++++++---------- crates/buzz-db/src/store/workflow.rs | 31 +++++---------- crates/buzz-db/tests/observability_source.rs | 3 +- 9 files changed, 146 insertions(+), 87 deletions(-) diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 8c051a7a793..e27b319bed4 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -99,7 +99,42 @@ impl DbError { DbError::Sqlx(sqlx::Error::Database(db)) if db.code().as_deref() == Some("57014") ) } + + /// Whether the failure is about reaching the database at all (pool + /// acquire timeout, closed pool, or socket I/O) rather than about the + /// statement or its tenant. Every later query on the pool is expected to + /// fail the same way. + pub fn is_connection_unavailable(&self) -> bool { + matches!( + self, + DbError::Sqlx(sqlx::Error::PoolTimedOut | sqlx::Error::PoolClosed | sqlx::Error::Io(_)) + ) + } } /// Convenience alias for `Result`. pub type Result = std::result::Result; + +#[cfg(test)] +mod tests { + use super::DbError; + + #[test] + fn connection_unavailable_covers_only_pool_and_io_failures() { + for error in [ + sqlx::Error::PoolTimedOut, + sqlx::Error::PoolClosed, + sqlx::Error::Io(std::io::Error::from(std::io::ErrorKind::ConnectionReset)), + ] { + assert!(DbError::Sqlx(error).is_connection_unavailable()); + } + for error in [ + DbError::Sqlx(sqlx::Error::RowNotFound), + DbError::Sqlx(sqlx::Error::Protocol("bad frame".into())), + DbError::AccessDenied("write-fenced".into()), + DbError::InvalidData("bad row".into()), + ] { + assert!(!error.is_connection_unavailable(), "{error:?}"); + } + } +} diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 477723e4443..529034b46fd 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -62,7 +62,7 @@ pub(crate) use runtime::{ begin_community_event_write_transaction, 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, + sweep_admitted_communities, ReadSessionInner, RouteDecision, RoutePredicate, }; pub use store::{ admin_moderation, allowlist, api_token, archived_identities, artifact, channel, diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index ee15f0c3aa5..4bda0d71a9c 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -216,6 +216,46 @@ pub(crate) async fn begin_community_sweep_transaction( } } +/// Run one admitted mutation per community for a periodic fleet sweep and +/// return the committed results in order. Fenced communities are skipped. +/// A tenant-local failure is logged and the sweep moves on; that tenant's +/// rows retry next tick. A connection-level failure stops the sweep instead, +/// because every remaining tenant would wait out the same acquire timeout. +pub(crate) async fn sweep_admitted_communities( + pool: &PgPool, + communities: impl IntoIterator, + operation: observability::WriterOperation, + sweep: &'static str, + mut mutate: F, +) -> Vec +where + F: FnMut(Uuid, AdmittedTx) -> Fut, + Fut: std::future::Future>, +{ + let mut committed = Vec::new(); + for id in communities { + let result = + match begin_community_sweep_transaction(pool, CommunityId::from_uuid(id), operation) + .await + { + Ok(None) => continue, + Ok(Some(tx)) => mutate(id, tx).await, + Err(error) => Err(error), + }; + match result { + Ok(value) => committed.push(value), + Err(error) if error.is_connection_unavailable() => { + tracing::warn!(sweep, community_id = %id, error = %error, + "Fleet sweep stopped: database unavailable; remaining tenants retry next tick"); + break; + } + Err(error) => tracing::warn!(sweep, community_id = %id, error = %error, + "Fleet sweep tenant failed; retained rows retry next tick"), + } + } + committed +} + /// Database handle. Clone is cheap (Arc-backed pool). #[derive(Clone, Debug)] pub struct Db { diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 3fe785134ca..d35923623da 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -77,6 +77,33 @@ async fn setup_db() -> Db { Db::from_pool(pool) } +/// A pool that becomes unavailable after the candidate read must end the +/// sweep at the first tenant, not cost one acquire attempt per tenant. +#[tokio::test] +async fn fleet_sweep_stops_at_first_unavailable_connection() { + let pool = sqlx::postgres::PgPoolOptions::new() + .connect_lazy(&crate::test_support::database_url()) + .expect("construct lazy sweep pool"); + pool.close().await; + let attempted = std::cell::Cell::new(0); + let candidates = (0..3).map(|_| { + attempted.set(attempted.get() + 1); + Uuid::new_v4() + }); + + let committed = sweep_admitted_communities( + &pool, + candidates, + observability::WriterOperation::Maintenance, + "test_sweep", + |_, _| async { Ok(()) }, + ) + .await; + + assert!(committed.is_empty()); + assert_eq!(attempted.get(), 1, "sweep must stop after the outage"); +} + #[tokio::test] async fn event_write_transaction_preserves_legacy_acquisition_metrics() { use metrics_util::debugging::{DebugValue, DebuggingRecorder}; diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 9b2c18dc879..cf8d15094ed 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -860,18 +860,12 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result> = 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 reaped = crate::sweep_admitted_communities( + pool, + communities, + crate::observability::WriterOperation::Maintenance, + "ephemeral_channel_reap", + |id, mut tx| async move { let rows = sqlx::query( "UPDATE channels AS ch SET archived_at = NOW() \ FROM communities AS c \ @@ -896,15 +890,10 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result>>()?; 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) + }, + ) + .await; + Ok(reaped.into_iter().flatten().collect()) } impl Db { diff --git a/crates/buzz-db/src/store/push.rs b/crates/buzz-db/src/store/push.rs index 88637ea4fe1..24ae312521e 100644 --- a/crates/buzz-db/src/store/push.rs +++ b/crates/buzz-db/src/store/push.rs @@ -1004,18 +1004,12 @@ pub async fn reap_exhausted_matches(pool: &PgPool) -> Result { .fetch_all(&mut *connection) .await?; drop(connection); - let mut deleted = 0; - for id in communities { - let batch: Result = 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 deleted = crate::sweep_admitted_communities( + pool, + communities, + crate::observability::WriterOperation::Maintenance, + "exhausted_match_reap", + |id, mut tx| async move { 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()))", @@ -1026,15 +1020,10 @@ pub async fn reap_exhausted_matches(pool: &PgPool) -> Result { .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) + }, + ) + .await; + Ok(deleted.into_iter().sum()) } /// Load active endpoint-enabled leases for one tenant. diff --git a/crates/buzz-db/src/store/relay_invite.rs b/crates/buzz-db/src/store/relay_invite.rs index 252f313f940..b6d76fb7349 100644 --- a/crates/buzz-db/src/store/relay_invite.rs +++ b/crates/buzz-db/src/store/relay_invite.rs @@ -207,19 +207,13 @@ pub async fn reap_expired_relay_invites(pool: &PgPool, cutoff: DateTime) -> ids_by_community.entry(community).or_default().push(id); } - let mut deleted = 0; - for id in ordered_communities { - let batch: Result = 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 ids_by_community = &ids_by_community; + let deleted = crate::sweep_admitted_communities( + pool, + ordered_communities, + crate::observability::WriterOperation::Maintenance, + "relay_invite_reap", + |id, mut tx| async move { let result = sqlx::query( "DELETE FROM relay_invites WHERE community_id = $1 AND id = ANY($2) AND expires_at < $3", ) @@ -230,15 +224,10 @@ pub async fn reap_expired_relay_invites(pool: &PgPool, cutoff: DateTime) -> .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) + }, + ) + .await; + Ok(deleted.into_iter().sum()) } /// Atomically claim a v2 relay invite. diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 662ceb015d5..6f7f64aa2da 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -640,18 +640,12 @@ pub async fn prune_scheduled_workflow_fires_before( .bind(older_than) .fetch_all(pool) .await?; - let mut deleted = 0; - for id in communities { - let batch: Result = 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 deleted = crate::sweep_admitted_communities( + pool, + communities, + crate::observability::WriterOperation::Maintenance, + "scheduled_workflow_fire_prune", + |id, mut tx| async move { let result = sqlx::query( "DELETE FROM scheduled_workflow_fires WHERE community_id = $1 AND claimed_at < $2", ) @@ -661,15 +655,10 @@ pub async fn prune_scheduled_workflow_fires_before( .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) + }, + ) + .await; + Ok(deleted.into_iter().sum()) } /// Update a workflow's name, definition, and definition_hash. diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 671a3d4d959..9b05eeb94aa 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -902,12 +902,13 @@ const GUARDED_TABLE_WRITE_MARKERS: [&str; 9] = [ const COMMUNITY_CHOKEPOINT_MARKER: &str = "begin_community_event_write_transaction("; const GENERAL_COMMUNITY_CHOKEPOINT_MARKER: &str = "begin_community_write_transaction("; +const SWEEP_COMMUNITY_CHOKEPOINT_MARKER: &str = "begin_community_sweep_transaction("; const COMMUNITY_CHOKEPOINT_LEGACY_MARKER: &str = "begin_community_event_write_transaction_with_legacy_metrics("; fn has_any_tenant_local_chokepoint(source: &str) -> bool { source.contains(COMMUNITY_CHOKEPOINT_MARKER) - || source.contains("begin_community_sweep_transaction(") + || source.contains(SWEEP_COMMUNITY_CHOKEPOINT_MARKER) || source.contains(GENERAL_COMMUNITY_CHOKEPOINT_MARKER) || source.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER) } From 88b3508faf0c66eac5dfc9cba7237c358d48710a Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 10:43:54 -0400 Subject: [PATCH 4/5] test(db): pin committed results when a fleet sweep stops on outage Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- .../src/store/admission_postgres_tests.rs | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/crates/buzz-db/src/store/admission_postgres_tests.rs b/crates/buzz-db/src/store/admission_postgres_tests.rs index af505529c71..9ae81f3a830 100644 --- a/crates/buzz-db/src/store/admission_postgres_tests.rs +++ b/crates/buzz-db/src/store/admission_postgres_tests.rs @@ -969,6 +969,40 @@ async fn channel_reap_returns_committed_rows_after_earlier_tenant_error() { assert_eq!(*retried[0].community_id.as_uuid(), ids[0]); } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn fleet_sweep_returns_committed_results_when_database_drops_mid_page() { + let (db, communities) = fixture().await; + let [first, second] = two_active_tenants(&db, communities[0]).await; + let candidates = [first, second, Uuid::new_v4()]; + let pulled = std::cell::Cell::new(0); + let pool = db.pool.clone(); + let committed = crate::sweep_admitted_communities( + &db.pool, + candidates + .iter() + .copied() + .inspect(|_| pulled.set(pulled.get() + 1)), + crate::observability::WriterOperation::Maintenance, + "test_sweep", + |id, tx| { + let pool = pool.clone(); + async move { + tx.commit().await?; + pool.close().await; + Ok(id) + } + }, + ) + .await; + assert_eq!(committed, vec![first], "rows committed before the outage"); + assert_eq!( + pulled.get(), + 2, + "the sweep must stop at the unavailable tenant" + ); +} + #[tokio::test] #[ignore = "requires Postgres"] async fn d_tag_backfill_admits_active_tenant_and_skips_quiescing_tenant() { From 5d37fbda699f56231930d949643490e35c70d2c8 Mon Sep 17 00:00:00 2001 From: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Date: Fri, 16 Oct 2026 11:11:13 -0400 Subject: [PATCH 5/5] test(db): pass gate_barrier to enqueue in the admission test after #8352 Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/runtime/push_enqueue.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/runtime/push_enqueue.rs b/crates/buzz-db/src/runtime/push_enqueue.rs index 6377e9a82c3..3cbfd7cc802 100644 --- a/crates/buzz-db/src/runtime/push_enqueue.rs +++ b/crates/buzz-db/src/runtime/push_enqueue.rs @@ -290,11 +290,13 @@ mod tests { event_id: vec![1_u8; 32], barrier: None, }; - let err = enqueue(&db.pool, &quiescing) + let err = enqueue(&db.pool, &quiescing, None) .await .expect_err("quiescing enqueue"); assert!(crate::test_support::is_admission_rejection(&err)); - enqueue(&db.pool, &active).await.expect("admitted enqueue"); + 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",