diff --git a/crates/marmot-app/src/client/audit.rs b/crates/marmot-app/src/client/audit.rs index fb74af25..a7032658 100644 --- a/crates/marmot-app/src/client/audit.rs +++ b/crates/marmot-app/src/client/audit.rs @@ -189,6 +189,21 @@ impl AppClient { ); } + /// Record an `epoch_stall_backfill_armed` forensic audit row at the arm + /// decision: the epoch the group was stalled at (`stalled_epoch`) and the + /// distinct-undecryptable threshold that armed the backfill (`threshold`). + /// Group-scoped, so `group_ref` carries the stalled group id. + pub(crate) fn record_epoch_stall_backfill_armed(&self, group_id: &GroupId, stalled_epoch: u64) { + self.runtime.session().record_audit_event( + Some(group_id), + None, + AuditEventKind::EpochStallBackfillArmed { + stalled_epoch, + threshold: self.epoch_stall.threshold() as u64, + }, + ); + } + /// Apply the audit-logging switch to this live session by swapping the /// recorder in place: a file-backed recorder when `enabled`, or a no-op /// recorder when off. Dropping the prior recorder flushes and closes any diff --git a/crates/marmot-app/src/client/epoch_stall.rs b/crates/marmot-app/src/client/epoch_stall.rs index fb4e89d7..9715b6a1 100644 --- a/crates/marmot-app/src/client/epoch_stall.rs +++ b/crates/marmot-app/src/client/epoch_stall.rs @@ -64,6 +64,14 @@ impl EpochStallDetector { } } + /// The distinct-undecryptable count at which this detector arms a backfill. + /// Reported on the `epoch_stall_backfill_armed` audit row so the row is + /// honest even when the detector was built with a non-default threshold + /// (unit tests, or a future configurable value). + pub(crate) fn threshold(&self) -> usize { + self.threshold + } + /// 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 diff --git a/crates/marmot-app/src/client/sync.rs b/crates/marmot-app/src/client/sync.rs index 57499338..7104cb1b 100644 --- a/crates/marmot-app/src/client/sync.rs +++ b/crates/marmot-app/src/client/sync.rs @@ -337,14 +337,27 @@ impl AppClient { 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) - { + if self.epoch_stall.observe_undecryptable( + group_id.clone(), + message_id_hex.to_owned(), + record.epoch, + ) { + // Record the arm decision before the replay side effect runs (the + // worker seam calls run_pending_epoch_backfill after this returns). + // Best-effort, fire-and-forget: recording can never block or fail + // the backfill. + self.record_epoch_stall_backfill_armed(&group_id, record.epoch.0); self.epoch_backfill_pending = true; } } + /// Whether an epoch-gap backfill is armed and awaiting its replay. Read by + /// the account worker to schedule a forensic audit-tracker upload for the + /// just-recorded `epoch_stall_backfill_armed` row without poking the field. + pub(crate) fn has_pending_epoch_backfill(&self) -> bool { + self.epoch_backfill_pending + } + /// 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 diff --git a/crates/marmot-app/src/runtime/account_worker.rs b/crates/marmot-app/src/runtime/account_worker.rs index abb99170..ff1603ed 100644 --- a/crates/marmot-app/src/runtime/account_worker.rs +++ b/crates/marmot-app/src/runtime/account_worker.rs @@ -449,14 +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), - ); - } + run_pending_epoch_backfill_reporting_arm( + &mut client, + &events, + &account_id_hex, + &account_label, + &shared, + ) + .await; if sync_summary_triggers_audit_tracker_update(&summary) { shared.schedule_audit_log_tracker_update("startup_sync"); } @@ -582,14 +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), - ); - } + run_pending_epoch_backfill_reporting_arm( + &mut client, + &events, + &account_id_hex, + &account_label, + &shared, + ) + .await; if sync_summary_triggers_audit_tracker_update(&summary) { shared.schedule_audit_log_tracker_update("receive"); } @@ -663,6 +663,14 @@ async fn handle_account_worker_command( let result = match client.sync().await { Ok(summary) => { publish_app_runtime_summary(events, account_id_hex, account_label, &summary); + // Catch-up arms without running the backfill (the next + // receive pass runs it), but the arm row is already durable + // and the arming traffic never trips the summary gate below + // — push it to the tracker from the seam that observed the + // arm rather than a hoped-for later delivery. + if client.has_pending_epoch_backfill() { + shared.schedule_audit_log_tracker_update("epoch_backfill_armed"); + } if sync_summary_triggers_audit_tracker_update(&summary) { shared.schedule_audit_log_tracker_update("catch_up"); } @@ -1425,6 +1433,36 @@ fn sync_summary_triggers_audit_tracker_update(summary: &SyncSummary) -> bool { !summary.joined_groups.is_empty() || !summary.messages.is_empty() || !summary.events.is_empty() } +/// Run any pending epoch-gap backfill and push its arm evidence to the audit +/// tracker. The arm state is captured *before* the replay drains it, and the +/// tracker is scheduled unconditionally on the replay outcome: the +/// `epoch_stall_backfill_armed` row is already durable, a failing replay is the +/// highest-value upload, and the arming pass returns an empty summary that +/// never trips the visible-activity gate. Shared by the startup and receive +/// seams so the capture-before-run ordering cannot drift between them; the +/// catch-up seam schedules the upload without running the backfill and stays +/// separate. +async fn run_pending_epoch_backfill_reporting_arm( + client: &mut AppClient, + events: &broadcast::Sender, + account_id_hex: &str, + account_label: &str, + shared: &RuntimeSharedServices, +) { + let backfill_armed = client.has_pending_epoch_backfill(); + 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 backfill_armed { + shared.schedule_audit_log_tracker_update("epoch_backfill_armed"); + } +} + fn publish_app_runtime_summary( events: &broadcast::Sender, account_id_hex: &str, diff --git a/crates/marmot-app/tests/epoch_stall_backfill_audit.rs b/crates/marmot-app/tests/epoch_stall_backfill_audit.rs new file mode 100644 index 00000000..69c614f4 --- /dev/null +++ b/crates/marmot-app/tests/epoch_stall_backfill_audit.rs @@ -0,0 +1,409 @@ +//! Forensic-audit coverage for the epoch-stall backfill arm decision. +//! +//! PR #892 ships an epoch-gap backfill: once a group accumulates +//! `EPOCH_STALL_BACKFILL_THRESHOLD` distinct undecryptable messages at one +//! stalled epoch, the runtime arms a full-history transport replay to recover +//! the missed commit. The threshold is an empirical, single-cohort estimate the +//! detector's own docstring flags as provisional. To firm it up against more +//! cohorts a field export must reveal when and why a backfill fired — so the arm +//! decision emits one `epoch_stall_backfill_armed` forensic audit row, carrying +//! the stalled epoch and the threshold in force. These tests pin that row +//! through the account's forensic audit JSONL, the sanctioned channel for this +//! group-scoped evidence. +//! +//! The arm is driven through `AppClient::next_event` exactly as +//! `next_event_backfill.rs` does — the deterministic seam that avoids the full +//! runtime. Harness helpers (`mock_relay`, `open_store`, `test_unix_now_seconds`, +//! `publish_garbage_group_message_at`) are copied from `next_event_backfill.rs` +//! and the audit-JSONL readers (`AuditRowTracker`, `rows_of_kind`) from +//! `cursor_persistence.rs`, following the established test convention of +//! duplicating these helpers rather than sharing them. + +use std::collections::HashMap; +use std::time::{Duration, Instant}; + +use marmot_account::AccountHome; +use marmot_app::{AuditLogSettings, 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 arm it returns as soon as the threshold-crossing delivery +/// fires; the bound guards against a permanent hang. +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 `next_event_backfill.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-audit-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"); +} + +/// Incremental reader over an account's forensic audit JSONL files: each call +/// returns only the rows appended since the previous call, so a burst can be +/// asserted in isolation. Copied from `cursor_persistence.rs`. +#[derive(Default)] +struct AuditRowTracker { + consumed_lines: HashMap, +} + +impl AuditRowTracker { + fn new_rows(&mut self, app: &MarmotApp, account_ref: &str) -> Vec { + let mut files = app.audit_log_files().unwrap(); + files.sort_by(|a, b| a.path.cmp(&b.path)); + let mut rows = Vec::new(); + for file in files.iter().filter(|file| file.account_ref == account_ref) { + let text = std::fs::read_to_string(&file.path).unwrap(); + let lines = text.lines().collect::>(); + let seen = self.consumed_lines.entry(file.path.clone()).or_insert(0); + for line in &lines[*seen..] { + rows.push(serde_json::from_str::(line).unwrap()); + } + *seen = lines.len(); + } + rows + } +} + +fn rows_of_kind<'rows>( + rows: &'rows [serde_json::Value], + kind: &str, +) -> Vec<&'rows serde_json::Value> { + rows.iter() + .filter(|row| row["kind"]["type"] == kind) + .collect() +} + +/// Sum the `deliveries` counter across every `sync_drain` row: positive +/// evidence that a drain window actually ingested traffic. The absence tests +/// assert this against their probe count first, so a too-short drain window +/// fails loudly instead of letting the no-armed-row assertion pass vacuously. +fn total_sync_drain_deliveries(rows: &[serde_json::Value]) -> u64 { + rows_of_kind(rows, "sync_drain") + .iter() + .filter_map(|row| row["kind"]["deliveries"].as_u64()) + .sum() +} + +/// Set up bob (the stalled account under test, own store) joined to a group +/// alice created (separate store), with forensic audit recording enabled. +/// Returns the mock relay (kept bound by the caller so it outlives the client), +/// bob's app handle and live client, the relay URL, the group id, and bob's +/// `nostr_group_id` route so the caller can spray undecryptable traffic at it. +async fn stalled_bob_in_a_live_group( + dir_bob: &tempfile::TempDir, + dir_alice: &tempfile::TempDir, +) -> ( + MockRelay, + MarmotApp, + marmot_app::AppClient, + String, + cgka_traits::GroupId, + String, +) { + let (relay, url) = mock_relay().await; + + // --- bob: the stalled account whose arm decision we pin (own store) --- + 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); + // Enable forensic audit recording before any client opens; the setting + // persists in the store so every client this account builds records. + app_bob + .set_audit_log_settings(AuditLogSettings { + enabled: true, + ..Default::default() + }) + .unwrap(); + { + 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 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-stall backfill audit", &[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; + + (relay, app_bob, bob, url, group_id, nostr_group_id_hex) +} + +/// Arming a backfill records exactly one `epoch_stall_backfill_armed` row, +/// group-scoped and carrying the stalled epoch and the threshold in force. +#[tokio::test] +async fn arming_a_backfill_records_one_epoch_stall_backfill_armed_row() { + let dir_bob = tempfile::tempdir().unwrap(); + let dir_alice = tempfile::tempdir().unwrap(); + let (_relay, app_bob, mut bob, url, group_id, nostr_group_id_hex) = + stalled_bob_in_a_live_group(&dir_bob, &dir_alice).await; + + let stalled_epoch = bob.group_mls_state(&group_id).unwrap().epoch; + + // 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. + 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; `next_event` returns. + 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"); + + let mut audit_rows = AuditRowTracker::default(); + let rows = audit_rows.new_rows(&app_bob, "bob"); + let armed = rows_of_kind(&rows, "epoch_stall_backfill_armed"); + assert_eq!( + armed.len(), + 1, + "arming a backfill must record exactly one epoch_stall_backfill_armed row: {rows:?}" + ); + let row = armed[0]; + assert_eq!( + row["kind"]["stalled_epoch"].as_u64(), + Some(stalled_epoch), + "the row must carry the epoch bob was stalled at: {row}" + ); + assert_eq!( + row["kind"]["threshold"].as_u64(), + Some(BACKFILL_THRESHOLD as u64), + "the row must carry the threshold in force: {row}" + ); + assert_eq!( + row["group_ref"].as_str(), + Some(hex::encode(group_id.as_slice()).as_str()), + "the row is group-scoped via group_ref, not a duplicated field: {row}" + ); +} + +/// Below-threshold undecryptable traffic never crosses the arm decision, so it +/// records no `epoch_stall_backfill_armed` row. Guards the +/// `observe_undecryptable == false` path. +#[tokio::test] +async fn below_threshold_undecryptable_traffic_records_no_backfill_armed_row() { + let dir_bob = tempfile::tempdir().unwrap(); + let dir_alice = tempfile::tempdir().unwrap(); + let (_relay, app_bob, mut bob, url, _group_id, nostr_group_id_hex) = + stalled_bob_in_a_live_group(&dir_bob, &dir_alice).await; + + // Consume the setup rows (join-phase drains) so the delivery evidence + // below counts the probes alone, not the welcome traffic. + let mut audit_rows = AuditRowTracker::default(); + let _setup_rows = audit_rows.new_rows(&app_bob, "bob"); + + // One below the threshold: enough to accumulate a stall run but never arm. + let created_at = test_unix_now_seconds(); + for probe in 0..(BACKFILL_THRESHOLD - 1) { + publish_garbage_group_message_at( + &url, + &nostr_group_id_hex, + created_at, + &format!("below-{probe}"), + ) + .await; + } + + // The probes arm nothing, so `next_event` would block — drain with `sync`, + // which returns after each drain window. Several passes give the mock relay + // ample time to deliver all of the pre-published probes. + for _ in 0..5 { + bob.sync().await.unwrap(); + sleep(Duration::from_millis(50)).await; + } + + let rows = audit_rows.new_rows(&app_bob, "bob"); + // Positive delivery evidence first: the drains must have ingested every + // probe, otherwise the absence assertion below would hold vacuously. + assert!( + total_sync_drain_deliveries(&rows) >= (BACKFILL_THRESHOLD - 1) as u64, + "the drain windows must have delivered all {} below-threshold probes: {rows:?}", + BACKFILL_THRESHOLD - 1 + ); + assert!( + rows_of_kind(&rows, "epoch_stall_backfill_armed").is_empty(), + "below-threshold traffic must not arm a backfill: {rows:?}" + ); +} + +/// A second burst of distinct undecryptable traffic at the *same* stalled epoch +/// records no further row: the row inherits the detector's once-per-stalled-epoch +/// debounce, since the recorder is only reached when `observe_undecryptable` +/// returns `true`. +#[tokio::test] +async fn a_second_stall_burst_at_the_same_epoch_records_no_further_row() { + let dir_bob = tempfile::tempdir().unwrap(); + let dir_alice = tempfile::tempdir().unwrap(); + let (_relay, app_bob, mut bob, url, group_id, nostr_group_id_hex) = + stalled_bob_in_a_live_group(&dir_bob, &dir_alice).await; + + let stalled_epoch = bob.group_mls_state(&group_id).unwrap().epoch; + let mut audit_rows = AuditRowTracker::default(); + + // First burst arms the backfill (row 1). + 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; + } + timeout(NEXT_EVENT_DEADLINE, bob.next_event()) + .await + .expect("next_event must return once the first burst arms the backfill") + .expect("next_event should not error"); + let arm_rows = audit_rows.new_rows(&app_bob, "bob"); + assert_eq!( + rows_of_kind(&arm_rows, "epoch_stall_backfill_armed").len(), + 1, + "the first burst must have armed and recorded exactly one row: {arm_rows:?}" + ); + + // A second burst of *distinct* undecryptables at the same stalled epoch: the + // group never advanced (undecryptable traffic does not move the epoch), so + // the detector's debounce holds and no further row may be recorded. These + // arm nothing, so drain with `sync` rather than the blocking `next_event`. + assert_eq!( + bob.group_mls_state(&group_id).unwrap().epoch, + stalled_epoch, + "sanity: undecryptable traffic must not have advanced bob's epoch", + ); + for again in 0..BACKFILL_THRESHOLD { + publish_garbage_group_message_at( + &url, + &nostr_group_id_hex, + created_at, + &format!("again-{again}"), + ) + .await; + } + for _ in 0..5 { + bob.sync().await.unwrap(); + sleep(Duration::from_millis(50)).await; + } + + let post_arm_rows = audit_rows.new_rows(&app_bob, "bob"); + // Positive delivery evidence first: the post-arm drains must have ingested + // the whole second burst, otherwise the debounce assertion below would + // hold vacuously. + assert!( + total_sync_drain_deliveries(&post_arm_rows) >= BACKFILL_THRESHOLD as u64, + "the drain windows must have delivered all {BACKFILL_THRESHOLD} second-burst probes: \ + {post_arm_rows:?}" + ); + assert!( + rows_of_kind(&post_arm_rows, "epoch_stall_backfill_armed").is_empty(), + "a second burst at the same stalled epoch must not record a further row: {post_arm_rows:?}" + ); +} diff --git a/crates/marmot-app/tests/relay_runtime.rs b/crates/marmot-app/tests/relay_runtime.rs index 9d290ac2..0b13438c 100644 --- a/crates/marmot-app/tests/relay_runtime.rs +++ b/crates/marmot-app/tests/relay_runtime.rs @@ -22,14 +22,17 @@ use marmot_app::{ use nostr::base64::Engine as _; use nostr::base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use nostr_relay_builder::MockRelay; -use nostr_sdk::prelude::Client as NostrSdkClient; +use nostr_sdk::prelude::{ + Alphabet, Client as NostrSdkClient, EventBuilder, Keys, Kind, SingleLetterTag, Tag, TagKind, + Timestamp as NostrTimestamp, +}; use sha2::{Digest, Sha256}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::{Mutex, oneshot}; -use tokio::time::{Duration, sleep, timeout}; +use tokio::sync::{Mutex, mpsc, oneshot}; +use tokio::time::{Duration, Instant, sleep, timeout}; use transport_nostr_adapter::{KIND_MARMOT_KEY_PACKAGE, NostrRelayClient, NostrSdkRelayClient}; -use transport_nostr_peeler::NostrTransportEvent; +use transport_nostr_peeler::{NOSTR_GROUP_CONTENT_MIN_LEN, NostrTransportEvent}; const AUDIT_TRACKER_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); const AUDIT_TRACKER_NON_BLOCKING_TIMEOUT: Duration = Duration::from_secs(5); @@ -125,6 +128,28 @@ async fn capture_delayed_audit_upload_with_overlap_probe( } } +/// Accept audit uploads in a loop, forwarding each request body over `bodies` +/// and answering 204 immediately. Unlike the gated `capture_delayed_*` helpers, +/// this drains a stream of uploads so a test can wait for the *one* whose body +/// carries a specific forensic row while ignoring earlier unrelated uploads. +async fn forward_audit_upload_bodies( + listener: TcpListener, + bodies: mpsc::UnboundedSender>, +) { + loop { + let Ok((mut stream, _peer)) = listener.accept().await else { + return; + }; + let Some(captured) = read_captured_audit_upload(&mut stream).await else { + continue; + }; + write_http_response(&mut stream, 204, "text/plain", b"").await; + if bodies.send(captured.body).is_err() { + return; + } + } +} + async fn read_captured_audit_upload(stream: &mut TcpStream) -> Option { let mut request = Vec::new(); let mut buffer = [0_u8; 4096]; @@ -331,6 +356,40 @@ async fn publish_nostr_event_at( .unwrap(); } +/// Publish an envelope-shaped undecryptable kind-445 probe with a fresh +/// ephemeral key, h-tagged to `nostr_group_id_hex`. Mirrors the probe publisher +/// in `next_event_backfill.rs`: real kind-445 senders always sign with a fresh +/// per-event key, and a zero-nonce marker body peels to a clean +/// `PeelFailed` — the `IngestOutcome::Stale` shape the epoch-stall detector +/// counts toward arming a backfill. Distinct `marker`s yield distinct event ids, +/// hence distinct undecryptables at the group's one stalled epoch. +async fn publish_garbage_group_message( + 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-armed-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(&[endpoint(relay_url)], &transport_event, 1) + .await + .expect("publish garbage kind-445 test event"); +} + async fn publish_account_relay_lists_at( home: &AccountHome, label: &str, @@ -1434,6 +1493,116 @@ async fn app_runtime_schedules_audit_tracker_update_after_inbound_welcome() { runtime.shutdown().await; } +/// The epoch-stall backfill arms on a run of undecryptable traffic that carries +/// NO visible `SyncSummary` content, so the summary-gated audit-tracker schedule +/// never fires for it. The arm still records an `epoch_stall_backfill_armed` +/// forensic row, and the runtime worker must push it to the field tracker — a +/// passively-stalled account whose only traffic is undecryptable is exactly the +/// case the field-evidence loop needs to observe. +#[tokio::test] +async fn app_runtime_uploads_armed_backfill_row_without_visible_activity() { + // Mirrors `EPOCH_STALL_BACKFILL_THRESHOLD` (crate-private): the distinct + // undecryptable messages at one stalled epoch that arm a backfill. + const BACKFILL_THRESHOLD: usize = 8; + + let dir = tempfile::tempdir().unwrap(); + let home = AccountHome::open(dir.path()); + home.create_account("alice").unwrap(); + home.create_account("bob").unwrap(); + let bob_id = home.account("bob").unwrap().account_id_hex; + + let (_relay, app, url) = mock_app(&dir).await; + app.set_audit_log_settings(AuditLogSettings { + enabled: true, + ..Default::default() + }) + .unwrap(); + let mut bob_setup = app.client("bob").await.unwrap(); + bob_setup.publish_key_package().await.unwrap(); + drop(bob_setup); + + let runtime = MarmotAppRuntime::new(app.clone()); + let mut events = runtime.subscribe(); + let (body_tx, mut body_rx) = mpsc::unbounded_channel(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(forward_audit_upload_bodies(listener, body_tx)); + runtime + .set_audit_log_tracker_config(AuditLogTrackerConfig { + endpoint: Some(format!("http://{addr}/api/v1/audit-logs/")), + authorization_bearer_token: Some("goggles_backfill_secret".to_owned()), + source: AuditLogUploadSource::default(), + }) + .unwrap(); + runtime.start().await.unwrap(); + + // bob's managed worker joins the group so it holds a live subscription on + // which the undecryptable probes below are delivered. + let mut alice = app.client("alice").await.unwrap(); + let group_id = alice + .create_group("runtime epoch backfill arm", &["bob"]) + .await + .unwrap(); + wait_for_event(&mut events, |event| { + matches!( + event, + MarmotAppEvent::GroupJoined { account_id_hex, group_id: joined_group, .. } + if account_id_hex == &bob_id && joined_group == &group_id + ) + }) + .await; + + let group_id_hex = hex::encode(group_id.as_slice()); + let nostr_group_id_hex = app + .group("bob", &group_id_hex) + .unwrap() + .expect("bob's group projection") + .nostr_routing + .nostr_group_id_hex; + + // Arm bob's detector with exactly the threshold of distinct undecryptable + // messages at his live epoch. Present-dated so they clear any since floor; + // none yields visible summary content, so the summary-gated schedule stays + // silent and only the armed-backfill schedule can deliver these rows. + let created_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + for arm in 0..BACKFILL_THRESHOLD { + publish_garbage_group_message(&url, &nostr_group_id_hex, created_at, &format!("arm-{arm}")) + .await; + } + + // Wait, across uploads, for the one carrying the armed row. Earlier uploads + // (e.g. the welcome-join schedule) are drained and ignored. + let deadline = Instant::now() + Duration::from_secs(20); + let mut carried_armed_row = false; + while Instant::now() < deadline { + match timeout( + deadline.saturating_duration_since(Instant::now()), + body_rx.recv(), + ) + .await + { + Ok(Some(body)) => { + if String::from_utf8_lossy(&body).contains("epoch_stall_backfill_armed") { + carried_armed_row = true; + break; + } + } + Ok(None) | Err(_) => break, + } + } + assert!( + carried_armed_row, + "the runtime must push the epoch_stall_backfill_armed row to the audit \ + tracker even though the arming traffic produced no visible activity", + ); + + server.abort(); + runtime.shutdown().await; +} + #[tokio::test] async fn app_runtime_coalesces_audit_tracker_updates_while_upload_is_in_flight() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/marmot-forensics/schema/audit-log-event.v2.schema.json b/crates/marmot-forensics/schema/audit-log-event.v2.schema.json index c418c2a5..4924569c 100644 --- a/crates/marmot-forensics/schema/audit-log-event.v2.schema.json +++ b/crates/marmot-forensics/schema/audit-log-event.v2.schema.json @@ -1639,6 +1639,22 @@ "$ref": "#/$defs/u64" } } + }, + { + "type": "object", + "additionalProperties": false, + "required": ["type", "stalled_epoch", "threshold"], + "properties": { + "type": { + "const": "epoch_stall_backfill_armed" + }, + "stalled_epoch": { + "$ref": "#/$defs/u64" + }, + "threshold": { + "$ref": "#/$defs/u64" + } + } } ] } diff --git a/crates/marmot-forensics/src/audit.rs b/crates/marmot-forensics/src/audit.rs index 5fbce35e..9264b350 100644 --- a/crates/marmot-forensics/src/audit.rs +++ b/crates/marmot-forensics/src/audit.rs @@ -1002,6 +1002,25 @@ pub enum AuditEventKind { #[serde(default, skip_serializing_if = "Option::is_none")] cursor_after_secs: Option, }, + /// A group crossed the epoch-stall backfill threshold — enough distinct + /// undecryptable messages at one stalled epoch — and armed a full-history + /// epoch-gap backfill (commit-loss recovery). Emitted once per (group, + /// stalled epoch) at the arm decision, *before* the replay side effect runs, + /// so a field export reveals when and why full-history replays fire — the + /// evidence loop for tuning the empirical backfill threshold. + /// + /// Group-scoped: the stalled group's id is on the enclosing + /// [`AuditEvent::group_ref`], exactly as the `human_action` group rows carry + /// it; it is deliberately not duplicated into a field here. `stalled_epoch` + /// is the group epoch the device was stuck at when it armed — correlate it + /// against the group's live epoch (visible on `group_context` / `epoch_*` + /// rows) to read the size of the gap that triggered recovery. `threshold` is + /// the distinct-undecryptable count that armed the backfill, carried so an + /// export is self-describing when the constant is retuned across builds. + /// + /// Privacy: two scalar counts only — no ids, relay URLs, message ids, or + /// payloads — so nothing here needs scrubbing in either [`AuditDataMode`]. + EpochStallBackfillArmed { stalled_epoch: u64, threshold: u64 }, } impl AuditEventKind { @@ -1052,6 +1071,7 @@ impl AuditEventKind { AuditEventKind::Rejection { .. } => "rejection", AuditEventKind::SubscriptionRebuild { .. } => "subscription_rebuild", AuditEventKind::SyncDrain { .. } => "sync_drain", + AuditEventKind::EpochStallBackfillArmed { .. } => "epoch_stall_backfill_armed", } } } diff --git a/crates/marmot-forensics/src/audit/tests.rs b/crates/marmot-forensics/src/audit/tests.rs index c8811a66..21546d04 100644 --- a/crates/marmot-forensics/src/audit/tests.rs +++ b/crates/marmot-forensics/src/audit/tests.rs @@ -394,6 +394,33 @@ fn sync_drain_round_trips_through_serde() { ); } +#[test] +fn epoch_stall_backfill_armed_roundtrips_and_carries_its_fields() { + let kind = AuditEventKind::EpochStallBackfillArmed { + stalled_epoch: 19, + threshold: 8, + }; + let event = AuditEvent { + schema_version: AUDIT_LOG_SCHEMA_VERSION.into(), + seq: 7, + wall_time_ms: 1_700_000_000_000, + audit_data_mode: AuditDataMode::ObfuscatedSensitiveData, + recorder_session_id: Some("recorder-1".into()), + account_ref: None, + engine_id: "engine-xyz".into(), + group_ref: None, + context: None, + kind: kind.clone(), + }; + let json = serde_json::to_string(&event).unwrap(); + let value: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(value["kind"]["type"], "epoch_stall_backfill_armed"); + assert_eq!(value["kind"]["stalled_epoch"], 19); + assert_eq!(value["kind"]["threshold"], 8); + let parsed: AuditEvent = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed.kind, kind); +} + fn sample_audit_event_kinds() -> Vec { vec![ AuditEventKind::RecorderStarted { @@ -804,6 +831,10 @@ fn sample_audit_event_kinds() -> Vec { cursor_before_secs: Some(1_699_999_880), cursor_after_secs: Some(1_700_000_000), }, + AuditEventKind::EpochStallBackfillArmed { + stalled_epoch: 19, + threshold: 8, + }, ] }