Improve profile fetch performance for large groups.

2026-08-06 13:31:18 +01:00 · 2026-05-19 15:13:35 -04:00
parent d682de08d2
commit 1661f3b5f7
10 changed files with 123 additions and 50 deletions
@@ -75,6 +75,10 @@ public class SignalIdentityKeyStore implements IdentityKeyStore {
    return baseStore.getIdentityRecord(recipientId);
  }

+  public @NonNull Optional<IdentityRecord> getIdentityRecord(@NonNull Recipient recipient) {
+    return baseStore.getIdentityRecord(recipient);
+  }
+
  public @NonNull IdentityRecordList getIdentityRecords(@NonNull List<Recipient> recipients) {
    return baseStore.getIdentityRecords(recipients);
  }
@@ -120,6 +120,7 @@ import java.util.LinkedList
 import java.util.Optional
 import kotlin.jvm.optionals.getOrNull
 import kotlin.math.max
+import kotlin.time.Duration

 open class RecipientTable(context: Context, databaseHelper: SignalDatabase) : DatabaseTable(context, databaseHelper) {

@@ -762,6 +763,27 @@ open class RecipientTable(context: Context, databaseHelper: SignalDatabase) : Da
    return (foundRecords + remappedRecords).associateBy { it.id }
  }

+  /**
+   * Returns recipient records eligible for a profile fetch.
+   * - Must have a service id (ACI or PNI)
+   * - Last profile fetch must be before [debounceThreshold] if non-null
+   */
+  fun getRecordsForProfileFetch(ids: Collection<RecipientId>, debounceThreshold: Duration?): List<RecipientRecord> {
+    if (ids.isEmpty()) {
+      return emptyList()
+    }
+
+    val prefix = "($ACI_COLUMN NOT NULL OR $PNI_COLUMN NOT NULL) ${if (debounceThreshold != null) " AND ($LAST_PROFILE_FETCH < ${debounceThreshold.inWholeMilliseconds}) AND " else ""}"
+    val idQuery = SqlUtil.buildFastCollectionQuery(ID, ids, prefix)
+
+    return readableDatabase
+      .select()
+      .from(TABLE_NAME)
+      .where(idQuery.where, idQuery.whereArgs)
+      .run()
+      .readToList { cursor -> RecipientTableCursorUtil.getRecord(context, cursor) }
+  }
+
  fun getRecord(id: RecipientId): RecipientRecord {
    val query = "$ID = ?"
    val args = arrayOf(id.serialize())
@@ -2489,13 +2511,34 @@ open class RecipientTable(context: Context, databaseHelper: SignalDatabase) : Da
  fun markUnregistered(id: RecipientId) {
    val record = getRecord(id)

-    if (record.aci != null && record.pni != null) {
+    val needsSplit = record.aci != null && record.pni != null
+    if (record.registered == RegisteredState.NOT_REGISTERED && !needsSplit) {
+      return
+    }
+
+    if (needsSplit) {
      markUnregisteredAndSplit(id, record)
    } else {
      markUnregisteredWithoutSplit(id)
    }
  }

+  fun markUnregistered(ids: Collection<RecipientId>) {
+    if (ids.isEmpty()) {
+      return
+    }
+
+    ids
+      .chunked(100)
+      .forEach { chunk ->
+        writableDatabase.withinTransaction {
+          for (id in chunk) {
+            markUnregistered(id)
+          }
+        }
+      }
+  }
+
  /**
   * Marks the user unregistered and also splits it into an ACI-only and PNI-only contact.
   * This is to allow a new user to register the number with a new ACI.
@@ -3773,13 +3816,19 @@ open class RecipientTable(context: Context, databaseHelper: SignalDatabase) : Da
      }
    }

-    return Recipient.resolvedList(recipientsWithinInteractionThreshold)
-      .asSequence()
-      .filterNot { it.isSelf }
-      .filter { it.lastProfileFetchTime < lastProfileFetchThreshold }
-      .take(limit)
-      .map { it.id }
-      .toMutableList()
+    if (Recipient.isSelfSet) {
+      recipientsWithinInteractionThreshold.remove(Recipient.self().id)
+    }
+
+    val select = SqlUtil.buildFastCollectionQuery(ID, recipientsWithinInteractionThreshold, "$LAST_PROFILE_FETCH < $lastProfileFetchThreshold AND")
+
+    return readableDatabase
+      .select(ID)
+      .from(TABLE_NAME)
+      .where(select.where, select.whereArgs)
+      .limit(limit)
+      .run()
+      .readToList { RecipientId.from(it.requireLong(ID)) }
  }

  fun markProfilesFetched(ids: Collection<RecipientId>, time: Long) {
@@ -3790,11 +3839,6 @@ open class RecipientTable(context: Context, databaseHelper: SignalDatabase) : Da
        db.update(TABLE_NAME, values, query.where, query.whereArgs)
      }
    }
-
-    // Invalidate recipient cache so that updated timestamps are reflected
-    ids.forEach { id ->
-      AppDependencies.databaseObserver.notifyRecipientChanged(id)
-    }
  }

  fun applyBlockedUpdate(blockedE164s: List<String>, blockedAcis: List<ACI>, blockedGroupIds: List<ByteArray?>) {
@@ -145,7 +145,6 @@ object RecipientTableCursorUtil {
      signalProfileAvatar = cursor.requireString(RecipientTable.PROFILE_AVATAR),
      profileAvatarFileDetails = AvatarHelper.getAvatarFileDetails(context, recipientId),
      profileSharing = cursor.requireBoolean(RecipientTable.PROFILE_SHARING),
-      lastProfileFetch = cursor.requireLong(RecipientTable.LAST_PROFILE_FETCH),
      notificationChannel = cursor.requireString(RecipientTable.NOTIFICATION_CHANNEL),
      sealedSenderAccessMode = RecipientTable.SealedSenderAccessMode.fromMode(cursor.requireInt(RecipientTable.SEALED_SENDER_MODE)),
      capabilities = readCapabilities(cursor),
@@ -59,7 +59,6 @@ data class RecipientRecord(
  val profileAvatarFileDetails: ProfileAvatarFileDetails,
  @get:JvmName("isProfileSharing")
  val profileSharing: Boolean,
-  val lastProfileFetch: Long,
  val notificationChannel: String?,
  val sealedSenderAccessMode: SealedSenderAccessMode,
  val capabilities: Capabilities,
@@ -22,6 +22,7 @@ import org.thoughtcrime.securesms.database.RecipientTable.Companion.maskCapabili
 import org.thoughtcrime.securesms.database.RecipientTable.PhoneNumberSharingState
 import org.thoughtcrime.securesms.database.RecipientTable.SealedSenderAccessMode
 import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.model.IdentityRecord
 import org.thoughtcrime.securesms.database.model.RecipientRecord
 import org.thoughtcrime.securesms.dependencies.AppDependencies
 import org.thoughtcrime.securesms.jobmanager.Job
@@ -32,6 +33,7 @@ import org.thoughtcrime.securesms.net.SignalNetwork
 import org.thoughtcrime.securesms.notifications.v2.ConversationId.Companion.forConversation
 import org.thoughtcrime.securesms.profiles.ProfileName
 import org.thoughtcrime.securesms.recipients.Recipient
+import org.thoughtcrime.securesms.recipients.RecipientCreator
 import org.thoughtcrime.securesms.recipients.RecipientId
 import org.thoughtcrime.securesms.recipients.RecipientUtil
 import org.thoughtcrime.securesms.storage.StorageSyncHelper
@@ -49,6 +51,7 @@ import org.whispersystems.signalservice.api.profiles.SignalServiceProfile
 import org.whispersystems.signalservice.api.util.ExpiringProfileCredentialUtil
 import java.io.IOException
 import java.util.concurrent.TimeUnit
+import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.minutes

 /**
@@ -90,32 +93,25 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val

    val stopwatch = Stopwatch("RetrieveProfile")

-    val recipients = recipientIds.map { Recipient.live(it).refresh().resolve() }
+    val debounceThreshold = if (skipDebounce) null else System.currentTimeMillis().milliseconds - PROFILE_FETCH_DEBOUNCE_TIME
+    val recipientsToFetch = SignalDatabase
+      .recipients
+      .getRecordsForProfileFetch(recipientIds, debounceThreshold)
+      .map { RecipientCreator.forRecord(context, it) }

-    RecipientUtil.ensureUuidsAreAvailable(
-      context,
-      recipients.filter { it.registered != RecipientTable.RegisteredState.NOT_REGISTERED }
-    )
-
-    stopwatch.split("resolve-ensure")
-
-    val currentTime = System.currentTimeMillis()
-    val debounceThreshold = currentTime - PROFILE_FETCH_DEBOUNCE_TIME_MS
-    val recipientsToFetch = recipients.filter { recipient ->
-      recipient.hasServiceId && (skipDebounce || recipient.lastProfileFetchTime < debounceThreshold)
-    }
+    stopwatch.split("resolve")

    if (recipientsToFetch.isEmpty()) {
-      Log.i(TAG, "All ${recipients.size} recipients have been fetched recently (within ${PROFILE_FETCH_DEBOUNCE_TIME_MS}ms). Skipping network requests.")
+      Log.i(TAG, "All ${recipientIds.size} recipients have been fetched recently (within $PROFILE_FETCH_DEBOUNCE_TIME) or are not eligible. Skipping network requests.")
      return
    }

-    if (recipientsToFetch.size < recipients.size) {
-      Log.i(TAG, "Debouncing: Fetching ${recipientsToFetch.size} of ${recipients.size} recipients (${recipients.size - recipientsToFetch.size} were fetched recently)")
+    if (recipientsToFetch.size < recipientIds.size) {
+      Log.i(TAG, "Fetching ${recipientsToFetch.size} of ${recipientIds.size} recipients (${recipientIds.size - recipientsToFetch.size} were ineligible or fetched recently)")
    }

    val fetchingRecipientIds = recipientsToFetch.map { it.id }.toSet()
-    val recipientsById: Map<RecipientId, Recipient> = recipients.associateBy { it.id }
+    val recipientsById: Map<RecipientId, Recipient> = recipientsToFetch.associateBy { it.id }

    val requests: List<ProfileFetchRequest<RecipientId>> = recipientsToFetch
      .map { recipient ->
@@ -176,12 +172,6 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val
        }
      }
    }
-    if (updatedProfiles.isNotEmpty()) {
-      StorageSyncHelper.scheduleSyncForDataChange()
-    }
-    if (avatarJobs.isNotEmpty()) {
-      AppDependencies.jobManager.addAll(avatarJobs)
-    }
    stopwatch.split("process")

    SignalDatabase.recipients.markProfilesFetched(successIds, System.currentTimeMillis())
@@ -193,9 +183,7 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val
    }
    if (response.unregistered.isNotEmpty()) {
      Log.i(TAG, "Marking ${response.unregistered.size} users as unregistered.")
-      for (recipientId in response.unregistered) {
-        SignalDatabase.recipients.markUnregistered(recipientId)
-      }
+      SignalDatabase.recipients.markUnregistered(response.unregistered)
    }
    stopwatch.split("registered-update")

@@ -214,9 +202,17 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val

    val keyCount = response.successes.mapNotNull { recipientsById[it.id] }.mapNotNull { it.profileKey }.count()

-    Log.d(TAG, "Started with ${recipients.size} recipient(s). Of those, ${recipientsToFetch.size} were outside the cache period. Found ${response.successes.size} profile(s), and had keys for $keyCount of them. Will retry ${response.retryableFailures.size}.")
+    Log.d(TAG, "Started with ${recipientIds.size} recipient(s). Of those, ${recipientsToFetch.size} were outside the cache period. Found ${response.successes.size} profile(s), and had keys for $keyCount of them. Will retry ${response.retryableFailures.size}.")
    stopwatch.stop(TAG)

+    if (avatarJobs.isNotEmpty()) {
+      AppDependencies.jobManager.addAll(avatarJobs)
+    }
+
+    if (updatedProfiles.isNotEmpty()) {
+      StorageSyncHelper.scheduleSyncForDataChange()
+    }
+
    recipientIds.clear()
    recipientIds.addAll(response.retryableFailures)

@@ -353,11 +349,19 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val
      }

      val identityKey = IdentityKey(decode(identityKeyValue), 0)
-      if (!AppDependencies.protocolStore.aci().identities().getIdentityRecord(recipient.id).isPresent) {
+      val existingIdentityKey = AppDependencies.protocolStore.aci().identities().getIdentityRecord(recipient)
+        .map { (_, identityKey): IdentityRecord -> identityKey }
+        .orElse(null)
+
+      if (existingIdentityKey == null) {
        Log.w(TAG, "Still first use for ${recipient.id}")
        return
      }

+      if (existingIdentityKey == identityKey) {
+        return
+      }
+
      IdentityUtil.saveIdentity(recipient.requireServiceId().toString(), identityKey)
    } catch (e: InvalidKeyException) {
      Log.w(TAG, e)
@@ -544,7 +548,7 @@ class RetrieveProfileJob private constructor(parameters: Parameters, private val
    private const val KEY_SKIP_DEBOUNCE = "skip_debounce"
    private const val QUEUE_PREFIX = "RetrieveProfileJob_"

-    private val PROFILE_FETCH_DEBOUNCE_TIME_MS = 5.minutes.inWholeMilliseconds
+    private val PROFILE_FETCH_DEBOUNCE_TIME = 5.minutes

    /**
     * Submits the necessary job to refresh the profile of the requested recipient. Works for any
@@ -105,7 +105,6 @@ class Recipient(
  val profileAvatarFileDetails: ProfileAvatarFileDetails = ProfileAvatarFileDetails.NO_DETAILS,
  val isProfileSharing: Boolean = false,
  val hiddenState: HiddenState = HiddenState.NOT_HIDDEN,
-  val lastProfileFetchTime: Long = 0,
  private val notificationChannelValue: String? = null,
  private val sealedSenderAccessModeValue: SealedSenderAccessMode = SealedSenderAccessMode.UNKNOWN,
  private val capabilities: RecipientRecord.Capabilities = RecipientRecord.Capabilities.UNKNOWN,
@@ -178,7 +178,6 @@ object RecipientCreator {
      profileAvatarFileDetails = record.profileAvatarFileDetails,
      isProfileSharing = record.profileSharing,
      hiddenState = record.hiddenState,
-      lastProfileFetchTime = record.lastProfileFetch,
      isSelf = isSelf,
      notificationChannelValue = record.notificationChannel,
      sealedSenderAccessModeValue = record.sealedSenderAccessMode,