From ef19b0b13d271608079ef988629ea2a312636731 Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Tue, 3 Nov 2026 00:24:28 +0200 Subject: [PATCH] fix(sync): ask a store only for keys it can hold --- .../io/github/aedev/flow/sync2/lan/LanPeer.kt | 16 ++++---- .../aedev/flow/sync2/store/GroupStore.kt | 7 +--- .../io/github/aedev/flow/sync2/store/Store.kt | 23 ++++++++++++ .../github/aedev/flow/sync2/HeldStoreTest.kt | 37 +++++++++++++++++++ 4 files changed, 69 insertions(+), 14 deletions(-) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanPeer.kt b/app/src/main/java/io/github/aedev/flow/sync2/lan/LanPeer.kt index 41ef41190..a66e11739 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/lan/LanPeer.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/lan/LanPeer.kt @@ -11,6 +11,7 @@ import io.github.aedev.flow.sync2.roster.parseEntry import io.github.aedev.flow.sync2.store.LocalStore import io.github.aedev.flow.sync2.store.StorePaths import io.github.aedev.flow.sync2.store.checkSeq +import io.github.aedev.flow.sync2.store.keyNames import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job @@ -213,15 +214,12 @@ class LanPeer( .mapNotNull { runCatching { parseEntry(it) }.getOrNull() }, ) } - val epochs = local.rosterAfter(-1).mapNotNull { it.epoch?.n } - for (epoch in epochs) { - for (name in listOf(session.me, "chain")) { - if (local.key(epoch, name) != null) continue - request( - "key", - mapOf("epoch" to JsonPrimitive(epoch), "name" to JsonPrimitive(name)), - )?.bytes?.let { local.putKey(epoch, name, it) } - } + for ((epoch, name) in keyNames(local.rosterAfter(-1), session.me)) { + if (local.key(epoch, name) != null) continue + request( + "key", + mapOf("epoch" to JsonPrimitive(epoch), "name" to JsonPrimitive(name)), + )?.bytes?.let { local.putKey(epoch, name, it) } } val vv = JsonObject(heldVersions().mapValues { (_, seq) -> JsonPrimitive(seq) }) val listed = request("list", mapOf("vv" to vv)) ?: return diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/GroupStore.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/GroupStore.kt index 93ed3ef2a..63daae294 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/GroupStore.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/GroupStore.kt @@ -67,11 +67,8 @@ class GroupStore( server.rosterAfter(local.rosterAfter(-1).lastOrNull()?.seq ?: -1), ) } - val epochs = local.rosterAfter(-1).mapNotNull { it.epoch?.n } - for (epoch in epochs) { - for (name in listOf(me, "chain")) { - if (local.key(epoch, name) == null) server.key(epoch, name)?.let { local.putKey(epoch, name, it) } - } + for ((epoch, name) in keyNames(local.rosterAfter(-1), me)) { + if (local.key(epoch, name) == null) server.key(epoch, name)?.let { local.putKey(epoch, name, it) } } if (!objects) return cursor if (server is FileLayout) { diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt index 30c48be4e..14b676f85 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt @@ -53,6 +53,29 @@ internal fun epochOf( chain: List, ): Int = entry.epoch?.n ?: chain.fold(null as RosterState?) { state, e -> applyEntry(state, e) }?.epoch ?: 0 +/** + * The wraps and chain boxes a store holds for [me] along [chain]: a chain box for every epoch after + * genesis, and [me]'s wrap only for the epochs since it joined. + */ +internal fun keyNames( + chain: List, + me: String, +): List> { + var epoch = 0 + val names = ArrayList>() + for (entry in chain) { + val started = entry.epoch + if (started != null) { + epoch = started.n + if (epoch > 0) names += epoch to "chain" + if (me in started.to) names += epoch to me + } else if (entry.add.orEmpty().any { it.did == me }) { + names += epoch to me + } + } + return names +} + /** A wrap or chain box as an append carries it, in base64url. */ internal fun wrapBytes(b64: String): ByteArray = runCatching { SyncBytes.b64urlDecode(b64) }.getOrNull() ?: reject(ProtocolError.ROSTER_FORMAT) diff --git a/app/src/test/java/io/github/aedev/flow/sync2/HeldStoreTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/HeldStoreTest.kt index 8125d028b..24b6ba1d9 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/HeldStoreTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/HeldStoreTest.kt @@ -7,6 +7,7 @@ import io.github.aedev.flow.sync.protocol.CollectionWire import io.github.aedev.flow.sync.protocol.SyncCollection import io.github.aedev.flow.sync2.data.LinkedDatabase import io.github.aedev.flow.sync2.engine.Session +import io.github.aedev.flow.sync2.linking.rotateEpoch import io.github.aedev.flow.sync2.roster.Op import io.github.aedev.flow.sync2.roster.Role import io.github.aedev.flow.sync2.roster.buildEntry @@ -14,6 +15,7 @@ import io.github.aedev.flow.sync2.roster.entryHash import io.github.aedev.flow.sync2.roster.memberSubject import io.github.aedev.flow.sync2.store.Appended import io.github.aedev.flow.sync2.store.GroupStore +import io.github.aedev.flow.sync2.store.Remote import io.github.aedev.flow.sync2.store.StorePaths import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.runTest @@ -115,4 +117,39 @@ class HeldStoreTest { storeB.download(b, 0, objects = true) assertTrue(a in Session.open(dbB, storeB, g.b).pull(setOf(LINK)).batches) } + + @Test + fun a_download_asks_only_for_keys_the_store_holds() = + runTest { + val g = LinkedGroup(this).apply { create() } + check(rotateEpoch(Session.open(g.deviceDb(), g.store, g.a), g.a)) + val state = g.store.head() + val c = newMember() + val add = + buildEntry( + g.group, + state, + draft(Op.ADD, g.a).copy(add = listOf(memberSubject(c.sigPublic, c.kemPublic, Role.DEVICE))), + listOf(g.a, c), + ) + val current = Session.open(g.deviceDb(), g.store, g.a).currentKey() + check(g.store.appendRoster(state.head, add, wraps(g.group, 1, current, listOf(c)), null) == Appended.Done) + for (member in listOf(g.b, c)) { + val asked = AskedKeys(g.store) + GroupStore(heldStore(emptyDb()), asked).download(SyncBytes.toHex(member.did), 0, objects = false) + assertTrue("${asked.missing}", asked.missing.isEmpty()) + } + } + + /** Records each key a download asks for that the store doesn't hold. */ + private class AskedKeys( + private val inner: MemoryStore, + ) : Remote by inner { + val missing = ArrayList>() + + override suspend fun key( + epoch: Int, + name: String, + ) = inner.key(epoch, name).also { if (it == null) missing += epoch to name } + } }