From 81d30a157baad539515352d7558b59f1946c6ddb 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/2] 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 8c051a7a79..db697ba730 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 2f306c35d0..477723e444 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 266003f5a0..ee15f0c3aa 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 40b1dcb52f..733e09059d 100644 --- a/crates/buzz-db/src/runtime/push_enqueue.rs +++ b/crates/buzz-db/src/runtime/push_enqueue.rs @@ -165,7 +165,6 @@ async fn enqueue(pool: &PgPool, job: &Job) -> crate::Result<()> { "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 \ @@ -203,4 +202,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 6f1d8cab68..eb46279951 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!( @@ -794,3 +750,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 13a93263e9..9bcce096fe 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -827,33 +827,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 183585e278..042268229c 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -2664,7 +2664,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( @@ -2672,19 +2672,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 4269c3f7e6..304858d1c6 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. @@ -2571,7 +2621,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"); @@ -2601,7 +2651,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 6487696151..a4fc89b4bc 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); - Ok(result.rows_affected()) + 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); + } + + 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 8f63d85555..f5f10ed2e9 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 e94b8b1449..4e0143261e 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 4d4edcf8e9..ba6026d444 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 4ed127dfff..4e93a81b95 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 2c151217897aa3b0cd33e20de2994eba22721c05 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/2] 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 db697ba730..8c051a7a79 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 733e09059d..b63ecf1f7e 100644 --- a/crates/buzz-db/src/runtime/push_enqueue.rs +++ b/crates/buzz-db/src/runtime/push_enqueue.rs @@ -213,22 +213,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]) @@ -248,7 +249,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 eb46279951..df89dbb2eb 100644 --- a/crates/buzz-db/src/store/admission_postgres_tests.rs +++ b/crates/buzz-db/src/store/admission_postgres_tests.rs @@ -856,11 +856,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) @@ -868,8 +913,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) \ @@ -881,49 +936,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) + let reaped = super::channel::reap_expired_ephemeral_channels(&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)] { + .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", ) @@ -933,6 +951,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] @@ -1038,3 +1067,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 9bcce096fe..bffc1cc0e9 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -835,7 +835,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 304858d1c6..099b773c21 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 a4fc89b4bc..252f313f94 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 f5f10ed2e9..662ceb015d 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 4e0143261e..4159e5d83f 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 ba6026d444..d04c2e2ca1 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 4e93a81b95..4ed127dfff 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;