Replace thread.sleep in sync exchange v2 with coroutine-based delay

Author: CDRussell

sync/sync-impl/src/main/java/com/duckduckgo/sync/impl/RealSyncCodeDispatcher.kt

@@ -125,7 +125,7 @@ class RealSyncCodeDispatcher @Inject constructor(
      * or null for intermediate events the caller should ignore. Both Host and Joiner roles are
      * reachable here; mirrors [mapV2LinkingEventToOutcome] (login on Joiner.Done, preserve abort reason).
      */
-    private fun mapV2PresentEventToOutcome(event: ExchangeV2Event, peerKind: PeerKind?): DispatchOutcome? = when (event) {
+    private suspend fun mapV2PresentEventToOutcome(event: ExchangeV2Event, peerKind: PeerKind?): DispatchOutcome? = when (event) {
         is ExchangeV2Event.SessionStarted -> event.linkingCode?.let { DispatchOutcome.LinkingCodeReady(it) }
         is ExchangeV2Event.Transition -> when (event.to) {
             ExchangeV2State.Joiner.Confirming ->
@@ -321,7 +321,7 @@ class RealSyncCodeDispatcher @Inject constructor(
      * Translate one runner event into a terminal [DispatchOutcome] for the v2 scanner flow,
      * or null for intermediate events the caller should ignore.
      */
-    private fun mapV2LinkingEventToOutcome(event: ExchangeV2Event, peerKind: PeerKind?): DispatchOutcome? = when (event) {
+    private suspend fun mapV2LinkingEventToOutcome(event: ExchangeV2Event, peerKind: PeerKind?): DispatchOutcome? = when (event) {
         is ExchangeV2Event.Transition -> when (event.to) {
             ExchangeV2State.Joiner.Confirming ->
                 DispatchOutcome.JoinerConfirmationRequested(peerName = runner.peerName, peerKind = peerKind)
@@ -376,7 +376,7 @@ class RealSyncCodeDispatcher @Inject constructor(
      * cid=ddg logs in directly; cid=3party upgrades the account via
      * [SyncAccountRepository.joinAccountFromThirdPartyRecoveryCode].
      */
-    private fun loginWithV2RecoveryCode(b64: String, peerKind: PeerKind?): DispatchOutcome {
+    private suspend fun loginWithV2RecoveryCode(b64: String, peerKind: PeerKind?): DispatchOutcome {
         val parsed = decodeV2Recovery(b64) ?: return DispatchOutcome.Failed("Couldn't parse received recovery code as v2.0")
         return when (parsed.cid) {
             CID_DDG -> {

sync/sync-impl/src/main/java/com/duckduckgo/sync/impl/SyncAccountRepository.kt

@@ -53,6 +53,7 @@ import com.squareup.anvil.annotations.*
 import com.squareup.moshi.*
 import dagger.*
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import logcat.LogPriority.ERROR
@@ -96,7 +97,7 @@ interface SyncAccountRepository {
      * adopted locally; otherwise a new one is created on the server. Either path leaves the
      * local store ready for [getThirdPartyRecoveryCode].
      */
-    fun createThirdPartyCredential(): Result<Boolean>
+    suspend fun createThirdPartyCredential(): Result<Boolean>
 
     /**
      * Fetches the 3party credential from the server, decrypts the SP using the account's secretKey,
@@ -124,7 +125,7 @@ interface SyncAccountRepository {
      * populated, credentialId=ddg, scopedPassword populated with SP. SyncStore is written
      * atomically only after both network calls succeed; observers never see an intermediate state.
      */
-    fun joinAccountFromThirdPartyRecoveryCode(pastedCode: String): Result<Boolean>
+    suspend fun joinAccountFromThirdPartyRecoveryCode(pastedCode: String): Result<Boolean>
 
     /**
      * Returns a recovery code that a 3rd-party browser can use to sign in and access this
@@ -631,7 +632,7 @@ class AppSyncAccountRepository @Inject constructor(
         }
     }
 
-    override fun createThirdPartyCredential(): Result<Boolean> = thirdPartyCredentialManager.create()
+    override suspend fun createThirdPartyCredential(): Result<Boolean> = thirdPartyCredentialManager.create()
 
     override fun refreshThirdPartyCredential(): Result<Boolean> = thirdPartyCredentialManager.refresh()
 
@@ -774,7 +775,7 @@ class AppSyncAccountRepository @Inject constructor(
         )
     }
 
-    override fun joinAccountFromThirdPartyRecoveryCode(pastedCode: String): Result<Boolean> {
+    override suspend fun joinAccountFromThirdPartyRecoveryCode(pastedCode: String): Result<Boolean> {
         if (!syncFeature.canUseV2ConnectFlow().isEnabled()) {
             return Error(reason = "JoinFrom3party: canUseV2ConnectFlow is disabled")
         }
@@ -1239,7 +1240,7 @@ class AppSyncAccountRepository @Inject constructor(
      * Returns the [LoginResponse] for the caller to commit alongside the new local key
      * material.
      */
-    private fun performDdgLoginForUpgrade(
+    private suspend fun performDdgLoginForUpgrade(
         userId: String,
         deviceId: String,
         deviceName: String,
@@ -1417,15 +1418,15 @@ class AppSyncAccountRepository @Inject constructor(
         return this
     }
 
-    private fun <T> retryingOnTransientError(block: () -> Result<T>): Result<T> {
+    private suspend fun <T> retryingOnTransientError(block: () -> Result<T>): Result<T> {
         var attempt = 0
         while (true) {
             val result = block()
             val code = (result as? Error)?.code
             if (code != null && isRetryableTransient(code) && attempt < MAX_UPGRADE_RETRIES) {
                 attempt++
                 logcat { "Sync-ScopedToken: upgrade call transient error (code=$code); retry $attempt/$MAX_UPGRADE_RETRIES" }
-                runCatching { Thread.sleep(upgradeRetryDelayMillis * attempt) }
+                delay(upgradeRetryDelayMillis * attempt)
                 continue
             }
             return result

sync/sync-impl/src/main/java/com/duckduckgo/sync/impl/ThirdPartyCredentialManager.kt

@@ -30,6 +30,7 @@ import com.duckduckgo.sync.store.SyncStore
 import com.squareup.anvil.annotations.ContributesBinding
 import com.squareup.moshi.Moshi
 import dagger.SingleInstanceIn
+import kotlinx.coroutines.delay
 import logcat.LogPriority.ERROR
 import logcat.logcat
 import javax.inject.Inject
@@ -46,7 +47,7 @@ interface ThirdPartyCredentialManager {
      * already created it, the existing credential is adopted locally; otherwise a new one is created
      * on the server. Either path leaves [SyncStore.scopedPassword] populated.
      */
-    fun create(): Result<Boolean>
+    suspend fun create(): Result<Boolean>
 
     /**
      * Fetches the 3party credential from the server, decrypts its SP using the account's primary
@@ -79,7 +80,7 @@ class RealThirdPartyCredentialManager @Inject constructor(
     @VisibleForTesting
     internal var rateLimitRetryDelayMillis: Long = DEFAULT_RATE_LIMIT_RETRY_DELAY_MILLIS
 
-    override fun create(): Result<Boolean> {
+    override suspend fun create(): Result<Boolean> {
         val inputs = when (val r = validateCreatePreconditions()) {
             is Success -> r.data
             is Error -> return r
@@ -151,7 +152,7 @@ class RealThirdPartyCredentialManager @Inject constructor(
         return AdoptResult.Adopted
     }
 
-    private fun mintAndPostNewCredential(inputs: CreateInputs): Result<Boolean> {
+    private suspend fun mintAndPostNewCredential(inputs: CreateInputs): Result<Boolean> {
         logcat { "Sync-ScopedToken: generating new 3party credential" }
 
         // Re-auth against the existing ddg credential's twice_hashed_password.
@@ -214,7 +215,7 @@ class RealThirdPartyCredentialManager @Inject constructor(
      *   - 3party-side: RFC 7516 JWE compact, kid="3party", AES-256-GCM,
      *     key = HKDF(SP, salt=user_id, info="Main Key"). Cross-platform readable.
      */
-    private fun reEncryptExistingDdgKeysFor3party(
+    private suspend fun reEncryptExistingDdgKeysFor3party(
         token: String,
         newSpBase64: String,
         hkdfSalt: ByteArray,
@@ -398,7 +399,7 @@ class RealThirdPartyCredentialManager @Inject constructor(
      * Retry an idempotent scoped-credential call on rate-limit.
      * A bounded linear backoff recovers since the limit is per-window and the call is idempotent.
      */
-    private fun <T> retryingOnRateLimit(block: () -> Result<T>): Result<T> {
+    private suspend fun <T> retryingOnRateLimit(block: () -> Result<T>): Result<T> {
         var attempt = 0
         while (true) {
             val result = block()
@@ -407,7 +408,7 @@ class RealThirdPartyCredentialManager @Inject constructor(
             if (rateLimited && attempt < MAX_RATE_LIMIT_RETRIES) {
                 attempt++
                 logcat { "Sync-ScopedToken: rate-limited (code=$code); retry $attempt/$MAX_RATE_LIMIT_RETRIES" }
-                runCatching { Thread.sleep(rateLimitRetryDelayMillis * attempt) }
+                delay(rateLimitRetryDelayMillis * attempt)
                 continue
             }
             return result

sync/sync-impl/src/main/java/com/duckduckgo/sync/impl/exchange/v2/ExchangeV2Runner.kt

@@ -431,10 +431,18 @@ class RealExchangeV2Runner @Inject constructor(
     /**
      * Composite of [SideEffect.RequestRecoveryCodeShare]: fetch + send the recovery code
      * (or `recovery_code_unavailable` if it can't be produced), then advance the SM.
+     *
+     * The fetch+send runs inside the launched coroutine so the underlying retry backoff can use
+     * `delay` rather than blocking the dispatcher thread. This means the SM mutex is NOT held
+     * while the network work runs, so a concurrent [cancelLocked] (user back-out / 5-min timeout /
+     * explicit [cancel]) can race with this body. The short-circuit below catches the common case;
+     * any work that proceeds past it is best-effort, and any orphaned server-side credential is
+     * cleaned up by the BE's 5-min TTL (Unified Algorithm, Backend-supported Alternative).
      */
     private fun shareRecoveryCodeAndAdvanceLocked() {
-        val responseOk = sendRecoveryCodeResponse()
         appScope.launch(dispatchers.io()) {
+            if (session == null) return@launch
+            val responseOk = sendRecoveryCodeResponse()
             mutex.withLock {
                 val s = session ?: return@withLock
                 if (s.currentState != ExchangeV2State.Host.Sending) return@withLock
@@ -657,7 +665,7 @@ class RealExchangeV2Runner @Inject constructor(
      * `recovery_code_unavailable` to the peer + emitted a SessionError event, and the SM
      * should be driven to [ExchangeV2State.Host.Aborted] rather than Done).
      */
-    private fun sendRecoveryCodeResponse(): Boolean {
+    private suspend fun sendRecoveryCodeResponse(): Boolean {
         val peer = peerChannelId ?: return false
         val peerKey = peerPublicKey ?: return false
         // Recovery code per peer kind. Per spec §"Exchange Share Recovery Code": create the host
@@ -703,7 +711,7 @@ class RealExchangeV2Runner @Inject constructor(
             }
         }
 
-    private fun provisionForThirdPartyPeer(): Result<String> {
+    private suspend fun provisionForThirdPartyPeer(): Result<String> {
         when (val ddg = recoveryCodeProvider.createDdgAccountIfNeeded()) {
             is Result.Success -> Unit
             is Result.Error -> {

sync/sync-impl/src/main/java/com/duckduckgo/sync/impl/exchange/v2/RecoveryCodeProvider.kt

@@ -56,7 +56,7 @@ interface RecoveryCodeProvider {
      * 3party, if needed, extend the account."* No-op if a 3party credential already exists locally
      * ([SyncStore.scopedPassword] set); otherwise creates one. Failure surfaces as [Result.Error].
      */
-    fun createThirdPartyCredentialIfNeeded(): Result<Unit>
+    suspend fun createThirdPartyCredentialIfNeeded(): Result<Unit>
 }
 
 @ContributesBinding(AppScope::class)
@@ -73,7 +73,7 @@ class RealRecoveryCodeProvider @Inject constructor(
         }
     }
 
-    override fun createThirdPartyCredentialIfNeeded(): Result<Unit> {
+    override suspend fun createThirdPartyCredentialIfNeeded(): Result<Unit> {
         if (syncStore.scopedPassword != null) return Result.Success(Unit)
         return when (val r = syncAccountRepository.createThirdPartyCredential()) {
             is Result.Success -> Result.Success(Unit)

sync/sync-impl/src/test/java/com/duckduckgo/sync/impl/AppSyncAccountRepositoryTest.kt

@@ -84,6 +84,7 @@ import com.duckduckgo.sync.store.ScopedPassword
 import com.duckduckgo.sync.store.SyncStore
 import com.squareup.moshi.Moshi
 import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runTest
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
 import org.junit.Assert.assertNull
@@ -1247,7 +1248,7 @@ class AppSyncAccountRepositoryTest {
     // Per-manager logic is covered in ThirdPartyCredentialManagerTest and ProtectedKeyManagerTest.
 
     @Test
-    fun whenCreateThirdPartyCredentialThenDelegatesToManager() {
+    fun whenCreateThirdPartyCredentialThenDelegatesToManager() = runTest {
         whenever(thirdPartyCredentialManager.create()).thenReturn(Success(true))
 
         val result = syncRepo.createThirdPartyCredential()
@@ -1307,7 +1308,7 @@ class AppSyncAccountRepositoryTest {
     // joinAccountFromThirdPartyRecoveryCode
 
     @Test
-    fun whenJoinAccountFromThirdPartyAndAllStepsSucceedThenAtomicStoreUsesDdgTokenFromSecondLogin() {
+    fun whenJoinAccountFromThirdPartyAndAllStepsSucceedThenAtomicStoreUsesDdgTokenFromSecondLogin() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode()
 
         val result = syncRepo.joinAccountFromThirdPartyRecoveryCode(pastedCode)
@@ -1325,7 +1326,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenJoinAccountFromThirdPartyThenStep7aPostUpgradeLoginUsesUnrestrictedScope() {
+    fun whenJoinAccountFromThirdPartyThenStep7aPostUpgradeLoginUsesUnrestrictedScope() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode()
 
         syncRepo.joinAccountFromThirdPartyRecoveryCode(pastedCode)
@@ -1350,7 +1351,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenStep7aPostUpgradeLoginReturns401ThenAbortsWithoutAtomicStore() {
+    fun whenStep7aPostUpgradeLoginReturns401ThenAbortsWithoutAtomicStore() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode(step7aResult = step7aLoginUnauthorized)
 
         val result = syncRepo.joinAccountFromThirdPartyRecoveryCode(pastedCode)
@@ -1361,7 +1362,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenStep7aPostUpgradeLoginFailsWithNon401ThenAbortsWithoutAtomicStore() {
+    fun whenStep7aPostUpgradeLoginFailsWithNon401ThenAbortsWithoutAtomicStore() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode(step7aResult = step7aLoginServerError)
 
         val result = syncRepo.joinAccountFromThirdPartyRecoveryCode(pastedCode)
@@ -1375,7 +1376,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenJoinAccountFromThirdPartyAndAccountAlreadyHasDdgCredentialThenAbortsBeforePost() {
+    fun whenJoinAccountFromThirdPartyAndAccountAlreadyHasDdgCredentialThenAbortsBeforePost() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode(accountAlreadyHasDdg = true)
 
         val result = syncRepo.joinAccountFromThirdPartyRecoveryCode(pastedCode)
@@ -1388,7 +1389,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenStep7PostRateLimitedThenRetriesAndSucceeds() {
+    fun whenStep7PostRateLimitedThenRetriesAndSucceeds() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode()
         whenever(syncApi.createAccessCredential(anyString(), eq("ddg"), any()))
             .thenReturn(Error(code = API_CODE.TOO_MANY_REQUESTS_2.code, reason = "rate limited"), Success(true))
@@ -1400,7 +1401,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenStep7aLoginServerErrorThenRetriesAndSucceeds() {
+    fun whenStep7aLoginServerErrorThenRetriesAndSucceeds() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode()
         whenever(syncApi.login(eq(userId), any(), eq(deviceId), any(), any(), eq<String?>(null)))
             .thenReturn(step7aLoginServerError, step7aLoginSuccess)
@@ -1420,7 +1421,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenStep7aLoginTransportErrorThenRetriesAndSucceeds() {
+    fun whenStep7aLoginTransportErrorThenRetriesAndSucceeds() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode()
         whenever(syncApi.login(eq(userId), any(), eq(deviceId), any(), any(), eq<String?>(null)))
             .thenReturn(step7aLoginTransportError, step7aLoginSuccess)
@@ -1432,7 +1433,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenStep7aLogin401ThenAbortsWithoutRetry() {
+    fun whenStep7aLogin401ThenAbortsWithoutRetry() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode(step7aResult = step7aLoginUnauthorized)
 
         val result = syncRepo.joinAccountFromThirdPartyRecoveryCode(pastedCode)
@@ -1443,7 +1444,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenStep7Post409AlreadyExistsThenAbortsWithoutRetry() {
+    fun whenStep7Post409AlreadyExistsThenAbortsWithoutRetry() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode()
         whenever(syncApi.createAccessCredential(anyString(), eq("ddg"), any()))
             .thenReturn(Error(code = API_CODE.COUNT_LIMIT.code, reason = "already exists"))
@@ -1456,7 +1457,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenStep7PostRateLimitedBeyondRetryLimitThenAbortsWithoutStore() {
+    fun whenStep7PostRateLimitedBeyondRetryLimitThenAbortsWithoutStore() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode()
         whenever(syncApi.createAccessCredential(anyString(), eq("ddg"), any()))
             .thenReturn(Error(code = API_CODE.TOO_MANY_REQUESTS_1.code, reason = "rate limited"))
@@ -1470,7 +1471,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenJoinAccountFromThirdPartyAndCodeIsLaterMinorVersionThenAccepted() {
+    fun whenJoinAccountFromThirdPartyAndCodeIsLaterMinorVersionThenAccepted() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode(version = "2.5")
 
         val result = syncRepo.joinAccountFromThirdPartyRecoveryCode(pastedCode)
@@ -1479,7 +1480,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenJoinAccountFromThirdPartyAndCodeIsBareMajorVersionThenAccepted() {
+    fun whenJoinAccountFromThirdPartyAndCodeIsBareMajorVersionThenAccepted() = runTest {
         // "2" is the spec's common shorthand for "2.0" (Transport TD 1214486492252757).
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode(version = "2")
 
@@ -1489,7 +1490,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenJoinAccountFromThirdPartyAndCodeIsMajorVersion3ThenRejectedWithoutNetwork() {
+    fun whenJoinAccountFromThirdPartyAndCodeIsMajorVersion3ThenRejectedWithoutNetwork() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode(version = "3.0")
 
         val result = syncRepo.joinAccountFromThirdPartyRecoveryCode(pastedCode)
@@ -1499,7 +1500,7 @@ class AppSyncAccountRepositoryTest {
     }
 
     @Test
-    fun whenJoinAccountFromThirdPartyAndCodeCidIsNotThirdPartyThenRejectedWithoutNetwork() {
+    fun whenJoinAccountFromThirdPartyAndCodeCidIsNotThirdPartyThenRejectedWithoutNetwork() = runTest {
         val pastedCode = prepareForJoinAccountFromThirdPartyRecoveryCode(cid = "ddg")
 
         val result = syncRepo.joinAccountFromThirdPartyRecoveryCode(pastedCode)

sync/sync-impl/src/test/java/com/duckduckgo/sync/impl/RealSyncCodeDispatcherTest.kt

@@ -448,7 +448,7 @@ class RealSyncCodeDispatcherTest {
         assertTrue((outcome as DispatchOutcome.Failed).reason.contains("future-credential"))
     }
 
-    @Test fun `Legacy path - dispatcher never invokes processCode (caller owns that)`() {
+    @Test fun `Legacy path - dispatcher never invokes processCode (caller owns that)`() = runTest {
         setV2(false)
         whenever(syncAccountRepository.parseSyncAuthCode(any())).thenReturn(SyncAuthCode.Recovery(mock()))