Add storage sync hot loop detection and rate limiting.

2026-09-13 12:07:57 +01:00 · 2026-08-19 19:05:46 -04:00
parent e3fd0897e8
commit 9c01310df4
13 changed files with 1064 additions and 18 deletions
@@ -26,6 +26,9 @@ import org.thoughtcrime.securesms.notifications.NotificationIds
 import org.thoughtcrime.securesms.util.RemoteConfig
 import java.io.ByteArrayOutputStream
 import java.io.PrintStream
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.days
+import kotlin.time.Duration.Companion.minutes

 /**
 * Records noteworthy runtime issues to the [LogDatabase] issue table on a low-priority background thread.
@@ -41,6 +44,7 @@ object IssueReporter {
  const val ISSUE_SLOW_DATABASE_WRITE = "Slow Database Write"
  const val ISSUE_SLOW_DATABASE_READ = "Slow Database Read"
  const val ISSUE_SLOW_DATABASE_LOCK = "Slow Database Lock"
+  const val ISSUE_STORAGE_SYNC_LOOP = "Storage Sync Loop"

  const val SLOW_WRITE_LOW_PRIORITY_MS = 1_000L
  const val SLOW_WRITE_MEDIUM_PRIORITY_MS = 5_000L
@@ -51,6 +55,12 @@ object IssueReporter {

  private const val NON_INTERNAL_DEBOUNCE_MS = 5_000L

+  /** Applies per issue name, so a persistently-firing issue can't bury every other notification. */
+  private val DEFAULT_NOTIFY_COOLDOWN = 30.minutes
+
+  /** Ages entries out of the persisted map. Must exceed the longest cooldown a caller passes. */
+  private val MAX_NOTIFY_TIME_AGE = 7.days
+
  private val IGNORED_DB_STACK_TRACE_CLASSES = listOf(
    "BackupRepository",
    "BackupMessagesJob",
@@ -72,9 +82,7 @@ object IssueReporter {
  /**
   * Records a generic issue. Safe to call from any thread.
   */
-  @JvmStatic
-  @JvmOverloads
-  fun report(name: String, description: String, throwable: Throwable? = null, priority: IssuePriority = IssuePriority.LOW, duration: Long? = null) {
+  fun report(name: String, description: String, throwable: Throwable? = null, priority: IssuePriority = IssuePriority.LOW, duration: Long? = null, notifyCooldown: Duration = DEFAULT_NOTIFY_COOLDOWN) {
    val now = System.currentTimeMillis()

    if (!RemoteConfig.internalUser) {
@@ -86,7 +94,9 @@ object IssueReporter {

    requests.add(IssueRequest(now, BuildConfig.VERSION_NAME, name, description, throwable, priority, duration))

-    maybeNotify(name, priority)
+    if (RemoteConfig.internalUser) {
+      maybeNotify(name, priority, notifyCooldown)
+    }
  }

  @JvmStatic
@@ -138,12 +148,17 @@ object IssueReporter {
    report(ISSUE_SLOW_DATABASE_READ, query?.trim() ?: "", throwable, priority = priority, duration = durationMs)
  }

-  private fun maybeNotify(name: String, priority: IssuePriority) {
-    if (!RemoteConfig.internalUser) {
+  /** Synchronized because the cooldown is a read-modify-write of a persisted value, and this is safe to call anywhere. */
+  @Synchronized
+  private fun maybeNotify(name: String, priority: IssuePriority, cooldown: Duration) {
+    if (priority.value < SignalStore.internal.issueNotificationPriority.value) {
      return
    }

-    if (priority.value < SignalStore.internal.issueNotificationPriority.value) {
+    val now = System.currentTimeMillis()
+    val notifyTimes = SignalStore.internal.issueNotifyTimes
+
+    if (now - (notifyTimes.lastNotifyTimeByName[name] ?: 0) < cooldown.inWholeMilliseconds) {
      return
    }

@@ -160,6 +175,15 @@ object IssueReporter {
      .build()

    NotificationManagerCompat.from(context).notify(NotificationIds.INTERNAL_ERROR, notification)
+
+    val updatedTimes = notifyTimes
+      .lastNotifyTimeByName
+      .filterValues { now < it || now - it < MAX_NOTIFY_TIME_AGE.inWholeMilliseconds }
+      .plus(name to now)
+
+    SignalStore.internal.issueNotifyTimes = notifyTimes.copy(
+      lastNotifyTimeByName = updatedTimes
+    )
  }

  private fun isExpectedSlowDatabaseOperation(): Boolean {
@@ -11,10 +11,12 @@ import org.signal.libsignal.protocol.InvalidKeyException
 import org.signal.network.service.StorageServiceService
 import org.signal.network.service.StorageServiceService.ManifestIfDifferentVersionResult
 import org.thoughtcrime.securesms.database.ChatFolderTables.ChatFolderTable
+import org.thoughtcrime.securesms.database.IssueReporter
 import org.thoughtcrime.securesms.database.NotificationProfileTables
 import org.thoughtcrime.securesms.database.RecipientTable
 import org.thoughtcrime.securesms.database.SignalDatabase
 import org.thoughtcrime.securesms.database.StickerTables
+import org.thoughtcrime.securesms.database.model.IssuePriority
 import org.thoughtcrime.securesms.dependencies.AppDependencies
 import org.thoughtcrime.securesms.jobmanager.Job
 import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint
@@ -31,6 +33,7 @@ import org.thoughtcrime.securesms.storage.NotificationProfileRecordProcessor
 import org.thoughtcrime.securesms.storage.StickerPackRecordProcessor
 import org.thoughtcrime.securesms.storage.StorageSyncHelper
 import org.thoughtcrime.securesms.storage.StorageSyncHelper.WriteOperationResult
+import org.thoughtcrime.securesms.storage.StorageSyncLoopDetector
 import org.thoughtcrime.securesms.storage.StorageSyncModels
 import org.thoughtcrime.securesms.storage.StorageSyncValidations
 import org.thoughtcrime.securesms.storage.StoryDistributionListRecordProcessor
@@ -64,6 +67,7 @@ import org.whispersystems.signalservice.internal.storage.protos.ManifestRecord
 import java.io.IOException
 import java.util.concurrent.TimeUnit
 import java.util.stream.Collectors
+import kotlin.time.Duration.Companion.days
 import kotlin.time.Duration.Companion.milliseconds

 /**
@@ -257,7 +261,8 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc
    val repository = StorageServiceService(SignalNetwork.storageService)

    val localManifest = SignalStore.storageService.manifest
-    val remoteManifest = if (localManifestOutOfDate || localManifest.version < 1 || runAttempt >= 3) {
+    val fetchRemoteManifest = localManifestOutOfDate || localManifest.version < 1 || runAttempt >= 3
+    val remoteManifest = if (fetchRemoteManifest) {
      Log.i(TAG, "Local manifest is invalid. Fetching remote manifest. (localManifestOutOfDate: $localManifestOutOfDate, localManifest.version: ${localManifest.version}, runAttempt: $runAttempt)")
      when (val result = repository.getStorageManifestIfDifferentVersion(storageServiceKey, localManifest.version)) {
        is ManifestIfDifferentVersionResult.DifferentVersion -> result.manifest
@@ -444,7 +449,26 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc
    }
    stopwatch.split("local-data-transaction")

-    if (!remoteWriteOperation.isEmpty) {
+    val loopCheck = if (remoteWriteOperation.isEmpty) {
+      StorageSyncLoopDetector.Decision.Allowed
+    } else {
+      StorageSyncLoopDetector.onWriteAttempt(remoteWriteOperation, fetchRemoteManifest, isRetry = runAttempt > 0)
+    }
+
+    if (remoteWriteOperation.isEmpty) {
+      Log.i(TAG, "No remote writes needed. Still at version: " + remoteManifest.versionString)
+      StorageSyncLoopDetector.onConverged()
+    } 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")
+
+      IssueReporter.report(
+        name = IssueReporter.ISSUE_STORAGE_SYNC_LOOP,
+        description = "Throttling storage service writes. Cause: ${loopCheck.cause}, level: ${loopCheck.level}. $remoteWriteOperation",
+        throwable = Throwable(),
+        priority = IssuePriority.HIGH,
+        notifyCooldown = 3.days
+      )
+    } else {
      Log.i(TAG, "We have something to write remotely.")
      Log.i(TAG, "WriteOperationResult :: $remoteWriteOperation")

@@ -452,9 +476,16 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc

      when (val result = repository.writeStorageRecords(storageServiceKey, remoteWriteOperation.manifest, remoteWriteOperation.inserts, remoteWriteOperation.deletes)) {
        StorageServiceService.WriteStorageRecordsResult.Success -> Unit
-        is StorageServiceService.WriteStorageRecordsResult.StatusCodeError -> throw result.exception
-        is StorageServiceService.WriteStorageRecordsResult.NetworkError -> throw result.exception
+        is StorageServiceService.WriteStorageRecordsResult.StatusCodeError -> {
+          StorageSyncLoopDetector.onWriteFailed()
+          throw result.exception
+        }
+        is StorageServiceService.WriteStorageRecordsResult.NetworkError -> {
+          StorageSyncLoopDetector.onWriteFailed()
+          throw result.exception
+        }
        StorageServiceService.WriteStorageRecordsResult.ConflictError -> {
+          StorageSyncLoopDetector.onWriteFailed()
          Log.w(TAG, "Hit a conflict when trying to resolve the conflict! Retrying.")
          localManifestOutOfDate = true
          throw RetryLaterException()
@@ -468,8 +499,6 @@ class StorageSyncJob private constructor(parameters: Parameters, private var loc
      stopwatch.split("remote-write")

      needsMultiDeviceSync = true
-    } else {
-      Log.i(TAG, "No remote writes needed. Still at version: " + remoteManifest.versionString)
    }

    if (needsForcePush && SignalStore.account.isPrimaryDevice) {
@@ -4,6 +4,7 @@ import org.signal.archive.proto.BackupDebugInfo
 import org.signal.ringrtc.CallManager.DataMode
 import org.thoughtcrime.securesms.BuildConfig
 import org.thoughtcrime.securesms.database.model.IssuePriority
+import org.thoughtcrime.securesms.keyvalue.protos.IssueNotifyTimes
 import org.thoughtcrime.securesms.util.Environment.Calling.defaultSfuUrl
 import org.thoughtcrime.securesms.util.RemoteConfig

@@ -43,6 +44,7 @@ class InternalValues internal constructor(store: KeyValueStore) : SignalStoreVal
    const val USE_NEW_MEDIA_ACTIVITY: String = "internal.use_new_media_activity"
    const val ANR_DETECTION_CRASH: String = "internal.anr_detection_crash"
    const val ISSUE_NOTIFICATION_PRIORITY: String = "internal.issue_notification_priority"
+    const val ISSUE_NOTIFY_TIMES: String = "internal.issue_notify_times"
  }

  public override fun onFirstEverAppLaunch() = Unit
@@ -225,6 +227,9 @@ class InternalValues internal constructor(store: KeyValueStore) : SignalStoreVal
    get() = IssuePriority.fromValue(getInteger(ISSUE_NOTIFICATION_PRIORITY, IssuePriority.HIGH.value))
    set(value) = putInteger(ISSUE_NOTIFICATION_PRIORITY, value.value)

+  /** Persisted so an issue's notification cooldown isn't reset by process death. */
+  var issueNotifyTimes: IssueNotifyTimes by protoValue(ISSUE_NOTIFY_TIMES, IssueNotifyTimes(), IssueNotifyTimes.ADAPTER)
+
  var showArchiveStateHint by booleanValue(SHOW_ARCHIVE_STATE_HINT, false).defaultForExternalUsers()

  /** Whether or not we should include a debuglog in the backup debug info when generating a backup. */
@@ -3,6 +3,7 @@ package org.thoughtcrime.securesms.keyvalue;
 import androidx.annotation.AnyThread;
 import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
+import androidx.annotation.VisibleForTesting;
 import androidx.annotation.WorkerThread;

 import org.signal.core.util.ThreadUtil;
@@ -14,7 +15,7 @@ import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executor;

 /**
 * An replacement for {@link android.content.SharedPreferences} that stores key-value pairs in our
@@ -30,13 +31,18 @@ public final class KeyValueStore implements KeyValueReader {

  private static final String TAG = Log.tag(KeyValueStore.class);

-  private final ExecutorService           executor;
+  private final Executor                  executor;
  private final KeyValuePersistentStorage storage;

  private KeyValueDataSet dataSet;

  public KeyValueStore(@NonNull KeyValuePersistentStorage storage) {
-    this.executor = SignalExecutors.newCachedSingleThreadExecutor("signal-KeyValueStore", ThreadUtil.PRIORITY_BACKGROUND_THREAD);
+    this(storage, SignalExecutors.newCachedSingleThreadExecutor("signal-KeyValueStore", ThreadUtil.PRIORITY_BACKGROUND_THREAD));
+  }
+
+  @VisibleForTesting
+  public KeyValueStore(@NonNull KeyValuePersistentStorage storage, @NonNull Executor executor) {
+    this.executor = executor;
    this.storage  = storage;
  }

@@ -2,6 +2,7 @@ package org.thoughtcrime.securesms.keyvalue

 import org.signal.core.models.storageservice.StorageKey
 import org.signal.core.util.logging.Log
+import org.thoughtcrime.securesms.keyvalue.protos.StorageSyncLoopState
 import org.whispersystems.signalservice.api.storage.SignalStorageManifest

 class StorageServiceValues internal constructor(store: KeyValueStore) : SignalStoreValues(store) {
@@ -11,6 +12,7 @@ class StorageServiceValues internal constructor(store: KeyValueStore) : SignalSt
    private const val LAST_SYNC_TIME = "storage.last_sync_time"
    private const val NEEDS_ACCOUNT_RESTORE = "storage.needs_account_restore"
    private const val MANIFEST = "storage.manifest"
+    private const val SYNC_LOOP_STATE = "storage.sync_loop_state"
  }

  public override fun onFirstEverAppLaunch() = Unit
@@ -26,6 +28,9 @@ class StorageServiceValues internal constructor(store: KeyValueStore) : SignalSt

  var needsAccountRestore: Boolean by booleanValue(NEEDS_ACCOUNT_RESTORE, false)

+  /** Owned by [org.thoughtcrime.securesms.storage.StorageSyncLoopDetector] */
+  var syncLoopState: StorageSyncLoopState by protoValue(SYNC_LOOP_STATE, StorageSyncLoopState(), StorageSyncLoopState.ADAPTER)
+
  var manifest: SignalStorageManifest
    get() {
      val data = getBlob(MANIFEST, null)
@@ -0,0 +1,171 @@
+/*
+ * Copyright 2026 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.storage
+
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.storage.StorageSyncHelper.WriteOperationResult
+import org.thoughtcrime.securesms.util.LeakyBucket
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.hours
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.minutes
+
+/**
+ * Throttles storage service writes when another device keeps undoing them. Left alone, the two devices ping-pong
+ * manifest versions indefinitely, which burns a large amount of battery and network.
+ *
+ * Declining the write is what actually breaks the loop: we only ask other devices to sync after a successful write, so
+ * not writing stops us from prodding the other device, which stops it from prodding us back.
+ *
+ * Two [LeakyBucket]s, both persisted so a loop can't shake them off by killing the process:
+ *
+ * - The content bucket is only charged when a write repeats one of the last [FINGERPRINT_HISTORY] payload fingerprints
+ *   on a run that fetched the remote manifest. Novel content is always free, and so is a run that reused the local
+ *   manifest, because a loop only ever writes in reaction to a peer's manifest. Storage ids and deletes are left out of the
+ *   fingerprint because they rotate on every write, loop or not.
+ * - The rate bucket is charged for every write on a run that fetched the remote manifest, regardless of content. It is
+ *   much larger, and exists to bound loops the content bucket can't see -- ones whose payloads aren't stable.
+ */
+object StorageSyncLoopDetector {
+
+  /** How many past payload fingerprints a write is compared against. */
+  private const val FINGERPRINT_HISTORY = 3
+
+  private val contentBucket = LeakyBucket(
+    capacity = 3,
+    dripInterval = 1.hours,
+    state = ContentBucketState
+  )
+
+  private val rateBucket = LeakyBucket(
+    capacity = 100,
+    dripInterval = 10.minutes,
+    state = RateBucketState
+  )
+
+  /**
+   * Charges a write against whichever buckets apply and reports whether it may proceed. Call once per write attempt:
+   * levels rise here, not on success, so a write that fails still costs what it would have.
+   *
+   * Job retries are exempt entirely, which keeps that cost at once per failure rather than once per attempt.
+   */
+  @Synchronized
+  fun onWriteAttempt(write: WriteOperationResult, fetchedRemoteManifest: Boolean, isRetry: Boolean, now: Duration = System.currentTimeMillis().milliseconds): Decision {
+    if (!SignalStore.account.isMultiDevice) {
+      return Decision.Allowed
+    }
+
+    if (isRetry) {
+      return Decision.Allowed
+    }
+
+    val fingerprint = fingerprint(write)
+    val chargeContent = fetchedRemoteManifest && fingerprint != null && SignalStore.storageService.syncLoopState.recentFingerprints.contains(fingerprint)
+
+    if (chargeContent && !contentBucket.hasRoom(now)) {
+      return Decision.Denied(Cause.REPEATED_PAYLOAD, contentBucket.level(now))
+    }
+
+    if (fetchedRemoteManifest && !rateBucket.hasRoom(now)) {
+      return Decision.Denied(Cause.WRITE_RATE, rateBucket.level(now))
+    }
+
+    if (chargeContent) {
+      contentBucket.use(now)
+    }
+
+    if (fetchedRemoteManifest) {
+      rateBucket.use(now)
+    }
+
+    if (fingerprint != null) {
+      remember(fingerprint)
+    }
+
+    return Decision.Allowed
+  }
+
+  /**
+   * Refunds a write that likely never landed.
+   */
+  @Synchronized
+  fun onWriteFailed(now: Duration = System.currentTimeMillis().milliseconds) {
+    contentBucket.refund(now)
+    rateBucket.refund(now)
+  }
+
+  /**
+   * Called when a sync had nothing to write, meaning we agree with the remote state. Only the content bucket clears:
+   * agreement says the payload disagreement resolved, but says nothing about the write volume the rate bucket tracks.
+   */
+  @Synchronized
+  fun onConverged() {
+    contentBucket.clear()
+  }
+
+  private fun remember(fingerprint: Int) {
+    val state = SignalStore.storageService.syncLoopState
+
+    SignalStore.storageService.syncLoopState = state.copy(
+      recentFingerprints = (listOf(fingerprint) + state.recentFingerprints).distinct().take(FINGERPRINT_HISTORY)
+    )
+  }
+
+  /** Null when there's nothing content-addressable to compare, i.e. a write of nothing but deletes. */
+  private fun fingerprint(write: WriteOperationResult): Int? {
+    if (write.inserts.isEmpty()) {
+      return null
+    }
+
+    return write.inserts
+      .map { it.proto.encode().contentHashCode() }
+      .sorted()
+      .hashCode()
+  }
+
+  private object ContentBucketState : LeakyBucket.State {
+    override val level: Int
+      get() = SignalStore.storageService.syncLoopState.contentLevel
+
+    override val levelUpdatedAt: Long
+      get() = SignalStore.storageService.syncLoopState.contentLevelAsOf
+
+    override fun update(level: Int, levelAsOf: Long) {
+      SignalStore.storageService.syncLoopState = SignalStore.storageService.syncLoopState.copy(
+        contentLevel = level,
+        contentLevelAsOf = levelAsOf
+      )
+    }
+  }
+
+  private object RateBucketState : LeakyBucket.State {
+    override val level: Int
+      get() = SignalStore.storageService.syncLoopState.rateLevel
+
+    override val levelUpdatedAt: Long
+      get() = SignalStore.storageService.syncLoopState.rateLevelAsOf
+
+    override fun update(level: Int, levelAsOf: Long) {
+      SignalStore.storageService.syncLoopState = SignalStore.storageService.syncLoopState.copy(
+        rateLevel = level,
+        rateLevelAsOf = levelAsOf
+      )
+    }
+  }
+
+  enum class Cause {
+    REPEATED_PAYLOAD,
+    WRITE_RATE
+  }
+
+  sealed interface Decision {
+    /** The write should proceed as normal. */
+    data object Allowed : Decision
+
+    /** The write found no room in a bucket and should be skipped. */
+    data class Denied(val cause: Cause, val level: Int) : Decision
+  }
+}
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2026 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.util
+
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.milliseconds
+
+/**
+ * A leaky bucket, in the same spirit as [LeakyBucketLimiter]: each use raises the level, and the level drips back down
+ * one per [dripInterval]. Once the level reaches [capacity] there is no room until enough has dripped away.
+ *
+ * Not thread safe. [hasRoom] and [use] are separate calls so a caller can check several buckets before charging any of
+ * them, so callers need to serialize access themselves anyway.
+ */
+class LeakyBucket(
+  private val capacity: Int,
+  private val dripInterval: Duration,
+  private val state: State
+) {
+
+  /** The level after everything that has dripped away is credited, without recording that drip. */
+  fun level(now: Duration = System.currentTimeMillis().milliseconds): Int {
+    return calculateStateForCurrentTime(now).level
+  }
+
+  fun hasRoom(now: Duration = System.currentTimeMillis().milliseconds): Boolean {
+    return level(now) < capacity
+  }
+
+  /** Raises the level by one, recording any drip along the way. Only call when [hasRoom]. */
+  fun use(now: Duration = System.currentTimeMillis().milliseconds) {
+    val currentState = calculateStateForCurrentTime(now)
+
+    state.update(currentState.level + 1, currentState.levelUpdatedAt)
+  }
+
+  /** Lowers the level by one, recording any drip along the way. Pairs with [use] for work that ended up not happening. */
+  fun refund(now: Duration = System.currentTimeMillis().milliseconds) {
+    val currentState = calculateStateForCurrentTime(now)
+
+    state.update((currentState.level - 1).coerceAtLeast(0), currentState.levelUpdatedAt)
+  }
+
+  fun clear() {
+    state.update(0, 0)
+  }
+
+  /**
+   * [Snapshot.levelUpdatedAt] advances by whole drip intervals only, so a partially elapsed one carries toward the next drip
+   * rather than being discarded.
+   */
+  private fun calculateStateForCurrentTime(now: Duration): Snapshot {
+    val level = state.level
+    val levelAsOf = state.levelUpdatedAt
+    val elapsed = now - levelAsOf.milliseconds
+
+    // The level only rises when there was room, so a clock behind levelAsOf would report a full bucket forever with
+    // nothing left to advance the timestamp that drains it. Empty it rather than wedge until the clock catches up.
+    if (level <= 0 || elapsed < Duration.ZERO) {
+      return Snapshot(0, now.inWholeMilliseconds)
+    }
+
+    val drips = (elapsed / dripInterval).toInt()
+
+    return Snapshot((level - drips).coerceAtLeast(0), levelAsOf + (dripInterval * drips).inWholeMilliseconds)
+  }
+
+  private data class Snapshot(val level: Int, val levelUpdatedAt: Long)
+
+  interface State {
+    val level: Int
+    val levelUpdatedAt: Long
+
+    fun update(level: Int, levelAsOf: Long)
+  }
+
+  class InMemoryState : State {
+    override var level: Int = 0
+      private set
+
+    override var levelUpdatedAt: Long = 0
+      private set
+
+    override fun update(level: Int, levelAsOf: Long) {
+      this.level = level
+      this.levelUpdatedAt = levelAsOf
+    }
+  }
+}
@@ -103,4 +103,16 @@ message BackupDownloadNotifierState {
    uint64 lastSheetDisplaySeconds = 2;
    uint64 intervalSeconds = 3;
    Type type = 4;
-}
+}
+
+message IssueNotifyTimes {
+    map<string, uint64> lastNotifyTimeByName = 1;
+}
+
+message StorageSyncLoopState {
+    repeated sint32 recentFingerprints = 1;
+    uint32 contentLevel                = 2;
+    uint64 contentLevelAsOf            = 3;
+    uint32 rateLevel                   = 4;
+    uint64 rateLevelAsOf               = 5;
+}
@@ -7,9 +7,12 @@ package org.thoughtcrime.securesms.jobs

 import android.app.Application
 import androidx.test.core.app.ApplicationProvider
+import io.mockk.CapturingSlot
 import io.mockk.every
 import io.mockk.mockkObject
+import io.mockk.slot
 import io.mockk.unmockkObject
+import io.mockk.verify
 import okio.ByteString.Companion.toByteString
 import org.junit.After
 import org.junit.Assert.assertArrayEquals
@@ -31,8 +34,10 @@ import org.signal.core.util.Util
 import org.signal.core.util.logging.Log
 import org.signal.core.util.update
 import org.signal.core.util.withinTransaction
+import org.thoughtcrime.securesms.database.IssueReporter
 import org.thoughtcrime.securesms.database.RecipientTable
 import org.thoughtcrime.securesms.database.SignalDatabase
+import org.thoughtcrime.securesms.database.model.IssuePriority
 import org.thoughtcrime.securesms.database.model.StickerPackId
 import org.thoughtcrime.securesms.groups.GroupId
 import org.thoughtcrime.securesms.jobmanager.Job
@@ -91,6 +96,7 @@ class StorageSyncJobTest {
  @After
  fun tearDown() {
    unmockkObject(RemoteConfig)
+    unmockkObject(IssueReporter)
  }

  @Test
@@ -384,6 +390,79 @@ class StorageSyncJobTest {
    assertEquals(0, remoteStorage.records.count { it.proto.contact != null })
  }

+  @Test
+  fun `given another device keeps undoing my write, when I run again, then I stop writing`() {
+    stubIssueReporter()
+
+    val throttledRun = loopWritesUntilThrottled()
+
+    assertTrue(throttledRun.isSuccess)
+    assertEquals(0, remoteStorage.writeCount)
+    assertEquals(0, remoteStorage.records.count { it.proto.contact?.givenName == "Loop" })
+  }
+
+  @Test
+  fun `given another device keeps undoing my write, when I throttle it, then I report it at high priority`() {
+    val priority = stubIssueReporter()
+
+    loopWritesUntilThrottled()
+
+    verify { IssueReporter.report(any(), any(), any(), any(), any(), any()) }
+    assertEquals(IssuePriority.HIGH, priority.captured)
+  }
+
+  @Test
+  fun `given a local-only contact the other device leaves alone, when I run repeatedly, then I keep writing`() {
+    stubIssueReporter()
+    every { recipients.signalStore.account.isMultiDevice } returns true
+    SignalDatabase.recipients.rotateStorageId(recipients.createRecipient("Local Contact"))
+
+    check(runJob(StorageSyncJob.forLocalChange()).isSuccess)
+
+    repeat(4) {
+      SignalDatabase.recipients.rotateStorageId(recipients.createRecipient("Local Contact ${it + 2}"))
+      remoteStorage.resetCounters()
+
+      assertTrue(runJob(StorageSyncJob.forLocalChange()).isSuccess)
+      assertEquals(1, remoteStorage.writeCount)
+    }
+  }
+
+  /**
+   * Writes the same contact, with the other device deleting it again after each, until a run gets throttled. Returns
+   * that run, leaving [FakeStorageServiceRule.writeCount] counting only it. Driven by observation rather than a fixed
+   * count because the record's payload settles after its first sync, so the number of passes isn't fixed.
+   */
+  private fun loopWritesUntilThrottled(): Job.Result {
+    every { recipients.signalStore.account.isMultiDevice } returns true
+    SignalDatabase.recipients.rotateStorageId(recipients.createRecipient("Loop Contact"))
+
+    repeat(12) { pass ->
+      remoteStorage.resetCounters()
+
+      val result = runJob(StorageSyncJob.forRemoteChange())
+      check(result.isSuccess) { "Loop write ${pass + 1} failed!" }
+
+      if (remoteStorage.writeCount == 0) {
+        return result
+      }
+
+      val withoutLoopContact = remoteStorage.records.filterNot { record -> record.proto.contact?.givenName == "Loop" }
+      remoteStorage.setRemoteState(withoutLoopContact, version = remoteStorage.manifest!!.version + 1)
+    }
+
+    throw AssertionError("Writes were never throttled!")
+  }
+
+  private fun stubIssueReporter(): CapturingSlot<IssuePriority> {
+    val priority = slot<IssuePriority>()
+
+    mockkObject(IssueReporter)
+    every { IssueReporter.report(any(), any(), any(), capture(priority), any(), any()) } returns Unit
+
+    return priority
+  }
+
  /**
   * Gets us to a steady state: remote holds our account record at version 1, then a sync pushes up everything else
   * the fresh database came with (the default chat folder), leaving both sides at [BASE_MANIFEST_VERSION].
@@ -0,0 +1,436 @@
+/*
+ * Copyright 2026 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.storage
+
+import android.app.Application
+import org.junit.Assert.assertEquals
+import org.junit.Before
+import org.junit.Rule
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import org.robolectric.annotation.Config
+import org.signal.core.util.Util
+import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.keyvalue.protos.StorageSyncLoopState
+import org.thoughtcrime.securesms.storage.StorageSyncHelper.WriteOperationResult
+import org.thoughtcrime.securesms.testutil.SignalStoreRule
+import org.thoughtcrime.securesms.testutil.TestHelpers
+import org.whispersystems.signalservice.api.storage.SignalStorageManifest
+import org.whispersystems.signalservice.api.storage.SignalStorageRecord
+import org.whispersystems.signalservice.api.storage.StorageId
+import org.whispersystems.signalservice.internal.storage.protos.ContactRecord
+import org.whispersystems.signalservice.internal.storage.protos.StorageRecord
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.hours
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.minutes
+
+@RunWith(RobolectricTestRunner::class)
+@Config(manifest = Config.NONE, application = Application::class)
+class StorageSyncLoopDetectorTest {
+
+  companion object {
+    private val NOW = 1_700_000_000_000L.milliseconds
+  }
+
+  @get:Rule
+  val signalStore = SignalStoreRule()
+
+  @Before
+  fun setup() {
+    SignalStore.storageService.syncLoopState = StorageSyncLoopState()
+
+    // Throttling only applies when another device could be undoing our writes.
+    SignalStore.account.isMultiDevice = true
+  }
+
+  @Test
+  fun `given no linked devices, when I repeat the same write, then I expect it to always be allowed`() {
+    SignalStore.account.isMultiDevice = false
+
+    repeat(20) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1)))
+    }
+  }
+
+  @Test
+  fun `given no linked devices, when I repeat the same write, then I expect nothing to be charged`() {
+    SignalStore.account.isMultiDevice = false
+
+    repeat(20) { attempt(writeOf(1)) }
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+    assertEquals(0, SignalStore.storageService.syncLoopState.rateLevel)
+  }
+
+  @Test
+  fun `given a write that failed, when I refund it, then I expect the charge to come back`() {
+    attempt(writeOf(1))
+    attempt(writeOf(1))
+    val chargedLevel = SignalStore.storageService.syncLoopState.contentLevel
+
+    StorageSyncLoopDetector.onWriteFailed(NOW)
+
+    assertEquals(chargedLevel - 1, SignalStore.storageService.syncLoopState.contentLevel)
+  }
+
+  @Test
+  fun `given writes that keep failing, when I retry forever, then I expect to never be throttled`() {
+    repeat(20) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1)))
+      StorageSyncLoopDetector.onWriteFailed(NOW)
+    }
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+    assertEquals(0, SignalStore.storageService.syncLoopState.rateLevel)
+  }
+
+  @Test
+  fun `given nothing charged, when I refund, then I expect the level to stay at zero`() {
+    StorageSyncLoopDetector.onWriteFailed(NOW)
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+    assertEquals(0, SignalStore.storageService.syncLoopState.rateLevel)
+  }
+
+  @Test
+  fun `allows a write nothing like the ones before it`() {
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1)))
+  }
+
+  @Test
+  fun `never charges the content bucket for novel payloads`() {
+    repeat(50) { index ->
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(index), fetchedRemoteManifest = false))
+    }
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+  }
+
+  @Test
+  fun `charges the content bucket for a repeated payload and denies once it is dry`() {
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1)))
+
+    repeat(3) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1)))
+    }
+
+    assertEquals(
+      StorageSyncLoopDetector.Decision.Denied(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, 3),
+      attempt(writeOf(1))
+    )
+  }
+
+  @Test
+  fun `compares against the last three payloads, not just the previous one`() {
+    attempt(writeOf(1))
+    attempt(writeOf(2))
+    attempt(writeOf(3))
+
+    repeat(3) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1)))
+    }
+
+    assertEquals(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, denialCause(attempt(writeOf(1))))
+  }
+
+  @Test
+  fun `forgets a payload once three newer ones have been written`() {
+    attempt(writeOf(1))
+    attempt(writeOf(2))
+    attempt(writeOf(3))
+    attempt(writeOf(4))
+
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1)))
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+  }
+
+  @Test
+  fun `restores one content permit per hour`() {
+    exhaustContentBucket()
+
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), now = NOW + 1.hours))
+    assertEquals(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, denialCause(attempt(writeOf(1), now = NOW + 1.hours)))
+  }
+
+  @Test
+  fun `restores several content permits at once when enough time has passed`() {
+    exhaustContentBucket()
+
+    repeat(3) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), now = NOW + 3.hours))
+    }
+
+    assertEquals(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, denialCause(attempt(writeOf(1), now = NOW + 3.hours)))
+  }
+
+  @Test
+  fun `never restores more permits than the bucket holds`() {
+    exhaustContentBucket()
+
+    repeat(3) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), now = NOW + 500.hours))
+    }
+
+    assertEquals(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, denialCause(attempt(writeOf(1), now = NOW + 500.hours)))
+  }
+
+  @Test
+  fun `refills rather than wedging when the clock moves backwards`() {
+    exhaustContentBucket()
+
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), now = NOW - 5.hours))
+  }
+
+  /** The rate bucket has no convergence reset, so a wedge here would deny manifest-fetching writes indefinitely. */
+  @Test
+  fun `refills the rate bucket when the clock moves backwards`() {
+    exhaustRateBucket()
+
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(100), now = NOW - 5.hours))
+  }
+
+  @Test
+  fun `recovers normal throttling after a backwards clock write`() {
+    exhaustContentBucket()
+
+    val past = NOW - 5.hours
+
+    repeat(3) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), now = past))
+    }
+
+    assertEquals(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, denialCause(attempt(writeOf(1), now = past)))
+  }
+
+  /** A partially elapsed interval carries toward the next drip instead of restarting, unlike a strike counter. */
+  @Test
+  fun `keeps a partially elapsed interval toward the next drip`() {
+    exhaustContentBucket()
+
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), now = NOW + 1.hours + 59.minutes))
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), now = NOW + 2.hours + 1.minutes))
+  }
+
+  /** Twice the drip rate accumulates a level of one per hour, which a strike counter reset by any gap would not. */
+  @Test
+  fun `fills the bucket when writes outpace the drip rate`() {
+    repeat(6) { index ->
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), now = NOW + (index * 30).minutes))
+    }
+
+    assertEquals(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, denialCause(attempt(writeOf(1), now = NOW + 180.minutes)))
+  }
+
+  @Test
+  fun `does not credit the same elapsed time twice`() {
+    exhaustContentBucket()
+
+    repeat(4) {
+      assertEquals(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, denialCause(attempt(writeOf(1), now = NOW + 59.minutes)))
+    }
+  }
+
+  @Test
+  fun `never charges the content bucket for a delete-only write`() {
+    repeat(30) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(deletes = intArrayOf(1)), fetchedRemoteManifest = false))
+    }
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+  }
+
+  @Test
+  fun `never charges the content bucket for a repeated payload when the local manifest was reused`() {
+    repeat(30) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), fetchedRemoteManifest = false))
+    }
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+  }
+
+  /** Toggling a setting back and forth reproduces a previous payload, but it is a genuine local change rather than a loop. */
+  @Test
+  fun `never throttles a setting toggled back and forth`() {
+    repeat(20) { index ->
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(index % 2), fetchedRemoteManifest = false))
+    }
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+  }
+
+  /** Local writes still populate the history, so a peer that starts undoing them is caught on the first repeat we see. */
+  @Test
+  fun `builds fingerprint history from writes that reused the local manifest`() {
+    repeat(4) {
+      attempt(writeOf(1), fetchedRemoteManifest = false)
+    }
+
+    repeat(3) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1)))
+    }
+
+    assertEquals(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, denialCause(attempt(writeOf(1))))
+  }
+
+  @Test
+  fun `does not charge the rate bucket when the local manifest was reused`() {
+    repeat(50) { index ->
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(index), fetchedRemoteManifest = false))
+    }
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.rateLevel)
+  }
+
+  @Test
+  fun `charges the rate bucket for every write on a manifest-fetching run`() {
+    repeat(100) { index ->
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(index)))
+    }
+
+    assertEquals(100, SignalStore.storageService.syncLoopState.rateLevel)
+    assertEquals(StorageSyncLoopDetector.Cause.WRITE_RATE, denialCause(attempt(writeOf(999))))
+  }
+
+  /** The rate bucket is content-blind, so a run of delete-only writes drains it just the same. */
+  @Test
+  fun `charges the rate bucket for delete-only writes too`() {
+    repeat(100) { index ->
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(deletes = intArrayOf(index))))
+    }
+
+    assertEquals(StorageSyncLoopDetector.Cause.WRITE_RATE, denialCause(attempt(writeOf(deletes = intArrayOf(99)))))
+  }
+
+  @Test
+  fun `drips one rate level every ten minutes`() {
+    exhaustRateBucket()
+
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1000), now = NOW + 10.minutes))
+    assertEquals(StorageSyncLoopDetector.Cause.WRITE_RATE, denialCause(attempt(writeOf(1001), now = NOW + 10.minutes)))
+  }
+
+  @Test
+  fun `converging resets the content bucket`() {
+    exhaustContentBucket()
+
+    StorageSyncLoopDetector.onConverged()
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+  }
+
+  @Test
+  fun `converging leaves the rate bucket alone`() {
+    exhaustRateBucket()
+
+    StorageSyncLoopDetector.onConverged()
+
+    assertEquals(100, SignalStore.storageService.syncLoopState.rateLevel)
+    assertEquals(StorageSyncLoopDetector.Cause.WRITE_RATE, denialCause(attempt(writeOf(999))))
+  }
+
+  @Test
+  fun `converging lets a throttled write through again`() {
+    exhaustContentBucket()
+    check(attempt(writeOf(1)) is StorageSyncLoopDetector.Decision.Denied) { "Should be throttled before converging!" }
+
+    StorageSyncLoopDetector.onConverged()
+
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1)))
+  }
+
+  @Test
+  fun `persists bucket state so it survives a process restart`() {
+    exhaustContentBucket()
+
+    assertEquals(3, SignalStore.storageService.syncLoopState.contentLevel)
+    assertEquals(NOW.inWholeMilliseconds, SignalStore.storageService.syncLoopState.contentLevelAsOf)
+  }
+
+  /** Spends every content permit, leaving [writeOf] 1 as a repeat that needs one. */
+  private fun exhaustContentBucket() {
+    repeat(4) {
+      check(attempt(writeOf(1)) is StorageSyncLoopDetector.Decision.Allowed) { "Content bucket drained early!" }
+    }
+  }
+
+  @Test
+  fun `never charges a retry`() {
+    repeat(50) {
+      assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), isRetry = true))
+    }
+
+    assertEquals(0, SignalStore.storageService.syncLoopState.contentLevel)
+    assertEquals(0, SignalStore.storageService.syncLoopState.rateLevel)
+  }
+
+  @Test
+  fun `a retry does not build up fingerprint history`() {
+    repeat(5) {
+      attempt(writeOf(1), isRetry = true)
+    }
+
+    assertEquals(emptyList<Int>(), SignalStore.storageService.syncLoopState.recentFingerprints)
+  }
+
+  @Test
+  fun `allows a retry of a repeated payload even when the content bucket is full`() {
+    repeat(4) {
+      attempt(writeOf(1))
+    }
+
+    assertEquals(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, denialCause(attempt(writeOf(1))))
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(1), isRetry = true))
+  }
+
+  @Test
+  fun `allows a retry even when the rate bucket is exhausted`() {
+    exhaustRateBucket()
+
+    assertEquals(StorageSyncLoopDetector.Cause.WRITE_RATE, denialCause(attempt(writeOf(500))))
+    assertEquals(StorageSyncLoopDetector.Decision.Allowed, attempt(writeOf(500), isRetry = true))
+  }
+
+  private fun exhaustRateBucket() {
+    repeat(100) { index ->
+      check(attempt(writeOf(index)) is StorageSyncLoopDetector.Decision.Allowed) { "Rate bucket drained early!" }
+    }
+  }
+
+  private fun attempt(
+    write: WriteOperationResult,
+    fetchedRemoteManifest: Boolean = true,
+    isRetry: Boolean = false,
+    now: Duration = NOW
+  ): StorageSyncLoopDetector.Decision {
+    return StorageSyncLoopDetector.onWriteAttempt(write, fetchedRemoteManifest, isRetry, now)
+  }
+
+  private fun denial(decision: StorageSyncLoopDetector.Decision): StorageSyncLoopDetector.Decision.Denied {
+    return decision as? StorageSyncLoopDetector.Decision.Denied ?: throw AssertionError("Expected a denial, got $decision")
+  }
+
+  private fun denialCause(decision: StorageSyncLoopDetector.Decision): StorageSyncLoopDetector.Cause {
+    return denial(decision).cause
+  }
+
+  private fun writeOf(vararg payloads: Int, deletes: IntArray = IntArray(0)): WriteOperationResult {
+    return WriteOperationResult(
+      manifest = SignalStorageManifest.EMPTY,
+      inserts = payloads.map { contactRecord(it) },
+      deletes = deletes.map { TestHelpers.byteArray(it) }
+    )
+  }
+
+  /** One distinct payload per [payload], under a storage id that rotates on every call the way a real write does. */
+  private fun contactRecord(payload: Int): SignalStorageRecord {
+    return SignalStorageRecord(
+      id = StorageId.forContact(Util.getSecretBytes(16)),
+      proto = StorageRecord(contact = ContactRecord(e164 = "+1555000$payload"))
+    )
+  }
+}
@@ -19,6 +19,7 @@ import org.signal.network.service.StorageServiceService
 import org.thoughtcrime.securesms.keyvalue.PhoneNumberPrivacyValues.PhoneNumberDiscoverabilityMode
 import org.thoughtcrime.securesms.keyvalue.PhoneNumberPrivacyValues.PhoneNumberSharingMode
 import org.thoughtcrime.securesms.keyvalue.SignalStore
+import org.thoughtcrime.securesms.keyvalue.protos.StorageSyncLoopState
 import org.thoughtcrime.securesms.net.SignalNetwork
 import org.thoughtcrime.securesms.util.RemoteConfig
 import org.whispersystems.signalservice.api.storage.RecordIkm
@@ -108,12 +109,16 @@ class FakeStorageServiceRule(val storageKey: StorageKey = StorageKey(Util.getSec
   */
  fun stubDefaults(store: MockSignalStoreRule) {
    var localManifest = SignalStorageManifest.EMPTY
+    var syncLoopState = StorageSyncLoopState()

    every { store.storageService.manifest } answers { localManifest }
    every { store.storageService.manifest = any() } answers { localManifest = firstArg() }
    every { store.storageService.storageKey } returns storageKey
    every { store.storageService.storageKeyForInitialDataRestore } returns null

+    every { store.storageService.syncLoopState } answers { syncLoopState }
+    every { store.storageService.syncLoopState = any() } answers { syncLoopState = firstArg() }
+
    every { store.svr.hasPin() } returns true
    every { store.svr.hasOptedOut() } returns false

@@ -22,7 +22,7 @@ class SignalStoreRule : ExternalResource() {

  override fun before() {
    val application = ApplicationProvider.getApplicationContext<Application>()
-    SignalStore.testInject(SignalStore(application, KeyValueStore(MockKeyValuePersistentStorage.withDataSet(KeyValueDataSet()))))
+    SignalStore.testInject(SignalStore(application, KeyValueStore(MockKeyValuePersistentStorage.withDataSet(KeyValueDataSet()), DirectExecutor())))
  }

  override fun after() {
@@ -0,0 +1,182 @@
+/*
+ * Copyright 2026 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.thoughtcrime.securesms.util
+
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.hours
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.minutes
+
+class LeakyBucketTest {
+
+  companion object {
+    private val NOW = 1_700_000_000_000L.milliseconds
+    private val DRIP = 15.minutes
+  }
+
+  private val state = LeakyBucket.InMemoryState()
+  private val bucket = LeakyBucket(capacity = 3, dripInterval = DRIP, state = state)
+
+  @Test
+  fun `starts empty`() {
+    assertEquals(0, bucket.level(NOW))
+    assertTrue(bucket.hasRoom(NOW))
+  }
+
+  @Test
+  fun `fills to capacity and then has no room`() {
+    repeat(3) {
+      assertTrue(bucket.hasRoom(NOW))
+      bucket.use(NOW)
+    }
+
+    assertEquals(3, bucket.level(NOW))
+    assertFalse(bucket.hasRoom(NOW))
+  }
+
+  @Test
+  fun `drips one level per interval`() {
+    fill()
+
+    assertEquals(2, bucket.level(NOW + DRIP))
+    assertTrue(bucket.hasRoom(NOW + DRIP))
+  }
+
+  @Test
+  fun `drips several levels at once when enough time has passed`() {
+    fill()
+
+    assertEquals(1, bucket.level(NOW + DRIP * 2))
+    assertEquals(0, bucket.level(NOW + DRIP * 3))
+  }
+
+  @Test
+  fun `never drips below empty`() {
+    fill()
+
+    assertEquals(0, bucket.level(NOW + DRIP * 500))
+  }
+
+  @Test
+  fun `does not drip for a partially elapsed interval`() {
+    fill()
+
+    assertEquals(3, bucket.level(NOW + DRIP - 1.milliseconds))
+  }
+
+  @Test
+  fun `carries a partially elapsed interval toward the next drip`() {
+    fill()
+
+    // A drip and most of a second one. Using the bucket banks the drip and keeps the remainder.
+    bucket.use(NOW + DRIP + 14.minutes)
+
+    assertEquals(NOW.inWholeMilliseconds + DRIP.inWholeMilliseconds, state.levelUpdatedAt)
+    assertEquals(3, state.level)
+
+    // One more minute crosses the second interval, rather than restarting from the last use.
+    assertEquals(2, bucket.level(NOW + DRIP + 15.minutes))
+  }
+
+  @Test
+  fun `reading the level does not record the drip`() {
+    fill()
+
+    bucket.level(NOW + DRIP)
+
+    assertEquals(3, state.level)
+    assertEquals(NOW.inWholeMilliseconds, state.levelUpdatedAt)
+  }
+
+  @Test
+  fun `using the bucket records the drip`() {
+    fill()
+
+    bucket.use(NOW + DRIP)
+
+    assertEquals(3, state.level)
+    assertEquals(NOW.inWholeMilliseconds + DRIP.inWholeMilliseconds, state.levelUpdatedAt)
+  }
+
+  @Test
+  fun `empties rather than wedging when the clock moves backwards`() {
+    fill()
+
+    assertEquals(0, bucket.level(NOW - 5.hours))
+    assertTrue(bucket.hasRoom(NOW - 5.hours))
+  }
+
+  @Test
+  fun `throttles again after a backwards clock use`() {
+    fill()
+
+    val past = NOW - 5.hours
+
+    repeat(3) {
+      assertTrue(bucket.hasRoom(past))
+      bucket.use(past)
+    }
+
+    assertFalse(bucket.hasRoom(past))
+  }
+
+  @Test
+  fun `clear empties the bucket`() {
+    fill()
+
+    bucket.clear()
+
+    assertEquals(0, bucket.level(NOW))
+    assertEquals(0, state.level)
+  }
+
+  @Test
+  fun `fills when uses outpace the drip rate`() {
+    var now = NOW
+
+    // Twice the drip rate, so the level nets up one per interval and a capacity of three fills on the fifth use.
+    repeat(5) {
+      assertTrue(bucket.hasRoom(now))
+      bucket.use(now)
+      now += DRIP / 2
+    }
+
+    assertFalse(bucket.hasRoom(now))
+  }
+
+  @Test
+  fun `never fills when uses match the drip rate`() {
+    var now = NOW
+
+    repeat(50) {
+      assertTrue(bucket.hasRoom(now))
+      bucket.use(now)
+      now += DRIP
+    }
+  }
+
+  @Test
+  fun `reads its level from the state it was given`() {
+    state.update(level = 2, levelAsOf = NOW.inWholeMilliseconds)
+
+    assertEquals(2, bucket.level(NOW))
+    assertTrue(bucket.hasRoom(NOW))
+
+    bucket.use(NOW)
+
+    assertFalse(bucket.hasRoom(NOW))
+  }
+
+  private fun fill(now: Duration = NOW) {
+    repeat(3) {
+      bucket.use(now)
+    }
+  }
+}