Skip to content

fix(cgka-engine): classify malformed group messages as terminal stale instead of aborting ingest #893

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jgmontoya wants to merge 5 commits into
base: master
Choose a base branch
from
Open

fix(cgka-engine): classify malformed group messages as terminal stale instead of aborting ingest #893

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
7dd6659
fix(cgka-engine): classify malformed group messages as terminal stale…
jgmontoya Jul 16, 2026
f7fb509
refactor(cgka-engine): move raw-retry-row retirement into store.rs pe…
jgmontoya Jul 17, 2026
df32c33
fix(cgka-engine): retire malformed snapshot-fallback peels as termina…
jgmontoya Jul 17, 2026
b2451d5
fix(cgka-engine): retire non-deferred raw rows on terminal replay out…
jgmontoya Jul 17, 2026
28cb0c8
fix(cgka-engine): guard retry_deferred_peels retirement against clobb…
jgmontoya Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
116 changes: 82 additions & 34 deletions crates/cgka-engine/src/message_processor/ingest.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -254,14 +254,33 @@ impl<S: StorageProvider> Engine<S> {
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 {
Expand Down Expand Up @@ -329,14 +348,33 @@ impl<S: StorageProvider> Engine<S> {
}
}
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 {
Expand All @@ -357,7 +395,7 @@ impl<S: StorageProvider> Engine<S> {
// (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",
)?;
Expand All @@ -380,6 +418,24 @@ impl<S: StorageProvider> Engine<S> {
});
}
}
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.
return self.malformed_terminal_stale(
&raw_msg_id,
&msg.id,
"malformed_payload",
);
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Err(e) => return Err(EngineError::Peeler(e)),
};
let mls_bytes = match peeled.content {
Expand All @@ -388,7 +444,7 @@ impl<S: StorageProvider> Engine<S> {
// 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",
)?;
Expand Down Expand Up @@ -461,7 +517,7 @@ impl<S: StorageProvider> Engine<S> {
// 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",
)?;
Expand Down Expand Up @@ -498,7 +554,7 @@ impl<S: StorageProvider> Engine<S> {
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,
});
Expand Down Expand Up @@ -577,7 +633,7 @@ impl<S: StorageProvider> Engine<S> {
.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)?;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
return Ok(IngestOutcome::Stale {
reason: StaleReason::PeelFailed,
});
Expand Down Expand Up @@ -717,7 +773,7 @@ impl<S: StorageProvider> Engine<S> {
// 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",
)?;
Expand All @@ -737,7 +793,7 @@ impl<S: StorageProvider> Engine<S> {
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",
)?;
Expand Down Expand Up @@ -1572,31 +1628,23 @@ impl<S: StorageProvider> Engine<S> {
Ok(())
}

fn mark_raw_transport_message_failed_if_deferred(
/// 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,
reason: &str,
) -> Result<(), EngineError> {
match self.storage.get_message(raw_msg_id) {
Ok(record) if record.state == MessageState::PeelDeferred => {
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,
),
);
self.note_peel_deferred_row_retired(&record.group_id, raw_msg_id);
Ok(())
}
Ok(_) | Err(StorageError::NotFound) => Ok(()),
Err(err) => Err(EngineError::Storage(err)),
}
msg_id: &MessageId,
reason: &'static str,
) -> Result<IngestOutcome, EngineError> {
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`
Expand Down
81 changes: 66 additions & 15 deletions crates/cgka-engine/src/message_processor/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -593,15 +593,38 @@ impl<S: StorageProvider> Engine<S> {
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;
}
Expand Down Expand Up @@ -880,21 +903,49 @@ impl<S: StorageProvider> Engine<S> {
}
}
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 deferred. A terminal-after-peel `PeelFailed` already
// retired its raw deferred row inside `ingest_group_message`
// via `mark_raw_transport_message_failed_if_deferred`.
// 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.
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
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);
}
}
Expand Down
Loading