diff --git a/crates/ant-control/src/command.rs b/crates/ant-control/src/command.rs index ec08c29..9aa50f9 100644 --- a/crates/ant-control/src/command.rs +++ b/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, }, /// Walk the manifest at `reference`, resolve `path`, then join the diff --git a/crates/ant-gateway/src/subscribe.rs b/crates/ant-gateway/src/subscribe.rs index 0b49208..307cc70 100644 --- a/crates/ant-gateway/src/subscribe.rs +++ b/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, + /// **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::(SUB_CHANNEL_CAP); let cmd = ControlCommand::LurkerSubscribe { target, gsoc_addresses, pss_topics, + history, ack: ack_tx, }; if handle.commands.send(cmd).await.is_err() { diff --git a/crates/ant-gateway/tests/subscribe_ws.rs b/crates/ant-gateway/tests/subscribe_ws.rs index 6fa13fe..0994268 100644 --- a/crates/ant-gateway/tests/subscribe_ws.rs +++ b/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; diff --git a/crates/ant-p2p/src/behaviour.rs b/crates/ant-p2p/src/behaviour.rs index 047c62f..bd65f6f 100644 --- a/crates/ant-p2p/src/behaviour.rs +++ b/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 diff --git a/crates/ant-p2p/src/lurker.rs b/crates/ant-p2p/src/lurker.rs index 923835b..e9ae220 100644 --- a/crates/ant-p2p/src/lurker.rs +++ b/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 { 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, 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); diff --git a/crates/ant-p2p/src/messaging.rs b/crates/ant-p2p/src/messaging.rs index efc53ba..fdfedbb 100644 --- a/crates/ant-p2p/src/messaging.rs +++ b/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());