fix(relay): gate push producers and cancel workers during rollback

Signed-off-by: Tom Brow <tomb@block.xyz>

.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
 

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,

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
                     {

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))")

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;
+}

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.

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]

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<Output = ()>,
+) {
+    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<AppState>) {
+    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<AppState>) {
         {
             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<AppState>) {
+    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<AppState>) {
                         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<AtomicBool>);
+        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));
+    }
+}

crates/buzz-relay/src/state.rs

@@ -1193,6 +1193,8 @@ pub struct AppState {
     pub community_connections: Arc<CommunityConnectionRegistry>,
     /// 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<AtomicU64>,
     /// 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)]

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
 

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$;

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.

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