From 1c48ce0a9066b05cc479dce828820686aff252b9 Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 7 Nov 2026 15:58:38 +0200 Subject: [PATCH 1/2] fix(sync): let one bad roster file, object or snapshot stop only itself --- .../2026-11-07-fix-sync2-oversized-item.md | 2 + .../github/aedev/flow/sync2/engine/Session.kt | 12 +- .../github/aedev/flow/sync2/roster/Forks.kt | 39 ++- .../aedev/flow/sync2/store/FileLayout.kt | 59 ++++- .../aedev/flow/sync2/store/GroupStore.kt | 18 +- .../aedev/flow/sync2/store/LocalStore.kt | 9 + .../io/github/aedev/flow/sync2/store/Store.kt | 17 ++ .../github/aedev/flow/sync2/OneBadItemTest.kt | 234 ++++++++++++++++++ 8 files changed, 363 insertions(+), 27 deletions(-) create mode 100644 .github/nightly-notes/2026-11-07-fix-sync2-oversized-item.md create mode 100644 app/src/test/java/io/github/aedev/flow/sync2/OneBadItemTest.kt diff --git a/.github/nightly-notes/2026-11-07-fix-sync2-oversized-item.md b/.github/nightly-notes/2026-11-07-fix-sync2-oversized-item.md new file mode 100644 index 000000000..92574da4d --- /dev/null +++ b/.github/nightly-notes/2026-11-07-fix-sync2-oversized-item.md @@ -0,0 +1,2 @@ +## Fixed +- Linked devices: one file or object a store can't hand over no longer stops the whole sync diff --git a/app/src/main/java/io/github/aedev/flow/sync2/engine/Session.kt b/app/src/main/java/io/github/aedev/flow/sync2/engine/Session.kt index 3cacc3f33..d3a4156f3 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/engine/Session.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/engine/Session.kt @@ -353,10 +353,14 @@ class Session private constructor( .maxByOrNull { it.manifest.merged() } val tips = newest?.manifest?.vv.orEmpty() val records = - newest?.let { - val parsed = verified(store.getObject(StorePaths.log(it.did, it.seq)) ?: return null, it.did, it.seq) - mapOf(it.did to snapshotBatches(it.did, parsed.header.epoch, it.manifest)) - } ?: emptyMap() + try { + newest?.let { + val parsed = verified(store.getObject(StorePaths.log(it.did, it.seq)) ?: return null, it.did, it.seq) + mapOf(it.did to snapshotBatches(it.did, parsed.header.epoch, it.manifest)) + } ?: emptyMap() + } catch (_: ProtocolException) { + return null + } db.profiles().saveReads( positions .filter { it.did != me } diff --git a/app/src/main/java/io/github/aedev/flow/sync2/roster/Forks.kt b/app/src/main/java/io/github/aedev/flow/sync2/roster/Forks.kt index bd6a74f7f..db23a6e39 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/roster/Forks.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/roster/Forks.kt @@ -52,24 +52,45 @@ private class Node( fun mayStand() = waiting == 0 && (parent == null || parent.status == Status.STANDS) } +/** + * The entries that count (spec/protocol.md, Forks): each valid against the chain from genesis to its own `prev`. + * Entries go in with lower `seq` first. A null [group] takes any group, and [genesis] pins the only genesis that counts. + */ +internal class ValidEntries( + private val group: String?, + private val genesis: String?, +) { + private val states = HashMap() + + /** The state after [entry], or null when it doesn't count or is already in. */ + fun admit(entry: Entry): RosterState? { + val hash = entryHash(entry) + if (hash in states || (group != null && entry.group != group)) return null + if (entry.op == Op.GENESIS && genesis != null && hash != genesis) return null + val parent = if (entry.op == Op.GENESIS) null else states[entry.prev] ?: return null + val state = + try { + applyEntry(parent, entry) + } catch (_: ProtocolException) { + return null + } + states[hash] = state + return state + } +} + /** Every held entry that is valid against its parent, as a tree. */ private fun validTree( group: String, held: List, genesis: String?, ): List { + val valid = ValidEntries(group, genesis) val nodes = LinkedHashMap() for (entry in held.sortedBy { it.seq }) { + val state = valid.admit(entry) ?: continue val hash = entryHash(entry) - if (hash in nodes || entry.group != group) continue - if (entry.op == Op.GENESIS && genesis != null && hash != genesis) continue - val parent = if (entry.op == Op.GENESIS) null else nodes[entry.prev] ?: continue - val state = - try { - applyEntry(parent?.state, entry) - } catch (_: ProtocolException) { - continue - } + val parent = nodes[entry.prev].takeIf { entry.op != Op.GENESIS } val node = Node(entry, hash, state, parent) nodes[hash] = node parent?.children?.add(node) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/FileLayout.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/FileLayout.kt index 53e23debe..b0fd918b8 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/FileLayout.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/FileLayout.kt @@ -11,6 +11,10 @@ import java.util.concurrent.ConcurrentHashMap private val DID = Regex("[0-9a-f]{32}") private val LOG_NAME = Regex("([0-9a-f]{16})\\.o") +/** How many files and bytes of `roster/` one read takes. Far above a real group's: a roster holds at most 4,096 small entries. */ +private const val ROSTER_FILES = 16_384 +private const val ROSTER_BYTES = 64L * 1024 * 1024 + /** * The store layout (spec/store.md) on plain files. Roster appends aren't linearized, so competing * entries both stay and every device picks the same winner; [me] deletes only its own paths. @@ -19,13 +23,28 @@ class FileLayout( private val files: FileStore, private val me: String, ) : Remote { - private suspend fun entry(path: String): Entry? = - files.get(path)?.let { runCatching { parseEntry(Json.parseToJsonElement(it.decodeToString())) }.getOrNull() } - - suspend fun rosterPaths(): List = files.list("roster").filter { it.endsWith(".r") }.map { "roster/$it" } - - /** Every roster entry stored, fork candidates included. */ - private suspend fun entries(): List = rosterPaths().mapNotNull { entry(it) } + private fun entry(bytes: ByteArray): Entry? = runCatching { parseEntry(Json.parseToJsonElement(bytes.decodeToString())) }.getOrNull() + + suspend fun rosterPaths(): List = + files + .list("roster") + .filter { it.endsWith(".r") } + .map { "roster/$it" } + .sorted() + + /** Every roster entry stored, fork candidates included. All of them stay in memory, so past the limits nothing is answered. */ + private suspend fun entries(): List { + val paths = rosterPaths() + if (paths.size > ROSTER_FILES) throw StoreException.Malformed() + var bytes = 0L + return paths.mapNotNull { path -> + oneItem { files.get(path) }?.let { file -> + bytes += file.size + if (bytes > ROSTER_BYTES) throw StoreException.Malformed() + entry(file) + } + } + } @Volatile private var pins: Set = emptySet() @@ -74,10 +93,23 @@ class FileLayout( if (path.startsWith("dev/$me/")) files.delete(path) } - /** Copies the roster entries [local] doesn't hold, fork candidates included, so it resolves forks as every device does. */ + /** + * Copies the roster entries [local] doesn't hold, fork candidates included, so it resolves forks as every device does. + * One file at a time, lowest `seq` first, keeping only what counts: anything else is read again next time, up to the limits. + */ suspend fun pullRoster(local: LocalStore) { val held = local.entries().map { StorePaths.roster(it.seq, entryHash(it)) }.toSet() - local.addEntries(rosterPaths().filter { it !in held }.mapNotNull { entry(it) }) + val fresh = rosterPaths().filter { it !in held }.take(ROSTER_FILES) + if (fresh.isEmpty()) return + val valid = local.validEntries() + var bytes = 0L + for (path in fresh) { + if (bytes > ROSTER_BYTES) return + val file = oneItem { files.get(path) } ?: continue + bytes += file.size + val entry = entry(file) ?: continue + if (valid.admit(entry) != null) local.addEntries(listOf(entry)) + } } /** @@ -95,7 +127,14 @@ class FileLayout( var seq = tips[did]?.plus(1) ?: oldest(did) ?: continue while (true) { val path = StorePaths.log(did, seq) - val bytes = files.get(path) ?: break + val bytes = + try { + files.get(path) + } catch (e: StoreException) { + if (!e.stopsOnlyItsItem) throw e + whole = false + break + } ?: break // A store that answers every path, with an error page say, would otherwise never end the log. if (!check.admits(path, bytes)) { whole = false 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 98e72883e..ba21a71cd 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 @@ -63,8 +63,13 @@ class GroupStore( bytes: ByteArray, ) = local.putObject(path, bytes) - override suspend fun getObject(path: String): ByteArray? = - local.getObject(path) ?: (server as? FileLayout)?.blob(path)?.takeIf { namesBlob(path, it) }?.also { local.putObject(path, it) } + override suspend fun getObject(path: String): ByteArray? = local.getObject(path) ?: blobFromFiles(path) + + /** A snapshot blob a file store holds and this device doesn't yet. */ + private suspend fun blobFromFiles(path: String): ByteArray? { + val files = server as? FileLayout ?: return null + return oneItem { files.blob(path) }?.takeIf { namesBlob(path, it) }?.also { local.putObject(path, it) } + } // The server first: a delete it refuses leaves the object held, so a later collection tries again. override suspend fun deleteObject(path: String) { @@ -100,7 +105,7 @@ class GroupStore( } val resolved = local.resolved() for ((epoch, kc, name) in keyNames(resolved.chain, resolved.standing, me)) { - if (local.key(epoch, kc, name) == null) server.key(epoch, kc, name)?.let { local.putKey(epoch, kc, name, it) } + if (local.key(epoch, kc, name) == null) oneItem { server.key(epoch, kc, name) }?.let { local.putKey(epoch, kc, name, it) } } if (!objects) return cursor if (server is FileLayout) { @@ -130,7 +135,12 @@ class GroupStore( check: ObjectCheck, ): Boolean { if (local.getObject(path) != null) return true - val bytes = server.getObject(path) ?: return true + val bytes = + try { + server.getObject(path) ?: return true + } catch (e: StoreException) { + if (e.stopsOnlyItsItem) return false else throw e + } if (!check.admits(path, bytes)) return false local.putObject(path, bytes) return true diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/LocalStore.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/LocalStore.kt index fa7e0f26b..e95df024b 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/LocalStore.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/LocalStore.kt @@ -4,6 +4,7 @@ import io.github.aedev.flow.sync2.data.HeldDao import io.github.aedev.flow.sync2.data.HeldRow import io.github.aedev.flow.sync2.roster.Entry import io.github.aedev.flow.sync2.roster.ForkResolution +import io.github.aedev.flow.sync2.roster.ValidEntries import io.github.aedev.flow.sync2.roster.canonicalEntry import io.github.aedev.flow.sync2.roster.entryHash import io.github.aedev.flow.sync2.roster.parseEntry @@ -59,6 +60,14 @@ class LocalStore( return resolveForks(genesis?.group, entries(), held.rosterPins().toSet(), genesis?.let(::entryHash)) } + /** What decides whether a new entry counts, given what this device holds. */ + internal suspend fun validEntries(): ValidEntries { + val genesis = held.genesis()?.let { parseEntry(Json.parseToJsonElement(it)) } + val valid = ValidEntries(genesis?.group, genesis?.let(::entryHash)) + entries().forEach(valid::admit) + return valid + } + suspend fun addEntries(entries: List) = withContext(Dispatchers.IO) { entries.forEach { write(StorePaths.roster(it.seq, entryHash(it)), canonicalEntry(it).toByteArray()) } 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 0919e1adc..4b2856bfe 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 @@ -169,4 +169,21 @@ sealed class StoreException( class InstanceChanged : StoreException("This is not the server instance this device paired with") class Malformed : StoreException("The server sent an invalid response") + + /** Whether this read failed for the one item asked for, over its limit or refused, so the rest of the sync can go on (spec/store.md, Untrusted input). */ + val stopsOnlyItsItem: Boolean + get() = + when (this) { + is Malformed -> true + is Refused -> !rateLimited + else -> false + } } + +/** What [read] answers, or null when it failed for that one item. */ +internal suspend fun oneItem(read: suspend () -> T?): T? = + try { + read() + } catch (e: StoreException) { + if (e.stopsOnlyItsItem) null else throw e + } diff --git a/app/src/test/java/io/github/aedev/flow/sync2/OneBadItemTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/OneBadItemTest.kt new file mode 100644 index 000000000..714710334 --- /dev/null +++ b/app/src/test/java/io/github/aedev/flow/sync2/OneBadItemTest.kt @@ -0,0 +1,234 @@ +package io.github.aedev.flow.sync2 + +import android.app.Application +import androidx.test.ext.junit.runners.AndroidJUnit4 +import io.github.aedev.flow.sync.crypto.SyncBytes +import io.github.aedev.flow.sync.protocol.CollectionWire +import io.github.aedev.flow.sync.protocol.SyncCollection +import io.github.aedev.flow.sync2.engine.Session +import io.github.aedev.flow.sync2.engine.writeSnapshot +import io.github.aedev.flow.sync2.roster.canonicalEntry +import io.github.aedev.flow.sync2.roster.entryHash +import io.github.aedev.flow.sync2.store.ChangeList +import io.github.aedev.flow.sync2.store.FileLayout +import io.github.aedev.flow.sync2.store.FileStore +import io.github.aedev.flow.sync2.store.GroupStore +import io.github.aedev.flow.sync2.store.MAX_OBJECT_SIZE +import io.github.aedev.flow.sync2.store.Mailbox +import io.github.aedev.flow.sync2.store.Remote +import io.github.aedev.flow.sync2.store.StoreException +import io.github.aedev.flow.sync2.store.StorePaths +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.annotation.Config + +/** One item a store can't hand over stops only that item (spec/store.md, Untrusted input). */ +@RunWith(AndroidJUnit4::class) +@Config(sdk = [34], application = Application::class) +class OneBadItemTest { + private val watched = mapOf(SyncCollection.WATCH_HISTORY to CollectionWire(listOf("""{"title":"t","videoId":"aaaaaaaaaaa"}"""), 1, "")) + + /** Files in memory, where a read of a path in [failing] throws what it names. */ + private class Files( + val held: MutableMap = HashMap(), + val failing: MutableMap = HashMap(), + ) : FileStore { + var served = 0L + var tagged: String? = null + override val location = "memory" + + override suspend fun get(path: String): ByteArray? { + failing[path]?.let { throw it } + return held[path]?.also { served += it.size } + } + + override suspend fun put( + path: String, + bytes: ByteArray, + ) { + held[path] = bytes + } + + override suspend fun delete(path: String) { + held.remove(path) + } + + override suspend fun list(folder: String) = + held.keys + .filter { it.startsWith("$folder/") } + .map { it.removePrefix("$folder/").substringBefore('/') } + .distinct() + + override suspend fun tag(folder: String) = tagged + } + + private fun me(g: LinkedGroup) = SyncBytes.toHex(g.b.did) + + private fun Files.publish(g: LinkedGroup) { + g.store.roster.forEach { held[StorePaths.roster(it.seq, entryHash(it))] = canonicalEntry(it).toByteArray() } + } + + private suspend fun TestScope.batchFromA(g: LinkedGroup): String { + Session.open(g.deviceDb(), g.store, g.a).push(watched, LINK) + return StorePaths.log(SyncBytes.toHex(g.a.did), 1) + } + + @Test + fun a_roster_file_that_can_t_be_read_leaves_the_others_read() = + runTest { + val g = LinkedGroup(this).apply { create() } + val files = Files().apply { publish(g) } + val bad = "roster/0000000000000000-0000000000000000.r" + files.held[bad] = ByteArray(8) + files.failing[bad] = StoreException.Malformed() + val local = heldStore(g.deviceDb()) + + GroupStore(local, FileLayout(files, me(g))).download(me(g), 0, objects = false) + + assertEquals(g.store.roster.size, local.entries().size) + } + + @Test + fun a_log_file_that_can_t_be_read_stops_only_its_sender_and_is_tried_again() = + runTest { + val g = LinkedGroup(this).apply { create() } + val good = batchFromA(g) + val first = "00".repeat(16) + val bad = StorePaths.log(first, 1) + val files = Files().apply { publish(g) } + files.tagged = "same" + files.held[good] = g.store.take(good) + files.held[bad] = ByteArray(8) + files.failing[bad] = StoreException.Refused(403, "webdav") + val local = heldStore(g.deviceDb()) + + GroupStore(local, FileLayout(files, me(g))).download(me(g), 0, objects = true) + assertNotNull("the sender after it was read", local.getObject(good)) + + files.failing.clear() + val asked = ArrayList() + val spy = + object : FileStore by files { + override suspend fun get(path: String): ByteArray? { + asked += path + return files.get(path) + } + } + GroupStore(local, FileLayout(spy, me(g))).download(me(g), 0, objects = true) + assertTrue("an unread file keeps the folder's tag from being trusted", bad in asked) + } + + @Test + fun being_rate_limited_or_refused_a_login_still_ends_the_sync() = + runTest { + val g = LinkedGroup(this).apply { create() } + val bad = + StorePaths.roster( + g.store.roster + .last() + .seq, + entryHash(g.store.roster.last()), + ) + for (stop in listOf(StoreException.Refused(429, "webdav"), StoreException.LoginRefused())) { + val files = Files().apply { publish(g) } + files.failing[bad] = stop + val store = GroupStore(heldStore(g.deviceDb()), FileLayout(files, me(g))) + assertEquals(stop, runCatching { store.download(me(g), 0, objects = false) }.exceptionOrNull()) + } + } + + @Test + fun roster_files_that_don_t_chain_are_not_held() = + runTest { + val g = LinkedGroup(this).apply { create() } + val files = Files().apply { publish(g) } + val last = g.store.roster.last() + val orphan = last.copy(prev = "ab".repeat(32), meta = mapOf("ab".repeat(16) to "x".repeat(1024))) + files.held[StorePaths.roster(orphan.seq, entryHash(orphan))] = canonicalEntry(orphan).toByteArray() + val forged = last.copy(meta = mapOf("ab".repeat(16) to "x".repeat(1024))) + files.held[StorePaths.roster(forged.seq, entryHash(forged))] = canonicalEntry(forged).toByteArray() + val local = heldStore(g.deviceDb()) + + GroupStore(local, FileLayout(files, me(g))).download(me(g), 0, objects = false) + + assertEquals( + g.store.roster + .map(::entryHash) + .toSet(), + local.entries().map(::entryHash).toSet(), + ) + } + + @Test + fun one_pull_stops_reading_the_roster_folder_past_its_limit() = + runTest { + val g = LinkedGroup(this).apply { create() } + val files = Files().apply { publish(g) } + val junk = ByteArray(MAX_OBJECT_SIZE) + repeat(40) { files.held["roster/ffffffffffffff%02x-0000000000000000.r".format(it)] = junk } + + val local = heldStore(g.deviceDb()) + GroupStore(local, FileLayout(files, me(g))).download(me(g), 0, objects = false) + + assertEquals("what comes first is read", g.store.roster.size, local.entries().size) + assertTrue("read ${files.served} bytes", files.served < 70L * 1024 * 1024) + } + + /** The Flow server, refusing [refused]. */ + private class PickyServer( + private val real: MemoryStore, + private val refused: String, + ) : Remote, + ChangeList by real, + Mailbox by real { + override suspend fun getObject(path: String): ByteArray? { + if (path == refused) throw StoreException.Malformed() + return real.getObject(path) + } + } + + @Test + fun an_object_a_server_can_t_hand_over_leaves_the_others_held_and_its_page_is_read_again() = + runTest { + val g = LinkedGroup(this).apply { create() } + val a = Session.open(g.deviceDb(), g.store, g.a) + a.push(watched, LINK) + a.push(watched, "other-link") + val aHex = SyncBytes.toHex(g.a.did) + val (bad, good) = StorePaths.log(aHex, 1) to StorePaths.log(aHex, 2) + val local = heldStore(g.deviceDb()) + + val next = GroupStore(local, PickyServer(g.store, bad)).download(me(g), 0, objects = true) + + assertNull(local.getObject(bad)) + assertNotNull(local.getObject(good)) + assertEquals(0, next) + } + + @Test + fun a_snapshot_that_can_t_be_read_yet_leaves_the_link_for_a_later_sync() = + runTest { + val g = LinkedGroup(this).apply { create() } + val a = Session.open(g.deviceDb(), g.store, g.a) + a.writeSnapshot(watched, "other-link") + val db = g.deviceDb() + val b = Session.open(db, g.store, g.b) + b.pull(setOf(LINK)).also { b.finish(it) } + val blobs = + g.store.objects.keys + .filter { StorePaths.isBlob(it) } + assertTrue(blobs.isNotEmpty()) + val taken = blobs.associateWith { g.store.take(it) } + + assertNull(Session.open(db, g.store, g.b).startReading("other-link")) + + taken.forEach { (path, bytes) -> g.store.put(path, bytes) } + assertNotNull(Session.open(db, g.store, g.b).startReading("other-link")) + } +} From 639ba65bdcfb5f5e63b3a67abb05fdeb6a7949a4 Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 7 Nov 2026 22:44:26 +0200 Subject: [PATCH 2/2] fix(sync): fail the sync on a transient store error and let only a bad item stop itself --- .../2026-11-07-fix-sync2-oversized-item.md | 2 +- .../aedev/flow/sync2/service/LinkedError.kt | 2 +- .../io/github/aedev/flow/sync2/store/Store.kt | 12 +- .../aedev/flow/sync2/store/Transport.kt | 2 +- .../github/aedev/flow/sync2/OneBadItemTest.kt | 141 +++++++++++++++++- .../aedev/flow/sync2/ServerStoreTest.kt | 4 +- .../aedev/flow/sync2/WebDavStoreTest.kt | 4 +- 7 files changed, 155 insertions(+), 12 deletions(-) diff --git a/.github/nightly-notes/2026-11-07-fix-sync2-oversized-item.md b/.github/nightly-notes/2026-11-07-fix-sync2-oversized-item.md index 92574da4d..cca35fb3a 100644 --- a/.github/nightly-notes/2026-11-07-fix-sync2-oversized-item.md +++ b/.github/nightly-notes/2026-11-07-fix-sync2-oversized-item.md @@ -1,2 +1,2 @@ ## Fixed -- Linked devices: one file or object a store can't hand over no longer stops the whole sync +- Linked devices: one file or object that is too large or refused for itself no longer stops the whole sync. A server error still does, and the next sync picks up where it stopped diff --git a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedError.kt b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedError.kt index 80c3019a7..18f7144ad 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedError.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedError.kt @@ -84,7 +84,7 @@ fun Throwable.linkedError(): LinkedError = if (error == ProtocolError.ROSTER_GROUP) LinkedError.GROUP_OUTDATED else LinkedError.PROTOCOL } - is StoreException.Malformed -> { + is StoreException.Malformed, is StoreException.TooLarge -> { LinkedError.PROTOCOL } 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 4b2856bfe..a0c6b2c51 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 @@ -170,16 +170,22 @@ sealed class StoreException( class Malformed : StoreException("The server sent an invalid response") - /** Whether this read failed for the one item asked for, over its limit or refused, so the rest of the sync can go on (spec/store.md, Untrusted input). */ + /** An answer past the size it may have (spec/store.md, Untrusted input). */ + class TooLarge : StoreException("The server sent more than an answer may hold") + + /** Whether this read failed for the one item asked for, not because the store is out of reach, wants a login or asks for time (spec/store.md, Untrusted input). */ val stopsOnlyItsItem: Boolean get() = when (this) { - is Malformed -> true - is Refused -> !rateLimited + is TooLarge -> true + is Refused -> status in 400..499 && status !in NOT_ABOUT_THE_ITEM else -> false } } +/** 4xx answers that say something about the store or this device rather than the item asked for. */ +private val NOT_ABOUT_THE_ITEM = setOf(401, 407, 408, 429) + /** What [read] answers, or null when it failed for that one item. */ internal suspend fun oneItem(read: suspend () -> T?): T? = try { diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/Transport.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/Transport.kt index cce59de09..a368af515 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/Transport.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/Transport.kt @@ -14,7 +14,7 @@ internal const val MAX_RESPONSE = 16L * 1024 * 1024 /** Reads a body of at most [limit] bytes, failing as soon as more arrive. */ internal fun ResponseBody.bytesUpTo(limit: Long): ByteArray { val source = source() - if (source.request(limit + 1)) throw StoreException.Malformed() + if (source.request(limit + 1)) throw StoreException.TooLarge() return source.readByteArray() } diff --git a/app/src/test/java/io/github/aedev/flow/sync2/OneBadItemTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/OneBadItemTest.kt index 714710334..3ce1cc221 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/OneBadItemTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/OneBadItemTest.kt @@ -7,12 +7,14 @@ import io.github.aedev.flow.sync.protocol.CollectionWire import io.github.aedev.flow.sync.protocol.SyncCollection import io.github.aedev.flow.sync2.engine.Session import io.github.aedev.flow.sync2.engine.writeSnapshot +import io.github.aedev.flow.sync2.linking.rotateEpoch import io.github.aedev.flow.sync2.roster.canonicalEntry import io.github.aedev.flow.sync2.roster.entryHash import io.github.aedev.flow.sync2.store.ChangeList import io.github.aedev.flow.sync2.store.FileLayout import io.github.aedev.flow.sync2.store.FileStore import io.github.aedev.flow.sync2.store.GroupStore +import io.github.aedev.flow.sync2.store.KeyName import io.github.aedev.flow.sync2.store.MAX_OBJECT_SIZE import io.github.aedev.flow.sync2.store.Mailbox import io.github.aedev.flow.sync2.store.Remote @@ -86,7 +88,7 @@ class OneBadItemTest { val files = Files().apply { publish(g) } val bad = "roster/0000000000000000-0000000000000000.r" files.held[bad] = ByteArray(8) - files.failing[bad] = StoreException.Malformed() + files.failing[bad] = StoreException.TooLarge() val local = heldStore(g.deviceDb()) GroupStore(local, FileLayout(files, me(g))).download(me(g), 0, objects = false) @@ -188,7 +190,7 @@ class OneBadItemTest { ChangeList by real, Mailbox by real { override suspend fun getObject(path: String): ByteArray? { - if (path == refused) throw StoreException.Malformed() + if (path == refused) throw StoreException.TooLarge() return real.getObject(path) } } @@ -231,4 +233,139 @@ class OneBadItemTest { taken.forEach { (path, bytes) -> g.store.put(path, bytes) } assertNotNull(Session.open(db, g.store, g.b).startReading("other-link")) } + + /** The Flow server, where reading a path or a key's name in [failing] throws what it names. */ + private class FailingServer( + private val real: MemoryStore, + ) : Remote, + ChangeList by real, + Mailbox by real { + val failing = HashMap() + + override suspend fun getObject(path: String): ByteArray? { + failing[path]?.let { throw it } + return real.getObject(path) + } + + override suspend fun key( + epoch: Int, + kc: String, + name: String, + ): ByteArray? { + failing[name]?.let { throw it } + return real.key(epoch, kc, name) + } + } + + private val notAboutTheItem = listOf(302, 401, 407, 408, 429, 500, 503) + private val aboutTheItem = listOf(400, 403, 410, 451) + + private fun refused(status: Int) = StoreException.Refused(status, "test") + + private suspend fun TestScope.chainBoxOf(g: LinkedGroup): KeyName { + rotateEpoch(Session.open(g.deviceDb(), g.store, g.a), g.a) + return g.store.keys.keys + .first { it.name == "chain" } + } + + @Test + fun a_status_not_about_an_object_fails_the_sync_and_the_next_read_gets_it() = + runTest { + val g = LinkedGroup(this).apply { create() } + val bad = batchFromA(g) + for (status in notAboutTheItem) { + val stop = refused(status) + val server = FailingServer(g.store).apply { failing[bad] = stop } + val local = heldStore(g.deviceDb()) + assertEquals( + "$status", + stop, + runCatching { GroupStore(local, server).download(me(g), 0, objects = true) }.exceptionOrNull(), + ) + server.failing.clear() + GroupStore(local, server).download(me(g), 0, objects = true) + assertNotNull("$status", local.getObject(bad)) + } + } + + @Test + fun a_status_not_about_a_chain_box_fails_the_sync_and_the_next_read_gets_it() = + runTest { + val g = LinkedGroup(this).apply { create() } + val box = chainBoxOf(g) + for (status in notAboutTheItem) { + val stop = refused(status) + val server = FailingServer(g.store).apply { failing["chain"] = stop } + val local = heldStore(g.deviceDb()) + assertEquals( + "$status", + stop, + runCatching { GroupStore(local, server).download(me(g), 0, objects = false) }.exceptionOrNull(), + ) + server.failing.clear() + GroupStore(local, server).download(me(g), 0, objects = false) + assertNotNull("$status", local.key(box.epoch, box.kc, box.name)) + } + } + + @Test + fun a_status_not_about_a_roster_file_fails_the_sync_and_the_next_read_gets_it() = + runTest { + val g = LinkedGroup(this).apply { create() } + val last = g.store.roster.last() + val path = StorePaths.roster(last.seq, entryHash(last)) + for (status in notAboutTheItem) { + val stop = refused(status) + val files = Files().apply { publish(g) } + files.failing[path] = stop + val local = heldStore(g.deviceDb()) + val store = GroupStore(local, FileLayout(files, me(g))) + assertEquals("$status", stop, runCatching { store.download(me(g), 0, objects = false) }.exceptionOrNull()) + files.failing.clear() + store.download(me(g), 0, objects = false) + assertEquals("$status", g.store.roster.size, local.entries().size) + } + } + + @Test + fun a_status_not_about_a_log_file_fails_the_sync_and_the_next_read_gets_it() = + runTest { + val g = LinkedGroup(this).apply { create() } + val bad = batchFromA(g) + for (status in notAboutTheItem) { + val stop = refused(status) + val files = Files().apply { publish(g) } + files.tagged = "same-$status" + files.held[bad] = checkNotNull(g.store.objects[bad]).second + files.failing[bad] = stop + val local = heldStore(g.deviceDb()) + val store = GroupStore(local, FileLayout(files, me(g))) + assertEquals("$status", stop, runCatching { store.download(me(g), 0, objects = true) }.exceptionOrNull()) + files.failing.clear() + store.download(me(g), 0, objects = true) + assertNotNull("$status", local.getObject(bad)) + } + } + + @Test + fun an_object_a_chain_box_or_a_roster_file_refused_for_itself_stops_only_itself() = + runTest { + val g = LinkedGroup(this).apply { create() } + val bad = batchFromA(g) + chainBoxOf(g) + val last = g.store.roster.last() + for (stop in aboutTheItem.map(::refused) + StoreException.TooLarge()) { + val server = FailingServer(g.store).apply { failing[bad] = stop } + server.failing["chain"] = stop + val local = heldStore(g.deviceDb()) + GroupStore(local, server).download(me(g), 0, objects = true) + assertNull(local.getObject(bad)) + + val files = Files().apply { publish(g) } + files.failing[StorePaths.roster(last.seq, entryHash(last))] = stop + val held = heldStore(g.deviceDb()) + GroupStore(held, FileLayout(files, me(g))).download(me(g), 0, objects = false) + assertEquals(g.store.roster.size - 1, held.entries().size) + } + } } diff --git a/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt index b05c889f0..6fa354980 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt @@ -169,8 +169,8 @@ class ServerStoreTest { } } assertEquals(object4MiB, store().getObject("fits")?.size) - assertTrue(runCatching { store().getObject("big") }.exceptionOrNull() is StoreException.Malformed) - assertTrue(runCatching { store().changes(0) }.exceptionOrNull() is StoreException.Malformed) + assertTrue(runCatching { store().getObject("big") }.exceptionOrNull() is StoreException.TooLarge) + assertTrue(runCatching { store().changes(0) }.exceptionOrNull() is StoreException.TooLarge) } @Test diff --git a/app/src/test/java/io/github/aedev/flow/sync2/WebDavStoreTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/WebDavStoreTest.kt index 46b7b1bab..c9a4e5b8a 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/WebDavStoreTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/WebDavStoreTest.kt @@ -109,8 +109,8 @@ class WebDavStoreTest { } }.use { standIn -> val dav = WebDavStore.connect(standIn.url, "u", "p") - assertTrue(runCatching { dav.get("dev/x.o") }.exceptionOrNull() is StoreException.Malformed) - assertTrue(runCatching { dav.list("dev/") }.exceptionOrNull() is StoreException.Malformed) + assertTrue(runCatching { dav.get("dev/x.o") }.exceptionOrNull() is StoreException.TooLarge) + assertTrue(runCatching { dav.list("dev/") }.exceptionOrNull() is StoreException.TooLarge) } }