diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 8c051a7a79..e27b319bed 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 2f306c35d0..529034b46f 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, + 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 266003f5a0..4bda0d71a9 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -202,6 +202,60 @@ 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), + } +} + +/// 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/push_enqueue.rs b/crates/buzz-db/src/runtime/push_enqueue.rs index ac9c6e2f5d..3cbfd7cc80 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,70 @@ mod tests { assert_eq!(receiver.len(), CAPACITY - 1); producer.join().await; } + + mod postgres_tests { + use super::*; + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn enqueue_admits_active_tenant_and_rejects_quiescing_tenant() { + let (db, communities) = crate::test_support::admission_fixture().await; + crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await; + for (index, community) in communities[..2].iter().enumerate() { + sqlx::query( + "INSERT INTO events \ + (community_id, id, pubkey, created_at, kind, tags, content, sig) \ + VALUES ($1, $2, $3, to_timestamp(1), 9, '[]', '', $4)", + ) + .bind(community.as_uuid()) + .bind(vec![index as u8; 32]) + .bind(vec![1_u8; 32]) + .bind(vec![2_u8; 64]) + .execute(&db.pool) + .await + .expect("seed event"); + sqlx::query( + "INSERT INTO push_leases (community_id, author, installation_id, \ + source_event_id, source_created_at, generation, active, endpoint_hash, \ + endpoint_grant, max_class, subscriptions, expires_at, updated_at) \ + VALUES ($1, $2, 'test', $3, 1, 1, true, $4, 'grant', 'default', '[]', \ + EXTRACT(EPOCH FROM now())::bigint + 3600, now() - interval '1 hour')", + ) + .bind(community.as_uuid()) + .bind(vec![3_u8; 32]) + .bind(vec![4_u8; 32]) + .bind(vec![5_u8; 32]) + .execute(&db.pool) + .await + .expect("seed lease"); + } + let active = Job { + community: communities[0], + event_id: vec![0_u8; 32], + barrier: None, + }; + let quiescing = Job { + community: communities[1], + event_id: vec![1_u8; 32], + barrier: None, + }; + let err = enqueue(&db.pool, &quiescing, None) + .await + .expect_err("quiescing enqueue"); + assert!(crate::test_support::is_admission_rejection(&err)); + enqueue(&db.pool, &active, None) + .await + .expect("admitted enqueue"); + for (community, expected) in [(communities[0], 1), (communities[1], 0)] { + let count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM push_match_queue WHERE community_id = $1", + ) + .bind(community.as_uuid()) + .fetch_one(&db.pool) + .await + .expect("read enqueue state"); + assert_eq!(count, expected, "queue state for {community}"); + } + } + } } diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 3fe785134c..d35923623d 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/admission_postgres_tests.rs b/crates/buzz-db/src/store/admission_postgres_tests.rs index 2c65beecb4..9ae81f3a83 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,552 @@ async fn owner_mutations_report_pending_deletion_when_admission_refuses() { assert_eq!(count_relay_members(&db, quiescing).await, 0); assert_eq!(count_relay_members(&db, fenced).await, 0); } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn invite_sweep_admits_active_tenant_and_skips_quiescing_tenant() { + let (db, communities) = fixture().await; + crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await; + let cutoff = chrono::Utc::now(); + for (index, community) in communities[..2].iter().enumerate() { + sqlx::query( + "INSERT INTO relay_invites (community_id, token_hash, expires_at, created_by) \ + VALUES ($1, $2, $3, 'test')", + ) + .bind(community.as_uuid()) + .bind(vec![index as u8; 32]) + .bind(cutoff - chrono::Duration::minutes(1)) + .execute(&db.pool) + .await + .expect("seed invite"); + } + assert_eq!( + super::relay_invite::reap_expired_relay_invites(&db.pool, cutoff) + .await + .unwrap(), + 1 + ); + for (community, expected) in [(communities[0], 0), (communities[1], 1)] { + let count: i64 = + sqlx::query_scalar("SELECT count(*) FROM relay_invites WHERE community_id = $1") + .bind(community.as_uuid()) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(count, expected, "invite state for {community}"); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_prune_admits_active_tenant_and_skips_quiescing_tenant() { + let (db, communities) = fixture().await; + crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await; + let cutoff = chrono::Utc::now(); + for community in &communities[..2] { + sqlx::query( + "INSERT INTO scheduled_workflow_fires \ + (community_id, workflow_id, scheduled_for, claimed_at) VALUES ($1, $2, $3, $4)", + ) + .bind(community.as_uuid()) + .bind(Uuid::new_v4()) + .bind(cutoff - chrono::Duration::hours(2)) + .bind(cutoff - chrono::Duration::hours(1)) + .execute(&db.pool) + .await + .expect("seed fire"); + } + assert_eq!( + super::workflow::prune_scheduled_workflow_fires_before(&db.pool, cutoff) + .await + .unwrap(), + 1 + ); + for (community, expected) in [(communities[0], 0), (communities[1], 1)] { + let count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM scheduled_workflow_fires WHERE community_id = $1", + ) + .bind(community.as_uuid()) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(count, expected, "fire state for {community}"); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn channel_reap_admits_active_tenant_and_skips_quiescing_tenant() { + let (db, communities) = fixture().await; + crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await; + for community in &communities[..2] { + sqlx::query( + "INSERT INTO channels (community_id, name, created_by, ttl_seconds, ttl_deadline) \ + VALUES ($1, 'expired', $2, 60, now() - interval '1 minute')", + ) + .bind(community.as_uuid()) + .bind(vec![1_u8; 32]) + .execute(&db.pool) + .await + .expect("seed channel"); + } + let reaped = super::channel::reap_expired_ephemeral_channels(&db.pool) + .await + .unwrap(); + assert_eq!(reaped.len(), 1); + assert_eq!(reaped[0].community_id, communities[0]); + for (community, expected) in [(communities[0], true), (communities[1], false)] { + let archived: bool = sqlx::query_scalar( + "SELECT archived_at IS NOT NULL FROM channels WHERE community_id = $1", + ) + .bind(community.as_uuid()) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(archived, expected, "archive state for {community}"); + } +} + +/// Keep every failure-injection object, including the ALWAYS trigger's table, +/// in this call's shadow schema so a shared-database run cannot leak it. +async fn fail_tenant_mutations(pool: &sqlx::PgPool, schema: &str, table: &str, id: Uuid) { + // Dynamic identifiers come only from our UUID schema and literal table names. + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE TABLE {schema}.{table} (LIKE public.{table} INCLUDING ALL)" + ))) + .execute(pool) + .await + .expect("create test-local serving table"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE TABLE {schema}.fail_tenant (id uuid PRIMARY KEY)" + ))) + .execute(pool) + .await + .expect("create failure target"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "INSERT INTO {schema}.fail_tenant (id) VALUES ($1)" + ))) + .bind(id) + .execute(pool) + .await + .expect("select failing tenant"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION {schema}.fail_tenant_mutation() RETURNS trigger \ + LANGUAGE plpgsql AS $$ BEGIN \ + IF EXISTS (SELECT 1 FROM {schema}.fail_tenant \ + WHERE id = COALESCE(NEW.community_id, OLD.community_id)) \ + THEN RAISE EXCEPTION 'tenant mutation failed'; END IF; \ + IF TG_OP = 'DELETE' THEN RETURN OLD; END IF; RETURN NEW; END $$", + ))) + .execute(pool) + .await + .expect("create failure trigger function"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE TRIGGER fail_tenant_mutation BEFORE UPDATE OR DELETE ON {schema}.{table} \ + FOR EACH ROW EXECUTE FUNCTION {schema}.fail_tenant_mutation()", + ))) + .execute(pool) + .await + .expect("create test-local failure trigger"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "ALTER TABLE {schema}.{table} ENABLE ALWAYS TRIGGER fail_tenant_mutation" + ))) + .execute(pool) + .await + .expect("enable trigger with replica role"); +} + +async fn two_active_tenants(db: &Db, first: CommunityId) -> [Uuid; 2] { + let extra = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(extra) + .bind(format!("admission-extra-{extra}.example")) + .execute(&db.pool) + .await + .expect("seed second active tenant"); + let mut ids = [*first.as_uuid(), extra]; + ids.sort(); + ids +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn channel_reap_returns_committed_rows_after_earlier_tenant_error() { + let (db, communities) = fixture().await; + let schema = crate::test_support::reject_legacy_sql_fence_predicate(&db.pool).await; + let ids = two_active_tenants(&db, communities[0]).await; + fail_tenant_mutations(&db.pool, &schema, "channels", ids[0]).await; + for id in ids { + sqlx::query( + "INSERT INTO channels (community_id, name, created_by, ttl_seconds, ttl_deadline) \ + VALUES ($1, 'expired', $2, 60, now() - interval '1 minute')", + ) + .bind(id) + .bind(vec![1_u8; 32]) + .execute(&db.pool) + .await + .expect("seed expired channel"); + } + let reaped = super::channel::reap_expired_ephemeral_channels(&db.pool) + .await + .expect("tenant failure must not stop the fleet sweep"); + assert_eq!(reaped.len(), 1); + assert_eq!(*reaped[0].community_id.as_uuid(), ids[1]); + for (id, expected) in [(ids[0], false), (ids[1], true)] { + let archived: bool = sqlx::query_scalar( + "SELECT archived_at IS NOT NULL FROM channels WHERE community_id = $1", + ) + .bind(id) + .fetch_one(&db.pool) + .await + .expect("read archive state"); + assert_eq!(archived, expected, "archive state for {id}"); + } + sqlx::query(sqlx::AssertSqlSafe(format!( + "DELETE FROM {schema}.fail_tenant" + ))) + .execute(&db.pool) + .await + .unwrap(); + let retried = super::channel::reap_expired_ephemeral_channels(&db.pool) + .await + .unwrap(); + assert_eq!(retried.len(), 1); + assert_eq!(*retried[0].community_id.as_uuid(), ids[0]); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn 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() { + 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}"); + } +} + +#[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 ffc60694ee..cf8d15094e 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -847,33 +847,53 @@ 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", + 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 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 \ + 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; + Ok(reaped.into_iter().flatten().collect()) } impl Db { diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 3e76d75190..9474623da9 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 26eaee65e4..24ae312521 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,47 @@ 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?; + // 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 \ + WHERE q.attempts < $1 AND q.next_attempt_at <= now() \ + AND (q.state = 'pending' OR (q.state = 'matching' AND q.lease_until < now())) \ + AND c.deletion_state = 'active' AND c.deleted_at IS NULL \ + AND q.community_id <> ALL($2) \ + ORDER BY q.next_attempt_at, q.created_at LIMIT 1", + ) + .bind(MAX_MATCH_ATTEMPTS) + .bind(&refused) + .fetch_optional(&mut *connection) + .await?; + drop(connection); + let Some(id) = candidate else { + break; + }; + let community = CommunityId::from_uuid(id); + let Some(mut tx) = crate::begin_community_sweep_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await? + else { + refused.push(id); + continue; + }; + let rows = sqlx::query( + r#" + WITH candidates AS ( SELECT q.community_id, q.event_id FROM push_match_queue q - JOIN target t ON q.community_id = t.community_id - WHERE q.attempts < $3 + WHERE q.community_id = $5 AND q.attempts < $3 AND q.next_attempt_at <= now() AND (q.state = 'pending' OR (q.state = 'matching' AND q.lease_until < now())) ORDER BY q.next_attempt_at, q.created_at @@ -893,22 +918,26 @@ where WHERE q.community_id=c.community_id AND q.event_id=c.event_id RETURNING q.community_id, q.event_id, q.attempts "#, - ) - .bind(claim_id) - .bind(lease_until) - .bind(MAX_MATCH_ATTEMPTS) - .bind(limit) - .fetch_all(&mut *connection) - .await?; + ) + .bind(claim_id) + .bind(lease_until) + .bind(MAX_MATCH_ATTEMPTS) + .bind(limit) + .bind(community.as_uuid()) + .fetch_all(tx.conn()) + .await?; + tx.commit().await?; + claimed = Some((community, rows)); + break; + } + let Some((community, rows)) = claimed else { + return Ok(None); + }; if rows.is_empty() { return Ok(None); } - // The claim query is a single autocommitted statement. Release its pool - // slot before loading source events, because the production loader owns a - // separately attributed acquisition. Holding this connection across the - // load would self-starve a supported size-one writer pool. - drop(connection); - let community = CommunityId::from_uuid(rows[0].try_get("community_id")?); + // The admitted transaction committed above, so no connection is held + // across the loader (size-one pools must work). let mut attempts = std::collections::HashMap::with_capacity(rows.len()); for row in &rows { let event_id: Vec = row.try_get("event_id")?; @@ -967,15 +996,34 @@ 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 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()))", + ) + .bind(id) + .bind(MAX_MATCH_ATTEMPTS) + .execute(tx.conn()) + .await?; + tx.commit().await?; + Ok(result.rows_affected()) + }, + ) + .await; + Ok(deleted.into_iter().sum()) } /// Load active endpoint-enabled leases for one tenant. @@ -2601,7 +2649,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 +2679,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..b6d76fb734 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,54 @@ 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. + // Exclude fenced tenants so they cannot fill the LIMIT page. + let candidates = sqlx::query( + "SELECT invite.community_id, invite.id FROM relay_invites invite \ + JOIN communities community ON community.id = invite.community_id \ + WHERE invite.expires_at < $1 AND community.deletion_state = 'active' \ + AND community.deleted_at IS NULL \ + ORDER BY invite.expires_at LIMIT $2", ) .bind(cutoff) .bind(RETENTION_SWEEP_BATCH_SIZE) - .execute(&mut *connection) + .fetch_all(&mut *connection) .await?; + drop(connection); - Ok(result.rows_affected()) + let mut ordered_communities = Vec::new(); + let mut ids_by_community = std::collections::HashMap::>::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 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", + ) + .bind(id) + .bind(&ids_by_community[&id]) + .bind(cutoff) + .execute(tx.conn()) + .await?; + tx.commit().await?; + Ok(result.rows_affected()) + }, + ) + .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 8f63d85555..6f7f64aa2d 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -634,18 +634,31 @@ 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 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", + ) + .bind(id) + .bind(older_than) + .execute(tx.conn()) + .await?; + tx.commit().await?; + Ok(result.rows_affected()) + }, + ) + .await; + Ok(deleted.into_iter().sum()) } /// 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..4159e5d83f 100644 --- a/crates/buzz-db/src/test_support.rs +++ b/crates/buzz-db/src/test_support.rs @@ -8,12 +8,155 @@ pub(crate) fn database_url() -> String { .unwrap_or_else(|_| DEFAULT_DATABASE_URL.to_owned()) } +/// One-session fixture that bypasses ordinary write-fence triggers while +/// retaining application admission. Communities are active, quiescing, fenced. +pub(crate) async fn admission_fixture() -> (crate::Db, [buzz_core::CommunityId; 3]) { + use buzz_core::CommunityId; + use uuid::Uuid; + + let db = crate::Db::new(&crate::DbConfig { + database_url: database_url(), + max_connections: 1, + min_connections: 1, + ..crate::DbConfig::default() + }) + .await + .expect("connect test database"); + let pool = &db.pool; + let mut communities = [CommunityId::from_uuid(Uuid::nil()); 3]; + for community in &mut communities { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(format!("admission-{id}.example")) + .execute(pool) + .await + .expect("create community"); + *community = CommunityId::from_uuid(id); + } + set_deletion_state(pool, *communities[1].as_uuid(), "quiescing").await; + set_deletion_state(pool, *communities[2].as_uuid(), "fenced").await; + // The one physical session serves every API call in this fixture. + sqlx::query("SET session_replication_role = replica") + .execute(pool) + .await + .expect("disable ordinary triggers"); + let role: String = sqlx::query_scalar("SHOW session_replication_role") + .fetch_one(pool) + .await + .expect("read replication role"); + assert_eq!(role, "replica"); + // A raw write to the fenced tenant proves the DB backstop is bypassed. + sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)") + .bind(communities[2].as_uuid()) + .bind(vec![0xfe_u8; 32]) + .execute(pool) + .await + .expect("raw fenced insert succeeds with triggers disabled"); + (db, communities) +} + +/// In one per-test database, make any return to the retired SQL predicate +/// observable without changing the production function in the public schema. +pub(crate) async fn reject_legacy_sql_fence_predicate(pool: &sqlx::PgPool) -> String { + let schema = format!("admission_test_{}", uuid::Uuid::new_v4().simple()); + sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {schema}"))) + .execute(pool) + .await + .expect("create test-only function schema"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION {schema}.community_write_allowed(uuid) RETURNS boolean \ + LANGUAGE plpgsql VOLATILE AS $$ BEGIN \ + RAISE EXCEPTION 'legacy SQL fence predicate called'; END $$", + ))) + .execute(pool) + .await + .expect("shadow legacy SQL predicate"); + sqlx::query(sqlx::AssertSqlSafe(format!( + "SET search_path = {schema}, public" + ))) + .execute(pool) + .await + .expect("resolve test-local predicate first"); + let shadowed: bool = sqlx::query_scalar( + "SELECT to_regprocedure('community_write_allowed(uuid)') = to_regprocedure($1)", + ) + .bind(format!("{schema}.community_write_allowed(uuid)")) + .fetch_one(pool) + .await + .expect("resolve shadowed predicate"); + assert!(shadowed); + schema +} + +/// Hold the executor's exclusive lock until a sweep reaches shared admission, +/// then quiesce in that same transaction before releasing it. The ordinary +/// second pool keeps lifecycle triggers enabled; the sweep fixture bypasses them. +pub(crate) async fn fence_while_sweep_waits(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 /// executor does: the tombstone trigger only accepts lifecycle changes from a /// transaction carrying the executor GUCs for that community, including its /// current fence generation. pub(crate) async fn set_deletion_state(pool: &sqlx::PgPool, id: uuid::Uuid, state: &str) { let mut tx = pool.begin().await.expect("begin lifecycle fixture"); + set_deletion_state_in_transaction(&mut tx, id, state).await; + tx.commit().await.expect("commit lifecycle fixture"); +} + +async fn set_deletion_state_in_transaction( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + id: uuid::Uuid, + state: &str, +) { sqlx::query( "SELECT set_config('buzz.deletion_executor_community', $1, true), \ set_config('buzz.deletion_fence_generation', \ @@ -21,7 +164,7 @@ pub(crate) async fn set_deletion_state(pool: &sqlx::PgPool, id: uuid::Uuid, stat ) .bind(id.to_string()) .bind(id) - .execute(&mut *tx) + .execute(&mut **tx) .await .expect("authorize lifecycle fixture"); sqlx::query( @@ -31,10 +174,9 @@ pub(crate) async fn set_deletion_state(pool: &sqlx::PgPool, id: uuid::Uuid, stat ) .bind(id) .bind(state) - .execute(&mut *tx) + .execute(&mut **tx) .await .expect("set lifecycle state"); - tx.commit().await.expect("commit lifecycle fixture"); } /// Move a disposable test community into `quiescing`, the state in which diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 06aa82c8e6..9b05eeb94a 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -164,7 +164,6 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() { .expect("d-tag backfill must precede discovery deletion") .0; assert!(backfill_d_tags.contains("WriterOperation::Bootstrap")); - assert!(backfill_d_tags.contains("execute(&mut *connection)")); let soft_delete_discovery = event .split_once("pub async fn soft_delete_discovery_events") .expect("event store must expose discovery-event deletion") @@ -903,11 +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(SWEEP_COMMUNITY_CHOKEPOINT_MARKER) || source.contains(GENERAL_COMMUNITY_CHOKEPOINT_MARKER) || source.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER) } @@ -915,10 +916,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] = [(