From 61976c5271aa189218a9fb6680df96d76ba87396 Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Fri, 9 Oct 2026 09:22:33 -0700 Subject: [PATCH 1/5] fix(relay): gate push producers and cancel workers during rollback Signed-off-by: Tom Brow --- .env.example | 2 + Justfile | 2 +- crates/buzz-db/src/runtime/mod.rs | 9 +- crates/buzz-db/src/store/event_follow_up.rs | 10 ++ .../store/event_follow_up_postgres_tests.rs | 156 ++++++++++++++++++ crates/buzz-relay/src/config.rs | 2 +- crates/buzz-relay/src/main.rs | 24 ++- crates/buzz-relay/src/push_runtime.rs | 107 +++++++++++- crates/buzz-relay/src/state.rs | 4 + docs/push-gateway-deployment.md | 49 +++++- migrations/0059_push_rollback_gate.sql | 44 +++++ schema/functions/enqueue_push_match_job.sql | 8 +- scripts/run-tests.sh | 3 + 13 files changed, 404 insertions(+), 16 deletions(-) create mode 100644 migrations/0059_push_rollback_gate.sql diff --git a/.env.example b/.env.example index f79eb23f734..8e1fe159391 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/Justfile b/Justfile index c85b620fd87..04c6e46e700 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/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 3369919f1e4..933ee4b029e 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -576,6 +576,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 +630,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, @@ -726,6 +730,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 +780,8 @@ 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)", ) .bind(lock_timeout_ms.to_string()) .bind(idle_txn_timeout_ms.to_string()) @@ -785,6 +791,7 @@ impl Db { } else { "off" }) + .bind(if push_enabled { "on" } else { "off" }) .execute(&mut *conn) .await { diff --git a/crates/buzz-db/src/store/event_follow_up.rs b/crates/buzz-db/src/store/event_follow_up.rs index 98fc08c29a1..2c79a55579d 100644 --- a/crates/buzz-db/src/store/event_follow_up.rs +++ b/crates/buzz-db/src/store/event_follow_up.rs @@ -76,6 +76,16 @@ pub(crate) async fn enqueue_push_match( 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))") 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 10c396dfb78..a357c63dacd 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() }; @@ -251,6 +252,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 +284,8 @@ 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; + 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}")); @@ -771,3 +793,137 @@ 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(pool, 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, + ..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(2), + 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); + disabled_events.push(message); + } + holder.rollback().await.unwrap(); + assert_eq!(match_count(pool, 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(); + assert_eq!(match_count(pool, community, &message).await, 1); + 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(pool, community, &message).await, 1); + for message in disabled_events { + assert_eq!(match_count(pool, community, &message).await, 0); + } + disabled.pool().close().await; +} diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index ca895b989e2..44662bb9e10 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 3df5c428ee3..6cef13d6e3c 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,14 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { jwks_refresh_handle, ) .await?; + state.push_cancel.cancel(); + 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 +3093,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 e25e7c0d172..44183a8f1c8 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 be6e9ed7c4e..f1d9540fb97 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,7 @@ 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(); } #[cfg(test)] diff --git a/docs/push-gateway-deployment.md b/docs/push-gateway-deployment.md index 65a3c8cf3d3..f1e2f0ec09b 100644 --- a/docs/push-gateway-deployment.md +++ b/docs/push-gateway-deployment.md @@ -249,12 +249,49 @@ 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. Both the application producer and the surviving overlap trigger +return before push locks, lease scans or queue writes when this setting is off. +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, +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. Enabled enqueue still runs +inside the message transaction and propagates errors; this change does not +isolate enabled message storage from push failures. + +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 00000000000..3c031610e30 --- /dev/null +++ b/migrations/0059_push_rollback_gate.sql @@ -0,0 +1,44 @@ +-- 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 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. + 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 0061fb2e526..ec1e3679105 100644 --- a/schema/functions/enqueue_push_match_job.sql +++ b/schema/functions/enqueue_push_match_job.sql @@ -1,10 +1,16 @@ --- 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 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. diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index fededb68421..4f5ed92ba36 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 From 6f330328b840299a6b446ab2e62d85945e9bcef3 Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Fri, 9 Oct 2026 09:28:37 -0700 Subject: [PATCH 2/5] test(db): register rollback migration in embedded schema contract Signed-off-by: Tom Brow --- crates/buzz-db/src/runtime/migration.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 63466f762f4..9766bb0ab79 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -705,7 +705,10 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 57); + assert_eq!(migrations.len(), 58); + // 0058 is reserved by the independently reviewed trigger retirement. + assert_eq!(migrations[57].version, 59); + assert!(migrations[57].sql.as_str().contains("buzz.push_enabled")); assert_eq!(migrations[55].version, 56); assert!(migrations[55] .sql From 3164b25cde5fc5e581d433eb5de01c39a3b0e1c0 Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Fri, 9 Oct 2026 10:06:50 -0700 Subject: [PATCH 3/5] 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 15b190f8aed..b4986b67b40 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 7e080808ac3..80e80921ec8 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 d26981804ff..fb0bb4f41fb 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 933ee4b029e..7be9402da40 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 00000000000..ca726942578 --- /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 9dd4793d0db..183585e2785 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 2c79a55579d..5219af60624 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 a357c63dacd..d2eb6805197 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 f8a5ce15ee0..011a9434eda 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 200d44cb65d..1bce5766a46 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 6cef13d6e3c..e42c6a49756 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 f1d9540fb97..ff506fb426e 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 f1e2f0ec09b..25756e4bddb 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 3c031610e30..125823b6cb6 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 ec1e3679105..ae085e76ea0 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 809ad6b0c77..423c69b4bf3 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), From 9d3d710dc15aa3d0bf0923a9ca51928f109b35cb Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Fri, 9 Oct 2026 12:47:05 -0700 Subject: [PATCH 4/5] docs(push): clarify acceptable enrollment overlap Signed-off-by: Tom Brow --- crates/buzz-db/src/runtime/push_enqueue.rs | 6 +++--- crates/buzz-db/src/store/event_follow_up_postgres_tests.rs | 5 +++-- docs/push-gateway-deployment.md | 7 +++++-- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/crates/buzz-db/src/runtime/push_enqueue.rs b/crates/buzz-db/src/runtime/push_enqueue.rs index ca726942578..40b1dcb52f8 100644 --- a/crates/buzz-db/src/runtime/push_enqueue.rs +++ b/crates/buzz-db/src/runtime/push_enqueue.rs @@ -158,9 +158,9 @@ async fn enqueue(pool: &PgPool, job: &Job) -> crate::Result<()> { .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. + // 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 \ 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 d2eb6805197..5a61e288d36 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 @@ -998,8 +998,9 @@ async fn assert_async_isolation(db: &Db, community: CommunityId, keys: &Keys) { .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. + // 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))") diff --git a/docs/push-gateway-deployment.md b/docs/push-gateway-deployment.md index 25756e4bddb..b6474946b08 100644 --- a/docs/push-gateway-deployment.md +++ b/docs/push-gateway-deployment.md @@ -294,8 +294,11 @@ 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. +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. 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. From 12f1d1e81418fe95649c21e60baf27ed159339a3 Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Fri, 9 Oct 2026 12:54:57 -0700 Subject: [PATCH 5/5] docs(push): retain lease renewal suppression caveat Signed-off-by: Tom Brow --- docs/push-gateway-deployment.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/push-gateway-deployment.md b/docs/push-gateway-deployment.md index b6474946b08..9a460274905 100644 --- a/docs/push-gateway-deployment.md +++ b/docs/push-gateway-deployment.md @@ -297,7 +297,9 @@ 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. Activation never scans +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