Never cut a sync push between its chunks

app/src/main/java/com/pombo/android/AppViewModel.kt

@@ -1628,6 +1628,7 @@ class AppViewModel(app: Application) : AndroidViewModel(app), PomboBridge.Listen
         contactsStore.memoryOnly = guest
         settingsStore.scopeAddress = address
         syncStore.scopeAddress = address
+        sync.cancelAutoPush()
         sync.cancelPushConfirmation()
         epochKeyStore.scopeAddress = address
         epochKeyStore.memoryOnly = guest
@@ -2354,6 +2355,7 @@ class AppViewModel(app: Application) : AndroidViewModel(app), PomboBridge.Listen
 
     fun disconnect() {
         manager.closeCurrent()
+        sync.cancelAutoPush()
         store.clear()
         _accounts.value = store.accounts()
         _address.value = store.address

app/src/main/java/com/pombo/android/SyncManager.kt

@@ -92,6 +92,13 @@ class SyncManager(
     private var autoPushJob: Job? = null
     private var autoPushRetries = 0
     private var pushQueued = false
+    /**
+     * An auto push that has started publishing is never cancelled: a run cut
+     * between its chunks stays on storage and never joins.
+     */
+    @Volatile private var autoPushPublishing = false
+    /** The delay a change asked for while the auto push was publishing. */
+    @Volatile private var queuedDelayMs: Long? = null
     /** Bumped on every change; a push clears `dirty` only if none came after its export. */
     private val changeSeq = java.util.concurrent.atomic.AtomicLong()
     /**
@@ -114,37 +121,60 @@ class SyncManager(
         changeSeq.incrementAndGet()
         store.dirty = true
         if (syncMode() == com.pombo.android.data.SyncMode.MANUAL_ONLY) return
+        if (autoPushPublishing) {
+            queuedDelayMs = delayMs
+            return
+        }
         autoPushJob?.cancel()
         autoPushJob = scope.launch {
             delay(delayMs)
-            try {
+            val owner = myAddress()?.lowercase()
+            autoPushPublishing = true
+            val failed = try {
                 if (pushSync() != null) autoPushRetries = 0
+                false
             } catch (e: Exception) {
+                true
+            } finally {
+                autoPushPublishing = false
+            }
+            val queued = queuedDelayMs
+            queuedDelayMs = null
+            if (owner == null || accountChanged(owner)) return@launch
+            if (failed) {
                 autoPushRetries++
                 // Web: exponential backoff capped at five minutes.
                 val retry = minOf(autoPushDelayMs * (1L shl autoPushRetries), 300_000L)
                 scheduleAutoPush(retry)
+            } else if (queued != null) {
+                scheduleAutoPush(queued)
             }
         }
     }
 
     fun cancelAutoPush() {
-        autoPushJob?.cancel()
-        autoPushJob = null
+        if (!autoPushPublishing) {
+            autoPushJob?.cancel()
+            autoPushJob = null
+        }
         autoPushRetries = 0
         pushQueued = false
+        queuedDelayMs = null
     }
 
     /** Publishes the local snapshot. Returns the payload timestamp, or null if skipped. */
     suspend fun pushSync(): Long? {
         if (isGuest()) return null
         if (_syncing.value) { pushQueued = true; return null }
-        val inbox = inboxId() ?: return null
+        val me = myAddress()?.lowercase() ?: return null
+        val key = myPrivateKey() ?: return null
+        val inbox = "$me/Pombo-DM-1"
         if (!inboxExists()) return null
 
         _syncing.value = true
         try {
             val seq = changeSeq.get()
+            if (accountChanged(me)) return null
             val data = exportLocal()
             val hash = stateHash(data)
             if (isConfirmedState(hash) || hash == pendingHash) {
@@ -171,13 +201,15 @@ class SyncManager(
             publishing = true
             try {
                 for (message in messages) {
-                    sealPublishToSelf(inbox, StreamConstants.P_SYNC, message)?.let {
+                    if (accountChanged(me)) return null
+                    sealPublishToSelf(inbox, StreamConstants.P_SYNC, message, key, me).let {
                         rows.add(rowKey(it.optLong("timestamp"), it.optString("publisherId")))
                     }
                 }
             } finally {
                 publishing = false
             }
+            if (accountChanged(me)) return null
             noteOwnRows(rows)
             confirmPush(inbox, hash, rows)
 
@@ -193,6 +225,13 @@ class SyncManager(
         }
     }
 
+    /**
+     * A push started for one account stops when another takes over: what is
+     * left of it, and the store it would write to, are no longer that account's.
+     */
+    private fun accountChanged(address: String): Boolean =
+        (myAddress()?.lowercase() != address).also { if (it) Log.d(TAG, "push stopped: the account changed") }
+
     private fun stateHash(data: JSONObject): String =
         java.security.MessageDigest.getInstance("SHA-256")
             .digest(com.pombo.android.core.SyncStateKey.key(data).toByteArray(Charsets.UTF_8))
@@ -435,6 +474,10 @@ class SyncManager(
         // Backgrounding is still an unprompted publish. `dirty` survives, so the
         // next manual run carries the change.
         if (syncMode() == com.pombo.android.data.SyncMode.MANUAL_ONLY) return
+        if (autoPushPublishing) {
+            queuedDelayMs = 0L
+            return
+        }
         cancelAutoPush()
         scope.launch {
             try { pushSync() } catch (e: Exception) { /* dirty survives for the next run */ }
@@ -448,20 +491,21 @@ class SyncManager(
     /** Publishes locally-held images that no device has synced yet. */
     suspend fun pushImageBlobs() {
         if (isGuest() || _syncing.value) return
-        val inbox = inboxId() ?: return
+        val me = myAddress()?.lowercase() ?: return
+        val key = myPrivateKey() ?: return
+        val inbox = "$me/Pombo-DM-1"
         val pending = blobStore.unsynced()
         if (pending.isEmpty()) return
         if (!inboxExists()) return
 
-        if (myAddress() == null) return
-
         // Sealed-to-self v2 under a throwaway publisher, one seal per message.
         suspend fun publish(payload: JSONObject) {
-            sealPublishToSelf(inbox, StreamConstants.P_SYNC_BLOBS, payload)
+            sealPublishToSelf(inbox, StreamConstants.P_SYNC_BLOBS, payload, key, me)
         }
 
         var published = 0
         for (record in pending) {
+            if (accountChanged(me)) break
             val data = blobStore.load(record.imageId)
             if (data.isNullOrEmpty()) {
                 // The blob is gone but the ledger still lists it; drop the entry
@@ -650,12 +694,16 @@ class SyncManager(
      * WebView; the throwaway key crosses because it is also the publishing
      * identity (bridge publishAs).
      */
-    private suspend fun sealPublishToSelf(streamId: String, partition: Int, payload: JSONObject): JSONObject? {
-        val myPk = myPrivateKey() ?: return null
-        val me = myAddress()?.lowercase() ?: return null
-        val recipientPub = com.pombo.android.core.EthereumSigner.compressedPublicKey(myPk)
+    private suspend fun sealPublishToSelf(
+        streamId: String,
+        partition: Int,
+        payload: JSONObject,
+        privateKey: String,
+        address: String
+    ): JSONObject {
+        val recipientPub = com.pombo.android.core.EthereumSigner.compressedPublicKey(privateKey)
         val (envelope, ephemeralPk) = kotlinx.coroutines.withContext(kotlinx.coroutines.Dispatchers.Default) {
-            com.pombo.android.core.SealedSenderCrypto.seal(payload, myPk, me, recipientPub)
+            com.pombo.android.core.SealedSenderCrypto.seal(payload, privateKey, address, recipientPub)
         }
         // The ceiling applies to the envelope, not the payload: JSON-escaping
         // the slice and base64 of the ciphertext both inflate it. Measured, a

app/src/test/java/com/pombo/android/SyncManagerLightTest.kt

@@ -7,6 +7,7 @@ import com.pombo.android.data.SyncStore
 import io.mockk.coEvery
 import io.mockk.every
 import io.mockk.mockk
+import io.mockk.verify
 import java.util.concurrent.CopyOnWriteArrayList
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -35,6 +36,9 @@ class SyncManagerLightTest {
     @Volatile private var rows = JSONArray()
     @Volatile private var resendMs = 0L
     @Volatile private var onPublish: () -> Unit = {}
+    @Volatile private var export = JSONObject("""{"channels":[],"username":"Bob","sliceTs":{"username":1}}""")
+    @Volatile private var address = ME
+    @Volatile private var key = PK
     private var mode = SyncMode.AUTOMATIC
     private val calls = CopyOnWriteArrayList<String>()
     private val left = CopyOnWriteArrayList<Int>()
@@ -84,8 +88,8 @@ class SyncManagerLightTest {
         }
         sync = SyncManager(
             bridge, scope, store, blobStore,
-            myAddress = { ME }, myPrivateKey = { PK }, isGuest = { false },
-            exportLocal = { JSONObject("""{"channels":[],"username":"Bob","sliceTs":{"username":1}}""") },
+            myAddress = { address }, myPrivateKey = { key }, isGuest = { false },
+            exportLocal = { JSONObject(export.toString()) },
             importMerged = {}, syncMode = { mode },
             checkIntervalMs = 50L, confirmAtMs = longArrayOf(10L, 20L, 30L, 40L), blobLeaveAfterMs = 10L
         )
@@ -135,6 +139,112 @@ class SyncManagerLightTest {
         assertTrue(dirty)
     }
 
+    /** A state big enough to go out as a run of several chunks. */
+    private fun bigState(name: String) = JSONObject().put("channels", JSONArray()).put("username", name)
+        .put("note", name.repeat(400_000 / name.length))
+
+    @Test
+    fun `a change during an auto push lets it finish, and goes out after it`() {
+        rows = JSONArray().put(row(5000, "0xeph"))
+        export = bigState("first")
+        val runSize = com.pombo.android.core.SyncChunks.split(JSONObject().put("type", "sync").put("v", 1)
+            .put("ts", System.currentTimeMillis()).put("data", export), "0123456789abcdef").size
+        assertTrue(runSize >= 3)
+        val inFlight = kotlinx.coroutines.CompletableDeferred<Unit>()
+        val published = java.util.concurrent.atomic.AtomicInteger()
+        coEvery { bridge.call("publishAs", any(), any()) } coAnswers {
+            if (published.incrementAndGet() == 2) inFlight.await()
+            JSONObject().put("ok", true).put("timestamp", 5000L).put("publisherId", "0xEph")
+        }
+
+        sync.scheduleAutoPush(10L)
+        waitFor { published.get() == 2 }
+        export = bigState("second")
+        sync.scheduleAutoPush(10L)
+        inFlight.complete(Unit)
+        waitFor { published.get() >= 2 * runSize && !dirty }
+
+        assertEquals(2 * runSize, published.get())
+        assertFalse(dirty)
+    }
+
+    @Test
+    fun `an auto push that fails midway is retried with backoff, and a change queued meanwhile adds no push`() {
+        rows = JSONArray().put(row(5000, "0xeph"))
+        export = bigState("first")
+        val inFlight = kotlinx.coroutines.CompletableDeferred<Unit>()
+        val published = java.util.concurrent.atomic.AtomicInteger()
+        coEvery { bridge.call("publishAs", any(), any()) } coAnswers {
+            if (published.incrementAndGet() == 2) {
+                inFlight.await()
+                throw IllegalStateException("publish failed")
+            }
+            JSONObject().put("ok", true).put("timestamp", 5000L).put("publisherId", "0xEph")
+        }
+
+        sync.scheduleAutoPush(10L)
+        waitFor { published.get() == 2 }
+        export = bigState("second")
+        sync.scheduleAutoPush(10L)
+        inFlight.complete(Unit)
+        // The queued change asked for 10 ms; the retry after a failure waits 30 s.
+        Thread.sleep(500)
+
+        assertEquals(2, published.get())
+        assertTrue(dirty)
+        sync.cancelAutoPush()
+    }
+
+    @Test
+    fun `an account switch during a push stops it before the next chunk and leaves the store alone`() {
+        rows = JSONArray().put(row(5000, "0xeph"))
+        export = bigState("first")
+        val inFlight = kotlinx.coroutines.CompletableDeferred<Unit>()
+        val published = java.util.concurrent.atomic.AtomicInteger()
+        coEvery { bridge.call("publishAs", any(), any()) } coAnswers {
+            if (published.incrementAndGet() == 2) inFlight.await()
+            JSONObject().put("ok", true).put("timestamp", 5000L).put("publisherId", "0xEph")
+        }
+
+        sync.scheduleAutoPush(10L)
+        waitFor { published.get() == 2 }
+        address = OTHER
+        key = OTHER_PK
+        inFlight.complete(Unit)
+        waitFor { !sync.syncing.value }
+
+        assertEquals(2, published.get())
+        assertTrue(dirty)
+        assertNull(confirmedHash)
+        verify(exactly = 0) { store.recordApplied(any()) }
+        verify(exactly = 0) { store.lastSyncTs = any() }
+    }
+
+    @Test
+    fun `a push that fails because the account switched is not retried for the new account`() {
+        rows = JSONArray().put(row(5000, "0xeph"))
+        export = bigState("first")
+        val inFlight = kotlinx.coroutines.CompletableDeferred<Unit>()
+        val published = java.util.concurrent.atomic.AtomicInteger()
+        coEvery { bridge.call("publishAs", any(), any()) } coAnswers {
+            if (published.incrementAndGet() == 2) {
+                inFlight.await()
+                throw IllegalStateException("bridge reloaded")
+            }
+            JSONObject().put("ok", true).put("timestamp", 5000L).put("publisherId", "0xEph")
+        }
+
+        sync.scheduleAutoPush(10L)
+        waitFor { published.get() == 2 }
+        address = OTHER
+        key = OTHER_PK
+        inFlight.complete(Unit)
+        waitFor { !sync.syncing.value }
+
+        verify(exactly = 1) { store.dirty = true }
+        assertEquals(2, published.get())
+    }
+
     @Test
     fun `an unconfirmed push leaves too and its state is sent again`() = runBlocking {
         rows = JSONArray()
@@ -216,5 +326,7 @@ class SyncManagerLightTest {
     private companion object {
         const val PK = "0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d"
         const val ME = "0x70997970c51812dc3a010c7d01b50e0d17dc79c8"
+        const val OTHER_PK = "0x5de4111afa1a4b94908f83103eb1f1706367c2e68ca870fc3fb9a804cdab365a"
+        const val OTHER = "0x3c44cdddb6a900fa2b585dd299e03d12fa4293bc"
     }
 }