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,
@@ -127,7 +128,7 @@ interface SyncAccountRepository {
      * populated, credentialId=ddg, scopedPassword populated with SP. SyncStore is written only
      * after all three 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
@@ -625,7 +626,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()
 
@@ -758,7 +759,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")
         }
@@ -1338,15 +1339,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
@@ -78,7 +79,7 @@ class RealThirdPartyCredentialManager @Inject constructor(
     @VisibleForTesting
     internal var rateLimitRetryDelayMillis: Long = 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
@@ -150,7 +151,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(
      *    the credential POST means the server writes them atomically, removing the need for a
      *    separate `POST /sync/keys/.../set-if-absent` call (and the race window that opened).
      */
-    private fun buildKeysForNewThirdPartyCredential(
+    private suspend fun buildKeysForNewThirdPartyCredential(
         token: String,
         newSpBase64: String,
         hkdfSalt: ByteArray,
@@ -317,7 +318,7 @@ class RealThirdPartyCredentialManager @Inject constructor(
         )
     }
 
-    private fun postCreateAccessCredential(
+    private suspend fun postCreateAccessCredential(
         inputs: CreateInputs,
         request: CreateAccessCredentialRequest,
         newSpBase64: String,
@@ -359,7 +360,7 @@ class RealThirdPartyCredentialManager @Inject constructor(
      * Other 409 variants (e.g. `credential_already_exists`) are NOT retried — they're returned
      * to the caller, which handles them via the adopt path.
      */
-    private fun postWithConcurrentModificationRetry(
+    private suspend fun postWithConcurrentModificationRetry(
         token: String,
         request: CreateAccessCredentialRequest,
     ): Result<Boolean> {
@@ -372,7 +373,7 @@ class RealThirdPartyCredentialManager @Inject constructor(
             if (isConcurrentMod && attempt < MAX_CONCURRENT_MODIFICATION_RETRIES) {
                 attempt++
                 logcat { "Sync-ScopedToken: 3party credential POST hit concurrent_modification; retry $attempt/$MAX_CONCURRENT_MODIFICATION_RETRIES" }
-                runCatching { Thread.sleep(CONCURRENT_MODIFICATION_RETRY_DELAY_MILLIS * attempt) }
+                delay(CONCURRENT_MODIFICATION_RETRY_DELAY_MILLIS * attempt)
                 continue
             }
             return result
@@ -476,7 +477,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()
@@ -485,7 +486,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)