diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletionMirroringMessageStream.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletionMirroringMessageStream.java deleted file mode 100644 index 640ae5dc7..000000000 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletionMirroringMessageStream.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Copyright 2026 Signal Messenger, LLC - * SPDX-License-Identifier: AGPL-3.0-only - */ - -package org.whispersystems.textsecuregcm.storage; - -import java.util.UUID; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.Flow; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; -import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; -import org.whispersystems.textsecuregcm.storage.foundationdb.FoundationDbMessageStore; - -/// A temporary message stream that can mirror message acknowledgements (deletion requests) to FoundationDB -public class DeletionMirroringMessageStream implements MessageStream { - - private final RedisDynamoDbMessageStream redisDynamoDbMessageStream; - private final FoundationDbMessageStore foundationDbMessageStore; - private final ExperimentEnrollmentManager experimentEnrollmentManager; - - private final AciServiceIdentifier accountIdentifier; - private final byte deviceId; - - private static final Logger logger = LoggerFactory.getLogger(DeletionMirroringMessageStream.class); - - public DeletionMirroringMessageStream(final RedisDynamoDbMessageStream redisDynamoDbMessageStream, - final FoundationDbMessageStore foundationDbMessageStore, - final ExperimentEnrollmentManager experimentEnrollmentManager, - final UUID accountIdentifier, - final byte deviceId) { - - this.redisDynamoDbMessageStream = redisDynamoDbMessageStream; - this.foundationDbMessageStore = foundationDbMessageStore; - this.experimentEnrollmentManager = experimentEnrollmentManager; - - this.accountIdentifier = new AciServiceIdentifier(accountIdentifier); - this.deviceId = deviceId; - } - - @Override - public Flow.Publisher getMessages() { - return redisDynamoDbMessageStream.getMessages(); - } - - @Override - public CompletableFuture acknowledgeMessage(final UUID messageGuid, final long serverTimestamp) { - // All messages stored in FoundationDB use version 8 UUIDs; if a message has a version 4 UUID, then it only exists - // in Redis/DynamoDB - if (messageGuid.version() == 8 && - experimentEnrollmentManager.isEnrolled(accountIdentifier.uuid(), MessagesManager.MIRROR_DELETIONS_EXPERIMENT_NAME)) { - - foundationDbMessageStore.delete(accountIdentifier, deviceId, messageGuid) - .whenComplete((_, throwable) -> { - if (throwable != null) { - logger.warn("Failed to delete message {}/{}/{} from FoundationDb", accountIdentifier.uuid(), deviceId, - messageGuid, throwable); - } - }); - } - - return redisDynamoDbMessageStream.acknowledgeMessage(messageGuid, serverTimestamp); - } -} diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java index 735ac6703..fad154f43 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java @@ -283,12 +283,11 @@ public class MessagesManager { new RedisDynamoDbMessageStream(messagesDynamoDb, messagesCache, redisMessageAvailabilityManager, destinationUuid, destinationDevice); - return new DeletionMirroringMessageStream( + return experimentEnrollmentManager.isEnrolled(destinationUuid, MIRROR_DELETIONS_EXPERIMENT_NAME) + ? new AcknowledgementMirroringMessageStream( redisDynamoDbMessageStream, - foundationDbMessageStore, - experimentEnrollmentManager, - destinationUuid, - destinationDevice.getId()); + foundationDbMessageStore.getMessages(new AciServiceIdentifier(destinationUuid), destinationDevice.getId())) + : redisDynamoDbMessageStream; } Publisher getMessagesForDevice(final UUID destinationUuid, final Device destinationDevice) { diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStore.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStore.java index 9e1bf934c..790a9c710 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStore.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStore.java @@ -84,12 +84,6 @@ public class FoundationDbMessageStore { private static final Timer INSERT_MESSAGE_BATCH_TIMER = Metrics.timer(MetricsUtil.name(FoundationDbMessageStore.class, "insertMessageBatchTimer")); - private static final Counter DELETE_MESSAGE_COUNTER = - Metrics.counter(MetricsUtil.name(FoundationDbMessageStore.class, "deleteMessage")); - - private static final Timer DELETE_MESSAGE_TIMER = - Metrics.timer(MetricsUtil.name(FoundationDbMessageStore.class, "deleteMessageTimer")); - /// Result of inserting a message for a particular device /// /// @param versionstamp the versionstamp of the transaction in which this device's message was inserted, empty @@ -349,27 +343,6 @@ public class FoundationDbMessageStore { }); } - // Note that this method is intended only for initial migration support; in general, callers should clear messages - // by acknowledging messages via a `FoundationDbMessageStream`. - public CompletableFuture delete(final AciServiceIdentifier aci, final byte deviceId, final UUID messageGuid) { - return delete(aci, deviceId, versionstampUUIDCipher.decryptVersionstamp(messageGuid, aci.uuid(), deviceId)); - } - - private CompletableFuture delete(final AciServiceIdentifier aci, final byte deviceId, final Versionstamp versionstamp) { - final Timer.Sample sample = Timer.start(); - - final byte[] messageKey = getDeviceQueueSubspace(aci, deviceId).pack(Tuple.from(versionstamp)); - - return databasesByEpoch[getConfigurationEpoch(versionstamp)][getShardId(versionstamp)].runAsync(transaction -> { - transaction.clear(messageKey); - return CompletableFuture.completedFuture(null); - }) - .thenRun(() -> { - sample.stop(DELETE_MESSAGE_TIMER); - DELETE_MESSAGE_COUNTER.increment(); - }); - } - public void clearAll(final AciServiceIdentifier aci) { doForAllDatabasesWithMessages(aci, database -> database.run(transaction -> { transaction.clear(getAccountSubspace(aci).range()); diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletionMirroringMessageStreamTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletionMirroringMessageStreamTest.java deleted file mode 100644 index 0ef449fac..000000000 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletionMirroringMessageStreamTest.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * Copyright 2026 Signal Messenger, LLC - * SPDX-License-Identifier: AGPL-3.0-only - */ - -package org.whispersystems.textsecuregcm.storage; - -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -import java.util.List; -import java.util.UUID; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.Arguments; -import org.junit.jupiter.params.provider.MethodSource; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; -import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; -import org.whispersystems.textsecuregcm.storage.foundationdb.FoundationDbMessageStore; - -class DeletionMirroringMessageStreamTest { - - private FoundationDbMessageStore foundationDbMessageStore; - private ExperimentEnrollmentManager experimentEnrollmentManager; - - private DeletionMirroringMessageStream deletionMirroringMessageStream; - - private static final AciServiceIdentifier ACCOUNT_IDENTIFIER = new AciServiceIdentifier(UUID.randomUUID()); - private static final byte DEVICE_ID = Device.PRIMARY_ID; - - @BeforeEach - void setUp() { - foundationDbMessageStore = mock(FoundationDbMessageStore.class); - experimentEnrollmentManager = mock(ExperimentEnrollmentManager.class); - - deletionMirroringMessageStream = new DeletionMirroringMessageStream( - mock(RedisDynamoDbMessageStream.class), - foundationDbMessageStore, - experimentEnrollmentManager, - ACCOUNT_IDENTIFIER.uuid(), - DEVICE_ID); - } - - @ParameterizedTest - @MethodSource - void acknowledgeMessage(final boolean enrolled, final UUID messageGuid, final boolean expectFoundationDbDeletion) { - when(experimentEnrollmentManager.isEnrolled(any(UUID.class), eq(MessagesManager.MIRROR_DELETIONS_EXPERIMENT_NAME))) - .thenReturn(enrolled); - - deletionMirroringMessageStream.acknowledgeMessage(messageGuid, System.currentTimeMillis()); - - verify(foundationDbMessageStore, times(expectFoundationDbDeletion ? 1 : 0)) - .delete(ACCOUNT_IDENTIFIER, DEVICE_ID, messageGuid); - } - - private static List acknowledgeMessage() { - return List.of( - Arguments.argumentSet("Not enrolled, v4 UUID", false, UUID.randomUUID(), false), - Arguments.argumentSet("Not enrolled, v8 UUID", false, MessageGuidUtil.generateRandomV8UUID(), false), - Arguments.argumentSet("Enrolled, v4 UUID", true, UUID.randomUUID(), false), - Arguments.argumentSet("Enrolled, v8 UUID", true, MessageGuidUtil.generateRandomV8UUID(), true) - ); - } -} diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStoreTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStoreTest.java index c8e71794d..2cf3d54a4 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStoreTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStoreTest.java @@ -493,36 +493,6 @@ class FoundationDbMessageStoreTest { Map.of(generateRandomAciForShard(0), Collections.emptyMap()))); } - @Test - void delete() { - final AciServiceIdentifier deletedMessageAci = new AciServiceIdentifier(UUID.randomUUID()); - final byte deletedMessageDeviceId = Device.PRIMARY_ID; - final MessageProtos.Envelope deletedMessage = generateRandomMessage(false); - - final AciServiceIdentifier retainedMessageAci = new AciServiceIdentifier(UUID.randomUUID()); - final byte retainedMessageDeviceId = Device.PRIMARY_ID; - final MessageProtos.Envelope retainedMessage = generateRandomMessage(false); - - final UUID deletedMessageGuid = - foundationDbMessageStore.insert(deletedMessageAci, Map.of(deletedMessageDeviceId, deletedMessage)).join() - .get(deletedMessageDeviceId).messageGuid().orElseThrow(); - - foundationDbMessageStore.insert(retainedMessageAci, Map.of(retainedMessageDeviceId, retainedMessage)).join(); - - assertArrayEquals(deletedMessage.toByteArray(), - getItemsInDeviceQueue(deletedMessageAci, deletedMessageDeviceId).getFirst().getValue()); - - assertArrayEquals(retainedMessage.toByteArray(), - getItemsInDeviceQueue(retainedMessageAci, retainedMessageDeviceId).getFirst().getValue()); - - foundationDbMessageStore.delete(deletedMessageAci, deletedMessageDeviceId, deletedMessageGuid).join(); - - assertTrue(getItemsInDeviceQueue(deletedMessageAci, deletedMessageDeviceId).isEmpty()); - - assertArrayEquals(retainedMessage.toByteArray(), - getItemsInDeviceQueue(retainedMessageAci, retainedMessageDeviceId).getFirst().getValue()); - } - @Test void clearAllForAccount() { final AciServiceIdentifier deletedAccountIdentifier = new AciServiceIdentifier(UUID.randomUUID());