fix(engine): retire a terminal group's deferred-peel rows so they stop charging the account budget (#1784)

* fix(engine): retire a terminal group's deferred-peel rows so they stop charging the account budget

When this device's copy of a group became terminal, its PeelDeferred rows
were never retired. The sweep that releases such rows is reached only
through prepare_convergence_input_advance, which refuses a removed copy
outright and a disbanded one via EpochState::Disbanded, while
ensure_peel_deferred_usage_initialized re-counts every group's rows into
the deferred-peel account budget on each open. Terminal groups therefore
held their per-group row slots and their share of the account byte
budget forever; a disbanded group also showed pending deferred work in
conformance snapshots beside zeroed unresolved inputs.

Retire the rows silently at every marker site: inside the disband settle
transaction beside delete_deferred_peel_generation (its idempotent
re-entry never re-runs the body, so a purge outside it could be skipped
forever after a crash), at realize_self_eviction, at the commit-apply
self-removed arm (realize_self_eviction early-returns once removed is
written, so that seam owns its own purge), at the convergence-reorg
marker, and once more in the terminal gate as crash-window recovery. The
durable flip enumerates payload-free metadata and runs inside the
transaction; the in-memory budget release runs after commit, safe because
the counters are derived state rebuilt from durable rows on open. Rows go
to Failed like every other retired deferred row, and nothing on this path
emits TransportObjectResourceRefused.

Consolidate the engine's hand-rolled terminal predicates onto
Group::is_terminal where the predicate is genuinely either kind of
terminal; the send gates stay on removed because the tombstone gate above
each already refuses a disbanded copy and their error names removal.

Tests drive every behavior through ingest and advance_convergence. The
regression guard for a row terminalized during a sweep is re-pointed at a
live-group shape so it enumerates rows again.

* fix(engine): flip a terminal group's deferred rows in one transaction

Outside a transaction each update_message_state autocommitted, so a
storage error mid-loop left earlier rows durably Failed with their budget
slot still charged and no transition audit, and a retry that enumerates
PeelDeferred only never saw them. Run the durable half of the retire in
one transaction; a failed retire now leaves every row PeelDeferred for
the next pass. Nesting is safe: the SQLite backend reuses a same-thread
outer transaction, so the disband settle is unaffected.

Pinned with a FaultStorage test that fails the second Failed-state write
during a removal and asserts both rows stay deferred, then retires both
on the next advance.

Author: jgmontoya

crates/cgka-engine/src/disband.rs

@@ -15,6 +15,7 @@ use cgka_traits::app_components::{
 use cgka_traits::engine::SendResult;
 use cgka_traits::engine_state::{EpochState, StagedCommitHandle};
 use cgka_traits::error::EngineError;
+use cgka_traits::message::DeferredMessageMetadata;
 use cgka_traits::storage::{
     DisbandCandidate, DisbandFailureReason, DisbandRequest, DisbandRequestStatus, StorageError,
     StorageProvider,
@@ -545,8 +546,8 @@ impl<S: StorageProvider> Engine<S> {
             announced: false,
         };
 
-        self.storage
-            .with_transaction(|storage| -> Result<(), EngineError> {
+        let retired_deferred_rows = self.storage.with_transaction(
+            |storage| -> Result<Vec<DeferredMessageMetadata>, EngineError> {
                 storage.put_disband_tombstone(group_id, &tombstone)?;
                 let mut group = storage.get_group(group_id)?;
                 group.epoch = epoch;
@@ -573,6 +574,16 @@ impl<S: StorageProvider> Engine<S> {
                 }
                 storage.delete_convergence_pass(group_id)?;
                 storage.delete_deferred_peel_generation(group_id)?;
+                // The generation barrier is gone and the rows it tracked
+                // must go with it (see
+                // `Engine::retire_deferred_peel_rows_for_terminal_group`).
+                // Durably retired on this transaction, not after it: the early
+                // return above makes a re-entry after a crash skip this body
+                // forever.
+                let retired_deferred_rows =
+                    crate::message_processor::fail_deferred_peel_rows_in_terminal_group(
+                        storage, group_id,
+                    )?;
                 for snapshot in storage.list_group_snapshots(group_id)? {
                     storage.release_group_snapshot(group_id, &snapshot)?;
                 }
@@ -597,8 +608,10 @@ impl<S: StorageProvider> Engine<S> {
                 mls_group.delete(tx_provider.storage()).map_err(|error| {
                     EngineError::Backend(format!("delete MLS group: {error:?}"))
                 })?;
-                Ok(())
-            })?;
+                Ok(retired_deferred_rows)
+            },
+        )?;
+        self.release_retired_deferred_peel_rows(&retired_deferred_rows);
 
         self.transport_group_id_index
             .retain(|_, mapped_group| mapped_group != group_id);

crates/cgka-engine/src/distributed_convergence.rs

@@ -1816,6 +1816,11 @@ impl<S: StorageProvider> Engine<S> {
                 // removed-copy send gate.
                 self.discard_queued_outbound_intents_for_removed_group(group_id)
                     .map_err(|e| OpenMlsProjectionError::Storage(format!("{e:?}")))?;
+                // Same for retained inbound rows: see
+                // `retire_deferred_peel_rows_for_terminal_group` for why no
+                // later sweep can reach them.
+                self.retire_deferred_peel_rows_for_terminal_group(group_id)
+                    .map_err(|e| OpenMlsProjectionError::Storage(format!("{e:?}")))?;
             }
             self.push_group_state_change(
                 group_id,

crates/cgka-engine/src/message_processor/ingest.rs

@@ -1599,6 +1599,13 @@ impl<S: StorageProvider> Engine<S> {
                     // intents so later drains do not re-fail them forever
                     // against the removed-copy send gate.
                     self.discard_queued_outbound_intents_for_removed_group(&group_id)?;
+                    // And retire the deferred-peel backlog (see
+                    // `retire_deferred_peel_rows_for_terminal_group`). This
+                    // seam owns that here rather than deferring to
+                    // `realize_self_eviction`: the transaction above already
+                    // wrote `removed`, so a later realization early-returns
+                    // without ever reaching it.
+                    self.retire_deferred_peel_rows_for_terminal_group(&group_id)?;
                 } else if after_ids.contains(self.identity.self_id()) {
                     if self.load_leave_request_state(&group_id)?.is_some() {
                         // A SelfRemove proposal is valid only in its
@@ -2246,6 +2253,10 @@ impl<S: StorageProvider> Engine<S> {
         // leaving them to re-fail through the removed-copy send gate on every
         // later drain.
         self.discard_queued_outbound_intents_for_removed_group(group_id)?;
+        // Same for retained inbound work: see
+        // `retire_deferred_peel_rows_for_terminal_group` for why no later
+        // sweep can reach these rows.
+        self.retire_deferred_peel_rows_for_terminal_group(group_id)?;
         // Deliberately LAST, after the marker write — not before it like the
         // convergence path (which has no attribution read). The notification
         // is already enqueued above, so a failure here cannot lose it; it only

crates/cgka-engine/src/message_processor/mod.rs

@@ -14,6 +14,7 @@ mod store;
 
 pub(crate) use ingest::avatar_component_snapshot;
 pub(crate) use send::merge_capabilities;
+pub(crate) use store::fail_deferred_peel_rows_in_terminal_group;
 #[cfg(feature = "test-conformance-snapshot")]
 pub(crate) use store::normalized_deferred_peel_lifecycle;
 pub(crate) use store::transition_staged_invite_welcomes;
@@ -752,7 +753,8 @@ impl<S: StorageProvider> Engine<S> {
         // Terminal gate before queueing: a local copy marked removed (realized
         // self-eviction) must never accept or queue outbound work. Checked
         // again in `do_send_ready` so queued-intent drains for a copy removed
-        // after queueing hit the same deterministic error.
+        // after queueing hit the same deterministic error. Disband reaches the
+        // tombstone gate above first, so the message below stays accurate.
         if group.as_ref().is_some_and(|group| group.removed) {
             return Err(EngineError::InvalidTransition(
                 cgka_traits::engine_state::InvalidTransition {
@@ -965,14 +967,19 @@ impl<S: StorageProvider> Engine<S> {
         if self.sync_unrecoverable_halt_from_record(group_id, group.as_ref()) {
             return Ok(false);
         }
-        // Terminal: a removed copy must never publish, and the removed-copy
+        // Terminal: a terminal copy must never publish, and the terminal-copy
         // gate in `do_send_ready` would turn every queued record into a
         // permanent drain error that the app retries forever. Discard the
-        // queue and report nothing to drain. This is the defense-in-depth
-        // side; the marker sites (realization, commit-apply seam, convergence
-        // reorg) also purge at the moment the copy becomes removed.
-        if group.is_some_and(|group| group.removed) {
+        // queue, retire the deferred-peel backlog, and report nothing to
+        // drain. This gate is also what keeps the sweep away from a *removed*
+        // copy, which stays `Stable` — so it owes the retirement itself, and
+        // it is the only recovery if a crash lands between a marker site's
+        // record write and its own purge. The marker sites (realization,
+        // commit-apply seam, convergence reorg, disband settle) purge both at
+        // the moment the copy becomes terminal.
+        if group.is_some_and(|group| group.is_terminal()) {
             self.discard_queued_outbound_intents_for_removed_group(group_id)?;
+            self.retire_deferred_peel_rows_for_terminal_group(group_id)?;
             return Ok(false);
         }
         if let Some(state) = self.epoch_manager.state(group_id)
@@ -1655,7 +1662,7 @@ impl<S: StorageProvider> Engine<S> {
         if delay.is_some()
             && self
                 .stored_group_record(group_id)?
-                .is_none_or(|group| group.removed)
+                .is_none_or(|group| group.is_terminal())
         {
             // Input-only convergence can realize our eviction while fanout
             // blocks the outbound drain. Removed copies exit that drain before
@@ -2921,10 +2928,21 @@ impl<S: StorageProvider> Engine<S> {
     /// cap-rejection audit re-arms so a fresh cap-full episode is recorded
     /// once more.
     pub(crate) fn note_peel_deferred_row_retired(&mut self, record: &MessageRecord) {
+        self.note_peel_deferred_row_retired_by_id(&record.group_id, &record.id);
+    }
+
+    /// [`Self::note_peel_deferred_row_retired`] keyed by the only two fields
+    /// it reads, for callers holding metadata rather than a whole record (the
+    /// payload bytes come from `deferred_payload_bytes_by_id`, not the row).
+    pub(crate) fn note_peel_deferred_row_retired_by_id(
+        &mut self,
+        group_id: &GroupId,
+        id: &MessageId,
+    ) {
         let account_was_full = self.deferred_peel_account.counted
             && self.deferred_peel_account.bytes >= self.deferred_peel_account_byte_limit;
-        let local_reopened = if let Some(state) = self.deferred_peel.get_mut(&record.group_id) {
-            let Some(payload_bytes) = state.deferred_payload_bytes_by_id.remove(&record.id) else {
+        let local_reopened = if let Some(state) = self.deferred_peel.get_mut(group_id) {
+            let Some(payload_bytes) = state.deferred_payload_bytes_by_id.remove(id) else {
                 return;
             };
             state.deferred_rows = state.deferred_rows.saturating_sub(1);
@@ -2950,7 +2968,7 @@ impl<S: StorageProvider> Engine<S> {
             for state in self.deferred_peel.values_mut() {
                 state.cap_rejection_audited = false;
             }
-        } else if local_reopened && let Some(state) = self.deferred_peel.get_mut(&record.group_id) {
+        } else if local_reopened && let Some(state) = self.deferred_peel.get_mut(group_id) {
             state.cap_rejection_audited = false;
         }
     }

crates/cgka-engine/src/message_processor/send.rs

@@ -54,7 +54,9 @@ impl<S: StorageProvider> Engine<S> {
         // fail with an opaque `UseAfterEviction` backend error anyway. Gate
         // here (not just `do_send`) so queued-intent drains hit the same
         // deterministic terminal error. Every intent kind is blocked,
-        // including `Leave` — there is nothing left to leave.
+        // including `Leave` — there is nothing left to leave. Disband reaches
+        // the tombstone gate above first, so the message below stays
+        // accurate.
         let group = self.stored_group_record(&group_id)?;
         if group.as_ref().is_some_and(|group| group.removed) {
             return Err(EngineError::InvalidTransition(

crates/cgka-engine/src/message_processor/store.rs

@@ -32,6 +32,39 @@ fn fresh_deferred_peel_lifecycle(
     }
 }
 
+/// Durable half of `Engine::retire_deferred_peel_rows_for_terminal_group`:
+/// flip every `PeelDeferred` row of a terminal group to `Failed` and hand back
+/// what was retired, for the caller's in-memory reconciliation. Takes `&S` so
+/// it runs inside the caller's storage transaction.
+///
+/// All rows flip or none do, and that is a contract on the caller: run this
+/// inside a transaction — the disband settle's own, or the one
+/// `Engine::retire_deferred_peel_rows_for_terminal_group` opens. Without one,
+/// each write autocommits, so a mid-loop failure would leave earlier rows
+/// durably `Failed` while the error skips the in-memory release: those rows
+/// would keep their capacity slot with no transition audit, and a retry
+/// enumerates only `PeelDeferred` rows so it would never revisit them. Inside
+/// a transaction a failed retire instead leaves every row `PeelDeferred` for
+/// the next pass, and nothing is ever charged without its audit row.
+///
+/// Enumerates metadata, never payloads: this runs inside the disband write
+/// transaction, where loading up to `MAX_PEEL_DEFERRED_BYTES_PER_GROUP` of
+/// ciphertext nobody reads would be copied and held across the commit.
+#[must_use = "every returned row still owes its in-memory capacity slot; pass \
+              them to Engine::release_retired_deferred_peel_rows once the \
+              durable flip is committed, or the group's deferred-peel budget \
+              stays charged for the rest of this engine incarnation"]
+pub(crate) fn fail_deferred_peel_rows_in_terminal_group<S: StorageProvider>(
+    storage: &S,
+    group_id: &GroupId,
+) -> Result<Vec<DeferredMessageMetadata>, EngineError> {
+    let retired = storage.list_deferred_message_metadata(group_id)?;
+    for row in &retired {
+        storage.update_message_state(&row.id, MessageState::Failed)?;
+    }
+    Ok(retired)
+}
+
 /// Promote or retire the non-deliverable Welcome artifacts produced beside one
 /// staged invite commit.
 ///
@@ -723,6 +756,80 @@ impl<S: StorageProvider> Engine<S> {
         Ok(())
     }
 
+    /// Retire this group's whole `PeelDeferred` backlog because the local copy
+    /// has become terminal — removed or disbanded.
+    ///
+    /// Why nothing else will: the only production driver that reaches the
+    /// deferred-peel sweep is `advance_convergence_inputs`, and its single
+    /// door is `prepare_convergence_input_advance`, whose terminal gate
+    /// refuses a terminal group before any sweep runs. A *disbanded* copy is
+    /// refused twice over (its `EpochState::Disbanded` also fails that
+    /// function's `Stable` check), but a *removed* copy stays `Stable` — the
+    /// terminal gate is the whole reason its rows never come back. (The `pub`
+    /// `retry_deferred_peels` would sweep a removed group happily; no
+    /// production path calls it.) So without this the rows sit forever holding
+    /// their share of the account byte budget — reconstructed from durable
+    /// rows on every open — and their per-group row slots.
+    ///
+    /// Silent by construction: the rows leave the retry lifecycle the way
+    /// [`Self::mark_raw_transport_message_failed_if_awaiting_retry`] retires
+    /// one, with no `TransportObjectResourceRefused` — nothing was refused,
+    /// the group they belonged to is gone.
+    ///
+    /// The tradeoff, taken deliberately: `removed` is reversible (branch
+    /// selection can supersede the removal that set it — see
+    /// `cgka_traits::group::Group::removed` and the heal in
+    /// `distributed_convergence::emit_convergence_events`), and a `Failed` row
+    /// blocks same-id redelivery, so a heal cannot get these rows back. That
+    /// is the same bet
+    /// [`Self::discard_queued_outbound_intents_for_removed_group`] already
+    /// makes beside every call site here, and the window is narrow because
+    /// the terminal gate refuses further advances until the heal lands.
+    pub(crate) fn retire_deferred_peel_rows_for_terminal_group(
+        &mut self,
+        group_id: &GroupId,
+    ) -> Result<(), EngineError> {
+        // One durable unit: see the free function's contract. Nesting is
+        // safe — the SQLite backend reuses a same-thread outer transaction
+        // rather than beginning a second one — so a caller that already holds
+        // one loses nothing by coming through here.
+        let retired = self.storage.with_transaction(|storage| {
+            fail_deferred_peel_rows_in_terminal_group(storage, group_id)
+        })?;
+        self.release_retired_deferred_peel_rows(&retired);
+        Ok(())
+    }
+
+    /// In-memory half of [`Self::retire_deferred_peel_rows_for_terminal_group`]:
+    /// return each retired row's capacity slot and record the transition.
+    ///
+    /// Split from the durable half so a caller already inside a storage
+    /// transaction — `disband::settle_disband_after_convergence`, whose
+    /// idempotent re-entry never re-runs the settle body — can keep the row
+    /// flip inside that transaction and reconcile the engine's own counters
+    /// after it commits. Safe in that order: the counters are derived state,
+    /// rebuilt from the durable rows by
+    /// `ensure_peel_deferred_usage_initialized` on the next open, so a crash
+    /// between the two loses nothing.
+    pub(crate) fn release_retired_deferred_peel_rows(
+        &mut self,
+        retired: &[DeferredMessageMetadata],
+    ) {
+        for row in retired {
+            self.audit_group(
+                &row.group_id,
+                crate::audit_helpers::message_state_transition_event(
+                    hex::encode(row.id.as_slice()),
+                    Some(MessageState::PeelDeferred),
+                    MessageState::Failed,
+                    Some(row.epoch),
+                    "terminal_group",
+                ),
+            );
+            self.note_peel_deferred_row_retired_by_id(&row.group_id, &row.id);
+        }
+    }
+
     pub(crate) fn mark_raw_transport_message_failed_if_awaiting_retry(
         &mut self,
         raw_msg_id: &MessageId,

crates/cgka-engine/src/own_commit_intent.rs

@@ -171,7 +171,7 @@ impl<S: StorageProvider> Engine<S> {
                 None,
             )));
         };
-        if group.removed || group.disbanded.is_some() {
+        if group.is_terminal() {
             return Ok(Some((
                 report(
                     SupersededIntentOutcome::NotMember,
@@ -468,10 +468,7 @@ impl<S: StorageProvider> Engine<S> {
             outcome,
             reason,
         };
-        if group
-            .as_ref()
-            .is_none_or(|group| group.removed || group.disbanded.is_some())
-        {
+        if group.as_ref().is_none_or(|group| group.is_terminal()) {
             self.storage.delete_own_commit_intent(commit_id)?;
             return Ok(Some(report(
                 SupersededIntentOutcome::NotMember,