fix: Reticulum LXMF catch-up loop and private-IP Bonjour hints

docs/reticulum-sidecar-ipc.md

@@ -87,14 +87,14 @@ Routing bias between **RF** (LoRa / RNode) and **network** (TCP/UDP/I2P/gateway/
 
 ### LXMF and contacts
 
-| Method | Path                           | Body / notes                                                                    | Response                                                                                                                                                                                                                                                                                                                                                                       |
-| ------ | ------------------------------ | ------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
-| POST   | `/api/v1/lxmf/send`            | `{ destination_hash, text, reply_to_hash?, reply_to_id?, reply_preview_text? }` | Live: stamps LXMF `FIELD_REPLY_TO` (0x30) / optional `FIELD_REPLY_QUOTE` (0x31) before sign; `{ ok, delivery_method?, delivery_status?, sent_via?, message? }` or `{ ok: false, error: "no_propagation_node" }`. **`delivery_status` on this response is initial enqueue state only** (`queued` or `sending`) — not delivery confirmation. Stub: `{ ok, sent_via?, message? }` |
-| POST   | `/api/v1/lxmf/reaction`        | `{ destination_hash, target_hash, emoji }`                                      | `{ ok, message? }`                                                                                                                                                                                                                                                                                                                                                             |
-| GET    | `/api/v1/lxmf/recent`          | `?since_ts=` (ms, optional), `?limit=` (default 200, max 500)                   | `{ messages: [], ring_len }` — ring buffer of recent **inbound** LXMF payloads for WS lag/reconnect catch-up (not durable across sidecar restart; capped at 200); `ring_len` is current buffer occupancy                                                                                                                                                                       |
-| DELETE | `/api/v1/lxmf/messages/{hash}` |                                                                                 | `{ ok }`                                                                                                                                                                                                                                                                                                                                                                       |
-| GET    | `/api/v1/contacts`             |                                                                                 | `{ contacts: [] }` — overlays announce/peer/Nomad labels onto nameless or hash-prefix contact `display_name` values (does not overwrite a real name) and may persist fills                                                                                                                                                                                                     |
-| DELETE | `/api/v1/contacts`             |                                                                                 | `{ ok, cleared }` — clears LXMF contacts after demoting them into the peer cache (keeps Peers; does not delete chat messages)                                                                                                                                                                                                                                                  |
+| Method | Path                           | Body / notes                                                                                               | Response                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
+| ------ | ------------------------------ | ---------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| POST   | `/api/v1/lxmf/send`            | `{ destination_hash, text, reply_to_hash?, reply_to_id?, reply_preview_text? }`                            | Live: stamps LXMF `FIELD_REPLY_TO` (0x30) / optional `FIELD_REPLY_QUOTE` (0x31) before sign; `{ ok, delivery_method?, delivery_status?, sent_via?, message? }` or `{ ok: false, error: "no_propagation_node" }`. **`delivery_status` on this response is initial enqueue state only** (`queued` or `sending`) — not delivery confirmation. Stub: `{ ok, sent_via?, message? }`                                                                                                                                                                                                              |
+| POST   | `/api/v1/lxmf/reaction`        | `{ destination_hash, target_hash, emoji }`                                                                 | `{ ok, message? }`                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
+| GET    | `/api/v1/lxmf/recent`          | `?since_ts=` (ms, optional), `?since_seq=` (opaque `ring_seq`, optional), `?limit=` (default 200, max 500) | `{ messages: [], ring_len }` — ring buffer of recent **inbound** LXMF payloads for WS lag/reconnect catch-up (not durable across sidecar restart; capped at 200). Rows are chronological (oldest→newest) and each accepted row is stamped with monotonic `ring_seq`. Cursor: `since_ts` alone keeps `timestamp > since_ts`; with `since_seq`, keep rows after the complete `(since_ts, since_seq)` cursor (`timestamp > since_ts` **or** same-ms with `ring_seq > since_seq`) so same-ms twins remain recoverable without reprocessing the boundary; `ring_len` is current buffer occupancy |
+| DELETE | `/api/v1/lxmf/messages/{hash}` |                                                                                                            | `{ ok }`                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
+| GET    | `/api/v1/contacts`             |                                                                                                            | `{ contacts: [] }` — overlays announce/peer/Nomad labels onto nameless or hash-prefix contact `display_name` values (does not overwrite a real name) and may persist fills                                                                                                                                                                                                                                                                                                                                                                                                                  |
+| DELETE | `/api/v1/contacts`             |                                                                                                            | `{ ok, cleared }` — clears LXMF contacts after demoting them into the peer cache (keeps Peers; does not delete chat messages)                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
 
 ### Peers, topology, and propagation
 

reticulum-sidecar/src/api/lxmf.rs

@@ -128,9 +128,15 @@ pub async fn lxmf_delete_message(
 
 #[derive(Debug, Deserialize)]
 pub struct RecentLxmfQuery {
-    /// Inclusive lower bound on payload `timestamp` (ms). Omit to return the full ring.
+    /// Exclusive lower-bound cursor on payload `timestamp` (ms). Omit with `since_seq` to return
+    /// the full ring. Rows are chronological (oldest→newest). With `since_seq`, keep rows after
+    /// the complete `(since_ts, since_seq)` cursor so same-ms twins remain recoverable.
     #[serde(default)]
     pub since_ts: Option<i64>,
+    /// Opaque monotonic `ring_seq` stamped by the inbound ring. Pair with `since_ts`; ignored when
+    /// `since_ts` is omitted. Without `since_seq`, filtering is timestamp-only exclusive.
+    #[serde(default)]
+    pub since_seq: Option<u64>,
     /// Max rows (default 200, capped at 500).
     #[serde(default)]
     pub limit: Option<usize>,
@@ -142,7 +148,7 @@ pub async fn list_recent_lxmf(
     Query(q): Query<RecentLxmfQuery>,
 ) -> Json<serde_json::Value> {
     let limit = q.limit.unwrap_or(200).clamp(1, 500);
-    let messages = stack.list_recent_inbound_lxmf(q.since_ts, limit);
+    let messages = stack.list_recent_inbound_lxmf(q.since_ts, q.since_seq, limit);
     let ring_len = stack.inbound_lxmf_ring_len();
     Json(serde_json::json!({ "messages": messages, "ring_len": ring_len }))
 }

reticulum-sidecar/src/stack/live.rs

@@ -346,7 +346,9 @@ impl LiveBridge {
                 Some(&inbound_sender_name),
             );
             // Buffer before WS emit so lag/reconnect can catch up via GET /api/v1/lxmf/recent.
-            inbound_lxmf_cb.push(payload.clone());
+            // Stamp `ring_seq` on accepted rows so live watermark and catch-up share a cursor.
+            // Deduped / lock-failed pushes return None — still emit the original payload.
+            let payload = inbound_lxmf_cb.push(payload.clone()).unwrap_or(payload);
             let message_hash = payload
                 .get("message_hash")
                 .and_then(|v| v.as_str())

reticulum-sidecar/src/stack/lxmf_inbound_log.rs

@@ -9,21 +9,26 @@ pub const MAX_LXMF_INBOUND_LOG: usize = 200;
 #[derive(Debug)]
 pub struct LxmfInboundBuffer {
     max: usize,
+    /// Monotonic opaque sequence stamped onto accepted rows as `ring_seq`.
+    next_seq: Mutex<u64>,
     inner: Mutex<VecDeque<serde_json::Value>>,
 }
 
 impl LxmfInboundBuffer {
     pub fn new(max: usize) -> Self {
         Self {
             max: max.max(1),
+            next_seq: Mutex::new(1),
             inner: Mutex::new(VecDeque::new()),
         }
     }
 
     /// Push an inbound `lxmf_message` payload. Dedupes by `message_hash` when present.
-    pub fn push(&self, payload: serde_json::Value) {
+    /// Stamps a monotonic opaque `ring_seq` on accepted rows for catch-up cursors.
+    /// Returns the stamped payload when accepted, or `None` when deduped / lock failed.
+    pub fn push(&self, mut payload: serde_json::Value) -> Option<serde_json::Value> {
         let Ok(mut buf) = self.inner.lock() else {
-            return;
+            return None;
         };
         if let Some(hash) = payload
             .get("message_hash")
@@ -35,35 +40,52 @@ impl LxmfInboundBuffer {
                     .and_then(|v| v.as_str())
                     .is_some_and(|h| h.eq_ignore_ascii_case(hash))
             }) {
-                return;
+                return None;
             }
         }
+        let Ok(mut next) = self.next_seq.lock() else {
+            return None;
+        };
+        let seq = *next;
+        *next = next.saturating_add(1);
+        drop(next);
+        if let Some(obj) = payload.as_object_mut() {
+            obj.insert("ring_seq".into(), serde_json::json!(seq));
+        }
         if buf.len() >= self.max {
             buf.pop_front();
         }
-        buf.push_back(payload);
+        buf.push_back(payload.clone());
+        Some(payload)
     }
 
     pub fn len(&self) -> usize {
         self.inner.lock().map(|buf| buf.len()).unwrap_or(0)
     }
 
-    /// Snapshot newest-first filtered by optional `since_ts` (inclusive, ms), then reverse to
-    /// chronological order for ingest catch-up.
-    pub fn snapshot(&self, since_ts: Option<i64>, limit: usize) -> Vec<serde_json::Value> {
+    /// Snapshot in chronological push order (oldest→newest via `push_back` / `VecDeque::iter`),
+    /// filtered by an optional exclusive `(since_ts, since_seq)` cursor, then truncated to
+    /// the newest `limit` rows.
+    ///
+    /// Cursor semantics:
+    /// - `since_ts` alone: keep rows with `timestamp > since_ts` (legacy exclusive ms bound).
+    /// - `since_ts` + `since_seq`: keep rows after that complete cursor —
+    ///   `timestamp > since_ts` **or** (`timestamp == since_ts` **and** `ring_seq > since_seq`).
+    ///   Same-ms twins after the stamped sequence are therefore recoverable without
+    ///   re-returning already-processed boundary rows.
+    pub fn snapshot(
+        &self,
+        since_ts: Option<i64>,
+        since_seq: Option<u64>,
+        limit: usize,
+    ) -> Vec<serde_json::Value> {
         let limit = limit.max(1);
         let Ok(buf) = self.inner.lock() else {
             return Vec::new();
         };
         let mut out: Vec<serde_json::Value> = buf
             .iter()
-            .filter(|row| match since_ts {
-                None => true,
-                Some(min_ts) => row
-                    .get("timestamp")
-                    .and_then(serde_json::Value::as_i64)
-                    .is_some_and(|ts| ts >= min_ts),
-            })
+            .filter(|row| after_catch_up_cursor(row, since_ts, since_seq))
             .cloned()
             .collect();
         if out.len() > limit {
@@ -73,6 +95,41 @@ impl LxmfInboundBuffer {
     }
 }
 
+fn row_timestamp(row: &serde_json::Value) -> Option<i64> {
+    row.get("timestamp").and_then(serde_json::Value::as_i64)
+}
+
+fn row_ring_seq(row: &serde_json::Value) -> u64 {
+    row.get("ring_seq")
+        .and_then(serde_json::Value::as_u64)
+        .unwrap_or(0)
+}
+
+fn after_catch_up_cursor(
+    row: &serde_json::Value,
+    since_ts: Option<i64>,
+    since_seq: Option<u64>,
+) -> bool {
+    let Some(min_ts) = since_ts else {
+        return true;
+    };
+    let Some(ts) = row_timestamp(row) else {
+        return false;
+    };
+    if ts > min_ts {
+        return true;
+    }
+    if ts < min_ts {
+        return false;
+    }
+    // Same millisecond as the cursor: require a sequence past `since_seq`.
+    match since_seq {
+        Some(min_seq) => row_ring_seq(row) > min_seq,
+        // Timestamp-only exclusive bound (legacy): drop the entire ms bucket.
+        None => false,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -90,25 +147,78 @@ mod tests {
     #[test]
     fn ring_evicts_oldest_and_dedupes_hash() {
         let buf = LxmfInboundBuffer::new(2);
-        buf.push(msg("h1", 1, "a"));
-        buf.push(msg("h2", 2, "b"));
-        buf.push(msg("h1", 1, "a-dup"));
-        buf.push(msg("h3", 3, "c"));
-        let rows = buf.snapshot(None, 10);
+        assert!(buf.push(msg("h1", 1, "a")).is_some());
+        assert!(buf.push(msg("h2", 2, "b")).is_some());
+        assert!(buf.push(msg("h1", 1, "a-dup")).is_none());
+        assert!(buf.push(msg("h3", 3, "c")).is_some());
+        let rows = buf.snapshot(None, None, 10);
         assert_eq!(rows.len(), 2);
         assert_eq!(rows[0]["message_hash"], "h2");
         assert_eq!(rows[1]["message_hash"], "h3");
     }
 
     #[test]
-    fn since_ts_filters_and_limit_keeps_newest() {
+    fn since_ts_filters_exclusive_and_limit_keeps_newest() {
         let buf = LxmfInboundBuffer::new(10);
         buf.push(msg("h1", 100, "a"));
         buf.push(msg("h2", 200, "b"));
         buf.push(msg("h3", 300, "c"));
-        let rows = buf.snapshot(Some(200), 2);
-        assert_eq!(rows.len(), 2);
-        assert_eq!(rows[0]["message_hash"], "h2");
-        assert_eq!(rows[1]["message_hash"], "h3");
+        let rows = buf.snapshot(Some(200), None, 2);
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0]["message_hash"], "h3");
+    }
+
+    #[test]
+    fn since_ts_at_boundary_returns_empty() {
+        let buf = LxmfInboundBuffer::new(10);
+        buf.push(msg("h2", 200, "b"));
+        let rows = buf.snapshot(Some(200), None, 10);
+        assert!(rows.is_empty());
+    }
+
+    #[test]
+    fn since_ts_none_returns_full_chronological_buffer() {
+        let buf = LxmfInboundBuffer::new(10);
+        buf.push(msg("h1", 100, "a"));
+        buf.push(msg("h2", 200, "b"));
+        buf.push(msg("h3", 300, "c"));
+        let rows = buf.snapshot(None, None, 10);
+        assert_eq!(rows.len(), 3);
+        assert_eq!(rows[0]["message_hash"], "h1");
+        assert_eq!(rows[1]["message_hash"], "h2");
+        assert_eq!(rows[2]["message_hash"], "h3");
+    }
+
+    #[test]
+    fn same_ms_twins_recoverable_via_ring_seq_cursor() {
+        let buf = LxmfInboundBuffer::new(10);
+        let a = buf.push(msg("h_a", 200, "a")).expect("accepted");
+        let b = buf.push(msg("h_b", 200, "b")).expect("accepted");
+        let seq_a = a["ring_seq"].as_u64().expect("seq a");
+        let seq_b = b["ring_seq"].as_u64().expect("seq b");
+        assert!(seq_b > seq_a);
+
+        // After processing only the first twin, the second same-ms row must still be returned.
+        let after_a = buf.snapshot(Some(200), Some(seq_a), 10);
+        assert_eq!(after_a.len(), 1);
+        assert_eq!(after_a[0]["message_hash"], "h_b");
+        assert_eq!(after_a[0]["ring_seq"], seq_b);
+
+        // Complete cursor at the second twin — no reprocessing.
+        let after_b = buf.snapshot(Some(200), Some(seq_b), 10);
+        assert!(after_b.is_empty());
+
+        // Timestamp-only exclusive bound still drops the whole ms bucket (legacy clients).
+        let ts_only = buf.snapshot(Some(200), None, 10);
+        assert!(ts_only.is_empty());
+    }
+
+    #[test]
+    fn push_stamps_monotonic_ring_seq_on_accepted_rows() {
+        let buf = LxmfInboundBuffer::new(10);
+        let a = buf.push(msg("h1", 1, "a")).expect("a");
+        let b = buf.push(msg("h2", 2, "b")).expect("b");
+        assert_eq!(a["ring_seq"], 1);
+        assert_eq!(b["ring_seq"], 2);
     }
 }