diff --git a/lib/src/main/java/org/asamk/signal/manager/helper/StorageHelper.java b/lib/src/main/java/org/asamk/signal/manager/helper/StorageHelper.java index 93015188..a7aa7892 100644 --- a/lib/src/main/java/org/asamk/signal/manager/helper/StorageHelper.java +++ b/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()); diff --git a/lib/src/main/java/org/asamk/signal/manager/syncStorage/StorageSyncLoopDetector.java b/lib/src/main/java/org/asamk/signal/manager/syncStorage/StorageSyncLoopDetector.java new file mode 100644 index 00000000..6f3a1e3e --- /dev/null +++ b/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 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 {} + } +} diff --git a/lib/src/main/java/org/asamk/signal/manager/util/LeakyBucket.java b/lib/src/main/java/org/asamk/signal/manager/util/LeakyBucket.java new file mode 100644 index 00000000..8f8b7121 --- /dev/null +++ b/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); + } +} diff --git a/lib/src/test/java/org/asamk/signal/manager/syncStorage/StorageSyncLoopDetectorTest.java b/lib/src/test/java/org/asamk/signal/manager/syncStorage/StorageSyncLoopDetectorTest.java new file mode 100644 index 00000000..ed84aef8 --- /dev/null +++ b/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); + } +} diff --git a/lib/src/test/java/org/asamk/signal/manager/util/LeakyBucketTest.java b/lib/src/test/java/org/asamk/signal/manager/util/LeakyBucketTest.java new file mode 100644 index 00000000..e7ed5ef2 --- /dev/null +++ b/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; + } + } +}