From a2333210dc688b976d562faf2d183b0fd936a2d6 Mon Sep 17 00:00:00 2001 From: Jordan Mecom Date: Tue, 14 Jul 2026 18:11:13 -0700 Subject: [PATCH] fix(push): reap terminal wake outbox rows (BUZZ-SEC-062) --- crates/buzz-db/src/lib.rs | 13 + crates/buzz-db/src/migration.rs | 10 +- crates/buzz-db/src/push.rs | 325 ++++++++++++++++-- crates/buzz-relay/src/main.rs | 9 + crates/buzz-relay/src/push_runtime.rs | 116 ++++++- docs/push-gateway-deployment.md | 13 + .../0022_push_wake_outbox_retention.sql | 68 ++++ 7 files changed, 518 insertions(+), 36 deletions(-) create mode 100644 migrations/0022_push_wake_outbox_retention.sql diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index eb1fe864a..84c9c95ad 100644 --- a/crates/buzz-db/src/lib.rs +++ b/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, + batch_size: u32, + ) -> Result { + 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, diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index bff6430c3..141a5cdb4 100644 --- a/crates/buzz-db/src/migration.rs +++ b/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] diff --git a/crates/buzz-db/src/push.rs b/crates/buzz-db/src/push.rs index be7a7cc68..a8ac04c62 100644 --- a/crates/buzz-db/src/push.rs +++ b/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 = 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 = 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, + batch_size: u32, ) -> Result { + 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 = + 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" ); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index b11ac3857..e853d71a2 100644 --- a/crates/buzz-relay/src/main.rs +++ b/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 diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index 12c8fd1f0..9f53a1b37 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/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) { } } +/// 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, 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( + period: Duration, + cancel: CancellationToken, + mut tick: Tick, +) where + Tick: FnMut() -> TickFuture, + TickFuture: std::future::Future, +{ + 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(); diff --git a/docs/push-gateway-deployment.md b/docs/push-gateway-deployment.md index 54511a0c1..167cbb7bf 100644 --- a/docs/push-gateway-deployment.md +++ b/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 diff --git a/migrations/0022_push_wake_outbox_retention.sql b/migrations/0022_push_wake_outbox_retention.sql new file mode 100644 index 000000000..a300b32ef --- /dev/null +++ b/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();