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/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/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 3369919f1e..933ee4b029 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 98fc08c29a..2c79a55579 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 10c396dfb7..a357c63dac 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 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..6cef13d6e3 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 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..f1d9540fb9 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 65a3c8cf3d..f1e2f0ec09 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 0000000000..3c031610e3 --- /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 0061fb2e52..ec1e367910 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 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