From a39435d3045d6f1bc12996e3f27150c91a36e876 Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Thu, 16 Jul 2026 11:10:44 -0400 Subject: [PATCH 1/3] fix(transport-nostr-adapter): register routing state before issuing subscriptions --- crates/transport-nostr-adapter/src/lib.rs | 35 +++-- .../tests/inbound_routing.rs | 122 ++++++++++++++++++ 2 files changed, 143 insertions(+), 14 deletions(-) diff --git a/crates/transport-nostr-adapter/src/lib.rs b/crates/transport-nostr-adapter/src/lib.rs index 7c081b2a..fadb136d 100644 --- a/crates/transport-nostr-adapter/src/lib.rs +++ b/crates/transport-nostr-adapter/src/lib.rs @@ -526,6 +526,27 @@ impl TransportAdapter for NostrTransportAdapter { for group in &activation.group_subscriptions { issued.push(group_subscription(&account_id, group, activation.since)); } + // Register routing/telemetry state BEFORE the relay REQs go out: a + // relay may stream stored events the moment it sees a subscription, + // and an event arriving before the routes exist is dropped as + // unroutable — stored catch-up history would be lost with nothing to + // re-request it. + { + let now_ms = self.now_ms(); + let mut state = self.state.write().await; + // Reactivation replaces the account's routes; evict telemetry for + // the old subscription ids first (before recording the new starts, + // since unchanged endpoint sets reuse the same ids). + state.forget_account_subscription_starts(&account_id); + if replaced_count > 0 { + // The blanket `unsubscribe_account` above supersedes any queued + // per-subscription unsubscribes for this account. + state.clear_pending_unsubscribes_for_account(&account_id); + } + state.record_subscription_starts(&issued, now_ms); + state.activate(activation, replaced_count); + } + self.subscribe_all("activate_account", &issued).await?; tracing::debug!( target: "transport_nostr_adapter::adapter", @@ -533,20 +554,6 @@ impl TransportAdapter for NostrTransportAdapter { issued_count = issued.len(), "all transport subscriptions issued" ); - - let now_ms = self.now_ms(); - let mut state = self.state.write().await; - // Reactivation replaces the account's routes; evict telemetry for the - // old subscription ids first (before recording the new starts, since - // unchanged endpoint sets reuse the same ids). - state.forget_account_subscription_starts(&account_id); - if replaced_count > 0 { - // The blanket `unsubscribe_account` above supersedes any queued - // per-subscription unsubscribes for this account. - state.clear_pending_unsubscribes_for_account(&account_id); - } - state.record_subscription_starts(&issued, now_ms); - state.activate(activation, replaced_count); Ok(()) } diff --git a/crates/transport-nostr-adapter/tests/inbound_routing.rs b/crates/transport-nostr-adapter/tests/inbound_routing.rs index 318f8408..27a2e8e7 100644 --- a/crates/transport-nostr-adapter/tests/inbound_routing.rs +++ b/crates/transport-nostr-adapter/tests/inbound_routing.rs @@ -94,6 +94,79 @@ impl NostrRelayClient for ConcurrentSubscribeRelayClient { } } +/// Simulates a relay holding stored events: the moment it sees a group REQ it +/// streams a matching stored event back through `handle_relay_event`, before +/// `subscribe` returns — i.e. inside the activation's subscribe window. The +/// adapter owns the relay client, so the back-reference used for the replay is +/// attached after construction. +#[derive(Default)] +struct StoredReplayRelayClient { + adapter: Mutex>, + replayed_deliveries: AtomicUsize, +} + +impl StoredReplayRelayClient { + fn attach(&self, adapter: &NostrTransportAdapter) { + *self.adapter.lock().unwrap() = Some(adapter.clone()); + } +} + +#[async_trait] +impl NostrRelayClient for StoredReplayRelayClient { + async fn subscribe( + &self, + subscription: transport_nostr_adapter::NostrSubscription, + ) -> Result<(), cgka_traits::TransportAdapterError> { + let NostrSubscription::Group { + transport_group_id, + endpoints, + .. + } = &subscription + else { + return Ok(()); + }; + let adapter = self + .adapter + .lock() + .unwrap() + .clone() + .expect("adapter attached before subscribe"); + let delivered = adapter + .handle_relay_event(NostrRelayEvent { + endpoint: endpoints[0].clone(), + subscription_id: Some(subscription.subscription_id()), + event: group_event("30", transport_group_id), + }) + .await?; + self.replayed_deliveries + .fetch_add(delivered, Ordering::SeqCst); + Ok(()) + } + + async fn unsubscribe( + &self, + _subscription: transport_nostr_adapter::NostrSubscription, + ) -> Result<(), cgka_traits::TransportAdapterError> { + Ok(()) + } + + async fn unsubscribe_account( + &self, + _account_id: &MemberId, + ) -> Result<(), cgka_traits::TransportAdapterError> { + Ok(()) + } + + async fn publish_event( + &self, + endpoints: &[TransportEndpoint], + _event: &NostrTransportEvent, + _required_acks: usize, + ) -> Result { + Ok(NostrPublishOutcome::accepted(endpoints.to_vec())) + } +} + #[derive(Default)] struct FakeRelayClient { subscriptions: Mutex>, @@ -544,6 +617,55 @@ async fn subscribed_group_event_becomes_account_scoped_delivery() { ); } +// Regression for the cold-boot catch-up race: a relay may stream stored events +// the moment a REQ is issued, so a stored group event can arrive inside +// `activate_account`'s subscribe window. Routing state must already cover the +// account by then; an event arriving before the routes exist is dropped as +// unroutable, and that stored history is lost — nothing re-requests it. +#[tokio::test] +async fn stored_event_replayed_during_activation_subscribe_is_delivered() { + let relay = Arc::new(StoredReplayRelayClient::default()); + let adapter = NostrTransportAdapter::new(relay.clone()); + relay.attach(&adapter); + let account_id = MemberId::new(vec![0xA1; 32]); + let group_id = cgka_traits::GroupId::new(vec![0xB2; 32]); + let transport_group_id = vec![0xC3; 32]; + let endpoint = TransportEndpoint("wss://group.example".into()); + + adapter + .activate_account(TransportAccountActivation { + account_id: account_id.clone(), + inbox_endpoints: vec![TransportEndpoint("wss://inbox.example".into())], + group_subscriptions: vec![TransportGroupSubscription { + group_id: group_id.clone(), + transport_group_id: transport_group_id.clone(), + endpoints: vec![endpoint.clone()], + }], + since: Some(Timestamp(1_700_000_000)), + }) + .await + .expect("activation succeeds"); + + assert_eq!( + relay.replayed_deliveries.load(Ordering::SeqCst), + 1, + "the stored event replayed during subscribe must route to the account" + ); + let delivery = adapter + .receive() + .await + .expect("receive succeeds") + .expect("delivery available"); + assert_eq!(delivery.account_id, account_id); + assert_eq!(delivery.group_id_hint, Some(group_id)); + assert_eq!(delivery.source.plane, TransportDeliveryPlane::Group); + assert_eq!(delivery.source.endpoint, Some(endpoint)); + assert_eq!( + delivery.message.envelope, + TransportEnvelope::GroupMessage { transport_group_id } + ); +} + #[tokio::test] async fn observe_relay_event_records_every_relay_copy_for_spread() { let relay = Arc::new(FakeRelayClient::default()); From 90da57c44d95b33f2d9b42948762c9d31c42849f Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Thu, 16 Jul 2026 11:14:36 -0400 Subject: [PATCH 2/3] feat(marmot-app): recover commit-loss backlog by replaying full transport history when a group stalls below its live epoch (epoch-gap backfill) --- crates/marmot-app/src/client/epoch_stall.rs | 200 ++++++++++++++ crates/marmot-app/src/client/mod.rs | 9 + crates/marmot-app/src/client/sync.rs | 56 +++- crates/marmot-app/src/lib.rs | 2 + .../marmot-app/src/runtime/account_worker.rs | 16 ++ crates/marmot-app/tests/since_floor.rs | 252 ++++++++++++++++-- 6 files changed, 519 insertions(+), 16 deletions(-) create mode 100644 crates/marmot-app/src/client/epoch_stall.rs diff --git a/crates/marmot-app/src/client/epoch_stall.rs b/crates/marmot-app/src/client/epoch_stall.rs new file mode 100644 index 00000000..fb4e89d7 --- /dev/null +++ b/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, + /// The epoch a backfill was last signalled for, so the detector signals at + /// most once per stalled epoch. + fired_at_epoch: Option, +} + +/// 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, +} + +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" + ); + } +} diff --git a/crates/marmot-app/src/client/mod.rs b/crates/marmot-app/src/client/mod.rs index 813c0723..bcf89468 100644 --- a/crates/marmot-app/src/client/mod.rs +++ b/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, + /// 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 diff --git a/crates/marmot-app/src/client/sync.rs b/crates/marmot-app/src/client/sync.rs index 1eaeaa14..f6d0c8d0 100644 --- a/crates/marmot-app/src/client/sync.rs +++ b/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 { @@ -288,11 +289,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 +306,57 @@ 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, + 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 !std::mem::take(&mut self.epoch_backfill_pending) { + return Ok(()); + } + self.runtime.activate_transport(None).await?; + self.epoch_stall.mark_replayed(); + Ok(()) + } + pub(crate) async fn advance_convergence_after_runtime_sync( &mut self, group_id: &cgka_traits::GroupId, diff --git a/crates/marmot-app/src/lib.rs b/crates/marmot-app/src/lib.rs index 66889c57..6ecbe509 100644 --- a/crates/marmot-app/src/lib.rs +++ b/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 / diff --git a/crates/marmot-app/src/runtime/account_worker.rs b/crates/marmot-app/src/runtime/account_worker.rs index c61815ed..abb99170 100644 --- a/crates/marmot-app/src/runtime/account_worker.rs +++ b/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"); } diff --git a/crates/marmot-app/tests/since_floor.rs b/crates/marmot-app/tests/since_floor.rs index e8c22f1d..1802340f 100644 --- a/crates/marmot-app/tests/since_floor.rs +++ b/crates/marmot-app/tests/since_floor.rs @@ -18,6 +18,13 @@ //! below-floor probe should start arriving (via backfill) and this file's //! expectations must be updated accordingly. //! +//! Epoch-gap backfill has since landed, splitting this file in two: +//! `cold_restart_since_floor_permanently_drops_backlog_below_it` still pins the +//! UNARMED floor-drop (one undecryptable probe stays under the backfill +//! threshold, so it stays dropped forever), while +//! `stalled_epoch_backfill_recovers_below_floor_backlog_when_armed` pins the +//! ARMED recovery of below-floor backlog through the full-history backfill. +//! //! # Why one account per store //! //! `MarmotRelayPlane` runs one shared router per `MarmotApp`/store @@ -98,7 +105,7 @@ use nostr_sdk::prelude::{ }; use tokio::time::sleep; use transport_nostr_adapter::{NostrRelayClient, NostrSdkRelayClient}; -use transport_nostr_peeler::NostrTransportEvent; +use transport_nostr_peeler::{NOSTR_GROUP_CONTENT_MIN_LEN, NostrTransportEvent}; /// Mirrors `marmot_app::lib.rs`'s private `APP_RUNTIME_RELAY_REBUILD_LOOKBACK` /// (120s). Not importable (crate-private); every production runtime plane @@ -106,6 +113,12 @@ use transport_nostr_peeler::NostrTransportEvent; /// consistent with itself. const REBUILD_LOOKBACK_SECS: u64 = 120; +/// Mirrors `marmot_app::client::epoch_stall::EPOCH_STALL_BACKFILL_THRESHOLD` +/// (crate-private, hence mirrored): the number of distinct undecryptable +/// messages a group must accumulate at one stalled epoch before the runtime +/// arms a full-history epoch-gap backfill. +const BACKFILL_THRESHOLD: usize = 8; + /// How long a cold boot's initial catch-up is allowed to take before this /// test gives up waiting for it. Generous relative to the crate's own /// `SDK_FIRST_SYNC_WAIT` (750ms) / `SDK_DRAIN_WAIT` (250ms) internals. @@ -187,32 +200,62 @@ async fn inbound_events_delivered(app: &MarmotApp) -> usize { app.relay_telemetry().await.metrics.inbound_events_delivered } +/// Poll delivery telemetry until at least `target` inbound events have been +/// routed to the account, failing at `CATCH_UP_DEADLINE`. A behavioral await +/// on observable state: the epoch-gap backfill replay runs *after* the initial +/// catch-up op is recorded, so `wait_for_first_catch_up` alone cannot see it — +/// the delivered count is the signal that the replay has landed. +async fn wait_for_inbound_delivered(app: &MarmotApp, target: usize) { + let deadline = Instant::now() + CATCH_UP_DEADLINE; + loop { + let delivered = inbound_events_delivered(app).await; + if delivered >= target { + return; + } + assert!( + Instant::now() < deadline, + "inbound deliveries stalled at {delivered}, expected at least \ + {target} within the deadline", + ); + sleep(Duration::from_millis(25)).await; + } +} + /// Publish a kind-445 group-message event at the wire level with a fresh, /// arbitrary (non-member) ephemeral signing key and a caller-chosen /// `created_at`. This is legitimate wire traffic, not a forged event: real /// kind-445 senders are always a fresh per-event key, never the sender's /// Marmot account identity. This mirrors the existing `signed_group_event_dto` -/// precedent in `transport-nostr-adapter/src/sdk_client.rs`'s own tests. The -/// content is undecryptable garbage — this test exercises delivery, not -/// decryption. +/// precedent in `transport-nostr-adapter/src/sdk_client.rs`'s own tests. +/// +/// The content is an *envelope-shaped* undecryptable probe. Per +/// `spec/transports/nostr.md`, kind-445 content is `base64(nonce || +/// ciphertext)`; the probe carries a zero nonce plus marker bytes that +/// authenticate under no key, so the recipient peels it to a clean +/// `PeelerError::DecryptFailed` — the same shape as real traffic sealed under +/// an exporter secret this device does not hold, and the shape the epoch-stall +/// detector counts as `IngestOutcome::Stale { reason: PeelFailed }`. (A +/// shorter-than-envelope body would instead be `Malformed`, a hard ingest +/// error rather than an undecryptable-message observation.) async fn publish_garbage_group_message_at( relay_url: &str, nostr_group_id_hex: &str, created_at: u64, marker: &str, ) { + // 12-byte zero nonce, then the marker as the unauthenticatable ciphertext. + let mut envelope = vec![0u8; 12]; + envelope.extend_from_slice(format!("since-floor-probe:{marker}").as_bytes()); + assert!(envelope.len() >= NOSTR_GROUP_CONTENT_MIN_LEN); let ephemeral = Keys::generate(); - let signed = EventBuilder::new( - Kind::MlsGroupMessage, - BASE64_STANDARD.encode(format!("since-floor-probe:{marker}").as_bytes()), - ) - .tags([Tag::custom( - TagKind::SingleLetter(SingleLetterTag::lowercase(Alphabet::H)), - [nostr_group_id_hex.to_owned()], - )]) - .custom_created_at(NostrTimestamp::from_secs(created_at)) - .sign_with_keys(&ephemeral) - .expect("sign ephemeral kind-445 test event"); + let signed = EventBuilder::new(Kind::MlsGroupMessage, BASE64_STANDARD.encode(envelope)) + .tags([Tag::custom( + TagKind::SingleLetter(SingleLetterTag::lowercase(Alphabet::H)), + [nostr_group_id_hex.to_owned()], + )]) + .custom_created_at(NostrTimestamp::from_secs(created_at)) + .sign_with_keys(&ephemeral) + .expect("sign ephemeral kind-445 test event"); let transport_event = NostrTransportEvent::from_nostr_event(&signed).expect("dto from signed event"); let relay_client = NostrSdkRelayClient::new(NostrSdkClient::builder().build()); @@ -356,3 +399,182 @@ async fn cold_restart_since_floor_permanently_drops_backlog_below_it() { ); runtime_bob_boot3.shutdown().await; } + +/// The armed counterpart to +/// [`cold_restart_since_floor_permanently_drops_backlog_below_it`]: with enough +/// undecryptable traffic at a stalled epoch, epoch-gap backfill recovers the +/// backlog the rebuilt subscription's since floor would otherwise drop forever. +/// +/// The probes play two distinct roles on either side of the floor: +/// +/// - **Arming** (`arm-0`..`arm-7`, above the floor): the cold boot's ordinary +/// floored catch-up delivers them, each fails to peel, and — being distinct +/// event ids — each counts as a distinct undecryptable message at bob's +/// stalled epoch. The eighth crosses `EPOCH_STALL_BACKFILL_THRESHOLD` and +/// arms a full-history transport replay (`since = None`). +/// - **Recovery target** (`below-target`, below the floor): the floored +/// catch-up can never deliver it; the only path to ingest is the unfloored +/// backfill replay the arming probes trigger. +/// +/// Boot 2 pins the heal itself. Boot 3 pins that the heal is *durable and +/// debounced*: ingested event ids persist in the account's seen-event state, +/// so a later cold boot neither re-arms the detector nor replays full history +/// again — recovery costs one backfill, not one per boot. +#[tokio::test] +async fn stalled_epoch_backfill_recovers_below_floor_backlog_when_armed() { + let (_relay, url) = mock_relay().await; + + // Same one-account-per-store split as the floor-drop test above. + let dir_bob = tempfile::tempdir().unwrap(); + let home_bob = AccountHome::open(dir_bob.path()); + home_bob.create_account("bob").unwrap(); + let bob_id = home_bob.account("bob").unwrap().account_id_hex; + + let dir_alice = tempfile::tempdir().unwrap(); + let home_alice = AccountHome::open(dir_alice.path()); + home_alice.create_account("alice").unwrap(); + let app_alice = open_store(&dir_alice, &url); + + // --- boot 1: bob live, joins the group, receives one ordinary message --- + let app_bob_boot1 = open_store(&dir_bob, &url); + { + let mut bob_setup = app_bob_boot1.client("bob").await.unwrap(); + bob_setup.publish_key_package().await.unwrap(); + } + let runtime_bob_boot1 = MarmotAppRuntime::new(app_bob_boot1.clone()); + let mut events_bob_boot1 = runtime_bob_boot1.subscribe(); + runtime_bob_boot1.start().await.unwrap(); + + let mut alice_client = app_alice.client("alice").await.unwrap(); + let group_id = alice_client + .create_group("epoch-gap backfill recovery", &[bob_id.as_str()]) + .await + .unwrap(); + wait_for_event(&mut events_bob_boot1, |event| { + matches!( + event, + MarmotAppEvent::GroupJoined { account_id_hex, group_id: joined, .. } + if account_id_hex == &bob_id && joined == &group_id + ) + }) + .await; + + alice_client + .send(&group_id, b"ordinary traffic advances the cursor") + .await + .unwrap(); + wait_for_event(&mut events_bob_boot1, |event| { + matches!( + event, + MarmotAppEvent::MessageReceived(message) + if message.account_id_hex == bob_id + && message.message.group_id == group_id + && message.message.plaintext == "ordinary traffic advances the cursor" + ) + }) + .await; + + let group_id_hex = hex::encode(group_id.as_slice()); + let nostr_group_id_hex = app_bob_boot1 + .group("bob", &group_id_hex) + .unwrap() + .expect("bob's group projection") + .nostr_routing + .nostr_group_id_hex; + + // Measured, not assumed, exactly as in the floor-drop test: how many + // deliveries a fresh cold boot legitimately re-fetches (the welcome and + // the one ordinary message). + let legitimate_delivery_count = inbound_events_delivered(&app_bob_boot1).await; + + // Same floor placement as the floor-drop test above. + let reference_now = test_unix_now_seconds(); + let floor_estimate = reference_now.saturating_sub(REBUILD_LOOKBACK_SECS); + let below_floor_created_at = floor_estimate.saturating_sub(180); + let above_floor_created_at = floor_estimate.saturating_add(60); + assert!(above_floor_created_at < reference_now); + + // bob must be fully offline before any probe is published (see the module + // doc comment on live subscriptions ignoring `since`). + runtime_bob_boot1.shutdown().await; + + // The recovery target: below the floor, unreachable by any floored + // catch-up — only an unfloored backfill replay can deliver it. + publish_garbage_group_message_at( + &url, + &nostr_group_id_hex, + below_floor_created_at, + "below-target", + ) + .await; + // The arming probes: above the floor, so the ordinary catch-up delivers + // them. Distinct markers make distinct event ids, hence distinct + // undecryptable messages at bob's stalled epoch. + for arm in 0..BACKFILL_THRESHOLD { + publish_garbage_group_message_at( + &url, + &nostr_group_id_hex, + above_floor_created_at, + &format!("arm-{arm}"), + ) + .await; + } + + // Expected exact delivery count for the armed cold boot: + // legitimate_delivery_count — the catch-up re-fetches the welcome and + // the one ordinary message; + // + BACKFILL_THRESHOLD — the above-floor arming probes arrive + // through the floored catch-up and arm the + // detector at bob's stalled epoch; + // + 1 — the armed backfill's `since = None` replay + // recovers the below-floor probe. + // The unfloored replay re-serves every already-seen event too, but + // nostr-sdk emits an `Event` notification only for events new to its + // database, so re-fetches are not re-counted and the total is exact — a + // higher count would mean double-counted replays, a lower one a dropped + // probe. + let expected_healed = legitimate_delivery_count + BACKFILL_THRESHOLD + 1; + + // --- boot 2: cold restart; catch-up arms the detector, backfill heals --- + let app_bob_boot2 = open_store(&dir_bob, &url); + let runtime_bob_boot2 = MarmotAppRuntime::new(app_bob_boot2.clone()); + runtime_bob_boot2.start().await.unwrap(); + wait_for_first_catch_up(&runtime_bob_boot2).await; + wait_for_inbound_delivered(&app_bob_boot2, expected_healed).await; + // Settle grace so any spurious extra delivery lands before the + // exact-equality check (mirrors the floor-drop test's settle). + sleep(TELEMETRY_SETTLE_GRACE).await; + assert_eq!( + inbound_events_delivered(&app_bob_boot2).await, + expected_healed, + "an armed cold boot must deliver exactly the legitimate re-fetches, \ + the arming probes, and — via the epoch-gap backfill replay — the \ + below-floor probe, with no replayed duplicate re-counted", + ); + runtime_bob_boot2.shutdown().await; + + // --- boot 3: the heal is durable, not a per-boot replay storm. Boot 2 + // consumed the arming evidence (ingested event ids persist in the + // account's seen-event state and are skipped before ingest), so this + // independent cold boot re-fetches the above-floor history at the + // transport layer without re-arming the per-boot, in-memory detector: no + // second full-history replay fires, and the below-floor probe — already + // recovered once — stays below the rebuilt floor without being + // re-delivered. Exactly the legitimate re-fetches plus the eight + // above-floor probes arrive, and nothing else. --- + let expected_after_heal = legitimate_delivery_count + BACKFILL_THRESHOLD; + let app_bob_boot3 = open_store(&dir_bob, &url); + let runtime_bob_boot3 = MarmotAppRuntime::new(app_bob_boot3.clone()); + runtime_bob_boot3.start().await.unwrap(); + wait_for_first_catch_up(&runtime_bob_boot3).await; + wait_for_inbound_delivered(&app_bob_boot3, expected_after_heal).await; + sleep(TELEMETRY_SETTLE_GRACE).await; + assert_eq!( + inbound_events_delivered(&app_bob_boot3).await, + expected_after_heal, + "a cold boot after the heal must not replay full history again: the \ + backfill is debounced by the durable seen-event state, so only the \ + floored catch-up's above-floor re-fetches arrive", + ); + runtime_bob_boot3.shutdown().await; +} From 865d5b248a1734c31d2bb3f634fac1b4aa82d8c7 Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Thu, 16 Jul 2026 13:29:50 -0400 Subject: [PATCH 3/3] address PR review comments --- crates/marmot-app/src/client/sync.rs | 4 +- .../marmot-app/tests/next_event_backfill.rs | 200 ++++++++++++++++++ 2 files changed, 203 insertions(+), 1 deletion(-) create mode 100644 crates/marmot-app/tests/next_event_backfill.rs diff --git a/crates/marmot-app/src/client/sync.rs b/crates/marmot-app/src/client/sync.rs index f6d0c8d0..1217682f 100644 --- a/crates/marmot-app/src/client/sync.rs +++ b/crates/marmot-app/src/client/sync.rs @@ -214,6 +214,7 @@ impl AppClient { && summary.messages.is_empty() && summary.events.is_empty() && self.pending_convergence_groups.is_empty() + && !self.epoch_backfill_pending { continue; } @@ -349,11 +350,12 @@ impl AppClient { /// 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 !std::mem::take(&mut self.epoch_backfill_pending) { + if !self.epoch_backfill_pending { return Ok(()); } self.runtime.activate_transport(None).await?; self.epoch_stall.mark_replayed(); + self.epoch_backfill_pending = false; Ok(()) } diff --git a/crates/marmot-app/tests/next_event_backfill.rs b/crates/marmot-app/tests/next_event_backfill.rs new file mode 100644 index 00000000..525d88c3 --- /dev/null +++ b/crates/marmot-app/tests/next_event_backfill.rs @@ -0,0 +1,200 @@ +//! Steady-state coverage for the epoch-gap backfill seam in `next_event`. +//! +//! When a *live* delivery arms the epoch-gap backfill — a run of undecryptable +//! traffic at a stalled epoch crossing `EPOCH_STALL_BACKFILL_THRESHOLD` — the +//! arming produces no visible `SyncSummary` content (an undecryptable kind-445 +//! yields `IngestOutcome::Stale`, no `MarmotAppEvent`). `next_event` must still +//! RETURN so the worker's post-`next_event` +//! `run_pending_epoch_backfill` seam (`runtime/account_worker.rs`) runs. If the +//! empty-summary guard does not also check `epoch_backfill_pending`, the call +//! instead `continue`s and blocks on the next `receive()` — for a permanently +//! stalled group that next visible event may never arrive, so the armed backfill +//! is never acted on: a liveness gap. +//! +//! This drives `AppClient::next_event` directly (the worker's exact call, at +//! `account_worker.rs`) rather than through the full runtime. The observable is +//! precisely "did `next_event` return", which the telemetry-based `since_floor` +//! harness cannot see: it exercises the cold-boot catch-up seam +//! (`run_pending_epoch_backfill` after the initial `sync()`), not the live +//! `next_event` seam this test pins. Before the guard learned about +//! `epoch_backfill_pending` this call hangs, and the timeout below is the RED +//! signal. + +use std::time::{Duration, Instant}; + +use marmot_account::AccountHome; +use marmot_app::{MarmotApp, MarmotAppConfig}; +use nostr::base64::Engine as _; +use nostr::base64::engine::general_purpose::STANDARD as BASE64_STANDARD; +use nostr_relay_builder::MockRelay; +use nostr_sdk::prelude::{ + Alphabet, Client as NostrSdkClient, EventBuilder, Keys, Kind, SingleLetterTag, Tag, TagKind, + Timestamp as NostrTimestamp, +}; +use tokio::time::{sleep, timeout}; +use transport_nostr_adapter::{NostrRelayClient, NostrSdkRelayClient}; +use transport_nostr_peeler::{NOSTR_GROUP_CONTENT_MIN_LEN, NostrTransportEvent}; + +/// Mirrors `marmot_app::client::epoch_stall::EPOCH_STALL_BACKFILL_THRESHOLD` +/// (crate-private, hence mirrored): the number of distinct undecryptable +/// messages a group must accumulate at one stalled epoch before the runtime +/// arms a full-history epoch-gap backfill. +const BACKFILL_THRESHOLD: usize = 8; + +/// How long the arming `next_event` call is allowed to run before the test +/// gives up. With the fix it returns as soon as the threshold-crossing delivery +/// arms the backfill; without it the call blocks forever on the next +/// `receive()`, and this bound is the RED failure. +const NEXT_EVENT_DEADLINE: Duration = Duration::from_secs(10); + +/// Deadline for bob's welcome to arrive and the group to become live. +const JOIN_DEADLINE: Duration = Duration::from_secs(10); + +async fn mock_relay() -> (MockRelay, String) { + let relay = MockRelay::run().await.unwrap(); + let url = relay.url().await.to_string(); + (relay, url) +} + +fn open_store(dir: &tempfile::TempDir, relay_url: &str) -> MarmotApp { + MarmotApp::with_relay_and_config( + dir.path(), + relay_url.to_owned(), + MarmotAppConfig::default().with_allow_loopback_relay_endpoints(true), + ) +} + +fn test_unix_now_seconds() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +/// Publish an envelope-shaped undecryptable kind-445 probe with a fresh +/// ephemeral key and a caller-chosen `created_at`, h-tagged to +/// `nostr_group_id_hex`. Copied from `since_floor.rs`: real kind-445 senders are +/// always a fresh per-event key, and a zero-nonce marker body peels to a clean +/// `PeelerError::DecryptFailed`, the `IngestOutcome::Stale { PeelFailed }` shape +/// the epoch-stall detector counts. +async fn publish_garbage_group_message_at( + relay_url: &str, + nostr_group_id_hex: &str, + created_at: u64, + marker: &str, +) { + let mut envelope = vec![0u8; 12]; + envelope.extend_from_slice(format!("backfill-seam-probe:{marker}").as_bytes()); + assert!(envelope.len() >= NOSTR_GROUP_CONTENT_MIN_LEN); + let ephemeral = Keys::generate(); + let signed = EventBuilder::new(Kind::MlsGroupMessage, BASE64_STANDARD.encode(envelope)) + .tags([Tag::custom( + TagKind::SingleLetter(SingleLetterTag::lowercase(Alphabet::H)), + [nostr_group_id_hex.to_owned()], + )]) + .custom_created_at(NostrTimestamp::from_secs(created_at)) + .sign_with_keys(&ephemeral) + .expect("sign ephemeral kind-445 test event"); + let transport_event = + NostrTransportEvent::from_nostr_event(&signed).expect("dto from signed event"); + let relay_client = NostrSdkRelayClient::new(NostrSdkClient::builder().build()); + relay_client + .publish_event( + &[cgka_traits::TransportEndpoint(relay_url.to_owned())], + &transport_event, + 1, + ) + .await + .expect("publish garbage kind-445 test event"); +} + +/// A live delivery that arms the epoch-gap backfill but yields no visible +/// summary content must still make `next_event` return, so the worker reaches +/// its `run_pending_epoch_backfill` seam. +#[tokio::test] +async fn next_event_returns_when_a_live_delivery_arms_the_epoch_backfill() { + let (_relay, url) = mock_relay().await; + + // --- bob: the stalled account whose `next_event` seam we pin --- + let dir_bob = tempfile::tempdir().unwrap(); + let home_bob = AccountHome::open(dir_bob.path()); + home_bob.create_account("bob").unwrap(); + let bob_id = home_bob.account("bob").unwrap().account_id_hex; + let app_bob = open_store(&dir_bob, &url); + { + let mut bob_setup = app_bob.client("bob").await.unwrap(); + bob_setup.publish_key_package().await.unwrap(); + } + + // --- alice: separate store, only creates the group so bob has a live + // group subscription to receive undecryptable traffic on --- + let dir_alice = tempfile::tempdir().unwrap(); + let home_alice = AccountHome::open(dir_alice.path()); + home_alice.create_account("alice").unwrap(); + let app_alice = open_store(&dir_alice, &url); + let mut alice_client = app_alice.client("alice").await.unwrap(); + let group_id = alice_client + .create_group("epoch-gap backfill next_event seam", &[bob_id.as_str()]) + .await + .unwrap(); + let group_id_hex = hex::encode(group_id.as_slice()); + + // bob joins by draining the welcome; poll `sync` until the group is live so + // its route is subscribed and later garbage is delivered on it. + let mut bob = app_bob.client("bob").await.unwrap(); + let deadline = Instant::now() + JOIN_DEADLINE; + loop { + bob.sync().await.unwrap(); + if app_bob.group("bob", &group_id_hex).unwrap().is_some() { + break; + } + assert!( + Instant::now() < deadline, + "bob did not join the group within the deadline", + ); + sleep(Duration::from_millis(50)).await; + } + + let nostr_group_id_hex = app_bob + .group("bob", &group_id_hex) + .unwrap() + .expect("bob's group projection") + .nostr_routing + .nostr_group_id_hex; + + // Arm the detector with exactly `BACKFILL_THRESHOLD` distinct undecryptable + // messages at bob's live epoch. Present-dated so they clear any since floor + // and reach bob's live subscription; distinct markers make distinct event + // ids, hence distinct undecryptables at the one stalled epoch. None produces + // a visible `MarmotAppEvent`, so the summary stays empty across all of them + // and only the armed backfill can force the return. + let created_at = test_unix_now_seconds(); + for arm in 0..BACKFILL_THRESHOLD { + publish_garbage_group_message_at( + &url, + &nostr_group_id_hex, + created_at, + &format!("arm-{arm}"), + ) + .await; + } + + // The threshold-crossing delivery arms the backfill with an empty summary. + // With the guard fix `next_event` returns here; without it the call blocks + // on the next `receive()` and this timeout is the RED failure. + let summary = timeout(NEXT_EVENT_DEADLINE, bob.next_event()) + .await + .expect("next_event must return once a live delivery arms the epoch backfill") + .expect("next_event should not error"); + + // The return is driven purely by the armed backfill: the undecryptable + // probes carry no visible content, so an empty summary is the expected and + // harmless payload the worker forwards to `publish_app_runtime_summary`. + assert!( + summary.joined_groups.is_empty() + && summary.messages.is_empty() + && summary.events.is_empty(), + "the arming deliveries are undecryptable and must not surface visible \ + summary content; the return is forced by the armed backfill alone", + ); +}