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 7409a9731..ba298ca04 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 @@ -47,8 +47,9 @@ internal suspend fun Session.bootstrap(links: Set?): 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) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/engine/GarbageCollection.kt b/app/src/main/java/io/github/aedev/flow/sync2/engine/GarbageCollection.kt index 33a93a047..fcb6e6d6a 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/engine/GarbageCollection.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/engine/GarbageCollection.kt @@ -170,8 +170,9 @@ private suspend fun Session.heldPaths(): Set { 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 } } 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 fd4173877..d5e32512c 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 @@ -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) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/Envelope.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/Envelope.kt index 1bb80093b..fd05b6506 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/Envelope.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/Envelope.kt @@ -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, 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 5ffd133d2..349a1948f 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 @@ -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++ } } 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 b0b10f277..93ed3ef2a 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 @@ -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 } } diff --git a/app/src/test/java/io/github/aedev/flow/sync2/FileLayoutTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/FileLayoutTest.kt index 778b30bfc..3327fa8be 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/FileLayoutTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/FileLayoutTest.kt @@ -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 "Sign in".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 { diff --git a/app/src/test/java/io/github/aedev/flow/sync2/GarbageCollectionTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/GarbageCollectionTest.kt index 31da5d6cc..c16c1271b 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/GarbageCollectionTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/GarbageCollectionTest.kt @@ -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 { diff --git a/app/src/test/java/io/github/aedev/flow/sync2/HeldStoreTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/HeldStoreTest.kt index 7ead49202..8125d028b 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/HeldStoreTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/HeldStoreTest.kt @@ -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 { diff --git a/app/src/test/java/io/github/aedev/flow/sync2/LinkedGroup.kt b/app/src/test/java/io/github/aedev/flow/sync2/LinkedGroup.kt index 992faba54..aa60a6d69 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/LinkedGroup.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/LinkedGroup.kt @@ -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, Long> { + check(++pages < 100) { "the same page was read forever" } + return store.changes(0).first to cursor + } +} + fun wraps( group: ByteArray, epoch: Int, diff --git a/app/src/test/java/io/github/aedev/flow/sync2/SessionTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/SessionTest.kt index 215070adc..840ac8969 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/SessionTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/SessionTest.kt @@ -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 {