From 7f26a6f37e4f639eb36190e9f617a541751f308f Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 7 Nov 2026 15:49:00 +0200 Subject: [PATCH] fix(sync): finish the sync when a store refuses an upload --- .../2026-11-07-fix-sync2-upload-refusal.md | 2 + .../aedev/flow/sync2/service/LinkedSync.kt | 5 +- .../aedev/flow/sync2/store/GroupStore.kt | 23 +++- .../aedev/flow/sync2/UploadRefusalTest.kt | 105 ++++++++++++++++++ 4 files changed, 129 insertions(+), 6 deletions(-) create mode 100644 .github/nightly-notes/2026-11-07-fix-sync2-upload-refusal.md create mode 100644 app/src/test/java/io/github/aedev/flow/sync2/UploadRefusalTest.kt diff --git a/.github/nightly-notes/2026-11-07-fix-sync2-upload-refusal.md b/.github/nightly-notes/2026-11-07-fix-sync2-upload-refusal.md new file mode 100644 index 000000000..22113bca7 --- /dev/null +++ b/.github/nightly-notes/2026-11-07-fix-sync2-upload-refusal.md @@ -0,0 +1,2 @@ +## Fixed +- Linked devices: when a store refuses one change, the device still finishes the rest of its sync and shows the error diff --git a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt index ce6e3476e..f55e5763f 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt @@ -233,12 +233,15 @@ class LinkedSync exchange(session, links, announce) if (offline != null) throw offline session.collectGarbage(links.mapTo(HashSet()) { it.link } + null) - db.groups().setUploaded(store.upload(me, db.groups().group()?.uploaded ?: 0)) + val uploaded = store.upload(me, db.groups().group()?.uploaded ?: 0) + db.groups().setUploaded(uploaded.cursor) // The sync is done; a failed look for devices asking to join only waits for the next one. try { server(record, device)?.joinRequests()?.firstOrNull()?.let { NotificationHelper.showJoinRequest(context, it.name) } } catch (_: StoreException) { } + // Everything else in this sync is done; the user still has to hear that the device's changes aren't going out. + uploaded.refused?.let { throw it } } /** 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..40e7f2a9a 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 @@ -136,20 +136,33 @@ class GroupStore( return true } - /** Sends the server this device's own objects held after [heldCursor]; returns how far it got. */ + /** + * Sends the server this device's own objects held after [heldCursor], in order. The first one it refuses + * ends the upload: skipping it would leave a gap other devices wait at. + */ suspend fun upload( me: String, heldCursor: Long, - ): Long { - val server = server ?: return heldCursor + ): Uploaded { + val server = server ?: return Uploaded(heldCursor, null) var next = heldCursor while (true) { val (changed, after) = local.changes(next) - if (changed.isEmpty()) return next + if (changed.isEmpty()) return Uploaded(next, null) for (change in changed.filter { it.path.startsWith("dev/$me/") }) { - local.getObject(change.path)?.let { server.putObject(change.path, it) } + try { + local.getObject(change.path)?.let { server.putObject(change.path, it) } + } catch (e: StoreException.Refused) { + return Uploaded(next, e) + } } next = after } } } + +/** How far an upload got, and the refusal that ended it early. */ +class Uploaded( + val cursor: Long, + val refused: StoreException.Refused?, +) diff --git a/app/src/test/java/io/github/aedev/flow/sync2/UploadRefusalTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/UploadRefusalTest.kt new file mode 100644 index 000000000..217d67d20 --- /dev/null +++ b/app/src/test/java/io/github/aedev/flow/sync2/UploadRefusalTest.kt @@ -0,0 +1,105 @@ +package io.github.aedev.flow.sync2 + +import android.app.Application +import androidx.test.ext.junit.runners.AndroidJUnit4 +import io.github.aedev.flow.sync2.store.ChangeList +import io.github.aedev.flow.sync2.store.GroupStore +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.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 + +/** A store that won't take an object must not end a sync, and the device's log keeps its order. */ +@RunWith(AndroidJUnit4::class) +@Config(sdk = [34], application = Application::class) +class UploadRefusalTest { + private val me = "ab".repeat(16) + + private fun own(seq: Long) = StorePaths.log(me, seq) + + /** Answers with [refusal] for the object at [refused]. */ + private class Picky( + private val real: MemoryStore, + var refused: String?, + private val refusal: StoreException = StoreException.Refused(403, "forbidden"), + ) : Remote, + ChangeList by real, + Mailbox by real { + val tried = ArrayList() + + override suspend fun putObject( + path: String, + bytes: ByteArray, + ) { + tried += path + if (path == refused) throw refusal + real.putObject(path, bytes) + } + } + + @Test + fun a_refused_object_ends_the_upload_there_and_is_reported_not_thrown() = + runTest { + val local = heldStore(emptyDb()) + (1L..3L).forEach { local.putObject(own(it), byteArrayOf(it.toByte())) } + val remote = Picky(MemoryStore(), own(2)) + + val uploaded = GroupStore(local, remote).upload(me, 0) + + assertNotNull(uploaded.refused) + assertEquals("nothing after the refused object goes out", listOf(own(1), own(2)), remote.tried) + assertEquals("the cursor stays before the refused object", 0, uploaded.cursor) + } + + @Test + fun the_refused_object_goes_out_first_once_the_store_takes_it() = + runTest { + val local = heldStore(emptyDb()) + (1L..3L).forEach { local.putObject(own(it), byteArrayOf(it.toByte())) } + val real = MemoryStore() + val remote = Picky(real, own(2)) + val store = GroupStore(local, remote) + + val stuck = store.upload(me, 0) + remote.refused = null + val done = store.upload(me, stuck.cursor) + + assertNull(done.refused) + assertTrue(done.cursor > 0) + assertEquals(setOf(own(1), own(2), own(3)), real.objects.keys) + } + + @Test + fun pages_before_the_refused_one_are_kept_as_uploaded() = + runTest { + val local = heldStore(emptyDb()) + val pastThePage = 501L + (1L..pastThePage).forEach { local.putObject(own(it), byteArrayOf(1)) } + val remote = Picky(MemoryStore(), own(pastThePage)) + + val uploaded = GroupStore(local, remote).upload(me, 0) + + assertNotNull(uploaded.refused) + assertTrue("a full page went out and need not go again", uploaded.cursor > 0) + } + + @Test + fun a_refused_login_still_ends_the_sync() = + runTest { + val local = heldStore(emptyDb()) + local.putObject(own(1), byteArrayOf(1)) + val remote = Picky(MemoryStore(), own(1), StoreException.LoginRefused()) + + val outcome = runCatching { GroupStore(local, remote).upload(me, 0) } + + assertTrue(outcome.exceptionOrNull() is StoreException.LoginRefused) + } +}