fix(sync): finish the sync when a store refuses an upload

.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

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<String?>()) { 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 }
         }
 
         /**

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?,
+)

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