diff --git a/app/src/main/java/io/github/aedev/flow/sync2/engine/Bootstrap.kt b/app/src/main/java/io/github/aedev/flow/sync2/engine/Bootstrap.kt index f3bc4fce5..88a08be86 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/engine/Bootstrap.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/engine/Bootstrap.kt @@ -57,60 +57,64 @@ internal suspend fun Session.bootstrap(links: Set?): Start? { val first = logs.mapValues { (_, seqs) -> seqs.min() } if (first.values.all { it == 1L }) return null - val best = HashMap() + val candidates = HashMap>() for ((did, seqs) in logs) { - for (found in newestManifests(did, seqs.sorted(), links)) { - val link = found.manifest.link - if (found.manifest.covers(first) && best[link].let { it == null || it.manifest.merged() < found.manifest.merged() }) { - best[link] = found - } + for (found in manifests(did, seqs.sorted(), links)) { + if (found.manifest.covers(first)) candidates.getOrPut(found.manifest.link) { ArrayList() } += found + } + } + val best = HashMap() + val starts = ArrayList>() + val batches = HashMap>() + for ((link, found) in candidates) { + // The newest snapshot of the link this device can open (spec/protocol.md, Reading a link from its snapshot). + for (candidate in found.sortedByDescending { it.manifest.merged() }) { + val (at, read) = + try { + positions(candidate.manifest) to snapshotBatches(candidate.did, candidate.epoch, candidate.manifest) + } catch (_: ProtocolException) { + continue + } + best[link] = candidate + starts += at + batches.getOrPut(candidate.did) { ArrayList() } += read + break } } if (best.isEmpty()) return null val positions = - best.values - .map { found -> - found.manifest.vv.filterKeys { it != me }.mapValues { (_, tip) -> - val hash = - runCatching { SyncBytes.fromHex(tip.hash) }.getOrNull()?.takeIf { it.size == 32 } ?: reject(ProtocolError.PREV) - StreamPosition(tip.seq, hash) - } - }.reduce { lowest, next -> - // A log one snapshot leaves out is read from its first object. - lowest.filterKeys { it in next }.mapValues { (did, at) -> minOf(at, next.getValue(did), compareBy { it.seq }) } - } + starts.reduce { lowest, next -> + // A log one snapshot leaves out is read from its first object. + lowest.filterKeys { it in next }.mapValues { (did, at) -> minOf(at, next.getValue(did), compareBy { it.seq }) } + } val floors = best.flatMap { (link, found) -> found.manifest.vv.map { (did, tip) -> (link to did) to tip.seq } }.toMap() - val batches = HashMap>() - for (found in best.values) batches.getOrPut(found.did) { ArrayList() } += snapshotBatches(found.did, found.epoch, found.manifest) return Start(batches, positions, floors) } -/** The newest snapshot manifest of each of [links] (every link when null) in [did]'s log, searched from its last object back. */ -private suspend fun Session.newestManifests( +/** Where each other log goes on after [manifest]'s snapshot, chained to the hash it names. */ +private fun Session.positions(manifest: Manifest): Map = + manifest.vv.filterKeys { it != me }.mapValues { (_, tip) -> + val hash = runCatching { SyncBytes.fromHex(tip.hash) }.getOrNull()?.takeIf { it.size == 32 } ?: reject(ProtocolError.PREV) + StreamPosition(tip.seq, hash) + } + +/** Every snapshot manifest of [links] (every link when null) in [did]'s log. */ +private suspend fun Session.manifests( did: String, seqs: List, links: Set?, -): List { - val found = HashMap() - for (seq in seqs.asReversed()) { - if (links != null && found.keys.containsAll(links)) break - val bytes = store.getObject(StorePaths.log(did, seq)) ?: continue - val (epoch, manifest) = - try { - val parsed = verified(bytes, did, seq) - if (parsed.header.type != ObjectType.SNAPSHOT) continue - parsed.header.epoch to (openContent(parsed) as Content.Snapshot).manifest - } catch (_: ProtocolException) { - continue - } - if ((links == null || manifest.link in links) && - manifest.link !in found - ) { - found[manifest.link] = Found(did, epoch, manifest) +): List = + seqs.mapNotNull { seq -> + val bytes = store.getObject(StorePaths.log(did, seq)) ?: return@mapNotNull null + try { + val parsed = verified(bytes, did, seq) + if (parsed.header.type != ObjectType.SNAPSHOT) return@mapNotNull null + val manifest = (openContent(parsed) as Content.Snapshot).manifest + Found(did, parsed.header.epoch, manifest).takeIf { links == null || manifest.link in links } + } catch (_: ProtocolException) { + null } } - return found.values.toList() -} internal suspend fun Session.snapshotBatches( writer: String, diff --git a/app/src/test/java/io/github/aedev/flow/sync2/SnapshotsTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/SnapshotsTest.kt index 3b691105b..c2cb01d0d 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/SnapshotsTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/SnapshotsTest.kt @@ -176,4 +176,31 @@ class SnapshotsTest { .isEmpty(), ) } + + @Test + fun a_new_device_falls_back_to_an_older_snapshot_when_a_blob_is_bad() = + runTest { + val breaks = + listOf<(MemoryStore, String) -> Unit>( + { store, path -> store.take(path) }, + { store, path -> store.put(path, store.take(path).also { it[it.size - 1] = (it[it.size - 1] + 1).toByte() }) }, + ) + for (bad in breaks) { + val g = LinkedGroup(this).apply { create() } + val did = SyncBytes.toHex(g.a.did) + val a = Session.open(g.deviceDb(), g.store, g.a) + a.push(watched("aaaaaaaaaaa"), LINK) + a.writeSnapshot(watched("aaaaaaaaaaa"), LINK) + a.push(watched("bbbbbbbbbbb"), LINK) + a.writeSnapshot(watched("aaaaaaaaaaa", "bbbbbbbbbbb"), LINK) + a.push(watched("ccccccccccc"), LINK) + g.store.take(StorePaths.log(did, 1)) + val newest = parseObject(checkNotNull(g.store.objects[StorePaths.log(did, 4)]).second, g.group) + val manifest = parseManifest(decodePayload(newest.open(purposeKey(g.epochKey, g.group, 0, Purpose.SNAPSHOT)))) + bad(g.store, StorePaths.blob(did, manifest.parts.single().blob)) + + val pulled = Session.open(g.deviceDb(), g.store, g.b).pull(setOf(LINK)) + assertEquals(listOf("aaaaaaaaaaa", "bbbbbbbbbbb", "ccccccccccc"), videos(pulled.batches.values).distinct()) + } + } }