fix: stop refetching the full MLS kind:445 backlog on every restart

The Marmot subscription since, the processed-event dedup set, and the application ratchet position (group state persists only at commits) are all in-memory only. On restart, relays therefore redeliver the group's entire kind:445 history and the rewound ratchet re-decrypts old application messages as if they had just arrived — wasted decryption work and, when a replay beats the disk restore, duplicate entries appended to the persisted plaintext message log. Two defenses: - MarmotManager.restoreAll() now seeds each restored group's subscription since from the newest persisted decrypted message, minus a one-day overlap window for late/out-of-order publishes. Seeding happens before syncWithGroupManager registers default entries, so even the first filter set sent to relays carries it. The CLI is unaffected: it builds group filters from its own persisted since. - MarmotMessageStore appends are now explicitly idempotent (contract was previously ambiguous and both real stores appended blindly): the Android and CLI file stores skip an entry that is already in the group's log, so replays inside the overlap window cannot grow it. Covered by MarmotManagerRestoreTest in commons jvmTest — placed there rather than androidHostTest because CI only runs :commons:jvmTest (the androidHostTest task currently fails on android.util.Log stubs even for the pre-existing Marmot test).
2026-08-12 09:13:23 +00:00 · 2026-06-10 23:03:48 +00:00
parent af7d61f361
commit 2eb6510eec
6 changed files with 267 additions and 5 deletions
@@ -78,6 +78,10 @@ class AndroidMarmotMessageStore(
        writeMutex.withLock {
            try {
                val existing = readAll(nostrGroupId).toMutableList()
+                if (innerEventJson in existing) {
+                    Log.d(TAG) { "appendMessage($nostrGroupId): duplicate entry skipped" }
+                    return@withLock
+                }
                existing.add(innerEventJson)
                writeAll(nostrGroupId, existing)
                Log.d(TAG) {
@@ -140,9 +140,12 @@ class FileMarmotMessageStore(
        nostrGroupId: String,
        innerEventJson: String,
    ) {
-        // Each line is one inner event JSON. The store doc tolerates duplicates
-        // so we don't bother deduping here — readers can do it.
-        SecureFileIO.appendText(file(nostrGroupId), innerEventJson.replace("\n", " ") + "\n")
+        // Each line is one inner event JSON. Appends must be idempotent:
+        // post-restart relay replays re-decrypt already-persisted messages.
+        val line = innerEventJson.replace("\n", " ")
+        val target = file(nostrGroupId)
+        if (target.exists() && target.readLines().any { it == line }) return
+        SecureFileIO.appendText(target, line + "\n")
    }

    override suspend fun loadMessages(nostrGroupId: String): List<String> = file(nostrGroupId).takeIf { it.exists() }?.readLines()?.filter { it.isNotBlank() } ?: emptyList()
@@ -279,7 +279,8 @@ private class InMemoryMarmotMessageStore : MarmotMessageStore {
        nostrGroupId: String,
        innerEventJson: String,
    ) {
-        messages.getOrPut(nostrGroupId) { mutableListOf() }.add(innerEventJson)
+        val log = messages.getOrPut(nostrGroupId) { mutableListOf() }
+        if (innerEventJson !in log) log.add(innerEventJson)
    }

    override suspend fun loadMessages(nostrGroupId: String): List<String> = messages[nostrGroupId]?.toList() ?: emptyList()
@@ -88,6 +88,13 @@ class MarmotManager(
        try {
            groupManager.restoreAll()
            val activeIds = groupManager.activeGroupIds()
+            // Register restored groups with a seeded `since` BEFORE
+            // syncWithGroupManager fills in default (since = null) entries,
+            // so even the first filter set sent to relays skips the
+            // already-processed kind:445 backlog.
+            subscriptionSinceFromStoredMessages(activeIds).forEach { (groupId, since) ->
+                subscriptionManager.subscribeGroup(groupId, since)
+            }
            subscriptionManager.syncWithGroupManager(activeIds)
            // Also restore previously-published KeyPackage bundles so that
            // Welcomes referencing them remain processable across restarts.
@@ -98,6 +105,41 @@ class MarmotManager(
        }
    }

+    /**
+     * Computes a per-group kind:445 subscription `since` from the newest
+     * persisted decrypted message of each group.
+     *
+     * Subscription state does not survive a restart, and neither does the
+     * application ratchet position (group state is persisted only at
+     * commits). Without a `since`, every restart re-downloads the group's
+     * full kind:445 backlog, and the rewound ratchet re-decrypts old
+     * application messages as if they had just arrived.
+     *
+     * [GROUP_EVENT_REFETCH_OVERLAP_SEC] of overlap is kept so late or
+     * out-of-order events published shortly before the newest stored message
+     * are still fetched; replays inside the window are deduplicated by the
+     * message store and by note identity in the chatroom.
+     */
+    private suspend fun subscriptionSinceFromStoredMessages(groupIds: Set<HexKey>): Map<HexKey, Long> {
+        if (messageStore == null) return emptyMap()
+        val result = mutableMapOf<HexKey, Long>()
+        for (groupId in groupIds) {
+            val newest =
+                loadStoredMessages(groupId).maxOfOrNull { json ->
+                    try {
+                        Event.fromJson(json).createdAt
+                    } catch (e: Exception) {
+                        Log.w("MarmotManager", "Unparseable persisted message for $groupId: ${e.message}", e)
+                        0L
+                    }
+                } ?: continue
+            if (newest > GROUP_EVENT_REFETCH_OVERLAP_SEC) {
+                result[groupId] = newest - GROUP_EVENT_REFETCH_OVERLAP_SEC
+            }
+        }
+        return result
+    }
+
    // --- Inbound Processing ---

    /**
@@ -704,6 +746,17 @@ class MarmotManager(
                "(leafs=${members.map { it.leafIndex }})"
        }
    }
+
+    companion object {
+        /**
+         * Overlap window (seconds) subtracted from the newest persisted
+         * message's createdAt when seeding a restored group's kind:445
+         * subscription `since`. One day is generous: inner and outer events
+         * are timestamped at the same send, so the window only needs to
+         * absorb relay/system clock skew and out-of-order publishes.
+         */
+        internal const val GROUP_EVENT_REFETCH_OVERLAP_SEC = 24L * 60 * 60
+    }
 }

 /**
@@ -0,0 +1,196 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.marmot
+
+import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageBundleStore
+import com.vitorpamplona.quartz.marmot.mip01Groups.MarmotGroupData
+import com.vitorpamplona.quartz.marmot.mls.group.MarmotMessageStore
+import com.vitorpamplona.quartz.marmot.mls.group.MlsGroupStateStore
+import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
+import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
+import kotlinx.coroutines.runBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+
+/**
+ * Restart behavior of [MarmotManager.restoreAll].
+ *
+ * The kind:445 subscription state and the application ratchet position are
+ * both in-memory only (group state persists only at commits), so without a
+ * seeded `since` every restart re-downloads the full kind:445 backlog and
+ * re-decrypts old application messages as if they were new. These tests pin
+ * down the two defenses:
+ *
+ *   1. `restoreAll` seeds each restored group's subscription `since` from
+ *      the newest persisted decrypted message, minus the refetch overlap.
+ *   2. [MarmotMessageStore.appendMessage] is idempotent, so replays that do
+ *      slip through the overlap window cannot grow the on-disk log.
+ */
+class MarmotManagerRestoreTest {
+    private val nostrGroupId = "b".repeat(64)
+    private val relay =
+        RelayUrlNormalizer.normalizeOrNull("wss://example.invalid/")
+            ?: error("test relay must normalize")
+
+    @Test
+    fun testRestoreAllSeedsSubscriptionSinceFromStoredMessages() {
+        runBlocking {
+            val signer = NostrSignerInternal(KeyPair())
+            val mlsStore = InMemoryStateStore()
+            val messageStore = InMemoryMessageStore()
+            val kpStore = InMemoryBundleStore()
+
+            val manager = MarmotManager(signer, mlsStore, messageStore, kpStore)
+            manager.createGroup(
+                nostrGroupId,
+                MarmotGroupData(
+                    nostrGroupId = nostrGroupId,
+                    name = "restore-since",
+                    relays = listOf(relay.url),
+                ),
+            )
+            val bundle = manager.buildTextMessage(nostrGroupId, "survives restart")
+
+            // Simulate an app restart: fresh manager over the same stores.
+            val restarted = MarmotManager(signer, mlsStore, messageStore, kpStore)
+            restarted.restoreAll()
+
+            val filter = restarted.subscriptionManager.activeGroupFilters().single()
+            assertEquals(
+                bundle.innerEvent.createdAt - MarmotManager.GROUP_EVENT_REFETCH_OVERLAP_SEC,
+                filter.since,
+                "restored kind:445 filter must resume just behind the newest persisted message",
+            )
+        }
+    }
+
+    @Test
+    fun testRestoreAllLeavesSinceUnsetWithoutStoredMessages() {
+        runBlocking {
+            val signer = NostrSignerInternal(KeyPair())
+            val mlsStore = InMemoryStateStore()
+            val messageStore = InMemoryMessageStore()
+            val kpStore = InMemoryBundleStore()
+
+            val manager = MarmotManager(signer, mlsStore, messageStore, kpStore)
+            manager.createGroup(
+                nostrGroupId,
+                MarmotGroupData(
+                    nostrGroupId = nostrGroupId,
+                    name = "restore-no-messages",
+                    relays = listOf(relay.url),
+                ),
+            )
+
+            val restarted = MarmotManager(signer, mlsStore, messageStore, kpStore)
+            restarted.restoreAll()
+
+            val filter = restarted.subscriptionManager.activeGroupFilters().single()
+            assertNull(filter.since, "a group with no persisted messages must fetch its full history")
+        }
+    }
+
+    @Test
+    fun testPersistDecryptedMessageIsIdempotent() {
+        runBlocking {
+            val signer = NostrSignerInternal(KeyPair())
+            val manager =
+                MarmotManager(signer, InMemoryStateStore(), InMemoryMessageStore(), InMemoryBundleStore())
+
+            val json = """{"id":"00","kind":9,"content":"hi","created_at":1700000000}"""
+            manager.persistDecryptedMessage(nostrGroupId, json)
+            manager.persistDecryptedMessage(nostrGroupId, json)
+
+            assertEquals(
+                1,
+                manager.loadStoredMessages(nostrGroupId).size,
+                "replaying an already-persisted message must not grow the log",
+            )
+        }
+    }
+}
+
+// Minimal in-memory stores; the file-backed implementations live in the
+// platform modules (amethyst, cli) and follow the same contracts.
+
+private class InMemoryStateStore : MlsGroupStateStore {
+    private val states = mutableMapOf<String, ByteArray>()
+    private val retained = mutableMapOf<String, List<ByteArray>>()
+
+    override suspend fun save(
+        nostrGroupId: String,
+        state: ByteArray,
+    ) {
+        states[nostrGroupId] = state
+    }
+
+    override suspend fun load(nostrGroupId: String): ByteArray? = states[nostrGroupId]
+
+    override suspend fun delete(nostrGroupId: String) {
+        states.remove(nostrGroupId)
+        retained.remove(nostrGroupId)
+    }
+
+    override suspend fun listGroups(): List<String> = states.keys.toList()
+
+    override suspend fun saveRetainedEpochs(
+        nostrGroupId: String,
+        retainedSecrets: List<ByteArray>,
+    ) {
+        retained[nostrGroupId] = retainedSecrets
+    }
+
+    override suspend fun loadRetainedEpochs(nostrGroupId: String): List<ByteArray> = retained[nostrGroupId] ?: emptyList()
+}
+
+private class InMemoryMessageStore : MarmotMessageStore {
+    private val messages = mutableMapOf<String, MutableList<String>>()
+
+    override suspend fun appendMessage(
+        nostrGroupId: String,
+        innerEventJson: String,
+    ) {
+        val log = messages.getOrPut(nostrGroupId) { mutableListOf() }
+        if (innerEventJson !in log) log.add(innerEventJson)
+    }
+
+    override suspend fun loadMessages(nostrGroupId: String): List<String> = messages[nostrGroupId]?.toList() ?: emptyList()
+
+    override suspend fun delete(nostrGroupId: String) {
+        messages.remove(nostrGroupId)
+    }
+}
+
+private class InMemoryBundleStore : KeyPackageBundleStore {
+    private var snapshot: ByteArray? = null
+
+    override suspend fun save(snapshot: ByteArray) {
+        this.snapshot = snapshot
+    }
+
+    override suspend fun load(): ByteArray? = snapshot
+
+    override suspend fun delete() {
+        snapshot = null
+    }
+}
@@ -39,7 +39,12 @@ package com.vitorpamplona.quartz.marmot.mls.group
 interface MarmotMessageStore {
    /**
     * Append a decrypted inner event JSON to the group's persisted message log.
-     * Implementations should be tolerant to duplicate appends.
+     *
+     * Appends MUST be idempotent: appending a JSON string that is already in
+     * the group's log is a no-op. After a restart the MLS ratchet rewinds to
+     * the last persisted commit, so relays replaying recent kind:445 events
+     * can re-decrypt — and re-persist — messages already captured in a
+     * previous session; without dedup the log grows on every restart.
     *
     * @param nostrGroupId hex-encoded Nostr group ID
     * @param innerEventJson the decrypted inner Nostr event JSON (e.g., kind:9 chat)