From 3164b25cde5fc5e581d433eb5de01c39a3b0e1c0 Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Fri, 9 Oct 2026 10:06:50 -0700 Subject: [PATCH] fix(relay): enqueue push work after message commit Signed-off-by: Tom Brow --- Cargo.lock | 1 + crates/buzz-db/Cargo.toml | 1 + crates/buzz-db/src/runtime/admitted_tx.rs | 28 +++ crates/buzz-db/src/runtime/mod.rs | 36 ++- crates/buzz-db/src/runtime/push_enqueue.rs | 206 ++++++++++++++++ crates/buzz-db/src/store/event.rs | 5 + crates/buzz-db/src/store/event_follow_up.rs | 79 +----- .../store/event_follow_up_postgres_tests.rs | 227 +++++++++++++++--- crates/buzz-db/src/store/push.rs | 13 +- crates/buzz-db/src/store/replaceable.rs | 14 +- crates/buzz-relay/src/main.rs | 2 + crates/buzz-relay/src/state.rs | 1 + docs/push-gateway-deployment.md | 29 ++- migrations/0059_push_rollback_gate.sql | 6 +- schema/functions/enqueue_push_match_job.sql | 6 +- schema/tables/public/push_match_queue.sql | 8 +- 16 files changed, 523 insertions(+), 139 deletions(-) create mode 100644 crates/buzz-db/src/runtime/push_enqueue.rs diff --git a/Cargo.lock b/Cargo.lock index 15b190f8ae..b4986b67b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1203,6 +1203,7 @@ dependencies = [ "sqlx", "thiserror 2.0.18", "tokio", + "tokio-util", "tracing", "uuid", ] diff --git a/crates/buzz-db/Cargo.toml b/crates/buzz-db/Cargo.toml index 7e080808ac..80e80921ec 100644 --- a/crates/buzz-db/Cargo.toml +++ b/crates/buzz-db/Cargo.toml @@ -12,6 +12,7 @@ buzz-core = { workspace = true } buzz-datastore-tracing = { workspace = true } sqlx = { workspace = true } tokio = { workspace = true } +tokio-util = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } uuid = { workspace = true } diff --git a/crates/buzz-db/src/runtime/admitted_tx.rs b/crates/buzz-db/src/runtime/admitted_tx.rs index d26981804f..fb0bb4f41f 100644 --- a/crates/buzz-db/src/runtime/admitted_tx.rs +++ b/crates/buzz-db/src/runtime/admitted_tx.rs @@ -73,6 +73,8 @@ pub struct AdmittedTx { community: CommunityId, /// Channels whose TTL deadline [`AdmittedTx::commit`] refreshes. ttl_refresh_channels: BTreeSet, + push_enqueue: Option, + push_events: Vec>, } impl AdmittedTx { @@ -87,6 +89,8 @@ impl AdmittedTx { tx, community, ttl_refresh_channels: BTreeSet::new(), + push_enqueue: None, + push_events: Vec::new(), }) } @@ -104,9 +108,28 @@ impl AdmittedTx { tx, community: lease.community_id, ttl_refresh_channels: BTreeSet::new(), + push_enqueue: None, + push_events: Vec::new(), }) } + pub(crate) fn set_push_enqueue(&mut self, producer: Option) { + self.push_enqueue = producer; + } + + pub(crate) fn record_push_event(&mut self, event_id: &[u8], kind: i32) { + if self.push_enqueue.is_some() + && crate::store::event_follow_up::PUSH_MATCH_KINDS.contains(&kind) + { + // Bound transaction-local staging as well as the shared worker queue. + if self.push_events.len() < super::push_enqueue::CAPACITY { + self.push_events.push(event_id.to_vec()); + } else { + super::push_enqueue::record_drop("transaction_full"); + } + } + } + /// The community this transaction was admitted for. pub fn community(&self) -> CommunityId { self.community @@ -145,6 +168,11 @@ impl AdmittedTx { .await?; } self.tx.commit().await?; + if let Some(producer) = self.push_enqueue { + for event_id in self.push_events { + producer.submit(self.community, event_id); + } + } Ok(()) } diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 933ee4b029..7be9402da4 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -3,6 +3,7 @@ mod cold_start; mod connection_observability; pub mod migration; pub(crate) mod observability; +pub(crate) mod push_enqueue; pub mod replica_fence; pub use admitted_tx::AdmittedTx; @@ -189,6 +190,8 @@ pub(crate) async fn begin_community_event_write_transaction_with_legacy_metrics( #[derive(Clone, Debug)] pub struct Db { pub(crate) pool: PgPool, + /// Optional bounded post-commit producer, shared across handle clones. + pub(crate) push_enqueue: Option, /// Maximum connections configured for this pool (from [`DbConfig::max_connections`]). pub(crate) max_connections: u32, /// Optional read-replica pool (from [`DbConfig::read_database_url`]). @@ -705,7 +708,11 @@ impl Db { let replica_read_max_age = read_budget_from_ms(config.replica_read_max_age_ms); let usage_metrics_replica_max_age = read_budget_from_ms(config.usage_metrics_replica_max_age_ms); + let push_enqueue = config + .push_enabled + .then(|| push_enqueue::PushEnqueue::new(&pool)); Ok(Self { + push_enqueue, pool, max_connections: config.max_connections, read_pool, @@ -781,7 +788,8 @@ impl Db { set_config('idle_in_transaction_session_timeout', $2, false), \ set_config('statement_timeout', $3, false), \ set_config('default_transaction_read_only', $4, false), \ - set_config('buzz.push_enabled', $5, false)", + set_config('buzz.push_enabled', $5, false), \ + set_config('buzz.push_async_enqueue', 'on', false)", ) .bind(lock_timeout_ms.to_string()) .bind(idle_txn_timeout_ms.to_string()) @@ -942,8 +950,12 @@ impl Db { } /// Creates a `Db` from an existing `PgPool` (useful in tests). + /// + /// Does not start a push producer or configure sessions. Serving writers + /// that need push must use [`Db::new`] with `push_enabled` instead. pub fn from_pool(pool: PgPool) -> Self { Self { + push_enqueue: None, max_connections: pool.options().get_max_connections(), read_max_connections: pool.options().get_max_connections(), pool, @@ -964,6 +976,7 @@ impl Db { /// [`Db::fence`]). pub fn from_pools(pool: PgPool, read_pool: PgPool) -> Self { Self { + push_enqueue: None, max_connections: pool.options().get_max_connections(), read_max_connections: read_pool.options().get_max_connections(), pool, @@ -1324,12 +1337,28 @@ impl Db { &self, community: CommunityId, ) -> Result { - begin_community_event_write_transaction_with_legacy_metrics( + let mut tx = begin_community_event_write_transaction_with_legacy_metrics( &self.pool, community, observability::WriterOperation::EventWrite, ) - .await + .await?; + tx.set_push_enqueue(self.push_enqueue.clone()); + Ok(tx) + } + + /// Stop accepting best-effort push enqueue work and cancel in-flight work. + pub fn cancel_push_enqueue(&self) { + if let Some(producer) = &self.push_enqueue { + producer.cancel(); + } + } + + /// Wait for the cancelled producer to release its dedicated connection. + pub async fn join_push_enqueue(&self) { + if let Some(producer) = &self.push_enqueue { + producer.join().await; + } } /// Insert an event while holding and validating an admitted serving-write @@ -1359,6 +1388,7 @@ impl Db { let tx = sqlx::Transaction::begin(connection, None).await?; let mut tx = AdmittedTx::admit_with_serving_lease(tx, &self.deletion_store(), lease).await?; + tx.set_push_enqueue(self.push_enqueue.clone()); event::acquire_canvas_event_write_lock_if_needed(&mut tx, event, channel_id).await?; let result = event::insert_event_with_thread_metadata_tx(&mut tx, event, channel_id, None).await?; diff --git a/crates/buzz-db/src/runtime/push_enqueue.rs b/crates/buzz-db/src/runtime/push_enqueue.rs new file mode 100644 index 0000000000..ca72694257 --- /dev/null +++ b/crates/buzz-db/src/runtime/push_enqueue.rs @@ -0,0 +1,206 @@ +//! Bounded best-effort push production, strictly after event commit. + +use std::sync::Arc; +use std::time::Duration; + +use buzz_core::CommunityId; +use sqlx::{postgres::PgPoolOptions, PgPool}; +use tokio::sync::{mpsc, Mutex}; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +pub(crate) const CAPACITY: usize = 256; +const ENQUEUE_TIMEOUT: Duration = Duration::from_secs(2); + +#[derive(Clone, Debug)] +pub(crate) struct PushEnqueue { + sender: mpsc::Sender, + cancel: CancellationToken, + task: Arc>>>, +} + +#[derive(Debug)] +struct Job { + community: CommunityId, + event_id: Vec, + #[cfg(test)] + barrier: Option>, +} + +impl PushEnqueue { + pub(crate) fn new(writer: &PgPool) -> Self { + // No connection from the serving pool is ever consumed by push work. + // This lazy pool cannot make relay startup depend on push availability. + let pool = PgPoolOptions::new() + .max_connections(1) + .min_connections(0) + .acquire_timeout(ENQUEUE_TIMEOUT) + .connect_lazy_with( + (*writer.connect_options()) + .clone() + .application_name("buzz-push-enqueue"), + ); + let (sender, mut receiver) = mpsc::channel::(CAPACITY); + let cancel = CancellationToken::new(); + let stopped = cancel.clone(); + let task = tokio::spawn(async move { + loop { + let job = tokio::select! { + biased; + _ = stopped.cancelled() => break, + job = receiver.recv() => match job { + Some(job) => job, + None => break, + }, + }; + #[cfg(test)] + if let Some(barrier) = job.barrier { + let _ = barrier.send(()); + continue; + } + let result = tokio::select! { + biased; + _ = stopped.cancelled() => { + record_drop("shutdown"); + break; + }, + result = tokio::time::timeout(ENQUEUE_TIMEOUT, enqueue(&pool, &job)) => result, + }; + match result { + Ok(Ok(())) => { + metrics::counter!("buzz_push_enqueue_total", "result" => "completed") + .increment(1); + } + Ok(Err(error)) => { + record_drop("error"); + tracing::warn!(%error, community = %job.community, "push enqueue failed after message commit"); + } + Err(_) => record_drop("timeout"), + } + } + receiver.close(); + while receiver.try_recv().is_ok() { + record_drop("shutdown"); + } + if tokio::time::timeout(ENQUEUE_TIMEOUT, pool.close()) + .await + .is_err() + { + tracing::warn!("push enqueue pool close exceeded shutdown deadline"); + } + }); + Self { + sender, + cancel, + task: Arc::new(Mutex::new(Some(task))), + } + } + + pub(crate) fn submit(&self, community: CommunityId, event_id: Vec) { + if self.cancel.is_cancelled() { + record_drop("shutdown"); + return; + } + if let Err(error) = self.sender.try_send(Job { + community, + event_id, + #[cfg(test)] + barrier: None, + }) { + record_drop(match error { + mpsc::error::TrySendError::Full(_) => "full", + mpsc::error::TrySendError::Closed(_) => "closed", + }); + } + } + + #[cfg(test)] + pub(crate) async fn flush(&self) { + let (sender, receiver) = tokio::sync::oneshot::channel(); + self.sender + .send(Job { + community: CommunityId::from_uuid(uuid::Uuid::nil()), + event_id: Vec::new(), + barrier: Some(sender), + }) + .await + .expect("producer open"); + receiver.await.expect("producer completed preceding work"); + } + + pub(crate) fn cancel(&self) { + self.cancel.cancel(); + } + + pub(crate) async fn join(&self) { + if let Some(task) = self.task.lock().await.take() { + if let Err(error) = task.await { + tracing::warn!(%error, "push enqueue worker failed"); + } + } + } +} + +pub(crate) fn record_drop(reason: &'static str) { + metrics::counter!("buzz_push_enqueue_total", "result" => reason).increment(1); +} + +async fn enqueue(pool: &PgPool, job: &Job) -> crate::Result<()> { + let mut tx = pool.begin().await?; + // Server-side deadlines also bound work if the client disconnects or the + // task is cancelled. These apply only to the dedicated push connection. + sqlx::query("SELECT set_config('lock_timeout', '500ms', true), set_config('statement_timeout', '1500ms', true)") + .execute(&mut *tx).await?; + crate::deletion::DeletionStore::new(pool.clone()) + .guard_transaction(&mut tx, job.community) + .await?; + sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))") + .bind(crate::push::push_gate_lock_key(job.community)) + .execute(&mut *tx) + .await?; + // Do not enqueue solely because a lease first appeared after the + // message arrived. Updates racing this check may drop a wake, which is + // permitted by the best-effort contract; message storage is authoritative. + sqlx::query( + "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 \ + AND l.expires_at > EXTRACT(EPOCH FROM now())::bigint) \ + ON CONFLICT DO NOTHING", + ) + .bind(job.community.as_uuid()) + .bind(&job.event_id) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn saturation_and_shutdown_never_wait_for_a_receiver() { + let (sender, mut receiver) = mpsc::channel(CAPACITY); + let producer = PushEnqueue { + sender, + cancel: CancellationToken::new(), + task: Arc::new(Mutex::new(None)), + }; + let community = CommunityId::from_uuid(uuid::Uuid::new_v4()); + for _ in 0..CAPACITY + 1 { + producer.submit(community, vec![1; 32]); + } + assert_eq!(receiver.len(), CAPACITY); + receiver.try_recv().unwrap(); + producer.cancel(); + producer.submit(community, vec![2; 32]); + assert_eq!(receiver.len(), CAPACITY - 1); + producer.join().await; + } +} diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 9dd4793d0d..183585e278 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -413,6 +413,9 @@ pub async fn huddle_started_link_exists_in_transaction( /// Insert a Nostr event. Rejects AUTH and ephemeral kinds. /// +/// This raw-pool helper does not start application push production. Serving +/// callers should use [`Db::insert_event`] with a configured [`Db`]. +/// /// Returns `(StoredEvent, was_inserted)` — `was_inserted` is `false` on duplicate. pub async fn insert_event( pool: &PgPool, @@ -2116,6 +2119,7 @@ impl Db { crate::observability::WriterOperation::EventWrite, ) .await?; + tx.set_push_enqueue(self.push_enqueue.clone()); let result = crate::event::insert_event_in_transaction(&mut tx, event, channel_id).await?; if result.1 { crate::insert_mentions_in_transaction(&mut tx, event, channel_id).await?; @@ -2641,6 +2645,7 @@ impl Db { crate::observability::WriterOperation::EventWrite, ) .await?; + tx.set_push_enqueue(self.push_enqueue.clone()); crate::event::acquire_canvas_event_write_lock_if_needed(&mut tx, event, channel_id).await?; let result = crate::event::insert_event_with_thread_metadata_tx( &mut tx, diff --git a/crates/buzz-db/src/store/event_follow_up.rs b/crates/buzz-db/src/store/event_follow_up.rs index 2c79a55579..5219af6062 100644 --- a/crates/buzz-db/src/store/event_follow_up.rs +++ b/crates/buzz-db/src/store/event_follow_up.rs @@ -2,9 +2,9 @@ //! //! Two pieces of work follow every durable event insert: //! -//! - **Push match enqueue.** Push-eligible kinds get a `push_match_queue` row -//! when the community has an eligible lease. This runs right after the -//! INSERT, in the same transaction, and any failure rejects the event. +//! - **Push match enqueue.** Eligible event IDs are staged in memory. Commit +//! submits them to a bounded best-effort worker using a dedicated connection. +//! Push errors, delays, overload, and shutdown can drop wakes, never messages. //! - **Channel TTL refresh.** A channel-scoped event pushes its ephemeral //! channel's `ttl_deadline` forward. [`AdmittedTx`] records the channel and //! runs the refresh as the last statement before COMMIT (see @@ -13,10 +13,9 @@ //! //! This is the application-side replacement for the `events_enqueue_push_match` //! and `events_refresh_channel_ttl` triggers (migrations 0023, 0024, 0040). -//! Both run during the migration window; the push enqueue is idempotent -//! (`ON CONFLICT DO NOTHING`) and a second TTL refresh is harmless. Every -//! production `INSERT INTO events` must call this module, which -//! `tests/observability_source.rs` enforces. +//! Updated writer connections suppress the surviving push trigger. Legacy +//! writers retain transactional enqueue until replaced. Every production +//! `INSERT INTO events` must call this module, enforced by source-policy tests. use std::collections::BTreeSet; @@ -38,79 +37,19 @@ const SQLSTATE_LOCK_NOT_AVAILABLE: &str = "55P03"; /// Follow-up for an event row inserted in an admitted transaction. /// -/// Call only when the INSERT affected a row. Enqueues the push match job now -/// and records the channel for the pre-commit TTL refresh. -/// -/// If the insert ran inside a savepoint that may still roll back, call -/// [`enqueue_push_match`] inside the savepoint and -/// [`AdmittedTx::record_channel_event`] after it is released instead. +/// Call only when the INSERT affected a row, after releasing any savepoint +/// that may roll it back. Stages push work for after commit and records TTL work. pub(crate) async fn after_admitted_insert( tx: &mut AdmittedTx, event_id: &[u8], kind: i32, channel_id: Option, ) -> Result<()> { - let community = tx.community(); - enqueue_push_match(tx.conn(), community, event_id, kind).await?; + tx.record_push_event(event_id, kind); tx.record_channel_event(channel_id, kind); Ok(()) } -/// Enqueue a push match job for a just-inserted event, if its kind is -/// push-eligible and the community has an eligible lease. -/// -/// Takes the push gate SHARED and holds it to transaction end. Lease -/// activations take it EXCLUSIVE (`acquire_push_gate_lock` in `push.rs`), so -/// an event either sees the committed lease or strictly precedes the -/// activation, in which case no wake was owed. The lock must be its own -/// statement: under READ COMMITTED the eligibility check needs a snapshot -/// taken after the lock is granted. -/// -/// Errors propagate and reject the event, as the trigger did. -pub(crate) async fn enqueue_push_match( - conn: &mut PgConnection, - community: CommunityId, - event_id: &[u8], - kind: i32, -) -> Result<()> { - if !PUSH_MATCH_KINDS.contains(&kind) { - return Ok(()); - } - // Missing means a legacy writer during rollout overlap. New writer pools - // always set this explicitly, including replacement connections. - let enabled: bool = sqlx::query_scalar( - "SELECT COALESCE(NULLIF(current_setting('buzz.push_enabled', true), '')::boolean, true)", - ) - .fetch_one(&mut *conn) - .await?; - if !enabled { - return Ok(()); - } - crate::observability::observe_advisory_lock( - crate::observability::LockType::PushGate, - sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))") - .bind(crate::push::push_gate_lock_key(community)) - .execute(&mut *conn), - ) - .await?; - sqlx::query( - "INSERT INTO push_match_queue (community_id, event_id) \ - SELECT $1, $2 \ - WHERE EXISTS ( \ - SELECT 1 FROM push_leases \ - WHERE community_id = $1 \ - AND active \ - AND endpoint_enabled \ - AND expires_at > EXTRACT(EPOCH FROM now())::bigint) \ - ON CONFLICT DO NOTHING", - ) - .bind(community.as_uuid()) - .bind(event_id) - .execute(&mut *conn) - .await?; - Ok(()) -} - /// The channel whose TTL an event of `kind` in `channel_id` refreshes, if any. pub(crate) fn ttl_refresh_channel(channel_id: Option, kind: i32) -> Option { channel_id.filter(|_| kind != KIND_CHANNEL_CREATE) diff --git a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs index a357c63dac..d2eb680519 100644 --- a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs +++ b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs @@ -86,7 +86,11 @@ fn signed_event(keys: &Keys, kind: u16, content: &str) -> Event { .expect("sign test event") } -async fn match_count(pool: &PgPool, community: CommunityId, event: &Event) -> i64 { +async fn match_count(db: &Db, community: CommunityId, event: &Event) -> i64 { + if let Some(producer) = &db.push_enqueue { + producer.flush().await; + } + let pool = db.pool(); sqlx::query_scalar( "SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2", ) @@ -285,6 +289,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { activate_lease(pool, community, &keys, 11, now + 3600).await; assert_rollback(&db, community, channel, &keys).await; + assert_async_isolation(&db, community, &keys).await; for kind in [9, 40002, 45001, 45003] { let deadline_before_first_insert = reset_deadline(pool, community, channel).await; @@ -296,7 +301,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .1 ); assert_eq!(event_count(pool, community, &message).await, 1); - assert_eq!(match_count(pool, community, &message).await, 1); + assert_eq!(match_count(&db, community, &message).await, 1); let deadline_after_first_insert = deadline(pool, community, channel).await; assert!( deadline_after_first_insert > deadline_before_first_insert, @@ -308,7 +313,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .expect("deduplicate event") .1 ); - assert_eq!(match_count(pool, community, &message).await, 1); + assert_eq!(match_count(&db, community, &message).await, 1); assert_eq!( deadline(pool, community, channel).await, deadline_after_first_insert, @@ -323,7 +328,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .expect("insert reaction") .1 ); - assert_eq!(match_count(pool, community, &reaction).await, 0); + assert_eq!(match_count(&db, community, &reaction).await, 0); let no_lease_community = create_community(pool).await; let no_lease_message = signed_event(&keys, 9, "no-lease"); @@ -334,7 +339,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .1 ); assert_eq!( - match_count(pool, no_lease_community, &no_lease_message).await, + match_count(&db, no_lease_community, &no_lease_message).await, 0 ); @@ -386,7 +391,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .1 ); assert_eq!(event_count(pool, predicate_community, &message).await, 1); - assert_eq!(match_count(pool, predicate_community, &message).await, 0); + assert_eq!(match_count(&db, predicate_community, &message).await, 0); } let mut push_gate_holder = pool.begin().await.expect("begin push-gate lock holder"); @@ -401,16 +406,16 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .execute(gated_tx.conn()) .await .expect("bound push-gate lock wait"); - let gated_error = event::insert_event_in_transaction(&mut gated_tx, &gated, None) - .await - .expect_err("push-eligible insert must fail closed when its gate cannot be acquired"); - assert_sqlstate(&gated_error, "55P03"); - gated_tx - .rollback() - .await - .expect("rollback failed push-gate transaction"); - assert_eq!(event_count(pool, community, &gated).await, 0); - assert_eq!(match_count(pool, community, &gated).await, 0); + timeout(Duration::from_secs(5), async { + event::insert_event_in_transaction(&mut gated_tx, &gated, None) + .await + .unwrap(); + gated_tx.commit().await.unwrap(); + }) + .await + .expect("push contention must not delay message commit"); + assert_eq!(event_count(pool, community, &gated).await, 1); + assert_eq!(match_count(&db, community, &gated).await, 0); let ungated = signed_event(&keys, 7, "push-gate-non-eligible-control"); let mut ungated_tx = begin_caller_owned_event_transaction(&db, community).await; @@ -429,7 +434,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .await .expect("commit non-push control while gate held"); assert_eq!(event_count(pool, community, &ungated).await, 1); - assert_eq!(match_count(pool, community, &ungated).await, 0); + assert_eq!(match_count(&db, community, &ungated).await, 0); push_gate_holder .rollback() .await @@ -442,7 +447,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .1 ); assert_eq!(event_count(pool, community, &gated_control).await, 1); - assert_eq!(match_count(pool, community, &gated_control).await, 1); + assert_eq!(match_count(&db, community, &gated_control).await, 1); let rollback_deadline = reset_deadline(pool, community, channel).await; let rolled_back = signed_event(&keys, 9, "rolled-back"); @@ -461,13 +466,13 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .fetch_one(rollback_tx.conn()) .await .expect("read uncommitted follow-up"); - assert_eq!(in_transaction, 1); + assert_eq!(in_transaction, 0, "push work starts only after commit"); rollback_tx .rollback() .await .expect("rollback event and follow-up"); assert_eq!(event_count(pool, community, &rolled_back).await, 0); - assert_eq!(match_count(pool, community, &rolled_back).await, 0); + assert_eq!(match_count(&db, community, &rolled_back).await, 0); assert_eq!(deadline(pool, community, channel).await, rollback_deadline); let before_commit = reset_deadline(pool, community, channel).await; @@ -500,7 +505,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { >= database_time_before_commit + chrono::Duration::seconds(60), "the deferred trigger must refresh TTL at commit" ); - assert_eq!(match_count(pool, community, &committed).await, 1); + assert_eq!(match_count(&db, community, &committed).await, 1); let before_skips = reset_deadline(pool, community, channel).await; let channel_create = signed_event(&keys, 9007, "channel-create"); @@ -511,7 +516,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .1 ); assert_eq!(deadline(pool, community, channel).await, before_skips); - assert_eq!(match_count(pool, community, &channel_create).await, 0); + assert_eq!(match_count(&db, community, &channel_create).await, 0); let channelless = signed_event(&keys, 9, "channel-null"); assert!( db.insert_event(community, &channelless, None) @@ -519,7 +524,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .expect("insert channel-less event") .1 ); - assert_eq!(match_count(pool, community, &channelless).await, 1); + assert_eq!(match_count(&db, community, &channelless).await, 1); let no_ttl_channel = channel_with_ttl(pool, community, &keys.public_key().to_bytes(), None).await; @@ -630,7 +635,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .expect_err("a cancelled TTL refresh must reject the event"); assert_sqlstate(&cancel_error, "57014"); assert_eq!(event_count(pool, community, &cancelled_event).await, 0); - assert_eq!(match_count(pool, community, &cancelled_event).await, 0); + assert_eq!(match_count(&db, community, &cancelled_event).await, 0); assert_eq!( deadline(pool, community, channel).await, before_lock_timeout @@ -640,7 +645,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { .await .expect("release exclusive TTL lock"); assert_eq!(event_count(pool, community, &locked_event).await, 1); - assert_eq!(match_count(pool, community, &locked_event).await, 1); + assert_eq!(match_count(&db, community, &locked_event).await, 1); assert_entry_points(&db, community, channel, &keys).await; } @@ -660,9 +665,25 @@ async fn assert_entry_points(db: &Db, community: CommunityId, channel: Uuid, key .expect("insert through the thread-metadata writer") .1 ); - assert_eq!(match_count(pool, community, &threaded).await, 1); + assert_eq!(match_count(db, community, &threaded).await, 1); assert!(deadline(pool, community, channel).await > before); + // Internal/external-effect completion uses the same post-commit producer. + let lease = db + .deletion_store() + .acquire_serving_write_lease(community, "push-contract", "test", Duration::from_secs(60)) + .await + .unwrap(); + let guarded = signed_event(keys, 40002, "serving-lease-writer"); + db.insert_event_with_serving_write_guard(&lease, &guarded, Some(channel)) + .await + .unwrap(); + assert_eq!(match_count(db, community, &guarded).await, 1); + db.deletion_store() + .release_serving_write_lease(&lease) + .await + .unwrap(); + // Replaceable writer (NIP-16 and NIP-29 discovery state). let before = reset_deadline(pool, community, channel).await; let replaceable = signed_event(keys, 39000, "replaceable-writer"); @@ -672,7 +693,7 @@ async fn assert_entry_points(db: &Db, community: CommunityId, channel: Uuid, key .expect("insert through the replaceable writer") .1 ); - assert_eq!(match_count(pool, community, &replaceable).await, 0); + assert_eq!(match_count(db, community, &replaceable).await, 0); assert!(deadline(pool, community, channel).await > before); // Parameterized writer: the insert runs inside a savepoint, and the @@ -801,7 +822,7 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: & db.insert_event(community, &retained, Some(channel)) .await .unwrap(); - assert_eq!(match_count(pool, community, &retained).await, 1); + assert_eq!(match_count(db, community, &retained).await, 1); let lease_before: serde_json::Value = sqlx::query_scalar("SELECT to_jsonb(p) FROM push_leases p WHERE community_id=$1") .bind(community.as_uuid()) @@ -840,6 +861,7 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: & push_enabled: false, max_connections: 1, min_connections: 0, + lock_timeout_ms: 100, ..DbConfig::default() }) .await @@ -864,18 +886,18 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: & .unwrap(); let message = signed_event(keys, kind, &format!("disabled-{kind}")); timeout( - Duration::from_secs(2), + Duration::from_secs(5), disabled.insert_event(community, &message, Some(channel)), ) .await .expect("disabled write must not wait for the push gate") .unwrap(); assert_eq!(event_count(pool, community, &message).await, 1); - assert_eq!(match_count(pool, community, &message).await, 0); + assert_eq!(match_count(db, community, &message).await, 0); disabled_events.push(message); } holder.rollback().await.unwrap(); - assert_eq!(match_count(pool, community, &retained).await, 1); + assert_eq!(match_count(db, community, &retained).await, 1); let lease_after: serde_json::Value = sqlx::query_scalar("SELECT to_jsonb(p) FROM push_leases p WHERE community_id=$1") .bind(community.as_uuid()) @@ -912,7 +934,16 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: & .insert_event(community, &message, Some(channel)) .await .unwrap(); - assert_eq!(match_count(pool, community, &message).await, 1); + let trigger_exists: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_trigger WHERE tgname='events_enqueue_push_match')", + ) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!( + match_count(db, community, &message).await, + i64::from(trigger_exists) + ); legacy.pool().close().await; // Reactivation is an operator decision about retained queues. Reopening @@ -921,9 +952,137 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: & db.insert_event(community, &message, Some(channel)) .await .unwrap(); - assert_eq!(match_count(pool, community, &message).await, 1); + assert_eq!(match_count(db, community, &message).await, 1); for message in disabled_events { - assert_eq!(match_count(pool, community, &message).await, 0); + assert_eq!(match_count(db, community, &message).await, 0); } disabled.pool().close().await; } + +async fn assert_async_isolation(db: &Db, community: CommunityId, keys: &Keys) { + // A one-connection serving pool proves a stalled push job cannot monopolize + // its only connection. The lock holder uses the independent fixture pool. + let isolated = Db::new(&DbConfig { + database_url: std::env::var("BUZZ_TEST_DATABASE_URL").unwrap(), + max_connections: 1, + min_connections: 0, + lock_timeout_ms: 100, + push_enabled: true, + ..DbConfig::default() + }) + .await + .unwrap(); + let mut holder = db.pool().begin().await.unwrap(); + sqlx::query("LOCK TABLE push_match_queue IN ACCESS EXCLUSIVE MODE") + .execute(&mut *holder) + .await + .unwrap(); + let message = signed_event(keys, 9, "async-queue-lock"); + timeout( + Duration::from_secs(5), + isolated.insert_event(community, &message, None), + ) + .await + .expect("push table lock must not delay storage") + .unwrap(); + assert_eq!(event_count(isolated.pool(), community, &message).await, 1); + isolated.push_enqueue.as_ref().unwrap().flush().await; + // The background statement failed its lock timeout; the message survived. + holder.rollback().await.unwrap(); + assert_eq!(match_count(&isolated, community, &message).await, 0); + assert_eq!(event_count(isolated.pool(), community, &message).await, 1); + let healthy = signed_event(keys, 9, "async-after-error"); + isolated + .insert_event(community, &healthy, None) + .await + .unwrap(); + assert_eq!(match_count(&isolated, community, &healthy).await, 1); + + // If activation finishes while a post-commit job waits, it must not create + // a wake solely for a lease that appeared after the message was received. + let later = create_community(db.pool()).await; + let mut activation = db.pool().begin().await.unwrap(); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(push::push_gate_lock_key(later)) + .execute(&mut *activation) + .await + .unwrap(); + let before_enrollment = signed_event(keys, 9, "before-async-enrollment"); + isolated + .insert_event(later, &before_enrollment, None) + .await + .unwrap(); + sqlx::query( + "INSERT INTO push_leases (community_id, author, installation_id, source_event_id, \ + source_created_at, generation, active, endpoint_enabled, app_profile, endpoint_hash, \ + endpoint_grant, max_class, subscriptions, expires_at, updated_at) \ + SELECT $1, author, installation_id, source_event_id, source_created_at, generation, \ + active, endpoint_enabled, app_profile, endpoint_hash, endpoint_grant, max_class, \ + subscriptions, expires_at, clock_timestamp() FROM push_leases WHERE community_id=$2", + ) + .bind(later.as_uuid()) + .bind(community.as_uuid()) + .execute(&mut *activation) + .await + .unwrap(); + activation.commit().await.unwrap(); + assert_eq!(match_count(&isolated, later, &before_enrollment).await, 0); + assert_eq!( + event_count(isolated.pool(), later, &before_enrollment).await, + 1 + ); + + let mut holder = db.pool().begin().await.unwrap(); + sqlx::query("LOCK TABLE push_match_queue IN ACCESS EXCLUSIVE MODE") + .execute(&mut *holder) + .await + .unwrap(); + let cancelled = signed_event(keys, 9, "async-cancel-inflight"); + isolated + .insert_event(community, &cancelled, None) + .await + .unwrap(); + timeout(Duration::from_secs(2), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_stat_activity \ + WHERE datname=current_database() AND application_name='buzz-push-enqueue' \ + AND wait_event_type='Lock' AND query LIKE 'INSERT INTO push_match_queue%')", + ) + .fetch_one(db.pool()) + .await + .unwrap(); + if waiting { + break; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("observe actual enqueue blocked on the table"); + isolated.cancel_push_enqueue(); + timeout(Duration::from_secs(5), isolated.join_push_enqueue()) + .await + .expect("shutdown cancels in-flight enqueue without waiting for lock timeout"); + holder.rollback().await.unwrap(); + assert_eq!(event_count(isolated.pool(), community, &cancelled).await, 1); + let queued: i64 = sqlx::query_scalar( + "SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2", + ) + .bind(community.as_uuid()) + .bind(cancelled.id.as_bytes().as_slice()) + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(queued, 0); + let after_stop = signed_event(keys, 9, "async-message-after-stop"); + isolated + .insert_event(community, &after_stop, None) + .await + .unwrap(); + assert_eq!( + event_count(isolated.pool(), community, &after_stop).await, + 1 + ); + isolated.pool().close().await; +} diff --git a/crates/buzz-db/src/store/push.rs b/crates/buzz-db/src/store/push.rs index f8a5ce15ee..011a9434ed 100644 --- a/crates/buzz-db/src/store/push.rs +++ b/crates/buzz-db/src/store/push.rs @@ -29,14 +29,11 @@ async fn begin_operation_transaction( Ok(sqlx::Transaction::begin(connection, None).await?) } -/// Namespace for the per-community push-gate advisory lock. Event inserts -/// take it SHARED in `event_follow_up::enqueue_push_match` (and in the -/// `enqueue_push_match_job` trigger from migration 0023 until it is retired); -/// 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. +/// Namespace for the per-community push-gate advisory lock. The post-commit +/// enqueue worker takes it SHARED, while lease transitions take it EXCLUSIVE. +/// The worker checks receipt-time eligibility after obtaining the lock. Legacy +/// trigger writers still hold the shared lock in their event transaction. +/// Distinct key domain from the audit and lease address/author locks. const PUSH_GATE_LOCK_NAMESPACE: &str = "buzz_push_gate:"; /// The push-gate advisory lock key for `community`. Both sides of the lock diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 200d44cb65..1bce5766a4 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -333,15 +333,6 @@ async fn replace_parameterized_event_in_transaction_impl( if insert_result.rows_affected() == 0 { return Ok(false); } - // Inside the savepoint, so a later failure here also drops the job. - // The TTL refresh is recorded only once the savepoint is released. - crate::store::event_follow_up::enqueue_push_match( - tx.conn(), - community_id, - incoming_id, - kind_i32, - ) - .await?; if is_nip_rs { sqlx::query( @@ -390,7 +381,8 @@ async fn replace_parameterized_event_in_transaction_impl( .execute(tx.conn()) .await?; if status == ParameterizedReplaceStatus::Inserted { - tx.record_channel_event(channel_id, kind_i32); + crate::store::event_follow_up::after_admitted_insert(tx, incoming_id, kind_i32, channel_id) + .await?; } Ok(ParameterizedReplaceResult::new( @@ -436,6 +428,7 @@ impl Db { observability::WriterOperation::EventWrite, ) .await?; + tx.set_push_enqueue(self.push_enqueue.clone()); let transaction_timer = observability::TransactionTimer::start( observability::TransactionOperation::ReplaceAddressableEvent, ); @@ -610,6 +603,7 @@ impl Db { observability::WriterOperation::EventWrite, ) .await?; + tx.set_push_enqueue(self.push_enqueue.clone()); let transaction_timer = observability::TransactionTimer::start(TransactionOperation::ReplaceParameterizedEvent); transaction_timer diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 6cef13d6e3..e42c6a4975 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1672,6 +1672,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { ) .await?; state.push_cancel.cancel(); + state.db.cancel_push_enqueue(); + state.db.join_push_enqueue().await; if let Some((matcher, delivery)) = push_workers { for result in [matcher.await, delivery.await] { if let Err(error) = result { diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index f1d9540fb9..ff506fb426 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1589,6 +1589,7 @@ impl AppState { pub fn begin_shutdown(&self) { self.shutting_down.store(true, Ordering::Release); self.push_cancel.cancel(); + self.db.cancel_push_enqueue(); } #[cfg(test)] diff --git a/docs/push-gateway-deployment.md b/docs/push-gateway-deployment.md index f1e2f0ec09..25756e4bdd 100644 --- a/docs/push-gateway-deployment.md +++ b/docs/push-gateway-deployment.md @@ -182,6 +182,7 @@ installation, event, or request identifier as a label. | Metric | Type | Labels | Meaning | |---|---|---|---| | `buzz_push_enabled` | gauge | — | `1` only when the deployment opt-in is active. | +| `buzz_push_enqueue_total` | counter | `result` = `completed` \| `full` \| `transaction_full` \| `closed` \| `error` \| `timeout` \| `shutdown` | Best-effort post-commit enqueue outcomes. | | `buzz_push_match_jobs_total` | counter | `result` = `matched` \| `unmatched` \| `error` \| `context_error` | Accepted message events evaluated by the matcher. | | `buzz_push_match_queue_seconds` | histogram | — | Relay receipt to matcher evaluation latency. | | `buzz_push_wakes_total` | counter | `result` = `enqueued` \| `duplicate` \| `inactive_lease` | Durable wake-enqueue outcomes. | @@ -260,15 +261,17 @@ gateway grants unchanged. Do not stop a shared gateway as part of relay rollback The relay copies the flag into `DbConfig::push_enabled`. Every physical writer connection, including the audit pool and replacement connections, sets the `buzz.push_enabled` PostgreSQL session setting. Other `DbConfig` callers default -to disabled. Both the application producer and the surviving overlap trigger -return before push locks, lease scans or queue writes when this setting is off. +to disabled. Updated writers also set `buzz.push_async_enqueue=on`, which makes +the surviving trigger skip enqueueing regardless of the enabled flag. The +application producer is created only when enabled. Disabled writes perform no +push locks, lease scans or queue writes. The overlap migration updates only an existing trigger function; it does not recreate a retired trigger or function. Deploy that migration before relying on the flag with a surviving trigger. Older images and raw SQL writers with a missing setting retain legacy enqueue behavior; inventory and replace or explicitly disable all such connections before declaring rollback complete. -On SIGTERM/Ctrl-C, the relay cancels matcher and delivery futures immediately, +On SIGTERM/Ctrl-C, the relay cancels enqueue, matcher and delivery futures immediately, before its listener drain. Uncommitted transactions roll back; committed claims remain leased until their normal expiry. No cancellation cleanup deletes or acknowledges queue entries. An HTTP request already accepted by the gateway may @@ -279,9 +282,23 @@ Verify all old writer sessions and workers have exited, eligible messages still store and arrive in chat, no new matcher jobs appear, and gateway attempts stop once in-flight work settles. Inspect enqueue activity as well as queue counts: a stable count alone can conceal additions and removals. Ordinary reads, writes, -authentication and reconnects must remain healthy. Enabled enqueue still runs -inside the message transaction and propagates errors; this change does not -isolate enabled message storage from push failures. +authentication and reconnects must remain healthy. + +When enabled, updated application writers stage at most 256 event IDs per +transaction and submit them only after successful commit using a non-waiting +send. One background worker has a 256-item queue and a dedicated one-connection +pool; it never borrows a message-write connection. Each attempt has a two-second +deadline, a 500 ms lock timeout and a 1.5 second statement timeout. Saturation, +errors, deadlines, process loss and shutdown can drop notifications. They cannot +reject or roll back a committed message, and there is no replay/backfill of +in-memory work. Monitor `buzz_push_enqueue_total` by result (`completed`, `full`, +`transaction_full`, `closed`, `error`, `timeout`, `shutdown`). `completed` includes +jobs with no eligible lease. Database enqueue is conditional on a live event and +an eligible lease last updated no later than event receipt. A concurrent lease +update can suppress a notification under this best-effort contract. + +This isolation requires the overlap migration and updated writers. Old images +and raw SQL writers can still run the legacy transactional trigger until replaced. Pending matcher jobs and pending/claimed delivery jobs remain dormant. Do not process or delete them during urgent rollback. **Before re-enabling**, inspect diff --git a/migrations/0059_push_rollback_gate.sql b/migrations/0059_push_rollback_gate.sql index 3c031610e3..125823b6cb 100644 --- a/migrations/0059_push_rollback_gate.sql +++ b/migrations/0059_push_rollback_gate.sql @@ -14,12 +14,14 @@ BEGIN -- Legacy connections without the setting preserve their existing behavior. -- Updated writer pools always set it, including on reconnect. Check before -- taking push locks or touching leases/queues so disabled chat stays healthy. + IF COALESCE(current_setting('buzz.push_async_enqueue', true), '') = 'on' THEN + RETURN NEW; + END IF; IF NOT COALESCE(NULLIF(current_setting('buzz.push_enabled', true), '')::boolean, true) THEN RETURN NEW; END IF; -- Keep this allowlist identical to the relay's validated NIP-PL descriptor. - -- Centralizing it on the events table covers every durable producer, - -- including internal paths that bypass live dispatch. + -- This compatibility path covers legacy writers until they are replaced. IF NEW.kind IN (9, 40002, 45001, 45003) THEN PERFORM pg_advisory_xact_lock_shared( hashtextextended('buzz_push_gate:' || NEW.community_id::text, 0)); diff --git a/schema/functions/enqueue_push_match_job.sql b/schema/functions/enqueue_push_match_job.sql index ec1e367910..ae085e76ea 100644 --- a/schema/functions/enqueue_push_match_job.sql +++ b/schema/functions/enqueue_push_match_job.sql @@ -8,12 +8,14 @@ BEGIN -- Legacy connections without the setting preserve their existing behavior. -- Updated writer pools always set it, including on reconnect. Check before -- taking push locks or touching leases/queues so disabled chat stays healthy. + IF COALESCE(current_setting('buzz.push_async_enqueue', true), '') = 'on' THEN + RETURN NEW; + END IF; IF NOT COALESCE(NULLIF(current_setting('buzz.push_enabled', true), '')::boolean, true) THEN RETURN NEW; END IF; -- Keep this allowlist identical to the relay's validated NIP-PL descriptor. - -- Centralizing it on the events table covers every durable producer, - -- including internal paths that bypass live dispatch. + -- This compatibility path covers legacy writers until they are replaced. IF NEW.kind IN (9, 40002, 45001, 45003) THEN PERFORM pg_advisory_xact_lock_shared( hashtextextended('buzz_push_gate:' || NEW.community_id::text, 0)); diff --git a/schema/tables/public/push_match_queue.sql b/schema/tables/public/push_match_queue.sql index 809ad6b0c7..423c69b4bf 100644 --- a/schema/tables/public/push_match_queue.sql +++ b/schema/tables/public/push_match_queue.sql @@ -1,7 +1,7 @@ --- Durable event-to-push matching follower. The trigger runs in the event insert --- transaction, so every accepted persistent event has a crash-safe match job and --- rejected/rolled-back events never do. Processing is idempotent through the --- push_wake_outbox endpoint/event unique key. +-- Durable jobs after best-effort post-commit enqueue. Updated application +-- writers may drop push work on overload/failure/shutdown; messages remain +-- authoritative. Legacy writers may still enqueue in their event transaction. +-- Processing is idempotent through the outbox endpoint/event unique key. CREATE TABLE push_match_queue ( community_id UUID NOT NULL REFERENCES communities(id), event_id BYTEA NOT NULL CHECK (length(event_id) = 32),