feat(lurker): PSS mailbox mode — offline delivery via bounded backlog sweep

crates/ant-control/src/command.rs

@@ -117,6 +117,11 @@ pub enum ControlCommand {
         /// PSS topics to watch (`keccak256(topic_string)`); reception uses
         /// the node's PSS key if present, else topic-broadcast.
         pss_topics: Vec<[u8; 32]>,
+        /// **Mailbox mode**: sweep the trojan-bin backlog on subscribe so
+        /// PSS messages sent while the receiver was offline are recovered,
+        /// rather than only tailing live traffic. Backs the WebSocket
+        /// `?history=true` option. See `ant_p2p`'s `WatchState::history`.
+        history: bool,
         ack: mpsc::Sender<ControlAck>,
     },
     /// Walk the manifest at `reference`, resolve `path`, then join the

crates/ant-gateway/src/subscribe.rs

@@ -51,6 +51,11 @@ pub struct PssSubscribeQuery {
     /// messages there. Absent ⇒ the node's own neighborhood (directed PSS
     /// to this node).
     neighborhood: Option<String>,
+    /// **Mailbox mode** (`?history=true`): on subscribe, sweep the
+    /// trojan-bin backlog so messages sent while this client was offline
+    /// are delivered before live traffic — not just tail from now.
+    #[serde(default)]
+    history: bool,
 }
 
 /// `GET /gsoc/subscribe/{address}` upgrade handler.
@@ -64,9 +69,11 @@ pub async fn gsoc_subscribe(
         return params_error(ParamKind::Path, reasons);
     };
     // Watch exactly this SOC address; reside in its neighborhood.
+    // GSOC has no mailbox mode (a SOC has a latest value, not a message
+    // backlog) — always live.
     let cmd_target = address;
     ws.on_upgrade(move |socket| {
-        run_subscription(handle, socket, cmd_target, vec![address], Vec::new())
+        run_subscription(handle, socket, cmd_target, vec![address], Vec::new(), false)
     })
 }
 
@@ -90,7 +97,10 @@ pub async fn pss_subscribe(
         }
         None => [0u8; 32],
     };
-    ws.on_upgrade(move |socket| run_subscription(handle, socket, target, Vec::new(), vec![topic]))
+    let history = query.history;
+    ws.on_upgrade(move |socket| {
+        run_subscription(handle, socket, target, Vec::new(), vec![topic], history)
+    })
 }
 
 /// Drive one subscription: open the lurker on the node, forward each
@@ -102,12 +112,14 @@ async fn run_subscription(
     target: [u8; 32],
     gsoc_addresses: Vec<[u8; 32]>,
     pss_topics: Vec<[u8; 32]>,
+    history: bool,
 ) {
     let (ack_tx, mut ack_rx) = mpsc::channel::<ControlAck>(SUB_CHANNEL_CAP);
     let cmd = ControlCommand::LurkerSubscribe {
         target,
         gsoc_addresses,
         pss_topics,
+        history,
         ack: ack_tx,
     };
     if handle.commands.send(cmd).await.is_err() {

crates/ant-gateway/tests/subscribe_ws.rs

@@ -67,6 +67,21 @@ async fn pss_subscribe_delivers_the_topic_message() {
     );
 }
 
+/// Mailbox mode: `?history=true` parses and upgrades cleanly (guards the
+/// query wiring end to end; the sweep behaviour itself is unit-tested in
+/// `ant_p2p::lurker::start_bin_id`).
+#[tokio::test]
+async fn pss_subscribe_accepts_history_query() {
+    let base = serve().await;
+    let url = format!("{base}/pss/subscribe/test?history=true");
+    let (mut ws, _resp) = tokio_tungstenite::connect_async(&url).await.expect("ws");
+    let frame = ws.next().await.expect("frame").expect("ok");
+    assert_eq!(
+        frame,
+        Message::Binary(b"fixture-lurker-payload".to_vec().into())
+    );
+}
+
 #[tokio::test]
 async fn capped_subscription_is_rejected_with_a_close_reason() {
     let base = serve().await;

crates/ant-p2p/src/behaviour.rs

@@ -2502,6 +2502,7 @@ fn handle_control_command(
             target,
             gsoc_addresses,
             pss_topics,
+            history,
             ack,
         } => {
             use crate::lurker::{self, LurkerConfig};
@@ -2514,6 +2515,7 @@ fn handle_control_command(
                 // only (the topic-derived key handles it). Directed PSS to
                 // the node's key lands when a pss.key is persisted.
                 pss_secret: None,
+                history,
             };
             if watch.is_empty() {
                 // Nothing to watch — tell the subscriber why before the

crates/ant-p2p/src/lurker.rs

@@ -69,6 +69,30 @@ const HANDOVER_MAX_OVERLAP: Duration = Duration::from_mins(1);
 /// peers, so delivery is at-least-once, not N-times. Replaced pullers
 /// don't use this: they resume exactly where their predecessor stopped.
 const PULL_BACKLOG: u64 = 8;
+/// Earliest binID (reserves are 1-indexed; binID 0 is never used).
+const HISTORY_FLOOR: u64 = 1;
+/// Mailbox lookback, in binIDs, per (peer, bin). Mailbox mode starts a
+/// fresh puller this far behind the peer's cursor instead of at
+/// [`PULL_BACKLOG`], recovering messages sent while offline.
+///
+/// It is a **bounded** window on purpose. binIDs count chunks that
+/// landed in one bin on one peer, and a light node pulls a shallow
+/// covering peer's bin (`b_p ≈ 9-14`), which is busy — so an unbounded
+/// `start = 1` sweep would drag the whole history of a hot bin. This
+/// window caps the sweep at a few thousand recent chunks per bin, which
+/// on a busy bin is a recent-history mailbox (minutes-to-hours,
+/// depending on the bin's fill rate) rather than the complete backlog.
+///
+/// A *complete* backlog sweep would need the trojan concentrated into a
+/// sparse deep bin (a deeper mining prefix pulled by a deeply-resident
+/// receiver) — see [`PSS_MINED_PREFIX_BITS`] for why that trade doesn't
+/// pay at light-node residency. So today the mailbox is "recent", and a
+/// larger [`HISTORY_BACKLOG`] simply extends how far back it reaches at
+/// linear cost.
+///
+/// A sweep may exceed [`SEEN_CAP`] and re-deliver its oldest chunks;
+/// that is within the documented at-least-once/may-duplicate contract.
+const HISTORY_BACKLOG: u64 = 4096;
 /// Number of closest connected peers to pull from concurrently. A
 /// freshly-pushed chunk lands on the storer(s) nearest its address and
 /// replicates outward; pulling several covering peers catches it
@@ -432,9 +456,13 @@ pub async fn run(
         // may have added or removed GSOC addresses / PSS topics since
         // the last one, and the desired-set handover below then grows
         // or retires pullers to match.
-        let (want_gsoc, want_pss) = {
+        let (want_gsoc, want_pss, history) = {
             let w = watch.read().unwrap_or_else(PoisonError::into_inner);
-            (!w.gsoc_addresses.is_empty(), !w.pss_topics.is_empty())
+            (
+                !w.gsoc_addresses.is_empty(),
+                !w.pss_topics.is_empty(),
+                w.history,
+            )
         };
 
         // Which covering (peer, bin)s do we want, and which peers still
@@ -487,13 +515,16 @@ pub async fn run(
                     .copied()
                     .filter(|(epoch, _)| *epoch == cursors.epoch)
                     .map(|(_, start)| start);
-                let start = resume.unwrap_or_else(|| {
-                    cursor.saturating_add(1).saturating_sub(PULL_BACKLOG).max(1)
-                });
+                // A replaced puller resumes exactly where it stopped.
+                // Otherwise: mailbox mode sweeps the whole bin backlog
+                // (offline delivery); the default just tails live with a
+                // small backlog to cover the reside/cursor-read window.
+                let start = start_bin_id(resume, cursor, history);
                 tracing::info!(
                     target: "ant_p2p::lurker",
                     peer = %peer_id, bin, start, epoch = cursors.epoch,
-                    resumed = resume.is_some(), "lurker pulling neighborhood bin",
+                    resumed = resume.is_some(), history,
+                    "lurker pulling neighborhood bin",
                 );
                 next_generation += 1;
                 let handle = tokio::spawn(pull_bin(
@@ -741,6 +772,32 @@ fn covering_bins(b_p: u8, want_gsoc: bool, want_pss: bool) -> Vec<u8> {
     bins
 }
 
+/// The binID a fresh or resumed puller starts at.
+///
+/// - **Resume** (epoch-matched replacement puller): exactly where the
+///   predecessor stopped — never re-pull, never gap.
+/// - **Mailbox** (`history`, fresh puller): [`HISTORY_BACKLOG`] behind
+///   the cursor — sweep the recent bin backlog so messages sent while
+///   offline are recovered. The seen-set dedups the sweep against live
+///   traffic.
+/// - **Default** (fresh puller): a short [`PULL_BACKLOG`] behind the
+///   cursor — covers the window between a storer accepting a chunk and
+///   us reading its cursor, without pulling history.
+fn start_bin_id(resume: Option<u64>, cursor: u64, history: bool) -> u64 {
+    if let Some(start) = resume {
+        return start;
+    }
+    let backlog = if history {
+        HISTORY_BACKLOG
+    } else {
+        PULL_BACKLOG
+    };
+    cursor
+        .saturating_add(1)
+        .saturating_sub(backlog)
+        .max(HISTORY_FLOOR)
+}
+
 /// The `n` connected peers whose overlays are closest to `target`,
 /// deepest first. Marks the snapshot seen (`borrow_and_update`) so the
 /// driver's `changed()` wait really waits for the *next* change.
@@ -914,6 +971,34 @@ mod tests {
         assert!(covering_bins(12, false, false).is_empty());
     }
 
+    /// Mailbox mode: a FRESH puller sweeps a bounded backlog behind the
+    /// cursor (`HISTORY_BACKLOG`), recovering offline messages; the
+    /// default only tails a short `PULL_BACKLOG`.
+    #[test]
+    fn start_bin_id_mailbox_sweeps_the_backlog_window() {
+        // Deep cursor: history reaches HISTORY_BACKLOG back, default only PULL_BACKLOG.
+        assert_eq!(
+            start_bin_id(None, 50_000, true),
+            50_000 + 1 - HISTORY_BACKLOG
+        );
+        assert_eq!(start_bin_id(None, 50_000, false), 50_000 + 1 - PULL_BACKLOG);
+        // Sparse bin (fewer chunks than the window): backlog underflows
+        // to the floor → the mailbox recovers the ENTIRE bin history.
+        assert_eq!(start_bin_id(None, 500, true), HISTORY_FLOOR);
+        // A near-empty bin can't go below the floor either way.
+        assert_eq!(start_bin_id(None, 2, false), HISTORY_FLOOR);
+    }
+
+    /// A resumed (epoch-matched replacement) puller ALWAYS continues
+    /// exactly where its predecessor stopped — mailbox mode must not
+    /// rewind it to the floor and re-pull the whole backlog every
+    /// handover.
+    #[test]
+    fn start_bin_id_resume_overrides_mailbox() {
+        assert_eq!(start_bin_id(Some(12_345), 50_000, true), 12_345);
+        assert_eq!(start_bin_id(Some(12_345), 50_000, false), 12_345);
+    }
+
     #[test]
     fn closest_n_orders_deepest_first_and_marks_seen() {
         let target = overlay(0xff);

crates/ant-p2p/src/messaging.rs

@@ -37,6 +37,17 @@ pub struct WatchState {
     /// node's key. `None` still receives **topic-broadcast** PSS (messages
     /// with no explicit recipient, decryptable via the topic-derived key).
     pub pss_secret: Option<[u8; 32]>,
+    /// **Mailbox mode**: sweep a bounded trojan-bin backlog on subscribe
+    /// rather than only tailing live traffic. When set, fresh PSS pullers
+    /// start a bounded window behind the cursor (see
+    /// `lurker::HISTORY_BACKLOG`) instead of just behind it, so messages
+    /// sent while the receiver was offline are recovered — a recent-history
+    /// window (its reach depends on how busy the swept bin is). Union-OR
+    /// across subscribers: if any asks for history, the shared lurker
+    /// sweeps. Intended for discrete PSS messages; a GSOC watcher wants
+    /// the latest SOC value, not every historical version, so GSOC-only
+    /// subscribers leave this unset.
+    pub history: bool,
 }
 
 impl WatchState {
@@ -62,6 +73,9 @@ impl WatchState {
         if self.pss_secret.is_none() {
             self.pss_secret = other.pss_secret;
         }
+        // Any subscriber asking for history makes the shared lurker
+        // sweep the backlog.
+        self.history |= other.history;
     }
 }
 
@@ -231,6 +245,39 @@ mod tests {
         assert!(classify(&address, &data, &other).is_none());
     }
 
+    #[test]
+    fn merge_from_unions_topics_secret_and_history() {
+        let mut base = WatchState {
+            pss_topics: vec![[1u8; 32]],
+            ..Default::default()
+        };
+        // A history-wanting subscriber attaches: the shared lurker must
+        // now sweep (history OR), and the union grows.
+        base.merge_from(&WatchState {
+            pss_topics: vec![[2u8; 32]],
+            pss_secret: Some([9u8; 32]),
+            history: true,
+            ..Default::default()
+        });
+        assert!(
+            base.history,
+            "any history subscriber makes the lurker sweep"
+        );
+        assert_eq!(base.pss_topics, vec![[1u8; 32], [2u8; 32]]);
+        assert_eq!(base.pss_secret, Some([9u8; 32]));
+
+        // A later non-history subscriber must NOT turn the sweep back off.
+        base.merge_from(&WatchState {
+            gsoc_addresses: HashSet::from([[3u8; 32]]),
+            ..Default::default()
+        });
+        assert!(
+            base.history,
+            "history stays set once any subscriber wanted it"
+        );
+        assert!(base.gsoc_addresses.contains(&[3u8; 32]));
+    }
+
     #[test]
     fn empty_watch_short_circuits() {
         assert!(WatchState::default().is_empty());