Prevent storage service loops on ACI identity key changes.

app/src/androidTest/java/org/thoughtcrime/securesms/storage/ContactRecordProcessorTest.kt

@@ -52,7 +52,7 @@ class ContactRecordProcessorTest {
     )
 
     // WHEN
-    val subject = ContactRecordProcessor()
+    val subject = ContactRecordProcessor(mutableSetOf())
     subject.process(listOf(remote1, remote2), StorageSyncHelper.KEY_GENERATOR)
 
     // THEN
@@ -81,7 +81,7 @@ class ContactRecordProcessorTest {
     )
 
     // WHEN
-    val subject = ContactRecordProcessor()
+    val subject = ContactRecordProcessor(mutableSetOf())
     subject.process(listOf(remote), StorageSyncHelper.KEY_GENERATOR)
 
     // THEN
@@ -119,7 +119,7 @@ class ContactRecordProcessorTest {
     )
 
     // WHEN
-    val subject = ContactRecordProcessor()
+    val subject = ContactRecordProcessor(mutableSetOf())
     subject.process(listOf(remote1, remote2), StorageSyncHelper.KEY_GENERATOR)
 
     // THEN

app/src/main/java/org/thoughtcrime/securesms/jobs/RetrieveProfileJob.kt

@@ -58,8 +58,8 @@ import kotlin.time.Duration.Companion.minutes
 /**
  * Retrieves a users profile and sets the appropriate local fields.
  */
-class RetrieveProfileJob private constructor(parameters: Parameters, private val recipientIds: MutableSet<RecipientId>, private val skipDebounce: Boolean) : BaseJob(parameters) {
-  private constructor(recipientIds: Set<RecipientId>, skipDebounce: Boolean) : this(
+class RetrieveProfileJob private constructor(parameters: Parameters, private val recipientIds: MutableSet<RecipientId>, private val skipDebounce: Boolean, private val syncConfirmedIdentityKey: Boolean) : BaseJob(parameters) {
+  private constructor(recipientIds: Set<RecipientId>, skipDebounce: Boolean, syncConfirmedIdentityKey: Boolean = false) : this(
     parameters = Parameters.Builder()
       .addConstraint(NetworkConstraint.KEY)
       .addConstraint(DataRestoreConstraint.KEY)
@@ -72,13 +72,15 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val
       .setMaxAttempts(3)
       .build(),
     recipientIds = recipientIds.toMutableSet(),
-    skipDebounce = skipDebounce
+    skipDebounce = skipDebounce,
+    syncConfirmedIdentityKey = syncConfirmedIdentityKey
   )
 
   override fun serialize(): ByteArray? {
     return JsonJobData.Builder()
       .putStringListAsArray(KEY_RECIPIENTS, recipientIds.map { it.serialize() })
       .putBoolean(KEY_SKIP_DEBOUNCE, skipDebounce)
+      .putBoolean(KEY_SYNC_CONFIRMED_IDENTITY_KEY, syncConfirmedIdentityKey)
       .serialize()
   }
 
@@ -361,6 +363,11 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val
       }
 
       if (existingIdentityKey == identityKey) {
+        if (syncConfirmedIdentityKey) {
+          Log.i(TAG, "Server confirmed our identity key for ${recipient.id}. Syncing it so the conflicting peer record is replaced.")
+          SignalDatabase.recipients.markNeedsSync(recipient.id)
+          StorageSyncHelper.scheduleSyncForDataChange()
+        }
         return
       }
 
@@ -538,8 +545,9 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val
       val data = JsonJobData.deserialize(serializedData)
       val recipientIds: MutableSet<RecipientId> = data.getStringArray(KEY_RECIPIENTS).map { RecipientId.from(it) }.toMutableSet()
       val skipDebounce: Boolean = data.getBooleanOrDefault(KEY_SKIP_DEBOUNCE, false)
+      val syncConfirmedIdentityKey: Boolean = data.getBooleanOrDefault(KEY_SYNC_CONFIRMED_IDENTITY_KEY, false)
 
-      return RetrieveProfileJob(parameters, recipientIds, skipDebounce)
+      return RetrieveProfileJob(parameters, recipientIds, skipDebounce, syncConfirmedIdentityKey)
     }
   }
 
@@ -548,6 +556,7 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val
     private val TAG = Log.tag(RetrieveProfileJob::class.java)
     private const val KEY_RECIPIENTS = "recipients"
     private const val KEY_SKIP_DEBOUNCE = "skip_debounce"
+    private const val KEY_SYNC_CONFIRMED_IDENTITY_KEY = "sync_confirmed_identity_key"
     private const val QUEUE_PREFIX = "RetrieveProfileJob_"
 
     private val PROFILE_FETCH_DEBOUNCE_TIME = 5.minutes
@@ -610,6 +619,15 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val
       }
     }
 
+    /**
+     * Only to be used when confirming an identity key change indicated by storage service. The recipient must be an individual and is used as-is. If
+     * the identity key matches we will write to storage service.
+     */
+    @WorkerThread
+    fun enqueueToResolveIdentityKeyConflict(recipientId: RecipientId) {
+      AppDependencies.jobManager.add(RetrieveProfileJob(setOf(recipientId), skipDebounce = true, syncConfirmedIdentityKey = true))
+    }
+
     /**
      * Will fetch some profiles to ensure we're decently up-to-date if we haven't done so within a
      * certain time period.

app/src/main/java/org/thoughtcrime/securesms/jobs/StorageSyncJob.kt

@@ -280,6 +280,7 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc
     var self = freshSelf()
     var needsMultiDeviceSync = false
     var needsForcePush = false
+    val identityConflictsPendingRepair = mutableSetOf<StorageId>()
 
     if (self.storageId == null) {
       Log.w(TAG, "No storageId for self. Generating.")
@@ -342,7 +343,7 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc
         try {
           Log.i(TAG, "[Remote Sync] Remote-Only :: Contacts: ${remoteOnly.contacts.size}, GV2: ${remoteOnly.gv2.size}, Account: ${remoteOnly.account.size}, DLists: ${remoteOnly.storyDistributionLists.size}, call links: ${remoteOnly.callLinkRecords.size}, chat folders: ${remoteOnly.chatFolderRecords.size}, notification profiles: ${remoteOnly.notificationProfileRecords.size}, sticker packs: ${remoteOnly.stickerPackRecords.size}")
 
-          processKnownRecords(context, remoteOnly)
+          processKnownRecords(context, remoteOnly, identityConflictsPendingRepair)
 
           val unknownInserts: List<SignalStorageRecord> = remoteOnly.unknown
           val unknownDeletes = idDifference.localOnlyIds.stream().filter { obj: StorageId -> obj.isUnknown }.collect(Collectors.toList())
@@ -395,7 +396,7 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc
       Log.i(TAG, "Found ${remote.size} of the known-unknowns remotely.")
 
       db.withinTransaction {
-        processKnownRecords(context, records)
+        processKnownRecords(context, records, identityConflictsPendingRepair)
         SignalDatabase.unknownStorageIds.deleteAllWithTypes(knownTypes)
       }
     }
@@ -455,7 +456,9 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc
     }
     stopwatch.split("local-data-transaction")
 
-    val loopCheck = if (remoteWriteOperation.isEmpty) {
+    val onlyIdentityConflictsPendingRepair = remoteWriteOperation.inserts.isNotEmpty() && remoteWriteOperation.inserts.all { it.id in identityConflictsPendingRepair }
+
+    val loopCheck = if (remoteWriteOperation.isEmpty || onlyIdentityConflictsPendingRepair) {
       StorageSyncLoopDetector.Decision.Allowed
     } else {
       StorageSyncLoopDetector.onWriteAttempt(remoteWriteOperation, fetchRemoteManifest, isRetry = runAttempt > 0)
@@ -464,6 +467,8 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc
     if (remoteWriteOperation.isEmpty) {
       Log.i(TAG, "No remote writes needed. Still at version: " + remoteManifest.versionString)
       StorageSyncLoopDetector.onConverged()
+    } else if (onlyIdentityConflictsPendingRepair) {
+      Log.w(TAG, "Deferring remote write until the profile fetch says whose identity key is correct. WriteOperationResult :: $remoteWriteOperation")
     } else if (loopCheck is StorageSyncLoopDetector.Decision.Denied) {
       Log.w(TAG, "Skipping remote write, another device is likely undoing it. Cause: ${loopCheck.cause}, level: ${loopCheck.level}. WriteOperationResult :: $remoteWriteOperation")
 
@@ -519,8 +524,8 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc
   }
 
   @Throws(IOException::class)
-  private fun processKnownRecords(context: Context, records: StorageRecordCollection) {
-    ContactRecordProcessor().process(records.contacts, StorageSyncHelper.KEY_GENERATOR)
+  private fun processKnownRecords(context: Context, records: StorageRecordCollection, identityConflictsPendingRepair: MutableSet<StorageId>) {
+    ContactRecordProcessor(identityConflictsPendingRepair).process(records.contacts, StorageSyncHelper.KEY_GENERATOR)
     GroupV2RecordProcessor().process(records.gv2, StorageSyncHelper.KEY_GENERATOR)
     NotificationProfileRecordProcessor().process(records.notificationProfileRecords, StorageSyncHelper.KEY_GENERATOR)
     AccountRecordProcessor(context, freshSelf()).process(records.account, StorageSyncHelper.KEY_GENERATOR)

app/src/main/java/org/thoughtcrime/securesms/storage/ContactRecordProcessor.kt

@@ -31,12 +31,15 @@ import java.util.regex.Pattern
 /**
  * Record processor for [SignalContactRecord].
  * Handles merging and updating our local store when processing remote contact storage records.
+ *
+ * @param identityConflictsPendingRepair Populated with storage ids where the only difference is identity key
  */
 class ContactRecordProcessor(
   private val selfAci: ACI?,
   private val selfPni: PNI?,
   private val selfE164: String?,
-  private val recipientTable: RecipientTable
+  private val recipientTable: RecipientTable,
+  val identityConflictsPendingRepair: MutableSet<StorageId>
 ) : DefaultStorageRecordProcessor<SignalContactRecord>() {
 
   companion object {
@@ -51,11 +54,12 @@ class ContactRecordProcessor(
 
   private var rotateProfileKeyOnBlock = true
 
-  constructor() : this(
+  constructor(identityConflictsPendingRepair: MutableSet<StorageId>) : this(
     selfAci = SignalStore.account.aci,
     selfPni = SignalStore.account.pni,
     selfE164 = SignalStore.account.e164,
-    recipientTable = SignalDatabase.recipients
+    recipientTable = SignalDatabase.recipients,
+    identityConflictsPendingRepair = identityConflictsPendingRepair
   )
 
   /**
@@ -232,7 +236,7 @@ class ContactRecordProcessor(
       if (conflictAci != null) {
         Log.w(TAG, "Identity keys conflict for $conflictAci. Enqueueing a profile fetch.")
         SignalDatabase.runPostSuccessfulTransaction {
-          RetrieveProfileJob.enqueue(Recipient.trustedPush(conflictAci, mergedPni, mergedE164).id, true)
+          RetrieveProfileJob.enqueueToResolveIdentityKeyConflict(Recipient.trustedPush(conflictAci, mergedPni, mergedE164).id)
         }
       } else {
         Log.w(TAG, "Identity keys conflict for $localPni. No ACI, so no profile fetch is possible.")
@@ -279,6 +283,9 @@ class ContactRecordProcessor(
     return if (matchesRemote) {
       remote
     } else if (matchesLocal) {
+      if (identityKeysExistsAndConflict && conflictAci != null) {
+        identityConflictsPendingRepair += local.id
+      }
       local
     } else {
       merged

app/src/test/java/org/thoughtcrime/securesms/storage/ContactRecordProcessorTest.kt

@@ -55,7 +55,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, normal, false`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -75,7 +75,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, missing ACI and PNI, true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -93,7 +93,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, unknown ACI and PNI, true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -113,7 +113,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, e164 matches self, true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -132,7 +132,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, aci matches self, true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -150,7 +150,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, pni matches self as pni, true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -169,7 +169,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, valid E164, true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -188,7 +188,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, invalid E164 (missing +), true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -207,7 +207,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, invalid E164 (contains letters), true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -226,7 +226,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, invalid E164 (no numbers), true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -245,7 +245,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, invalid E164 (too many numbers), true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -264,7 +264,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `isInvalid, invalid E164 (starts with zero), true`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val record = buildRecord(
       record = ContactRecord(
@@ -283,7 +283,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, e164MatchesButPnisDont pnpEnabled, keepLocal`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -315,7 +315,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, pnisMatchButE164sDont pnpEnabled, keepLocal`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -347,7 +347,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, e164AndPniChange pnpEnabled, useRemote`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -379,7 +379,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, nickname change, useRemote`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -411,7 +411,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, identityKeys conflict on primary, keepLocal`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -442,7 +442,7 @@ class ContactRecordProcessorTest {
   fun `merge, identityKeys conflict on linked device, useRemote`() {
     // GIVEN
     every { SignalStore.account.isPrimaryDevice } returns false
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -477,10 +477,10 @@ class ContactRecordProcessorTest {
     mockkObject(RetrieveProfileJob.Companion)
     mockkObject(SignalDatabase.Companion)
     every { Recipient.trustedPush(any(), any(), any()) } returns mockk(relaxed = true)
-    every { RetrieveProfileJob.enqueue(any<RecipientId>(), any()) } returns Unit
+    every { RetrieveProfileJob.enqueueToResolveIdentityKeyConflict(any<RecipientId>()) } returns Unit
     every { SignalDatabase.runPostSuccessfulTransaction(any<Runnable>()) } answers { firstArg<Runnable>().run() }
 
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -505,14 +505,153 @@ class ContactRecordProcessorTest {
 
     // THEN the profile fetch can repair this, so we keep our own key rather than deferring
     assertEquals(IDENTITY_KEY_A, result.proto.identityKey)
-    verify { RetrieveProfileJob.enqueue(any<RecipientId>(), any()) }
+    verify { RetrieveProfileJob.enqueueToResolveIdentityKeyConflict(any<RecipientId>()) }
+    assertEquals(setOf(STORAGE_ID_A), subject.identityConflictsPendingRepair)
+  }
+
+  @Test
+  fun `merge, identityKeys conflict with ACI, defersPushingOurRecord`() {
+    // GIVEN
+    mockkObject(Recipient.Companion)
+    mockkObject(RetrieveProfileJob.Companion)
+    mockkObject(SignalDatabase.Companion)
+    every { Recipient.trustedPush(any(), any(), any()) } returns mockk(relaxed = true)
+    every { RetrieveProfileJob.enqueueToResolveIdentityKeyConflict(any<RecipientId>()) } returns Unit
+    every { SignalDatabase.runPostSuccessfulTransaction(any<Runnable>()) } answers { firstArg<Runnable>().run() }
+
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
+
+    val local = buildRecord(
+      STORAGE_ID_A,
+      record = ContactRecord(
+        aciBinary = ACI_B.toByteString(),
+        e164 = E164_B,
+        identityKey = IDENTITY_KEY_A
+      )
+    )
+
+    val remote = buildRecord(
+      STORAGE_ID_B,
+      record = ContactRecord(
+        aciBinary = ACI_B.toByteString(),
+        e164 = E164_B,
+        identityKey = IDENTITY_KEY_B
+      )
+    )
+
+    // WHEN
+    val result = subject.merge(remote, local, TestKeyGenerator(STORAGE_ID_C))
+
+    // THEN we keep our key, but flag our id so the caller holds the write until the fetch resolves
+    assertEquals(IDENTITY_KEY_A, result.proto.identityKey)
+    assertEquals(STORAGE_ID_A, result.id)
+    assertEquals(setOf(STORAGE_ID_A), subject.identityConflictsPendingRepair)
+  }
+
+  @Test
+  fun `merge, identityKeys conflict with ACI, addsToCallersExistingSet`() {
+    // GIVEN
+    mockkObject(Recipient.Companion)
+    mockkObject(RetrieveProfileJob.Companion)
+    mockkObject(SignalDatabase.Companion)
+    every { Recipient.trustedPush(any(), any(), any()) } returns mockk(relaxed = true)
+    every { RetrieveProfileJob.enqueueToResolveIdentityKeyConflict(any<RecipientId>()) } returns Unit
+    every { SignalDatabase.runPostSuccessfulTransaction(any<Runnable>()) } answers { firstArg<Runnable>().run() }
+
+    val callerSet = mutableSetOf(STORAGE_ID_C)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, callerSet)
+
+    val local = buildRecord(
+      STORAGE_ID_A,
+      record = ContactRecord(
+        aciBinary = ACI_B.toByteString(),
+        e164 = E164_B,
+        identityKey = IDENTITY_KEY_A
+      )
+    )
+
+    val remote = buildRecord(
+      STORAGE_ID_B,
+      record = ContactRecord(
+        aciBinary = ACI_B.toByteString(),
+        e164 = E164_B,
+        identityKey = IDENTITY_KEY_B
+      )
+    )
+
+    // WHEN
+    subject.merge(remote, local, TestKeyGenerator(STORAGE_ID_C))
+
+    // THEN the caller accumulates across processors, so a pre-existing entry has to survive
+    assertEquals(setOf(STORAGE_ID_C, STORAGE_ID_A), callerSet)
+  }
+
+  @Test
+  fun `merge, identityKeys conflict without ACI, doesNotDefer`() {
+    // GIVEN
+    val callerSet = mutableSetOf(STORAGE_ID_C)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, callerSet)
+
+    val local = buildRecord(
+      STORAGE_ID_A,
+      record = ContactRecord(
+        pniBinary = PNI_B.toByteStringWithoutPrefix(),
+        e164 = E164_B,
+        identityKey = IDENTITY_KEY_A
+      )
+    )
+
+    val remote = buildRecord(
+      STORAGE_ID_B,
+      record = ContactRecord(
+        pniBinary = PNI_B.toByteStringWithoutPrefix(),
+        e164 = E164_B,
+        identityKey = IDENTITY_KEY_B
+      )
+    )
+
+    // WHEN
+    subject.merge(remote, local, TestKeyGenerator(STORAGE_ID_C))
+
+    // THEN no profile fetch is possible, so deferring would stall forever
+    assertEquals(setOf(STORAGE_ID_C), callerSet)
+  }
+
+  @Test
+  fun `merge, identityKeys match, doesNotDefer`() {
+    // GIVEN
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
+
+    val local = buildRecord(
+      STORAGE_ID_A,
+      record = ContactRecord(
+        aciBinary = ACI_B.toByteString(),
+        e164 = E164_B,
+        identityKey = IDENTITY_KEY_A
+      )
+    )
+
+    val remote = buildRecord(
+      STORAGE_ID_B,
+      record = ContactRecord(
+        aciBinary = ACI_B.toByteString(),
+        e164 = E164_B,
+        identityKey = IDENTITY_KEY_A
+      )
+    )
+
+    // WHEN
+    subject.merge(remote, local, TestKeyGenerator(STORAGE_ID_C))
+
+    // THEN
+    assertTrue(subject.identityConflictsPendingRepair.isEmpty())
   }
 
   @Test
   fun `merge, identityKeys match on linked device, keepLocal`() {
     // GIVEN
     every { SignalStore.account.isPrimaryDevice } returns false
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -542,7 +681,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, local identityKey missing on primary, useRemote`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -572,7 +711,7 @@ class ContactRecordProcessorTest {
   fun `merge, remote identityKey missing on linked device, keepLocal`() {
     // GIVEN
     every { SignalStore.account.isPrimaryDevice } returns false
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -601,7 +740,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, pniSignatureVerified but no PNI, clearsFlag`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -630,7 +769,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, pniSignatureVerified with PNI, keepsFlag`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -661,7 +800,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, notifyForCallsIfMuted set remotely and locally, useRemote`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -691,7 +830,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, notifyForCallsIfMuted unset remotely, keepLocal`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -721,7 +860,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, showUnreadReminders set remotely and locally, useRemote`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,
@@ -751,7 +890,7 @@ class ContactRecordProcessorTest {
   @Test
   fun `merge, showUnreadReminders unset remotely, keepLocal`() {
     // GIVEN
-    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable)
+    val subject = ContactRecordProcessor(ACI_A, PNI_A, E164_A, recipientTable, mutableSetOf())
 
     val local = buildRecord(
       STORAGE_ID_A,