diff --git a/.env.example b/.env.example index f79eb23f73..8e1fe15939 100644 --- a/.env.example +++ b/.env.example @@ -95,6 +95,8 @@ RELAY_URL=ws://localhost:3000 # NIP-PL mobile push is an explicit deployment opt-in. A gateway URL alone # never enables it. When enabled, the exact HTTP(S) delivery URL is required. +# Rollback requires replacing every enabled writer/worker process. Inspect and +# explicitly decide retained queue disposition before re-enabling (see docs/push-gateway-deployment.md). BUZZ_PUSH_ENABLED=false BUZZ_PUSH_GATEWAY_DELIVERY_URL=https://push.buzz.xyz/v1/deliveries/apns 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/Justfile b/Justfile index c85b620fd8..04c6e46e70 100644 --- a/Justfile +++ b/Justfile @@ -528,7 +528,7 @@ test-unit: # are infra-free; config parses env into a struct and never connects. # `^config::` is anchored so it does not also select nip_fi_config. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::admin::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' # Note on audio::join::tests scope: the full suite is infra-free (no DB, # no Redis). The infra-free audio/FI regression witnesses — bootstrap # ordering barrier, CommitConfirmed arm, pending-close invisibility, 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/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 8acf79ada0..a7376a1e6a 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -705,7 +705,9 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 58); + assert_eq!(migrations.len(), 59); + assert_eq!(migrations[58].version, 59); + assert!(migrations[58].sql.as_str().contains("buzz.push_enabled")); assert_eq!(migrations[55].version, 56); assert!(migrations[55] .sql diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 3369919f1e..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`]). @@ -576,6 +579,9 @@ pub enum DbReadinessOutcome { /// Configuration for the Postgres connection pool. #[derive(Debug, Clone)] pub struct DbConfig { + /// Enable push producers on every writer connection. Defaults to false. + /// Relay startup copies `BUZZ_PUSH_ENABLED`; auxiliary writers stay off. + pub push_enabled: bool, /// Postgres connection URL (usually sourced from `DATABASE_URL`). pub database_url: String, /// Optional read-replica connection URL (usually sourced from @@ -627,6 +633,7 @@ impl Default for DbConfig { /// At 20 main + 5 audit = 25/pod, four relay pods fit within the PG limit. fn default() -> Self { Self { + push_enabled: false, database_url: "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string(), // sadscan:disable np.postgres.1 read_database_url: None, max_connections: 20, @@ -701,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, @@ -726,6 +737,7 @@ impl Db { classify_pool_outcome, record_milestone, DbConnectionStep, DbConnectionStepAttempt, }; + let push_enabled = config.push_enabled; let lock_timeout_ms = config.lock_timeout_ms; let idle_txn_timeout_ms = config.idle_txn_timeout_ms; let statement_timeout_ms = config.statement_timeout_ms; @@ -775,7 +787,9 @@ impl Db { "SELECT set_config('lock_timeout', $1, false), \ 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('default_transaction_read_only', $4, 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()) @@ -785,6 +799,7 @@ impl Db { } else { "off" }) + .bind(if push_enabled { "on" } else { "off" }) .execute(&mut *conn) .await { @@ -935,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, @@ -957,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, @@ -1317,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 @@ -1352,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..40b1dcb52f --- /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?; + // Compare recorded receipt/update times without imposing commit ordering. + // Registration overlapping message arrival may send or suppress a wake. + // This bounded job never scans historical messages on lease activation. + 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 98fc08c29a..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,69 +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(()); - } - 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 10c396dfb7..5a61e288d3 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 @@ -25,6 +25,7 @@ async fn connect(migration_schema: bool) -> Db { Ok(expected_mode) ); let config = DbConfig { + push_enabled: true, database_url, ..DbConfig::default() }; @@ -85,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", ) @@ -251,6 +256,25 @@ async fn apply_arm(pool: &PgPool, arm: Arm) { remaining, 0, "app-only arm must leave no trigger on any partition" ); + sqlx::query("DROP FUNCTION enqueue_push_match_job()") + .execute(pool) + .await + .unwrap(); + sqlx::raw_sql(include_str!( + "../../../../migrations/0059_push_rollback_gate.sql" + )) + .execute(pool) + .await + .unwrap(); + let absent: bool = + sqlx::query_scalar("SELECT to_regprocedure('enqueue_push_match_job()') IS NULL") + .fetch_one(pool) + .await + .unwrap(); + assert!( + absent, + "rollback migration must not resurrect a retired function" + ); } } @@ -264,6 +288,9 @@ async fn assert_contract(migration_schema: bool, arm: Arm) { let now = Utc::now().timestamp(); 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; let message = signed_event(&keys, kind, &format!("allowlisted-{kind}")); @@ -274,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, @@ -286,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, @@ -301,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"); @@ -312,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 ); @@ -364,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"); @@ -379,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; @@ -407,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 @@ -420,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"); @@ -439,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; @@ -478,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"); @@ -489,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) @@ -497,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; @@ -608,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 @@ -618,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; } @@ -638,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"); @@ -650,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 @@ -771,3 +814,276 @@ async fn migration_schema_event_follow_up_contract_dual() { async fn migration_schema_event_follow_up_contract_app_only() { assert_contract(true, Arm::AppOnly).await; } + +// Runs with both schema sources and with/without the overlap trigger. +async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &Keys) { + let pool = db.pool(); + let retained = signed_event(keys, 9, "queued-before-rollback"); + db.insert_event(community, &retained, Some(channel)) + .await + .unwrap(); + 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()) + .fetch_one(pool) + .await + .unwrap(); + + // Include pending, claimed, and completed delivery records: rollback is + // not a queue cleanup or a retention operation. + for (marker, state) in [(81_u8, "pending"), (82, "sending"), (83, "delivered")] { + sqlx::query( + "INSERT INTO push_wake_outbox (community_id, author, installation_id, \ + lease_generation, endpoint_hash, event_id, class, expires_at, state, \ + claim_id, lease_until) \ + SELECT community_id, author, installation_id, generation, endpoint_hash, \ + $2, 'default', expires_at, $3, gen_random_uuid(), now() + interval '30 seconds' \ + FROM push_leases WHERE community_id=$1", + ) + .bind(community.as_uuid()) + .bind(vec![marker; 32]) + .bind(state) + .execute(pool) + .await + .unwrap(); + } + let wakes_before: serde_json::Value = sqlx::query_scalar( + "SELECT jsonb_agg(to_jsonb(w) ORDER BY id) FROM push_wake_outbox w WHERE community_id=$1", + ) + .bind(community.as_uuid()) + .fetch_one(pool) + .await + .unwrap(); + + let disabled = Db::new(&DbConfig { + database_url: std::env::var("BUZZ_TEST_DATABASE_URL").unwrap(), + push_enabled: false, + max_connections: 1, + min_connections: 0, + lock_timeout_ms: 100, + ..DbConfig::default() + }) + .await + .unwrap(); + let mut holder = pool.begin().await.unwrap(); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(push::push_gate_lock_key(community)) + .execute(&mut *holder) + .await + .unwrap(); + let mut disabled_events = Vec::new(); + for kind in [9, 40002, 45001, 45003] { + // Force a fresh physical connection each time, proving after_connect + // reapplies the flag instead of relying on a single startup session. + disabled + .pool() + .acquire() + .await + .unwrap() + .close() + .await + .unwrap(); + let message = signed_event(keys, kind, &format!("disabled-{kind}")); + timeout( + 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(db, community, &message).await, 0); + disabled_events.push(message); + } + holder.rollback().await.unwrap(); + 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()) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!( + lease_before, lease_after, + "rollback preserves leases and subscriptions" + ); + + let wakes_after: serde_json::Value = sqlx::query_scalar( + "SELECT jsonb_agg(to_jsonb(w) ORDER BY id) FROM push_wake_outbox w WHERE community_id=$1", + ) + .bind(community.as_uuid()) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!( + wakes_before, wakes_after, + "rollback preserves all delivery records" + ); + + // An old connection with no GUC retains the legacy producer behavior. + // It must be replaced before operators declare rollback complete. + let legacy_pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&std::env::var("BUZZ_TEST_DATABASE_URL").unwrap()) + .await + .unwrap(); + let legacy = Db::from_pool(legacy_pool); + let message = signed_event(keys, 9, "legacy-overlap-writer"); + legacy + .insert_event(community, &message, Some(channel)) + .await + .unwrap(); + 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 + // an enabled writer produces only new work, with no lease backfill. + let message = signed_event(keys, 9, "new-after-reactivation"); + db.insert_event(community, &message, Some(channel)) + .await + .unwrap(); + assert_eq!(match_count(db, community, &message).await, 1); + for message in disabled_events { + 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); + + // Exercise the timestamp filter with a lease timestamp explicitly newer + // than this message. This is not a production activation-order regression: + // registration overlapping message arrival may send or suppress a wake. + 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/config.rs b/crates/buzz-relay/src/config.rs index ca895b989e..44662bb9e1 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -361,7 +361,7 @@ pub struct Config { /// Used to authenticate internal policy endpoint requests. pub git_hook_hmac_secret: String, - /// Whether NIP-PL push discovery, lease acceptance, matching, and delivery + /// Whether NIP-PL push discovery, lease acceptance, enqueueing, matching, and delivery /// are enabled for this deployment. Defaults to false. pub push_enabled: bool, /// Descriptor key identifier accepted in kind:30350 `exec` tags. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 3df5c428ee..e42c6a4975 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -441,6 +441,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { ); let db_config = DbConfig { + push_enabled: config.push_enabled, database_url: config.database_url.clone(), read_database_url: config.read_database_url.clone(), replica_read_max_age_ms: config.replica_read_max_age_ms, @@ -1214,15 +1215,17 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { // NIP-PL matcher and worker are enabled as one unit behind the explicit // deployment opt-in. The gateway URL alone never enables push. - if state.config.push_enabled { - tokio::spawn(buzz_relay::push_runtime::run_matcher(Arc::clone(&state))); - tokio::spawn(buzz_relay::push_runtime::run_delivery_worker(Arc::clone( + let push_workers = if state.config.push_enabled { + let matcher = tokio::spawn(buzz_relay::push_runtime::run_matcher(Arc::clone(&state))); + let delivery = tokio::spawn(buzz_relay::push_runtime::run_delivery_worker(Arc::clone( &state, ))); info!("NIP-PL push matcher and delivery worker started"); + Some((matcher, delivery)) } else { info!("NIP-PL push disabled by BUZZ_PUSH_ENABLED"); - } + None + }; // Registration cleanup is independent of configured routes. Delivery is // opt-in, while accepted events populate the outbox transactionally. @@ -1668,6 +1671,16 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { jwks_refresh_handle, ) .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 { + tracing::warn!(%error, "push worker failed during shutdown"); + } + } + } state.community_revalidator_cancel.cancel(); state.dependency_sampler_cancel.cancel(); state.dependency_completion_publisher_cancel.cancel(); @@ -3082,6 +3095,11 @@ mod tests { assert_eq!(lock, "500ms"); assert_eq!(idle, "1min"); assert_eq!(statement, "0"); + let push: String = sqlx::query_scalar("SELECT current_setting('buzz.push_enabled')") + .fetch_one(&pool) + .await + .expect("audit pool push setting"); + assert_eq!(push, "off"); let lock_key = i64::from_be_bytes( Uuid::new_v4().as_bytes()[..8] diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index e25e7c0d17..44183a8f1c 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -13,6 +13,24 @@ use tracing::{error, warn}; use crate::{handlers::push_lease::Subscription, nip98::nip98_header, state::AppState}; +// Dropping a SQLx transaction rolls it back; committed claims remain leased +// until expiry. Dropping HTTP cannot recall requests already accepted by the +// gateway. Do not acknowledge, delete, or retry a job merely on cancellation. +async fn run_push_task( + enabled: bool, + cancel: &tokio_util::sync::CancellationToken, + work: impl std::future::Future, +) { + if !enabled { + return; + } + tokio::select! { + biased; + _ = cancel.cancelled() => {}, + _ = work => {}, + } +} + const CLAIM_SECS: i64 = 30; const EVENT_USEFUL_SECS: i64 = 3600; const MAX_ATTEMPTS: i32 = 8; @@ -56,6 +74,15 @@ enum DeliveryResponse { /// 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) { + run_push_task( + state.config.push_enabled, + &state.push_cancel, + matcher_loop(&state), + ) + .await; +} + +async fn matcher_loop(state: &AppState) { let mut idle_delay = IDLE_POLL_FLOOR; let mut last_reap = tokio::time::Instant::now(); loop { @@ -75,7 +102,7 @@ pub async fn run_matcher(state: Arc) { { Ok(Some(batch)) => { idle_delay = IDLE_POLL_FLOOR; - process_match_batch(&state, batch).await; + process_match_batch(state, batch).await; } Ok(None) => { tokio::time::sleep(idle_delay).await; @@ -336,6 +363,15 @@ fn push_filter_authorized_for_event( /// Continuously claim due wakes and deliver them through the push gateway. pub async fn run_delivery_worker(state: Arc) { + run_push_task( + state.config.push_enabled, + &state.push_cancel, + delivery_worker_loop(&state), + ) + .await; +} + +async fn delivery_worker_loop(state: &AppState) { let http = match reqwest::Client::builder() .timeout(state.config.push_gateway_timeout) .build() @@ -359,7 +395,7 @@ pub async fn run_delivery_worker(state: Arc) { Ok(wakes) => { for wake in wakes { found = true; - deliver_one(&state, &http, wake).await; + deliver_one(state, &http, wake).await; } } Err(e) => warn!(%community, "push wake claim failed: {e}"), @@ -774,3 +810,70 @@ mod tests { assert_eq!(bodies[1]["request_id"], request_id.to_string()); } } + +#[cfg(test)] +mod shutdown_tests { + use super::run_push_task; + use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }; + use tokio_util::sync::CancellationToken; + + #[tokio::test] + async fn relay_shutdown_stops_both_worker_entry_points() { + let mut state = crate::state::tests::test_state().await; + let config = &mut Arc::get_mut(&mut state).unwrap().config; + Arc::get_mut(config).unwrap().push_enabled = true; + state.begin_shutdown(); + tokio::time::timeout(std::time::Duration::from_secs(1), async { + tokio::join!( + super::run_matcher(state.clone()), + super::run_delivery_worker(state.clone()) + ); + }) + .await + .expect("both real workers must honor shutdown before touching dependencies"); + } + + #[tokio::test] + async fn disabled_or_cancelled_worker_never_polls_work() { + for enabled in [false, true] { + let cancel = CancellationToken::new(); + if enabled { + cancel.cancel(); + } + run_push_task(enabled, &cancel, async { panic!("worker must not start") }).await; + } + } + + #[tokio::test] + async fn shutdown_drops_in_flight_work_without_waiting_for_it() { + struct OnDrop(Arc); + impl Drop for OnDrop { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + let cancel = CancellationToken::new(); + let worker_cancel = cancel.clone(); + let dropped = Arc::new(AtomicBool::new(false)); + let guard = OnDrop(dropped.clone()); + let (started, ready) = tokio::sync::oneshot::channel(); + let worker = tokio::spawn(async move { + run_push_task(true, &worker_cancel, async move { + let _guard = guard; + started.send(()).unwrap(); + std::future::pending::<()>().await; + }) + .await; + }); + ready.await.unwrap(); + cancel.cancel(); + tokio::time::timeout(std::time::Duration::from_secs(1), worker) + .await + .unwrap() + .unwrap(); + assert!(dropped.load(Ordering::SeqCst)); + } +} diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index be6e9ed7c4..ff506fb426 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1193,6 +1193,8 @@ pub struct AppState { pub community_connections: Arc, /// Stops only the periodic lifecycle revalidator during graceful shutdown. pub community_revalidator_cancel: CancellationToken, + /// Cancels push claims and in-flight delivery on process shutdown. + pub push_cancel: CancellationToken, /// Test/telemetry counter for archive disconnect publication attempts. pub community_disconnect_publish_attempts: Arc, /// Semaphore limiting total concurrent connections. @@ -1479,6 +1481,7 @@ impl AppState { conn_manager: Arc::new(ConnectionManager::new()), community_connections: Arc::new(CommunityConnectionRegistry::new()), community_revalidator_cancel: CancellationToken::new(), + push_cancel: CancellationToken::new(), community_disconnect_publish_attempts: Arc::new(AtomicU64::new(0)), conn_semaphore: Arc::new(Semaphore::new(max_connections)), handler_semaphore: Arc::new(Semaphore::new(max_concurrent_handlers)), @@ -1585,6 +1588,8 @@ impl AppState { /// the readiness gauge on its next request. 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 65a3c8cf3d..9a46027490 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. | @@ -249,12 +250,70 @@ triggering message. Keep fallback-to-channel and placeholder/failure cases as explicit counts in the manual sample until privacy-preserving client telemetry is designed. -Rollback does not require deleting credentials or mutating existing leases. -Set `BUZZ_PUSH_ENABLED=false` on the enabled relays to stop advertisement, lease -acceptance, matching, workers, and new gateway traffic. If the gateway itself -is unhealthy, disable the gateway deployment only after relay delivery is off. -Existing leases and gateway authorities then expire naturally. Adding an App -Store application profile is outside this internal evaluation. +### Relay rollback and reactivation + +Set `BUZZ_PUSH_ENABLED=false` in the approved configuration for **every writer +workload in the affected database scope**, then replace all previously enabled +relay processes. This is a startup setting, not a live configuration reload. +Keep the delivery URL, image, relay keys, schema, leases, client preferences and +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. 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 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 +still produce a notification. The process shutdown deadline remains the hard +backstop; a stalled replacement with old workers alive is incomplete rollback. + +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. + +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 whose recorded update timestamp is no later than event receipt. +These timestamps do not establish transaction commit order. When registration +and message arrival overlap, either sending or suppressing the notification is +acceptable; there is no exact enrollment-order guarantee. A concurrent lease +update, including renewal of an active lease, can also suppress a notification +under this best-effort contract. Activation never scans +historical messages or deliberately backfills older notifications. + +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 +retained work across the full affected scope and record an explicit disposition +decision. Cleanup size, batching and scoped commands are an operator decision; +preserve leases and handle completed records through retention policy. There is +no automatic purge or startup queue interlock: enabling workers resumes eligible +retained work under existing expiry/retry rules. Thus an unchecked nonempty queue +must not be silently reactivated. Messages stored while disabled are not +backfilled on lease activation or re-enablement. ## Helm production inputs diff --git a/migrations/0059_push_rollback_gate.sql b/migrations/0059_push_rollback_gate.sql new file mode 100644 index 0000000000..125823b6cb --- /dev/null +++ b/migrations/0059_push_rollback_gate.sql @@ -0,0 +1,46 @@ +-- Only update a surviving overlap trigger function. Never resurrect the +-- function after the independent trigger-retirement migration has removed it. +DO $migration$ +BEGIN + IF to_regprocedure('enqueue_push_match_job()') IS NOT NULL THEN + EXECUTE $function$ +-- T1b push gate (migrations 0023 and 0059). 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 OR REPLACE FUNCTION enqueue_push_match_job() RETURNS trigger +LANGUAGE plpgsql AS $$ +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. + -- 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)); + 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 +$$; +$function$; + END IF; +END +$migration$; diff --git a/schema/functions/enqueue_push_match_job.sql b/schema/functions/enqueue_push_match_job.sql index 0061fb2e52..ae085e76ea 100644 --- a/schema/functions/enqueue_push_match_job.sql +++ b/schema/functions/enqueue_push_match_job.sql @@ -1,13 +1,21 @@ --- T1b push gate (keep in sync with migrations/0023). Enqueue only when the +-- T1b push gate (migrations 0023 and 0059). 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 + -- 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), diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index fededb6842..4f5ed92ba3 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -172,6 +172,9 @@ run_unit_tests() { # Mirror the relay filters from `just test-unit`: the three handler modules, # storage-snapshot helpers, readiness and router unit suites, and the single # scoped admission regression in state::tests, plus the REQ lifecycle tests. + run_test_step "buzz-relay push runtime tests" \ + cargo test -p buzz-relay --lib push_runtime:: -- --nocapture + run_test_step "buzz-relay channel authorization tests" \ cargo test -p buzz-relay --lib handlers::channel_authz:: -- --nocapture