From 60de1db0b8813766d8e004297798f2ff759d3089 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 16 Oct 2026 15:46:37 -0400 Subject: [PATCH] fix(db): admit fleet sweeps per tenant instead of in-SQL fence predicate (#8318) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What Removes every production call to the SQL fence predicate `community_write_allowed()` from application SQL, so no application mutation depends on the boolean predicate anymore. This is one prerequisite for BUZZ-315 part B; #8322 (tolerating absent fence objects) is a separate one. The triggers and SQL functions stay in this PR because mixed-version pods still depend on them. The predicate isn't inlined. It's a VOLATILE plpgsql function that takes the deletion lock and *then* reads lifecycle state on a fresh snapshot, and an inline expression would read state from the statement's start snapshot. A separate earlier check would race deletion. So each sweep now: 1. reads its candidate community IDs with a plain read, bounded the same way the original statement was, 2. opens one `AdmittedTx` per community (`begin_community_sweep_transaction`, a thin wrapper over `begin_community_write_transaction` that turns an admission refusal into "skip this tenant"; the four periodic sweeps go through `sweep_admitted_communities`, which also owns their error handling), 3. runs the original mutation scoped to `community_id = $1` and commits. | Site | Change | | --- | --- | | `runtime/push_enqueue.rs` | Deleted the redundant predicate. This transaction already calls `guard_transaction`. | | `store/relay_invite.rs` `reap_expired_relay_invites` | Keeps the global expiry order and the 1,000-row ceiling. Candidates are pre-filtered to active tenants so fenced invites can't fill the batch. Admission is still what authorizes the write. | | `store/workflow.rs` `prune_scheduled_workflow_fires_before` | Prunes per admitted community and sums the row counts. | | `store/channel.rs` `reap_expired_ephemeral_channels` | Archives per admitted community and returns the committed rows, so the relay still runs their system and discovery follow-ups. The candidate read has no lifecycle pre-filter (it has no `LIMIT`, so admission alone decides). | | `store/event.rs` `backfill_d_tags` | Backfills per admitted community at startup. | | `store/push.rs` `claim_due_match_batch_with_loader` | Picks the earliest due active community, admits it, and claims under the same `FOR UPDATE OF q SKIP LOCKED` and batch limit. It skips at most 16 tenants that fence between the read and admission. | | `store/push.rs` `reap_exhausted_matches` | Reaps per admitted community. | **Tenant failures.** The four periodic sweeps (invite reap, workflow fire prune, channel reap, exhausted match reap) log a warning and continue when one tenant fails. That tenant's rows are left unmodified and stay eligible, so the next tick retries them. A tenant error doesn't abort processing of the other tenants already in the selected candidate page. Invite cleanup keeps its oldest-first 1,000-row selection, so if one tenant owns that entire page and its delete keeps failing, tenants outside the page wait. No persistent tenant-local failure for that delete is known in the current schema, and base aborted the whole page on any error. The returned count or rows cover the tenants that committed. Claim, enqueue and the startup backfill still return the error to the caller. **Database outages.** A connection-level error (pool acquire timeout, closed pool, socket I/O; `DbError::is_connection_unavailable`) isn't tenant-local, so it stops the periodic sweep at that tenant instead of moving on. Without this, an outage that began after the candidate read cost one writer acquire timeout and one warning per remaining tenant, and invite cleanup runs inside the usage-metrics leader tick. Rows already committed are still returned, so reaped channels keep their follow-ups. Remaining tenants retry next tick. **Known limit (pre-existing, tracked separately):** - *Claim at a locked head.* `claim_due_match_batch_with_loader` returns `None` once an admitted tenant claims zero rows. If every due row in the oldest active tenant is row-locked, younger tenants aren't tried on that call. Base behaved the same way. This PR doesn't change it. ## Tests All tests use a one-session fixture (`test_support::admission_fixture`) with `session_replication_role = replica`, so the trigger backstop is off and only application admission protects the fenced tenants. The fixture shadows `community_write_allowed` with a function that raises, in a per-call `admission_test_` schema. - One test per site: a quiescing tenant's rows are untouched while an active tenant in the same sweep still progresses. - Race tests for invite reap, channel reap and claim. These are the sites where the candidate read already filters out fenced tenants. Each test holds the tenant's deletion lock, waits until the sweep is blocked in admission, fences the tenant in that transaction, and asserts its rows are untouched. The claim test also asserts that a second active tenant is claimed in the same call. - `channel_reap_returns_committed_rows_after_earlier_tenant_error` and `periodic_reaps_continue_after_tenant_error_and_retry_retained_rows` inject a per-tenant failure. They check that later tenants commit, that the failed tenant's rows are retained, and that the next call processes them. - `fleet_sweep_stops_at_first_unavailable_connection` runs the sweep helper against a closed pool with three candidates and asserts only the first is attempted. `fleet_sweep_returns_committed_results_when_database_drops_mid_page` (PostgreSQL) commits the first tenant, closes the pool, and asserts the first tenant's result is returned and the sweep stops at the second of three candidates. `connection_unavailable_covers_only_pool_and_io_failures` pins the classification; the per-tenant failure tests above confirm a database-raised error still continues. Falsification, each with the tests failing as expected: | Mutation | Result | | --- | --- | | Original SQL restored at invite sweep, push claim, d-tag backfill | 0/3 pass (`legacy SQL fence predicate called`) | | `guard_community_write` returns `Ok(())` after taking the lock | race tests + channel reap test 0/4 pass | | Per-tenant errors abort the sweep again | 0/2 pass | | Sweep helper `continue`s instead of stopping on a connection-level error | `fleet_sweep_stops_at_first_unavailable_connection` and `fleet_sweep_returns_committed_results_when_database_drops_mid_page` fail | | Sweep helper drops already-committed results when it stops (`return Vec::new()`) | `fleet_sweep_returns_committed_results_when_database_drops_mid_page` fails | | Sweep helper removed from the source-policy chokepoint check | `serving_table_writes_expose_syntactic_chokepoint_or_guarded_tx_routes` fails | The PostgreSQL tests are under `postgres_tests` modules with `#[ignore = "requires Postgres"]`, which puts them in the `.config/nextest.toml` PostgreSQL filter. CI evidence is under Verification. ## Verification CI at head `5d37fbda` ([run 38925996275](https://github.com/block/buzz/actions/runs/38925996275)) is green. The PostgreSQL job ([job 117062895683](https://github.com/block/buzz/actions/runs/38925996275/job/117062895683)) passed 1017/1017. That includes named PASS lines for all 30 `admission_postgres_tests`, `fleet_sweep_returns_committed_results_when_database_drops_mid_page`, and both writer-pool tests listed below. Unit Tests ([job 117058136482](https://github.com/block/buzz/actions/runs/38925996275/job/117058136482)) has PASS lines for `connection_unavailable_covers_only_pool_and_io_failures` and `fleet_sweep_stops_at_first_unavailable_connection`. Local gates ran on a Linux cloud workstation against private PostgreSQL 16 and Redis. The tree was `735170df`, which is this head rebased on `main` `c0ab789`, so the gates built the merge result. | Gate | Result | | --- | --- | | `cargo fmt --all -- --check` | pass | | `cargo clippy --workspace --all-targets -- -D warnings` | pass | | `cargo nextest run -p buzz-db --lib --tests` | 185 passed | | `scripts/test-postgres-test-discovery.sh` | pass | | PostgreSQL lane for buzz-db (excluding the two writer-pool tests below) | 530 passed | | PostgreSQL lanes for buzz-admin, buzz-audit, buzz-deletion, buzz-push-gateway, buzz-search, buzz-workflow | all pass | | PostgreSQL lane for buzz-relay | 413/426. The 13 `group_state_postgres_tests` fail with `PoolTimedOut` on unmodified `main` `c0ab789` too (see below) | | The three mutations in the falsification table above (`continue` ×2, `return Vec::new()`) | each fails its target test | These failures happen only on the workstation, and each one passes in CI's PostgreSQL job at this head: - `writer_pool_metrics_stop_after_session_timeout_setup_failure` stalls. It's an unchanged writer-pool configuration test, and it stalled the same way at earlier heads. - The 13 relay `group_state_postgres_tests` fail at the same point, `ensure community: Sqlx(PoolTimedOut)`, on unmodified `main` `c0ab789`. This PR doesn't touch `buzz-relay`. ## What got simpler - No application mutation calls the boolean predicate `community_write_allowed()` anymore. Dropping the fence objects still has other dependencies: serving-write lease acquisition calls `assert_community_write_allowed()` (`store/deletion.rs:3068`), catalog validation requires both functions (`store/deletion.rs:1195-1198`), and the triggers and compatibility SQL still use them. Those belong to #8322 and part B, not this PR. - The claim's `target` CTE is gone, and every sweep uses the same admission-and-skip helper. - The four periodic sweeps share one loop (`sweep_admitted_communities`) for admission, skip, continue-on-tenant-error and stop-on-outage, instead of four copies of it. - The source policy treats that helper as a tenant-local chokepoint, so `backfill_d_tags` no longer needs a reviewed exception. - The per-tenant loops add code, because each community now gets its own admitted transaction. --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- crates/buzz-db/src/error.rs | 35 + crates/buzz-db/src/lib.rs | 6 +- crates/buzz-db/src/runtime/mod.rs | 54 ++ crates/buzz-db/src/runtime/push_enqueue.rs | 67 +- crates/buzz-db/src/runtime/tests.rs | 27 + .../src/store/admission_postgres_tests.rs | 597 ++++++++++++++++-- crates/buzz-db/src/store/channel.rs | 60 +- crates/buzz-db/src/store/event.rs | 39 +- crates/buzz-db/src/store/push.rs | 128 ++-- crates/buzz-db/src/store/relay_invite.rs | 55 +- crates/buzz-db/src/store/workflow.rs | 31 +- crates/buzz-db/src/test_support.rs | 148 ++++- crates/buzz-db/tests/observability_source.rs | 6 +- 13 files changed, 1108 insertions(+), 145 deletions(-) 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] = [(