Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 7 additions & 9 deletions app/src/main/java/io/github/aedev/flow/sync2/lan/LanPeer.kt
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
23 changes: 23 additions & 0 deletions app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,29 @@ internal fun epochOf(
chain: List<Entry>,
): 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<Entry>,
me: String,
): List<Pair<Int, String>> {
var epoch = 0
val names = ArrayList<Pair<Int, String>>()
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)
Expand Down
37 changes: 37 additions & 0 deletions app/src/test/java/io/github/aedev/flow/sync2/HeldStoreTest.kt
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,15 @@ 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
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
Expand Down Expand Up @@ -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<Pair<Int, String>>()

override suspend fun key(
epoch: Int,
name: String,
) = inner.key(epoch, name).also { if (it == null) missing += epoch to name }
}
}
Loading