From d422051ccf7e784bb8f32e720fed3c487ede2f27 Mon Sep 17 00:00:00 2001 From: David Perez Date: Thu, 5 Nov 2026 13:20:51 -0600 Subject: [PATCH] PM-46434: bug: Handle multiple sync requests --- .../vault/manager/VaultSyncManagerImpl.kt | 75 ++++++++---- .../vault/manager/VaultSyncManagerTest.kt | 111 ++++++++++++++++++ 2 files changed, 166 insertions(+), 20 deletions(-) diff --git a/app/src/main/kotlin/com/x8bit/bitwarden/data/vault/manager/VaultSyncManagerImpl.kt b/app/src/main/kotlin/com/x8bit/bitwarden/data/vault/manager/VaultSyncManagerImpl.kt index e88a269b18..05cb58d90a 100644 --- a/app/src/main/kotlin/com/x8bit/bitwarden/data/vault/manager/VaultSyncManagerImpl.kt +++ b/app/src/main/kotlin/com/x8bit/bitwarden/data/vault/manager/VaultSyncManagerImpl.kt @@ -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 = 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") diff --git a/app/src/test/kotlin/com/x8bit/bitwarden/data/vault/manager/VaultSyncManagerTest.kt b/app/src/test/kotlin/com/x8bit/bitwarden/data/vault/manager/VaultSyncManagerTest.kt index 0c1387f39f..e57f678271 100644 --- a/app/src/test/kotlin/com/x8bit/bitwarden/data/vault/manager/VaultSyncManagerTest.kt +++ b/app/src/test/kotlin/com/x8bit/bitwarden/data/vault/manager/VaultSyncManagerTest.kt @@ -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>() + 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 /**