From 9c01310df4bb86a0ecb8479b382b4f7a60260570 Mon Sep 17 00:00:00 2001 From: Cody Henthorne Date: Fri, 14 Aug 2026 16:08:10 -0400 Subject: [PATCH] Add storage sync hot loop detection and rate limiting. --- .../securesms/database/IssueReporter.kt | 38 +- .../securesms/jobs/StorageSyncJob.kt | 41 +- .../securesms/keyvalue/InternalValues.kt | 5 + .../securesms/keyvalue/KeyValueStore.java | 12 +- .../keyvalue/StorageServiceValues.kt | 5 + .../storage/StorageSyncLoopDetector.kt | 171 +++++++ .../securesms/util/LeakyBucket.kt | 92 ++++ app/src/main/protowire/KeyValue.proto | 14 +- .../securesms/jobs/StorageSyncJobTest.kt | 79 ++++ .../storage/StorageSyncLoopDetectorTest.kt | 436 ++++++++++++++++++ .../testutil/FakeStorageServiceRule.kt | 5 + .../securesms/testutil/SignalStoreRule.kt | 2 +- .../securesms/util/LeakyBucketTest.kt | 182 ++++++++ 13 files changed, 1064 insertions(+), 18 deletions(-) create mode 100644 app/src/main/java/org/thoughtcrime/securesms/storage/StorageSyncLoopDetector.kt create mode 100644 app/src/main/java/org/thoughtcrime/securesms/util/LeakyBucket.kt create mode 100644 app/src/test/java/org/thoughtcrime/securesms/storage/StorageSyncLoopDetectorTest.kt create mode 100644 app/src/test/java/org/thoughtcrime/securesms/util/LeakyBucketTest.kt diff --git a/app/src/main/java/org/thoughtcrime/securesms/database/IssueReporter.kt b/app/src/main/java/org/thoughtcrime/securesms/database/IssueReporter.kt index 666beb73b7..d545b14589 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/database/IssueReporter.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/database/IssueReporter.kt @@ -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 { diff --git a/app/src/main/java/org/thoughtcrime/securesms/jobs/StorageSyncJob.kt b/app/src/main/java/org/thoughtcrime/securesms/jobs/StorageSyncJob.kt index e23a8288be..0627a27375 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/jobs/StorageSyncJob.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/jobs/StorageSyncJob.kt @@ -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) { diff --git a/app/src/main/java/org/thoughtcrime/securesms/keyvalue/InternalValues.kt b/app/src/main/java/org/thoughtcrime/securesms/keyvalue/InternalValues.kt index 3e2ed8c20a..42bc5bb300 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/keyvalue/InternalValues.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/keyvalue/InternalValues.kt @@ -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. */ diff --git a/app/src/main/java/org/thoughtcrime/securesms/keyvalue/KeyValueStore.java b/app/src/main/java/org/thoughtcrime/securesms/keyvalue/KeyValueStore.java index 8bc2a7f35e..6c22affcd7 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/keyvalue/KeyValueStore.java +++ b/app/src/main/java/org/thoughtcrime/securesms/keyvalue/KeyValueStore.java @@ -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; } diff --git a/app/src/main/java/org/thoughtcrime/securesms/keyvalue/StorageServiceValues.kt b/app/src/main/java/org/thoughtcrime/securesms/keyvalue/StorageServiceValues.kt index ba0b4aae3d..d6e70115a9 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/keyvalue/StorageServiceValues.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/keyvalue/StorageServiceValues.kt @@ -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) diff --git a/app/src/main/java/org/thoughtcrime/securesms/storage/StorageSyncLoopDetector.kt b/app/src/main/java/org/thoughtcrime/securesms/storage/StorageSyncLoopDetector.kt new file mode 100644 index 0000000000..983a9fd824 --- /dev/null +++ b/app/src/main/java/org/thoughtcrime/securesms/storage/StorageSyncLoopDetector.kt @@ -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 + } +} diff --git a/app/src/main/java/org/thoughtcrime/securesms/util/LeakyBucket.kt b/app/src/main/java/org/thoughtcrime/securesms/util/LeakyBucket.kt new file mode 100644 index 0000000000..11e38e2698 --- /dev/null +++ b/app/src/main/java/org/thoughtcrime/securesms/util/LeakyBucket.kt @@ -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 + } + } +} diff --git a/app/src/main/protowire/KeyValue.proto b/app/src/main/protowire/KeyValue.proto index d71d918ed5..b5f1dbfa59 100644 --- a/app/src/main/protowire/KeyValue.proto +++ b/app/src/main/protowire/KeyValue.proto @@ -103,4 +103,16 @@ message BackupDownloadNotifierState { uint64 lastSheetDisplaySeconds = 2; uint64 intervalSeconds = 3; Type type = 4; -} \ No newline at end of file +} + +message IssueNotifyTimes { + map lastNotifyTimeByName = 1; +} + +message StorageSyncLoopState { + repeated sint32 recentFingerprints = 1; + uint32 contentLevel = 2; + uint64 contentLevelAsOf = 3; + uint32 rateLevel = 4; + uint64 rateLevelAsOf = 5; +} diff --git a/app/src/test/java/org/thoughtcrime/securesms/jobs/StorageSyncJobTest.kt b/app/src/test/java/org/thoughtcrime/securesms/jobs/StorageSyncJobTest.kt index a6e99ef7a8..8b59c0c488 100644 --- a/app/src/test/java/org/thoughtcrime/securesms/jobs/StorageSyncJobTest.kt +++ b/app/src/test/java/org/thoughtcrime/securesms/jobs/StorageSyncJobTest.kt @@ -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 { + val priority = slot() + + 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]. diff --git a/app/src/test/java/org/thoughtcrime/securesms/storage/StorageSyncLoopDetectorTest.kt b/app/src/test/java/org/thoughtcrime/securesms/storage/StorageSyncLoopDetectorTest.kt new file mode 100644 index 0000000000..34f717f295 --- /dev/null +++ b/app/src/test/java/org/thoughtcrime/securesms/storage/StorageSyncLoopDetectorTest.kt @@ -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(), 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")) + ) + } +} diff --git a/app/src/test/java/org/thoughtcrime/securesms/testutil/FakeStorageServiceRule.kt b/app/src/test/java/org/thoughtcrime/securesms/testutil/FakeStorageServiceRule.kt index 6347cac3ba..3391c1cbfe 100644 --- a/app/src/test/java/org/thoughtcrime/securesms/testutil/FakeStorageServiceRule.kt +++ b/app/src/test/java/org/thoughtcrime/securesms/testutil/FakeStorageServiceRule.kt @@ -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 diff --git a/app/src/test/java/org/thoughtcrime/securesms/testutil/SignalStoreRule.kt b/app/src/test/java/org/thoughtcrime/securesms/testutil/SignalStoreRule.kt index 13ce3c65eb..e437d8e43e 100644 --- a/app/src/test/java/org/thoughtcrime/securesms/testutil/SignalStoreRule.kt +++ b/app/src/test/java/org/thoughtcrime/securesms/testutil/SignalStoreRule.kt @@ -22,7 +22,7 @@ class SignalStoreRule : ExternalResource() { override fun before() { val application = ApplicationProvider.getApplicationContext() - SignalStore.testInject(SignalStore(application, KeyValueStore(MockKeyValuePersistentStorage.withDataSet(KeyValueDataSet())))) + SignalStore.testInject(SignalStore(application, KeyValueStore(MockKeyValuePersistentStorage.withDataSet(KeyValueDataSet()), DirectExecutor()))) } override fun after() { diff --git a/app/src/test/java/org/thoughtcrime/securesms/util/LeakyBucketTest.kt b/app/src/test/java/org/thoughtcrime/securesms/util/LeakyBucketTest.kt new file mode 100644 index 0000000000..e57198d3df --- /dev/null +++ b/app/src/test/java/org/thoughtcrime/securesms/util/LeakyBucketTest.kt @@ -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) + } + } +}