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

.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/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 }

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(())
     }
 

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

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