Nov 2, 2026 · Nov 2, 2026
@@ -47,8 +47,9 @@ internal suspend fun Session.bootstrap(links: Set<String?>?): Start? {
     var cursor = 0L
     while (true) {
         val (changes, next) = store.changes(cursor)
-        if (changes.isEmpty()) break
         changes.mapNotNull { StorePaths.parseLog(it.path) }.forEach { (did, seq) -> logs.getOrPut(did) { ArrayList() } += seq }
+        // A cursor that doesn't move would hand out the same page forever.
+        if (changes.isEmpty() || next <= cursor) break
         cursor = next
     }
     logs.remove(me)

@@ -170,8 +170,9 @@ private suspend fun Session.heldPaths(): Set<String> {
     var cursor = 0L
     while (true) {
         val (changed, next) = store.changes(cursor)
-        if (changed.isEmpty()) return paths
         changed.mapTo(paths) { it.path }
+        // A cursor that doesn't move would hand out the same page forever.
+        if (changed.isEmpty() || next <= cursor) return paths
         cursor = next
     }
 }
@@ -257,6 +257,7 @@ class Session private constructor(
         while (true) {
             val (changes, next) = store.changes(cursor)
             if (changes.isEmpty()) break
+            val before = positions.mapValues { it.value.seq }
             // Objects of one sender may be listed out of order: retry while a pass makes progress.
             var waiting =
                 changes
@@ -298,6 +299,8 @@ class Session private constructor(
                 stalled = true
                 break
             }
+            // A page that moves neither the cursor nor any log would be handed out forever.
+            if (next <= cursor && positions.all { (did, at) -> before[did] == at.seq }) break
             cursor = next
         }
         return Pulled(batches, positions, cursor, consumed, stalled, acks, manifests, named, reads, readsDone)

@@ -20,6 +20,8 @@ private val MAGIC = byteArrayOf('F'.code.toByte(), 'S'.code.toByte(), '2'.code.t
 const val SUITE: Byte = 0x01
 const val HEADER_SIZE = 82
 private const val SIGNATURE_SIZE = 64
+private const val SENDER_AT = 26
+private const val SEQ_AT = 42
 const val MIN_OBJECT_SIZE = HEADER_SIZE + BOX_OVERHEAD + 256 + SIGNATURE_SIZE
 const val MAX_OBJECT_SIZE = 4 * 1024 * 1024
 val ZERO_HASH = ByteArray(32)
@@ -102,14 +104,24 @@ fun parseObject(
             type = type,
             group = bytes.copyOfRange(6, 22),
             epoch = buffer.getInt(22),
-            sender = bytes.copyOfRange(26, 42),
-            seq = buffer.getLong(42),
+            sender = bytes.copyOfRange(SENDER_AT, SEQ_AT),
+            seq = buffer.getLong(SEQ_AT),
             prev = bytes.copyOfRange(50, 82),
         )
     if (!header.group.contentEquals(group)) reject(ProtocolError.GROUP)
     return ParsedObject(header, bytes)
 }
 
+/** Whether [bytes] start as [sender]'s log object [seq]; anything else, such as an error page, is no object at all. */
+fun namesLogObject(
+    bytes: ByteArray,
+    sender: ByteArray,
+    seq: Long,
+): Boolean =
+    bytes.size >= HEADER_SIZE &&
+        bytes.copyOfRange(SENDER_AT, SEQ_AT).contentEquals(sender) &&
+        ByteBuffer.wrap(bytes).getLong(SEQ_AT) == seq
+
 class StreamPosition(
     val seq: Long,
     val hash: ByteArray,

@@ -1,5 +1,6 @@
 package io.github.aedev.flow.sync2.store
 
+import io.github.aedev.flow.sync.crypto.SyncBytes
 import io.github.aedev.flow.sync2.roster.Entry
 import io.github.aedev.flow.sync2.roster.canonicalEntry
 import io.github.aedev.flow.sync2.roster.entryHash
@@ -70,7 +71,7 @@ class FileLayout(
 
     /**
      * Copies every other sender's log objects after the last one [local] holds, reading the next `seq`
-     * until there is none (spec/store.md, WebDAV). A sender never read starts at its oldest object.
+     * until there is none, or a file that isn't that object (spec/store.md, WebDAV). A sender never read starts at its oldest object.
      * When the store tags folders, nothing is read while the tag stays the same.
      */
     suspend fun pullLogs(local: LocalStore) {
@@ -81,7 +82,10 @@ class FileLayout(
             var seq = tips[did]?.plus(1) ?: oldest(did) ?: continue
             while (true) {
                 val path = StorePaths.log(did, seq)
-                local.putObject(path, files.get(path) ?: break)
+                val bytes = files.get(path) ?: break
+                // A store that answers every path, with an error page say, would otherwise never end the log.
+                if (!namesLogObject(bytes, SyncBytes.fromHex(did), seq)) break
+                local.putObject(path, bytes)
                 seq++
             }
         }

@@ -82,10 +82,11 @@ class GroupStore(
         var next = cursor
         while (true) {
             val (changed, after) = server.changes(next)
-            if (changed.isEmpty()) return next
             for (change in changed.filter { StorePaths.isObject(it.path) }) {
                 if (local.getObject(change.path) == null) server.getObject(change.path)?.let { local.putObject(change.path, it) }
             }
+            // A cursor that doesn't move would hand out the same page forever.
+            if (changed.isEmpty() || after <= next) return next
             next = after
         }
     }

@@ -85,6 +85,41 @@ class FileLayoutTest {
             assertTrue("an unchanged Nextcloud folder isn't read again", server.requests.none { it.startsWith("GET flow/dev/") })
         }
 
+    @Test
+    fun a_store_that_answers_every_path_ends_a_log_at_the_first_file_that_isn_t_its_object() =
+        runTest {
+            val did = SyncBytes.toHex(newMember().did)
+            val answersEverything =
+                object : FileStore {
+                    private var gets = 0
+                    override val location = "memory"
+
+                    override suspend fun get(path: String): ByteArray {
+                        check(++gets < 100) { "the same file was read forever" }
+                        return "<html>Sign in</html>".toByteArray()
+                    }
+
+                    override suspend fun put(
+                        path: String,
+                        bytes: ByteArray,
+                    ) = Unit
+
+                    override suspend fun delete(path: String) = Unit
+
+                    override suspend fun list(folder: String) =
+                        when (folder) {
+                            "dev" -> listOf(did)
+                            "dev/$did/log" -> listOf("0000000000000000")
+                            else -> listOf("0000000000000001.o")
+                        }
+
+                    override suspend fun tag(folder: String): String? = null
+                }
+            val local = heldStore(LinkedGroup(this).deviceDb())
+            FileLayout(answersEverything, SyncBytes.toHex(newMember().did)).pullLogs(local)
+            assertTrue(local.logTips().isEmpty())
+        }
+
     @Test
     fun competing_roster_entries_both_stay_and_every_device_picks_the_same_winner() =
         runTest {

@@ -95,6 +95,17 @@ class GarbageCollectionTest {
         assertEquals(mapOf("a" to 1L..5L), plan(setOf("a", "b"), setOf("a"), acks, both, links).logs)
     }
 
+    @Test
+    fun a_store_whose_cursor_never_moves_still_lets_collection_finish() =
+        runTest {
+            val g = LinkedGroup(this).apply { create() }
+            val db = g.deviceDb()
+            val watched = mapOf(SyncCollection.WATCH_HISTORY to CollectionWire(listOf("""{"title":"t","videoId":"aaaaaaaaaaa"}"""), 1, ""))
+            Session.open(db, g.store, g.a).push(watched, LINK)
+            Session.open(db, StuckPages(g.store), g.a).collectGarbage(setOf(LINK))
+            assertTrue(StorePaths.log(SyncBytes.toHex(g.a.did), 1) in g.store.objects)
+        }
+
     @Test
     fun old_objects_go_once_acknowledged_and_covered_by_a_snapshot() =
         runTest {

@@ -80,6 +80,21 @@ class HeldStoreTest {
             assertEquals(Appended.Conflict(entryHash(winner)), stale)
         }
 
+    @Test
+    fun a_server_whose_cursor_never_moves_is_downloaded_once() =
+        runTest {
+            val g = LinkedGroup(this).apply { create() }
+            val a = SyncBytes.toHex(g.a.did)
+            Session.open(g.deviceDb(), g.store, g.a).push(
+                mapOf(SyncCollection.WATCH_HISTORY to CollectionWire(listOf("""{"title":"t","videoId":"aaaaaaaaaaa"}"""), 1, "")),
+                LINK,
+            )
+            val db = g.deviceDb()
+            val store = GroupStore(heldStore(db), StuckPages(g.store))
+            assertEquals(0L, store.download(SyncBytes.toHex(g.b.did), 0, objects = true))
+            assertTrue(store.local.getObject(StorePaths.log(a, 1)) != null)
+        }
+
     @Test
     fun a_server_group_syncs_through_what_each_device_holds() =
         runTest {

@@ -24,6 +24,7 @@ import io.github.aedev.flow.sync2.store.Appended
 import io.github.aedev.flow.sync2.store.ChangeList
 import io.github.aedev.flow.sync2.store.ChangedObject
 import io.github.aedev.flow.sync2.store.LocalStore
+import io.github.aedev.flow.sync2.store.Mailbox
 import io.github.aedev.flow.sync2.store.Remote
 import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.TestScope
@@ -124,6 +125,20 @@ class MemoryStore :
     }
 }
 
+/** A store whose cursor never moves: every page lists every object again. */
+class StuckPages(
+    private val store: MemoryStore,
+) : Remote,
+    ChangeList by store,
+    Mailbox by store {
+    private var pages = 0
+
+    override suspend fun changes(cursor: Long): Pair<List<ChangedObject>, Long> {
+        check(++pages < 100) { "the same page was read forever" }
+        return store.changes(0).first to cursor
+    }
+}
+
 fun wraps(
     group: ByteArray,
     epoch: Int,

@@ -110,6 +110,16 @@ class SessionTest {
             assertTrue(appB.watched.isEmpty())
         }
 
+    @Test
+    fun a_store_whose_cursor_never_moves_is_read_once() =
+        runTest {
+            val g = LinkedGroup(this).apply { create() }
+            sync(g.deviceDb(), g.store, g.a, App().apply { watch("aaaaaaaaaaa") })
+            val appB = App()
+            sync(g.deviceDb(), StuckPages(g.store), g.b, appB)
+            assertEquals(listOf("aaaaaaaaaaa"), appB.watched.keys.toList())
+        }
+
     @Test
     fun a_missing_object_stalls_reading_until_it_arrives() =
         runTest {