From 7d1814e74764f1681dbb82ca21604dc6532e9864 Mon Sep 17 00:00:00 2001 From: Nico Gervois Date: Tue, 10 Nov 2026 10:48:20 -0500 Subject: [PATCH] feat(relay): add durable scheduled message delivery Persist owner-scoped signed messages and dispatch due work atomically. Protect cancellation, retries, replica completeness and community deletion behind a default-off capability. Co-authored-by: Codex Signed-off-by: Nico Gervois --- .env.example | 3 + crates/buzz-db/src/lib.rs | 3 + crates/buzz-db/src/runtime/migration.rs | 3 +- crates/buzz-db/src/runtime/replica_fence.rs | 14 +- crates/buzz-db/src/store/deletion.rs | 18 + crates/buzz-db/src/store/mod.rs | 3 + crates/buzz-db/src/store/scheduled.rs | 301 +++++++++++++ .../src/store/scheduled/postgres_tests.rs | 267 ++++++++++++ crates/buzz-relay/src/api/mod.rs | 2 + .../buzz-relay/src/api/scheduled_messages.rs | 396 ++++++++++++++++++ crates/buzz-relay/src/config.rs | 4 + crates/buzz-relay/src/handlers/ingest.rs | 97 ++++- crates/buzz-relay/src/main.rs | 3 + crates/buzz-relay/src/nip11.rs | 9 + crates/buzz-relay/src/router.rs | 4 + docs/database.md | 8 + docs/scheduled-messages.md | 121 ++++++ migrations/0087_scheduled_messages.sql | 23 + schema/schema.sql | 1 + schema/tables/public/scheduled_messages.sql | 21 + 20 files changed, 1282 insertions(+), 19 deletions(-) create mode 100644 crates/buzz-db/src/store/scheduled.rs create mode 100644 crates/buzz-db/src/store/scheduled/postgres_tests.rs create mode 100644 crates/buzz-relay/src/api/scheduled_messages.rs create mode 100644 docs/scheduled-messages.md create mode 100644 migrations/0087_scheduled_messages.sql create mode 100644 schema/tables/public/scheduled_messages.sql diff --git a/.env.example b/.env.example index eb9a91a259b..f08994e0921 100644 --- a/.env.example +++ b/.env.example @@ -77,6 +77,9 @@ TYPESENSE_URL=http://localhost:8108 BUZZ_BIND_ADDR=0.0.0.0:3000 # Public WebSocket URL — used in NIP-42 auth challenges RELAY_URL=ws://localhost:3000 +# Two-phase rollout required: first replace every replica-fence probe while off. +# See docs/scheduled-messages.md before enabling or downgrading with pending work. +# BUZZ_SCHEDULED_MESSAGES=on # Stable relay signing key (required). `just bootstrap` generates a random key in # the gitignored .env file. Preserve that value across restarts and backups. # BUZZ_RELAY_PRIVATE_KEY=<32-byte hex private key> diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 92eeb42f882..bbc33991d48 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -46,6 +46,9 @@ macro_rules! superseded_with_mentions_sql { mod runtime; mod store; +/// Durable scheduled-message records and dispatch claims. +pub use store::scheduled; + /// Concurrency caps whose limit can change at runtime. pub mod concurrency_gate; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index f7f0888e96f..e285d2ffde5 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -902,7 +902,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 86); + assert_eq!(migrations.len(), 87); assert!(migrations[83] .sql .as_str() @@ -3042,6 +3042,7 @@ mod postgres_tests { .expect("apply migrations for schema parity"); for table in [ + "scheduled_messages", "media_uploaders", "media_publications", "relay_members", diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs index 7278fcc3b2b..fb3b456c2d9 100644 --- a/crates/buzz-db/src/runtime/replica_fence.rs +++ b/crates/buzz-db/src/runtime/replica_fence.rs @@ -610,6 +610,8 @@ fn rand_id() -> [u8; 32] { /// One writer-side sample of the ordered handshake. #[derive(Debug)] struct WriterSample { + /// Below every pending scheduled timestamp (including send-now clock slack). + scheduled_wall: Option>, /// `S`: writer `clock_timestamp()` captured first. sampled_at: DateTime, /// Oldest open transaction among other client backends at scan time, @@ -866,6 +868,12 @@ async fn sample_writer( }) .await?; + let scheduled_wall: Option> = sqlx::query_scalar( + "SELECT to_timestamp(min(send_at)-301) FROM scheduled_messages WHERE status='scheduled'", + ) + .fetch_one(&mut *tx) + .await?; + // 3. Token commit LAST, on the same pinned connection. The single-row // UPDATE serializes concurrent pods' probes, so RETURNING token is // globally commit-ordered. `committed_at` is captured before the @@ -883,6 +891,7 @@ async fn sample_writer( .ok_or(ProbeError::HeartbeatRowMissing)?; let sample = WriterSample { + scheduled_wall, sampled_at, oldest_xact_start, token: row.get("token"), @@ -982,7 +991,10 @@ async fn probe_untraced( timings: &mut ProbeTimings, ) -> Result { let sample = sample_writer(writer, timings).await?; - let wall = fence_wall(sample.sampled_at, sample.oldest_xact_start); + let ordinary_wall = fence_wall(sample.sampled_at, sample.oldest_xact_start); + let wall = sample + .scheduled_wall + .map_or(ordinary_wall, |reservation| ordinary_wall.min(reservation)); match fence.record(sample.token, sample.epoch, sample.committed_at, wall) { RecordOutcome::Recorded => Ok(TokenEntry { token: sample.token, diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 3a3d4e6e4fd..6675dd191a4 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -78,6 +78,7 @@ pub const RELAY_ADMIN_TABLES: &[(&str, &str)] = &[ /// approval and again before PostgreSQL purge. A new tenant table therefore /// blocks deletion until this manifest is intentionally updated. pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ + "scheduled_messages", "nip_fi_profile_heads", "api_tokens", "archived_identities", @@ -132,6 +133,7 @@ const EXCLUDED_SCOPED_TABLES: &[&str] = &[ /// Foreign-key-safe child-before-parent order for the PostgreSQL purge. pub const PURGE_SCOPED_TABLES: &[&str] = &[ + "scheduled_messages", "nip_fi_profile_heads", "personal_read_frontiers", "personal_read_accounts", @@ -8657,6 +8659,12 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .as_uuid() .to_owned(); seed_relay_admin_rows(&db, bystander).await; + for tenant in [community, bystander] { + sqlx::query("INSERT INTO scheduled_messages(community_id,owner,id,event,event_id,send_at,next_attempt_at) VALUES($1,$2,$3,'{}',$4,1,1)") + .bind(tenant).bind([7u8;32].as_slice()).bind(Uuid::new_v4()) + .bind([8u8;32].as_slice()).execute(&db.pool).await.unwrap(); + } + // Operator workflow history is retained even when its provenance // community is purged; it is not a tenant-owned serving table. let entity = db @@ -8762,6 +8770,16 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); serde_json::to_value(approvals_after).unwrap() ); + assert_eq!(deleted.get("scheduled_messages"), Some(&1)); + let survivors: Vec = sqlx::query_scalar( + "SELECT community_id FROM scheduled_messages WHERE community_id IN ($1,$2)", + ) + .bind(community) + .bind(bystander) + .fetch_all(&db.pool) + .await + .unwrap(); + assert_eq!(survivors, vec![bystander]); assert_eq!(deleted.get("relay_admin_outbox"), Some(&2)); assert_eq!(deleted.get("relay_admin_actions"), Some(&2)); assert_eq!(relay_admin_row_counts(&db, community).await, (0, 0)); diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index d1cb647694b..145dd7ce5e8 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -1,5 +1,8 @@ //! Domain-owned persistence implementations. +/// Durable scheduled messages. +pub mod scheduled; + /// Explicit deployment-global admin report reads. pub mod admin_moderation; /// Community-scoped authentication allowlist persistence. diff --git a/crates/buzz-db/src/store/scheduled.rs b/crates/buzz-db/src/store/scheduled.rs new file mode 100644 index 00000000000..d80109f0461 --- /dev/null +++ b/crates/buzz-db/src/store/scheduled.rs @@ -0,0 +1,301 @@ +//! Durable, owner-scoped scheduled delivery. A dispatch holds its row lock until +//! the normal message ingest has returned, serializing cancellation with sending. + +#[cfg(test)] +#[path = "scheduled/postgres_tests.rs"] +mod postgres_tests; + +use buzz_core::CommunityId; +use serde::Serialize; +use sqlx::Row; +use uuid::Uuid; + +use crate::{ + runtime::{begin_community_write_transaction, observability::WriterOperation}, + AdmittedTx, Db, DbError, Result, +}; + +/// A scheduled message; the signed event is never returned to another owner. +#[derive(Debug, Clone, Serialize)] +pub struct ScheduledMessage { + /// Caller-generated idempotency key. + pub id: Uuid, + /// Compare-and-swap revision. + pub revision: i64, + /// Exact user-signed event to publish. + pub event: serde_json::Value, + /// Unix seconds chosen for delivery. + pub send_at: i64, + /// Scheduled, sent, failed or cancelled. + pub status: String, + /// Sanitized delivery failure, if any. + pub error: Option, +} + +fn row_message(row: &sqlx::postgres::PgRow) -> Result { + Ok(ScheduledMessage { + id: row.try_get("id")?, + revision: row.try_get("revision")?, + event: row.try_get("event")?, + send_at: row.try_get("send_at")?, + status: row.try_get("status")?, + error: row.try_get("error")?, + }) +} + +fn conflict() -> DbError { + DbError::InvalidData("scheduled-message-conflict".into()) +} + +/// Worker-owned admitted transaction. Dropping it rolls back the claim. +pub struct ScheduledDispatch { + tx: Option, + pool: sqlx::PgPool, + community: CommunityId, + owner: Vec, + /// Exact queued message. + pub message: ScheduledMessage, + attempts: i32, +} + +impl ScheduledDispatch { + /// Community admitted by the held transaction. + pub fn community(&self) -> CommunityId { + self.community + } + /// Commit delivery, a permanent refusal, or a bounded transient backoff. + pub async fn finish(mut self, status: &str, error: Option<&str>, now: i64) -> Result<()> { + if !matches!(status, "sent" | "failed" | "scheduled") { + return Err(conflict()); + } + let mut tx = if let Some(tx) = self.tx.take() { + tx + } else { + // Persistence failure rolls its transaction back. Reacquire only + // this unchanged pending revision to record retry state; a cancel, + // edit, or successful concurrent delivery must win unchanged. + if status != "scheduled" { + return Ok(()); + } + let mut tx = begin_community_write_transaction( + &self.pool, + self.community, + WriterOperation::EventWrite, + ) + .await?; + let row = sqlx::query("SELECT attempts FROM scheduled_messages WHERE community_id=$1 AND owner=$2 AND id=$3 AND revision=$4 AND event=$5 AND status='scheduled' FOR UPDATE SKIP LOCKED") + .bind(self.community.as_uuid()).bind(&self.owner).bind(self.message.id) + .bind(self.message.revision).bind(&self.message.event).fetch_optional(tx.conn()).await?; + let Some(row) = row else { + return Ok(()); + }; + self.attempts = row.try_get("attempts")?; + tx + }; + let delay = 2_i64.pow((self.attempts as u32).min(8)).min(300); + sqlx::query("UPDATE scheduled_messages SET status=$4, error=$5, attempts=attempts+1, next_attempt_at=$6, updated_at=now() WHERE community_id=$1 AND owner=$2 AND id=$3") + .bind(self.community.as_uuid()).bind(&self.owner).bind(self.message.id).bind(status).bind(error).bind(now+delay) + .execute(tx.conn()).await?; + tx.commit().await + } + + /// Store the approved message, mentions, thread counters, push work and sent + /// state on the claim transaction. A lost claim can never publish afterward. + /// Pending queue timestamps reserve the replica fence until this commit. + pub async fn persist( + &mut self, + db: &Db, + event: &nostr::Event, + channel: Option, + thread: Option>, + ) -> Result<(buzz_core::StoredEvent, bool)> { + if event.kind.as_u16() != 9 || serde_json::to_value(event)? != self.message.event { + return Err(conflict()); + } + let mut tx = self.tx.take().ok_or_else(conflict)?; + // Only this claim's approved old timestamp is permitted. The probe caps + // its wall below all pending reservations before publishing a heartbeat. + sqlx::query("SELECT set_config('buzz.created_at_floor', GREATEST(960,ceil(extract(epoch FROM clock_timestamp())-$1)+960)::bigint::text,true)") + .bind(event.created_at.as_secs() as i64).execute(tx.conn()).await?; + tx.set_push_enqueue(db.push_enqueue.clone()); + let result = + crate::event::insert_event_with_thread_metadata_tx(&mut tx, event, channel, thread) + .await?; + if result.1 { + crate::insert_mentions_in_transaction(&mut tx, event, channel).await?; + } + sqlx::query("UPDATE scheduled_messages SET status='sent',error=NULL,updated_at=now() WHERE community_id=$1 AND owner=$2 AND id=$3") + .bind(self.community.as_uuid()).bind(&self.owner).bind(self.message.id).execute(tx.conn()).await?; + tx.commit().await?; + Ok(result) + } +} + +impl Db { + /// Return all active items (capped at creation) and the last 100 sent items. + pub async fn scheduled_list( + &self, + community: CommunityId, + owner: &[u8], + ) -> Result> { + let rows = sqlx::query("(SELECT * FROM scheduled_messages WHERE community_id=$1 AND owner=$2 AND status IN ('scheduled','failed') ORDER BY send_at LIMIT 256) UNION ALL (SELECT * FROM scheduled_messages WHERE community_id=$1 AND owner=$2 AND status='sent' ORDER BY send_at DESC LIMIT 100)") + .bind(community.as_uuid()).bind(owner).fetch_all(&self.pool).await?; + rows.iter().map(row_message).collect() + } + + /// Insert/update/cancel under the owner's lock, using an explicit revision. + /// Identical retries succeed after an acknowledgement was lost. + #[allow(clippy::too_many_arguments)] + pub async fn scheduled_mutate( + &self, + community: CommunityId, + owner: &[u8], + id: Uuid, + expected: Option, + event: Option<&serde_json::Value>, + send_at: i64, + ) -> Result<()> { + let mut tx = + begin_community_write_transaction(&self.pool, community, WriterOperation::EventWrite) + .await?; + let key = format!("scheduled:{community}:{}", hex::encode(owner)); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1,0))") + .bind(key) + .execute(tx.conn()) + .await?; + let existing = sqlx::query("SELECT * FROM scheduled_messages WHERE community_id=$1 AND owner=$2 AND id=$3 FOR UPDATE") + .bind(community.as_uuid()).bind(owner).bind(id).fetch_optional(tx.conn()).await?; + let event_id = event + .map(|v| { + v["id"] + .as_str() + .and_then(|s| hex::decode(s).ok()) + .ok_or_else(conflict) + }) + .transpose()?; + if let Some(row) = existing { + let current = row_message(&row)?; + let stored: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2)", + ) + .bind(community.as_uuid()) + .bind(row.try_get::, _>("event_id")?) + .fetch_one(tx.conn()) + .await?; + if stored && current.status != "sent" { + sqlx::query("UPDATE scheduled_messages SET status='sent',error=NULL,updated_at=now() WHERE community_id=$1 AND owner=$2 AND id=$3") + .bind(community.as_uuid()).bind(owner).bind(id).execute(tx.conn()).await?; + tx.commit().await?; + return Err(conflict()); + } + if current.status != "cancelled" + && event.is_some_and(|v| *v == current.event) + && current.send_at == send_at + { + return tx.commit().await; + } + if event.is_none() + && current.status == "cancelled" + && expected == Some(current.revision - 1) + { + return tx.commit().await; + } + if expected != Some(current.revision) + || !matches!(current.status.as_str(), "scheduled" | "failed") + { + return Err(conflict()); + } + validate_reservation(&mut tx, event, send_at).await?; + sqlx::query("UPDATE scheduled_messages SET revision=revision+1, event=COALESCE($4,event), event_id=COALESCE($5,event_id), send_at=CASE WHEN $4 IS NULL THEN send_at ELSE $6 END, next_attempt_at=$6, status=$7, error=NULL, attempts=0, updated_at=now() WHERE community_id=$1 AND owner=$2 AND id=$3") + .bind(community.as_uuid()).bind(owner).bind(id).bind(event).bind(event_id).bind(send_at) + .bind(if event.is_some() { "scheduled" } else { "cancelled" }).execute(tx.conn()).await?; + } else { + if expected.is_some() || event.is_none() { + return Err(conflict()); + } + validate_reservation(&mut tx, event, send_at).await?; + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM scheduled_messages WHERE community_id=$1 AND owner=$2 AND status IN ('scheduled','failed')") + .bind(community.as_uuid()).bind(owner).fetch_one(tx.conn()).await?; + if count >= 256 { + return Err(DbError::InvalidData("scheduled-message-limit".into())); + } + let inserted = sqlx::query("INSERT INTO scheduled_messages(community_id,owner,id,event,event_id,send_at,next_attempt_at) VALUES($1,$2,$3,$4,$5,$6,$6) ON CONFLICT (community_id,owner,event_id) DO NOTHING") + .bind(community.as_uuid()).bind(owner).bind(id).bind(event).bind(event_id).bind(send_at).execute(tx.conn()).await?; + if inserted.rows_affected() != 1 { + return Err(conflict()); + } + } + tx.commit().await + } + + /// Bounded fleet scan; each candidate is rechecked under admission and lock. + pub async fn scheduled_due( + &self, + now: i64, + ) -> Result, Uuid)>> { + let rows = sqlx::query("SELECT s.community_id,c.host,s.owner,s.id FROM scheduled_messages s JOIN communities c ON c.id=s.community_id WHERE s.status='scheduled' AND s.next_attempt_at<=$1 AND c.deletion_state='active' ORDER BY s.next_attempt_at LIMIT 100") + .bind(now).fetch_all(&self.pool).await?; + rows.iter() + .map(|r| { + Ok(( + CommunityId::from_uuid(r.try_get("community_id")?), + r.try_get("host")?, + r.try_get("owner")?, + r.try_get("id")?, + )) + }) + .collect() + } + + /// Claim a due item on one worker. Other workers skip the held row. + pub async fn scheduled_claim( + &self, + community: CommunityId, + owner: &[u8], + id: Uuid, + now: i64, + ) -> Result> { + let mut tx = + begin_community_write_transaction(&self.pool, community, WriterOperation::EventWrite) + .await?; + let row = sqlx::query("SELECT * FROM scheduled_messages WHERE community_id=$1 AND owner=$2 AND id=$3 AND status='scheduled' AND next_attempt_at<=$4 FOR UPDATE SKIP LOCKED") + .bind(community.as_uuid()).bind(owner).bind(id).bind(now).fetch_optional(tx.conn()).await?; + row.map(|r| { + Ok(ScheduledDispatch { + message: row_message(&r)?, + attempts: r.try_get("attempts")?, + tx: Some(tx), + pool: self.pool.clone(), + community, + owner: owner.to_vec(), + }) + }) + .transpose() + } +} + +// A new reservation must be newer than every previously proved fence wall. +// Use the writer's clock, including after lock waits; host clock skew must not +// admit an event behind an already published replica completeness proof. +async fn validate_reservation( + tx: &mut AdmittedTx, + event: Option<&serde_json::Value>, + send_at: i64, +) -> Result<()> { + let Some(event) = event else { + return Ok(()); + }; + let created = event["created_at"].as_i64().ok_or_else(conflict)?; + let now: i64 = + sqlx::query_scalar("SELECT floor(extract(epoch FROM clock_timestamp()))::bigint") + .fetch_one(tx.conn()) + .await?; + if send_at < now - 300 + || send_at > now + 120 * 86_400 + || created < send_at - 300 + || created > send_at + 5 + { + return Err(conflict()); + } + Ok(()) +} diff --git a/crates/buzz-db/src/store/scheduled/postgres_tests.rs b/crates/buzz-db/src/store/scheduled/postgres_tests.rs new file mode 100644 index 00000000000..a4e8ab243e4 --- /dev/null +++ b/crates/buzz-db/src/store/scheduled/postgres_tests.rs @@ -0,0 +1,267 @@ +use super::*; +use nostr::{Event, EventBuilder, Keys, Kind, Tag, Timestamp}; + +async fn fixture() -> (Db, crate::test_support::ScratchDb, CommunityId, Keys, Uuid) { + let (pool, scratch) = + crate::test_support::ScratchDb::migrated(&crate::test_support::database_url(), "scheduled") + .await; + let db = Db::from_pool(pool); + let community = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities(id,host) VALUES($1,$2)") + .bind(community.as_uuid()) + .bind(format!("scheduled-{}.example", community.as_uuid())) + .execute(db.pool()) + .await + .unwrap(); + let keys = Keys::generate(); + let channel = db + .create_channel( + community, + "scheduled", + buzz_core::channel::ChannelType::Stream, + buzz_core::channel::ChannelVisibility::Open, + None, + &keys.public_key().to_bytes(), + None, + ) + .await + .unwrap(); + (db, scratch, community, keys, channel.id) +} +fn queued_event(keys: &Keys, channel: Uuid, at: i64) -> Event { + EventBuilder::new(Kind::Custom(9), "A scheduled message") + .tags([Tag::parse(["h".to_string(), channel.to_string()]).unwrap()]) + .custom_created_at(Timestamp::from(at as u64)) + .sign_with_keys(keys) + .unwrap() +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn scheduled_owner_isolation_revision_and_cancelled_duplicate() { + let (db, _scratch, c, keys, channel) = fixture().await; + let at = chrono::Utc::now().timestamp() + 3600; + let value = serde_json::to_value(queued_event(&keys, channel, at)).unwrap(); + let id = Uuid::new_v4(); + let owner = keys.public_key().to_bytes(); + db.scheduled_mutate(c, &owner, id, None, Some(&value), at) + .await + .unwrap(); + db.scheduled_mutate(c, &owner, id, None, Some(&value), at) + .await + .unwrap(); + assert_eq!(db.scheduled_list(c, &owner).await.unwrap().len(), 1); + assert!(db.scheduled_list(c, &[7; 32]).await.unwrap().is_empty()); + assert!(db + .scheduled_mutate(c, &owner, id, Some(2), None, at) + .await + .is_err()); + db.scheduled_mutate(c, &owner, id, Some(1), None, at) + .await + .unwrap(); + assert!(db.scheduled_list(c, &owner).await.unwrap().is_empty()); + assert!(db + .scheduled_mutate(c, &owner, Uuid::new_v4(), None, Some(&value), at) + .await + .is_err()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn scheduled_overdue_dispatch_atomically_stores_message_and_sent_state() { + let (db, _scratch, c, keys, channel) = fixture().await; + let at = chrono::Utc::now().timestamp() - 7200; + let event = queued_event(&keys, channel, at); + let value = serde_json::to_value(&event).unwrap(); + let id = Uuid::new_v4(); + let owner = keys.public_key().to_bytes(); + assert!(db + .scheduled_mutate(c, &owner, id, None, Some(&value), at) + .await + .is_err()); + let future = chrono::Utc::now().timestamp() + 3600; + let initial = serde_json::to_value(queued_event(&keys, channel, future)).unwrap(); + db.scheduled_mutate(c, &owner, id, None, Some(&initial), future) + .await + .unwrap(); + // Simulate elapsed downtime without waiting hours. Only the fixture may + // bypass admission; production cannot introduce a retroactive reservation. + sqlx::query("UPDATE scheduled_messages SET event=$2,event_id=$3,send_at=$4,next_attempt_at=$4 WHERE id=$1") + .bind(id).bind(&value).bind(event.id.to_bytes().as_slice()).bind(at).execute(db.pool()).await.unwrap(); + let fence = crate::replica_fence::ReplicaFence::new(); + let reserved = crate::replica_fence::probe_once(db.pool(), &fence) + .await + .unwrap(); + assert_eq!(reserved.fence_wall.timestamp(), at - 301); + assert!(!fence.covers(chrono::DateTime::from_timestamp(at, 0).unwrap())); + let mut claim = db + .scheduled_claim(c, &owner, id, chrono::Utc::now().timestamp()) + .await + .unwrap() + .unwrap(); + sqlx::query("SET LOCAL buzz.created_at_floor='960'") + .execute(claim.tx.as_mut().unwrap().conn()) + .await + .unwrap(); + let (stored, inserted) = claim + .persist(&db, &event, Some(channel), None) + .await + .unwrap(); + assert_eq!(stored.event.id, event.id); + assert!(inserted); + let rows = db.scheduled_list(c, &owner).await.unwrap(); + assert_eq!(rows[0].status, "sent"); + let released = crate::replica_fence::probe_once(db.pool(), &fence) + .await + .unwrap(); + assert!(released.token > reserved.token); + assert!(released.fence_wall.timestamp() > at); + assert!(db + .get_event_by_id(c, &event.id.to_bytes()) + .await + .unwrap() + .is_some()); + assert!(db + .scheduled_claim(c, &owner, id, chrono::Utc::now().timestamp()) + .await + .unwrap() + .is_none()); + assert!(db + .scheduled_mutate(c, &owner, id, Some(1), None, at) + .await + .is_err()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn scheduled_claim_rollback_allows_recovery_and_excludes_other_workers() { + let (db, _scratch, c, keys, channel) = fixture().await; + let at = chrono::Utc::now().timestamp(); + let value = serde_json::to_value(queued_event(&keys, channel, at)).unwrap(); + let id = Uuid::new_v4(); + let owner = keys.public_key().to_bytes(); + db.scheduled_mutate(c, &owner, id, None, Some(&value), at) + .await + .unwrap(); + let claim = db + .scheduled_claim(c, &owner, id, at) + .await + .unwrap() + .unwrap(); + assert!(db + .scheduled_claim(c, &owner, id, at) + .await + .unwrap() + .is_none()); + drop(claim); + let claim = db + .scheduled_claim(c, &owner, id, at) + .await + .unwrap() + .unwrap(); + claim + .finish("scheduled", Some("transient failure"), at) + .await + .unwrap(); + assert!(db + .scheduled_claim(c, &owner, id, at) + .await + .unwrap() + .is_none()); + let mut recovered = db + .scheduled_claim(c, &owner, id, at + 10) + .await + .unwrap() + .unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(recovered.tx.as_mut().unwrap().conn()) + .await + .unwrap(); + sqlx::query("SELECT pg_terminate_backend($1)") + .bind(pid) + .execute(db.pool()) + .await + .unwrap(); + let event: Event = serde_json::from_value(value.clone()).unwrap(); + assert!(recovered + .persist(&db, &event, Some(channel), None) + .await + .is_err()); + recovered + .finish("scheduled", Some("Persistence interrupted"), at + 10) + .await + .unwrap(); + let (attempts, retry): (i32, i64) = + sqlx::query_as("SELECT attempts,next_attempt_at FROM scheduled_messages WHERE id=$1") + .bind(id) + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(attempts, 2); + assert_eq!(retry, at + 12); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn scheduled_lost_claim_cannot_publish_after_successful_cancellation() { + let (db, _scratch, c, keys, channel) = fixture().await; + let at = chrono::Utc::now().timestamp(); + let event = queued_event(&keys, channel, at); + let value = serde_json::to_value(&event).unwrap(); + let id = Uuid::new_v4(); + let owner = keys.public_key().to_bytes(); + db.scheduled_mutate(c, &owner, id, None, Some(&value), at) + .await + .unwrap(); + let mut claim = db + .scheduled_claim(c, &owner, id, at) + .await + .unwrap() + .unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(claim.tx.as_mut().unwrap().conn()) + .await + .unwrap(); + sqlx::query("SELECT pg_terminate_backend($1)") + .bind(pid) + .execute(db.pool()) + .await + .unwrap(); + db.scheduled_mutate(c, &owner, id, Some(1), None, at) + .await + .unwrap(); + assert!(claim + .persist(&db, &event, Some(channel), None) + .await + .is_err()); + assert!(db + .get_event_by_id(c, &event.id.to_bytes()) + .await + .unwrap() + .is_none()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn scheduled_writers_and_claims_honor_community_deletion_admission() { + let (db, _scratch, c, keys, channel) = fixture().await; + let at = chrono::Utc::now().timestamp(); + let value = serde_json::to_value(queued_event(&keys, channel, at)).unwrap(); + let id = Uuid::new_v4(); + let owner = keys.public_key().to_bytes(); + db.scheduled_mutate(c, &owner, id, None, Some(&value), at) + .await + .unwrap(); + crate::test_support::set_deletion_state(db.pool(), *c.as_uuid(), "fenced").await; + assert!(db.scheduled_claim(c, &owner, id, at).await.is_err()); + assert!(db + .scheduled_mutate(c, &owner, id, Some(1), None, at) + .await + .is_err()); + assert!(db + .scheduled_due(at) + .await + .unwrap() + .iter() + .all(|(community, _, _, _)| *community != c)); +} diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index b0641a09ce9..a33e387683d 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/crates/buzz-relay/src/api/mod.rs @@ -15,6 +15,8 @@ pub mod nip05; pub mod nip_fi; pub mod operator; pub mod profiles; +/// Owner-only scheduled message queue. +pub mod scheduled_messages; pub mod workflows; // Re-export imeta helpers used by ingest pipeline. diff --git a/crates/buzz-relay/src/api/scheduled_messages.rs b/crates/buzz-relay/src/api/scheduled_messages.rs new file mode 100644 index 00000000000..70766bb9855 --- /dev/null +++ b/crates/buzz-relay/src/api/scheduled_messages.rs @@ -0,0 +1,396 @@ +//! Owner-only scheduled delivery. The queue is relay-owned execution state; +//! dispatched messages still use the ordinary signed-event ingest pipeline. + +use crate::{ + api::{api_error, bridge::authenticate_member_post, internal_error}, + state::AppState, +}; +use axum::{ + body::Bytes, + extract::State, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, + Json, +}; +use buzz_core::TenantContext; +use nostr::Event; +use serde::Deserialize; +use serde_json::{json, Value}; +use std::sync::Arc; +use uuid::Uuid; + +/// Fixed, NIP-98-authenticated queue route. +pub const PATH: &str = "/scheduled-messages"; +const MAX_DAYS: i64 = 120 * 86_400; + +#[derive(Deserialize)] +#[serde(tag = "action", rename_all = "snake_case", deny_unknown_fields)] +enum Request { + List {}, + Create { + id: Uuid, + event: Event, + }, + Update { + id: Uuid, + revision: i64, + event: Event, + }, + Cancel { + id: Uuid, + revision: i64, + }, + SendNow { + id: Uuid, + revision: i64, + event: Event, + }, +} + +fn validate_message( + event: &Event, + owner: nostr::PublicKey, + immediate: bool, + now: i64, +) -> Result<(Uuid, i64), &'static str> { + if event.pubkey != owner + || event.kind.as_u16() != 9 + || event.content.trim().is_empty() + || event.content.len() > 24 * 1024 + { + return Err("Invalid scheduled message"); + } + let at = i64::try_from(event.created_at.as_secs()).map_err(|_| "Invalid send time")?; + if (immediate && !(now - 300..=now + 5).contains(&at)) + || (!immediate && (at <= now || at > now + MAX_DAYS)) + { + return Err("Choose a future time within 120 days"); + } + let channels: Vec<_> = event + .tags + .iter() + .filter(|t| t.as_slice().first().is_some_and(|s| s == "h")) + .collect(); + let [channel] = channels.as_slice() else { + return Err("Choose one conversation"); + }; + let id = channel + .as_slice() + .get(1) + .and_then(|s| Uuid::parse_str(s).ok()) + .ok_or("Invalid conversation")?; + if serde_json::to_vec(event).map_or(true, |v| v.len() > 32 * 1024) { + return Err("Message is too large"); + } + Ok((id, at)) +} + +/// Handle bounded queue reads and compare-and-swap mutations for the proved owner. +pub async fn handle( + State(state): State>, + slot: crate::access::AccessSlot, + headers: HeaderMap, + body: Bytes, +) -> Response { + if !state.config.scheduled_messages_enabled { + return api_error(StatusCode::NOT_FOUND, "Scheduled delivery is not enabled") + .into_response(); + } + if body.len() > 40 * 1024 { + return api_error(StatusCode::PAYLOAD_TOO_LARGE, "Message is too large").into_response(); + } + let (tenant, owner) = match authenticate_member_post(&state, &slot, &headers, PATH, &body).await + { + Ok(v) => v, + Err(e) => return e, + }; + let request: Request = match serde_json::from_slice(&body) { + Ok(v) => v, + Err(_) => { + return api_error(StatusCode::BAD_REQUEST, "Invalid scheduled-message request") + .into_response() + } + }; + let owner_bytes = owner.to_bytes(); + if matches!(request, Request::List {}) { + return match state + .db + .scheduled_list(tenant.community(), &owner_bytes) + .await + { + Ok(messages) => { + Json(json!({"messages":messages,"sent_history_limit":100})).into_response() + } + Err(_) => internal_error("Could not read scheduled messages").into_response(), + }; + } + let (id, revision, event, immediate) = match request { + Request::Create { id, event } => (id, None, Some(event), false), + Request::Update { + id, + revision, + event, + } => (id, Some(revision), Some(event), false), + Request::Cancel { id, revision } => (id, Some(revision), None, false), + Request::SendNow { + id, + revision, + event, + } => (id, Some(revision), Some(event), true), + Request::List {} => { + return api_error(StatusCode::BAD_REQUEST, "Invalid operation").into_response() + } + }; + let now = chrono::Utc::now().timestamp(); + let mut at = now; + if let Some(event) = &event { + let (channel, send_at) = match validate_message(event, owner, immediate, now) { + Ok(v) => v, + Err(reason) => return api_error(StatusCode::BAD_REQUEST, reason).into_response(), + }; + at = if immediate { now } else { send_at }; + let verify = event.clone(); + if !tokio::task::spawn_blocking(move || { + buzz_core::verification::verify_event(&verify).is_ok() + }) + .await + .unwrap_or(false) + { + return api_error(StatusCode::BAD_REQUEST, "Invalid message signature").into_response(); + } + match state + .db + .get_member_role(tenant.community(), channel, &owner_bytes) + .await + { + Ok(Some(role)) if role != "guest" => {} + Ok(_) => { + return api_error( + StatusCode::FORBIDDEN, + "You cannot post in this conversation", + ) + .into_response() + } + Err(_) => return internal_error("Could not check conversation access").into_response(), + } + if immediate { + let items = match state + .db + .scheduled_list(tenant.community(), &owner_bytes) + .await + { + Ok(v) => v, + Err(_) => { + return internal_error("Could not read scheduled message").into_response() + } + }; + let matches = items.iter().any(|item| { + item.id == id + && item.event["content"] == event.content + && item.event["tags"] + == serde_json::to_value(&event.tags).unwrap_or(Value::Null) + }); + if !matches { + return api_error( + StatusCode::CONFLICT, + "Message changed; refresh before sending", + ) + .into_response(); + } + } + } + let value = event.as_ref().map(serde_json::to_value).transpose(); + let value = match value { + Ok(v) => v, + Err(_) => return api_error(StatusCode::BAD_REQUEST, "Invalid message").into_response(), + }; + match state + .db + .scheduled_mutate( + tenant.community(), + &owner_bytes, + id, + revision, + value.as_ref(), + at, + ) + .await + { + Ok(()) => Json(json!({"accepted":true})).into_response(), + Err(buzz_db::DbError::InvalidData(_)) => api_error( + StatusCode::CONFLICT, + "Message changed or the queue is full; refresh and try again", + ) + .into_response(), + Err(_) => internal_error("Could not update scheduled message").into_response(), + } +} + +/// One bounded sweep, shared by the daemon and live integration tests. +pub async fn dispatch_due(state: &Arc) -> buzz_db::Result { + let now = chrono::Utc::now().timestamp(); + let due = state.db.scheduled_due(now).await?; + let mut sent = 0; + for (community, host, owner, id) in due { + let Some(mut claim) = state.db.scheduled_claim(community, &owner, id, now).await? else { + continue; + }; + let event: Event = match serde_json::from_value(claim.message.event.clone()) { + Ok(event) => event, + Err(_) => { + claim + .finish( + "failed", + Some("Stored message is invalid. Cancel and recreate it."), + now, + ) + .await?; + continue; + } + }; + let tenant = TenantContext::resolved(community, host); + let stored = state + .db + .get_event_by_id_including_deleted(community, &event.id.to_bytes()) + .await?; + if stored.is_some() { + claim.finish("sent", None, now).await?; + sent += 1; + continue; + } + let member = crate::api::relay_members::enforce_relay_membership( + state, + &tenant, + &owner, + None, + Some(now as u64), + None, + ) + .await; + if let Err(refusal) = member { + if refusal.status().is_server_error() { + claim + .finish( + "scheduled", + Some("Access check delayed; the relay will retry."), + now, + ) + .await?; + continue; + } + claim + .finish( + "failed", + Some("Community access could not be confirmed. Review before retrying."), + now, + ) + .await?; + continue; + } + let outcome = crate::handlers::ingest::ingest_scheduled( + state, + &tenant, + &mut claim, + crate::handlers::ingest::IngestAuth::Http { + pubkey: event.pubkey, + scopes: buzz_auth::Scope::all_known(), + auth_method: crate::handlers::ingest::HttpAuthMethod::Nip98, + }, + ) + .await; + match outcome { + Ok(result) if result.accepted => { + claim.finish("sent", None, now).await?; + sent += 1; + } + Err(crate::handlers::ingest::IngestError::Internal(_)) + | Err(crate::handlers::ingest::IngestError::Ban { + refusal: crate::handlers::ingest::BanRefusal::Unavailable, + .. + }) => { + claim + .finish( + "scheduled", + Some("Delivery delayed; the relay will retry."), + now, + ) + .await?; + } + _ => { + claim.finish("failed",Some("Delivery was refused. Review conversation access and message before retrying."),now).await?; + } + } + } + Ok(sent) +} + +/// Run even when every client is closed. Row locks coordinate relay replicas. +pub async fn run(state: Arc) { + tokio::time::sleep(std::time::Duration::from_millis( + (Uuid::new_v4().as_u128() % 1000) as u64, + )) + .await; + let mut ticks = tokio::time::interval(std::time::Duration::from_secs(1)); + ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + ticks.tick().await; + if dispatch_due(&state).await.is_err() { + tracing::warn!("Scheduled delivery sweep failed"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + #[test] + fn validates_owner_time_and_single_conversation() { + let keys = Keys::generate(); + let other = Keys::generate(); + let channel = Uuid::new_v4(); + let now = 1_800_000_000; + let build = |kind, at, content: &str, tags: Vec| { + EventBuilder::new(Kind::Custom(kind), content) + .tags(tags) + .custom_created_at(Timestamp::from(at as u64)) + .sign_with_keys(&keys) + .unwrap() + }; + let tags = || vec![Tag::parse(["h".to_string(), channel.to_string()]).unwrap()]; + let good = build(9, now + 3600, "Tomorrow", tags()); + assert_eq!( + validate_message(&good, keys.public_key(), false, now), + Ok((channel, now + 3600)) + ); + assert!(validate_message(&good, other.public_key(), false, now).is_err()); + for bad in [ + build(1, now + 3600, "Wrong kind", tags()), + build(9, now, "Past", tags()), + build(9, now + MAX_DAYS + 1, "Too far", tags()), + build(9, now + 3600, " ", tags()), + build(9, now + 3600, "Missing channel", vec![]), + build(9, now + 3600, "Two channels", [tags(), tags()].concat()), + build(9, now + 3600, &"x".repeat(24 * 1024 + 1), tags()), + ] { + assert!(validate_message(&bad, keys.public_key(), false, now).is_err()); + } + for (at, valid) in [ + (now - 301, false), + (now - 300, true), + (now + 5, true), + (now + 6, false), + ] { + assert_eq!( + validate_message(&build(9, at, "Now", tags()), keys.public_key(), true, now) + .is_ok(), + valid + ); + } + assert!(serde_json::from_value::( + json!({"action":"list","owner":other.public_key().to_hex()}) + ) + .is_err()); + } +} diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 76dee7b7b63..02c3e73e49e 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -183,6 +183,8 @@ pub struct Config { /// Whether REST API requests must present a valid token. Independent of /// WebSocket protocol auth, which is *always* required by REQ/EVENT/COUNT. pub require_auth_token: bool, + /// Enable scheduled delivery only after the reservation-aware fleet rollout. + pub scheduled_messages_enabled: bool, /// Opt-in private accessory API; disabled until explicitly deployed. pub buzz_v1_enabled: bool, /// Opt-in NIP-CL compact membership projection; dark by default. @@ -1495,6 +1497,8 @@ impl Config { client_ip_header, auth, require_auth_token, + scheduled_messages_enabled: std::env::var("BUZZ_SCHEDULED_MESSAGES") + .is_ok_and(|v| v == "on"), buzz_v1_enabled, membership_snapshot_enabled: std::env::var("BUZZ_MEMBERSHIP_SNAPSHOT_ENABLED") .is_ok_and(|v| v == "true"), diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 46a1c81f282..d87fff57dbc 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -2460,6 +2460,35 @@ pub async fn ingest_event( tenant: &TenantContext, event: Event, auth: IngestAuth, +) -> Result { + ingest_event_with_due_time(state, tenant, event, auth, None).await +} + +/// Only a locked, admitted queue claim can authorize a late signed timestamp. +/// Future timestamps and every other ingest/auth check remain unchanged. +pub(crate) async fn ingest_scheduled( + state: &Arc, + tenant: &TenantContext, + claim: &mut buzz_db::scheduled::ScheduledDispatch, + auth: IngestAuth, +) -> Result { + let event: Event = serde_json::from_value(claim.message.event.clone()) + .map_err(|_| IngestError::Rejected("invalid scheduled event".into()))?; + if claim.community() != tenant.community() + || event.kind.as_u16() != 9 + || claim.message.send_at > chrono::Utc::now().timestamp() + { + return Err(IngestError::Rejected("invalid scheduled claim".into())); + } + ingest_event_with_due_time(state, tenant, event, auth, Some(claim)).await +} + +async fn ingest_event_with_due_time( + state: &Arc, + tenant: &TenantContext, + event: Event, + auth: IngestAuth, + scheduled: Option<&mut buzz_db::scheduled::ScheduledDispatch>, ) -> Result { // Captured before `event` moves into the inner fn: the stored-events // counter below is emitted at this shared seam so WebSocket and HTTP @@ -2476,7 +2505,7 @@ pub async fn ingest_event( "ingest_event_exited_without_trace", ); - let result = ingest_event_inner(state, &tracer, tenant, event, auth).await; + let result = ingest_event_inner(state, &tracer, tenant, event, auth, scheduled).await; // Fleet-wide stored counter: kind + author_type only, no community tag // (see the cardinality rationale on buzz_events_received_total — @@ -2576,6 +2605,7 @@ async fn ingest_event_inner( tenant: &TenantContext, event: Event, auth: IngestAuth, + scheduled: Option<&mut buzz_db::scheduled::ScheduledDispatch>, ) -> Result { let event_id_hex = event.id.to_hex(); let kind_u32 = event_kind_u32(&event); @@ -2670,7 +2700,9 @@ async fn ingest_event_inner( const MAX_TIMESTAMP_DRIFT_SECS: i64 = 900; // ±15 minutes let now = chrono::Utc::now().timestamp(); let event_ts = event.created_at.as_secs() as i64; - if (event_ts - now).abs() > MAX_TIMESTAMP_DRIFT_SECS { + if event_ts > now + MAX_TIMESTAMP_DRIFT_SECS + || (scheduled.is_none() && event_ts < now - MAX_TIMESTAMP_DRIFT_SECS) + { return Err(IngestError::Rejected( "invalid: event timestamp too far from server time".into(), )); @@ -3749,6 +3781,18 @@ async fn ingest_event_inner( buzz_db::ChannelHeadWriteStatus::Inserted => (stored_event, true), buzz_db::ChannelHeadWriteStatus::Duplicate => (stored_event, false), } + } else if let Some(claim) = scheduled { + claim + .persist( + &state.db, + &event, + channel_id, + thread_meta.as_ref().map(|m| m.as_params()), + ) + .await + .map_err(|error| { + IngestError::Internal(format!("scheduled persistence: {}", error.kind())) + })? } else { let thread_params = thread_meta.as_ref().map(|m| m.as_params()); match state @@ -6723,7 +6767,7 @@ mod postgres_tests { let tracer: Arc = Arc::new(VecTracer::default()); // ── Submit through the real ingest path ─────────────────────────────── - let result = ingest_event_inner(&state, &tracer, &tenant, event, auth).await; + let result = ingest_event_inner(&state, &tracer, &tenant, event, auth, None).await; // The canvas-specific guard must fire before the h-tag check. // created_at = relay_now + 600 is 300 s above the canvas ceiling, so @@ -7109,13 +7153,13 @@ mod postgres_tests { .membership_cache .seed_member(member_key.clone(), true); let control = - ingest_event_inner(&healthy, &tracer, &tenant, post("control"), auth()).await; + ingest_event_inner(&healthy, &tracer, &tenant, post("control"), auth(), None).await; // Only the channel lookup fails: the post must be denied and not stored. failing.membership_cache.seed_member(member_key, true); let event = post("lookup fails"); let event_id = event.id.to_bytes().to_vec(); - let denied = ingest_event_inner(&failing, &tracer, &tenant, event, auth()).await; + let denied = ingest_event_inner(&failing, &tracer, &tenant, event, auth(), None).await; let stored: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE id = $1") .bind(&event_id) .fetch_one(&admin) @@ -7318,7 +7362,7 @@ mod postgres_tests { .expect("sign first canvas"); let first_id_hex = first.id.to_hex(); - ingest_event_inner(&state, &tracer, &tenant, first, make_auth(&author)) + ingest_event_inner(&state, &tracer, &tenant, first, make_auth(&author), None) .await .expect("first canvas write must succeed"); @@ -7332,7 +7376,7 @@ mod postgres_tests { .sign_with_keys(&author) .expect("sign second canvas"); - ingest_event_inner(&state, &tracer, &tenant, second, make_auth(&author)) + ingest_event_inner(&state, &tracer, &tenant, second, make_auth(&author), None) .await .expect("second canvas write must succeed"); @@ -7351,7 +7395,9 @@ mod postgres_tests { let stale_id_bytes = stale.id.as_bytes().to_vec(); let err = - match ingest_event_inner(&state, &tracer, &tenant, stale, make_auth(&author)).await { + match ingest_event_inner(&state, &tracer, &tenant, stale, make_auth(&author), None) + .await + { Ok(_) => panic!("stale precondition must be rejected, but ingest returned Ok"), Err(e) => e, }; @@ -7382,7 +7428,7 @@ mod postgres_tests { .sign_with_keys(&author) .expect("sign untagged canvas"); - ingest_event_inner(&state, &tracer, &tenant, untagged, make_auth(&author)) + ingest_event_inner(&state, &tracer, &tenant, untagged, make_auth(&author), None) .await .expect("untagged canvas write must append unconditionally"); } @@ -7456,7 +7502,7 @@ mod postgres_tests { global.clone(), in_scope.clone(), ] { - ingest_event_inner(&state, &tracer, &tenant, event, author_auth()) + ingest_event_inner(&state, &tracer, &tenant, event, author_auth(), None) .await .expect("store message"); } @@ -7473,6 +7519,7 @@ mod postgres_tests { &tenant, delete(&deleted.id.to_hex(), &author), author_auth(), + None, ) .await .expect("soft-delete message"); @@ -7493,8 +7540,15 @@ mod postgres_tests { global.id.to_hex(), ] { let auth = scoped_auth(keys, channels[1]); - match ingest_event_inner(&state, &tracer, &tenant, delete(&target, keys), auth) - .await + match ingest_event_inner( + &state, + &tracer, + &tenant, + delete(&target, keys), + auth, + None, + ) + .await { Err(IngestError::Rejected(msg)) => assert_eq!( msg, "invalid: deletion target not found or not deletable by you", @@ -7524,6 +7578,7 @@ mod postgres_tests { &tenant, redact(&target), scoped_auth(&author, channels[1]), + None, ) .await { @@ -7542,14 +7597,22 @@ mod postgres_tests { &tenant, delete(&in_scope.id.to_hex(), &author), scoped_auth(&author, channels[0]), + None, ) .await .expect("in-scope token deletes the author's message"); // Re-deleting a soft-deleted message stores the deletion in its channel. let redeletion = delete(&deleted.id.to_hex(), &author); - ingest_event_inner(&state, &tracer, &tenant, redeletion.clone(), author_auth()) - .await - .expect("author re-deletes a soft-deleted message"); + ingest_event_inner( + &state, + &tracer, + &tenant, + redeletion.clone(), + author_auth(), + None, + ) + .await + .expect("author re-deletes a soft-deleted message"); let stored = state .db .get_event_by_id(community, redeletion.id.as_bytes()) @@ -7719,7 +7782,7 @@ mod postgres_tests { scopes: vec![Scope::MessagesWrite], auth_method: HttpAuthMethod::Nip98, }; - ingest_event_inner(&state, &tracer, &tenant, event, auth) + ingest_event_inner(&state, &tracer, &tenant, event, auth, None) .await .expect("store target"); } @@ -7757,7 +7820,7 @@ mod postgres_tests { event: Event, auth: IngestAuth, ) -> Result { - ingest_event_inner(&self.state, &self.tracer, &self.tenant, event, auth).await + ingest_event_inner(&self.state, &self.tracer, &self.tenant, event, auth, None).await } } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 532207a3dbc..6bbe6beea8f 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -750,6 +750,9 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { }, ); let state = Arc::new(app_state); + if state.config.scheduled_messages_enabled { + tokio::spawn(buzz_relay::api::scheduled_messages::run(Arc::clone(&state))); + } // Concurrency caps follow their flags from the first request: apply them // once before serving, then keep re-reading them. let concurrency_gates = buzz_relay::concurrency_limits::relay_gates(&state); diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index 6c1eee3aba1..f323f34e2b7 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -194,6 +194,8 @@ fn relay_limitation(max_message_length: usize, advertise_fi: bool) -> RelayLimit /// clippy threshold. #[derive(Default, Clone, Copy)] pub(crate) struct RelayCapabilityFlags { + /// Reservation-aware scheduled delivery was explicitly enabled. + pub scheduled_messages: bool, /// Whether NIP-43 (relay membership) is advertised in `supported_nips`. pub advertise_nip43: bool, /// Whether NIP-FI (federated identity) is advertised. @@ -240,6 +242,7 @@ impl RelayInfo { let RelayCapabilityFlags { advertise_nip43, advertise_fi, + scheduled_messages, } = flags; debug_assert!( !advertise_nip43 || relay_self.is_some(), @@ -263,6 +266,9 @@ impl RelayInfo { // checks this before opening a channel as unlisted. "buzz-unlisted-channels".to_string(), ]; + if scheduled_messages { + supported_extensions.push("buzz-scheduled-messages-v1".to_string()); + } let gif = gif_provider.map(|provider| { supported_extensions.push("buzz-gif".to_string()); GifDescriptor { @@ -465,6 +471,7 @@ pub(crate) async fn nip11_document( RelayCapabilityFlags { advertise_nip43, advertise_fi, + scheduled_messages: state.config.scheduled_messages_enabled, }, state.config.max_frame_bytes, state.config.pairing_relay_url.as_deref(), @@ -938,6 +945,7 @@ mod tests { RelayCapabilityFlags { advertise_nip43: true, advertise_fi: false, + scheduled_messages: false, }, DEFAULT_MAX_FRAME_BYTES, None, @@ -960,6 +968,7 @@ mod tests { RelayCapabilityFlags { advertise_nip43: true, advertise_fi: false, + scheduled_messages: false, }, DEFAULT_MAX_FRAME_BYTES, None, diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index f271b4e58d3..acb6ca0be80 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -467,6 +467,10 @@ pub fn build_router(state: Arc) -> Router { .route("/events", post(api::bridge::submit_event)) .route("/query", post(api::bridge::query_events)) .route("/count", post(api::bridge::count_events)) + .route( + api::scheduled_messages::PATH, + post(api::scheduled_messages::handle), + ) // Relay-owned third-party GIF metadata proxy (NIP-98 auth). .route(api::gifs::SEARCH_PATH, post(api::gifs::search)) .route(api::gifs::SHARE_PATH, post(api::gifs::share)) diff --git a/docs/database.md b/docs/database.md index b96c0c6876a..7fdaa930270 100644 --- a/docs/database.md +++ b/docs/database.md @@ -485,3 +485,11 @@ description: - [crates/buzz-db/TESTING.md](../crates/buzz-db/TESTING.md): desired-state and migration-mode tests. - `schema/schema.sql` header: the desired-state layout. + +## Scheduled-delivery timestamp reservations + +Scheduled messages reserve their future history position until event storage and +sent state commit atomically. Replica-fence probes cap completeness proofs below +all pending reservations. See [scheduled messages](scheduled-messages.md#replica-completeness) +for the admission bounds, proof and mandatory two-phase rollout. Do not enable +the feature while a reservation-unaware probe can still publish heartbeat tokens. diff --git a/docs/scheduled-messages.md b/docs/scheduled-messages.md new file mode 100644 index 00000000000..44bc01a80bd --- /dev/null +++ b/docs/scheduled-messages.md @@ -0,0 +1,121 @@ +# Scheduled messages + +The relay owns a durable, private queue of exact user-signed kind:9 messages. +Delivery runs without an open client. The desktop/web composer prepares ordinary +messages, mentions, attachments and thread tags before asking the host signer to +sign with the selected future timestamp. No account key is stored in the queue. +The scheduled timestamp is the event's timestamp; after an outage a late delivery +appears at its originally selected time in event history. + +## Enable and roll back + +This is a two-phase rollout. The feature is off unless +`BUZZ_SCHEDULED_MESSAGES=on` (exact value). When off, the endpoint returns 404, +the worker does not run and NIP-11 omits `buzz-scheduled-messages-v1`. + +1. Apply migration 0087, or the matching desired-state table and indexes. It only + creates a table and indexes; it has no backfill, trigger, function or FK. + Older relays remain compatible with this expanded schema. +2. Replace **every relay process sharing this writer**, with the feature still + off, so every replica-fence probe understands pending queue reservations. + Confirm that no older probe process remains, including workers and canaries. +3. Enable the flag, then deploy clients that recognize the advertised extension. + Verify delivery with clients closed, queue access isolation and replica reads. + +To pause delivery, turn the flag off on every worker. Existing queue rows remain; +when re-enabled they are rechecked and delivered late. Turning it off alone does +not cancel work already in flight. Never downgrade to reservation-unaware probes +while pending rows exist. Keep the reservation-aware binary deployed, or hold +replica routing closed and explicitly resolve all queued work before downgrading. + +## Protocol + +`POST /scheduled-messages` uses the ordinary member HTTP admission path: NIP-98 +with an exact URL/method/body hash, replay protection, community resolution, +NIP-FI policy, rate limits and current relay membership. Owners cannot specify a +different community or queue owner. The body is at most 40 KiB; signed events at +most 32 KiB and content at most 24 KiB. Only kind:9, one UUID `h` destination and +future timestamps within 120 days are accepted. The sender must currently hold a +non-guest conversation membership. Dispatch rechecks normal ingest permissions. + +| Action | Body fields besides `action` | +| --- | --- | +| `list` | None | +| `create` | `id` (UUID idempotency key), `event` | +| `update` | `id`, `revision` (integer), `event` | +| `cancel` | `id`, `revision` | +| `send_now` | `id`, `revision`, freshly signed `event` | + +Mutations return `{"accepted":true}` after commit. Listing returns `messages` +with `id`, `revision`, `event`, `send_at`, `status`, `error`, plus +`sent_history_limit:100`. Active work is capped at 256 per owner/community; the +list includes all scheduled/failed rows and the last 100 sent rows. Cancelled +rows remain tombstones and are omitted. Queue state follows community deletion +inventory and purge. Sent rows and tombstones are retained until community purge; +the list limit is not a retention promise. + +Identical operation retries succeed after an acknowledgement is lost. Conflicting +revisions or rebinding an event already attached to another queue ID return 409. +Send-now preserves the reviewed content and tags and replaces only its timestamp +and signature. Refresh the list before changing another revision. A response loss +is uncertain acceptance: retry the same create ID/event, never invent a second +message to recover it. + +This HTTP resource holds relay execution state rather than a new public Nostr +event kind. The only delivered artifact is the exact ordinary signed event. Queue +contents are never broadcast or exposed in public event queries. + +## Atomic delivery and recovery + +Each one-second bounded sweep scans at most 100 due candidates, with randomized +startup phase. A worker admits one community transaction and claims one row with +`FOR UPDATE SKIP LOCKED`. Edits, cancellation and sending serialize on that row. +The shared ingest path performs ordinary signature, channel, ban, threading, +media and authorization checks. Event insertion, mentions, push work, thread +metadata and sent status commit on the held claim transaction. A lost claim +connection cannot later publish. Stable event IDs deduplicate crash recovery; +an already stored event, including a deleted event, is reconciled as sent. + +Transient storage/access failures retain work with exponential backoff. If a +failed insert has rolled back its claim, recording backoff reacquires only the +same pending revision; cancellation, edits and successful delivery are preserved. +Permanent refusals stay visible as failed; users may edit/reschedule or cancel. +Regular ingest side effects and fan-out remain shared. Like ordinary ingest, +external side effects after the database commit have their existing recovery +semantics; the queue guarantees one durable event, not exactly-once execution of +every external consumer. + +## Replica completeness + +Ordinary channel writes enforce a commit-time timestamp floor of 960 seconds. +A previously queued event may legitimately be older after downtime. Only a held +scheduled claim may widen its transaction-local floor, for its exact stored event. + +Every reservation-aware writer probe caps its ordinary proof wall at +`min(send_at)-301` across pending rows, before committing the heartbeat token. +The event timestamp is at least `send_at-300`, so it is strictly above every +proof that did not yet observe its commit. The reservation disappears in the +same commit that stores the event; a later heartbeat therefore orders after the +event's commit. Cancellation removes the reservation without publishing a row. + +Admission uses the **database clock after lock waits**: a new/revised reservation +must have `send_at >= DBnow-300`, and `created_at >= send_at-300`. Thus its earliest +possible timestamp, `DBnow-600`, stays newer than an already proved ordinary wall +(`S-965`). Transactions open during a probe are additionally bounded by its +oldest-open-transaction term. This preserves previously retained proofs even +when a reservation is created after their heartbeat. Host clock skew cannot +introduce a retroactive reservation. An overdue pending item may conservatively +hold the replica wall back until sent/cancelled; reads then use the writer. + +The proof depends on every probe using reservations, which is why enabling during +a mixed-version rollout and downgrading with pending work are unsupported. + +## Local verification + +Use an isolated PostgreSQL/Redis relay with the feature enabled. From the modern +app checkout, run `BUZZ_SCHEDULED_TEST_RELAY=http://127.0.0.1:3000 pnpm exec vitest +run src/features/relay/scheduled-messages.live.test.ts`. The test refuses remote +hosts, uses fictional random keys, and exercises authenticated queue operations +through the real app service. The relay's database regressions additionally cover +overdue insertion, replica reservation release, lost claims, stale revisions, +worker exclusion, retry backoff and community deletion. diff --git a/migrations/0087_scheduled_messages.sql b/migrations/0087_scheduled_messages.sql new file mode 100644 index 00000000000..1b4e5ea8814 --- /dev/null +++ b/migrations/0087_scheduled_messages.sql @@ -0,0 +1,23 @@ +SET LOCAL lock_timeout = '5s'; +SET LOCAL statement_timeout = '30s'; +CREATE TABLE scheduled_messages ( + community_id UUID NOT NULL, + owner BYTEA NOT NULL, + id UUID NOT NULL, + revision BIGINT NOT NULL DEFAULT 1, + event JSONB NOT NULL, + event_id BYTEA NOT NULL, + send_at BIGINT NOT NULL, + next_attempt_at BIGINT NOT NULL, + status TEXT NOT NULL DEFAULT 'scheduled', + attempts INTEGER NOT NULL DEFAULT 0, + error TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (community_id, owner, id), + UNIQUE (community_id, owner, event_id) +); +CREATE INDEX scheduled_messages_due ON scheduled_messages (next_attempt_at) + WHERE status = 'scheduled'; +CREATE INDEX scheduled_messages_owner ON scheduled_messages (community_id, owner, status, send_at); + +CREATE INDEX scheduled_messages_reservation ON scheduled_messages (send_at) WHERE status='scheduled'; diff --git a/schema/schema.sql b/schema/schema.sql index f804fb0bc4a..49417dc4539 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -223,3 +223,4 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i tables/public/media_uploaders.sql \i tables/public/media_publications.sql +\i tables/public/scheduled_messages.sql diff --git a/schema/tables/public/scheduled_messages.sql b/schema/tables/public/scheduled_messages.sql new file mode 100644 index 00000000000..e3307da9180 --- /dev/null +++ b/schema/tables/public/scheduled_messages.sql @@ -0,0 +1,21 @@ +CREATE TABLE scheduled_messages ( + community_id UUID NOT NULL, + owner BYTEA NOT NULL, + id UUID NOT NULL, + revision BIGINT NOT NULL DEFAULT 1, + event JSONB NOT NULL, + event_id BYTEA NOT NULL, + send_at BIGINT NOT NULL, + next_attempt_at BIGINT NOT NULL, + status TEXT NOT NULL DEFAULT 'scheduled', + attempts INTEGER NOT NULL DEFAULT 0, + error TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (community_id, owner, id), + UNIQUE (community_id, owner, event_id) +); +CREATE INDEX scheduled_messages_due ON scheduled_messages (next_attempt_at) + WHERE status = 'scheduled'; +CREATE INDEX scheduled_messages_owner ON scheduled_messages (community_id, owner, status, send_at); + +CREATE INDEX scheduled_messages_reservation ON scheduled_messages (send_at) WHERE status='scheduled';