From 3a2068a7920fb6d672b594ebe72f68bbe17f3d9b Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 23 Apr 2026 01:44:32 +0000 Subject: [PATCH] fix(marmot): buffer + retry UndecryptableOuterLayer events after epoch advance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Test 12 (offline catch-up) was almost passing: the retained-epoch fallback (c88923a3) brought through the 5 epoch-1 application messages the user sent before adding C, but 3 epoch-2 messages + the rename commit still went missing. Only on app restart did they appear — the relay re-delivered and the replayed arrival order happened to put the epoch-2-advancing commit before the epoch-2 events. Root cause: kind:445 events arrive from the relay in subscription order, not epoch order. When B drives an offline-catch-up sequence like msg-1..5 (epoch 1) → add C (epoch 1→2) → msg-6..8 (epoch 2) → rename (epoch 2→3) the receiver can easily see {msg-6, msg-7, msg-8, rename, add-C, msg-1..5}. Everything from the rename onward arrives before the add-C commit, so the outer ChaCha20-Poly1305 layer fails with UndecryptableOuterLayer — we don't have the epoch-2 exporter yet. Quartz correctly reports the failure, but the receiver just logged "likely from before our join" and dropped the event on the floor. Nothing retried when the add-C commit eventually landed and the epoch caught up. Fix it at the GroupEventHandler level, where we can both see the original GroupEvent and hook into every CommitProcessed: - Keep a bounded per-group ArrayDeque of events that failed with UndecryptableOuterLayer. FIFO-evict at 64 entries so a genuinely pre-join stream can't pin unbounded memory. - On every CommitProcessed for a group, drain that group's queue and re-run each event through the normal add() path. A successful retry may itself emit another CommitProcessed and trigger recursive draining — that's fine, epochs are strictly monotonic so recursion depth is bounded by the queue size. - Concurrent-safe: all queue mutation is guarded by a Mutex. Ambient discriminator: we can't tell "epoch too old, from before our join" from "epoch too new, commit hasn't arrived yet" just from UndecryptableOuterLayer — both report it. Buffering both is fine: truly-pre-join events stay buffered until FIFO eviction pushes them out (no correctness problem, just a few dozen bytes of memory for a bounded time). --- .../loggedIn/DecryptAndIndexProcessor.kt | 89 ++++++++++++++++++- 1 file changed, 85 insertions(+), 4 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt index b2a5d834d..b1ab71975 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt @@ -50,6 +50,8 @@ import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallRejectEvent import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallRenegotiateEvent import com.vitorpamplona.quartz.utils.Log import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock class EventProcessor( private val account: Account, @@ -549,6 +551,29 @@ class GroupEventHandler( private val account: Account, private val cache: LocalCache, ) : EventHandler { + /** + * Per-group buffer of kind:445 events whose outer ChaCha20-Poly1305 + * layer couldn't be decrypted yet — typically because they carry an + * epoch the receiver hasn't advanced to. The relay subscription + * doesn't guarantee arrival order, so a run like + * + * 1) B sends msg-1..5 at epoch 1 + * 2) B adds C (commit advances B → epoch 2) + * 3) B sends msg-6..8 at epoch 2 + * 4) B renames (commit advances B → epoch 3) + * + * can reach a receiver in the order {msg-6, msg-7, msg-8, rename, + * add-C, msg-1..5} depending on relay ordering. Everything from step + * 3+ above is `UndecryptableOuterLayer` until the add-C commit lands; + * we re-run those pending events every time an epoch-advancing + * CommitProcessed fires for the same group. + * + * Bounded per-group so a stuck stream of "truly undecryptable" + * pre-join events can't grow unbounded. FIFO eviction on overflow. + */ + private val pendingUndecryptable: MutableMap> = mutableMapOf() + private val pendingMutex = Mutex() + override suspend fun add( event: GroupEvent, eventNote: Note, @@ -642,6 +667,10 @@ class GroupEventHandler( // Sync MIP-01 metadata after epoch advance (extensions may have changed) val chatroom = account.marmotGroupList.getOrCreateGroup(result.groupId) manager.syncMetadataTo(result.groupId, chatroom) + // Epoch just advanced — drain any kind:445 events that + // previously failed as UndecryptableOuterLayer for this + // group. See `pendingUndecryptable` for the scenario. + retryPendingFor(result.groupId, eventNote, publicNote) } is GroupEventResult.CommitPending -> { @@ -655,12 +684,31 @@ class GroupEventHandler( } is GroupEventResult.UndecryptableOuterLayer -> { - // Expected for commits + application messages from epochs - // that predate our join — per MLS forward secrecy we - // never held those keys. Not a bug, not a warning. + // Two common causes for this result: + // (a) event's epoch predates our join — we never held + // those exporter keys, nothing to retry later. + // (b) event's epoch is ahead of our current one — the + // epoch-advancing commit for this group hasn't + // arrived yet but will in a moment (relays don't + // guarantee arrival order across a kind:445 + // subscription). We can't tell (a) from (b) here, + // so buffer the event and retry after any future + // CommitProcessed for this group advances us. + // + // Bounded per-group: keep the most recent + // MAX_PENDING_PER_GROUP events, FIFO-evict on overflow. + // Prevents a flood of genuinely pre-join events from + // pinning unbounded memory. Log.d("MarmotDbg") { "GroupEventHandler.add: undecryptable outer layer for group=${result.groupId.take(8)}… " + - "(current + ${result.retainedEpochCount} retained epoch key(s) tried) — likely from before our join" + "(current + ${result.retainedEpochCount} retained epoch key(s) tried) — buffering for post-commit retry" + } + pendingMutex.withLock { + val queue = pendingUndecryptable.getOrPut(result.groupId) { ArrayDeque() } + if (queue.size >= MAX_PENDING_PER_GROUP) { + queue.removeFirst() + } + queue.addLast(event) } } @@ -673,4 +721,37 @@ class GroupEventHandler( Log.e("MarmotDbg", "GroupEventHandler.add: exception processing kind:445", e) } } + + /** + * Drain the pending-undecryptable queue for [groupId] and re-run each + * event through `add`. A successful retry may itself emit another + * CommitProcessed and trigger recursive draining — that's fine, MLS + * epoch advances are strictly monotonic so recursion depth is bounded + * by the number of queued events. + * + * `eventNote` / `publicNote` are re-used for the retry path; they only + * feed into relay-activity tracking and compose-note lookup, both of + * which are safe to re-fire with the triggering event's notes. + */ + private suspend fun retryPendingFor( + groupId: String, + eventNote: Note, + publicNote: Note, + ) { + val drained = + pendingMutex.withLock { + pendingUndecryptable.remove(groupId)?.toList() ?: return + } + if (drained.isEmpty()) return + Log.d("MarmotDbg") { + "GroupEventHandler.retryPendingFor: replaying ${drained.size} previously-undecryptable kind:445 event(s) for group=${groupId.take(8)}…" + } + for (pending in drained) { + add(pending, eventNote, publicNote) + } + } + + companion object { + private const val MAX_PENDING_PER_GROUP = 64 + } }