fix(push): reap terminal wake outbox rows (BUZZ-SEC-062)

crates/buzz-db/src/lib.rs

@@ -1136,6 +1136,19 @@ impl Db {
         push::fail_wake(&self.pool, community, id, claim_id).await
     }
 
+    /// Delete one deployment-wide bounded batch of retained terminal or
+    /// undeliverable wakes.
+    ///
+    /// `batch_size` is capped at [`push::MAX_WAKE_PRUNE_BATCH_SIZE`] so callers
+    /// cannot turn a maintenance tick into an unbounded delete transaction.
+    pub async fn prune_push_wake_outbox(
+        &self,
+        before: DateTime<Utc>,
+        batch_size: u32,
+    ) -> Result<u64> {
+        push::prune_wake_outbox(&self.pool, before, batch_size).await
+    }
+
     /// Disable an endpoint only if the specified lease generation is current.
     pub async fn disable_push_endpoint(
         &self,

crates/buzz-db/src/migration.rs

@@ -549,7 +549,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 21);
+        assert_eq!(migrations.len(), 22);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -834,6 +834,14 @@ mod tests {
         assert!(push_admission.contains("queued_jobs < max_queued_jobs"));
         assert!(push_admission.contains("FROM push_leases"));
         assert!(push_admission.contains("WHERE singleton AND enabled"));
+
+        // The production outbox reaper walks old rows in creation order and
+        // then applies its terminal/undeliverable predicate to a bounded batch.
+        assert_eq!(migrations[21].version, 22);
+        let wake_retention = migrations[21].sql.as_str();
+        assert!(wake_retention.contains("CREATE INDEX push_wake_outbox_retention"));
+        assert!(wake_retention.contains("CREATE TABLE push_wake_outbox_community_state"));
+        assert!(wake_retention.contains("retained_rows < max_retained_rows"));
     }
 
     #[test]

crates/buzz-db/src/push.rs

@@ -1,7 +1,9 @@
 //! Community-scoped NIP-PL lease and durable wake-outbox persistence.
 //!
-//! Every operation requires a server-resolved [`CommunityId`]. Client-provided
-//! origins never select rows in this module.
+//! Every request operation requires a server-resolved [`CommunityId`].
+//! Client-provided origins never select rows in this module. The sole
+//! deployment-wide operation is the fixed-size retention sweep, which selects
+//! tenant keys only from persisted rows.
 
 use buzz_core::CommunityId;
 use chrono::{DateTime, Utc};
@@ -18,6 +20,9 @@ pub const MAX_MATCH_ATTEMPTS: i32 = 8;
 /// Maximum number of matcher rows removed in one disabled-mode transaction.
 pub const MATCH_DRAIN_BATCH: i64 = 1_000;
 
+/// Hard ceiling for one wake-outbox retention transaction.
+pub const MAX_WAKE_PRUNE_BATCH_SIZE: u32 = 1_000;
+
 /// Common signed-event ordering fields for a lease replacement.
 #[derive(Debug, Clone, Copy)]
 pub struct LeaseVersion<'a> {
@@ -66,6 +71,8 @@ pub enum EnqueueWakeOutcome {
     Duplicate(Uuid),
     /// No current active, unexpired lease matched the supplied generation.
     InactiveLease,
+    /// The community's durable wake budget is full.
+    CapacityExceeded,
 }
 
 /// Durable wake fields not copied from the effective lease.
@@ -537,13 +544,41 @@ pub async fn enqueue_wake(
     };
     let endpoint_hash: Vec<u8> = endpoint_hash.try_get("endpoint_hash")?;
 
+    // Serialize enqueue and duplicate detection within this community. The
+    // admission trigger uses the same row to enforce the retained-row budget,
+    // so concurrent endpoints cannot cross the cap or leak accounting.
+    sqlx::query(
+        "INSERT INTO push_wake_outbox_community_state (community_id) VALUES ($1) \
+         ON CONFLICT (community_id) DO NOTHING",
+    )
+    .bind(community.as_uuid())
+    .execute(&mut *tx)
+    .await?;
+    sqlx::query("SELECT retained_rows FROM push_wake_outbox_community_state WHERE community_id=$1 FOR UPDATE")
+        .bind(community.as_uuid())
+        .fetch_one(&mut *tx)
+        .await?;
+
+    let duplicate: Option<Uuid> = sqlx::query_scalar(
+        "SELECT id FROM push_wake_outbox \
+         WHERE community_id=$1 AND endpoint_hash=$2 AND event_id=$3",
+    )
+    .bind(community.as_uuid())
+    .bind(&endpoint_hash)
+    .bind(wake.event_id)
+    .fetch_optional(&mut *tx)
+    .await?;
+    if let Some(id) = duplicate {
+        tx.commit().await?;
+        return Ok(EnqueueWakeOutcome::Duplicate(id));
+    }
+
     let inserted = sqlx::query(
         r#"
         INSERT INTO push_wake_outbox (
             community_id, author, installation_id, lease_generation,
             endpoint_hash, event_id, class, expires_at
         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
-        ON CONFLICT (community_id, endpoint_hash, event_id) DO NOTHING
         RETURNING id
         "#,
     )
@@ -558,21 +593,9 @@ pub async fn enqueue_wake(
     .fetch_optional(&mut *tx)
     .await?;
 
-    let outcome = if let Some(row) = inserted {
-        EnqueueWakeOutcome::Enqueued(row.try_get("id")?)
-    } else {
-        // This is a separate statement so READ COMMITTED observes a competing
-        // transaction whose unique-key insert completed while ours waited.
-        let row = sqlx::query(
-            "SELECT id FROM push_wake_outbox \
-             WHERE community_id = $1 AND endpoint_hash = $2 AND event_id = $3",
-        )
-        .bind(community.as_uuid())
-        .bind(&endpoint_hash)
-        .bind(wake.event_id)
-        .fetch_one(&mut *tx)
-        .await?;
-        EnqueueWakeOutcome::Duplicate(row.try_get("id")?)
+    let outcome = match inserted {
+        Some(row) => EnqueueWakeOutcome::Enqueued(row.try_get("id")?),
+        None => EnqueueWakeOutcome::CapacityExceeded,
     };
     tx.commit().await?;
     Ok(outcome)
@@ -990,28 +1013,64 @@ pub async fn disable_endpoint_generation(
     Ok(result.rows_affected() == 1)
 }
 
-/// Delete terminal/expired outbox rows older than a retention cutoff.
+/// Delete a deployment-wide, bounded batch of terminal or undeliverable
+/// outbox rows older than a retention cutoff.
 ///
 /// NIP-RS hard purge only targets kind 30078, which is not push-eligible and
 /// therefore cannot have a matcher row; any other absent source is handled by
 /// the matcher's fenced load-miss deletion.
 pub async fn prune_wake_outbox(
     pool: &PgPool,
-    community: CommunityId,
     before: DateTime<Utc>,
+    batch_size: u32,
 ) -> Result<u64> {
+    if batch_size == 0 {
+        return Ok(0);
+    }
+    let batch_size = i64::from(batch_size.min(MAX_WAKE_PRUNE_BATCH_SIZE));
     let result = sqlx::query(
-        "DELETE FROM push_wake_outbox o \
-         WHERE o.community_id = $1 AND o.created_at < $2 \
-           AND (o.state IN ('delivered', 'failed') \
-                OR o.expires_at <= EXTRACT(EPOCH FROM now())::bigint) \
-           AND NOT EXISTS ( \
-               SELECT 1 FROM push_match_queue q \
-               WHERE q.community_id = o.community_id AND q.event_id = o.event_id \
-           )",
+        r#"
+        WITH candidates AS (
+            SELECT o.community_id, o.id
+            FROM push_wake_outbox o
+            WHERE o.created_at < $1
+              AND NOT EXISTS (
+                  SELECT 1
+                  FROM push_match_queue q
+                  WHERE q.community_id = o.community_id
+                    AND q.event_id = o.event_id
+              )
+              AND (
+                  o.state IN ('delivered', 'failed')
+                  OR o.expires_at <= EXTRACT(EPOCH FROM now())::bigint
+                  OR (
+                      o.state = 'sending'
+                      AND COALESCE(o.lease_until, o.created_at) < $1
+                  )
+                  OR NOT EXISTS (
+                      SELECT 1
+                      FROM push_leases l
+                      WHERE l.community_id = o.community_id
+                        AND l.author = o.author
+                        AND l.installation_id = o.installation_id
+                        AND l.generation = o.lease_generation
+                        AND l.endpoint_hash = o.endpoint_hash
+                        AND l.active
+                        AND l.endpoint_enabled
+                        AND l.expires_at > EXTRACT(EPOCH FROM now())::bigint
+                  )
+              )
+            ORDER BY o.created_at, o.community_id, o.id
+            FOR UPDATE OF o SKIP LOCKED
+            LIMIT $2
+        )
+        DELETE FROM push_wake_outbox o
+        USING candidates c
+        WHERE o.community_id = c.community_id AND o.id = c.id
+        "#,
     )
-    .bind(community.as_uuid())
     .bind(before)
+    .bind(batch_size)
     .execute(pool)
     .await?;
     Ok(result.rows_affected())
@@ -1316,6 +1375,7 @@ mod tests {
             ids.push(match task.await.expect("join") {
                 EnqueueWakeOutcome::Enqueued(id) | EnqueueWakeOutcome::Duplicate(id) => id,
                 EnqueueWakeOutcome::InactiveLease => panic!("lease unexpectedly inactive"),
+                EnqueueWakeOutcome::CapacityExceeded => panic!("wake budget unexpectedly full"),
             });
         }
         assert!(ids.iter().all(|id| *id == ids[0]));
@@ -1360,6 +1420,83 @@ mod tests {
         assert_eq!(total, 2, "same dedup key is independent per community");
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn wake_admission_caps_each_community_and_delete_releases_capacity() {
+        let pool = setup_pool().await;
+        let community = make_community(&pool).await;
+        let author = [71; 32];
+        activate(&pool, community, &author, "install", &[72; 32], 1).await;
+        sqlx::query(
+            "INSERT INTO push_wake_outbox_community_state \
+             (community_id, max_retained_rows) VALUES ($1, 1) \
+             ON CONFLICT (community_id) DO UPDATE SET max_retained_rows=1",
+        )
+        .bind(community.as_uuid())
+        .execute(&pool)
+        .await
+        .expect("set a small outbox budget");
+
+        assert!(matches!(
+            enqueue_wake(
+                &pool,
+                community,
+                &author,
+                "install",
+                NewWake {
+                    lease_generation: 1,
+                    event_id: &[73; 32],
+                    class: "default",
+                    expires_at: i64::MAX / 2,
+                },
+            )
+            .await
+            .expect("first wake"),
+            EnqueueWakeOutcome::Enqueued(_)
+        ));
+        assert_eq!(
+            enqueue_wake(
+                &pool,
+                community,
+                &author,
+                "install",
+                NewWake {
+                    lease_generation: 1,
+                    event_id: &[74; 32],
+                    class: "default",
+                    expires_at: i64::MAX / 2,
+                },
+            )
+            .await
+            .expect("capacity result"),
+            EnqueueWakeOutcome::CapacityExceeded
+        );
+
+        let state: (i64, i64) = sqlx::query_as(
+            "SELECT retained_rows, dropped_wakes \
+             FROM push_wake_outbox_community_state WHERE community_id=$1",
+        )
+        .bind(community.as_uuid())
+        .fetch_one(&pool)
+        .await
+        .expect("read outbox admission state");
+        assert_eq!(state, (1, 1));
+
+        sqlx::query("DELETE FROM push_wake_outbox WHERE community_id=$1")
+            .bind(community.as_uuid())
+            .execute(&pool)
+            .await
+            .expect("release outbox capacity");
+        let retained: i64 = sqlx::query_scalar(
+            "SELECT retained_rows FROM push_wake_outbox_community_state WHERE community_id=$1",
+        )
+        .bind(community.as_uuid())
+        .fetch_one(&pool)
+        .await
+        .expect("read released capacity");
+        assert_eq!(retained, 0);
+    }
+
     async fn enqueue_one(
         pool: &PgPool,
         community: CommunityId,
@@ -1756,6 +1893,120 @@ mod tests {
         assert_eq!(claimed, 1);
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn wake_reaper_prunes_only_retained_undeliverable_rows_in_bounded_batches() {
+        let pool = setup_pool().await;
+        let community = make_community(&pool).await;
+        let author = [60; 32];
+        activate(&pool, community, &author, "install", &[61; 32], 1).await;
+
+        let delivered = enqueue_one(&pool, community, &author, &[62; 32], 1).await;
+        let failed = enqueue_one(&pool, community, &author, &[63; 32], 1).await;
+        let expired_pending = enqueue_one(&pool, community, &author, &[64; 32], 1).await;
+        let abandoned_sending = enqueue_one(&pool, community, &author, &[65; 32], 1).await;
+        let live_pending = enqueue_one(&pool, community, &author, &[66; 32], 1).await;
+
+        let inactive_author = [67; 32];
+        activate(
+            &pool,
+            community,
+            &inactive_author,
+            "inactive-install",
+            &[68; 32],
+            1,
+        )
+        .await;
+        let inactive = sqlx::query_scalar::<_, Uuid>(
+            r#"
+            INSERT INTO push_wake_outbox (
+                community_id, author, installation_id, lease_generation,
+                endpoint_hash, event_id, class, expires_at
+            ) VALUES ($1, $2, 'inactive-install', 1, $3, $4, 'default', $5)
+            RETURNING id
+            "#,
+        )
+        .bind(community.as_uuid())
+        .bind(inactive_author)
+        .bind([68; 32])
+        .bind([69; 32])
+        .bind(i64::MAX / 2)
+        .fetch_one(&pool)
+        .await
+        .expect("insert inactive-lease wake");
+        assert_eq!(
+            revoke_lease(
+                &pool,
+                community,
+                &inactive_author,
+                "inactive-install",
+                version(70, 20, 2),
+            )
+            .await
+            .expect("revoke wake lease"),
+            ReplaceLeaseOutcome::Accepted
+        );
+
+        // The event trigger queues matcher work. Once matching completes, the
+        // retained outbox rows are eligible for production cleanup.
+        sqlx::query("DELETE FROM push_match_queue WHERE community_id=$1")
+            .bind(community.as_uuid())
+            .execute(&pool)
+            .await
+            .expect("complete matcher jobs");
+        sqlx::query(
+            "UPDATE push_wake_outbox SET created_at=now()-interval '2 days' \
+             WHERE community_id=$1",
+        )
+        .bind(community.as_uuid())
+        .execute(&pool)
+        .await
+        .expect("age wake rows");
+        sqlx::query(
+            "UPDATE push_wake_outbox SET state='delivered' WHERE community_id=$1 AND id=$2",
+        )
+        .bind(community.as_uuid())
+        .bind(delivered)
+        .execute(&pool)
+        .await
+        .expect("mark delivered");
+        sqlx::query("UPDATE push_wake_outbox SET state='failed' WHERE community_id=$1 AND id=$2")
+            .bind(community.as_uuid())
+            .bind(failed)
+            .execute(&pool)
+            .await
+            .expect("mark failed");
+        sqlx::query("UPDATE push_wake_outbox SET expires_at=0 WHERE community_id=$1 AND id=$2")
+            .bind(community.as_uuid())
+            .bind(expired_pending)
+            .execute(&pool)
+            .await
+            .expect("expire pending wake");
+        sqlx::query(
+            "UPDATE push_wake_outbox SET state='sending', claim_id=$3, \
+             lease_until=now()-interval '2 days' WHERE community_id=$1 AND id=$2",
+        )
+        .bind(community.as_uuid())
+        .bind(abandoned_sending)
+        .bind(Uuid::new_v4())
+        .execute(&pool)
+        .await
+        .expect("abandon claimed wake");
+
+        let cutoff = Utc::now() - chrono::Duration::days(1);
+        assert_eq!(prune_wake_outbox(&pool, cutoff, 3).await.unwrap(), 3);
+        assert_eq!(prune_wake_outbox(&pool, cutoff, 3).await.unwrap(), 2);
+
+        let remaining: Vec<Uuid> =
+            sqlx::query_scalar("SELECT id FROM push_wake_outbox WHERE community_id=$1")
+                .bind(community.as_uuid())
+                .fetch_all(&pool)
+                .await
+                .expect("load retained wake rows");
+        assert_eq!(remaining, vec![live_pending]);
+        assert!(!remaining.contains(&inactive));
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn delivered_wake_is_retained_while_rematch_is_queued() {
@@ -1777,7 +2028,9 @@ mod tests {
 
         let cutoff = Utc::now() - chrono::Duration::days(1);
         assert_eq!(
-            prune_wake_outbox(&pool, community, cutoff).await.unwrap(),
+            prune_wake_outbox(&pool, cutoff, MAX_WAKE_PRUNE_BATCH_SIZE)
+                .await
+                .unwrap(),
             0
         );
         sqlx::query("DELETE FROM push_match_queue WHERE community_id=$1 AND event_id=$2")
@@ -1787,7 +2040,9 @@ mod tests {
             .await
             .expect("complete rematch");
         assert_eq!(
-            prune_wake_outbox(&pool, community, cutoff).await.unwrap(),
+            prune_wake_outbox(&pool, cutoff, MAX_WAKE_PRUNE_BATCH_SIZE)
+                .await
+                .unwrap(),
             1
         );
     }
@@ -1834,7 +2089,9 @@ mod tests {
         .expect("mark old wake delivered");
         let cutoff = Utc::now() - chrono::Duration::days(1);
         assert_eq!(
-            prune_wake_outbox(&pool, community, cutoff).await.unwrap(),
+            prune_wake_outbox(&pool, cutoff, MAX_WAKE_PRUNE_BATCH_SIZE)
+                .await
+                .unwrap(),
             0
         );
         sqlx::query(
@@ -1863,7 +2120,9 @@ mod tests {
         .unwrap();
         assert_eq!(remaining, 0);
         assert_eq!(
-            prune_wake_outbox(&pool, community, cutoff).await.unwrap(),
+            prune_wake_outbox(&pool, cutoff, MAX_WAKE_PRUNE_BATCH_SIZE)
+                .await
+                .unwrap(),
             1,
             "reaped poison job must release delivered-wake retention"
         );

crates/buzz-relay/src/main.rs

@@ -641,6 +641,14 @@ async fn main() -> anyhow::Result<()> {
         });
     }
 
+    // Retention runs even while delivery is disabled so a configuration change
+    // cannot strand terminal or expired rows from an earlier enabled period.
+    let push_reaper_cancel = CancellationToken::new();
+    tokio::spawn(buzz_relay::push_runtime::run_wake_outbox_reaper(
+        Arc::clone(&state),
+        push_reaper_cancel.clone(),
+    ));
+
     // NIP-PL matcher and worker are enabled as one unit. Lease acceptance is
     // already disabled without the exact gateway URL, so discovery and runtime
     // cannot advertise or accumulate work for an undeliverable configuration.
@@ -986,6 +994,7 @@ async fn main() -> anyhow::Result<()> {
 
     serve(router, health_router, Arc::clone(&state)).await?;
     state.community_revalidator_cancel.cancel();
+    push_reaper_cancel.cancel();
 
     // Signal the audit worker to stop accepting, flush buffered entries, and
     // exit. Uses a CancellationToken so it works regardless of how many

crates/buzz-relay/src/push_runtime.rs

@@ -8,6 +8,7 @@ use chrono::{TimeDelta, Utc};
 use nostr::{EventBuilder, Filter, Kind, Tag};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest as _, Sha256};
+use tokio_util::sync::CancellationToken;
 use tracing::{error, info, warn};
 
 use crate::{handlers::push_lease::Subscription, state::AppState};
@@ -16,6 +17,15 @@ const CLAIM_SECS: i64 = 30;
 const EVENT_USEFUL_SECS: i64 = 3600;
 const MAX_ATTEMPTS: i32 = 8;
 
+// Never delete a wake until at least one day after it was created, then delete
+// at most 1,000 terminal or undeliverable rows per minute. These are fixed
+// safety bounds rather than deployment knobs so an accidental value cannot
+// turn a maintenance tick into an unbounded transaction or indefinite
+// retention.
+const WAKE_REAPER_INTERVAL: Duration = Duration::from_secs(60);
+const WAKE_RETENTION_SECS: i64 = 24 * 60 * 60;
+const WAKE_REAPER_BATCH_SIZE: u32 = 1_000;
+
 #[derive(Serialize)]
 struct DeliveryRequest<'a> {
     v: u8,
@@ -150,7 +160,7 @@ async fn process_match(state: &AppState, job: &buzz_db::push::ClaimedMatch) -> a
         if expires_at <= Utc::now().timestamp() {
             continue;
         }
-        let _ = state
+        let outcome = state
             .db
             .enqueue_push_wake(
                 job.community,
@@ -164,6 +174,9 @@ async fn process_match(state: &AppState, job: &buzz_db::push::ClaimedMatch) -> a
                 },
             )
             .await?;
+        if matches!(outcome, buzz_db::push::EnqueueWakeOutcome::CapacityExceeded) {
+            metrics::counter!("buzz_push_wake_admission_drops_total").increment(1);
+        }
     }
     Ok(())
 }
@@ -222,6 +235,74 @@ pub async fn run_delivery_worker(state: Arc<AppState>) {
     }
 }
 
+/// Reap retained push wakes until relay shutdown is signalled.
+///
+/// A tick runs immediately and then once per minute. Each transaction deletes
+/// at most 1,000 rows whose creation time is over 24 hours old and whose state
+/// is terminal, expired, abandoned, or no longer backed by an effective lease.
+pub async fn run_wake_outbox_reaper(state: Arc<AppState>, cancel: CancellationToken) {
+    info!(
+        interval_secs = WAKE_REAPER_INTERVAL.as_secs(),
+        retention_secs = WAKE_RETENTION_SECS,
+        batch_size = WAKE_REAPER_BATCH_SIZE,
+        "push wake outbox reaper started"
+    );
+    run_wake_outbox_reaper_loop(WAKE_REAPER_INTERVAL, cancel, move || {
+        let state = Arc::clone(&state);
+        async move { run_wake_outbox_reaper_tick(&state).await }
+    })
+    .await;
+    info!("push wake outbox reaper stopped");
+}
+
+async fn run_wake_outbox_reaper_loop<Tick, TickFuture>(
+    period: Duration,
+    cancel: CancellationToken,
+    mut tick: Tick,
+) where
+    Tick: FnMut() -> TickFuture,
+    TickFuture: std::future::Future<Output = ()>,
+{
+    let mut interval = tokio::time::interval(period);
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    loop {
+        tokio::select! {
+            biased;
+            _ = cancel.cancelled() => break,
+            _ = interval.tick() => tick().await,
+        }
+    }
+}
+
+async fn run_wake_outbox_reaper_tick(state: &AppState) {
+    let before = Utc::now() - TimeDelta::seconds(WAKE_RETENTION_SECS);
+    match state
+        .db
+        .prune_push_wake_outbox(before, WAKE_REAPER_BATCH_SIZE)
+        .await
+    {
+        Ok(deleted) => {
+            metrics::counter!(
+                "buzz_push_wake_reaper_ticks_total",
+                "outcome" => "success"
+            )
+            .increment(1);
+            metrics::counter!("buzz_push_wake_reaper_rows_total").increment(deleted);
+            if deleted > 0 {
+                info!(deleted, "pruned retained push wake rows");
+            }
+        }
+        Err(error) => {
+            metrics::counter!(
+                "buzz_push_wake_reaper_ticks_total",
+                "outcome" => "error"
+            )
+            .increment(1);
+            warn!(%error, "push wake outbox reaper tick failed");
+        }
+    }
+}
+
 async fn deliver_one(
     state: &AppState,
     http: &reqwest::Client,
@@ -455,9 +536,40 @@ mod tests {
     use super::*;
     use axum::{extract::State, routing::post, Json, Router};
     use serde_json::Value;
-    use std::{future::IntoFuture, sync::Arc};
+    use std::{
+        future::IntoFuture,
+        sync::{
+            atomic::{AtomicUsize, Ordering},
+            Arc,
+        },
+    };
     use tokio::sync::Mutex;
 
+    #[tokio::test(start_paused = true)]
+    async fn wake_reaper_ticks_immediately_and_cancels_without_waiting() {
+        let cancel = CancellationToken::new();
+        let task_cancel = cancel.clone();
+        let ticks = Arc::new(AtomicUsize::new(0));
+        let task_ticks = Arc::clone(&ticks);
+        let task = tokio::spawn(async move {
+            run_wake_outbox_reaper_loop(Duration::from_secs(60), task_cancel, move || {
+                let ticks = Arc::clone(&task_ticks);
+                async move {
+                    ticks.fetch_add(1, Ordering::Relaxed);
+                }
+            })
+            .await;
+        });
+
+        tokio::task::yield_now().await;
+        assert_eq!(ticks.load(Ordering::Relaxed), 1);
+        cancel.cancel();
+        tokio::time::timeout(Duration::from_millis(1), task)
+            .await
+            .expect("reaper cancellation must not wait for the next interval")
+            .expect("join reaper task");
+    }
+
     #[test]
     fn gift_wrap_match_requires_self_p_filter_and_recipient() {
         let recipient = nostr::Keys::generate();

docs/push-gateway-deployment.md

@@ -89,6 +89,19 @@ rotate claims by the least recently served community. When a community reaches
 its queue limit, Buzz keeps the accepted event but skips its push match job; the
 per-community `dropped_jobs` counter records that degradation for operators.
 
+The relay runs wake-outbox retention even when delivery is disabled, so jobs
+from an earlier enabled period cannot remain permanently. A row is never
+eligible for deletion until 24 hours after its creation; after that, the relay
+checks once per minute and deletes at most 1,000 terminal or otherwise
+undeliverable rows in one transaction. Monitor
+`buzz_push_wake_reaper_ticks_total{outcome="error"}` and
+`buzz_push_wake_reaper_rows_total`; repeated errors or a sustained full-batch
+deletion rate indicate the outbox is growing faster than retention can drain it.
+Each community may retain at most 100,000 wake rows. When that budget is full,
+the relay keeps the source event but skips additional wakes and increments
+`buzz_push_wake_admission_drops_total`; this prevents one tenant from consuming
+unbounded durable outbox capacity while old rows await retention.
+
 ## Relay integration status
 
 The operational relay integration is complete: per-origin event matching with

migrations/0022_push_wake_outbox_retention.sql

@@ -0,0 +1,68 @@
+-- Bound the recurring wake-outbox retention scan by creation order. The
+-- reaper filters terminal and no-longer-deliverable rows from this ordered
+-- prefix and deletes only a fixed-size batch on each tick.
+CREATE INDEX push_wake_outbox_retention
+    ON push_wake_outbox (created_at, community_id, id);
+
+-- A member can create multiple leases that match one event, so retention alone
+-- is not an admission boundary. Keep a race-safe row budget per community and
+-- drop only the wake when that budget is exhausted; the source event remains
+-- accepted and available through the relay.
+CREATE TABLE push_wake_outbox_community_state (
+    community_id UUID PRIMARY KEY REFERENCES communities(id),
+    retained_rows BIGINT NOT NULL DEFAULT 0 CHECK (retained_rows >= 0),
+    max_retained_rows BIGINT NOT NULL DEFAULT 100000 CHECK (max_retained_rows > 0),
+    dropped_wakes BIGINT NOT NULL DEFAULT 0 CHECK (dropped_wakes >= 0)
+);
+
+INSERT INTO push_wake_outbox_community_state (community_id, retained_rows)
+SELECT community_id, count(*)
+FROM push_wake_outbox
+GROUP BY community_id;
+
+CREATE FUNCTION admit_push_wake_outbox_row() RETURNS trigger
+LANGUAGE plpgsql AS $$
+BEGIN
+    INSERT INTO push_wake_outbox_community_state (community_id)
+    VALUES (NEW.community_id)
+    ON CONFLICT (community_id) DO NOTHING;
+
+    UPDATE push_wake_outbox_community_state
+    SET retained_rows = retained_rows + 1
+    WHERE community_id = NEW.community_id
+      AND retained_rows < max_retained_rows;
+
+    IF NOT FOUND THEN
+        UPDATE push_wake_outbox_community_state
+        SET dropped_wakes = dropped_wakes + 1
+        WHERE community_id = NEW.community_id;
+        RETURN NULL;
+    END IF;
+
+    RETURN NEW;
+END
+$$;
+
+CREATE TRIGGER push_wake_outbox_admission
+BEFORE INSERT ON push_wake_outbox
+FOR EACH ROW EXECUTE FUNCTION admit_push_wake_outbox_row();
+
+CREATE FUNCTION account_push_wake_outbox_deletes() RETURNS trigger
+LANGUAGE plpgsql AS $$
+BEGIN
+    UPDATE push_wake_outbox_community_state state
+    SET retained_rows = GREATEST(0, state.retained_rows - deleted.count)
+    FROM (
+        SELECT community_id, count(*) AS count
+        FROM deleted_push_wakes
+        GROUP BY community_id
+    ) deleted
+    WHERE state.community_id = deleted.community_id;
+    RETURN NULL;
+END
+$$;
+
+CREATE TRIGGER push_wake_outbox_delete_accounting
+AFTER DELETE ON push_wake_outbox
+REFERENCING OLD TABLE AS deleted_push_wakes
+FOR EACH STATEMENT EXECUTE FUNCTION account_push_wake_outbox_deletes();