fix(push): isolate push enqueue and stop workers on rollback (#8216)

Push enqueueing previously ran inside the message transaction, so a push lock or database error could block or reject a message. Configured relay writers now submit bounded best-effort work only after commit. A dedicated connection and bounded queue isolate enqueue waits and failures from message-write connections. Overload, errors, process loss, or shutdown may drop a notification; the committed message remains authoritative. `BUZZ_PUSH_ENABLED=false` disables the application producer and workers. Updated writer connections always bypass any surviving enqueue trigger. The overlap migration updates an existing function without resurrecting retired objects. Apply it before relying on isolation with a surviving trigger, and replace every old enabled process and writer connection: missing settings preserve legacy behavior during overlap. SIGTERM/Ctrl-C cancels enqueue, matcher, and delivery work before listener drain. Durable queues, claims, leases, preferences, and delivery configuration remain intact. Claims expire normally; gateway requests already accepted may still deliver. Reactivation requires an explicit operator decision about retained work. There is no automatic purge, startup interlock, or backfill of messages written while disabled. Built on merged [#8135](https://github.com/block/buzz/pull/8135), including [#8178](https://github.com/block/buzz/pull/8178). Trigger retirement [#8210](https://github.com/block/buzz/pull/8210) remains open; this change is tested with the trigger present and after removal. Operational details are in [the rollback and reactivation guide](https://github.com/block/buzz/blob/relay-push-rollback/docs/push-gateway-deployment.md#relay-rollback-and-reactivation). ### Validation [9d3d710](9d3d710dc1) changes documentation and Rust comments only, with no executable or assertion changes. Formatting and diff checks passed; [12f1d1e](12f1d1e814) additionally restores the reviewer-requested caveat that active lease renewal may suppress a notification. This is also documentation-only; Owner approval was confirmed on 12f1d1e. [a4f602d](a4f602df49) subsequently merges main to resolve a migration-count conflict after main added migration 0058. Both 0058 and rollback migration 0059 are retained; the focused embedded-migrator test, formatting, and applicable desktop pre-push checks passed. Current-head [owner approval](https://github.com/block/buzz/pull/8216#pullrequestreview-5474958334) is confirmed. [CI attempt 2 passed](https://github.com/block/buzz/actions/runs/37985516452/attempts/2), with required checks passing or appropriately skipped. The first attempt failed only the unchanged replica-floor probe test with `MaskedActivity { masked: 1 }`; one failed-jobs retry passed without code changes. All four push contract tests passed. The current-head [advisory security review](https://github.com/block/buzz/actions/runs/37985514346/job/114006457769) has remained in its review step since 20:18:51 UTC without a published result for over 25 minutes. It is stalled under the advisory-review policy, not a clean security review. [Required CI checks passed or were skipped](https://github.com/block/buzz/actions/runs/37964035783) on [3164b25](3164b25cde). The advisory security review was cancelled without a published result. Codex's durable-retry suggestion was [rebutted and resolved](https://github.com/block/buzz/pull/8216#discussion_r4232674980) under the explicit best-effort notification contract. The [lease activation-order observation](https://github.com/block/buzz/pull/8216#discussion_r4233051799) is correct. Under the clarified product requirement, either notification outcome is acceptable during overlapping registration and message arrival. Exact commit ordering is not required; deliberate historical backfill remains prohibited. The bot thread is resolved against that clarification; [Owner review accepted the clarified overlap contract](https://github.com/block/buzz/pull/8216#pullrequestreview-5474720278) on 9d3d710. All 488 PostgreSQL tests passed. The final four-way producer contracts passed on desired/migrated schemas, with/without triggers. Added coverage verifies message commit under push-gate and queue-table locks, recovery after enqueue failure, in-flight enqueue cancellation, the lease timestamp filter (not production activation ordering), serving-lease writers, rollback preservation, reconnects, and reactivation without backfill. Queue saturation and all three relay worker shutdown tests passed. Rust lint and all 17 source-policy checks passed. Full local CI/repository runs encountered the unchanged ACP `keepalive_resets_idle_past_deadline` timing failure. Earlier desktop subprocess timing tests also failed under parallel load and passed as an isolated module. Under Tom's local pre-push exception, this update excluded `rust-tests` and `desktop-tauri-checks`; remaining hooks stayed enabled. Required CI validation remains necessary before merge. A focused run competing with the full database suite hit wall-clock deadlines; tests now use a short database lock timeout to distinguish unwanted push dependencies, and their final isolated run passed. A local process-replacement rehearsal on 3164b25 used the real relay binary and a mock gateway holding a delivery request open. SIGTERM closed that request in under 1 ms; the relay exited in 5.02 seconds. Restarting the same binary with push disabled produced no further gateway requests over 35 seconds, beyond the 30-second claim expiry. The outbox row, lease, and its subscriptions remained byte-for-byte unchanged. This used a seeded outbox and did not exercise APNs or a mobile device. Production writer inventory, shutdown deadline, and retained-queue disposition still require operational validation before rollout. The PR is marked ready for review. The overlap behavior is accepted by the product requirement; owner review accepted that contract. No production deployment has occurred. --------- Signed-off-by: Tom Brow <tomb@block.xyz>
2026-10-27 04:30:52 +00:00 · 2026-10-09 13:48:06 -07:00
parent c1370cc654
commit 5b907b838a
22 changed files with 910 additions and 130 deletions
--- 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

--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1203,6 +1203,7 @@ dependencies = [
 "sqlx",
 "thiserror 2.0.18",
 "tokio",
+ "tokio-util",
 "tracing",
 "uuid",
 ]
--- a/2
+++ b/2
@@ -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,
--- 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 }
--- 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<Uuid>,
+    push_enqueue: Option<super::push_enqueue::PushEnqueue>,
+    push_events: Vec<Vec<u8>>,
 }

 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<super::push_enqueue::PushEnqueue>) {
+        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(())
    }

--- a/crates/buzz-db/src/runtime/migration.rs
+++ b/crates/buzz-db/src/runtime/migration.rs
@@ -705,7 +705,9 @@ mod postgres_tests {
        let mut migrations: Vec<_> = MIGRATOR.iter().collect();
        migrations.sort_by_key(|migration| migration.version);

-        assert_eq!(migrations.len(), 58);
+        assert_eq!(migrations.len(), 59);
+        assert_eq!(migrations[58].version, 59);
+        assert!(migrations[58].sql.as_str().contains("buzz.push_enabled"));
        assert_eq!(migrations[55].version, 56);
        assert!(migrations[55]
            .sql
--- 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<push_enqueue::PushEnqueue>,
    /// Maximum connections configured for this pool (from [`DbConfig::max_connections`]).
    pub(crate) max_connections: u32,
    /// Optional read-replica pool (from [`DbConfig::read_database_url`]).
@@ -576,6 +579,9 @@ pub enum DbReadinessOutcome {
 /// Configuration for the Postgres connection pool.
 #[derive(Debug, Clone)]
 pub struct DbConfig {
+    /// Enable push producers on every writer connection. Defaults to false.
+    /// Relay startup copies `BUZZ_PUSH_ENABLED`; auxiliary writers stay off.
+    pub push_enabled: bool,
    /// Postgres connection URL (usually sourced from `DATABASE_URL`).
    pub database_url: String,
    /// Optional read-replica connection URL (usually sourced from
@@ -627,6 +633,7 @@ impl Default for DbConfig {
    /// At 20 main + 5 audit = 25/pod, four relay pods fit within the PG limit.
    fn default() -> Self {
        Self {
+            push_enabled: false,
            database_url: "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string(), // sadscan:disable np.postgres.1
            read_database_url: None,
            max_connections: 20,
@@ -701,7 +708,11 @@ impl Db {
        let replica_read_max_age = read_budget_from_ms(config.replica_read_max_age_ms);
        let usage_metrics_replica_max_age =
            read_budget_from_ms(config.usage_metrics_replica_max_age_ms);
+        let push_enqueue = config
+            .push_enabled
+            .then(|| push_enqueue::PushEnqueue::new(&pool));
        Ok(Self {
+            push_enqueue,
            pool,
            max_connections: config.max_connections,
            read_pool,
@@ -726,6 +737,7 @@ impl Db {
            classify_pool_outcome, record_milestone, DbConnectionStep, DbConnectionStepAttempt,
        };

+        let push_enabled = config.push_enabled;
        let lock_timeout_ms = config.lock_timeout_ms;
        let idle_txn_timeout_ms = config.idle_txn_timeout_ms;
        let statement_timeout_ms = config.statement_timeout_ms;
@@ -775,7 +787,9 @@ impl Db {
                        "SELECT set_config('lock_timeout', $1, false), \
                                set_config('idle_in_transaction_session_timeout', $2, false), \
                                set_config('statement_timeout', $3, false), \
-                                set_config('default_transaction_read_only', $4, false)",
+                                set_config('default_transaction_read_only', $4, false), \
+                                set_config('buzz.push_enabled', $5, false), \
+                                set_config('buzz.push_async_enqueue', 'on', false)",
                    )
                    .bind(lock_timeout_ms.to_string())
                    .bind(idle_txn_timeout_ms.to_string())
@@ -785,6 +799,7 @@ impl Db {
                    } else {
                        "off"
                    })
+                    .bind(if push_enabled { "on" } else { "off" })
                    .execute(&mut *conn)
                    .await
                    {
@@ -935,8 +950,12 @@ impl Db {
    }

    /// Creates a `Db` from an existing `PgPool` (useful in tests).
+    ///
+    /// Does not start a push producer or configure sessions. Serving writers
+    /// that need push must use [`Db::new`] with `push_enabled` instead.
    pub fn from_pool(pool: PgPool) -> Self {
        Self {
+            push_enqueue: None,
            max_connections: pool.options().get_max_connections(),
            read_max_connections: pool.options().get_max_connections(),
            pool,
@@ -957,6 +976,7 @@ impl Db {
    /// [`Db::fence`]).
    pub fn from_pools(pool: PgPool, read_pool: PgPool) -> Self {
        Self {
+            push_enqueue: None,
            max_connections: pool.options().get_max_connections(),
            read_max_connections: read_pool.options().get_max_connections(),
            pool,
@@ -1317,12 +1337,28 @@ impl Db {
        &self,
        community: CommunityId,
    ) -> Result<AdmittedTx> {
-        begin_community_event_write_transaction_with_legacy_metrics(
+        let mut tx = begin_community_event_write_transaction_with_legacy_metrics(
            &self.pool,
            community,
            observability::WriterOperation::EventWrite,
        )
-        .await
+        .await?;
+        tx.set_push_enqueue(self.push_enqueue.clone());
+        Ok(tx)
+    }
+
+    /// Stop accepting best-effort push enqueue work and cancel in-flight work.
+    pub fn cancel_push_enqueue(&self) {
+        if let Some(producer) = &self.push_enqueue {
+            producer.cancel();
+        }
+    }
+
+    /// Wait for the cancelled producer to release its dedicated connection.
+    pub async fn join_push_enqueue(&self) {
+        if let Some(producer) = &self.push_enqueue {
+            producer.join().await;
+        }
    }

    /// Insert an event while holding and validating an admitted serving-write
@@ -1352,6 +1388,7 @@ impl Db {
        let tx = sqlx::Transaction::begin(connection, None).await?;
        let mut tx =
            AdmittedTx::admit_with_serving_lease(tx, &self.deletion_store(), lease).await?;
+        tx.set_push_enqueue(self.push_enqueue.clone());
        event::acquire_canvas_event_write_lock_if_needed(&mut tx, event, channel_id).await?;
        let result =
            event::insert_event_with_thread_metadata_tx(&mut tx, event, channel_id, None).await?;
--- a/crates/buzz-db/src/runtime/push_enqueue.rs
+++ 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<Job>,
+    cancel: CancellationToken,
+    task: Arc<Mutex<Option<JoinHandle<()>>>>,
+}
+
+#[derive(Debug)]
+struct Job {
+    community: CommunityId,
+    event_id: Vec<u8>,
+    #[cfg(test)]
+    barrier: Option<tokio::sync::oneshot::Sender<()>>,
+}
+
+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::<Job>(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<u8>) {
+        if self.cancel.is_cancelled() {
+            record_drop("shutdown");
+            return;
+        }
+        if let Err(error) = self.sender.try_send(Job {
+            community,
+            event_id,
+            #[cfg(test)]
+            barrier: None,
+        }) {
+            record_drop(match error {
+                mpsc::error::TrySendError::Full(_) => "full",
+                mpsc::error::TrySendError::Closed(_) => "closed",
+            });
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) async fn flush(&self) {
+        let (sender, receiver) = tokio::sync::oneshot::channel();
+        self.sender
+            .send(Job {
+                community: CommunityId::from_uuid(uuid::Uuid::nil()),
+                event_id: Vec::new(),
+                barrier: Some(sender),
+            })
+            .await
+            .expect("producer open");
+        receiver.await.expect("producer completed preceding work");
+    }
+
+    pub(crate) fn cancel(&self) {
+        self.cancel.cancel();
+    }
+
+    pub(crate) async fn join(&self) {
+        if let Some(task) = self.task.lock().await.take() {
+            if let Err(error) = task.await {
+                tracing::warn!(%error, "push enqueue worker failed");
+            }
+        }
+    }
+}
+
+pub(crate) fn record_drop(reason: &'static str) {
+    metrics::counter!("buzz_push_enqueue_total", "result" => reason).increment(1);
+}
+
+async fn enqueue(pool: &PgPool, job: &Job) -> crate::Result<()> {
+    let mut tx = pool.begin().await?;
+    // Server-side deadlines also bound work if the client disconnects or the
+    // task is cancelled. These apply only to the dedicated push connection.
+    sqlx::query("SELECT set_config('lock_timeout', '500ms', true), set_config('statement_timeout', '1500ms', true)")
+        .execute(&mut *tx).await?;
+    crate::deletion::DeletionStore::new(pool.clone())
+        .guard_transaction(&mut tx, job.community)
+        .await?;
+    sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))")
+        .bind(crate::push::push_gate_lock_key(job.community))
+        .execute(&mut *tx)
+        .await?;
+    // Compare recorded receipt/update times without imposing commit ordering.
+    // Registration overlapping message arrival may send or suppress a wake.
+    // This bounded job never scans historical messages on lease activation.
+    sqlx::query(
+        "INSERT INTO push_match_queue (community_id, event_id) \
+         SELECT $1, $2 FROM events e \
+         WHERE e.community_id = $1 AND e.id = $2 AND e.deleted_at IS NULL \
+           AND community_write_allowed($1) \
+           AND EXISTS (SELECT 1 FROM push_leases l \
+             WHERE l.community_id = $1 AND l.active AND l.endpoint_enabled \
+               AND l.updated_at <= e.received_at \
+               AND l.expires_at > EXTRACT(EPOCH FROM now())::bigint) \
+         ON CONFLICT DO NOTHING",
+    )
+    .bind(job.community.as_uuid())
+    .bind(&job.event_id)
+    .execute(&mut *tx)
+    .await?;
+    tx.commit().await?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn saturation_and_shutdown_never_wait_for_a_receiver() {
+        let (sender, mut receiver) = mpsc::channel(CAPACITY);
+        let producer = PushEnqueue {
+            sender,
+            cancel: CancellationToken::new(),
+            task: Arc::new(Mutex::new(None)),
+        };
+        let community = CommunityId::from_uuid(uuid::Uuid::new_v4());
+        for _ in 0..CAPACITY + 1 {
+            producer.submit(community, vec![1; 32]);
+        }
+        assert_eq!(receiver.len(), CAPACITY);
+        receiver.try_recv().unwrap();
+        producer.cancel();
+        producer.submit(community, vec![2; 32]);
+        assert_eq!(receiver.len(), CAPACITY - 1);
+        producer.join().await;
+    }
+}
--- 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,
--- a/crates/buzz-db/src/store/event_follow_up.rs
+++ b/crates/buzz-db/src/store/event_follow_up.rs
@@ -2,9 +2,9 @@
 //!
 //! Two pieces of work follow every durable event insert:
 //!
-//! - **Push match enqueue.** Push-eligible kinds get a `push_match_queue` row
-//!   when the community has an eligible lease. This runs right after the
-//!   INSERT, in the same transaction, and any failure rejects the event.
+//! - **Push match enqueue.** Eligible event IDs are staged in memory. Commit
+//!   submits them to a bounded best-effort worker using a dedicated connection.
+//!   Push errors, delays, overload, and shutdown can drop wakes, never messages.
 //! - **Channel TTL refresh.** A channel-scoped event pushes its ephemeral
 //!   channel's `ttl_deadline` forward. [`AdmittedTx`] records the channel and
 //!   runs the refresh as the last statement before COMMIT (see
@@ -13,10 +13,9 @@
 //!
 //! This is the application-side replacement for the `events_enqueue_push_match`
 //! and `events_refresh_channel_ttl` triggers (migrations 0023, 0024, 0040).
-//! Both run during the migration window; the push enqueue is idempotent
-//! (`ON CONFLICT DO NOTHING`) and a second TTL refresh is harmless. Every
-//! production `INSERT INTO events` must call this module, which
-//! `tests/observability_source.rs` enforces.
+//! Updated writer connections suppress the surviving push trigger. Legacy
+//! writers retain transactional enqueue until replaced. Every production
+//! `INSERT INTO events` must call this module, enforced by source-policy tests.

 use std::collections::BTreeSet;

@@ -38,69 +37,19 @@ const SQLSTATE_LOCK_NOT_AVAILABLE: &str = "55P03";

 /// Follow-up for an event row inserted in an admitted transaction.
 ///
-/// Call only when the INSERT affected a row. Enqueues the push match job now
-/// and records the channel for the pre-commit TTL refresh.
-///
-/// If the insert ran inside a savepoint that may still roll back, call
-/// [`enqueue_push_match`] inside the savepoint and
-/// [`AdmittedTx::record_channel_event`] after it is released instead.
+/// Call only when the INSERT affected a row, after releasing any savepoint
+/// that may roll it back. Stages push work for after commit and records TTL work.
 pub(crate) async fn after_admitted_insert(
    tx: &mut AdmittedTx,
    event_id: &[u8],
    kind: i32,
    channel_id: Option<Uuid>,
 ) -> Result<()> {
-    let community = tx.community();
-    enqueue_push_match(tx.conn(), community, event_id, kind).await?;
+    tx.record_push_event(event_id, kind);
    tx.record_channel_event(channel_id, kind);
    Ok(())
 }

-/// Enqueue a push match job for a just-inserted event, if its kind is
-/// push-eligible and the community has an eligible lease.
-///
-/// Takes the push gate SHARED and holds it to transaction end. Lease
-/// activations take it EXCLUSIVE (`acquire_push_gate_lock` in `push.rs`), so
-/// an event either sees the committed lease or strictly precedes the
-/// activation, in which case no wake was owed. The lock must be its own
-/// statement: under READ COMMITTED the eligibility check needs a snapshot
-/// taken after the lock is granted.
-///
-/// Errors propagate and reject the event, as the trigger did.
-pub(crate) async fn enqueue_push_match(
-    conn: &mut PgConnection,
-    community: CommunityId,
-    event_id: &[u8],
-    kind: i32,
-) -> Result<()> {
-    if !PUSH_MATCH_KINDS.contains(&kind) {
-        return Ok(());
-    }
-    crate::observability::observe_advisory_lock(
-        crate::observability::LockType::PushGate,
-        sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))")
-            .bind(crate::push::push_gate_lock_key(community))
-            .execute(&mut *conn),
-    )
-    .await?;
-    sqlx::query(
-        "INSERT INTO push_match_queue (community_id, event_id) \
-         SELECT $1, $2 \
-         WHERE EXISTS ( \
-             SELECT 1 FROM push_leases \
-             WHERE community_id = $1 \
-               AND active \
-               AND endpoint_enabled \
-               AND expires_at > EXTRACT(EPOCH FROM now())::bigint) \
-         ON CONFLICT DO NOTHING",
-    )
-    .bind(community.as_uuid())
-    .bind(event_id)
-    .execute(&mut *conn)
-    .await?;
-    Ok(())
-}
-
 /// The channel whose TTL an event of `kind` in `channel_id` refreshes, if any.
 pub(crate) fn ttl_refresh_channel(channel_id: Option<Uuid>, kind: i32) -> Option<Uuid> {
    channel_id.filter(|_| kind != KIND_CHANNEL_CREATE)
--- a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs
+++ b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs
@@ -25,6 +25,7 @@ async fn connect(migration_schema: bool) -> Db {
        Ok(expected_mode)
    );
    let config = DbConfig {
+        push_enabled: true,
        database_url,
        ..DbConfig::default()
    };
@@ -85,7 +86,11 @@ fn signed_event(keys: &Keys, kind: u16, content: &str) -> Event {
        .expect("sign test event")
 }

-async fn match_count(pool: &PgPool, community: CommunityId, event: &Event) -> i64 {
+async fn match_count(db: &Db, community: CommunityId, event: &Event) -> i64 {
+    if let Some(producer) = &db.push_enqueue {
+        producer.flush().await;
+    }
+    let pool = db.pool();
    sqlx::query_scalar(
        "SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2",
    )
@@ -251,6 +256,25 @@ async fn apply_arm(pool: &PgPool, arm: Arm) {
            remaining, 0,
            "app-only arm must leave no trigger on any partition"
        );
+        sqlx::query("DROP FUNCTION enqueue_push_match_job()")
+            .execute(pool)
+            .await
+            .unwrap();
+        sqlx::raw_sql(include_str!(
+            "../../../../migrations/0059_push_rollback_gate.sql"
+        ))
+        .execute(pool)
+        .await
+        .unwrap();
+        let absent: bool =
+            sqlx::query_scalar("SELECT to_regprocedure('enqueue_push_match_job()') IS NULL")
+                .fetch_one(pool)
+                .await
+                .unwrap();
+        assert!(
+            absent,
+            "rollback migration must not resurrect a retired function"
+        );
    }
 }

@@ -264,6 +288,9 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
    let now = Utc::now().timestamp();
    activate_lease(pool, community, &keys, 11, now + 3600).await;

+    assert_rollback(&db, community, channel, &keys).await;
+    assert_async_isolation(&db, community, &keys).await;
+
    for kind in [9, 40002, 45001, 45003] {
        let deadline_before_first_insert = reset_deadline(pool, community, channel).await;
        let message = signed_event(&keys, kind, &format!("allowlisted-{kind}"));
@@ -274,7 +301,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
                .1
        );
        assert_eq!(event_count(pool, community, &message).await, 1);
-        assert_eq!(match_count(pool, community, &message).await, 1);
+        assert_eq!(match_count(&db, community, &message).await, 1);
        let deadline_after_first_insert = deadline(pool, community, channel).await;
        assert!(
            deadline_after_first_insert > deadline_before_first_insert,
@@ -286,7 +313,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
                .expect("deduplicate event")
                .1
        );
-        assert_eq!(match_count(pool, community, &message).await, 1);
+        assert_eq!(match_count(&db, community, &message).await, 1);
        assert_eq!(
            deadline(pool, community, channel).await,
            deadline_after_first_insert,
@@ -301,7 +328,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
            .expect("insert reaction")
            .1
    );
-    assert_eq!(match_count(pool, community, &reaction).await, 0);
+    assert_eq!(match_count(&db, community, &reaction).await, 0);

    let no_lease_community = create_community(pool).await;
    let no_lease_message = signed_event(&keys, 9, "no-lease");
@@ -312,7 +339,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
            .1
    );
    assert_eq!(
-        match_count(pool, no_lease_community, &no_lease_message).await,
+        match_count(&db, no_lease_community, &no_lease_message).await,
        0
    );

@@ -364,7 +391,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
                .1
        );
        assert_eq!(event_count(pool, predicate_community, &message).await, 1);
-        assert_eq!(match_count(pool, predicate_community, &message).await, 0);
+        assert_eq!(match_count(&db, predicate_community, &message).await, 0);
    }

    let mut push_gate_holder = pool.begin().await.expect("begin push-gate lock holder");
@@ -379,16 +406,16 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
        .execute(gated_tx.conn())
        .await
        .expect("bound push-gate lock wait");
-    let gated_error = event::insert_event_in_transaction(&mut gated_tx, &gated, None)
-        .await
-        .expect_err("push-eligible insert must fail closed when its gate cannot be acquired");
-    assert_sqlstate(&gated_error, "55P03");
-    gated_tx
-        .rollback()
-        .await
-        .expect("rollback failed push-gate transaction");
-    assert_eq!(event_count(pool, community, &gated).await, 0);
-    assert_eq!(match_count(pool, community, &gated).await, 0);
+    timeout(Duration::from_secs(5), async {
+        event::insert_event_in_transaction(&mut gated_tx, &gated, None)
+            .await
+            .unwrap();
+        gated_tx.commit().await.unwrap();
+    })
+    .await
+    .expect("push contention must not delay message commit");
+    assert_eq!(event_count(pool, community, &gated).await, 1);
+    assert_eq!(match_count(&db, community, &gated).await, 0);

    let ungated = signed_event(&keys, 7, "push-gate-non-eligible-control");
    let mut ungated_tx = begin_caller_owned_event_transaction(&db, community).await;
@@ -407,7 +434,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
        .await
        .expect("commit non-push control while gate held");
    assert_eq!(event_count(pool, community, &ungated).await, 1);
-    assert_eq!(match_count(pool, community, &ungated).await, 0);
+    assert_eq!(match_count(&db, community, &ungated).await, 0);
    push_gate_holder
        .rollback()
        .await
@@ -420,7 +447,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
            .1
    );
    assert_eq!(event_count(pool, community, &gated_control).await, 1);
-    assert_eq!(match_count(pool, community, &gated_control).await, 1);
+    assert_eq!(match_count(&db, community, &gated_control).await, 1);

    let rollback_deadline = reset_deadline(pool, community, channel).await;
    let rolled_back = signed_event(&keys, 9, "rolled-back");
@@ -439,13 +466,13 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
    .fetch_one(rollback_tx.conn())
    .await
    .expect("read uncommitted follow-up");
-    assert_eq!(in_transaction, 1);
+    assert_eq!(in_transaction, 0, "push work starts only after commit");
    rollback_tx
        .rollback()
        .await
        .expect("rollback event and follow-up");
    assert_eq!(event_count(pool, community, &rolled_back).await, 0);
-    assert_eq!(match_count(pool, community, &rolled_back).await, 0);
+    assert_eq!(match_count(&db, community, &rolled_back).await, 0);
    assert_eq!(deadline(pool, community, channel).await, rollback_deadline);

    let before_commit = reset_deadline(pool, community, channel).await;
@@ -478,7 +505,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
            >= database_time_before_commit + chrono::Duration::seconds(60),
        "the deferred trigger must refresh TTL at commit"
    );
-    assert_eq!(match_count(pool, community, &committed).await, 1);
+    assert_eq!(match_count(&db, community, &committed).await, 1);

    let before_skips = reset_deadline(pool, community, channel).await;
    let channel_create = signed_event(&keys, 9007, "channel-create");
@@ -489,7 +516,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
            .1
    );
    assert_eq!(deadline(pool, community, channel).await, before_skips);
-    assert_eq!(match_count(pool, community, &channel_create).await, 0);
+    assert_eq!(match_count(&db, community, &channel_create).await, 0);
    let channelless = signed_event(&keys, 9, "channel-null");
    assert!(
        db.insert_event(community, &channelless, None)
@@ -497,7 +524,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
            .expect("insert channel-less event")
            .1
    );
-    assert_eq!(match_count(pool, community, &channelless).await, 1);
+    assert_eq!(match_count(&db, community, &channelless).await, 1);

    let no_ttl_channel =
        channel_with_ttl(pool, community, &keys.public_key().to_bytes(), None).await;
@@ -608,7 +635,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
        .expect_err("a cancelled TTL refresh must reject the event");
    assert_sqlstate(&cancel_error, "57014");
    assert_eq!(event_count(pool, community, &cancelled_event).await, 0);
-    assert_eq!(match_count(pool, community, &cancelled_event).await, 0);
+    assert_eq!(match_count(&db, community, &cancelled_event).await, 0);
    assert_eq!(
        deadline(pool, community, channel).await,
        before_lock_timeout
@@ -618,7 +645,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
        .await
        .expect("release exclusive TTL lock");
    assert_eq!(event_count(pool, community, &locked_event).await, 1);
-    assert_eq!(match_count(pool, community, &locked_event).await, 1);
+    assert_eq!(match_count(&db, community, &locked_event).await, 1);

    assert_entry_points(&db, community, channel, &keys).await;
 }
@@ -638,9 +665,25 @@ async fn assert_entry_points(db: &Db, community: CommunityId, channel: Uuid, key
            .expect("insert through the thread-metadata writer")
            .1
    );
-    assert_eq!(match_count(pool, community, &threaded).await, 1);
+    assert_eq!(match_count(db, community, &threaded).await, 1);
    assert!(deadline(pool, community, channel).await > before);

+    // Internal/external-effect completion uses the same post-commit producer.
+    let lease = db
+        .deletion_store()
+        .acquire_serving_write_lease(community, "push-contract", "test", Duration::from_secs(60))
+        .await
+        .unwrap();
+    let guarded = signed_event(keys, 40002, "serving-lease-writer");
+    db.insert_event_with_serving_write_guard(&lease, &guarded, Some(channel))
+        .await
+        .unwrap();
+    assert_eq!(match_count(db, community, &guarded).await, 1);
+    db.deletion_store()
+        .release_serving_write_lease(&lease)
+        .await
+        .unwrap();
+
    // Replaceable writer (NIP-16 and NIP-29 discovery state).
    let before = reset_deadline(pool, community, channel).await;
    let replaceable = signed_event(keys, 39000, "replaceable-writer");
@@ -650,7 +693,7 @@ async fn assert_entry_points(db: &Db, community: CommunityId, channel: Uuid, key
            .expect("insert through the replaceable writer")
            .1
    );
-    assert_eq!(match_count(pool, community, &replaceable).await, 0);
+    assert_eq!(match_count(db, community, &replaceable).await, 0);
    assert!(deadline(pool, community, channel).await > before);

    // Parameterized writer: the insert runs inside a savepoint, and the
@@ -771,3 +814,276 @@ async fn migration_schema_event_follow_up_contract_dual() {
 async fn migration_schema_event_follow_up_contract_app_only() {
    assert_contract(true, Arm::AppOnly).await;
 }
+
+// Runs with both schema sources and with/without the overlap trigger.
+async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &Keys) {
+    let pool = db.pool();
+    let retained = signed_event(keys, 9, "queued-before-rollback");
+    db.insert_event(community, &retained, Some(channel))
+        .await
+        .unwrap();
+    assert_eq!(match_count(db, community, &retained).await, 1);
+    let lease_before: serde_json::Value =
+        sqlx::query_scalar("SELECT to_jsonb(p) FROM push_leases p WHERE community_id=$1")
+            .bind(community.as_uuid())
+            .fetch_one(pool)
+            .await
+            .unwrap();
+
+    // Include pending, claimed, and completed delivery records: rollback is
+    // not a queue cleanup or a retention operation.
+    for (marker, state) in [(81_u8, "pending"), (82, "sending"), (83, "delivered")] {
+        sqlx::query(
+            "INSERT INTO push_wake_outbox (community_id, author, installation_id, \
+             lease_generation, endpoint_hash, event_id, class, expires_at, state, \
+             claim_id, lease_until) \
+             SELECT community_id, author, installation_id, generation, endpoint_hash, \
+             $2, 'default', expires_at, $3, gen_random_uuid(), now() + interval '30 seconds' \
+             FROM push_leases WHERE community_id=$1",
+        )
+        .bind(community.as_uuid())
+        .bind(vec![marker; 32])
+        .bind(state)
+        .execute(pool)
+        .await
+        .unwrap();
+    }
+    let wakes_before: serde_json::Value = sqlx::query_scalar(
+        "SELECT jsonb_agg(to_jsonb(w) ORDER BY id) FROM push_wake_outbox w WHERE community_id=$1",
+    )
+    .bind(community.as_uuid())
+    .fetch_one(pool)
+    .await
+    .unwrap();
+
+    let disabled = Db::new(&DbConfig {
+        database_url: std::env::var("BUZZ_TEST_DATABASE_URL").unwrap(),
+        push_enabled: false,
+        max_connections: 1,
+        min_connections: 0,
+        lock_timeout_ms: 100,
+        ..DbConfig::default()
+    })
+    .await
+    .unwrap();
+    let mut holder = pool.begin().await.unwrap();
+    sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+        .bind(push::push_gate_lock_key(community))
+        .execute(&mut *holder)
+        .await
+        .unwrap();
+    let mut disabled_events = Vec::new();
+    for kind in [9, 40002, 45001, 45003] {
+        // Force a fresh physical connection each time, proving after_connect
+        // reapplies the flag instead of relying on a single startup session.
+        disabled
+            .pool()
+            .acquire()
+            .await
+            .unwrap()
+            .close()
+            .await
+            .unwrap();
+        let message = signed_event(keys, kind, &format!("disabled-{kind}"));
+        timeout(
+            Duration::from_secs(5),
+            disabled.insert_event(community, &message, Some(channel)),
+        )
+        .await
+        .expect("disabled write must not wait for the push gate")
+        .unwrap();
+        assert_eq!(event_count(pool, community, &message).await, 1);
+        assert_eq!(match_count(db, community, &message).await, 0);
+        disabled_events.push(message);
+    }
+    holder.rollback().await.unwrap();
+    assert_eq!(match_count(db, community, &retained).await, 1);
+    let lease_after: serde_json::Value =
+        sqlx::query_scalar("SELECT to_jsonb(p) FROM push_leases p WHERE community_id=$1")
+            .bind(community.as_uuid())
+            .fetch_one(pool)
+            .await
+            .unwrap();
+    assert_eq!(
+        lease_before, lease_after,
+        "rollback preserves leases and subscriptions"
+    );
+
+    let wakes_after: serde_json::Value = sqlx::query_scalar(
+        "SELECT jsonb_agg(to_jsonb(w) ORDER BY id) FROM push_wake_outbox w WHERE community_id=$1",
+    )
+    .bind(community.as_uuid())
+    .fetch_one(pool)
+    .await
+    .unwrap();
+    assert_eq!(
+        wakes_before, wakes_after,
+        "rollback preserves all delivery records"
+    );
+
+    // An old connection with no GUC retains the legacy producer behavior.
+    // It must be replaced before operators declare rollback complete.
+    let legacy_pool = sqlx::postgres::PgPoolOptions::new()
+        .max_connections(1)
+        .connect(&std::env::var("BUZZ_TEST_DATABASE_URL").unwrap())
+        .await
+        .unwrap();
+    let legacy = Db::from_pool(legacy_pool);
+    let message = signed_event(keys, 9, "legacy-overlap-writer");
+    legacy
+        .insert_event(community, &message, Some(channel))
+        .await
+        .unwrap();
+    let trigger_exists: bool = sqlx::query_scalar(
+        "SELECT EXISTS (SELECT 1 FROM pg_trigger WHERE tgname='events_enqueue_push_match')",
+    )
+    .fetch_one(pool)
+    .await
+    .unwrap();
+    assert_eq!(
+        match_count(db, community, &message).await,
+        i64::from(trigger_exists)
+    );
+    legacy.pool().close().await;
+
+    // Reactivation is an operator decision about retained queues. Reopening
+    // an enabled writer produces only new work, with no lease backfill.
+    let message = signed_event(keys, 9, "new-after-reactivation");
+    db.insert_event(community, &message, Some(channel))
+        .await
+        .unwrap();
+    assert_eq!(match_count(db, community, &message).await, 1);
+    for message in disabled_events {
+        assert_eq!(match_count(db, community, &message).await, 0);
+    }
+    disabled.pool().close().await;
+}
+
+async fn assert_async_isolation(db: &Db, community: CommunityId, keys: &Keys) {
+    // A one-connection serving pool proves a stalled push job cannot monopolize
+    // its only connection. The lock holder uses the independent fixture pool.
+    let isolated = Db::new(&DbConfig {
+        database_url: std::env::var("BUZZ_TEST_DATABASE_URL").unwrap(),
+        max_connections: 1,
+        min_connections: 0,
+        lock_timeout_ms: 100,
+        push_enabled: true,
+        ..DbConfig::default()
+    })
+    .await
+    .unwrap();
+    let mut holder = db.pool().begin().await.unwrap();
+    sqlx::query("LOCK TABLE push_match_queue IN ACCESS EXCLUSIVE MODE")
+        .execute(&mut *holder)
+        .await
+        .unwrap();
+    let message = signed_event(keys, 9, "async-queue-lock");
+    timeout(
+        Duration::from_secs(5),
+        isolated.insert_event(community, &message, None),
+    )
+    .await
+    .expect("push table lock must not delay storage")
+    .unwrap();
+    assert_eq!(event_count(isolated.pool(), community, &message).await, 1);
+    isolated.push_enqueue.as_ref().unwrap().flush().await;
+    // The background statement failed its lock timeout; the message survived.
+    holder.rollback().await.unwrap();
+    assert_eq!(match_count(&isolated, community, &message).await, 0);
+    assert_eq!(event_count(isolated.pool(), community, &message).await, 1);
+    let healthy = signed_event(keys, 9, "async-after-error");
+    isolated
+        .insert_event(community, &healthy, None)
+        .await
+        .unwrap();
+    assert_eq!(match_count(&isolated, community, &healthy).await, 1);
+
+    // Exercise the timestamp filter with a lease timestamp explicitly newer
+    // than this message. This is not a production activation-order regression:
+    // registration overlapping message arrival may send or suppress a wake.
+    let later = create_community(db.pool()).await;
+    let mut activation = db.pool().begin().await.unwrap();
+    sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+        .bind(push::push_gate_lock_key(later))
+        .execute(&mut *activation)
+        .await
+        .unwrap();
+    let before_enrollment = signed_event(keys, 9, "before-async-enrollment");
+    isolated
+        .insert_event(later, &before_enrollment, None)
+        .await
+        .unwrap();
+    sqlx::query(
+        "INSERT INTO push_leases (community_id, author, installation_id, source_event_id, \
+         source_created_at, generation, active, endpoint_enabled, app_profile, endpoint_hash, \
+         endpoint_grant, max_class, subscriptions, expires_at, updated_at) \
+         SELECT $1, author, installation_id, source_event_id, source_created_at, generation, \
+         active, endpoint_enabled, app_profile, endpoint_hash, endpoint_grant, max_class, \
+         subscriptions, expires_at, clock_timestamp() FROM push_leases WHERE community_id=$2",
+    )
+    .bind(later.as_uuid())
+    .bind(community.as_uuid())
+    .execute(&mut *activation)
+    .await
+    .unwrap();
+    activation.commit().await.unwrap();
+    assert_eq!(match_count(&isolated, later, &before_enrollment).await, 0);
+    assert_eq!(
+        event_count(isolated.pool(), later, &before_enrollment).await,
+        1
+    );
+
+    let mut holder = db.pool().begin().await.unwrap();
+    sqlx::query("LOCK TABLE push_match_queue IN ACCESS EXCLUSIVE MODE")
+        .execute(&mut *holder)
+        .await
+        .unwrap();
+    let cancelled = signed_event(keys, 9, "async-cancel-inflight");
+    isolated
+        .insert_event(community, &cancelled, None)
+        .await
+        .unwrap();
+    timeout(Duration::from_secs(2), async {
+        loop {
+            let waiting: bool = sqlx::query_scalar(
+                "SELECT EXISTS (SELECT 1 FROM pg_stat_activity \
+                 WHERE datname=current_database() AND application_name='buzz-push-enqueue' \
+                 AND wait_event_type='Lock' AND query LIKE 'INSERT INTO push_match_queue%')",
+            )
+            .fetch_one(db.pool())
+            .await
+            .unwrap();
+            if waiting {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await
+    .expect("observe actual enqueue blocked on the table");
+    isolated.cancel_push_enqueue();
+    timeout(Duration::from_secs(5), isolated.join_push_enqueue())
+        .await
+        .expect("shutdown cancels in-flight enqueue without waiting for lock timeout");
+    holder.rollback().await.unwrap();
+    assert_eq!(event_count(isolated.pool(), community, &cancelled).await, 1);
+    let queued: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2",
+    )
+    .bind(community.as_uuid())
+    .bind(cancelled.id.as_bytes().as_slice())
+    .fetch_one(db.pool())
+    .await
+    .unwrap();
+    assert_eq!(queued, 0);
+    let after_stop = signed_event(keys, 9, "async-message-after-stop");
+    isolated
+        .insert_event(community, &after_stop, None)
+        .await
+        .unwrap();
+    assert_eq!(
+        event_count(isolated.pool(), community, &after_stop).await,
+        1
+    );
+    isolated.pool().close().await;
+}
--- 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
--- 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
--- 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.
--- a/crates/buzz-relay/src/main.rs
+++ b/crates/buzz-relay/src/main.rs
@@ -441,6 +441,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
    );

    let db_config = DbConfig {
+        push_enabled: config.push_enabled,
        database_url: config.database_url.clone(),
        read_database_url: config.read_database_url.clone(),
        replica_read_max_age_ms: config.replica_read_max_age_ms,
@@ -1214,15 +1215,17 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {

    // NIP-PL matcher and worker are enabled as one unit behind the explicit
    // deployment opt-in. The gateway URL alone never enables push.
-    if state.config.push_enabled {
-        tokio::spawn(buzz_relay::push_runtime::run_matcher(Arc::clone(&state)));
-        tokio::spawn(buzz_relay::push_runtime::run_delivery_worker(Arc::clone(
+    let push_workers = if state.config.push_enabled {
+        let matcher = tokio::spawn(buzz_relay::push_runtime::run_matcher(Arc::clone(&state)));
+        let delivery = tokio::spawn(buzz_relay::push_runtime::run_delivery_worker(Arc::clone(
            &state,
        )));
        info!("NIP-PL push matcher and delivery worker started");
+        Some((matcher, delivery))
    } else {
        info!("NIP-PL push disabled by BUZZ_PUSH_ENABLED");
-    }
+        None
+    };

    // Registration cleanup is independent of configured routes. Delivery is
    // opt-in, while accepted events populate the outbox transactionally.
@@ -1668,6 +1671,16 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
        jwks_refresh_handle,
    )
    .await?;
+    state.push_cancel.cancel();
+    state.db.cancel_push_enqueue();
+    state.db.join_push_enqueue().await;
+    if let Some((matcher, delivery)) = push_workers {
+        for result in [matcher.await, delivery.await] {
+            if let Err(error) = result {
+                tracing::warn!(%error, "push worker failed during shutdown");
+            }
+        }
+    }
    state.community_revalidator_cancel.cancel();
    state.dependency_sampler_cancel.cancel();
    state.dependency_completion_publisher_cancel.cancel();
@@ -3082,6 +3095,11 @@ mod tests {
        assert_eq!(lock, "500ms");
        assert_eq!(idle, "1min");
        assert_eq!(statement, "0");
+        let push: String = sqlx::query_scalar("SELECT current_setting('buzz.push_enabled')")
+            .fetch_one(&pool)
+            .await
+            .expect("audit pool push setting");
+        assert_eq!(push, "off");

        let lock_key = i64::from_be_bytes(
            Uuid::new_v4().as_bytes()[..8]
--- 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<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));
+    }
+}
--- 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<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,8 @@ impl AppState {
    /// the readiness gauge on its next request.
    pub fn begin_shutdown(&self) {
        self.shutting_down.store(true, Ordering::Release);
+        self.push_cancel.cancel();
+        self.db.cancel_push_enqueue();
    }

    #[cfg(test)]
--- a/docs/push-gateway-deployment.md
+++ b/docs/push-gateway-deployment.md
@@ -182,6 +182,7 @@ installation, event, or request identifier as a label.
 | Metric | Type | Labels | Meaning |
 |---|---|---|---|
 | `buzz_push_enabled` | gauge | — | `1` only when the deployment opt-in is active. |
+| `buzz_push_enqueue_total` | counter | `result` = `completed` \| `full` \| `transaction_full` \| `closed` \| `error` \| `timeout` \| `shutdown` | Best-effort post-commit enqueue outcomes. |
 | `buzz_push_match_jobs_total` | counter | `result` = `matched` \| `unmatched` \| `error` \| `context_error` | Accepted message events evaluated by the matcher. |
 | `buzz_push_match_queue_seconds` | histogram | — | Relay receipt to matcher evaluation latency. |
 | `buzz_push_wakes_total` | counter | `result` = `enqueued` \| `duplicate` \| `inactive_lease` | Durable wake-enqueue outcomes. |
@@ -249,12 +250,70 @@ triggering message. Keep fallback-to-channel and placeholder/failure cases as
 explicit counts in the manual sample until privacy-preserving client telemetry
 is designed.

-Rollback does not require deleting credentials or mutating existing leases.
-Set `BUZZ_PUSH_ENABLED=false` on the enabled relays to stop advertisement, lease
-acceptance, matching, workers, and new gateway traffic. If the gateway itself
-is unhealthy, disable the gateway deployment only after relay delivery is off.
-Existing leases and gateway authorities then expire naturally. Adding an App
-Store application profile is outside this internal evaluation.
+### Relay rollback and reactivation
+
+Set `BUZZ_PUSH_ENABLED=false` in the approved configuration for **every writer
+workload in the affected database scope**, then replace all previously enabled
+relay processes. This is a startup setting, not a live configuration reload.
+Keep the delivery URL, image, relay keys, schema, leases, client preferences and
+gateway grants unchanged. Do not stop a shared gateway as part of relay rollback.
+
+The relay copies the flag into `DbConfig::push_enabled`. Every physical writer
+connection, including the audit pool and replacement connections, sets the
+`buzz.push_enabled` PostgreSQL session setting. Other `DbConfig` callers default
+to disabled. Updated writers also set `buzz.push_async_enqueue=on`, which makes
+the surviving trigger skip enqueueing regardless of the enabled flag. The
+application producer is created only when enabled. Disabled writes perform no
+push locks, lease scans or queue writes.
+The overlap migration updates only an existing trigger function; it does not
+recreate a retired trigger or function. Deploy that migration before relying on
+the flag with a surviving trigger. Older images and raw SQL writers with a
+missing setting retain legacy enqueue behavior; inventory and replace or
+explicitly disable all such connections before declaring rollback complete.
+
+On SIGTERM/Ctrl-C, the relay cancels enqueue, matcher and delivery futures immediately,
+before its listener drain. Uncommitted transactions roll back; committed claims
+remain leased until their normal expiry. No cancellation cleanup deletes or
+acknowledges queue entries. An HTTP request already accepted by the gateway may
+still produce a notification. The process shutdown deadline remains the hard
+backstop; a stalled replacement with old workers alive is incomplete rollback.
+
+Verify all old writer sessions and workers have exited, eligible messages still
+store and arrive in chat, no new matcher jobs appear, and gateway attempts stop
+once in-flight work settles. Inspect enqueue activity as well as queue counts:
+a stable count alone can conceal additions and removals. Ordinary reads, writes,
+authentication and reconnects must remain healthy.
+
+When enabled, updated application writers stage at most 256 event IDs per
+transaction and submit them only after successful commit using a non-waiting
+send. One background worker has a 256-item queue and a dedicated one-connection
+pool; it never borrows a message-write connection. Each attempt has a two-second
+deadline, a 500 ms lock timeout and a 1.5 second statement timeout. Saturation,
+errors, deadlines, process loss and shutdown can drop notifications. They cannot
+reject or roll back a committed message, and there is no replay/backfill of
+in-memory work. Monitor `buzz_push_enqueue_total` by result (`completed`, `full`,
+`transaction_full`, `closed`, `error`, `timeout`, `shutdown`). `completed` includes
+jobs with no eligible lease. Database enqueue is conditional on a live event and
+an eligible lease whose recorded update timestamp is no later than event receipt.
+These timestamps do not establish transaction commit order. When registration
+and message arrival overlap, either sending or suppressing the notification is
+acceptable; there is no exact enrollment-order guarantee. A concurrent lease
+update, including renewal of an active lease, can also suppress a notification
+under this best-effort contract. Activation never scans
+historical messages or deliberately backfills older notifications.
+
+This isolation requires the overlap migration and updated writers. Old images
+and raw SQL writers can still run the legacy transactional trigger until replaced.
+
+Pending matcher jobs and pending/claimed delivery jobs remain dormant. Do not
+process or delete them during urgent rollback. **Before re-enabling**, inspect
+retained work across the full affected scope and record an explicit disposition
+decision. Cleanup size, batching and scoped commands are an operator decision;
+preserve leases and handle completed records through retention policy. There is
+no automatic purge or startup queue interlock: enabling workers resumes eligible
+retained work under existing expiry/retry rules. Thus an unchecked nonempty queue
+must not be silently reactivated. Messages stored while disabled are not
+backfilled on lease activation or re-enablement.

 ## Helm production inputs

--- a/migrations/0059_push_rollback_gate.sql
+++ b/migrations/0059_push_rollback_gate.sql
@@ -0,0 +1,46 @@
+-- Only update a surviving overlap trigger function. Never resurrect the
+-- function after the independent trigger-retirement migration has removed it.
+DO $migration$
+BEGIN
+    IF to_regprocedure('enqueue_push_match_job()') IS NOT NULL THEN
+        EXECUTE $function$
+-- T1b push gate (migrations 0023 and 0059). Enqueue only when the
+-- community has an active, endpoint-enabled, unexpired lease; the shared
+-- advisory lock pairs with the exclusive lock taken by lease activations
+-- (crates/buzz-db/src/push.rs) to close the lost-wake race.
+CREATE OR REPLACE FUNCTION enqueue_push_match_job() RETURNS trigger
+LANGUAGE plpgsql AS $$
+BEGIN
+    -- Legacy connections without the setting preserve their existing behavior.
+    -- Updated writer pools always set it, including on reconnect. Check before
+    -- taking push locks or touching leases/queues so disabled chat stays healthy.
+    IF COALESCE(current_setting('buzz.push_async_enqueue', true), '') = 'on' THEN
+        RETURN NEW;
+    END IF;
+    IF NOT COALESCE(NULLIF(current_setting('buzz.push_enabled', true), '')::boolean, true) THEN
+        RETURN NEW;
+    END IF;
+    -- Keep this allowlist identical to the relay's validated NIP-PL descriptor.
+    -- This compatibility path covers legacy writers until they are replaced.
+    IF NEW.kind IN (9, 40002, 45001, 45003) THEN
+        PERFORM pg_advisory_xact_lock_shared(
+            hashtextextended('buzz_push_gate:' || NEW.community_id::text, 0));
+        IF EXISTS (
+            SELECT 1 FROM push_leases
+            WHERE community_id = NEW.community_id
+              AND active
+              AND endpoint_enabled
+              AND expires_at > EXTRACT(EPOCH FROM now())::bigint
+        ) THEN
+            INSERT INTO push_match_queue (community_id, event_id)
+            VALUES (NEW.community_id, NEW.id)
+            ON CONFLICT DO NOTHING;
+        END IF;
+    END IF;
+    RETURN NEW;
+END
+$$;
+$function$;
+    END IF;
+END
+$migration$;
--- a/schema/functions/enqueue_push_match_job.sql
+++ b/schema/functions/enqueue_push_match_job.sql
@@ -1,13 +1,21 @@
-- T1b push gate (keep in sync with migrations/0023). Enqueue only when the
+-- T1b push gate (migrations 0023 and 0059). Enqueue only when the
 -- community has an active, endpoint-enabled, unexpired lease; the shared
 -- advisory lock pairs with the exclusive lock taken by lease activations
 -- (crates/buzz-db/src/push.rs) to close the lost-wake race.
 CREATE FUNCTION enqueue_push_match_job() RETURNS trigger
 LANGUAGE plpgsql AS $$
 BEGIN
+    -- Legacy connections without the setting preserve their existing behavior.
+    -- Updated writer pools always set it, including on reconnect. Check before
+    -- taking push locks or touching leases/queues so disabled chat stays healthy.
+    IF COALESCE(current_setting('buzz.push_async_enqueue', true), '') = 'on' THEN
+        RETURN NEW;
+    END IF;
+    IF NOT COALESCE(NULLIF(current_setting('buzz.push_enabled', true), '')::boolean, true) THEN
+        RETURN NEW;
+    END IF;
    -- Keep this allowlist identical to the relay's validated NIP-PL descriptor.
-    -- Centralizing it on the events table covers every durable producer,
-    -- including internal paths that bypass live dispatch.
+    -- This compatibility path covers legacy writers until they are replaced.
    IF NEW.kind IN (9, 40002, 45001, 45003) THEN
        PERFORM pg_advisory_xact_lock_shared(
            hashtextextended('buzz_push_gate:' || NEW.community_id::text, 0));
--- 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),
--- 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