fix(sync): let one bad roster file, object or snapshot stop only itself

.github/nightly-notes/2026-11-07-fix-sync2-oversized-item.md

@@ -0,0 +1,2 @@
+## Fixed
+- 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

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 }

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<String, RosterState>()
+
+    /** 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<Entry>,
     genesis: String?,
 ): List<Node> {
+    val valid = ValidEntries(group, genesis)
     val nodes = LinkedHashMap<String, Node>()
     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)

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
         }
 

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<String> = files.list("roster").filter { it.endsWith(".r") }.map { "roster/$it" }
-
-    /** Every roster entry stored, fork candidates included. */
-    private suspend fun entries(): List<Entry> = rosterPaths().mapNotNull { entry(it) }
+    private fun entry(bytes: ByteArray): Entry? = runCatching { parseEntry(Json.parseToJsonElement(bytes.decodeToString())) }.getOrNull()
+
+    suspend fun rosterPaths(): List<String> =
+        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<Entry> {
+        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<String> = 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

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

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<Entry>) =
         withContext(Dispatchers.IO) {
             entries.forEach { write(StorePaths.roster(it.seq, entryHash(it)), canonicalEntry(it).toByteArray()) }

app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt

@@ -169,4 +169,27 @@ 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")
+
+    /** 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 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 <T> oneItem(read: suspend () -> T?): T? =
+    try {
+        read()
+    } catch (e: StoreException) {
+        if (e.stopsOnlyItsItem) null else throw e
+    }

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()
 }