Implement storage sync loop detector

lib/src/main/java/org/asamk/signal/manager/helper/StorageHelper.java

@@ -14,6 +14,7 @@ import org.asamk.signal.manager.syncStorage.ContactRecordProcessor;
 import org.asamk.signal.manager.syncStorage.GroupV1RecordProcessor;
 import org.asamk.signal.manager.syncStorage.GroupV2RecordProcessor;
 import org.asamk.signal.manager.syncStorage.StickerPackRecordProcessor;
+import org.asamk.signal.manager.syncStorage.StorageSyncLoopDetector;
 import org.asamk.signal.manager.syncStorage.StorageSyncModels;
 import org.asamk.signal.manager.syncStorage.StorageSyncValidations;
 import org.asamk.signal.manager.syncStorage.WriteOperationResult;
@@ -61,11 +62,13 @@ public class StorageHelper {
     private final SignalAccount account;
     private final SignalDependencies dependencies;
     private final Context context;
+    private final StorageSyncLoopDetector storageSyncLoopDetector;
 
     public StorageHelper(final Context context) {
         this.account = context.getAccount();
         this.dependencies = context.getDependencies();
         this.context = context;
+        this.storageSyncLoopDetector = new StorageSyncLoopDetector(account::isMultiDevice);
     }
 
     public void syncDataWithStorage() throws IOException {
@@ -84,6 +87,7 @@ public class StorageHelper {
         final var storageServiceRepository = dependencies.getStorageServiceRepository();
         final var result = storageServiceRepository.getStorageManifestIfDifferentVersion(storageKey,
                 localManifestVersion);
+        final var fetchedRemoteManifest = result instanceof ManifestIfDifferentVersionResult.DifferentVersion;
 
         var needsForcePush = false;
         final var remoteManifest = switch (result) {
@@ -146,7 +150,10 @@ public class StorageHelper {
             needsForcePush = true;
         } else {
             try {
-                needsMultiDeviceSync = writeToStorage(storageKey, remoteManifest, needsForcePush);
+                needsMultiDeviceSync = writeToStorage(storageKey,
+                        remoteManifest,
+                        needsForcePush,
+                        fetchedRemoteManifest);
             } catch (RetryLaterException e) {
                 // TODO retry later
                 return;
@@ -288,7 +295,8 @@ public class StorageHelper {
     private boolean writeToStorage(
             final StorageKey storageKey,
             final SignalStorageManifest remoteManifest,
-            final boolean needsForcePush
+            final boolean needsForcePush,
+            final boolean fetchedRemoteManifest
     ) throws IOException, RetryLaterException {
         final WriteOperationResult remoteWriteOperation;
         try (final var connection = account.getAccountDatabase().getConnection()) {
@@ -326,6 +334,19 @@ public class StorageHelper {
 
         if (remoteWriteOperation.isEmpty()) {
             logger.debug("No remote writes needed. Still at version: {}", remoteManifest.version);
+            storageSyncLoopDetector.onConverged();
+            return false;
+        }
+
+        final var loopCheck = storageSyncLoopDetector.onWriteAttempt(remoteWriteOperation,
+                fetchedRemoteManifest,
+                false);
+        if (loopCheck instanceof StorageSyncLoopDetector.Decision.Denied denied) {
+            logger.warn(
+                    "Skipping remote write, another device is likely undoing it. Cause: {}, level: {}. WriteOperationResult :: {}",
+                    denied.cause(),
+                    denied.level(),
+                    remoteWriteOperation);
             return false;
         }
 
@@ -344,11 +365,18 @@ public class StorageHelper {
                         remoteWriteOperation.deletes());
         switch (result) {
             case WriteStorageRecordsResult.ConflictError ignored -> {
+                storageSyncLoopDetector.onWriteFailed();
                 logger.debug("Hit a conflict when trying to resolve the conflict! Retrying.");
                 throw new RetryLaterException();
             }
-            case WriteStorageRecordsResult.NetworkError networkError -> throw networkError.getException();
-            case WriteStorageRecordsResult.StatusCodeError statusCodeError -> throw statusCodeError.getException();
+            case WriteStorageRecordsResult.NetworkError networkError -> {
+                storageSyncLoopDetector.onWriteFailed();
+                throw networkError.getException();
+            }
+            case WriteStorageRecordsResult.StatusCodeError statusCodeError -> {
+                storageSyncLoopDetector.onWriteFailed();
+                throw statusCodeError.getException();
+            }
             case WriteStorageRecordsResult.Success ignored -> {
                 logger.debug("Saved new manifest. Now at version: {}", remoteWriteOperation.manifest().version);
                 storeManifestLocally(remoteWriteOperation.manifest());

lib/src/main/java/org/asamk/signal/manager/syncStorage/StorageSyncLoopDetector.java

@@ -0,0 +1,139 @@
+package org.asamk.signal.manager.syncStorage;
+
+import org.asamk.signal.manager.util.LeakyBucket;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.BooleanSupplier;
+
+public final class StorageSyncLoopDetector {
+
+    private static final int FINGERPRINT_HISTORY = 3;
+
+    private final BooleanSupplier isMultiDevice;
+    private final List<Integer> recentFingerprints = new ArrayList<>();
+    private final LeakyBucket contentBucket = new LeakyBucket(3,
+            Duration.ofHours(1).toMillis(),
+            new InMemoryBucketState());
+    private final LeakyBucket rateBucket = new LeakyBucket(100,
+            Duration.ofMinutes(10).toMillis(),
+            new InMemoryBucketState());
+
+    public StorageSyncLoopDetector(final BooleanSupplier isMultiDevice) {
+        this.isMultiDevice = isMultiDevice;
+    }
+
+    public synchronized Decision onWriteAttempt(
+            final WriteOperationResult write,
+            final boolean fetchedRemoteManifest,
+            final boolean isRetry
+    ) {
+        return onWriteAttempt(write, fetchedRemoteManifest, isRetry, System.currentTimeMillis());
+    }
+
+    synchronized Decision onWriteAttempt(
+            final WriteOperationResult write,
+            final boolean fetchedRemoteManifest,
+            final boolean isRetry,
+            final long now
+    ) {
+        if (!isMultiDevice.getAsBoolean() || isRetry) {
+            return Decision.Allowed.INSTANCE;
+        }
+
+        final var fingerprint = fingerprint(write);
+        final var chargeContent = fetchedRemoteManifest && fingerprint != null && recentFingerprints.contains(
+                fingerprint);
+
+        if (chargeContent && !contentBucket.hasRoom(now)) {
+            return new Decision.Denied(Cause.REPEATED_PAYLOAD, contentBucket.level(now));
+        }
+        if (fetchedRemoteManifest && !rateBucket.hasRoom(now)) {
+            return new 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.INSTANCE;
+    }
+
+    public synchronized void onWriteFailed() {
+        onWriteFailed(System.currentTimeMillis());
+    }
+
+    synchronized void onWriteFailed(final long now) {
+        contentBucket.refund(now);
+        rateBucket.refund(now);
+    }
+
+    public synchronized void onConverged() {
+        contentBucket.clear();
+    }
+
+    private void remember(final int fingerprint) {
+        recentFingerprints.remove(Integer.valueOf(fingerprint));
+        recentFingerprints.addFirst(fingerprint);
+        if (recentFingerprints.size() > FINGERPRINT_HISTORY) {
+            recentFingerprints.removeLast();
+        }
+    }
+
+    private static Integer fingerprint(final WriteOperationResult write) {
+        if (write.inserts().isEmpty()) {
+            return null;
+        }
+
+        return write.inserts()
+                .stream()
+                .map(record -> Arrays.hashCode(record.getProto().encode()))
+                .sorted()
+                .toList()
+                .hashCode();
+    }
+
+    private static final class InMemoryBucketState implements LeakyBucket.State {
+
+        private int level;
+        private long levelUpdatedAt;
+
+        @Override
+        public int level() {
+            return level;
+        }
+
+        @Override
+        public long levelUpdatedAt() {
+            return levelUpdatedAt;
+        }
+
+        @Override
+        public void update(final int level, final long levelUpdatedAt) {
+            this.level = level;
+            this.levelUpdatedAt = levelUpdatedAt;
+        }
+    }
+
+    public enum Cause {
+        REPEATED_PAYLOAD,
+        WRITE_RATE
+    }
+
+    public sealed interface Decision {
+
+        enum Allowed implements Decision {
+            INSTANCE
+        }
+
+        record Denied(Cause cause, int level) implements Decision {}
+    }
+}

lib/src/main/java/org/asamk/signal/manager/util/LeakyBucket.java

@@ -0,0 +1,60 @@
+package org.asamk.signal.manager.util;
+
+public final class LeakyBucket {
+
+    private final int capacity;
+    private final long dripIntervalMillis;
+    private final State state;
+
+    public LeakyBucket(final int capacity, final long dripIntervalMillis, final State state) {
+        this.capacity = capacity;
+        this.dripIntervalMillis = dripIntervalMillis;
+        this.state = state;
+    }
+
+    public int level(final long now) {
+        return calculateStateForCurrentTime(now).level();
+    }
+
+    public boolean hasRoom(final long now) {
+        return level(now) < capacity;
+    }
+
+    public void use(final long now) {
+        final var currentState = calculateStateForCurrentTime(now);
+        state.update(currentState.level() + 1, currentState.levelUpdatedAt());
+    }
+
+    public void refund(final long now) {
+        final var currentState = calculateStateForCurrentTime(now);
+        state.update(Math.max(currentState.level() - 1, 0), currentState.levelUpdatedAt());
+    }
+
+    public void clear() {
+        state.update(0, 0);
+    }
+
+    private Snapshot calculateStateForCurrentTime(final long now) {
+        final var level = state.level();
+        final var levelUpdatedAt = state.levelUpdatedAt();
+        final var elapsed = now - levelUpdatedAt;
+
+        if (level <= 0 || elapsed < 0) {
+            return new Snapshot(0, now);
+        }
+
+        final var drips = elapsed / dripIntervalMillis;
+        return new Snapshot((int) Math.max(level - drips, 0), levelUpdatedAt + dripIntervalMillis * drips);
+    }
+
+    private record Snapshot(int level, long levelUpdatedAt) {}
+
+    public interface State {
+
+        int level();
+
+        long levelUpdatedAt();
+
+        void update(int level, long levelUpdatedAt);
+    }
+}

lib/src/test/java/org/asamk/signal/manager/syncStorage/StorageSyncLoopDetectorTest.java

@@ -0,0 +1,102 @@
+package org.asamk.signal.manager.syncStorage;
+
+import org.junit.jupiter.api.Test;
+import org.whispersystems.signalservice.api.storage.SignalStorageRecord;
+import org.whispersystems.signalservice.api.storage.StorageId;
+import org.whispersystems.signalservice.internal.storage.protos.StorageRecord;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+class StorageSyncLoopDetectorTest {
+
+    private static final long NOW = 1_700_000_000_000L;
+
+    @Test
+    void repeatedPayloadIsDeniedAfterThreeCharges() {
+        final var detector = new StorageSyncLoopDetector(() -> true);
+        final var write = writeWithInsert(1);
+
+        assertAllowed(detector.onWriteAttempt(write, true, false, NOW));
+        for (var index = 0; index < 3; index++) {
+            assertAllowed(detector.onWriteAttempt(write, true, false, NOW));
+        }
+
+        assertEquals(new StorageSyncLoopDetector.Decision.Denied(StorageSyncLoopDetector.Cause.REPEATED_PAYLOAD, 3),
+                detector.onWriteAttempt(write, true, false, NOW));
+    }
+
+    @Test
+    void storageIdsAreExcludedFromPayloadFingerprint() {
+        final var detector = new StorageSyncLoopDetector(() -> true);
+
+        assertAllowed(detector.onWriteAttempt(writeWithInsert(1), true, false, NOW));
+        assertAllowed(detector.onWriteAttempt(writeWithInsert(2), true, false, NOW));
+        assertAllowed(detector.onWriteAttempt(writeWithInsert(3), true, false, NOW));
+        assertAllowed(detector.onWriteAttempt(writeWithInsert(4), true, false, NOW));
+
+        assertInstanceOf(StorageSyncLoopDetector.Decision.Denied.class,
+                detector.onWriteAttempt(writeWithInsert(5), true, false, NOW));
+    }
+
+    @Test
+    void convergenceClearsTheContentBucket() {
+        final var detector = new StorageSyncLoopDetector(() -> true);
+        final var write = writeWithInsert(1);
+        for (var index = 0; index < 4; index++) {
+            assertAllowed(detector.onWriteAttempt(write, true, false, NOW));
+        }
+
+        detector.onConverged();
+
+        assertAllowed(detector.onWriteAttempt(write, true, false, NOW));
+    }
+
+    @Test
+    void failedWritesAreRefunded() {
+        final var detector = new StorageSyncLoopDetector(() -> true);
+        final var write = writeWithInsert(1);
+
+        for (var index = 0; index < 20; index++) {
+            assertAllowed(detector.onWriteAttempt(write, true, false, NOW));
+            detector.onWriteFailed(NOW);
+        }
+    }
+
+    @Test
+    void rateBucketLimitsDeleteOnlyWrites() {
+        final var detector = new StorageSyncLoopDetector(() -> true);
+        final var write = new WriteOperationResult(null, List.of(), List.of(new byte[]{1}));
+
+        for (var index = 0; index < 100; index++) {
+            assertAllowed(detector.onWriteAttempt(write, true, false, NOW));
+        }
+
+        assertEquals(new StorageSyncLoopDetector.Decision.Denied(StorageSyncLoopDetector.Cause.WRITE_RATE, 100),
+                detector.onWriteAttempt(write, true, false, NOW));
+    }
+
+    @Test
+    void retriesAndSingleDeviceWritesAreExempt() {
+        final var retryDetector = new StorageSyncLoopDetector(() -> true);
+        final var singleDeviceDetector = new StorageSyncLoopDetector(() -> false);
+        final var write = writeWithInsert(1);
+
+        for (var index = 0; index < 20; index++) {
+            assertAllowed(retryDetector.onWriteAttempt(write, true, true, NOW));
+            assertAllowed(singleDeviceDetector.onWriteAttempt(write, true, false, NOW));
+        }
+    }
+
+    private static WriteOperationResult writeWithInsert(final int storageIdByte) {
+        final var storageId = new StorageId(99, new byte[]{(byte) storageIdByte});
+        final var record = new SignalStorageRecord(storageId, new StorageRecord.Builder().build());
+        return new WriteOperationResult(null, List.of(record), List.of());
+    }
+
+    private static void assertAllowed(final StorageSyncLoopDetector.Decision decision) {
+        assertEquals(StorageSyncLoopDetector.Decision.Allowed.INSTANCE, decision);
+    }
+}

lib/src/test/java/org/asamk/signal/manager/util/LeakyBucketTest.java

@@ -0,0 +1,66 @@
+package org.asamk.signal.manager.util;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class LeakyBucketTest {
+
+    private static final long NOW = 1_700_000_000_000L;
+
+    @Test
+    void dripsLevelsAndCarriesPartialIntervals() {
+        final var state = new TestState();
+        final var bucket = new LeakyBucket(3, Duration.ofHours(1).toMillis(), state);
+
+        bucket.use(NOW);
+        bucket.use(NOW);
+        bucket.use(NOW);
+
+        assertEquals(2, bucket.level(NOW + Duration.ofMinutes(90).toMillis()));
+        assertEquals(1, bucket.level(NOW + Duration.ofMinutes(121).toMillis()));
+    }
+
+    @Test
+    void refundNeverDropsBelowZero() {
+        final var bucket = new LeakyBucket(1, 1_000, new TestState());
+
+        bucket.refund(NOW);
+
+        assertEquals(0, bucket.level(NOW));
+        assertTrue(bucket.hasRoom(NOW));
+    }
+
+    @Test
+    void clockMovingBackwardsRefillsTheBucket() {
+        final var bucket = new LeakyBucket(1, 1_000, new TestState());
+        bucket.use(NOW);
+
+        assertTrue(bucket.hasRoom(NOW - 1));
+    }
+
+    private static final class TestState implements LeakyBucket.State {
+
+        private int level;
+        private long levelUpdatedAt;
+
+        @Override
+        public int level() {
+            return level;
+        }
+
+        @Override
+        public long levelUpdatedAt() {
+            return levelUpdatedAt;
+        }
+
+        @Override
+        public void update(final int level, final long levelUpdatedAt) {
+            this.level = level;
+            this.levelUpdatedAt = levelUpdatedAt;
+        }
+    }
+}