From 7dd6659601070579e5a3010df0f38308f2fc761c Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Thu, 16 Jul 2026 16:49:44 -0400 Subject: [PATCH 1/5] fix(cgka-engine): classify malformed group messages as terminal stale instead of aborting ingest --- .../src/message_processor/ingest.rs | 51 ++++- .../cgka-engine/src/message_processor/mod.rs | 8 +- crates/cgka-engine/tests/ingest.rs | 194 +++++++++++++++++ .../marmot-app/tests/malformed_445_drain.rs | 204 ++++++++++++++++++ 4 files changed, 444 insertions(+), 13 deletions(-) create mode 100644 crates/marmot-app/tests/malformed_445_drain.rs diff --git a/crates/cgka-engine/src/message_processor/ingest.rs b/crates/cgka-engine/src/message_processor/ingest.rs index 16286fe9..bce61ea6 100644 --- a/crates/cgka-engine/src/message_processor/ingest.rs +++ b/crates/cgka-engine/src/message_processor/ingest.rs @@ -357,7 +357,7 @@ impl Engine { // (if this came via the retry lifecycle) so it stops // holding a per-group cap slot (mdk#339); a no-op on // the direct path where no deferred row exists. - self.mark_raw_transport_message_failed_if_deferred( + self.mark_raw_transport_message_failed_if_awaiting_retry( &raw_msg_id, "stale_epoch_no_snapshot", )?; @@ -380,6 +380,27 @@ impl Engine { }); } } + Err(PeelerError::Malformed(_)) => { + // Structurally-invalid content is ordinary hostile input: + // anyone can publish to the cleartext routing tag without + // group membership, and no epoch context can ever peel it. + // Terminal — propagating an error here would abort the + // caller's whole transport drain on one garbage event, + // starving every message queued behind it. Dropped + // unpersisted for the same reason as the mdk#339 flood + // cap: transport ids are attacker-controllable, so durable + // rows keyed by them would grow without bound. Retire the + // raw deferred row if this arrived via the retry + // lifecycle; a no-op on the direct path. + self.mark_raw_transport_message_failed_if_awaiting_retry( + &raw_msg_id, + "malformed_payload", + )?; + self.seen_message_ids.insert(msg.id.clone()); + return Ok(IngestOutcome::Stale { + reason: StaleReason::PeelFailed, + }); + } Err(e) => return Err(EngineError::Peeler(e)), }; let mls_bytes = match peeled.content { @@ -388,7 +409,7 @@ impl Engine { // A group-message envelope that peels to a Welcome is // malformed: terminal. Retire the raw deferred row first // so its cap slot is released (mdk#339). - self.mark_raw_transport_message_failed_if_deferred( + self.mark_raw_transport_message_failed_if_awaiting_retry( &raw_msg_id, "peeled_welcome_in_group_message", )?; @@ -461,7 +482,7 @@ impl Engine { // Peeled successfully but the MLS body is neither a // public nor private message: terminal. Retire the raw // deferred row so it leaves the retry lifecycle (mdk#339). - self.mark_raw_transport_message_failed_if_deferred( + self.mark_raw_transport_message_failed_if_awaiting_retry( &raw_msg_id, "non_mls_message_body", )?; @@ -498,7 +519,7 @@ impl Engine { tag, ), ); - self.mark_raw_transport_message_failed_if_deferred(&raw_msg_id, tag)?; + self.mark_raw_transport_message_failed_if_awaiting_retry(&raw_msg_id, tag)?; return Ok(IngestOutcome::Stale { reason: StaleReason::PeelFailed, }); @@ -577,7 +598,7 @@ impl Engine { .tag() }; self.update_stored_message_state(&msg.id, MessageState::Failed)?; - self.mark_raw_transport_message_failed_if_deferred(&raw_msg_id, tag)?; + self.mark_raw_transport_message_failed_if_awaiting_retry(&raw_msg_id, tag)?; return Ok(IngestOutcome::Stale { reason: StaleReason::PeelFailed, }); @@ -717,7 +738,7 @@ impl Engine { // Peeled successfully but terminally rejected: retire // the raw deferred row so it leaves the retry // lifecycle instead of holding a cap slot (mdk#339). - self.mark_raw_transport_message_failed_if_deferred( + self.mark_raw_transport_message_failed_if_awaiting_retry( &raw_msg_id, "unattributable_sender", )?; @@ -737,7 +758,7 @@ impl Engine { self.update_stored_message_state(&msg.id, MessageState::Failed)?; // Peeled successfully but terminally rejected: retire // the raw deferred row (mdk#339). - self.mark_raw_transport_message_failed_if_deferred( + self.mark_raw_transport_message_failed_if_awaiting_retry( &raw_msg_id, "invalid_app_payload", )?; @@ -1572,13 +1593,18 @@ impl Engine { Ok(()) } - fn mark_raw_transport_message_failed_if_deferred( + fn mark_raw_transport_message_failed_if_awaiting_retry( &mut self, raw_msg_id: &MessageId, reason: &str, ) -> Result<(), EngineError> { match self.storage.get_message(raw_msg_id) { - Ok(record) if record.state == MessageState::PeelDeferred => { + Ok(record) + if matches!( + record.state, + MessageState::PeelDeferred | MessageState::Retryable + ) => + { self.storage .update_message_state(raw_msg_id, MessageState::Failed)?; self.audit_group( @@ -1591,7 +1617,12 @@ impl Engine { reason, ), ); - self.note_peel_deferred_row_retired(&record.group_id, raw_msg_id); + // Only a `PeelDeferred` row holds a flood-cap slot (mdk#339); + // a `Retryable` row — input buffered pre-peel while the group + // could not ingest — sits outside the cap. + if record.state == MessageState::PeelDeferred { + self.note_peel_deferred_row_retired(&record.group_id, raw_msg_id); + } Ok(()) } Ok(_) | Err(StorageError::NotFound) => Ok(()), diff --git a/crates/cgka-engine/src/message_processor/mod.rs b/crates/cgka-engine/src/message_processor/mod.rs index a29ab52f..bccf3e5e 100644 --- a/crates/cgka-engine/src/message_processor/mod.rs +++ b/crates/cgka-engine/src/message_processor/mod.rs @@ -883,9 +883,11 @@ impl Engine { reason: StaleReason::PeelFailed | StaleReason::Quarantined, }) => { // Still un-peelable, or the group is quarantined: leave the - // row deferred. A terminal-after-peel `PeelFailed` already - // retired its raw deferred row inside `ingest_group_message` - // via `mark_raw_transport_message_failed_if_deferred`. + // row in its retry state. A terminal-after-peel `PeelFailed` + // (malformed, stale epoch with no snapshot) already retired + // its raw row — `PeelDeferred` or `Retryable` alike — inside + // `ingest_group_message` via + // `mark_raw_transport_message_failed_if_awaiting_retry`. } Ok(_) => { // Applied (`Processed`) or terminally reclassified diff --git a/crates/cgka-engine/tests/ingest.rs b/crates/cgka-engine/tests/ingest.rs index 516ff5c8..ad432879 100644 --- a/crates/cgka-engine/tests/ingest.rs +++ b/crates/cgka-engine/tests/ingest.rs @@ -55,6 +55,11 @@ struct RecordingPeeler { struct FailOncePeeler { failed: Mutex>, } +/// Structural minimum mirroring the production Nostr peeler's content-length +/// gate: a payload too short to carry a nonce-prefixed ciphertext classifies +/// as `Malformed` before any decryption attempt. +const STRUCTURAL_MIN_PAYLOAD_LEN: usize = 28; +struct MalformedShortPayloadPeeler; impl FailOncePeeler { fn new() -> Self { @@ -222,6 +227,42 @@ impl TransportPeeler for FailOncePeeler { } } +#[async_trait] +impl TransportPeeler for MalformedShortPayloadPeeler { + async fn peel_group_message( + &self, + msg: &TransportMessage, + ctx: &GroupContextSnapshot, + ) -> Result { + if msg.payload.len() < STRUCTURAL_MIN_PAYLOAD_LEN { + return Err(PeelerError::Malformed( + "content too short for nonce-prefixed ciphertext".into(), + )); + } + MockPeeler.peel_group_message(msg, ctx).await + } + + async fn peel_welcome(&self, msg: &TransportMessage) -> Result { + MockPeeler.peel_welcome(msg).await + } + + async fn wrap_group_message( + &self, + payload: &EncryptedPayload, + ctx: &GroupContextSnapshot, + ) -> Result { + MockPeeler.wrap_group_message(payload, ctx).await + } + + async fn wrap_welcome( + &self, + payload: &EncryptedPayload, + recipient: &MemberId, + ) -> Result { + MockPeeler.wrap_welcome(payload, recipient).await + } +} + fn selfremove_registry() -> FeatureRegistry { let mut r = FeatureRegistry::new(); r.register( @@ -476,6 +517,159 @@ async fn peel_deferred_message_retries_instead_of_short_circuiting() { ); } +#[tokio::test] +async fn malformed_group_message_is_stale_and_does_not_wedge_ingest() { + let mut alice = build_client(b"alice"); + let mut bob = build_client_with_peeler(b"bob", Box::new(MalformedShortPayloadPeeler)); + let bob_kp = bob.fresh_key_package().await.unwrap(); + + let (group_id, result) = alice + .create_group(CreateGroupRequest { + name: "".into(), + description: "".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![], + }) + .await + .unwrap(); + let (pending, bob_welcome) = match result { + SendResult::GroupCreated { + pending, + mut welcomes, + } => (pending, welcomes.remove(0)), + _ => unreachable!(), + }; + alice.confirm_published(pending).await.unwrap(); + bob.join_welcome(bob_welcome).await.unwrap(); + + // Anyone can publish to a group's cleartext routing tag without being a + // member, so structurally-invalid content is ordinary hostile input. It + // must classify as stale — never abort ingest, or one garbage event + // starves every message queued behind it in a transport drain. + let garbage = TransportMessage { + id: hash_id(b"malformed garbage"), + payload: b"too short".to_vec(), + timestamp: Timestamp(0), + causal_deps: vec![], + source: TransportSource("mock".into()), + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + }; + let outcome = bob + .ingest(garbage) + .await + .expect("malformed input must classify as stale, not abort ingest"); + assert!( + matches!( + outcome, + IngestOutcome::Stale { + reason: StaleReason::PeelFailed + } + ), + "expected terminal PeelFailed for malformed content, got {outcome:?}" + ); + + let msg = match alice + .send(SendIntent::AppMessage { + group_id: group_id.clone(), + payload: app_payload_for(&alice, b"after the garbage"), + }) + .await + .unwrap() + { + SendResult::ApplicationMessage { msg } => TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + ..msg + }, + _ => unreachable!(), + }; + let after = bob.ingest(msg).await.unwrap(); + assert!(matches!(after, IngestOutcome::Processed)); + let events = bob.drain_events(); + assert!( + events.iter().any( + |event| matches!(event, GroupEvent::MessageReceived { payload, .. } if app_content(payload) == b"after the garbage") + ), + "expected the message behind the garbage to still deliver, got {events:?}" + ); +} + +#[tokio::test] +async fn malformed_message_buffered_during_pending_publish_lands_terminal_after_rollback() { + let mut alice = + build_client_with_peeler(b"alice-pending", Box::new(MalformedShortPayloadPeeler)); + let mut bob = build_client(b"bob-pending"); + let bob_kp = bob.fresh_key_package().await.unwrap(); + + let (group_id, result) = alice + .create_group(CreateGroupRequest { + name: "original".into(), + description: "".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![], + }) + .await + .unwrap(); + let pending = match result { + SendResult::GroupCreated { pending, .. } => pending, + _ => unreachable!(), + }; + alice.confirm_published(pending).await.unwrap(); + + // Stage a commit so the group sits in PendingPublish — the window where + // inbound input is persisted `Retryable` for replay BEFORE the peeler + // ever classifies it. + let staged = match alice + .send(SendIntent::UpdateGroupData { + group_id: group_id.clone(), + name: Some("doomed".into()), + description: None, + }) + .await + .unwrap() + { + SendResult::GroupEvolution { pending, .. } => pending, + _ => unreachable!(), + }; + + let garbage = TransportMessage { + id: hash_id(b"malformed garbage during pending publish"), + payload: b"too short".to_vec(), + timestamp: Timestamp(0), + causal_deps: vec![], + source: TransportSource("mock".into()), + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + }; + let buffered = alice.ingest(garbage.clone()).await.unwrap(); + assert!( + matches!(buffered, IngestOutcome::Buffered { .. }), + "pre-peel buffering during PendingPublish is the tested entry \ + condition, got {buffered:?}" + ); + + // Rollback returns the group to Stable and replays the buffered backlog; + // the garbage now peels `Malformed` for the first time. The terminal + // contract: once classification has run, the attacker-keyed row must not + // stay in a non-terminal state that re-enters replay forever. + alice.publish_failed(staged).await.unwrap(); + let after_replay = alice.ingest(garbage).await.unwrap(); + assert!( + matches!(after_replay, IngestOutcome::Stale { .. }), + "a malformed message must land terminal once classified — a \ + `Buffered` here means the stored row is still non-terminal and \ + perpetually reported as pending, got {after_replay:?}" + ); +} + #[tokio::test] async fn ingest_own_created_message_returns_own_echo() { // Alice sends an app message, then ingests her own outbound message diff --git a/crates/marmot-app/tests/malformed_445_drain.rs b/crates/marmot-app/tests/malformed_445_drain.rs new file mode 100644 index 00000000..18d3dbda --- /dev/null +++ b/crates/marmot-app/tests/malformed_445_drain.rs @@ -0,0 +1,204 @@ +//! Acceptance test for hostile-input resilience of the transport drain. +//! +//! Anyone can publish a kind-445 event to a group's cleartext routing tag +//! without being a member, so structurally-invalid content — too short to +//! carry the spec's `base64(nonce || ciphertext)` envelope — is ordinary +//! hostile wire input, not an exceptional condition. It must classify as +//! stale inside the engine and never surface as a sync failure: a peel error +//! that propagates as `Err` aborts the whole catch-up drain, loses the sync +//! summary (and with it every `MessageReceived` event the drain had +//! produced), skips the app-state save, and — because the garbage event +//! remains unremembered — re-aborts every subsequent catch-up the relay +//! replays it into. One garbage event starves every message queued behind it. +//! +//! The shape mirrors `since_floor.rs`'s cold-boot harness (see that file's +//! module docs for why one account per store and why a full shutdown, not +//! `restart_account`): bob goes fully offline, a malformed kind-445 and a +//! real message land in his backlog, and a cold boot's first catch-up must +//! deliver the real message and record zero sync failures. + +use std::time::Duration; + +use marmot_account::AccountHome; +use marmot_app::{MarmotApp, MarmotAppConfig, MarmotAppEvent, MarmotAppRuntime}; +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 transport_nostr_adapter::{NostrRelayClient, NostrSdkRelayClient}; +use transport_nostr_peeler::{NOSTR_GROUP_CONTENT_MIN_LEN, NostrTransportEvent}; + +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() +} + +async fn wait_for_event( + events: &mut tokio::sync::broadcast::Receiver, + mut matches_event: F, +) where + F: FnMut(&MarmotAppEvent) -> bool, +{ + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let event = events.recv().await.unwrap(); + if matches_event(&event) { + return; + } + } + }) + .await + .expect("runtime event") +} + +/// Publish a kind-445 whose content is structurally too short to be a +/// `base64(nonce || ciphertext)` envelope, with a fresh ephemeral (non-member) +/// signing key — the peeler classifies it `Malformed` before any decryption +/// attempt. Contrast `since_floor.rs`'s probe helper, which deliberately +/// builds an envelope-*shaped* payload so it classifies `DecryptFailed` +/// instead; this helper pins the other, structurally-invalid classification. +async fn publish_malformed_group_message_at( + relay_url: &str, + nostr_group_id_hex: &str, + created_at: u64, +) { + let short = b"too short"; + assert!(short.len() < NOSTR_GROUP_CONTENT_MIN_LEN); + let ephemeral = Keys::generate(); + let signed = EventBuilder::new(Kind::MlsGroupMessage, BASE64_STANDARD.encode(short)) + .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 malformed kind-445 test event"); +} + +fn account_sync_failures(runtime: &MarmotAppRuntime) -> u64 { + runtime + .shared_services() + .app_performance_telemetry() + .snapshot() + .account_sync + .failures +} + +#[tokio::test] +async fn malformed_group_message_does_not_starve_messages_behind_it() { + let (_relay, url) = mock_relay().await; + + // One account per store, exactly as documented in `since_floor.rs`. + 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 --- + 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("malformed drain resilience", &[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; + + 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; + + // bob fully offline before the hostile event and the real message land in + // his backlog (a live subscription would deliver them piecemeal; the + // production wedge is the cold catch-up drain that replays both at once). + runtime_bob_boot1.shutdown().await; + + // Hostile input first (slightly in the past), then a real message behind + // it — the delivery this test insists must survive the garbage. + publish_malformed_group_message_at( + &url, + &nostr_group_id_hex, + test_unix_now_seconds().saturating_sub(30), + ) + .await; + alice_client + .send(&group_id, b"delivered behind the garbage") + .await + .unwrap(); + + // --- boot 2: the cold catch-up must drain past the garbage --- + let app_bob_boot2 = open_store(&dir_bob, &url); + let runtime_bob_boot2 = MarmotAppRuntime::new(app_bob_boot2.clone()); + let mut events_bob_boot2 = runtime_bob_boot2.subscribe(); + runtime_bob_boot2.start().await.unwrap(); + wait_for_event(&mut events_bob_boot2, |event| { + matches!( + event, + MarmotAppEvent::MessageReceived(message) + if message.account_id_hex == bob_id + && message.message.group_id == group_id + && message.message.plaintext == "delivered behind the garbage" + ) + }) + .await; + assert_eq!( + account_sync_failures(&runtime_bob_boot2), + 0, + "hostile wire input must classify as stale inside the engine, never \ + surface as a sync failure", + ); + runtime_bob_boot2.shutdown().await; +} From f7fb50910db458b672703b621964e68bd7e29a71 Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Thu, 16 Jul 2026 23:40:14 -0400 Subject: [PATCH 2/5] refactor(cgka-engine): move raw-retry-row retirement into store.rs per module split mark_raw_transport_message_failed_if_awaiting_retry is durable stored-message state, so it belongs with store.rs's persistence helpers, not the ingest peel/classify/apply path. Pure relocation, no behavior change. --- .../src/message_processor/ingest.rs | 37 ------------------- .../src/message_processor/store.rs | 37 +++++++++++++++++++ 2 files changed, 37 insertions(+), 37 deletions(-) diff --git a/crates/cgka-engine/src/message_processor/ingest.rs b/crates/cgka-engine/src/message_processor/ingest.rs index bce61ea6..6aaa129d 100644 --- a/crates/cgka-engine/src/message_processor/ingest.rs +++ b/crates/cgka-engine/src/message_processor/ingest.rs @@ -1593,43 +1593,6 @@ impl Engine { Ok(()) } - fn mark_raw_transport_message_failed_if_awaiting_retry( - &mut self, - raw_msg_id: &MessageId, - reason: &str, - ) -> Result<(), EngineError> { - match self.storage.get_message(raw_msg_id) { - Ok(record) - if matches!( - record.state, - MessageState::PeelDeferred | MessageState::Retryable - ) => - { - self.storage - .update_message_state(raw_msg_id, MessageState::Failed)?; - self.audit_group( - &record.group_id, - crate::audit_helpers::message_state_transition_event( - hex::encode(raw_msg_id.as_slice()), - Some(record.state), - MessageState::Failed, - Some(record.epoch), - reason, - ), - ); - // Only a `PeelDeferred` row holds a flood-cap slot (mdk#339); - // a `Retryable` row — input buffered pre-peel while the group - // could not ingest — sits outside the cap. - if record.state == MessageState::PeelDeferred { - self.note_peel_deferred_row_retired(&record.group_id, raw_msg_id); - } - Ok(()) - } - Ok(_) | Err(StorageError::NotFound) => Ok(()), - Err(err) => Err(EngineError::Storage(err)), - } - } - /// Whether `msg_epoch` predates this device's membership in `group_id` /// (mdk#339): such a message was never decryptable here by design — /// OpenMLS holds no secrets for epochs before the welcome — so it is diff --git a/crates/cgka-engine/src/message_processor/store.rs b/crates/cgka-engine/src/message_processor/store.rs index 947a1719..a4340814 100644 --- a/crates/cgka-engine/src/message_processor/store.rs +++ b/crates/cgka-engine/src/message_processor/store.rs @@ -278,6 +278,43 @@ impl Engine { } Ok(()) } + + pub(crate) fn mark_raw_transport_message_failed_if_awaiting_retry( + &mut self, + raw_msg_id: &MessageId, + reason: &str, + ) -> Result<(), EngineError> { + match self.storage.get_message(raw_msg_id) { + Ok(record) + if matches!( + record.state, + MessageState::PeelDeferred | MessageState::Retryable + ) => + { + self.storage + .update_message_state(raw_msg_id, MessageState::Failed)?; + self.audit_group( + &record.group_id, + crate::audit_helpers::message_state_transition_event( + hex::encode(raw_msg_id.as_slice()), + Some(record.state), + MessageState::Failed, + Some(record.epoch), + reason, + ), + ); + // Only a `PeelDeferred` row holds a flood-cap slot (mdk#339); + // a `Retryable` row — input buffered pre-peel while the group + // could not ingest — sits outside the cap. + if record.state == MessageState::PeelDeferred { + self.note_peel_deferred_row_retired(&record.group_id, raw_msg_id); + } + Ok(()) + } + Ok(_) | Err(StorageError::NotFound) => Ok(()), + Err(err) => Err(EngineError::Storage(err)), + } + } } #[cfg(test)] From df32c33fa2a209c01480d0e843020a0a0a02b836 Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Thu, 16 Jul 2026 23:47:54 -0400 Subject: [PATCH 3/5] fix(cgka-engine): retire malformed snapshot-fallback peels as terminal stale instead of aborting the drain The two snapshot-fallback call sites propagated a PeelerError::Malformed with ? , aborting the whole transport drain on one garbage event. Give them the direct peel's terminal handling via a shared malformed_terminal_stale helper. Defense-in-depth for the public TransportPeeler contract, not a production-reachable Nostr bug; closes the one-seam guard gap (mdk#707). --- .../src/message_processor/ingest.rs | 74 ++++++- crates/cgka-engine/tests/ingest.rs | 195 +++++++++++++++++- 2 files changed, 258 insertions(+), 11 deletions(-) diff --git a/crates/cgka-engine/src/message_processor/ingest.rs b/crates/cgka-engine/src/message_processor/ingest.rs index 6aaa129d..d7c82dbc 100644 --- a/crates/cgka-engine/src/message_processor/ingest.rs +++ b/crates/cgka-engine/src/message_processor/ingest.rs @@ -254,14 +254,33 @@ impl Engine { let peeled = match peel_result { Ok(p) => p, Err(PeelerError::DecryptFailed) => { - if let Some(recovery) = self + let recovered = match self .try_peel_group_message_from_available_snapshots( msg, &group_id, current_epoch, ) - .await? + .await { + Ok(recovered) => recovered, + // Defense-in-depth for the public TransportPeeler + // contract: a `Malformed` verdict raised only by the + // snapshot fallback is terminal on this seam too — never + // an aborted drain (mdk#707). Not reachable via the + // production Nostr peeler (its Malformed detection is + // deterministic in the message bytes, so the direct peel + // catches it first); MissingContext / storage errors keep + // propagating as genuine engine faults. + Err(EngineError::Peeler(PeelerError::Malformed(_))) => { + return self.malformed_terminal_stale( + &raw_msg_id, + &msg.id, + "malformed_payload_snapshot_fallback", + ); + } + Err(e) => return Err(e), + }; + if let Some(recovery) = recovered { self.audit_group( &group_id, marmot_forensics::AuditEventKind::PeelerOutcome { @@ -329,14 +348,33 @@ impl Engine { } } Err(PeelerError::StaleEpoch { .. }) => { - if let Some(recovery) = self + let recovered = match self .try_peel_group_message_from_available_snapshots( msg, &group_id, current_epoch, ) - .await? + .await { + Ok(recovered) => recovered, + // Defense-in-depth for the public TransportPeeler + // contract: a `Malformed` verdict raised only by the + // snapshot fallback is terminal on this seam too — never + // an aborted drain (mdk#707). Not reachable via the + // production Nostr peeler (its Malformed detection is + // deterministic in the message bytes, so the direct peel + // catches it first); MissingContext / storage errors keep + // propagating as genuine engine faults. + Err(EngineError::Peeler(PeelerError::Malformed(_))) => { + return self.malformed_terminal_stale( + &raw_msg_id, + &msg.id, + "malformed_payload_snapshot_fallback", + ); + } + Err(e) => return Err(e), + }; + if let Some(recovery) = recovered { self.audit_group( &group_id, marmot_forensics::AuditEventKind::PeelerOutcome { @@ -392,14 +430,11 @@ impl Engine { // rows keyed by them would grow without bound. Retire the // raw deferred row if this arrived via the retry // lifecycle; a no-op on the direct path. - self.mark_raw_transport_message_failed_if_awaiting_retry( + return self.malformed_terminal_stale( &raw_msg_id, + &msg.id, "malformed_payload", - )?; - self.seen_message_ids.insert(msg.id.clone()); - return Ok(IngestOutcome::Stale { - reason: StaleReason::PeelFailed, - }); + ); } Err(e) => return Err(EngineError::Peeler(e)), }; @@ -1593,6 +1628,25 @@ impl Engine { Ok(()) } + /// Terminal disposition for a peel that resolved to `Malformed`: retire the + /// awaiting-retry raw transport row (a no-op on the direct path, where no + /// deferred row exists), mark the id seen so a redelivery short-circuits, + /// and classify the input `Stale { PeelFailed }`. Shared by the direct peel + /// and both snapshot-fallback seams so the terminal behavior has a single + /// definition — a guard living on one seam only is a bug (mdk#707). + fn malformed_terminal_stale( + &mut self, + raw_msg_id: &MessageId, + msg_id: &MessageId, + reason: &'static str, + ) -> Result { + self.mark_raw_transport_message_failed_if_awaiting_retry(raw_msg_id, reason)?; + self.seen_message_ids.insert(msg_id.clone()); + Ok(IngestOutcome::Stale { + reason: StaleReason::PeelFailed, + }) + } + /// Whether `msg_epoch` predates this device's membership in `group_id` /// (mdk#339): such a message was never decryptable here by design — /// OpenMLS holds no secrets for epochs before the welcome — so it is diff --git a/crates/cgka-engine/tests/ingest.rs b/crates/cgka-engine/tests/ingest.rs index ad432879..d143ee12 100644 --- a/crates/cgka-engine/tests/ingest.rs +++ b/crates/cgka-engine/tests/ingest.rs @@ -18,7 +18,7 @@ use cgka_traits::storage::MessageStorage; use cgka_traits::transport::{ EncryptedPayload, Timestamp, TransportEnvelope, TransportMessage, TransportSource, }; -use cgka_traits::types::{MemberId, MessageId}; +use cgka_traits::types::{EpochId, MemberId, MessageId}; use std::collections::HashSet; use std::sync::{Arc, Mutex}; use storage_sqlite::{SqlCipherKey, SqliteAccountStorage}; @@ -263,6 +263,62 @@ impl TransportPeeler for MalformedShortPayloadPeeler { } } +/// Garbage payload whose malformed verdict is context-dependent: it peels +/// `DecryptFailed` against the live epoch (so ingest falls through to the +/// retained-snapshot fallback) but `Malformed` against any older snapshot +/// context. This simulates a trait-permitted non-Nostr peeler: the production +/// Nostr peeler can never reach the fallback with a `Malformed` verdict, since +/// its malformed detection is a pure function of the message bytes and is +/// always caught by the direct peel first. The public `TransportPeeler` +/// contract, however, allows `StaleEpoch`-hint peelers whose malformed +/// detection is context-dependent, so the fallback seam must still treat that +/// verdict as terminal (mdk#707). +const SNAPSHOT_FALLBACK_GARBAGE: &[u8] = b"snapshot-fallback-garbage-payload"; + +struct SnapshotFallbackMalformedPeeler { + malformed_below_epoch: u64, +} + +#[async_trait] +impl TransportPeeler for SnapshotFallbackMalformedPeeler { + async fn peel_group_message( + &self, + msg: &TransportMessage, + ctx: &GroupContextSnapshot, + ) -> Result { + if msg.payload == SNAPSHOT_FALLBACK_GARBAGE { + return if ctx.epoch().0 < self.malformed_below_epoch { + Err(PeelerError::Malformed( + "malformed only against an older snapshot context".into(), + )) + } else { + Err(PeelerError::DecryptFailed) + }; + } + MockPeeler.peel_group_message(msg, ctx).await + } + + async fn peel_welcome(&self, msg: &TransportMessage) -> Result { + MockPeeler.peel_welcome(msg).await + } + + async fn wrap_group_message( + &self, + payload: &EncryptedPayload, + ctx: &GroupContextSnapshot, + ) -> Result { + MockPeeler.wrap_group_message(payload, ctx).await + } + + async fn wrap_welcome( + &self, + payload: &EncryptedPayload, + recipient: &MemberId, + ) -> Result { + MockPeeler.wrap_welcome(payload, recipient).await + } +} + fn selfremove_registry() -> FeatureRegistry { let mut r = FeatureRegistry::new(); r.register( @@ -670,6 +726,143 @@ async fn malformed_message_buffered_during_pending_publish_lands_terminal_after_ ); } +#[tokio::test] +async fn malformed_via_snapshot_fallback_is_stale_and_does_not_wedge_ingest() { + // Simulates a trait-permitted non-Nostr peeler whose `Malformed` verdict is + // context-dependent: the direct peel at the live epoch returns + // `DecryptFailed`, driving ingest into the retained-snapshot fallback, where + // the peel against the older snapshot context returns `Malformed`. The + // production Nostr peeler cannot reach this branch (its malformed detection + // is a pure function of the bytes, so the direct peel catches it first), but + // the fallback seam must handle the verdict identically to the direct seam: + // terminal stale, never an aborted drain (mdk#707). + let mut alice = build_client(b"alice"); + // The group reaches epoch 2 below (create -> 1, invite -> 2), so the + // retained epoch-1 anchor is the only past-peel snapshot older than live. + let live_epoch = 2u64; + let mut bob = build_client_with_peeler( + b"bob", + Box::new(SnapshotFallbackMalformedPeeler { + malformed_below_epoch: live_epoch, + }), + ); + let mut carol = build_client(b"carol"); + let bob_kp = bob.fresh_key_package().await.unwrap(); + + let (group_id, result) = alice + .create_group(CreateGroupRequest { + name: "".into(), + description: "".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![], + }) + .await + .unwrap(); + let (pending, bob_welcome) = match result { + SendResult::GroupCreated { + pending, + mut welcomes, + } => (pending, welcomes.remove(0)), + _ => unreachable!(), + }; + alice.confirm_published(pending).await.unwrap(); + bob.join_welcome(bob_welcome).await.unwrap(); + + // Alice invites Carol so a commit advances the group and Bob retains an + // anchor snapshot at the pre-commit epoch — the snapshot the fallback rolls + // back to and peels against below. + let carol_kp = carol.fresh_key_package().await.unwrap(); + let invite = match alice + .send(SendIntent::Invite { + group_id: group_id.clone(), + key_packages: vec![carol_kp], + }) + .await + .unwrap() + { + SendResult::GroupEvolution { msg, pending, .. } => { + alice.confirm_published(pending).await.unwrap(); + TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + ..msg + } + } + _ => unreachable!(), + }; + // Convergence settles a peer commit only after its quiescence window has + // elapsed, so buffer at one instant and converge at a later one (mirroring + // the retained-anchor convergence tests) to apply the commit and retain the + // pre-commit epoch anchor. + bob.buffer_openmls_convergence_message(&group_id, invite, 1_000) + .expect("invite commit buffered"); + bob.converge_stored_openmls_messages(&group_id, 1_000_000) + .expect("invite commit applies and retains the pre-commit anchor"); + assert_eq!( + bob.epoch(&group_id).unwrap(), + EpochId(live_epoch), + "Bob must advance to the live epoch so an older retained anchor exists" + ); + + // Garbage arrives: direct peel `DecryptFailed` -> snapshot fallback peels + // `Malformed`. The terminal contract must hold on this seam too — classify + // stale, do not abort the drain. + let garbage = TransportMessage { + id: hash_id(b"malformed via snapshot fallback"), + payload: SNAPSHOT_FALLBACK_GARBAGE.to_vec(), + timestamp: Timestamp(0), + causal_deps: vec![], + source: TransportSource("mock".into()), + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + }; + let outcome = bob + .ingest(garbage) + .await + .expect("malformed snapshot-fallback peel must classify stale, not abort ingest"); + assert!( + matches!( + outcome, + IngestOutcome::Stale { + reason: StaleReason::PeelFailed + } + ), + "expected terminal PeelFailed for a malformed snapshot-fallback peel, got {outcome:?}" + ); + + // The drain is not wedged: a well-formed message queued behind the garbage + // still processes. + let msg = match alice + .send(SendIntent::AppMessage { + group_id: group_id.clone(), + payload: app_payload_for(&alice, b"after the snapshot garbage"), + }) + .await + .unwrap() + { + SendResult::ApplicationMessage { msg } => TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + ..msg + }, + _ => unreachable!(), + }; + let after = bob.ingest(msg).await.unwrap(); + assert!(matches!(after, IngestOutcome::Processed)); + let events = bob.drain_events(); + assert!( + events.iter().any( + |event| matches!(event, GroupEvent::MessageReceived { payload, .. } if app_content(payload) == b"after the snapshot garbage") + ), + "expected the message behind the garbage to still deliver, got {events:?}" + ); +} + #[tokio::test] async fn ingest_own_created_message_returns_own_echo() { // Alice sends an app message, then ingests her own outbound message From b2451d50bcd58fd1d9a128fe94b1fff60d8ee70e Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Fri, 17 Jul 2026 10:40:12 -0400 Subject: [PATCH 4/5] fix(cgka-engine): retire non-deferred raw rows on terminal replay outcomes Closes a pre-existing gap in replay_buffered_messages where a raw transport row buffered Retryable during our own PendingPublish/Merging window was never retired when re-ingest returned a terminal Ok outcome (Processed, AlreadySeen, AlreadyAtEpoch/EpochInvalidated, OwnEcho, SelfEvicted): it lingered Retryable and was re-peeled on every later publish-cycle replay. The Ok(_) arm now retires the raw wrapper Processed (cap-slot release stays gated on PeelDeferred, so no mdk#339 accounting impact); UnknownGroup moves to the leave-in-retry arm since ingest deliberately re-buffers it. The retirement is guarded on the row's re-read current state so it fires only while the row is still awaiting retry: a terminal state ingest already committed to the same row during re-ingest (e.g. SelfEvicted persisting it Failed) is authoritative and is never clobbered to Processed. --- .../cgka-engine/src/message_processor/mod.rs | 52 +++- .../src/message_processor/store.rs | 25 ++ crates/cgka-engine/tests/fork_detection.rs | 242 +++++++++++++++++- crates/cgka-engine/tests/ingest.rs | 177 +++++++++++++ 4 files changed, 482 insertions(+), 14 deletions(-) diff --git a/crates/cgka-engine/src/message_processor/mod.rs b/crates/cgka-engine/src/message_processor/mod.rs index bccf3e5e..92bea2b4 100644 --- a/crates/cgka-engine/src/message_processor/mod.rs +++ b/crates/cgka-engine/src/message_processor/mod.rs @@ -880,23 +880,49 @@ impl Engine { } } Ok(IngestOutcome::Stale { - reason: StaleReason::PeelFailed | StaleReason::Quarantined, + reason: + StaleReason::PeelFailed | StaleReason::Quarantined | StaleReason::UnknownGroup, }) => { - // Still un-peelable, or the group is quarantined: leave the - // row in its retry state. A terminal-after-peel `PeelFailed` - // (malformed, stale epoch with no snapshot) already retired - // its raw row — `PeelDeferred` or `Retryable` alike — inside - // `ingest_group_message` via - // `mark_raw_transport_message_failed_if_awaiting_retry`. + // Leave the row in its retry state so a later pass re-attempts + // it. `PeelFailed`: still un-peelable, or already retired to + // `Failed` by a terminal-after-peel path inside + // `ingest_group_message` + // (`mark_raw_transport_message_failed_if_awaiting_retry`, + // `PeelDeferred`/`Retryable` alike). `Quarantined`: the group + // is frozen; the row replays once repair clears it. + // `UnknownGroup`: no local group matches yet — `ingest` + // deliberately re-buffered the row `Retryable` because a later + // welcome may create the group, so terminalizing it here would + // drop a recoverable message. } Ok(_) => { - // Applied (`Processed`) or terminally reclassified - // (`AlreadySeen`, `SelfEvicted`, ...): retire a raw - // deferred wrapper so it stops holding a cap slot - // (mdk#339). Non-deferred rows keep the pre-existing - // no-op behavior. - if was_peel_deferred { + // Terminal reclassification of the raw wrapper: the content- + // derived row now carries the real verdict — applied + // (`Processed`), a same-epoch fork the incumbent won + // (`AlreadyAtEpoch`, content row `EpochInvalidated`), a + // duplicate (`AlreadySeen`), our own echo (`OwnEcho`), or our + // own eviction (`SelfEvicted`). Retire the raw wrapper so it + // leaves the retry lifecycle instead of being re-peeled on + // every subsequent publish-cycle replay — but ONLY while it is + // still awaiting retry. `record.state` is the pre-ingest + // snapshot; `ingest_group_message` may have already committed a + // terminal state to this same row during the call. The + // reachable case is `SelfEvicted`: a buffered peer commit that + // evicts our leaf makes the next buffered row hit `!is_active`, + // which persists that row `Failed` (ingest.rs). That + // ingest-committed verdict is authoritative — relabeling an + // evicted-on row `Processed` would sweep it back into + // canonicalization (`openmls_projection` / + // `distributed_convergence` select on `Processed`). Re-read the + // row and retire only a state ingest left awaiting retry. + if self.raw_transport_row_awaiting_retry(&record.id)? { self.update_stored_message_state(&record.id, MessageState::Processed)?; + } + // Only a `PeelDeferred` row holds a flood-cap slot (mdk#339); + // its accounting tracks the original deferred count, so release + // the slot whenever the row leaves the retry lifecycle — whether + // this arm retired it or ingest already terminalized it. + if was_peel_deferred { self.note_peel_deferred_row_retired(group_id, &record.id); } } diff --git a/crates/cgka-engine/src/message_processor/store.rs b/crates/cgka-engine/src/message_processor/store.rs index a4340814..2ad93755 100644 --- a/crates/cgka-engine/src/message_processor/store.rs +++ b/crates/cgka-engine/src/message_processor/store.rs @@ -55,6 +55,31 @@ impl Engine { )) } + /// Whether the raw transport row `id` is STILL awaiting retry as of its + /// *current* stored state — the same `Created | Retryable | PeelDeferred` + /// set the replay / deferred-peel loops admit at entry. + /// + /// Retirement paths re-read through this rather than trusting a row state + /// snapshotted before re-ingest: `ingest_group_message` can commit a + /// terminal state to this same row during the call (e.g. the `SelfEvicted` + /// path persists it `Failed`, `ingest.rs`), and that verdict is + /// authoritative — overwriting it with `Processed` would relabel a row we + /// were evicted on as a canonicalization input. A vanished row + /// (`NotFound`) is not awaiting retry. + pub(crate) fn raw_transport_row_awaiting_retry( + &self, + id: &MessageId, + ) -> Result { + match self.storage.get_message(id) { + Ok(record) => Ok(matches!( + record.state, + MessageState::Created | MessageState::Retryable | MessageState::PeelDeferred + )), + Err(StorageError::NotFound) => Ok(false), + Err(e) => Err(EngineError::Storage(e)), + } + } + pub(crate) fn record_sent_message( &mut self, msg: &TransportMessage, diff --git a/crates/cgka-engine/tests/fork_detection.rs b/crates/cgka-engine/tests/fork_detection.rs index 6dea5853..11de5dc0 100644 --- a/crates/cgka-engine/tests/fork_detection.rs +++ b/crates/cgka-engine/tests/fork_detection.rs @@ -21,8 +21,9 @@ use cgka_traits::engine::{ use cgka_traits::error::PeelerError; use cgka_traits::group_context::GroupContextSnapshot; use cgka_traits::ingest::{PeeledContent, PeeledMessage}; +use cgka_traits::message::MessageState; use cgka_traits::peeler::TransportPeeler; -use cgka_traits::storage::{AccountDeviceSignerStorage, StorageProvider}; +use cgka_traits::storage::{AccountDeviceSignerStorage, MessageStorage, StorageProvider}; use cgka_traits::transport::{ EncryptedPayload, Timestamp, TransportEnvelope, TransportMessage, TransportSource, }; @@ -477,6 +478,245 @@ async fn convergence_privileged_remove_beats_grinding_ordinary_self_update() { ); } +/// A peer's same-epoch commit that arrives during our own `PendingPublish` +/// window is buffered `Retryable` before any peel. When our own commit confirms +/// and wins the fork, replay must reclassify the buffered commit terminal: the +/// content row lands `EpochInvalidated`, and the raw transport wrapper is +/// retired `Processed` — not left `Retryable` to be re-peeled on every later +/// publish-cycle replay. +#[tokio::test] +async fn buffered_losing_fork_commit_raw_row_is_retired_after_confirm_replay() { + use cgka_traits::ingest::IngestOutcome; + use sha2::{Digest, Sha256}; + + let (mut alice, alice_storage) = build_client_with_storage(b"alice-buffered-fork"); + let (mut bob, bob_storage) = build_client_with_storage(b"bob-buffered-fork"); + let bob_id = bob.self_id(); + + let bob_kp = bob.fresh_key_package().await.unwrap(); + let (group_id, create) = alice + .create_group(CreateGroupRequest { + name: "buffered-losing-fork".into(), + description: "".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![], + }) + .await + .unwrap(); + let welcome = match create { + SendResult::GroupCreated { + pending, + mut welcomes, + } => { + alice.confirm_published(pending).await.unwrap(); + welcomes.remove(0) + } + other => panic!("expected GroupCreated, got {other:?}"), + }; + bob.join_welcome(welcome).await.unwrap(); + assert_eq!(alice.epoch(&group_id).unwrap(), EpochId(1)); + + // Bob's ordinary self-update commit branches from epoch 1. A privileged + // admin commit always sorts ahead of an ordinary one, so the incumbent wins + // regardless of digest — no grinding needed to pin the winner. + let self_update = raw_self_update_commit(&bob_storage, &bob_id, &group_id); + let raw_id = self_update.id.clone(); + let content_id = MessageId::new(Sha256::digest(&self_update.payload).to_vec()); + + // Alice stages a privileged admin Remove(Bob) → PendingPublish, the window + // where inbound input is buffered `Retryable` before any peel. + let remove_pending = match alice + .send(SendIntent::RemoveMembers { + group_id: group_id.clone(), + members: vec![bob_id.clone()], + }) + .await + .unwrap() + { + SendResult::GroupEvolution { pending, .. } => pending, + other => panic!("expected RemoveMembers GroupEvolution, got {other:?}"), + }; + + let buffered = alice.ingest(self_update).await.unwrap(); + assert!( + matches!(buffered, IngestOutcome::Buffered { .. }), + "peer commit during PendingPublish must buffer, got {buffered:?}" + ); + assert_eq!( + alice_storage.get_message(&raw_id).unwrap().state, + MessageState::Retryable, + "the buffered raw transport row is persisted Retryable pending replay" + ); + + // Confirm advances Alice to epoch 2 and replays the backlog. The buffered + // commit loses the fork to the incumbent (privileged > ordinary). + alice.confirm_published(remove_pending).await.unwrap(); + assert_eq!(alice.epoch(&group_id).unwrap(), EpochId(2)); + + assert_eq!( + alice_storage.get_message(&content_id).unwrap().state, + MessageState::EpochInvalidated, + "the losing fork commit's content row carries the real verdict" + ); + assert_eq!( + alice_storage.get_message(&raw_id).unwrap().state, + MessageState::Processed, + "the raw transport wrapper must be retired terminal after replay — not \ + left Retryable to be re-peeled on every later publish cycle" + ); + assert!( + alice + .drain_events() + .iter() + .all(|event| !matches!(event, GroupEvent::ForkRecovered { .. })), + "the incumbent won the fork; no rollback / ForkRecovered is emitted" + ); +} + +/// Two peer rows buffered during our own `PendingPublish` window replay in a +/// single pass. The first is a privileged admin `RemoveMembers` that removes +/// our leaf; being privileged it deterministically wins the same-source-epoch +/// fork over our own ordinary `UpdateGroupData`, so on confirm-replay fork +/// recovery rolls us back and applies it inline — our local group state goes +/// `Inactive`. The second buffered row is then re-ingested against that +/// inactive state, so `ingest_group_message` classifies it `SelfEvicted` and +/// persists its raw row `Failed` (authenticated evidence we were removed). +/// +/// Replay retires the backlog, and that retirement must only touch rows STILL +/// awaiting retry after re-ingest. The unconditional retirement write introduced +/// in ae616488 violates this in two ways in exactly this scenario: it errors +/// (`Storage(NotFound)`) on the winning remove's raw row, which fork-recovery +/// rollback already swept away, and — the invariant this guards — it would +/// relabel the `SelfEvicted` row `Processed`, clobbering the `Failed` verdict +/// ingest committed. A `Processed` raw row is a canonicalization input +/// (`openmls_projection` / `distributed_convergence` select on it), so a row we +/// were evicted on must stay `Failed`, never be swept back into convergence. +#[tokio::test] +async fn buffered_self_evicted_row_stays_failed_not_swept_processed_on_replay() { + use cgka_traits::ingest::IngestOutcome; + + let (mut alice, alice_storage) = build_client_with_storage(b"alice-self-evict-replay"); + let (mut bob, bob_storage) = build_client_with_storage(b"bob-self-evict-replay"); + let alice_id = alice.self_id(); + let bob_id = bob.self_id(); + + // Bob is a co-admin so he can remove Alice; Alice (creator) is implicitly an + // admin, so removing her still leaves an admin (Bob) — MIP-03 §149. + let bob_kp = bob.fresh_key_package().await.unwrap(); + let (group_id, create) = alice + .create_group(CreateGroupRequest { + name: "self-evict-replay".into(), + description: "".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![bob_id.clone()], + }) + .await + .unwrap(); + let welcome = match create { + SendResult::GroupCreated { + pending, + mut welcomes, + } => { + alice.confirm_published(pending).await.unwrap(); + welcomes.remove(0) + } + other => panic!("expected GroupCreated, got {other:?}"), + }; + bob.join_welcome(welcome).await.unwrap(); + assert_eq!(alice.epoch(&group_id).unwrap(), EpochId(1)); + + let route = |msg: TransportMessage| TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + ..msg + }; + + // A second epoch-1 message from Bob, produced before he advances. On replay + // this is the row that classifies `SelfEvicted` once Alice's leaf is gone — + // its content is never peeled (the `!is_active` gate fires first). + let second = route(raw_self_update_commit(&bob_storage, &bob_id, &group_id)); + let second_id = second.id.clone(); + + // Bob (admin) removes Alice at epoch 1 → a privileged commit. + let remove_alice = match bob + .send(SendIntent::RemoveMembers { + group_id: group_id.clone(), + members: vec![alice_id.clone()], + }) + .await + .unwrap() + { + SendResult::GroupEvolution { msg, pending, .. } => { + bob.confirm_published(pending).await.unwrap(); + route(msg) + } + other => panic!("expected RemoveMembers GroupEvolution, got {other:?}"), + }; + assert_ne!(remove_alice.id, second_id, "distinct buffered rows"); + + // Alice stages her own ordinary commit → PendingPublish, the window where + // inbound input buffers `Retryable` before any peel. Buffer the eviction + // commit FIRST (lower insert_order → replayed first), then the second row. + let staged = match alice + .send(SendIntent::UpdateGroupData { + group_id: group_id.clone(), + name: Some("pending".into()), + description: None, + }) + .await + .unwrap() + { + SendResult::GroupEvolution { pending, .. } => pending, + other => panic!("expected UpdateGroupData GroupEvolution, got {other:?}"), + }; + + for msg in [remove_alice, second] { + let id = msg.id.clone(); + assert!( + matches!( + alice.ingest(msg).await.unwrap(), + IngestOutcome::Buffered { .. } + ), + "peer input during PendingPublish must buffer" + ); + assert_eq!( + alice_storage.get_message(&id).unwrap().state, + MessageState::Retryable, + "buffered raw rows are persisted Retryable pending replay" + ); + } + + // Confirm advances Alice to epoch 2 (her ordinary update applied) and + // replays the backlog. The buffered remove is a same-source-epoch fork the + // privileged committer wins, so fork recovery rolls Alice back and applies + // it inline — evicting her leaf — before the second row is replayed. + alice.confirm_published(staged).await.unwrap(); + + // The buffered remove really did evict Alice on replay. + assert!( + !alice + .members(&group_id) + .unwrap() + .iter() + .any(|member| member.id == alice_id), + "the buffered remove must have evicted Alice on replay" + ); + + // Regression: the `SelfEvicted` row's raw wrapper must keep the `Failed` + // state ingest committed — never clobbered to `Processed`. + assert_eq!( + alice_storage.get_message(&second_id).unwrap().state, + MessageState::Failed, + "a row we were evicted on must stay Failed after replay, not be swept \ + into canonicalization as Processed" + ); +} + #[tokio::test] async fn stale_commit_outside_rewind_horizon_is_not_treated_as_recoverable_fork() { let mut labels: Vec<&'static [u8]> = vec![ diff --git a/crates/cgka-engine/tests/ingest.rs b/crates/cgka-engine/tests/ingest.rs index d143ee12..3d0c16f9 100644 --- a/crates/cgka-engine/tests/ingest.rs +++ b/crates/cgka-engine/tests/ingest.rs @@ -319,6 +319,45 @@ impl TransportPeeler for SnapshotFallbackMalformedPeeler { } } +/// Pass-through peeler (identical to [`MockPeeler`]) that counts how many times +/// `peel_group_message` is invoked, so a test can assert a retired raw row is +/// not wastefully re-peeled on later replay passes. +struct CountingPeeler { + peels: Arc>, +} + +#[async_trait] +impl TransportPeeler for CountingPeeler { + async fn peel_group_message( + &self, + msg: &TransportMessage, + ctx: &GroupContextSnapshot, + ) -> Result { + *self.peels.lock().unwrap() += 1; + MockPeeler.peel_group_message(msg, ctx).await + } + + async fn peel_welcome(&self, msg: &TransportMessage) -> Result { + MockPeeler.peel_welcome(msg).await + } + + async fn wrap_group_message( + &self, + payload: &EncryptedPayload, + ctx: &GroupContextSnapshot, + ) -> Result { + MockPeeler.wrap_group_message(payload, ctx).await + } + + async fn wrap_welcome( + &self, + payload: &EncryptedPayload, + recipient: &MemberId, + ) -> Result { + MockPeeler.wrap_welcome(payload, recipient).await + } +} + fn selfremove_registry() -> FeatureRegistry { let mut r = FeatureRegistry::new(); r.register( @@ -1230,6 +1269,144 @@ async fn inbound_group_message_during_pending_publish_replays_after_rollback() { ); } +/// A peer message that arrives during our own `PendingPublish` window is +/// buffered `Retryable` (persisted before any peel). Once the publish cycle +/// resolves and replay applies it, the raw transport wrapper MUST reach a +/// terminal state: the content-derived row now carries the real verdict, so +/// leaving the raw row `Retryable` only makes replay re-peel it wastefully on +/// every subsequent publish cycle. +#[tokio::test] +async fn buffered_retryable_peer_message_is_retired_terminal_after_replay() { + let peels = Arc::new(Mutex::new(0usize)); + let storage = SqliteAccountStorage::in_memory().unwrap(); + let alice_storage = storage.clone(); + let mut alice = build_client_with_storage_and_peeler( + storage, + b"alice-buffered-retire", + Box::new(CountingPeeler { + peels: peels.clone(), + }), + ); + let mut bob = build_client(b"bob-buffered-retire"); + let bob_kp = bob.fresh_key_package().await.unwrap(); + + let (group_id, result) = alice + .create_group(CreateGroupRequest { + name: "".into(), + description: "".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![], + }) + .await + .unwrap(); + let (create_pending, bob_welcome) = match result { + SendResult::GroupCreated { + pending, + mut welcomes, + } => (pending, welcomes.remove(0)), + _ => unreachable!(), + }; + alice.confirm_published(create_pending).await.unwrap(); + bob.join_welcome(bob_welcome).await.unwrap(); + alice.drain_events(); + bob.drain_events(); + + let bob_msg = match bob + .send(SendIntent::AppMessage { + group_id: group_id.clone(), + payload: app_payload_for(&bob, b"arrived while alice was pending"), + }) + .await + .unwrap() + { + SendResult::ApplicationMessage { msg } => TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + ..msg + }, + _ => unreachable!(), + }; + let raw_id = bob_msg.id.clone(); + + // ── Cycle 1: buffer during PendingPublish, then replay on rollback. ── + let staged = match alice + .send(SendIntent::UpdateGroupData { + group_id: group_id.clone(), + name: Some("pending".into()), + description: None, + }) + .await + .unwrap() + { + SendResult::GroupEvolution { pending, .. } => pending, + _ => unreachable!(), + }; + + let buffered = alice.ingest(bob_msg).await.unwrap(); + assert!( + matches!(buffered, IngestOutcome::Buffered { .. }), + "peer message during PendingPublish must buffer, got {buffered:?}" + ); + assert_eq!( + *peels.lock().unwrap(), + 0, + "buffering is pre-peel: the PendingPublish window must not peel the row" + ); + assert_eq!( + alice_storage.get_message(&raw_id).unwrap().state, + MessageState::Retryable, + "the buffered raw transport row is persisted Retryable pending replay" + ); + + alice.publish_failed(staged).await.unwrap(); + assert!( + alice.drain_events().iter().any( + |e| matches!(e, GroupEvent::MessageReceived { payload, .. } if app_content(payload) == b"arrived while alice was pending"), + ), + "the buffered message must be delivered once replay runs" + ); + assert_eq!( + *peels.lock().unwrap(), + 1, + "replay peels the buffered row exactly once" + ); + assert_eq!( + alice_storage.get_message(&raw_id).unwrap().state, + MessageState::Processed, + "after replay applies it, the raw transport wrapper must be terminal \ + (Processed) — not left Retryable to be re-peeled forever" + ); + + // ── Cycle 2: a second publish cycle must not re-peel the retired row. ── + let staged_again = match alice + .send(SendIntent::UpdateGroupData { + group_id: group_id.clone(), + name: Some("pending-again".into()), + description: None, + }) + .await + .unwrap() + { + SendResult::GroupEvolution { pending, .. } => pending, + _ => unreachable!(), + }; + alice.publish_failed(staged_again).await.unwrap(); + assert_eq!( + *peels.lock().unwrap(), + 1, + "a retired raw row is excluded from replay — no wasted re-peel on the \ + next publish cycle" + ); + assert_eq!( + alice_storage.get_message(&raw_id).unwrap().state, + MessageState::Processed, + "the retired raw row stays terminal across publish cycles" + ); +} + // ── Content-derived dedup id (#238) ────────────────────────────────────────── /// Drive Alice + Bob into a shared group and return both engines plus the From 28cb0c830f452165f44e9ed1e44efa6eb6bd67a9 Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Fri, 17 Jul 2026 12:06:40 -0400 Subject: [PATCH 5/5] fix(cgka-engine): guard retry_deferred_peels retirement against clobbering ingest-committed terminal state Seam parity with replay_buffered_messages (5ae9a440). The deferred-peel sweep snapshots its PeelDeferred rows once, then re-ingests each via ingest_group_message. Both terminal-outcome arms (Ok(Buffered|Processed) and the Ok(Stale{..}) catch-all) retired the raw wrapper Processed UNCONDITIONALLY. But ingest can commit a terminal state to the same row during the call: a deferred future-epoch commit that re-ingests against a now-inactive group hits the !is_active gate, which persists the row Failed (SelfEvicted, ingest.rs). Relabeling that evicted-on row Processed swept it back into canonicalization (openmls_projection / distributed_convergence select on Processed); a fork-recovery rollback that deletes the row mid-sweep would instead crash the write with Storage(NotFound). Both writes now re-read the row's current state through the existing raw_transport_row_awaiting_retry helper and retire only a state ingest left awaiting retry, so ingest's terminal verdict is authoritative. The flood-cap release (note_peel_deferred_row_retired) stays outside the guard, mirroring 5ae9a440: the row leaves the retry lifecycle regardless of who terminalized it. The budget-terminal arm (runs before this row's ingest) and the ForkedEpoch arm (forked_epoch_fail_closed already wrote the row EpochInvalidated, then returns) are left unchanged. --- .../cgka-engine/src/message_processor/mod.rs | 31 ++++- .../tests/deferred_peel_lifecycle.rs | 113 ++++++++++++++++++ 2 files changed, 140 insertions(+), 4 deletions(-) diff --git a/crates/cgka-engine/src/message_processor/mod.rs b/crates/cgka-engine/src/message_processor/mod.rs index 92bea2b4..3ea14416 100644 --- a/crates/cgka-engine/src/message_processor/mod.rs +++ b/crates/cgka-engine/src/message_processor/mod.rs @@ -593,15 +593,38 @@ impl Engine { Ok(IngestOutcome::Buffered { .. } | IngestOutcome::Processed) => { // The peeled content now has its own content-derived record; // retire the raw transport wrapper so it does not keep - // re-entering this retry loop as a stale duplicate. - self.update_stored_message_state(&record.id, MessageState::Processed)?; + // re-entering this retry loop as a stale duplicate — but ONLY + // while it is still awaiting retry. `record` is the pre-ingest + // sweep snapshot; `ingest_group_message` may have committed a + // terminal state to this same row during the call (or + // fork-recovery rollback may have deleted it). That + // ingest-committed verdict is authoritative, so re-read the + // row and retire only a state ingest left awaiting retry + // (seam parity with `replay_buffered_messages`, 5ae9a440). + if self.raw_transport_row_awaiting_retry(&record.id)? { + self.update_stored_message_state(&record.id, MessageState::Processed)?; + } + // Release the flood-cap slot whenever the row leaves the + // retry lifecycle, whether this arm retired it or ingest + // already terminalized it. self.note_peel_deferred_row_retired(group_id, &record.id); progressed += 1; } Ok(IngestOutcome::Stale { .. }) => { // Terminal stale classifications are still successful - // reclassifications of this raw deferred row. - self.update_stored_message_state(&record.id, MessageState::Processed)?; + // reclassifications of this raw deferred row. Retire it only + // while it is still awaiting retry: the reachable case is + // `SelfEvicted`, where re-ingesting the deferred row against a + // now-inactive group makes ingest persist it `Failed` + // (ingest.rs). Relabeling that evicted-on row `Processed` + // would sweep it back into canonicalization, so re-read the + // row and never clobber ingest's terminal verdict (seam parity + // with `replay_buffered_messages`, 5ae9a440). + if self.raw_transport_row_awaiting_retry(&record.id)? { + self.update_stored_message_state(&record.id, MessageState::Processed)?; + } + // The cap-slot release stays outside the guard: the row has + // left the retry lifecycle regardless of who terminalized it. self.note_peel_deferred_row_retired(group_id, &record.id); progressed += 1; } diff --git a/crates/cgka-engine/tests/deferred_peel_lifecycle.rs b/crates/cgka-engine/tests/deferred_peel_lifecycle.rs index a5ae6f71..54c4902c 100644 --- a/crates/cgka-engine/tests/deferred_peel_lifecycle.rs +++ b/crates/cgka-engine/tests/deferred_peel_lifecycle.rs @@ -660,3 +660,116 @@ async fn join_epoch_recorded_on_welcome_join() { "creator records no pre-membership bound" ); } + +/// Seam parity with `replay_buffered_messages` (5ae9a440): the deferred-peel +/// sweep must NOT relabel `Processed` a `PeelDeferred` row that +/// `ingest_group_message` terminalized during the sweep. The reachable case is +/// `SelfEvicted`: a future-epoch commit sits `PeelDeferred`; once our own leaf +/// is removed the group is `!is_active` (still `Stable`, not quarantined), so +/// re-ingesting the deferred row hits the `!is_active` gate, which persists it +/// `Failed`. That ingest-committed verdict is authoritative — sweeping it back +/// to `Processed` would feed a row we were evicted on into canonicalization +/// (`openmls_projection` / `distributed_convergence` select on `Processed`). +#[tokio::test] +async fn deferred_peel_self_evicted_row_stays_failed_not_swept_processed() { + // alice (admin) removes carol; bob keeps the group non-trivial afterwards. + let (mut alice, _alice_storage) = build_client(b"alice-deferred-self-evict"); + let (mut bob, _bob_storage) = build_client(b"bob-deferred-self-evict"); + let (mut carol, carol_storage, _carol_peeler) = + build_counting_client(b"carol-deferred-self-evict"); + let carol_id = carol.self_id().clone(); + + let bob_kp = bob.fresh_key_package().await.unwrap(); + let carol_kp = carol.fresh_key_package().await.unwrap(); + let (group_id, create) = alice + .create_group(CreateGroupRequest { + name: "deferred-self-evict".into(), + description: "".into(), + members: vec![bob_kp, carol_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![alice.self_id()], + }) + .await + .unwrap(); + let (pending, welcomes) = match create { + SendResult::GroupCreated { pending, welcomes } => (pending, welcomes), + other => panic!("expected GroupCreated, got {other:?}"), + }; + alice.confirm_published(pending).await.unwrap(); + bob.join_welcome(welcome_for(&welcomes, b"bob-deferred-self-evict")) + .await + .unwrap(); + carol + .join_welcome(welcome_for(&welcomes, b"carol-deferred-self-evict")) + .await + .unwrap(); + carol.drain_events(); + assert_eq!(carol.epoch(&group_id).unwrap(), EpochId(1)); + + // Alice removes carol at epoch 1 (source epoch 1) → advances to epoch 2. + // Carol can peel & apply this at epoch 1; hold it back for now. + let (msg, pending) = evolution( + alice + .send(SendIntent::RemoveMembers { + group_id: group_id.clone(), + members: vec![carol_id.clone()], + }) + .await + .unwrap(), + ); + alice.confirm_published(pending).await.unwrap(); + let remove_carol = route(msg, &group_id); + + // Alice, now at epoch 2, produces a future commit (source epoch 2) that + // carol at epoch 1 cannot peel — it is retained `PeelDeferred`. + let (msg, pending) = evolution( + alice + .send(SendIntent::UpdateGroupData { + group_id: group_id.clone(), + name: Some("post-remove".into()), + description: None, + }) + .await + .unwrap(), + ); + alice.confirm_published(pending).await.unwrap(); + let future_commit = route(msg, &group_id); + + // Carol (epoch 1) ingests the future commit: undecryptable → PeelDeferred. + assert!(matches!( + carol.ingest(future_commit.clone()).await.unwrap(), + IngestOutcome::Stale { + reason: StaleReason::PeelFailed + } + )); + assert_eq!( + carol_storage.get_message(&future_commit.id).unwrap().state, + MessageState::PeelDeferred, + "the future-epoch commit is retained pending a later peel" + ); + + // Carol applies the removal: her leaf is gone and the group goes inactive + // (still `Stable`, not quarantined), so the deferred row will re-ingest + // against an inactive group. + carol.ingest(remove_carol).await.unwrap(); + assert!( + !carol + .members(&group_id) + .unwrap() + .iter() + .any(|member| member.id == carol_id), + "carol must have been evicted by the removal" + ); + + // The deferred future-commit row re-ingests against the inactive group → + // `SelfEvicted`, so ingest persists it `Failed`. Regression: the sweep must + // not clobber that `Failed` back to `Processed`. + carol.retry_deferred_peels(&group_id).await.unwrap(); + assert_eq!( + carol_storage.get_message(&future_commit.id).unwrap().state, + MessageState::Failed, + "a row we were evicted on must stay Failed after the deferred-peel \ + sweep, not be swept into canonicalization as Processed" + ); +}