feat(relay): add durable scheduled delivery

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

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;
 

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",

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<DateTime<Utc>>,
     /// `S`: writer `clock_timestamp()` captured first.
     sampled_at: DateTime<Utc>,
     /// Oldest open transaction among other client backends at scan time,
@@ -866,6 +868,12 @@ async fn sample_writer(
     })
     .await?;
 
+    let scheduled_wall: Option<DateTime<Utc>> = 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<TokenEntry, ProbeError> {
     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,

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<Uuid> = 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));

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.