PM-46434: bug: Handle multiple sync requests

2026-11-09 10:34:18 -06:00 · 2026-11-06 14:57:06 -06:00
parent 63d1e6457b
commit d422051ccf
2 changed files with 166 additions and 20 deletions
@@ -44,9 +44,13 @@ import com.x8bit.bitwarden.data.vault.repository.util.toEncryptedSdkFolderList
 import com.x8bit.bitwarden.data.vault.repository.util.toEncryptedSdkSendList
 import com.x8bit.bitwarden.data.vault.repository.util.toV2UpgradeToken
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Job
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
@@ -63,6 +67,8 @@ import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import java.time.Clock
 import java.time.temporal.ChronoUnit

@@ -95,7 +101,20 @@ internal class VaultSyncManagerImpl(
    private val unconfinedScope = CoroutineScope(dispatcherManager.unconfined)
    private val ioScope = CoroutineScope(dispatcherManager.io)

-    private var syncJob: Job = Job().apply { complete() }
+    /**
+     * A scope used specifically to make sync requests, it uses a [SupervisorJob] to ensure a
+     * single failure does not mean prevent other operations from occurring.
+     */
+    private val syncScope = CoroutineScope(dispatcherManager.io + SupervisorJob())
+
+    /**
+     * Guards reading and replacing [syncDeferred] so that only one sync can be started at a time.
+     */
+    private val syncMutex: Mutex = Mutex()
+
+    private var syncDeferred: Deferred<SyncVaultDataResult> = CompletableDeferred(
+        value = SyncVaultDataResult.Error(throwable = NoActiveUserException()),
+    )

    private val activeUserId: String? get() = authDiskSource.userState?.activeUserId

@@ -174,7 +193,7 @@ internal class VaultSyncManagerImpl(
                },
        )
            .onEach {
-                syncJob.cancel()
+                syncMutex.withLock { syncDeferred.cancel() }
                clearUnlockedData()
            }
            .launchIn(unconfinedScope)
@@ -233,14 +252,18 @@ internal class VaultSyncManagerImpl(
    }

    override fun sync(forced: Boolean) {
-        val userId = activeUserId ?: return
-        if (!syncJob.isCompleted) return
-        mutableDecryptCipherListResultFlow.updateToPendingOrLoading()
-        mutableDomainsStateFlow.updateToPendingOrLoading()
-        mutableFoldersStateFlow.updateToPendingOrLoading()
-        mutableCollectionsStateFlow.updateToPendingOrLoading()
-        mutableSendDataStateFlow.updateToPendingOrLoading()
-        syncJob = ioScope.launch { syncInternal(userId = userId, forced = forced) }
+        ioScope.launch {
+            syncMutex.withLock {
+                if (!syncDeferred.isCompleted) return@withLock
+                val userId = activeUserId ?: return@withLock
+                mutableDecryptCipherListResultFlow.updateToPendingOrLoading()
+                mutableDomainsStateFlow.updateToPendingOrLoading()
+                mutableFoldersStateFlow.updateToPendingOrLoading()
+                mutableCollectionsStateFlow.updateToPendingOrLoading()
+                mutableSendDataStateFlow.updateToPendingOrLoading()
+                syncDeferred = syncScope.async { syncInternal(userId = userId, forced = forced) }
+            }
+        }
    }

    override fun syncIfNecessary() {
@@ -258,16 +281,28 @@ internal class VaultSyncManagerImpl(
    }

    override suspend fun syncForResult(forced: Boolean): SyncVaultDataResult {
-        val userId = activeUserId ?: return SyncVaultDataResult.Error(NoActiveUserException())
-        syncJob = ioScope
-            .async { syncInternal(userId = userId, forced = forced) }
-            .also {
-                return try {
-                    it.await()
-                } catch (e: CancellationException) {
-                    SyncVaultDataResult.Error(throwable = e)
-                }
+        val deferred = syncMutex.withLock {
+            if (syncDeferred.isCompleted) {
+                activeUserId
+                    ?.let { userId ->
+                        syncScope
+                            .async { syncInternal(userId = userId, forced = forced) }
+                            .also { syncDeferred = it }
+                    }
+                    ?: CompletableDeferred(
+                        value = SyncVaultDataResult.Error(throwable = NoActiveUserException()),
+                    )
+            } else {
+                syncDeferred
            }
+        }
+        return try {
+            deferred.await()
+        } catch (e: CancellationException) {
+            // Rethrow if the caller was canceled, otherwise the sync itself was canceled.
+            currentCoroutineContext().ensureActive()
+            SyncVaultDataResult.Error(throwable = e)
+        }
    }

    @Suppress("LongMethod")
@@ -77,14 +77,22 @@ import io.mockk.runs
 import io.mockk.slot
 import io.mockk.unmockkConstructor
 import io.mockk.verify
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.async
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.update
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import java.net.UnknownHostException
@@ -1448,6 +1456,109 @@ class VaultSyncManagerTest {
            verify(exactly = 0) { fillAssistManager.syncIfNecessary() }
        }

+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun `syncForResult should return error when the sync is cancelled by a user switch`() =
+        runTest {
+            fakeAuthDiskSource.userState = MOCK_USER_STATE
+            coEvery { syncService.sync() } just awaits
+
+            val result = async { vaultSyncManager.syncForResult() }
+            runCurrent()
+            fakeAuthDiskSource.userState = UserStateJson(
+                activeUserId = "mockId-2",
+                accounts = mapOf("mockId-2" to mockk()),
+            )
+
+            val error = result.await() as SyncVaultDataResult.Error
+            assertTrue(error.throwable is CancellationException)
+        }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun `syncForResult should propagate cancellation of the calling coroutine`() = runTest {
+        fakeAuthDiskSource.userState = MOCK_USER_STATE
+        coEvery { syncService.sync() } just awaits
+        var isResultReceived = false
+
+        val job = launch {
+            vaultSyncManager.syncForResult()
+            isResultReceived = true
+        }
+        runCurrent()
+        job.cancel()
+        runCurrent()
+
+        assertTrue(job.isCancelled)
+        assertFalse(isResultReceived)
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun `sync should not start a new sync while syncForResult is in progress`() = runTest {
+        fakeAuthDiskSource.userState = MOCK_USER_STATE
+        coEvery { syncService.sync() } just awaits
+
+        backgroundScope.launch { vaultSyncManager.syncForResult() }
+        runCurrent()
+        vaultSyncManager.sync()
+
+        coVerify(exactly = 1) { syncService.sync() }
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun `syncForResult should join an in-progress sync and return its result`() =
+        runTest {
+            fakeAuthDiskSource.userState = MOCK_USER_STATE
+            val throwable = Throwable()
+            val syncResponse = CompletableDeferred<Result<SyncResponseJson>>()
+            coEvery { syncService.sync() } coAnswers { syncResponse.await() }
+
+            vaultSyncManager.sync()
+            val result = async { vaultSyncManager.syncForResult() }
+            runCurrent()
+            coVerify(exactly = 1) { syncService.sync() }
+
+            syncResponse.complete(throwable.asFailure())
+
+            assertEquals(SyncVaultDataResult.Error(throwable = throwable), result.await())
+            coVerify(exactly = 1) { syncService.sync() }
+        }
+
+    @Test
+    fun `syncForResult should return error when there is no active user`() = runTest {
+        fakeAuthDiskSource.userState = null
+
+        val result = vaultSyncManager.syncForResult()
+
+        assertEquals(SyncVaultDataResult.Error(throwable = NoActiveUserException()), result)
+        coVerify(exactly = 0) {
+            syncService.getAccountRevisionDateMillis()
+            syncService.sync()
+        }
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun `syncForResult should not sync when the in-progress sync is cancelled by a user switch`() =
+        runTest {
+            fakeAuthDiskSource.userState = MOCK_USER_STATE
+            coEvery { syncService.sync() } just awaits
+
+            vaultSyncManager.sync()
+            val result = async { vaultSyncManager.syncForResult() }
+            runCurrent()
+            fakeAuthDiskSource.userState = UserStateJson(
+                activeUserId = "mockId-2",
+                accounts = mapOf("mockId-2" to mockk()),
+            )
+
+            val error = result.await() as SyncVaultDataResult.Error
+            assertTrue(error.throwable is CancellationException)
+            coVerify(exactly = 1) { syncService.sync() }
+        }
+
    //region Helper functions

    /**