Feat/commit loss epoch gap backfill

crates/marmot-app/src/client/epoch_stall.rs

@@ -0,0 +1,200 @@
+//! Detection policy for epoch-gap backfill (commit-loss recovery).
+//!
+//! A device that misses a single commit sits stuck below its group's live
+//! epoch: it keeps receiving that group's later-epoch traffic but cannot decrypt
+//! any of it — the kind-445 envelope is sealed under the per-epoch exporter
+//! secret and carries no cleartext epoch, so every such message fails to peel.
+//! This detector turns that otherwise-invisible signal into a per-group "the
+//! group moved on without me" decision *without ever decrypting the traffic*: it
+//! counts the distinct undecryptable messages a group accumulates while its
+//! epoch does not advance, and signals a backfill once that count crosses a
+//! threshold.
+//!
+//! The policy is deliberately I/O-free so it can be unit-tested in isolation;
+//! the recovery action it triggers — a full-history transport replay — lives in
+//! the caller, which owns the runtime.
+
+use std::collections::{HashMap, HashSet};
+
+use cgka_traits::{EpochId, GroupId};
+
+/// Distinct undecryptable messages a group may accumulate at one stalled epoch
+/// before the runtime reads it as stuck and triggers an epoch-gap backfill.
+///
+/// This is an empirical estimate with structural safety — the `CATCH_UP_GRACE_MS`
+/// class, not a uniquely-derived bound like `EPOCH_DIVERGENCE_MIN_LAG`. It was
+/// chosen by replaying this detector over the two real forensic exports on hand
+/// (2026-07-15, a single cohort): the genuinely stuck device accumulated 45
+/// distinct undecryptables at its stalled epoch, while the healthy tip devices
+/// never exceeded 7 (a diverged peer's complete send burst). 8 is the smallest
+/// count above that healthy plateau. Its safety is structural on both sides: too
+/// low costs at most one debounced full-history replay per (group, epoch) — the
+/// same operation a key-package publish already performs — while too high only
+/// delays healing, since the count is monotone while a group stays stuck. Being
+/// single-cohort, it should be firmed up against more cohorts before it is
+/// treated as general.
+pub(crate) const EPOCH_STALL_BACKFILL_THRESHOLD: usize = 8;
+
+/// Per-group stall accounting.
+struct GroupStall {
+    /// The epoch the undecryptable messages accumulated at; a new epoch resets
+    /// the count, because advancing proves the group's commits are reaching us.
+    epoch: EpochId,
+    /// Distinct undecryptable message ids seen at `epoch` (hex), capped at the
+    /// threshold — the identity is attacker-mintable, so the set never needs to
+    /// grow past the point where it decides.
+    undecryptable: HashSet<String>,
+    /// The epoch a backfill was last signalled for, so the detector signals at
+    /// most once per stalled epoch.
+    fired_at_epoch: Option<EpochId>,
+}
+
+/// Decides, per group, when a run of undecryptable traffic at a stalled epoch
+/// means the group has advanced past this device and a backfill is warranted.
+pub(crate) struct EpochStallDetector {
+    threshold: usize,
+    groups: HashMap<GroupId, GroupStall>,
+}
+
+impl EpochStallDetector {
+    pub(crate) fn new(threshold: usize) -> Self {
+        Self {
+            threshold,
+            groups: HashMap::new(),
+        }
+    }
+
+    /// Record that an account-wide full-history replay was just triggered. One
+    /// replay re-fetches every group's history, so suppress a further backfill
+    /// for every currently-tracked group at its current epoch: N groups stuck at
+    /// once cost one replay, not N. A group re-arms only when its epoch advances
+    /// and it stalls again at the new epoch.
+    pub(crate) fn mark_replayed(&mut self) {
+        for stall in self.groups.values_mut() {
+            stall.fired_at_epoch = Some(stall.epoch);
+        }
+    }
+
+    /// Record one undecryptable message for `group` observed while the group is
+    /// at `epoch`. Returns `true` exactly once when the group crosses the
+    /// threshold at a stalled epoch and a backfill should be triggered.
+    pub(crate) fn observe_undecryptable(
+        &mut self,
+        group: GroupId,
+        message: String,
+        epoch: EpochId,
+    ) -> bool {
+        let stall = self.groups.entry(group).or_insert_with(|| GroupStall {
+            epoch,
+            undecryptable: HashSet::new(),
+            fired_at_epoch: None,
+        });
+        if stall.epoch != epoch {
+            stall.epoch = epoch;
+            stall.undecryptable.clear();
+            stall.fired_at_epoch = None;
+        }
+        // The message identity is attacker-mintable (a fresh envelope is a fresh
+        // id), so the set never needs to grow past the point where it decides.
+        if stall.undecryptable.len() < self.threshold {
+            stall.undecryptable.insert(message);
+        }
+        let crossed = stall.undecryptable.len() >= self.threshold;
+        if crossed && stall.fired_at_epoch != Some(epoch) {
+            stall.fired_at_epoch = Some(epoch);
+            true
+        } else {
+            false
+        }
+    }
+}
+
+impl Default for EpochStallDetector {
+    fn default() -> Self {
+        Self::new(EPOCH_STALL_BACKFILL_THRESHOLD)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn group(byte: u8) -> GroupId {
+        GroupId::new(vec![byte])
+    }
+
+    #[test]
+    fn signals_backfill_after_threshold_distinct_undecryptables_at_a_stable_epoch() {
+        let mut detector = EpochStallDetector::new(3);
+        let g = group(0x01);
+        let e = EpochId(19);
+
+        assert!(!detector.observe_undecryptable(g.clone(), "m1".into(), e));
+        assert!(!detector.observe_undecryptable(g.clone(), "m2".into(), e));
+        assert!(
+            detector.observe_undecryptable(g.clone(), "m3".into(), e),
+            "the threshold-crossing message should signal a backfill"
+        );
+    }
+
+    #[test]
+    fn signals_at_most_once_per_stalled_epoch() {
+        let mut detector = EpochStallDetector::new(3);
+        let g = group(0x01);
+        let e = EpochId(19);
+
+        detector.observe_undecryptable(g.clone(), "m1".into(), e);
+        detector.observe_undecryptable(g.clone(), "m2".into(), e);
+        assert!(detector.observe_undecryptable(g.clone(), "m3".into(), e));
+        // Further undecryptable traffic at the same stalled epoch must not
+        // re-signal: one replay per stalled epoch is enough, and re-signalling
+        // would let a burst (or a spray of attacker-minted ids) trigger a storm.
+        assert!(!detector.observe_undecryptable(g.clone(), "m4".into(), e));
+        assert!(!detector.observe_undecryptable(g.clone(), "m5".into(), e));
+    }
+
+    #[test]
+    fn mark_replayed_collapses_a_storm_of_simultaneously_stuck_groups() {
+        let mut detector = EpochStallDetector::new(3);
+        let a = group(0x0A);
+        let b = group(0x0B);
+        let e = EpochId(19);
+
+        // Group A crosses the threshold and the caller runs ONE account-wide
+        // replay (which re-fetches every group's history, B included).
+        detector.observe_undecryptable(a.clone(), "a1".into(), e);
+        detector.observe_undecryptable(a.clone(), "a2".into(), e);
+        assert!(detector.observe_undecryptable(a.clone(), "a3".into(), e));
+
+        // Group B was accumulating undecryptables at the same epoch in the same
+        // drain but had not yet crossed the threshold.
+        detector.observe_undecryptable(b.clone(), "b1".into(), e);
+        detector.observe_undecryptable(b.clone(), "b2".into(), e);
+
+        detector.mark_replayed();
+
+        // B crossing the threshold after the replay must NOT trigger a second
+        // one: the single replay already covered it.
+        assert!(
+            !detector.observe_undecryptable(b.clone(), "b3".into(), e),
+            "one account-wide replay should cover every stuck group at this epoch"
+        );
+    }
+
+    #[test]
+    fn an_epoch_advance_resets_the_count() {
+        let mut detector = EpochStallDetector::new(3);
+        let g = group(0x01);
+
+        detector.observe_undecryptable(g.clone(), "m1".into(), EpochId(19));
+        detector.observe_undecryptable(g.clone(), "m2".into(), EpochId(19));
+        // The group advanced to epoch 20 — its commits are reaching us again, so
+        // the earlier undecryptables must not count toward a stall at epoch 20.
+        assert!(!detector.observe_undecryptable(g.clone(), "m3".into(), EpochId(20)));
+        assert!(!detector.observe_undecryptable(g.clone(), "m4".into(), EpochId(20)));
+        assert!(
+            detector.observe_undecryptable(g.clone(), "m5".into(), EpochId(20)),
+            "the count should restart at the new epoch, not carry over"
+        );
+    }
+}

crates/marmot-app/src/client/mod.rs

@@ -44,10 +44,12 @@ use crate::{
 };
 
 mod audit;
+mod epoch_stall;
 mod projection;
 mod push;
 mod sync;
 
+use epoch_stall::EpochStallDetector;
 use push::notification_trigger_for_intent;
 // Re-exported so the crate's `tests` module can keep calling
 // `client::is_own_relay_echo`; the function itself lives in `client::sync`.
@@ -71,6 +73,13 @@ pub struct AppClient {
     /// `WelcomeDeliveryPending` event so callers learn a member is unjoinable
     /// without polling (mdk#352).
     pub(crate) pending_welcome_delivery_events: Vec<PendingWelcomeDelivery>,
+    /// Per-group detector for the epoch-gap backfill (commit-loss recovery): it
+    /// counts the distinct undecryptable messages a group accumulates at a
+    /// stalled epoch. Ephemeral session state, like the pending sets above.
+    pub(crate) epoch_stall: EpochStallDetector,
+    /// Set when [`epoch_stall`] arms a backfill during ingest; drained after the
+    /// sync by running the full-history transport replay.
+    pub(crate) epoch_backfill_pending: bool,
 }
 
 /// A point-in-time copy of the live session's read-only group projections

crates/marmot-app/src/client/sync.rs

@@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet};
 
 use cgka_traits::TransportAdapter;
 use cgka_traits::app_event::MARMOT_APP_EVENT_KIND_CHAT;
-use cgka_traits::ingest::IngestOutcome;
+use cgka_traits::ingest::{IngestOutcome, StaleReason};
 use storage_sqlite::clamp_to_max_future_skew;
 use tokio::time::timeout;
 use transport_nostr_peeler::NostrTransportEvent;
@@ -17,6 +17,7 @@ use crate::{
 };
 
 use super::AppClient;
+use crate::config::CursorPersistence;
 
 impl AppClient {
     pub(crate) fn take_pending_convergence_groups(&mut self) -> Vec<cgka_traits::GroupId> {
@@ -213,6 +214,7 @@ impl AppClient {
                 && summary.messages.is_empty()
                 && summary.events.is_empty()
                 && self.pending_convergence_groups.is_empty()
+                && !self.epoch_backfill_pending
             {
                 continue;
             }
@@ -288,11 +290,13 @@ impl AppClient {
     ) -> Result<(), AppError> {
         let source_message_id_hex = hex::encode(delivery.message.id.as_slice());
         let source_recorded_at = delivery.message.timestamp.0;
+        let group_id_hint = delivery.group_id_hint.clone();
         let effects = self.runtime.ingest_delivery(delivery).await?;
         fail_if_publish_failed(&effects.effects)?;
         self.remember_buffered_convergence_outcome(&effects.outcome);
         self.remember_pending_convergence_effects(&effects.effects);
         self.remember_transport_cursor(source_recorded_at);
+        self.detect_epoch_stall(group_id_hint, &source_message_id_hex, &effects.outcome);
         self.observe_account_device_effects(
             &effects.effects,
             display_names,
@@ -303,6 +307,58 @@ impl AppClient {
         .await
     }
 
+    /// Feed an undecryptable group delivery to the epoch-stall detector, arming a
+    /// backfill once a group has accumulated enough undecryptable traffic at a
+    /// stalled epoch (see [`super::epoch_stall`]). Only observed under
+    /// `CursorPersistence::Advance`: a `Frozen` wake-collection pass must not own
+    /// recovery, and the main app sees the same evidence on its own next sync.
+    fn detect_epoch_stall(
+        &mut self,
+        group_id_hint: Option<cgka_traits::GroupId>,
+        message_id_hex: &str,
+        outcome: &IngestOutcome,
+    ) {
+        if self.app.cursor_persistence() != CursorPersistence::Advance {
+            return;
+        }
+        if !matches!(
+            outcome,
+            IngestOutcome::Stale {
+                reason: StaleReason::PeelFailed
+            }
+        ) {
+            return;
+        }
+        let Some(group_id) = group_id_hint else {
+            return;
+        };
+        // A group we cannot resolve (unknown or quarantined) has its own recovery
+        // surface; do not track it here.
+        let Ok(record) = self.runtime.group_record(&group_id) else {
+            return;
+        };
+        if self
+            .epoch_stall
+            .observe_undecryptable(group_id, message_id_hex.to_owned(), record.epoch)
+        {
+            self.epoch_backfill_pending = true;
+        }
+    }
+
+    /// Recover any group that stalled below its live epoch during ingest by
+    /// replaying the account's full transport history (`since = None`). One replay
+    /// re-fetches every group, so the detector collapses simultaneously-stuck
+    /// groups into a single replay. A no-op when nothing stalled.
+    pub(crate) async fn run_pending_epoch_backfill(&mut self) -> Result<(), AppError> {
+        if !self.epoch_backfill_pending {
+            return Ok(());
+        }
+        self.runtime.activate_transport(None).await?;
+        self.epoch_stall.mark_replayed();
+        self.epoch_backfill_pending = false;
+        Ok(())
+    }
+
     pub(crate) async fn advance_convergence_after_runtime_sync(
         &mut self,
         group_id: &cgka_traits::GroupId,

crates/marmot-app/src/lib.rs

@@ -1081,6 +1081,8 @@ impl MarmotApp {
             pending_projection_updates: Vec::new(),
             pending_convergence_groups: std::collections::HashSet::new(),
             pending_welcome_delivery_events: Vec::new(),
+            epoch_stall: Default::default(),
+            epoch_backfill_pending: false,
         };
         // One-time upgrade backfill: derive `self_membership` for pre-0018 rows
         // from current engine state so groups the local account already left /

crates/marmot-app/src/runtime/account_worker.rs

@@ -449,6 +449,14 @@ async fn run_app_runtime_account_worker(
         Ok(summary) => {
             publish_app_runtime_summary(&events, &account_id_hex, &account_label, &summary);
             scheduled_convergence.schedule_groups(client.take_pending_convergence_groups());
+            if let Err(err) = client.run_pending_epoch_backfill().await {
+                publish_app_runtime_account_error(
+                    &events,
+                    &account_id_hex,
+                    &account_label,
+                    account_error_message("epoch-gap backfill failed", &err),
+                );
+            }
             if sync_summary_triggers_audit_tracker_update(&summary) {
                 shared.schedule_audit_log_tracker_update("startup_sync");
             }
@@ -574,6 +582,14 @@ async fn run_app_runtime_account_worker(
                         reconnect_backoff.reset();
                         publish_app_runtime_summary(&events, &account_id_hex, &account_label, &summary);
                         scheduled_convergence.schedule_groups(client.take_pending_convergence_groups());
+                        if let Err(err) = client.run_pending_epoch_backfill().await {
+                            publish_app_runtime_account_error(
+                                &events,
+                                &account_id_hex,
+                                &account_label,
+                                account_error_message("epoch-gap backfill failed", &err),
+                            );
+                        }
                         if sync_summary_triggers_audit_tracker_update(&summary) {
                             shared.schedule_audit_log_tracker_update("receive");
                         }