Cleanup bad notified state in background instead of during db migration.

2026-08-06 13:31:18 +01:00 · 2026-06-04 13:55:38 -04:00
parent 045bd9287b
commit a6311c87c1
5 changed files with 144 additions and 7 deletions
@@ -6,6 +6,9 @@ import org.thoughtcrime.securesms.database.SQLiteDatabase
 /**
 * Fix bad notified state across the message table so that we can use an index to improve query performance
 * when fetching notification state.
+ *
+ * Note: this intentionally does *not* clean up "dead" rows (read messages where notified is still 0) that bloat
+ * the partial index. That cleanup will happen over time as an app migration to prevent long migration startups.
 */
@Suppress("ClassName")
 object V318_AddMessageNotificationStateIndex : SignalDatabaseMigration {
@@ -15,8 +18,6 @@ object V318_AddMessageNotificationStateIndex : SignalDatabaseMigration {
    db.execSQL("UPDATE message SET reactions_unread = 0 WHERE reactions_unread = 1 AND (type & 31) NOT IN $outgoingBaseTypes")
    db.execSQL("UPDATE message SET votes_unread = 0 WHERE votes_unread = 1 AND (type & 31) NOT IN $outgoingBaseTypes")

-    db.execSQL("UPDATE message SET notified = 1 WHERE notified = 0 AND read = 1 AND reactions_unread = 0 AND votes_unread = 0")
-
    db.execSQL("CREATE INDEX IF NOT EXISTS message_notification_state_index ON message (date_received) WHERE notified = 0 AND story_type = 0 AND latest_revision_id IS NULL")
  }
 }
@@ -0,0 +1,96 @@
+package org.thoughtcrime.securesms.jobs
+
+import org.signal.core.util.fullWalCheckpoint
+import org.signal.core.util.logging.Log
+import org.signal.core.util.update
+import org.thoughtcrime.securesms.database.MessageTable
+import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.dependencies.AppDependencies
+import org.thoughtcrime.securesms.jobmanager.Job
+import kotlin.time.Duration.Companion.seconds
+
+/**
+ * Incrementally cleans up "dead" notification state in the message table by marking read, non-notified messages as
+ * notified. These rows would otherwise sit in the message_notification_state_index forever.
+ */
+class BackfillNotifiedStateJob private constructor(parameters: Parameters) : Job(parameters) {
+
+  companion object {
+    const val KEY = "BackfillNotifiedStateJob"
+
+    private val TAG = Log.tag(BackfillNotifiedStateJob::class.java)
+
+    private const val BATCH_SIZE = 1000
+    private val TIME_BUDGET = 3.seconds
+    private val RETRY_BACKOFF = 30.seconds
+
+    @JvmStatic
+    fun enqueue() {
+      AppDependencies.jobManager.add(BackfillNotifiedStateJob())
+    }
+  }
+
+  constructor() : this(
+    Parameters.Builder()
+      .setQueue(KEY)
+      .setMaxInstancesForFactory(1)
+      .setMaxAttempts(Parameters.UNLIMITED)
+      .setInitialDelay(30.seconds.inWholeMilliseconds)
+      .build()
+  )
+
+  override fun serialize(): ByteArray? = null
+  override fun getFactoryKey(): String = KEY
+  override fun onFailure() = Unit
+
+  override fun run(): Result {
+    val endTime = System.currentTimeMillis() + TIME_BUDGET.inWholeMilliseconds
+    var totalUpdated = 0
+    var lastBatchUpdateCount: Int
+
+    do {
+      lastBatchUpdateCount = updateBatch()
+      totalUpdated += lastBatchUpdateCount
+    } while (lastBatchUpdateCount > 0 && System.currentTimeMillis() < endTime)
+
+    Log.i(TAG, "Updated $totalUpdated rows this run.")
+
+    if (lastBatchUpdateCount > 0) {
+      return Result.retry(RETRY_BACKOFF.inWholeMilliseconds)
+    }
+
+    Log.i(TAG, "Backfill complete. Attempting to shrink WAL")
+    if (!SignalDatabase.writableDatabase.fullWalCheckpoint()) {
+      Log.w(TAG, "Failed to do a full WAL checkpoint after finished backfill.")
+    }
+
+    return Result.success()
+  }
+
+  /**
+   * Marks up to [BATCH_SIZE] read, non-notified messages as notified in a single transaction. Returns the number of
+   * rows updated, which is 0 once there is nothing left to clean up.
+   */
+  private fun updateBatch(): Int {
+    return SignalDatabase.writableDatabase
+      .update(MessageTable.TABLE_NAME)
+      .values(MessageTable.NOTIFIED to 1)
+      .where(
+        """
+        ${MessageTable.ID} IN (
+          SELECT ${MessageTable.ID}
+          FROM ${MessageTable.TABLE_NAME}
+          WHERE ${MessageTable.NOTIFIED} = 0 AND ${MessageTable.READ} = 1 AND ${MessageTable.REACTIONS_UNREAD} = 0 AND ${MessageTable.VOTES_UNREAD} = 0
+          LIMIT $BATCH_SIZE
+        )
+        """
+      )
+      .run()
+  }
+
+  class Factory : Job.Factory<BackfillNotifiedStateJob> {
+    override fun create(parameters: Parameters, serializedData: ByteArray?): BackfillNotifiedStateJob {
+      return BackfillNotifiedStateJob(parameters)
+    }
+  }
+}
@@ -55,6 +55,7 @@ import org.thoughtcrime.securesms.migrations.AvatarIdRemovalMigrationJob;
 import org.thoughtcrime.securesms.migrations.AvatarMigrationJob;
 import org.thoughtcrime.securesms.migrations.BackfillCollapsedEventsMigrationJob;
 import org.thoughtcrime.securesms.migrations.BackfillDigestsForDuplicatesMigrationJob;
+import org.thoughtcrime.securesms.migrations.BackfillNotifiedStateMigrationJob;
 import org.thoughtcrime.securesms.migrations.BackupJitterMigrationJob;
 import org.thoughtcrime.securesms.migrations.BackupNotificationMigrationJob;
 import org.thoughtcrime.securesms.migrations.BadE164MigrationJob;
@@ -144,8 +145,9 @@ public final class JobManagerFactories {
      put(AutomaticSessionResetJob.KEY,                new AutomaticSessionResetJob.Factory());
      put(AvatarGroupsV1DownloadJob.KEY,               new AvatarGroupsV1DownloadJob.Factory());
      put(AvatarGroupsV2DownloadJob.KEY,               new AvatarGroupsV2DownloadJob.Factory());
-      put(BackfillCollapsedMessageJob.KEY,              new BackfillCollapsedMessageJob.Factory());
+      put(BackfillCollapsedMessageJob.KEY,             new BackfillCollapsedMessageJob.Factory());
      put(BackfillDigestsForDataFileJob.KEY,           new BackfillDigestsForDataFileJob.Factory());
+      put(BackfillNotifiedStateJob.KEY,                new BackfillNotifiedStateJob.Factory());
      put(BackupDeleteJob.KEY,                         new BackupDeleteJob.Factory());
      put(BackupMessagesJob.KEY,                       new BackupMessagesJob.Factory());
      put(BackupRestoreMediaJob.KEY,                   new BackupRestoreMediaJob.Factory());
@@ -317,6 +319,7 @@ public final class JobManagerFactories {
      put(AvatarMigrationJob.KEY,                         new AvatarMigrationJob.Factory());
      put(BackfillCollapsedEventsMigrationJob.KEY,        new BackfillCollapsedEventsMigrationJob.Factory());
      put(BackfillDigestsForDuplicatesMigrationJob.KEY,   new BackfillDigestsForDuplicatesMigrationJob.Factory());
+      put(BackfillNotifiedStateMigrationJob.KEY,          new BackfillNotifiedStateMigrationJob.Factory());
      put(BackupJitterMigrationJob.KEY,                   new BackupJitterMigrationJob.Factory());
      put(BackupNotificationMigrationJob.KEY,             new BackupNotificationMigrationJob.Factory());
      put(BackupRefreshJob.KEY,                           new BackupRefreshJob.Factory());
@@ -203,10 +203,11 @@ public class ApplicationMigrations {
    static final int READ_INDEX_DB_MIGRATION       = 159;
    // Need to skip 160 due to release ordering issues
    static final int SVR2_ENCLAVE_UPDATE_6         = 161;
-    static final int NOTIFICATION_INDEX__MIGRATION = 162;
+    static final int NOTIFICATION_INDEX_MIGRATION  = 162;
+    static final int NOTIFICATION_STATE_CLEANUP    = 163;
  }

-  public static final int CURRENT_VERSION = 162;
+  public static final int CURRENT_VERSION = 163;

  /**
   * This *must* be called after the {@link JobManager} has been instantiated, but *before* the call
@@ -941,8 +942,12 @@ public class ApplicationMigrations {
      jobs.put(Version.SVR2_ENCLAVE_UPDATE_6, new Svr2MirrorMigrationJob());
    }

-    if (lastSeenVersion < Version.NOTIFICATION_INDEX__MIGRATION) {
-      jobs.put(Version.NOTIFICATION_INDEX__MIGRATION, new DatabaseMigrationJob());
+    if (lastSeenVersion < Version.NOTIFICATION_INDEX_MIGRATION) {
+      jobs.put(Version.NOTIFICATION_INDEX_MIGRATION, new DatabaseMigrationJob());
+    }
+
+    if (lastSeenVersion < Version.NOTIFICATION_STATE_CLEANUP) {
+      jobs.put(Version.NOTIFICATION_STATE_CLEANUP, new BackfillNotifiedStateMigrationJob());
    }

    return jobs;
@@ -0,0 +1,32 @@
+package org.thoughtcrime.securesms.migrations
+
+import org.thoughtcrime.securesms.jobmanager.Job
+import org.thoughtcrime.securesms.jobs.BackfillNotifiedStateJob
+
+/**
+ * Kicks off a background job to clean up dead notification state left behind by V318. See [BackfillNotifiedStateJob].
+ */
+internal class BackfillNotifiedStateMigrationJob(
+  parameters: Parameters = Parameters.Builder().build()
+) : MigrationJob(parameters) {
+
+  companion object {
+    const val KEY = "BackfillNotifiedStateMigrationJob"
+  }
+
+  override fun getFactoryKey(): String = KEY
+
+  override fun isUiBlocking(): Boolean = false
+
+  override fun performMigration() {
+    BackfillNotifiedStateJob.enqueue()
+  }
+
+  override fun shouldRetry(e: Exception): Boolean = false
+
+  class Factory : Job.Factory<BackfillNotifiedStateMigrationJob> {
+    override fun create(parameters: Parameters, serializedData: ByteArray?): BackfillNotifiedStateMigrationJob {
+      return BackfillNotifiedStateMigrationJob(parameters)
+    }
+  }
+}