nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(marmot): buffer + retry UndecryptableOuterLayer events after epoch advance

Browse Source 
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<GroupEvent> 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).
This commit is contained in:
Claude
2026-04-23 01:44:32 +00:00
parent ab064d3cd3
commit 3a2068a792
1 changed files with 85 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt
+85 -4
View File
@@ -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<GroupEvent> {
/**
* 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<String, ArrayDeque<GroupEvent>> = 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
}
}