From cf018e1bd97eaccafe21a41c8f722e0a96ede93e Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Date: Sun, 19 Jul 2026 18:36:59 -0400 Subject: [PATCH 1/2] relay: gate push enqueue on live leases; batch matcher pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two write-amplification fixes for the push pipeline (T1a repair, T1b, T2b of the write-amp plan): Migration 0023 (T1b push gate): skip the push_match_queue enqueue for communities with no active, endpoint-enabled, unexpired push lease. In lease-less communities (most of them) every durable message paid the full matcher cost — enqueue + claim + lease scan + delete — to conclude "notify no one". The gate lives in the events trigger so every durable producer is covered. The lost-wake race (event reads "no lease" while a lease activation commits concurrently) is closed with a per-community advisory lock: event inserts take it SHARED, lease transitions that can make eligibility true (accept_lease_event, replace_lease) take it EXCLUSIVE, plus a product-recovery backfill on activation. Migration 0024 (T1a repair): the 0022 TTL-refresh trigger took FOR UPDATE on the channel row before testing ttl_seconds, so every durable message in a permanent channel serialized at commit time on that tuple, each holding it across its WAL flush — one hot channel meant fully serialized commits (observed live: 0.07ms -> ~15ms commit latency at 200 QPS). The trigger now synchronizes on a shared per-channel advisory lock; TTL transitions in update_channel take the same key EXCLUSIVE, so the stale-prefetch proof from 0022 is preserved without serializing the permanent-channel hot path. T2b (matcher batching): the relay push matcher now claims per-community batches (single SKIP LOCKED CTE, cap 64) and loads events, leases, and channel membership pairs in one statement each; completes and retries are set-wise under the shared claim fence. One batch costs a constant number of statements regardless of size. The poison-job reap moves off the claim path to a 30s sweep (its scan is not served by the due partial index), and idle matcher/delivery workers back off 250ms -> 2s. The per-job claim API is deleted, not deprecated — no second code path. Tests: batch fence contract, single-community batch claim, T1a barrier (8 concurrent permanent-channel commits, zero channel tuple locks), T1b forced lost-wake ordering + activation backfill, exhausted-job sweep. Full buzz-db + buzz-relay suites green on a fresh 0001-0024 database (lone red: api::mesh_demo echo test, also red on main). Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-db/src/channel.rs | 56 +- crates/buzz-db/src/event.rs | 101 +++ crates/buzz-db/src/lib.rs | 48 +- crates/buzz-db/src/migration.rs | 31 +- crates/buzz-db/src/push.rs | 625 +++++++++++++++--- crates/buzz-relay/src/push_runtime.rs | 188 +++++- migrations/0023_push_match_gate.sql | 43 ++ .../0024_event_ttl_refresh_shared_lock.sql | 58 ++ schema/schema.sql | 67 +- scripts/attach-schema-partitions.sql | 8 + 10 files changed, 1092 insertions(+), 133 deletions(-) create mode 100644 migrations/0023_push_match_gate.sql create mode 100644 migrations/0024_event_ttl_refresh_shared_lock.sql diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 5c3bd2538c..2f20fc029f 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -538,6 +538,33 @@ pub async fn is_member( Ok(cnt > 0) } +/// Return which of the given (channel, pubkey) combinations are active +/// memberships, restricted to non-deleted channels — one statement for any +/// batch size (T2b). Semantics per pair match [`is_member`]. +pub async fn membership_pairs( + pool: &PgPool, + community_id: CommunityId, + channel_ids: &[Uuid], + pubkeys: &[Vec], +) -> Result)>> { + if channel_ids.is_empty() || pubkeys.is_empty() { + return Ok(Vec::new()); + } + let rows = sqlx::query( + "SELECT cm.channel_id, cm.pubkey FROM channel_members cm \ + JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \ + WHERE cm.community_id = $1 AND cm.channel_id = ANY($2) AND cm.pubkey = ANY($3) AND cm.removed_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(channel_ids) + .bind(pubkeys) + .fetch_all(pool) + .await?; + rows.into_iter() + .map(|row| Ok((row.try_get("channel_id")?, row.try_get("pubkey")?))) + .collect() +} + /// Returns all active members of the given channel. /// /// Returns an empty list if the channel has been soft-deleted. @@ -1076,9 +1103,32 @@ pub async fn update_channel( q = q.bind(community_id.as_uuid()); q = q.bind(channel_id); - let result = q.execute(pool).await?; - if result.rows_affected() == 0 { - return Err(DbError::ChannelNotFound(channel_id)); + // T1a repair: a TTL change can flip this channel's event-trigger fast + // path (migration 0024 reads ttl_seconds under a SHARED per-channel + // advisory lock). Take the same key EXCLUSIVE before the UPDATE so a + // concurrent event either sees the committed TTL or strictly precedes + // this transition — whose own deadline reset is then the latest word. + // Non-TTL updates don't touch the fast path and skip the lock. + if updates.ttl_seconds.is_some() { + let mut tx = pool.begin().await?; + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(format!( + "buzz_channel_ttl:{}:{}", + community_id.as_uuid(), + channel_id + )) + .execute(&mut *tx) + .await?; + let result = q.execute(&mut *tx).await?; + if result.rows_affected() == 0 { + return Err(DbError::ChannelNotFound(channel_id)); + } + tx.commit().await?; + } else { + let result = q.execute(pool).await?; + if result.rows_affected() == 0 { + return Err(DbError::ChannelNotFound(channel_id)); + } } get_channel(pool, community_id, channel_id).await diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 352a1fe404..71f3119943 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -1561,6 +1561,14 @@ mod tests { assert_eq!(stale_ttl, None); let mut activation = pool.begin().await.expect("begin TTL activation"); + // Model the repaired update_channel protocol (migration 0024): the + // TTL transition holds the per-channel advisory key EXCLUSIVE, which + // is what the event trigger's shared acquisition now waits on. + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(format!("buzz_channel_ttl:{community_uuid}:{racing}")) + .execute(&mut *activation) + .await + .expect("acquire exclusive channel TTL key"); let activation_deadline: DateTime = sqlx::query_scalar( "UPDATE channels \ SET ttl_seconds = 60, ttl_deadline = clock_timestamp() + interval '60 seconds' \ @@ -1605,6 +1613,99 @@ mod tests { ); } + /// T1a repair regression test (migration 0024): permanent-channel event + /// commits must not serialize on the channel row. The 0022 trigger took + /// `FOR UPDATE` on the channel tuple before testing `ttl_seconds`, so + /// concurrent commits into one hot permanent channel queued at commit + /// time (deferred trigger) — invisible to any single-connection test. + /// This holds N insert transactions at a barrier past their INSERTs, + /// then proves (a) while all N sit pre-commit, no transaction holds a + /// row-level lock on the channel tuple, and (b) all N commits succeed + /// with the channel row untouched (permanent ⇒ no deadline write). + #[tokio::test] + #[ignore = "requires Postgres"] + async fn permanent_channel_event_commits_do_not_lock_the_channel_row() { + const N: usize = 8; + // setup_pool's default cap (10) covers N held transactions plus the + // pg_locks inspector connection. + let pool = setup_pool().await; + let community_uuid = make_test_community(&pool).await; + let channel = make_test_channel(&pool, community_uuid, None).await; + + // Open N transactions, run the full event INSERT in each (the deferred + // trigger fires at COMMIT), and park them at a barrier. + let mut txs = Vec::new(); + for i in 0..N { + let mut tx = pool.begin().await.expect("begin insert txn"); + let event = make_text_event(&format!("hot channel event {i}")); + sqlx::query( + "INSERT INTO events (community_id,id,pubkey,created_at,kind,tags,content,sig,received_at,channel_id) \ + VALUES ($1,$2,$3,$4,9,$5,$6,$7,now(),$8)", + ) + .bind(community_uuid) + .bind(event.id.as_bytes().as_slice()) + .bind(event.pubkey.as_bytes().as_slice()) + .bind(DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).unwrap()) + .bind(serde_json::to_value(&event.tags).unwrap()) + .bind(&event.content) + .bind(event.sig.serialize().as_slice()) + .bind(channel) + .execute(&mut *tx) + .await + .expect("insert event inside held txn"); + txs.push(tx); + } + + // With all N transactions holding completed INSERTs, none may hold a + // row-level lock on the channels tuple. (The 0022 trigger would not + // have taken it yet either — it locks at COMMIT — so also verify the + // commit phase below completes without mutual blocking.) + let tuple_locks: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_locks l \ + JOIN pg_class c ON c.oid = l.relation \ + WHERE c.relname = 'channels' AND l.locktype = 'tuple'", + ) + .fetch_one(&pool) + .await + .expect("inspect pg_locks"); + assert_eq!(tuple_locks, 0, "no channel tuple locks while txns are held"); + + // Release all commits concurrently. Under 0022 these serialized on the + // channel row (each holding it across its WAL flush); under 0024 the + // shared advisory key admits them all. Join with a timeout so a + // regression fails fast instead of hanging the suite. + let commits = txs + .into_iter() + .map(|tx| tokio::spawn(async move { tx.commit().await })) + .collect::>(); + for c in commits { + tokio::time::timeout(std::time::Duration::from_secs(10), c) + .await + .expect("concurrent permanent-channel commits must not block") + .expect("join commit task") + .expect("commit succeeds"); + } + + let deadline: Option> = sqlx::query_scalar( + "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2", + ) + .bind(community_uuid) + .bind(channel) + .fetch_one(&pool) + .await + .expect("read deadline after commits"); + assert_eq!(deadline, None, "permanent channel must remain untouched"); + let stored: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events WHERE community_id = $1 AND channel_id = $2", + ) + .bind(community_uuid) + .bind(channel) + .fetch_one(&pool) + .await + .expect("count stored events"); + assert_eq!(stored as usize, N); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn get_event_by_id_is_scoped_when_event_id_collides_across_communities() { diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 552150e132..c059e5c7e3 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1220,12 +1220,13 @@ impl Db { event::get_events_by_ids(&self.pool, community_id, ids).await } - /// Exclusively claim the next due event-to-push matcher job. - pub async fn claim_due_push_match( + /// Exclusively claim a batch of due matcher jobs from one community. + pub async fn claim_due_push_match_batch( &self, + limit: i64, lease_until: DateTime, - ) -> Result> { - push::claim_due_match(&self.pool, lease_until).await + ) -> Result> { + push::claim_due_match_batch(&self.pool, limit, lease_until).await } /// Load active endpoint-enabled leases eligible for push matching. @@ -1236,18 +1237,30 @@ impl Db { push::active_match_leases(&self.pool, community).await } - /// Complete a matcher job if its claim fence is still held. - pub async fn complete_push_match(&self, job: &push::ClaimedMatch) -> Result { - push::complete_match(&self.pool, job).await + /// Complete matcher jobs from one claimed batch while the fence holds. + pub async fn complete_push_match_batch( + &self, + community: CommunityId, + claim_id: uuid::Uuid, + event_ids: &[Vec], + ) -> Result { + push::complete_match_batch(&self.pool, community, claim_id, event_ids).await } - /// Release a matcher claim for retry at the supplied time. - pub async fn retry_push_match( + /// Release fenced matcher claims from one batch for retry. + pub async fn retry_push_match_batch( &self, - job: &push::ClaimedMatch, + community: CommunityId, + claim_id: uuid::Uuid, + event_ids: &[Vec], next: DateTime, - ) -> Result { - push::retry_match(&self.pool, job, next).await + ) -> Result { + push::retry_match_batch(&self.pool, community, claim_id, event_ids, next).await + } + + /// Delete exhausted matcher jobs (periodic sweep, off the claim path). + pub async fn reap_exhausted_push_matches(&self) -> Result { + push::reap_exhausted_matches(&self.pool).await } /// Idempotently enqueue a wake for a matched lease and event. @@ -1534,6 +1547,17 @@ impl Db { channel::is_member(&self.pool, community_id, channel_id, pubkey).await } + /// Return the active (channel, pubkey) membership pairs among the given + /// sets, in one statement. + pub async fn membership_pairs( + &self, + community_id: CommunityId, + channel_ids: &[Uuid], + pubkeys: &[Vec], + ) -> Result)>> { + channel::membership_pairs(&self.pool, community_id, channel_ids, pubkeys).await + } + /// Returns all active members of a channel. pub async fn get_members( &self, diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index dce6ca6ad4..1674b0ec4d 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -560,7 +560,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 22); + assert_eq!(migrations.len(), 24); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -852,6 +852,33 @@ mod tests { assert!(ttl_refresh.contains("DEFERRABLE INITIALLY DEFERRED")); assert!(ttl_refresh.contains("clock_timestamp()")); assert!(ttl_refresh.contains("NEW.kind <> 9007")); + + // T1b push gate: the match-queue trigger only enqueues when the + // community has an eligible lease, ordered against lease activations + // through the shared/exclusive per-community advisory lock. + assert_eq!(migrations[22].version, 23); + let push_gate = migrations[22].sql.as_str(); + assert!(push_gate.contains("CREATE OR REPLACE FUNCTION enqueue_push_match_job")); + assert!(push_gate.contains("pg_advisory_xact_lock_shared")); + assert!(push_gate.contains("'buzz_push_gate:' || NEW.community_id::text")); + assert!(push_gate.contains("endpoint_enabled")); + + // T1a repair: the TTL refresh trigger synchronizes on a shared + // per-channel advisory lock instead of FOR UPDATE on the channel row, + // so permanent-channel commits no longer serialize. + assert_eq!(migrations[23].version, 24); + let ttl_shared = migrations[23].sql.as_str(); + assert!(ttl_shared + .contains("CREATE OR REPLACE FUNCTION refresh_channel_ttl_after_event_insert")); + assert!(ttl_shared.contains("pg_advisory_xact_lock_shared")); + assert!(ttl_shared.contains("'buzz_channel_ttl:' || NEW.community_id::text")); + // The row read must be a bare SELECT (comments describe the removed + // FOR UPDATE; the executable body must not reintroduce it). + assert!(ttl_shared.contains("SELECT ttl_seconds INTO channel_ttl")); + assert!(!strip_sql_comments(ttl_shared) + .to_lowercase() + .contains("for update")); + assert!(ttl_shared.contains("NEW.kind <> 9007")); } #[test] @@ -1094,7 +1121,7 @@ mod tests { run_migrations(&pool) .await .expect("retry succeeds after operator repair"); - assert_eq!(applied_versions(&pool).await.last().copied(), Some(22)); + assert_eq!(applied_versions(&pool).await.last().copied(), Some(24)); } #[tokio::test] diff --git a/crates/buzz-db/src/push.rs b/crates/buzz-db/src/push.rs index dc5920e1e9..b0042ec92f 100644 --- a/crates/buzz-db/src/push.rs +++ b/crates/buzz-db/src/push.rs @@ -12,6 +12,60 @@ use uuid::Uuid; use crate::error::Result; +/// Namespace for the per-community push-gate advisory lock. Must match the +/// key built inside the `enqueue_push_match_job` trigger (migration 0023): +/// event inserts take it SHARED there; every lease transition that can make +/// match eligibility true takes it EXCLUSIVE here, forcing a total order so +/// a concurrent event insert either sees the committed lease or strictly +/// precedes the activation (in which case no wake was owed). Distinct key +/// domain from the audit lock and the lease address/author locks. +const PUSH_GATE_LOCK_NAMESPACE: &str = "buzz_push_gate:"; + +async fn acquire_push_gate_lock( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community: CommunityId, +) -> Result<()> { + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(format!("{PUSH_GATE_LOCK_NAMESPACE}{}", community.as_uuid())) + .execute(&mut **tx) + .await?; + Ok(()) +} + +/// Recovery window for the activation backfill: recent events that the gate +/// legitimately skipped (no eligible lease at their commit time) are enqueued +/// when a lease activates, so a user who registers moments after a message +/// still gets woken. Product coverage only — the advisory-lock total order is +/// what makes the gate correct; see `PUSH_GATE_LOCK_NAMESPACE`. +const PUSH_GATE_BACKFILL_SECS: i64 = 120; + +/// Enqueue match jobs for recent gate-skipped events. MUST run inside the same +/// transaction that holds the exclusive push-gate lock: after this commit, +/// every event is either backfilled here or ordered after the activation and +/// enqueued by the trigger — running it post-commit would reopen the gap. +/// Keyed on relay `received_at` (not author-controlled `created_at`); the kind +/// list mirrors the trigger allowlist; `ON CONFLICT DO NOTHING` dedups against +/// rows the trigger already enqueued. +async fn backfill_push_match_jobs( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community: CommunityId, +) -> Result<()> { + sqlx::query( + "INSERT INTO push_match_queue (community_id, event_id) \ + SELECT community_id, id FROM events \ + WHERE community_id = $1 \ + AND kind IN (7, 9, 1059, 40007, 46010) \ + AND deleted_at IS NULL \ + AND received_at > now() - make_interval(secs => $2) \ + ON CONFLICT DO NOTHING", + ) + .bind(community.as_uuid()) + .bind(PUSH_GATE_BACKFILL_SECS as f64) + .execute(&mut **tx) + .await?; + Ok(()) +} + /// Maximum claims for a malformed matcher job before it is discarded. pub const MAX_MATCH_ATTEMPTS: i32 = 8; @@ -116,19 +170,6 @@ pub enum RevalidateWakeOutcome { Suppressed, } -/// One durably accepted event claimed for push matching. -#[derive(Debug, Clone)] -pub struct ClaimedMatch { - /// Tenant that owns both the event and matcher job. - pub community: CommunityId, - /// Non-deleted source event loaded after the claim commits. - pub event: buzz_core::StoredEvent, - /// Fencing token required to complete or retry this claim. - pub claim_id: Uuid, - /// Attempt number, starting at one for the first claim. - pub attempt: i32, -} - /// Current active lease candidate for matcher evaluation. #[derive(Debug, Clone)] pub struct MatchLease { @@ -197,6 +238,12 @@ pub async fn accept_lease_event( .bind(author_lock) .execute(&mut *tx) .await?; + // T1b: an activation can flip the community from "no eligible lease" to + // "eligible", so it must serialize against the trigger's shared gate lock. + // Acquired after the address/author locks to keep one global lock order. + if active.is_some() { + acquire_push_gate_lock(&mut tx, community).await?; + } if let Some(row) = sqlx::query( "SELECT author, installation_id FROM push_leases WHERE community_id=$1 AND source_event_id=$2", @@ -340,6 +387,9 @@ pub async fn accept_lease_event( } return Err(error.into()); } + if is_active { + backfill_push_match_jobs(&mut tx, community).await?; + } tx.commit().await?; Ok(AcceptLeaseOutcome::Accepted) } @@ -417,6 +467,15 @@ async fn replace_lease( None => (false, None, None, None, None, None), }; + // T1b: an activating replacement can flip the community from "no eligible + // lease" to "eligible"; serialize it against the trigger's shared gate + // lock (gate → lease row, matching accept_lease_event's global order). + // Revocations (is_active = false) never make eligibility true and skip it. + let mut tx = pool.begin().await?; + if is_active { + acquire_push_gate_lock(&mut tx, community).await?; + } + // The conflict predicate is the acceptance state machine. Keeping both // orderings in the upsert closes the missing-row race: concurrent initial // publications cannot bypass a preceding SELECT/row lock. @@ -464,10 +523,14 @@ async fn replace_lease( .bind(max_class) .bind(subscriptions) .bind(version.expires_at) - .fetch_optional(pool) + .fetch_optional(&mut *tx) .await?; if accepted.is_some() { + if is_active { + backfill_push_match_jobs(&mut tx, community).await?; + } + tx.commit().await?; return Ok(ReplaceLeaseOutcome::Accepted); } @@ -479,8 +542,9 @@ async fn replace_lease( .bind(community.as_uuid()) .bind(author) .bind(installation_id) - .fetch_one(pool) + .fetch_one(&mut *tx) .await?; + tx.commit().await?; let current_created_at: i64 = current.try_get("source_created_at")?; let current_event_id: Vec = current.try_get("source_event_id")?; let wins_event_order = version.source_created_at > current_created_at @@ -575,57 +639,88 @@ pub async fn enqueue_wake( Ok(outcome) } -/// Exclusively claim the next due matcher job and load its non-deleted event. -pub async fn claim_due_match( +/// One batch of matcher jobs claimed from a single community. +#[derive(Debug, Clone)] +pub struct ClaimedMatchBatch { + /// Tenant that owns every job in this batch. + pub community: CommunityId, + /// Fencing token shared by the whole batch. + pub claim_id: Uuid, + /// Claimed jobs whose non-deleted source events loaded successfully. + pub jobs: Vec, +} + +/// One claimed job inside a [`ClaimedMatchBatch`]. +#[derive(Debug, Clone)] +pub struct BatchedMatch { + /// Non-deleted source event loaded after the claim commits. + pub event: buzz_core::StoredEvent, + /// Attempt number, starting at one for the first claim. + pub attempt: i32, +} + +/// Exclusively claim up to `limit` due matcher jobs from ONE community and +/// load their non-deleted events in a single query (T2b). +/// +/// The batch is community-scoped so downstream lease and membership loads are +/// each one statement. Jobs whose source event is absent or soft-deleted are +/// completed inside this call (privacy-preserving terminal outcome, identical +/// to the previous per-job path). Poison reaping is NOT performed here — it +/// lives in [`reap_exhausted_matches`], off the claim path, because the reap +/// DELETE rescans the pending set and under backlog made every claim slower. +pub async fn claim_due_match_batch( pool: &PgPool, + limit: i64, lease_until: DateTime, -) -> Result> { - claim_due_match_with_loader(pool, lease_until, |pool, community, event_id| async move { - Ok( - crate::event::get_events_by_ids(&pool, community, &[&event_id]) - .await? - .into_iter() - .next(), - ) - }) +) -> Result> { + claim_due_match_batch_with_loader( + pool, + limit, + lease_until, + |pool, community, ids| async move { + let refs: Vec<&[u8]> = ids.iter().map(Vec::as_slice).collect(); + crate::event::get_events_by_ids(&pool, community, &refs).await + }, + ) .await } -async fn claim_due_match_with_loader( +async fn claim_due_match_batch_with_loader( pool: &PgPool, + limit: i64, lease_until: DateTime, load: F, -) -> Result> +) -> Result> where - F: FnOnce(PgPool, CommunityId, Vec) -> Fut, - Fut: std::future::Future>>, + F: FnOnce(PgPool, CommunityId, Vec>) -> Fut, + Fut: std::future::Future>>, { let claim_id = Uuid::new_v4(); - let mut tx = pool.begin().await?; - // Reap poison jobs before claiming so a worker crash on the final attempt - // cannot leave an unclaimable row pinning outbox retention forever. - sqlx::query( - "DELETE FROM push_match_queue WHERE attempts >= $1 \ - AND (state='pending' OR (state='matching' AND lease_until < now()))", - ) - .bind(MAX_MATCH_ATTEMPTS) - .execute(&mut *tx) - .await?; - let row = sqlx::query( + let rows = sqlx::query( r#" - WITH candidate AS ( - SELECT community_id, event_id + 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())) ORDER BY next_attempt_at, created_at - FOR UPDATE SKIP LOCKED LIMIT 1 + ), + 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 + 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 + FOR UPDATE OF q SKIP LOCKED + LIMIT $4 ) UPDATE push_match_queue q - SET state='matching', claim_id=$1, lease_until=$2, attempts=attempts+1 - FROM candidate c + SET state='matching', claim_id=$1, lease_until=$2, attempts=q.attempts+1 + FROM candidates c WHERE q.community_id=c.community_id AND q.event_id=c.event_id RETURNING q.community_id, q.event_id, q.attempts "#, @@ -633,40 +728,71 @@ where .bind(claim_id) .bind(lease_until) .bind(MAX_MATCH_ATTEMPTS) - .fetch_optional(&mut *tx) + .bind(limit) + .fetch_all(pool) .await?; - let Some(row) = row else { - tx.commit().await?; + if rows.is_empty() { return Ok(None); - }; - let community = CommunityId::from_uuid(row.try_get("community_id")?); - let event_id: Vec = row.try_get("event_id")?; - let attempt: i32 = row.try_get("attempts")?; - tx.commit().await?; - let event = load(pool.clone(), community, event_id.clone()).await?; - let Some(event) = event else { - // Source absence and soft deletion are deliberate privacy-preserving - // terminal outcomes. Query errors above propagate instead, leaving the - // fenced job recoverable after its claim lease expires. + } + let community = CommunityId::from_uuid(rows[0].try_get("community_id")?); + let mut attempts = std::collections::HashMap::with_capacity(rows.len()); + for row in &rows { + let event_id: Vec = row.try_get("event_id")?; + let attempt: i32 = row.try_get("attempts")?; + attempts.insert(event_id, attempt); + } + let ids: Vec> = attempts.keys().cloned().collect(); + let events = load(pool.clone(), community, ids).await?; + let mut jobs = Vec::with_capacity(events.len()); + for event in events { + let attempt = attempts + .remove(event.event.id.as_bytes().as_slice()) + .unwrap_or(1); + jobs.push(BatchedMatch { event, attempt }); + } + // Whatever is left in `attempts` had no loadable source event: absence and + // soft deletion are deliberate privacy-preserving terminal outcomes. + // Query errors above propagate instead, leaving the fenced jobs + // recoverable after their claim lease expires. + let gone: Vec> = attempts.into_keys().collect(); + if !gone.is_empty() { sqlx::query( "DELETE FROM push_match_queue \ - WHERE community_id=$1 AND event_id=$2 AND claim_id=$3 AND state='matching'", + WHERE community_id=$1 AND claim_id=$2 AND state='matching' AND event_id = ANY($3)", ) .bind(community.as_uuid()) - .bind(&event_id) .bind(claim_id) + .bind(&gone) .execute(pool) .await?; + } + if jobs.is_empty() { return Ok(None); - }; - Ok(Some(ClaimedMatch { + } + Ok(Some(ClaimedMatchBatch { community, - event, claim_id, - attempt, + jobs, })) } +/// Delete exhausted matcher jobs so a worker crash on the final attempt +/// cannot leave an unclaimable row pinning outbox retention forever. +/// +/// Runs on a periodic sweep, never inside the claim path: the scan is not +/// served by the due partial index, so putting it in every claim made claims +/// slower exactly when a backlog needed them fastest. +pub async fn reap_exhausted_matches(pool: &PgPool) -> Result { + Ok(sqlx::query( + "DELETE FROM push_match_queue WHERE attempts >= $1 \ + AND (state='pending' OR (state='matching' AND lease_until < now()))", + ) + .bind(MAX_MATCH_ATTEMPTS) + .execute(pool) + .await? + .rows_affected()) +} + /// Load active endpoint-enabled leases for one tenant. pub async fn active_match_leases(pool: &PgPool, community: CommunityId) -> Result> { let rows = sqlx::query( @@ -690,22 +816,54 @@ pub async fn active_match_leases(pool: &PgPool, community: CommunityId) -> Resul .collect() } -/// Delete a matcher job only while its claim fence is held. -pub async fn complete_match(pool: &PgPool, match_job: &ClaimedMatch) -> Result { - Ok(sqlx::query("DELETE FROM push_match_queue WHERE community_id=$1 AND event_id=$2 AND claim_id=$3 AND state='matching'") - .bind(match_job.community.as_uuid()).bind(match_job.event.event.id.as_bytes().as_slice()) - .bind(match_job.claim_id).execute(pool).await?.rows_affected() == 1) +/// Delete matcher jobs from one claimed batch while its fence is held (one +/// statement for any number of jobs). Returns how many rows were completed; +/// jobs whose fence was lost are left for their next claimant. +pub async fn complete_match_batch( + pool: &PgPool, + community: CommunityId, + claim_id: Uuid, + event_ids: &[Vec], +) -> Result { + if event_ids.is_empty() { + return Ok(0); + } + Ok(sqlx::query( + "DELETE FROM push_match_queue \ + WHERE community_id=$1 AND claim_id=$2 AND state='matching' AND event_id = ANY($3)", + ) + .bind(community.as_uuid()) + .bind(claim_id) + .bind(event_ids) + .execute(pool) + .await? + .rows_affected()) } -/// Release a fenced matcher claim for retry at the supplied time. -pub async fn retry_match( +/// Release fenced matcher claims from one batch for retry at the supplied +/// time (one statement for any number of jobs). +pub async fn retry_match_batch( pool: &PgPool, - match_job: &ClaimedMatch, + community: CommunityId, + claim_id: Uuid, + event_ids: &[Vec], next: DateTime, -) -> Result { - Ok(sqlx::query("UPDATE push_match_queue SET state='pending', claim_id=NULL, lease_until=NULL, next_attempt_at=$4 WHERE community_id=$1 AND event_id=$2 AND claim_id=$3 AND state='matching'") - .bind(match_job.community.as_uuid()).bind(match_job.event.event.id.as_bytes().as_slice()) - .bind(match_job.claim_id).bind(next).execute(pool).await?.rows_affected() == 1) +) -> Result { + if event_ids.is_empty() { + return Ok(0); + } + Ok(sqlx::query( + "UPDATE push_match_queue \ + SET state='pending', claim_id=NULL, lease_until=NULL, next_attempt_at=$4 \ + WHERE community_id=$1 AND claim_id=$2 AND state='matching' AND event_id = ANY($3)", + ) + .bind(community.as_uuid()) + .bind(claim_id) + .bind(event_ids) + .bind(next) + .execute(pool) + .await? + .rows_affected()) } /// Claim due jobs for one community, recovering expired worker leases. @@ -1467,7 +1625,16 @@ mod tests { #[ignore = "requires Postgres"] async fn matcher_trigger_is_allowlisted_and_deleted_events_are_discarded() { let pool = setup_pool().await; + // Global claim assertions below require a queue free of other tests' + // leftovers. + sqlx::query("DELETE FROM push_match_queue") + .execute(&pool) + .await + .expect("drain matcher queue"); let community = make_community(&pool).await; + // The T1b gate only enqueues match jobs for communities with an + // eligible lease; give this one an active lease first. + activate(&pool, community, &[77; 32], "install", &[78; 32], 1).await; let keys = nostr::Keys::generate(); let push_event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "push") .sign_with_keys(&keys) @@ -1500,7 +1667,7 @@ mod tests { .await .expect("soft delete before matching"); assert!( - claim_due_match(&pool, Utc::now() + chrono::Duration::minutes(1)) + claim_due_match_batch(&pool, 16, Utc::now() + chrono::Duration::minutes(1)) .await .expect("claim deleted event") .is_none() @@ -1519,6 +1686,8 @@ mod tests { async fn matcher_load_error_preserves_claimed_job_for_recovery() { let pool = setup_pool().await; let community = make_community(&pool).await; + // Eligible lease required for the T1b-gated trigger to enqueue. + activate(&pool, community, &[79; 32], "install", &[80; 32], 1).await; let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "retry me") .sign_with_keys(&nostr::Keys::generate()) .expect("sign event"); @@ -1526,10 +1695,11 @@ mod tests { .await .expect("insert event"); - let error = claim_due_match_with_loader( + let error = claim_due_match_batch_with_loader( &pool, + 16, Utc::now() - chrono::Duration::seconds(1), - |_pool, _community, _event_id| async { + |_pool, _community, _event_ids| async { Err(crate::DbError::InvalidData("injected load failure".into())) }, ) @@ -1546,7 +1716,7 @@ mod tests { .expect("load failure must preserve matcher row"); assert_eq!(row, ("matching".to_string(), 1)); assert!( - claim_due_match(&pool, Utc::now() + chrono::Duration::minutes(1)) + claim_due_match_batch(&pool, 16, Utc::now() + chrono::Duration::minutes(1)) .await .expect("expired claim remains recoverable") .is_some() @@ -1557,7 +1727,16 @@ mod tests { #[ignore = "requires Postgres"] async fn matcher_claim_is_exclusive_across_workers() { let pool = setup_pool().await; + // Drain leftovers from other tests sharing this database: a racing + // worker claiming an unrelated community's stale batch would count as + // a second success. + sqlx::query("DELETE FROM push_match_queue") + .execute(&pool) + .await + .expect("drain matcher queue"); let community = make_community(&pool).await; + // Eligible lease required for the T1b-gated trigger to enqueue. + activate(&pool, community, &[81; 32], "install", &[82; 32], 1).await; let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "one job") .sign_with_keys(&nostr::Keys::generate()) .expect("sign event"); @@ -1571,7 +1750,7 @@ mod tests { let barrier = Arc::clone(&barrier); tasks.push(tokio::spawn(async move { barrier.wait().await; - claim_due_match(&pool, Utc::now() + chrono::Duration::minutes(1)) + claim_due_match_batch(&pool, 16, Utc::now() + chrono::Duration::minutes(1)) .await .expect("claim matcher job") })); @@ -1623,6 +1802,12 @@ mod tests { #[ignore = "requires Postgres"] async fn exhausted_match_job_is_reaped_and_cannot_pin_retention() { let pool = setup_pool().await; + // Global claim assertions below require a queue free of other tests' + // leftovers. + sqlx::query("DELETE FROM push_match_queue") + .execute(&pool) + .await + .expect("drain matcher queue"); let community = make_community(&pool).await; let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "poison") .sign_with_keys(&nostr::Keys::generate()) @@ -1674,12 +1859,20 @@ mod tests { .execute(&pool) .await .expect("exhaust matcher job"); + // The reap lives off the claim path now (periodic sweep): a claim must + // skip the exhausted row, and the sweep must delete it. assert!( - claim_due_match(&pool, Utc::now() + chrono::Duration::minutes(1)) + claim_due_match_batch(&pool, 16, Utc::now() + chrono::Duration::minutes(1)) .await - .expect("reap exhausted matcher") + .expect("claim skips exhausted matcher") .is_none() ); + assert_eq!( + reap_exhausted_matches(&pool) + .await + .expect("reap exhausted matcher"), + 1 + ); let remaining: i64 = sqlx::query_scalar( "SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2", ) @@ -1695,4 +1888,268 @@ mod tests { "reaped poison job must release delivered-wake retention" ); } + + /// T2b batch contract: one claim returns jobs from exactly ONE community + /// (so downstream lease/membership loads are single statements), the + /// set-wise complete and retry honor the claim fence, and a retried job + /// becomes claimable again. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn batch_claim_is_single_community_and_setwise_ops_honor_the_fence() { + let pool = setup_pool().await; + // The batch claim targets the globally oldest due job, so leftover + // queue rows from other tests sharing this database would hijack the + // target community. Start from a drained queue. + sqlx::query("DELETE FROM push_match_queue") + .execute(&pool) + .await + .expect("drain matcher queue"); + let community_a = make_community(&pool).await; + let community_b = make_community(&pool).await; + activate(&pool, community_a, &[83; 32], "install", &[84; 32], 1).await; + activate(&pool, community_b, &[85; 32], "install", &[86; 32], 1).await; + let mut a_ids = Vec::new(); + for i in 0..3 { + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), format!("a{i}")) + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign event"); + crate::event::insert_event(&pool, community_a, &event, None) + .await + .expect("insert community-a event"); + a_ids.push(event.id.as_bytes().to_vec()); + } + let b_event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "b0") + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign event"); + crate::event::insert_event(&pool, community_b, &b_event, None) + .await + .expect("insert community-b event"); + + let batch = claim_due_match_batch(&pool, 16, Utc::now() + chrono::Duration::minutes(1)) + .await + .expect("claim first batch") + .expect("batch present"); + assert_eq!( + batch.jobs.len(), + 3, + "batch must take ALL due jobs from one community" + ); + assert_eq!(batch.community, community_a, "oldest community first"); + assert!(batch.jobs.iter().all(|job| job.attempt == 1)); + + let claimed_ids: Vec> = batch + .jobs + .iter() + .map(|job| job.event.event.id.as_bytes().to_vec()) + .collect(); + // A stale fence must not complete or retry anything. + assert_eq!( + complete_match_batch(&pool, batch.community, Uuid::new_v4(), &claimed_ids) + .await + .expect("stale complete"), + 0 + ); + assert_eq!( + retry_match_batch( + &pool, + batch.community, + Uuid::new_v4(), + &claimed_ids, + Utc::now() + ) + .await + .expect("stale retry"), + 0 + ); + // Complete two under the real fence, retry the third immediately. + assert_eq!( + complete_match_batch(&pool, batch.community, batch.claim_id, &claimed_ids[..2]) + .await + .expect("complete two"), + 2 + ); + assert_eq!( + retry_match_batch( + &pool, + batch.community, + batch.claim_id, + &claimed_ids[2..], + Utc::now() + ) + .await + .expect("retry one"), + 1 + ); + + // The retried job is claimable again, but its retry time is later + // than community B's untouched row, so B's batch comes first. + let second = claim_due_match_batch(&pool, 16, Utc::now() + chrono::Duration::minutes(1)) + .await + .expect("claim community-b batch") + .expect("community-b batch present"); + assert_eq!(second.community, community_b); + assert_eq!(second.jobs.len(), 1); + assert_eq!( + complete_match_batch( + &pool, + second.community, + second.claim_id, + &[second.jobs[0].event.event.id.as_bytes().to_vec()] + ) + .await + .expect("complete community-b job"), + 1 + ); + + // Then the retried community-a job, on its second attempt. + let third = claim_due_match_batch(&pool, 16, Utc::now() + chrono::Duration::minutes(1)) + .await + .expect("claim retried job") + .expect("retried job present"); + assert_eq!(third.community, community_a); + assert_eq!(third.jobs.len(), 1); + assert_eq!(third.jobs[0].attempt, 2); + assert_eq!( + complete_match_batch( + &pool, + third.community, + third.claim_id, + &[third.jobs[0].event.event.id.as_bytes().to_vec()] + ) + .await + .expect("complete retried job"), + 1 + ); + } + + /// T1b lost-wake race, forced (migration 0023): with no eligible lease the + /// events trigger must skip the match enqueue, and a lease activation + /// racing an in-flight event insert must be ordered AFTER it by the gate + /// lock — so the activation's backfill enqueues the event the trigger + /// skipped, exactly once. Without the shared/exclusive advisory pair, the + /// trigger could read "no lease" while the activation commits + /// concurrently, dropping that wake with no retry. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn gate_orders_lease_activation_after_in_flight_event_and_backfills_it() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + + // Phase 0: no lease anywhere in this community — a committed gated-kind + // event must not be enqueued. + let skipped = nostr::EventBuilder::new(nostr::Kind::Custom(9), "gate skips me") + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign skipped event"); + crate::event::insert_event(&pool, community, &skipped, None) + .await + .expect("insert lease-less event"); + let queued: i64 = + sqlx::query_scalar("SELECT count(*) FROM push_match_queue WHERE community_id=$1") + .bind(community.as_uuid()) + .fetch_one(&pool) + .await + .expect("count queue after lease-less insert"); + assert_eq!(queued, 0, "gate must skip enqueue with no eligible lease"); + + // Phase 1: hold an event-insert transaction open past its INSERT. The + // (non-deferred) trigger has already run: shared gate lock held, EXISTS + // saw no lease, enqueue skipped — the classic lost-wake window. + let raced = nostr::EventBuilder::new(nostr::Kind::Custom(9), "raced wake") + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign raced event"); + let mut insert_tx = pool.begin().await.expect("begin raced insert"); + sqlx::query( + "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at) \ + VALUES ($1, $2, $3, now(), 9, '[]', 'raced wake', $4, now())", + ) + .bind(community.as_uuid()) + .bind(raced.id.as_bytes().as_slice()) + .bind(raced.pubkey.as_bytes().as_slice()) + .bind(raced.sig.serialize().as_slice()) + .execute(&mut *insert_tx) + .await + .expect("insert raced event inside held txn"); + + // A concurrent activation must block on the exclusive gate lock until + // the insert transaction resolves. + let activation = { + let pool = pool.clone(); + tokio::spawn(async move { + replace_active_lease( + &pool, + community, + &[91; 32], + "install", + version(1, 10, 1), + ActiveLease { + app_profile: "ios-production", + endpoint_hash: &[92; 32], + endpoint_grant: "opaque-grant", + max_class: "default", + subscriptions: &serde_json::json!([]), + }, + ) + .await + .expect("activate racing lease") + }) + }; + + // Wait until the activation is provably parked on the advisory lock + // (not merely unscheduled), then confirm it has not completed. + let mut parked = false; + for _ in 0..100 { + let waiting: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted", + ) + .fetch_one(&pool) + .await + .expect("inspect advisory waiters"); + if waiting > 0 { + parked = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + assert!(parked, "activation must block on the exclusive gate lock"); + assert!(!activation.is_finished()); + + // Release the event; the activation acquires the gate, and its + // backfill must enqueue the event the trigger skipped — exactly once. + insert_tx.commit().await.expect("commit raced insert"); + assert_eq!( + tokio::time::timeout(std::time::Duration::from_secs(10), activation) + .await + .expect("activation completes once the gate is free") + .expect("join activation"), + ReplaceLeaseOutcome::Accepted + ); + let backfilled: Vec> = sqlx::query_scalar( + "SELECT event_id FROM push_match_queue WHERE community_id=$1 ORDER BY created_at", + ) + .bind(community.as_uuid()) + .fetch_all(&pool) + .await + .expect("read backfilled queue"); + assert!( + backfilled.contains(&raced.id.as_bytes().to_vec()), + "raced event must be recovered by the activation backfill" + ); + + // Phase 2: with the lease now active, the trigger enqueues directly and + // the backfill's ON CONFLICT dedup keeps it single. + let direct = nostr::EventBuilder::new(nostr::Kind::Custom(9), "direct enqueue") + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign direct event"); + crate::event::insert_event(&pool, community, &direct, None) + .await + .expect("insert post-activation event"); + let per_event: Vec = sqlx::query_scalar( + "SELECT count(*) FROM push_match_queue WHERE community_id=$1 GROUP BY event_id", + ) + .bind(community.as_uuid()) + .fetch_all(&pool) + .await + .expect("count queue rows per event"); + assert!(per_event.iter().all(|count| *count == 1)); + } } diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index 1c5b9c26fa..993db21805 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -15,6 +15,18 @@ use crate::{handlers::push_lease::Subscription, state::AppState}; const CLAIM_SECS: i64 = 30; const EVENT_USEFUL_SECS: i64 = 3600; const MAX_ATTEMPTS: i32 = 8; +/// Upper bound on one claimed matcher batch. Bounded well under +/// `get_events_by_ids`' 500-id batch-fetch contract. +const MATCH_BATCH_LIMIT: i64 = 64; +/// Idle poll floor and ceiling. An idle matcher previously issued a claim +/// transaction every 250ms forever; backing off to the ceiling folds that +/// steady-state cost while keeping first-message latency at the floor. +const IDLE_POLL_FLOOR: Duration = Duration::from_millis(250); +const IDLE_POLL_CEILING: Duration = Duration::from_secs(2); +/// Cadence of the poison-job sweep, which lives off the claim path: its scan +/// is not served by the due partial index, so running it inside every claim +/// made claims slower exactly when a backlog needed them fastest. +const REAP_INTERVAL: Duration = Duration::from_secs(30); #[derive(Serialize)] struct DeliveryRequest<'a> { @@ -37,29 +49,36 @@ enum DeliveryResponse { }, } -/// Continuously claim accepted events and match them against active leases. +/// Continuously claim accepted events in per-community batches and match +/// them against active leases (T2b). One batch costs one claim statement, +/// one event load, one lease scan, and one membership scan — plus one +/// complete/retry statement each — regardless of batch size. pub async fn run_matcher(state: Arc) { + let mut idle_delay = IDLE_POLL_FLOOR; + let mut last_reap = tokio::time::Instant::now(); loop { + if last_reap.elapsed() >= REAP_INTERVAL { + match state.db.reap_exhausted_push_matches().await { + Ok(reaped) if reaped > 0 => warn!(reaped, "reaped exhausted push match jobs"), + Ok(_) => {} + Err(e) => error!("push match reap failed: {e}"), + } + last_reap = tokio::time::Instant::now(); + } let until = Utc::now() + TimeDelta::seconds(CLAIM_SECS); - match state.db.claim_due_push_match(until).await { - Ok(Some(job)) => { - if let Err(e) = process_match(&state, &job).await { - warn!(event_id=%job.event.event.id, attempt=job.attempt, "push match failed: {e}"); - if job.attempt >= buzz_db::push::MAX_MATCH_ATTEMPTS { - // A poison event/lease must not retry forever or pin - // delivered outbox retention through the rematch guard. - let _ = state.db.complete_push_match(&job).await; - } else { - let _ = state - .db - .retry_push_match(&job, Utc::now() + TimeDelta::seconds(2)) - .await; - } - } else if let Err(e) = state.db.complete_push_match(&job).await { - warn!(event_id=%job.event.event.id, "push match completion failed: {e}"); - } + match state + .db + .claim_due_push_match_batch(MATCH_BATCH_LIMIT, until) + .await + { + Ok(Some(batch)) => { + idle_delay = IDLE_POLL_FLOOR; + process_match_batch(&state, batch).await; + } + Ok(None) => { + tokio::time::sleep(idle_delay).await; + idle_delay = (idle_delay * 2).min(IDLE_POLL_CEILING); } - Ok(None) => tokio::time::sleep(Duration::from_millis(250)).await, Err(e) => { error!("push matcher claim failed: {e}"); tokio::time::sleep(Duration::from_secs(2)).await; @@ -68,18 +87,123 @@ pub async fn run_matcher(state: Arc) { } } -async fn process_match(state: &AppState, job: &buzz_db::push::ClaimedMatch) -> anyhow::Result<()> { - let leases = state.db.active_push_match_leases(job.community).await?; - for lease in leases { +/// Per-batch state shared by every job: the community's active leases and +/// the exact (channel, lease author) membership pairs the jobs can consult. +struct MatchContext { + leases: Vec, + memberships: std::collections::HashSet<(uuid::Uuid, Vec)>, +} + +async fn load_match_context( + state: &AppState, + batch: &buzz_db::push::ClaimedMatchBatch, +) -> anyhow::Result { + let leases = state.db.active_push_match_leases(batch.community).await?; + let mut channels: Vec = batch + .jobs + .iter() + .filter_map(|job| job.event.channel_id) + .collect(); + channels.sort_unstable(); + channels.dedup(); + let mut authors: Vec> = leases.iter().map(|lease| lease.author.clone()).collect(); + authors.sort_unstable(); + authors.dedup(); + let memberships = state + .db + .membership_pairs(batch.community, &channels, &authors) + .await? + .into_iter() + .collect(); + Ok(MatchContext { + leases, + memberships, + }) +} + +async fn process_match_batch(state: &AppState, batch: buzz_db::push::ClaimedMatchBatch) { + let community = batch.community; + let context = match load_match_context(state, &batch).await { + Ok(context) => context, + Err(e) => { + // Shared context failed to load, so no job was evaluated: release + // the whole batch for retry. Jobs that keep failing are reaped by + // the periodic sweep once their attempts are exhausted. + warn!(%community, "push match context load failed: {e}"); + let ids: Vec> = batch + .jobs + .iter() + .map(|job| job.event.event.id.as_bytes().to_vec()) + .collect(); + if let Err(e) = state + .db + .retry_push_match_batch( + community, + batch.claim_id, + &ids, + Utc::now() + TimeDelta::seconds(2), + ) + .await + { + warn!(%community, "push match batch retry failed: {e}"); + } + return; + } + }; + let mut completed = Vec::new(); + let mut retry = Vec::new(); + for job in &batch.jobs { + let event_id = job.event.event.id.as_bytes().to_vec(); + match process_match(state, community, job, &context).await { + Ok(()) => completed.push(event_id), + Err(e) => { + warn!(event_id=%job.event.event.id, attempt=job.attempt, "push match failed: {e}"); + if job.attempt >= buzz_db::push::MAX_MATCH_ATTEMPTS { + // A poison event/lease must not retry forever or pin + // delivered outbox retention through the rematch guard. + completed.push(event_id); + } else { + retry.push(event_id); + } + } + } + } + if let Err(e) = state + .db + .complete_push_match_batch(community, batch.claim_id, &completed) + .await + { + warn!(%community, "push match batch completion failed: {e}"); + } + if let Err(e) = state + .db + .retry_push_match_batch( + community, + batch.claim_id, + &retry, + Utc::now() + TimeDelta::seconds(2), + ) + .await + { + warn!(%community, "push match batch retry failed: {e}"); + } +} + +async fn process_match( + state: &AppState, + community: buzz_core::CommunityId, + job: &buzz_db::push::BatchedMatch, + context: &MatchContext, +) -> anyhow::Result<()> { + for lease in &context.leases { let author_hex = hex::encode(&lease.author); if !reader_authorized_for_event(&job.event.event, &author_hex) { continue; } if let Some(channel) = job.event.channel_id { - if !state - .db - .is_member(job.community, channel, &lease.author) - .await? + if !context + .memberships + .contains(&(channel, lease.author.clone())) { continue; } @@ -126,7 +250,7 @@ async fn process_match(state: &AppState, job: &buzz_db::push::ClaimedMatch) -> a let _ = state .db .enqueue_push_wake( - job.community, + community, &lease.author, &lease.installation_id, buzz_db::push::NewWake { @@ -169,6 +293,7 @@ pub async fn run_delivery_worker(state: Arc) { .timeout(state.config.push_gateway_timeout) .build() .expect("push HTTP client"); + let mut idle_delay = Duration::from_millis(500); loop { let mut found = false; match state.db.usage_community_hosts().await { @@ -189,8 +314,13 @@ pub async fn run_delivery_worker(state: Arc) { } Err(e) => error!("push worker community scan failed: {e}"), } - if !found { - tokio::time::sleep(Duration::from_millis(500)).await; + if found { + idle_delay = Duration::from_millis(500); + } else { + // Empty sweeps back off so an idle worker stops paying a full + // per-community claim scan every 500ms forever. + tokio::time::sleep(idle_delay).await; + idle_delay = (idle_delay * 2).min(IDLE_POLL_CEILING); } } } diff --git a/migrations/0023_push_match_gate.sql b/migrations/0023_push_match_gate.sql new file mode 100644 index 0000000000..fe131226ff --- /dev/null +++ b/migrations/0023_push_match_gate.sql @@ -0,0 +1,43 @@ +-- T1b push gate: skip push_match_queue enqueue entirely for communities with +-- no active, endpoint-enabled, unexpired push lease. In lease-less communities +-- (most of them) every durable message currently pays the full matcher cost +-- (enqueue + claim + lease scan + delete) to conclude "notify no one". +-- +-- Correctness protocol (write-amp plan rev 3, [R2/R3]): +-- * The gate lives HERE, in the events trigger, so every durable producer is +-- covered — including internal paths that bypass live dispatch. +-- * Lost-wake race: a naive EXISTS check could read "no lease" while a lease +-- activation commits concurrently, silently dropping that user's wake with +-- no retry. Closed with a per-community advisory lock held to transaction +-- end: event inserts take the lock SHARED (concurrent with each other), +-- lease transitions that can make eligibility true take it EXCLUSIVE +-- (crates/buzz-db/src/push.rs: accept_lease_event and replace_lease). +-- The conflict forces a total order: either the event's check sees the +-- committed lease, or the activation strictly follows the event's commit — +-- in which case no lease existed when the event was accepted and no wake +-- was owed. The lease-activation backfill is product recovery coverage +-- only and is not part of this proof. +-- * Lock key domain 'buzz_push_gate:' is distinct from the audit lock +-- ('buzz_audit:') and both lease-address lock families. +CREATE OR REPLACE FUNCTION enqueue_push_match_job() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + -- Keep this allowlist identical to the relay's validated NIP-PL descriptor. + IF NEW.kind IN (7, 9, 1059, 40007, 46010) THEN + PERFORM pg_advisory_xact_lock_shared( + hashtextextended('buzz_push_gate:' || NEW.community_id::text, 0)); + IF EXISTS ( + SELECT 1 FROM push_leases + WHERE community_id = NEW.community_id + AND active + AND endpoint_enabled + AND expires_at > EXTRACT(EPOCH FROM now())::bigint + ) THEN + INSERT INTO push_match_queue (community_id, event_id) + VALUES (NEW.community_id, NEW.id) + ON CONFLICT DO NOTHING; + END IF; + END IF; + RETURN NEW; +END +$$; diff --git a/migrations/0024_event_ttl_refresh_shared_lock.sql b/migrations/0024_event_ttl_refresh_shared_lock.sql new file mode 100644 index 0000000000..77116b72bb --- /dev/null +++ b/migrations/0024_event_ttl_refresh_shared_lock.sql @@ -0,0 +1,58 @@ +-- T1a repair: the 0022 trigger takes FOR UPDATE on the channel row before +-- testing ttl_seconds, so every durable message in a PERMANENT channel +-- serializes on that tuple at commit time (deferred trigger) — one hot +-- channel means fully serialized commits, each holding the lock across its +-- WAL flush. Observed live at 200 QPS: commit latency 0.07ms -> ~15ms, +-- non-CPU DB load ~9/vCPU with CPU under 45%. +-- +-- Repair keeps 0022's stale-prefetch proof but moves the synchronization to +-- a per-channel advisory lock, shared on the hot path: +-- * Event insert: shared channel-key lock -> read ttl_seconds. NULL returns +-- with no tuple lock and no update; shared locks admit each other, so +-- permanent-channel commits proceed concurrently. +-- * Permanent->ephemeral (or TTL-change) transition (update_channel in +-- crates/buzz-db/src/channel.rs) takes the same key EXCLUSIVE before its +-- UPDATE. Either the transition commits first and the event's read sees +-- the TTL (and refreshes), or the event commits first and the +-- transition's own deadline reset is later than anything the event would +-- have written. No stale-NULL hole in either order. +-- * Ephemeral channels still run the conditional UPDATE; their row updates +-- serialize per channel, but only ephemeral channels pay that. +-- Lock key domain 'buzz_channel_ttl:' is distinct from 'buzz_push_gate:' +-- (migration 0023) and the audit/lease lock families. Lock order note: the +-- deferred trigger acquires this key at COMMIT, after any push-gate shared +-- lock taken during insert; no path acquires both domains exclusively. +CREATE OR REPLACE FUNCTION refresh_channel_ttl_after_event_insert() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + channel_ttl INTEGER; +BEGIN + -- Kind 9007 creates the channel and initializes its deadline itself. + IF NEW.channel_id IS NOT NULL AND NEW.kind <> 9007 THEN + BEGIN + PERFORM pg_advisory_xact_lock_shared(hashtextextended( + 'buzz_channel_ttl:' || NEW.community_id::text || ':' || NEW.channel_id::text, 0)); + + SELECT ttl_seconds INTO channel_ttl + FROM channels + WHERE community_id = NEW.community_id AND id = NEW.channel_id; + + IF channel_ttl IS NOT NULL THEN + UPDATE channels + SET ttl_deadline = clock_timestamp() + make_interval(secs => ttl_seconds) + WHERE community_id = NEW.community_id + AND id = NEW.channel_id + AND ttl_seconds IS NOT NULL + AND archived_at IS NULL + AND deleted_at IS NULL; + END IF; + EXCEPTION WHEN OTHERS THEN + -- Preserve the existing best-effort contract: a TTL refresh failure + -- must not reject an otherwise valid durable event. + RAISE WARNING 'channel TTL refresh failed for community %, channel %: %', + NEW.community_id, NEW.channel_id, SQLERRM; + END; + END IF; + RETURN NULL; +END +$$; diff --git a/schema/schema.sql b/schema/schema.sql index bd805895d3..1fd13941be 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -818,6 +818,10 @@ CREATE INDEX push_match_queue_due CREATE INDEX push_match_queue_recovery ON push_match_queue (lease_until) WHERE state = 'matching'; +-- T1b push gate (keep in sync with migrations/0023). Enqueue only when the +-- community has an active, endpoint-enabled, unexpired lease; the shared +-- advisory lock pairs with the exclusive lock taken by lease activations +-- (crates/buzz-db/src/push.rs) to close the lost-wake race. CREATE FUNCTION enqueue_push_match_job() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN @@ -825,9 +829,19 @@ BEGIN -- Centralizing it on the events table covers every durable producer, -- including internal paths that bypass live dispatch. IF NEW.kind IN (7, 9, 1059, 40007, 46010) THEN - INSERT INTO push_match_queue (community_id, event_id) - VALUES (NEW.community_id, NEW.id) - ON CONFLICT DO NOTHING; + PERFORM pg_advisory_xact_lock_shared( + hashtextextended('buzz_push_gate:' || NEW.community_id::text, 0)); + IF EXISTS ( + SELECT 1 FROM push_leases + WHERE community_id = NEW.community_id + AND active + AND endpoint_enabled + AND expires_at > EXTRACT(EPOCH FROM now())::bigint + ) THEN + INSERT INTO push_match_queue (community_id, event_id) + VALUES (NEW.community_id, NEW.id) + ON CONFLICT DO NOTHING; + END IF; END IF; RETURN NEW; END @@ -837,6 +851,53 @@ CREATE TRIGGER events_enqueue_push_match AFTER INSERT ON events FOR EACH ROW EXECUTE FUNCTION enqueue_push_match_job(); +-- Channel TTL refresh (keep in sync with migrations/0024). Runs deferred, in +-- the transaction that makes a channel-scoped event durable, so a TTL +-- transition committed while ingest was in flight is never missed. The +-- per-channel advisory lock is SHARED here — permanent-channel commits admit +-- each other — and taken EXCLUSIVE by TTL transitions (update_channel in +-- crates/buzz-db/src/channel.rs), which forces the same total order the +-- 0022 row lock provided without serializing the hot path. +CREATE FUNCTION refresh_channel_ttl_after_event_insert() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + channel_ttl INTEGER; +BEGIN + -- Kind 9007 creates the channel and initializes its deadline itself. + IF NEW.channel_id IS NOT NULL AND NEW.kind <> 9007 THEN + BEGIN + PERFORM pg_advisory_xact_lock_shared(hashtextextended( + 'buzz_channel_ttl:' || NEW.community_id::text || ':' || NEW.channel_id::text, 0)); + + SELECT ttl_seconds INTO channel_ttl + FROM channels + WHERE community_id = NEW.community_id AND id = NEW.channel_id; + + IF channel_ttl IS NOT NULL THEN + UPDATE channels + SET ttl_deadline = clock_timestamp() + make_interval(secs => ttl_seconds) + WHERE community_id = NEW.community_id + AND id = NEW.channel_id + AND ttl_seconds IS NOT NULL + AND archived_at IS NULL + AND deleted_at IS NULL; + END IF; + EXCEPTION WHEN OTHERS THEN + -- Preserve the existing best-effort contract: a TTL refresh failure + -- must not reject an otherwise valid durable event. + RAISE WARNING 'channel TTL refresh failed for community %, channel %: %', + NEW.community_id, NEW.channel_id, SQLERRM; + END; + END IF; + RETURN NULL; +END +$$; + +CREATE CONSTRAINT TRIGGER events_refresh_channel_ttl +AFTER INSERT ON events +DEFERRABLE INITIALLY DEFERRED +FOR EACH ROW EXECUTE FUNCTION refresh_channel_ttl_after_event_insert(); + -- Replica-fence floor guard (keep in sync with migrations/0021). A deferred -- constraint trigger re-checks, inside COMMIT processing, that channel-bearing -- event rows are no older than `buzz.created_at_floor` seconds before commit diff --git a/scripts/attach-schema-partitions.sql b/scripts/attach-schema-partitions.sql index a61ab099fd..d02fc7c82a 100644 --- a/scripts/attach-schema-partitions.sql +++ b/scripts/attach-schema-partitions.sql @@ -19,6 +19,7 @@ BEGIN -- triggers while attaching and rejects same-named child triggers -- (both the push-match trigger and the replica-fence floor guard). DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p_past; + DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p_past; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p_past; ALTER TABLE events ATTACH PARTITION events_p_past FOR VALUES FROM (MINVALUE) TO ('2026-01-01'); @@ -30,6 +31,7 @@ BEGIN AND inhrelid = 'events_p2026_01'::regclass ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_01; + DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_01; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_01; ALTER TABLE events ATTACH PARTITION events_p2026_01 FOR VALUES FROM ('2026-01-01') TO ('2026-02-01'); @@ -41,6 +43,7 @@ BEGIN AND inhrelid = 'events_p2026_02'::regclass ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_02; + DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_02; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_02; ALTER TABLE events ATTACH PARTITION events_p2026_02 FOR VALUES FROM ('2026-02-01') TO ('2026-03-01'); @@ -52,6 +55,7 @@ BEGIN AND inhrelid = 'events_p2026_03'::regclass ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_03; + DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_03; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_03; ALTER TABLE events ATTACH PARTITION events_p2026_03 FOR VALUES FROM ('2026-03-01') TO ('2026-04-01'); @@ -63,6 +67,7 @@ BEGIN AND inhrelid = 'events_p2026_04'::regclass ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_04; + DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_04; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_04; ALTER TABLE events ATTACH PARTITION events_p2026_04 FOR VALUES FROM ('2026-04-01') TO ('2026-05-01'); @@ -74,6 +79,7 @@ BEGIN AND inhrelid = 'events_p2026_05'::regclass ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_05; + DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_05; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_05; ALTER TABLE events ATTACH PARTITION events_p2026_05 FOR VALUES FROM ('2026-05-01') TO ('2026-06-01'); @@ -85,6 +91,7 @@ BEGIN AND inhrelid = 'events_p2026_06'::regclass ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p2026_06; + DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p2026_06; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p2026_06; ALTER TABLE events ATTACH PARTITION events_p2026_06 FOR VALUES FROM ('2026-06-01') TO ('2026-07-01'); @@ -96,6 +103,7 @@ BEGIN AND inhrelid = 'events_p_future'::regclass ) THEN DROP TRIGGER IF EXISTS events_enqueue_push_match ON events_p_future; + DROP TRIGGER IF EXISTS events_refresh_channel_ttl ON events_p_future; DROP TRIGGER IF EXISTS events_created_at_floor ON events_p_future; ALTER TABLE events ATTACH PARTITION events_p_future FOR VALUES FROM ('2026-07-01') TO (MAXVALUE); From 3173861ed5b72e47497e3ed1789018e64d1a2c64 Mon Sep 17 00:00:00 2001 From: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Date: Sun, 19 Jul 2026 20:30:50 -0400 Subject: [PATCH 2/2] relay: make T2b wake enqueue set-wise (one txn per matched batch) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Max's red-team of cf018e1bd found the remaining T2b hole: every successful (event, lease) match still called enqueue_push_wake inside the matcher loop — its own transaction, a lease SELECT ... FOR UPDATE, one outbox insert, one commit. Live-lease workload therefore stayed O(events x matching leases) transactions/WAL flushes, and the "constant statements per batch" claim only held up to the wake enqueue. New set-wise primitive push::enqueue_wakes: one transaction per batch — lease revalidation via a single ordered FOR UPDATE over the distinct (author, installation) set (the ORDER BY pins lock order against replace_lease and sibling pods), one multi-row UNNEST INSERT ... ON CONFLICT DO NOTHING RETURNING under the existing (community, endpoint_hash, event_id) dedup key, and one set-wise duplicate lookup so per-request outcomes (Enqueued / Duplicate / InactiveLease) stay exact. enqueue_wake becomes a batch-of-one wrapper — one code path, same shape as the claim API. The matcher goes pure: match_job evaluates leases with no DB access and returns the WakeRequests a job owes; process_match_batch flushes all of them in one enqueue_wakes call. A failed flush sends only the contributing jobs back for an idempotent rematch (the dedup key absorbs wakes that did commit); wake-less jobs still complete. Also honors TEST_DATABASE_URL in test_usage_metrics_lock_has_single_owner_and_releases_on_drop, which hardcoded the shared dev DB and failed whenever a live local relay held the usage-metrics advisory lock. Tests: setwise_enqueue_maps_outcomes_per_request covers fresh insert, duplicate-of-committed, same-batch dedup collision, stale generation, unknown lease, and a second live lease in one batch. buzz-db PG suite 202/202 green on a fresh isolated 0001-0024 database; full-gate rerun and scenario B before/after numbers to follow on this SHA. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-db/src/lib.rs | 13 +- crates/buzz-db/src/push.rs | 338 ++++++++++++++++++++++---- crates/buzz-relay/src/push_runtime.rs | 67 +++-- 3 files changed, 341 insertions(+), 77 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index c059e5c7e3..fdd72c3c32 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1274,6 +1274,15 @@ impl Db { push::enqueue_wake(&self.pool, community, author, installation_id, wake).await } + /// Set-wise [`Self::enqueue_push_wake`]: one transaction per batch. + pub async fn enqueue_push_wakes( + &self, + community: CommunityId, + requests: &[push::WakeRequest], + ) -> Result> { + push::enqueue_wakes(&self.pool, community, requests).await + } + /// Exclusively claim due wake jobs for one community. pub async fn claim_due_push_wakes( &self, @@ -4642,9 +4651,11 @@ mod tests { #[tokio::test] #[ignore = "requires Postgres"] async fn test_usage_metrics_lock_has_single_owner_and_releases_on_drop() { + let database_url = + std::env::var("TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.into()); let pool = PgPoolOptions::new() .max_connections(2) - .connect(TEST_DB_URL) + .connect(&database_url) .await .expect("connect to test DB"); let first = Db::from_pool(pool.clone()); diff --git a/crates/buzz-db/src/push.rs b/crates/buzz-db/src/push.rs index b0042ec92f..04b6a7ae36 100644 --- a/crates/buzz-db/src/push.rs +++ b/crates/buzz-db/src/push.rs @@ -557,11 +557,26 @@ async fn replace_lease( } } +/// One matcher-produced wake request inside a set-wise enqueue. +#[derive(Debug, Clone)] +pub struct WakeRequest { + /// Lease owner's raw public key. + pub author: Vec, + /// Installation address within the tenant. + pub installation_id: String, + /// Generation observed by the matcher. + pub lease_generation: i64, + /// Accepted event id that caused the wake (32 bytes). + pub event_id: Vec, + /// Effective wake class. + pub class: String, + /// Delivery deadline, in Unix seconds. + pub expires_at: i64, +} + /// Atomically enqueue at most one job per community, endpoint, and event. /// -/// Endpoint identity and the endpoint grant are copied from the current lease; -/// callers cannot redirect a wake by supplying either value. A generation that -/// lost a replacement race is ineligible in the same statement that inserts. +/// Batch-of-one shape of [`enqueue_wakes`]; see there for the protocol. pub async fn enqueue_wake( pool: &PgPool, community: CommunityId, @@ -569,74 +584,207 @@ pub async fn enqueue_wake( installation_id: &str, wake: NewWake<'_>, ) -> Result { + let outcomes = enqueue_wakes( + pool, + community, + &[WakeRequest { + author: author.to_vec(), + installation_id: installation_id.to_string(), + lease_generation: wake.lease_generation, + event_id: wake.event_id.to_vec(), + class: wake.class.to_string(), + expires_at: wake.expires_at, + }], + ) + .await?; + Ok(outcomes + .into_iter() + .next() + .expect("one outcome per request")) +} + +/// Set-wise counterpart of [`enqueue_wake`]: one transaction and a constant +/// number of statements for any number of matched (lease, event) pairs (T2b). +/// +/// Endpoint identity and the endpoint grant are copied from each current +/// lease; callers cannot redirect a wake by supplying either value. +/// Revalidation locks the distinct lease rows `FOR UPDATE` in one statement, +/// ordered by (author, installation_id) so concurrent batches and +/// `replace_active_lease` (single-row lock) acquire in a consistent order. If +/// enqueue wins a lock race, a later replacement can leave a durable job +/// queued, but worker revalidation suppresses it; if replacement wins, the +/// generation comparison fails and that request reports `InactiveLease`. +/// +/// Returns one outcome per request, index-aligned with `requests`. +pub async fn enqueue_wakes( + pool: &PgPool, + community: CommunityId, + requests: &[WakeRequest], +) -> Result> { + if requests.is_empty() { + return Ok(Vec::new()); + } let mut tx = pool.begin().await?; - // Serialize against lease replacement. If enqueue wins the lock, a later - // replacement can leave this durable job queued, but worker revalidation - // will suppress it; if replacement wins, the generation predicate fails. - let endpoint_hash = sqlx::query( + + // 1. Lock and read the current lease row for every distinct requested + // (author, installation), in deterministic order. + let mut pairs: Vec<(&[u8], &str)> = requests + .iter() + .map(|r| (r.author.as_slice(), r.installation_id.as_str())) + .collect(); + pairs.sort_unstable(); + pairs.dedup(); + let lock_authors: Vec<&[u8]> = pairs.iter().map(|(a, _)| *a).collect(); + let lock_installs: Vec<&str> = pairs.iter().map(|(_, i)| *i).collect(); + let lease_rows = sqlx::query( r#" - SELECT endpoint_hash + SELECT author, installation_id, generation, endpoint_hash FROM push_leases WHERE community_id = $1 - AND author = $2 - AND installation_id = $3 - AND generation = $4 + AND (author, installation_id) IN + (SELECT a, i FROM UNNEST($2::bytea[], $3::text[]) AS t(a, i)) AND active AND endpoint_enabled AND expires_at > EXTRACT(EPOCH FROM now())::bigint + ORDER BY author, installation_id FOR UPDATE "#, ) .bind(community.as_uuid()) - .bind(author) - .bind(installation_id) - .bind(wake.lease_generation) - .fetch_optional(&mut *tx) + .bind(&lock_authors) + .bind(&lock_installs) + .fetch_all(&mut *tx) .await?; - let Some(endpoint_hash) = endpoint_hash else { - return Ok(EnqueueWakeOutcome::InactiveLease); - }; - let endpoint_hash: Vec = endpoint_hash.try_get("endpoint_hash")?; + let mut leases: std::collections::HashMap<(Vec, String), (i64, Vec)> = + std::collections::HashMap::with_capacity(lease_rows.len()); + for row in lease_rows { + leases.insert( + (row.try_get("author")?, row.try_get("installation_id")?), + (row.try_get("generation")?, row.try_get("endpoint_hash")?), + ); + } - let inserted = sqlx::query( - r#" - INSERT INTO push_wake_outbox ( - community_id, author, installation_id, lease_generation, - endpoint_hash, event_id, class, expires_at - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) - ON CONFLICT (community_id, endpoint_hash, event_id) DO NOTHING - RETURNING id - "#, - ) - .bind(community.as_uuid()) - .bind(author) - .bind(installation_id) - .bind(wake.lease_generation) - .bind(&endpoint_hash) - .bind(wake.event_id) - .bind(wake.class) - .bind(wake.expires_at) - .fetch_optional(&mut *tx) - .await?; + // 2. Resolve per-request eligibility; collect the insert arrays. + // `None` marks InactiveLease; `Some(endpoint_hash)` carries the key half + // that maps insert/duplicate rows back to their requests. + let resolved: Vec>> = requests + .iter() + .map(|r| { + leases + .get(&(r.author.clone(), r.installation_id.clone())) + .filter(|(generation, _)| *generation == r.lease_generation) + .map(|(_, endpoint_hash)| endpoint_hash.clone()) + }) + .collect(); + let mut ins_authors: Vec<&[u8]> = Vec::new(); + let mut ins_installs: Vec<&str> = Vec::new(); + let mut ins_generations: Vec = Vec::new(); + let mut ins_endpoints: Vec<&[u8]> = Vec::new(); + let mut ins_events: Vec<&[u8]> = Vec::new(); + let mut ins_classes: Vec<&str> = Vec::new(); + let mut ins_expires: Vec = Vec::new(); + for (request, endpoint_hash) in requests.iter().zip(&resolved) { + let Some(endpoint_hash) = endpoint_hash else { + continue; + }; + ins_authors.push(&request.author); + ins_installs.push(&request.installation_id); + ins_generations.push(request.lease_generation); + ins_endpoints.push(endpoint_hash); + ins_events.push(&request.event_id); + ins_classes.push(&request.class); + ins_expires.push(request.expires_at); + } - let outcome = if let Some(row) = inserted { - EnqueueWakeOutcome::Enqueued(row.try_get("id")?) - } else { - // This is a separate statement so READ COMMITTED observes a competing - // transaction whose unique-key insert completed while ours waited. - let row = sqlx::query( - "SELECT id FROM push_wake_outbox \ - WHERE community_id = $1 AND endpoint_hash = $2 AND event_id = $3", + // 3. One multi-row insert. ON CONFLICT covers both pre-existing jobs and + // dedup-key collisions between rows of this same statement. + let mut job_ids: std::collections::HashMap<(Vec, Vec), Uuid> = + std::collections::HashMap::new(); + let mut inserted_keys: std::collections::HashSet<(Vec, Vec)> = + std::collections::HashSet::new(); + if !ins_events.is_empty() { + let inserted = sqlx::query( + r#" + INSERT INTO push_wake_outbox ( + community_id, author, installation_id, lease_generation, + endpoint_hash, event_id, class, expires_at + ) + SELECT $1, a, i, g, eh, ev, c, ex + FROM UNNEST( + $2::bytea[], $3::text[], $4::bigint[], $5::bytea[], + $6::bytea[], $7::text[], $8::bigint[] + ) AS t(a, i, g, eh, ev, c, ex) + ON CONFLICT (community_id, endpoint_hash, event_id) DO NOTHING + RETURNING endpoint_hash, event_id, id + "#, ) .bind(community.as_uuid()) - .bind(&endpoint_hash) - .bind(wake.event_id) - .fetch_one(&mut *tx) + .bind(&ins_authors) + .bind(&ins_installs) + .bind(&ins_generations) + .bind(&ins_endpoints) + .bind(&ins_events) + .bind(&ins_classes) + .bind(&ins_expires) + .fetch_all(&mut *tx) .await?; - EnqueueWakeOutcome::Duplicate(row.try_get("id")?) - }; + for row in inserted { + let key: (Vec, Vec) = (row.try_get("endpoint_hash")?, row.try_get("event_id")?); + job_ids.insert(key.clone(), row.try_get("id")?); + inserted_keys.insert(key); + } + // 4. Set-wise duplicate lookup for eligible requests the insert + // skipped. A separate statement so READ COMMITTED observes a + // competing transaction whose unique-key insert completed while + // ours waited. + if inserted_keys.len() < ins_events.len() { + let dup_rows = sqlx::query( + r#" + SELECT endpoint_hash, event_id, id FROM push_wake_outbox + WHERE community_id = $1 + AND (endpoint_hash, event_id) IN + (SELECT eh, ev FROM UNNEST($2::bytea[], $3::bytea[]) AS t(eh, ev)) + "#, + ) + .bind(community.as_uuid()) + .bind(&ins_endpoints) + .bind(&ins_events) + .fetch_all(&mut *tx) + .await?; + for row in dup_rows { + let key: (Vec, Vec) = + (row.try_get("endpoint_hash")?, row.try_get("event_id")?); + job_ids.entry(key).or_insert(row.try_get("id")?); + } + } + } tx.commit().await?; - Ok(outcome) + + // First request to claim an inserted key reports Enqueued; later + // same-key requests in this batch report Duplicate, matching what + // sequential single-request calls would have returned. + let mut reported: std::collections::HashSet<(Vec, Vec)> = + std::collections::HashSet::new(); + requests + .iter() + .zip(&resolved) + .map(|(request, endpoint_hash)| { + let Some(endpoint_hash) = endpoint_hash else { + return Ok(EnqueueWakeOutcome::InactiveLease); + }; + let key = (endpoint_hash.clone(), request.event_id.clone()); + let id = *job_ids.get(&key).ok_or_else(|| { + crate::error::DbError::InvalidData( + "wake enqueue resolved neither insert nor duplicate".into(), + ) + })?; + Ok(if inserted_keys.contains(&key) && reported.insert(key) { + EnqueueWakeOutcome::Enqueued(id) + } else { + EnqueueWakeOutcome::Duplicate(id) + }) + }) + .collect() } /// One batch of matcher jobs claimed from a single community. @@ -1434,6 +1582,90 @@ mod tests { assert_eq!(total, 2, "same dedup key is independent per community"); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn setwise_enqueue_maps_outcomes_per_request() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let alice = [21; 32]; + let bob = [22; 32]; + let alice_endpoint = [23; 32]; + let bob_endpoint = [24; 32]; + let event_x = [25; 32]; + let event_y = [26; 32]; + activate(&pool, community, &alice, "install", &alice_endpoint, 1).await; + activate(&pool, community, &bob, "install", &bob_endpoint, 2).await; + + // Pre-existing durable job for (alice, event_x): the batch's matching + // request must come back Duplicate with the same id. + let existing = enqueue_one(&pool, community, &alice, &event_x, 1).await; + 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(event_y) + .bind([42_u8; 32]) + .bind([43_u8; 64]) + .execute(&pool) + .await + .expect("insert second wake source event"); + + let request = |author: [u8; 32], event: [u8; 32], generation: i64| WakeRequest { + author: author.to_vec(), + installation_id: "install".into(), + lease_generation: generation, + event_id: event.to_vec(), + class: "default".into(), + expires_at: i64::MAX / 2, + }; + let outcomes = enqueue_wakes( + &pool, + community, + &[ + // Duplicate of the pre-existing job. + request(alice, event_x, 1), + // Fresh insert. + request(alice, event_y, 1), + // Same dedup key within this same batch. + request(alice, event_y, 1), + // Stale generation: lease revalidation must reject it. + request(bob, event_x, 7), + // Unknown lease entirely. + request([99; 32], event_x, 1), + // Fresh insert for the second lease. + request(bob, event_y, 2), + ], + ) + .await + .expect("set-wise enqueue"); + + assert_eq!(outcomes.len(), 6, "index-aligned outcomes"); + assert_eq!(outcomes[0], EnqueueWakeOutcome::Duplicate(existing)); + let EnqueueWakeOutcome::Enqueued(alice_y) = outcomes[1] else { + panic!( + "expected fresh insert for (alice, event_y), got {:?}", + outcomes[1] + ); + }; + assert_eq!( + outcomes[2], + EnqueueWakeOutcome::Duplicate(alice_y), + "same-batch dedup collision resolves to the row the batch inserted" + ); + assert_eq!(outcomes[3], EnqueueWakeOutcome::InactiveLease); + assert_eq!(outcomes[4], EnqueueWakeOutcome::InactiveLease); + assert!(matches!(outcomes[5], EnqueueWakeOutcome::Enqueued(id) if id != alice_y)); + + let total: i64 = + sqlx::query_scalar("SELECT count(*) FROM push_wake_outbox WHERE community_id = $1") + .bind(community.as_uuid()) + .fetch_one(&pool) + .await + .expect("count outbox rows"); + assert_eq!(total, 3, "one row per distinct (endpoint, event) key"); + } + async fn enqueue_one( pool: &PgPool, community: CommunityId, diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index 993db21805..49845067ea 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -51,8 +51,9 @@ enum DeliveryResponse { /// Continuously claim accepted events in per-community batches and match /// them against active leases (T2b). One batch costs one claim statement, -/// one event load, one lease scan, and one membership scan — plus one -/// complete/retry statement each — regardless of batch size. +/// one event load, one lease scan, one membership scan, and one wake-enqueue +/// transaction — plus one complete/retry statement each — regardless of +/// batch size or how many (event, lease) pairs match. pub async fn run_matcher(state: Arc) { let mut idle_delay = IDLE_POLL_FLOOR; let mut last_reap = tokio::time::Instant::now(); @@ -152,10 +153,18 @@ async fn process_match_batch(state: &AppState, batch: buzz_db::push::ClaimedMatc }; let mut completed = Vec::new(); let mut retry = Vec::new(); + // Jobs whose wakes are pending the set-wise enqueue below: their + // completion is decided by the flush, not by match evaluation. + let mut pending = Vec::new(); + let mut wakes: Vec = Vec::new(); for job in &batch.jobs { let event_id = job.event.event.id.as_bytes().to_vec(); - match process_match(state, community, job, &context).await { - Ok(()) => completed.push(event_id), + match match_job(job, &context) { + Ok(job_wakes) if job_wakes.is_empty() => completed.push(event_id), + Ok(job_wakes) => { + pending.push((event_id, job.attempt)); + wakes.extend(job_wakes); + } Err(e) => { warn!(event_id=%job.event.event.id, attempt=job.attempt, "push match failed: {e}"); if job.attempt >= buzz_db::push::MAX_MATCH_ATTEMPTS { @@ -168,6 +177,23 @@ async fn process_match_batch(state: &AppState, batch: buzz_db::push::ClaimedMatc } } } + // One transaction for every wake in the batch (T2b). InactiveLease and + // Duplicate outcomes are per-request non-errors; only a failed enqueue + // transaction sends the contributing jobs back for an idempotent rematch + // (the outbox dedup key absorbs any wakes that did commit elsewhere). + match state.db.enqueue_push_wakes(community, &wakes).await { + Ok(_) => completed.extend(pending.into_iter().map(|(event_id, _)| event_id)), + Err(e) => { + warn!(%community, "push wake batch enqueue failed: {e}"); + for (event_id, attempt) in pending { + if attempt >= buzz_db::push::MAX_MATCH_ATTEMPTS { + completed.push(event_id); + } else { + retry.push(event_id); + } + } + } + } if let Err(e) = state .db .complete_push_match_batch(community, batch.claim_id, &completed) @@ -189,12 +215,13 @@ async fn process_match_batch(state: &AppState, batch: buzz_db::push::ClaimedMatc } } -async fn process_match( - state: &AppState, - community: buzz_core::CommunityId, +/// Pure match evaluation: no DB access. Returns the wake requests this job +/// owes, which the caller flushes set-wise for the whole batch. +fn match_job( job: &buzz_db::push::BatchedMatch, context: &MatchContext, -) -> anyhow::Result<()> { +) -> anyhow::Result> { + let mut wakes = Vec::new(); for lease in &context.leases { let author_hex = hex::encode(&lease.author); if !reader_authorized_for_event(&job.event.event, &author_hex) { @@ -247,22 +274,16 @@ async fn process_match( if expires_at <= Utc::now().timestamp() { continue; } - let _ = state - .db - .enqueue_push_wake( - community, - &lease.author, - &lease.installation_id, - buzz_db::push::NewWake { - lease_generation: lease.generation, - event_id: job.event.event.id.as_bytes(), - class, - expires_at, - }, - ) - .await?; + wakes.push(buzz_db::push::WakeRequest { + author: lease.author.clone(), + installation_id: lease.installation_id.clone(), + lease_generation: lease.generation, + event_id: job.event.event.id.as_bytes().to_vec(), + class: class.to_string(), + expires_at, + }); } - Ok(()) + Ok(wakes) } /// Match-time counterpart of REQ's filter-level `#p` authorization gate.