Switch back to AcknowledgementMirroringMessageStream

service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletionMirroringMessageStream.java

@@ -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<MessageStreamEntry> getMessages() {
-    return redisDynamoDbMessageStream.getMessages();
-  }
-
-  @Override
-  public CompletableFuture<Void> 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);
-  }
-}

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<Envelope> getMessagesForDevice(final UUID destinationUuid, final Device destinationDevice) {

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<Void> delete(final AciServiceIdentifier aci, final byte deviceId, final UUID messageGuid) {
-    return delete(aci, deviceId, versionstampUUIDCipher.decryptVersionstamp(messageGuid, aci.uuid(), deviceId));
-  }
-
-  private CompletableFuture<Void> 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());

service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletionMirroringMessageStreamTest.java

@@ -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<Arguments> 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)
-    );
-  }
-}

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());