Retire the legacy message availability system

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -50,7 +50,6 @@ import java.util.List;
 import java.util.Optional;
 import java.util.ServiceLoader;
 import java.util.Set;
-import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -459,9 +458,6 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     FaultTolerantRedisClient pubsubClient =
         config.getRedisPubSubConfiguration().build("pubsub", sharedClientResources);
 
-    final BlockingQueue<Runnable> keyspaceNotificationDispatchQueue = new ArrayBlockingQueue<>(100_000);
-    Metrics.gaugeCollectionSize(name(getClass(), "keyspaceNotificationDispatchQueueSize"), Collections.emptyList(),
-        keyspaceNotificationDispatchQueue);
     final BlockingQueue<Runnable> receiptSenderQueue = new LinkedBlockingQueue<>();
     Metrics.gaugeCollectionSize(name(getClass(), "receiptSenderQueue"), Collections.emptyList(), receiptSenderQueue);
     final BlockingQueue<Runnable> fcmSenderQueue = new LinkedBlockingQueue<>();
@@ -474,14 +470,6 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         .scheduledExecutorService(name(getClass(), "recurringJob-%d")).threads(6).build();
     ScheduledExecutorService websocketScheduledExecutor = environment.lifecycle()
         .scheduledExecutorService(name(getClass(), "websocket-%d")).threads(8).build();
-    ExecutorService keyspaceNotificationDispatchExecutor = ExecutorServiceMetrics.monitor(Metrics.globalRegistry,
-        environment.lifecycle()
-            .executorService(name(getClass(), "keyspaceNotification-%d"))
-            .maxThreads(16)
-            .workQueue(keyspaceNotificationDispatchQueue)
-            .build(),
-        MetricsUtil.name(getClass(), "keyspaceNotificationExecutor"),
-        MetricsUtil.PREFIX);
     ExecutorService apnSenderExecutor = environment.lifecycle().executorService(name(getClass(), "apnSender-%d"))
         .maxThreads(1).minThreads(1).build();
     ExecutorService fcmSenderExecutor = environment.lifecycle().executorService(name(getClass(), "fcmSender-%d"))
@@ -611,8 +599,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         storageServiceExecutor, storageServiceRetryExecutor, config.getSecureStorageServiceConfiguration());
     PubSubClientEventManager pubSubClientEventManager = new PubSubClientEventManager(messagesCluster, clientEventExecutor);
     ProfilesManager profilesManager = new ProfilesManager(profiles, cacheCluster);
-    MessagesCache messagesCache = new MessagesCache(messagesCluster, keyspaceNotificationDispatchExecutor,
-        messageDeliveryScheduler, messageDeletionAsyncExecutor, clock, dynamicConfigurationManager);
+    MessagesCache messagesCache = new MessagesCache(messagesCluster, messageDeliveryScheduler,
+        messageDeletionAsyncExecutor, clock, dynamicConfigurationManager);
     ClientReleaseManager clientReleaseManager = new ClientReleaseManager(clientReleases,
         recurringJobExecutor,
         config.getClientReleaseConfiguration().refreshInterval(),
@@ -733,7 +721,6 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     environment.lifecycle().manage(apnSender);
     environment.lifecycle().manage(pushNotificationScheduler);
     environment.lifecycle().manage(provisioningManager);
-    environment.lifecycle().manage(messagesCache);
     environment.lifecycle().manage(pubSubClientEventManager);
     environment.lifecycle().manage(currencyManager);
     environment.lifecycle().manage(registrationServiceClient);

service/src/main/java/org/whispersystems/textsecuregcm/push/ClientEventListener.java

@@ -19,7 +19,7 @@ public interface ClientEventListener {
   /**
    * Indicates that messages for the client have been persisted from short-term storage to long-term storage.
    */
-  void handleMessagesPersistedPubSub();
+  void handleMessagesPersisted();
 
   /**
    * Indicates that the client's presence has been displaced and the listener should close the client's underlying

service/src/main/java/org/whispersystems/textsecuregcm/push/PubSubClientEventManager.java

@@ -371,7 +371,7 @@ public class PubSubClientEventManager extends RedisClusterPubSubAdapter<byte[],
 
         case DISCONNECT_REQUESTED -> listenerEventExecutor.execute(() -> listener.handleConnectionDisplaced(false));
 
-        case MESSAGES_PERSISTED -> listenerEventExecutor.execute(listener::handleMessagesPersistedPubSub);
+        case MESSAGES_PERSISTED -> listenerEventExecutor.execute(listener::handleMessagesPersisted);
 
         default -> logger.warn("Unexpected client event type: {}", clientEvent.getClass());
       }

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

@@ -1,26 +0,0 @@
-/*
- * Copyright 2013-2021 Signal Messenger, LLC
- * SPDX-License-Identifier: AGPL-3.0-only
- */
-
-package org.whispersystems.textsecuregcm.storage;
-
-/**
- * A message availability listener is notified when new messages are available for a specific device for a specific
- * account. Availability listeners are also notified when messages are moved from the message cache to long-term storage
- * as an optimization hint to implementing classes.
- */
-public interface MessageAvailabilityListener {
-
-  /**
-   * @return whether the listener is still active. {@code false} indicates the listener can no longer handle messages
-   * and may be discarded
-   */
-  boolean handleNewMessagesAvailable();
-
-  /**
-   * @return whether the listener is still active. {@code false} indicates the listener can no longer handle messages
-   * and may be discarded
-   */
-  boolean handleMessagesPersisted();
-}

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

@@ -10,12 +10,8 @@ import static com.codahale.metrics.MetricRegistry.name;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.InvalidProtocolBufferException;
-import io.dropwizard.lifecycle.Managed;
 import io.lettuce.core.ZAddArgs;
 import io.lettuce.core.cluster.SlotHash;
-import io.lettuce.core.cluster.event.ClusterTopologyChangedEvent;
-import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
-import io.lettuce.core.cluster.pubsub.RedisClusterPubSubAdapter;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.Timer;
@@ -28,18 +24,13 @@ import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
-import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Predicate;
-import javax.annotation.Nullable;
 import org.reactivestreams.Publisher;
 import org.signal.libsignal.protocol.SealedSenderMultiRecipientMessage;
 import org.signal.libsignal.protocol.ServiceId;
@@ -51,11 +42,9 @@ import org.whispersystems.textsecuregcm.entities.MessageProtos;
 import org.whispersystems.textsecuregcm.experiment.Experiment;
 import org.whispersystems.textsecuregcm.identity.ServiceIdentifier;
 import org.whispersystems.textsecuregcm.metrics.MetricsUtil;
-import org.whispersystems.textsecuregcm.redis.FaultTolerantPubSubClusterConnection;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisClusterClient;
 import org.whispersystems.textsecuregcm.util.Pair;
 import org.whispersystems.textsecuregcm.util.RedisClusterUtil;
-import org.whispersystems.textsecuregcm.util.Util;
 import reactor.core.observability.micrometer.Micrometer;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -118,13 +107,11 @@ import reactor.core.scheduler.Schedulers;
  * @see MessagesCacheRemoveRecipientViewFromMrmDataScript
  * @see MessagesCacheRemoveQueueScript
  */
-public class MessagesCache extends RedisClusterPubSubAdapter<String, String> implements Managed {
+public class MessagesCache {
 
   private final FaultTolerantRedisClusterClient redisCluster;
-  private final FaultTolerantPubSubClusterConnection<String, String> pubSubConnection;
   private final Clock clock;
 
-  private final ExecutorService notificationExecutorService;
   private final Scheduler messageDeliveryScheduler;
   private final ExecutorService messageDeletionExecutorService;
   // messageDeletionExecutorService wrapped into a reactor Scheduler
@@ -140,10 +127,6 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
   private final MessagesCacheGetQueuesToPersistScript getQueuesToPersistScript;
   private final MessagesCacheRemoveRecipientViewFromMrmDataScript removeRecipientViewFromMrmDataScript;
 
-  private final ReentrantLock messageListenersLock = new ReentrantLock();
-  private final Map<String, MessageAvailabilityListener> messageListenersByQueueName = new HashMap<>();
-  private final Map<MessageAvailabilityListener, String> queueNamesByMessageListener = new IdentityHashMap<>();
-
   private final Timer insertTimer = Metrics.timer(name(MessagesCache.class, "insert"));
   private final Timer insertSharedMrmPayloadTimer = Metrics.timer(name(MessagesCache.class, "insertSharedMrmPayload"));
   private final Timer getMessagesTimer = Metrics.timer(name(MessagesCache.class, "get"));
@@ -151,17 +134,8 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
   private final Timer removeByGuidTimer = Metrics.timer(name(MessagesCache.class, "removeByGuid"));
   private final Timer removeRecipientViewTimer = Metrics.timer(name(MessagesCache.class, "removeRecipientView"));
   private final Timer clearQueueTimer = Metrics.timer(name(MessagesCache.class, "clear"));
-  private final Counter pubSubMessageCounter = Metrics.counter(name(MessagesCache.class, "pubSubMessage"));
-  private final Counter newMessageNotificationCounter = Metrics.counter(
-      name(MessagesCache.class, "newMessageNotification"));
-  private final Counter queuePersistedNotificationCounter = Metrics.counter(
-      name(MessagesCache.class, "queuePersisted"));
   private final Counter staleEphemeralMessagesCounter = Metrics.counter(
       name(MessagesCache.class, "staleEphemeralMessages"));
-  private final Counter messageAvailabilityListenerRemovedAfterAddCounter = Metrics.counter(
-      name(MessagesCache.class, "messageAvailabilityListenerRemovedAfterAdd"));
-  private final Counter prunedStaleSubscriptionCounter = Metrics.counter(
-      name(MessagesCache.class, "prunedStaleSubscription"));
   private final Counter mrmContentRetrievedCounter = Metrics.counter(name(MessagesCache.class, "mrmViewRetrieved"));
   private final Counter sharedMrmDataKeyRemovedCounter = Metrics.counter(
       name(MessagesCache.class, "sharedMrmKeyRemoved"));
@@ -169,9 +143,6 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
   static final String NEXT_SLOT_TO_PERSIST_KEY = "user_queue_persist_slot";
   private static final byte[] LOCK_VALUE = "1".getBytes(StandardCharsets.UTF_8);
 
-  private static final String QUEUE_KEYSPACE_PREFIX = "__keyspace@0__:user_queue::";
-  private static final String PERSISTING_KEYSPACE_PREFIX = "__keyspace@0__:user_queue_persisting::";
-
   private static final String MRM_VIEWS_EXPERIMENT_NAME = "mrmViews";
 
   @VisibleForTesting
@@ -184,13 +155,15 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
 
   private static final Logger logger = LoggerFactory.getLogger(MessagesCache.class);
 
-  public MessagesCache(final FaultTolerantRedisClusterClient redisCluster, final ExecutorService notificationExecutorService,
-                       final Scheduler messageDeliveryScheduler, final ExecutorService messageDeletionExecutorService, final Clock clock,
-                       final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager)
+  public MessagesCache(final FaultTolerantRedisClusterClient redisCluster,
+      final Scheduler messageDeliveryScheduler,
+      final ExecutorService messageDeletionExecutorService,
+      final Clock clock,
+      final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager)
       throws IOException {
+
     this(
         redisCluster,
-        notificationExecutorService,
         messageDeliveryScheduler,
         messageDeletionExecutorService,
         clock,
@@ -206,8 +179,9 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
   }
 
   @VisibleForTesting
-  MessagesCache(final FaultTolerantRedisClusterClient redisCluster, final ExecutorService notificationExecutorService,
-                final Scheduler messageDeliveryScheduler, final ExecutorService messageDeletionExecutorService, final Clock clock,
+  MessagesCache(final FaultTolerantRedisClusterClient redisCluster,
+                final Scheduler messageDeliveryScheduler,
+                final ExecutorService messageDeletionExecutorService, final Clock clock,
                 final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager,
                 final MessagesCacheInsertScript insertScript,
                 final MessagesCacheInsertSharedMultiRecipientPayloadAndViewsScript insertMrmScript,
@@ -218,10 +192,8 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
       throws IOException {
 
     this.redisCluster = redisCluster;
-    this.pubSubConnection = redisCluster.createPubSubConnection();
     this.clock = clock;
 
-    this.notificationExecutorService = notificationExecutorService;
     this.messageDeliveryScheduler = messageDeliveryScheduler;
     this.messageDeletionExecutorService = messageDeletionExecutorService;
     this.messageDeletionScheduler = Schedulers.fromExecutorService(messageDeletionExecutorService, "messageDeletion");
@@ -237,34 +209,6 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
     this.removeRecipientViewFromMrmDataScript = removeRecipientViewFromMrmDataScript;
   }
 
-  @Override
-  public void start() {
-    pubSubConnection.usePubSubConnection(connection -> connection.addListener(this));
-    pubSubConnection.subscribeToClusterTopologyChangedEvents(this::resubscribeAll);
-  }
-
-  @Override
-  public void stop() {
-    pubSubConnection.usePubSubConnection(connection -> connection.sync().upstream().commands().unsubscribe());
-  }
-
-  private void resubscribeAll(final ClusterTopologyChangedEvent event) {
-
-    final Set<String> queueNames;
-
-    messageListenersLock.lock();
-    try {
-      queueNames = new HashSet<>(messageListenersByQueueName.keySet());
-    } finally {
-      messageListenersLock.unlock();
-    }
-
-    for (final String queueName : queueNames) {
-      // avoid overwhelming a newly recovered node by processing synchronously, rather than using CompletableFuture.allOf()
-      subscribeForKeyspaceNotifications(queueName).join();
-    }
-  }
-
   public long insert(final UUID guid, final UUID destinationUuid, final byte destinationDevice,
       final MessageProtos.Envelope message) {
     final MessageProtos.Envelope messageWithGuid = message.toBuilder().setServerGuid(guid.toString()).build();
@@ -642,146 +586,6 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
         connection -> connection.sync().del(getPersistInProgressKey(accountUuid, deviceId)));
   }
 
-  public void addMessageAvailabilityListener(final UUID destinationUuid, final byte deviceId,
-      final MessageAvailabilityListener listener) {
-    final String queueName = getQueueName(destinationUuid, deviceId);
-
-    final CompletableFuture<Void> subscribeFuture;
-    messageListenersLock.lock();
-    try {
-      messageListenersByQueueName.put(queueName, listener);
-      queueNamesByMessageListener.put(listener, queueName);
-      // Submit to the Redis queue while holding the lock, but don’t wait until exiting
-      subscribeFuture = subscribeForKeyspaceNotifications(queueName);
-    } finally {
-      messageListenersLock.unlock();
-    }
-
-    subscribeFuture.join();
-  }
-
-  public void removeMessageAvailabilityListener(final MessageAvailabilityListener listener) {
-    @Nullable final String queueName;
-    messageListenersLock.lock();
-    try {
-      queueName = queueNamesByMessageListener.get(listener);
-    } finally {
-      messageListenersLock.unlock();
-    }
-
-    if (queueName != null) {
-
-      final CompletableFuture<Void> unsubscribeFuture;
-      messageListenersLock.lock();
-      try {
-        queueNamesByMessageListener.remove(listener);
-        if (messageListenersByQueueName.remove(queueName, listener)) {
-          // Submit to the Redis queue holding the lock, but don’t wait until exiting
-          unsubscribeFuture = unsubscribeFromKeyspaceNotifications(queueName);
-        } else {
-          messageAvailabilityListenerRemovedAfterAddCounter.increment();
-          unsubscribeFuture = CompletableFuture.completedFuture(null);
-        }
-      } finally {
-        messageListenersLock.unlock();
-      }
-
-      unsubscribeFuture.join();
-    }
-  }
-
-  private void pruneStaleSubscription(final String channel) {
-    unsubscribeFromKeyspaceNotifications(getQueueNameFromKeyspaceChannel(channel))
-        .thenRun(prunedStaleSubscriptionCounter::increment);
-  }
-
-  private CompletableFuture<Void> subscribeForKeyspaceNotifications(final String queueName) {
-    final int slot = SlotHash.getSlot(queueName);
-
-    return pubSubConnection.withPubSubConnection(
-            connection -> connection.async()
-                .nodes(node -> node.is(RedisClusterNode.NodeFlag.UPSTREAM) && node.hasSlot(slot))
-            .commands()
-                .subscribe(getKeyspaceChannels(queueName))).toCompletableFuture()
-        .thenRun(Util.NOOP);
-  }
-
-  private CompletableFuture<Void> unsubscribeFromKeyspaceNotifications(final String queueName) {
-    final int slot = SlotHash.getSlot(queueName);
-
-    return pubSubConnection.withPubSubConnection(
-            connection -> connection.async()
-                .nodes(node -> node.is(RedisClusterNode.NodeFlag.UPSTREAM) && node.hasSlot(slot))
-            .commands()
-                .unsubscribe(getKeyspaceChannels(queueName)))
-        .toCompletableFuture()
-        .thenRun(Util.NOOP);
-  }
-
-  private static String[] getKeyspaceChannels(final String queueName) {
-    return new String[]{
-        QUEUE_KEYSPACE_PREFIX + "{" + queueName + "}",
-        PERSISTING_KEYSPACE_PREFIX + "{" + queueName + "}"
-    };
-  }
-
-  @Override
-  public void message(final RedisClusterNode node, final String channel, final String message) {
-    pubSubMessageCounter.increment();
-
-    if (channel.startsWith(QUEUE_KEYSPACE_PREFIX) && "zadd".equals(message)) {
-      newMessageNotificationCounter.increment();
-      notificationExecutorService.execute(() -> {
-        try {
-          findListener(channel).ifPresentOrElse(listener -> {
-            if (!listener.handleNewMessagesAvailable()) {
-              removeMessageAvailabilityListener(listener);
-            }
-          }, () -> pruneStaleSubscription(channel));
-        } catch (final Exception e) {
-          logger.warn("Unexpected error handling new message", e);
-        }
-      });
-    } else if (channel.startsWith(PERSISTING_KEYSPACE_PREFIX) && "del".equals(message)) {
-      queuePersistedNotificationCounter.increment();
-      notificationExecutorService.execute(() -> {
-        try {
-          findListener(channel).ifPresentOrElse(listener -> {
-            if (!listener.handleMessagesPersisted()) {
-              removeMessageAvailabilityListener(listener);
-            }
-          }, () -> pruneStaleSubscription(channel));
-        } catch (final Exception e) {
-          logger.warn("Unexpected error handling messages persisted", e);
-        }
-      });
-    }
-  }
-
-  private Optional<MessageAvailabilityListener> findListener(final String keyspaceChannel) {
-    final String queueName = getQueueNameFromKeyspaceChannel(keyspaceChannel);
-
-    messageListenersLock.lock();
-    try {
-      return Optional.ofNullable(messageListenersByQueueName.get(queueName));
-    } finally {
-      messageListenersLock.unlock();
-    }
-  }
-
-  @VisibleForTesting
-  static String getQueueName(final UUID accountUuid, final byte deviceId) {
-    return accountUuid + "::" + deviceId;
-  }
-
-  @VisibleForTesting
-  static String getQueueNameFromKeyspaceChannel(final String channel) {
-    final int startOfHashTag = channel.indexOf('{');
-    final int endOfHashTag = channel.lastIndexOf('}');
-
-    return channel.substring(startOfHashTag + 1, endOfHashTag);
-  }
-
   static byte[] getMessageQueueKey(final UUID accountUuid, final byte deviceId) {
     return ("user_queue::{" + accountUuid.toString() + "::" + deviceId + "}").getBytes(StandardCharsets.UTF_8);
   }

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

@@ -24,7 +24,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
-import org.whispersystems.textsecuregcm.identity.ServiceIdentifier;
 import org.whispersystems.textsecuregcm.metrics.MetricsUtil;
 import org.whispersystems.textsecuregcm.util.Pair;
 import reactor.core.observability.micrometer.Micrometer;
@@ -128,10 +127,6 @@ public class MessagesManager {
         .tap(Micrometer.metrics(Metrics.globalRegistry));
   }
 
-  public Mono<Long> getEarliestUndeliveredTimestampForDevice(UUID destinationUuid, Device destinationDevice) {
-    return Mono.from(messagesDynamoDb.load(destinationUuid, destinationDevice, 1)).map(Envelope::getServerTimestamp);
-  }
-
   public CompletableFuture<Void> clear(UUID destinationUuid) {
     return messagesCache.clear(destinationUuid);
   }
@@ -190,17 +185,6 @@ public class MessagesManager {
     return messagesRemovedFromCache;
   }
 
-  public void addMessageAvailabilityListener(
-      final UUID destinationUuid,
-      final byte destinationDeviceId,
-      final MessageAvailabilityListener listener) {
-    messagesCache.addMessageAvailabilityListener(destinationUuid, destinationDeviceId, listener);
-  }
-
-  public void removeMessageAvailabilityListener(final MessageAvailabilityListener listener) {
-    messagesCache.removeMessageAvailabilityListener(listener);
-  }
-
   /**
    * Inserts the shared multi-recipient message payload to storage.
    *
@@ -211,15 +195,4 @@ public class MessagesManager {
       final SealedSenderMultiRecipientMessage sealedSenderMultiRecipientMessage) {
     return messagesCache.insertSharedMultiRecipientMessagePayload(sealedSenderMultiRecipientMessage);
   }
-
-  /**
-   * Removes the recipient's view from shared MRM data if necessary
-   */
-  public void removeRecipientViewFromMrmData(final byte destinationDeviceId, final Envelope message) {
-    if (message.hasSharedMrmKey()) {
-      messagesCache.removeRecipientViewFromMrmData(List.of(message.getSharedMrmKey().toByteArray()),
-          ServiceIdentifier.valueOf(message.getDestinationServiceId()),
-          destinationDeviceId);
-    }
-  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/websocket/AuthenticatedConnectListener.java

@@ -19,7 +19,6 @@ import org.whispersystems.textsecuregcm.push.PubSubClientEventManager;
 import org.whispersystems.textsecuregcm.push.PushNotificationManager;
 import org.whispersystems.textsecuregcm.push.PushNotificationScheduler;
 import org.whispersystems.textsecuregcm.push.ReceiptSender;
-import org.whispersystems.textsecuregcm.redis.RedisOperation;
 import org.whispersystems.textsecuregcm.storage.ClientReleaseManager;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
 import org.whispersystems.websocket.session.WebSocketSessionContext;
@@ -109,23 +108,12 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
         pubSubClientEventManager.handleClientDisconnected(auth.getAccount().getUuid(),
             auth.getAuthenticatedDevice().getId());
 
-        // Next, we stop listening for inbound messages. If a message arrives after this call, the websocket connection
-        // will not be notified and will not change its state, but that's okay because it has already closed and
-        // attempts to deliver mesages via this connection will not succeed.
-        RedisOperation.unchecked(() -> messagesManager.removeMessageAvailabilityListener(connection));
-
         // Finally, stop trying to deliver messages and send a push notification if the connection is aware of any
         // undelivered messages.
         connection.stop();
       });
 
       try {
-        // Once we add this connection as a message availability listener, it will be notified any time a new message
-        // arrives in the message cache. This updates the connection's "may have messages" state. It's important that
-        // we do this first because we want to make sure we're accurately tracking message availability in the
-        // connection's internal state.
-        messagesManager.addMessageAvailabilityListener(auth.getAccount().getUuid(), auth.getAuthenticatedDevice().getId(), connection);
-
         // Once we "start" the websocket connection, we'll cancel any scheduled "you may have new messages" push
         // notifications and begin delivering any stored messages for the connected device. We have not yet declared the
         // client as "present" yet. If a message arrives at this point, we will update the message availability state

service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java

@@ -51,7 +51,6 @@ import org.whispersystems.textsecuregcm.push.PushNotificationScheduler;
 import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.storage.ClientReleaseManager;
 import org.whispersystems.textsecuregcm.storage.Device;
-import org.whispersystems.textsecuregcm.storage.MessageAvailabilityListener;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
 import org.whispersystems.textsecuregcm.util.HeaderUtils;
 import org.whispersystems.websocket.WebSocketClient;
@@ -63,7 +62,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
 
-public class WebSocketConnection implements MessageAvailabilityListener, ClientEventListener {
+public class WebSocketConnection implements ClientEventListener {
 
   private static final DistributionSummary messageTime = Metrics.summary(
       name(MessageController.class, "messageDeliveryDuration"));
@@ -81,8 +80,6 @@ public class WebSocketConnection implements MessageAvailabilityListener, ClientE
   private static final String DISPLACEMENT_COUNTER_NAME = name(WebSocketConnection.class, "displacement");
   private static final String NON_SUCCESS_RESPONSE_COUNTER_NAME = name(WebSocketConnection.class,
       "clientNonSuccessResponse");
-  private static final String CLIENT_CLOSED_MESSAGE_AVAILABLE_COUNTER_NAME = name(WebSocketConnection.class,
-      "messageAvailableAfterClientClosed");
   private static final String SEND_MESSAGES_FLUX_NAME = MetricsUtil.name(WebSocketConnection.class,
       "sendMessages");
   private static final String SEND_MESSAGE_ERROR_COUNTER = MetricsUtil.name(WebSocketConnection.class,
@@ -460,21 +457,6 @@ public class WebSocketConnection implements MessageAvailabilityListener, ClientE
     }
   }
 
-  @Override
-  public boolean handleNewMessagesAvailable() {
-    if (!client.isOpen()) {
-      // The client may become closed without successful removal of references to the `MessageAvailabilityListener`
-      Metrics.counter(CLIENT_CLOSED_MESSAGE_AVAILABLE_COUNTER_NAME).increment();
-      return false;
-    }
-
-    Metrics.counter(MESSAGE_AVAILABLE_COUNTER_NAME,
-        PRESENCE_MANAGER_TAG, "legacy")
-        .increment();
-
-    return true;
-  }
-
   @Override
   public void handleNewMessageAvailable() {
     Metrics.counter(MESSAGE_AVAILABLE_COUNTER_NAME,
@@ -487,22 +469,7 @@ public class WebSocketConnection implements MessageAvailabilityListener, ClientE
   }
 
   @Override
-  public boolean handleMessagesPersisted() {
-    if (!client.isOpen()) {
-      // The client may become without successful removal of references to the `MessageAvailabilityListener`
-      Metrics.counter(CLIENT_CLOSED_MESSAGE_AVAILABLE_COUNTER_NAME).increment();
-      return false;
-    }
-
-    Metrics.counter(MESSAGES_PERSISTED_COUNTER_NAME,
-        PRESENCE_MANAGER_TAG, "legacy")
-        .increment();
-
-    return true;
-  }
-
-  @Override
-  public void handleMessagesPersistedPubSub() {
+  public void handleMessagesPersisted() {
     Metrics.counter(MESSAGES_PERSISTED_COUNTER_NAME,
             PRESENCE_MANAGER_TAG, "pubsub")
         .increment();

service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java

@@ -119,8 +119,6 @@ record CommandDependencies(
 
     Scheduler messageDeliveryScheduler = Schedulers.fromExecutorService(
         environment.lifecycle().executorService("messageDelivery").minThreads(4).maxThreads(4).build());
-    ExecutorService keyspaceNotificationDispatchExecutor = environment.lifecycle()
-        .executorService(name(name, "keyspaceNotification-%d")).minThreads(4).maxThreads(4).build();
     ExecutorService messageDeletionExecutor = environment.lifecycle()
         .executorService(name(name, "messageDeletion-%d")).minThreads(4).maxThreads(4).build();
     ExecutorService secureValueRecoveryServiceExecutor = environment.lifecycle()
@@ -209,7 +207,7 @@ record CommandDependencies(
     SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
         storageServiceExecutor, storageServiceRetryExecutor, configuration.getSecureStorageServiceConfiguration());
     PubSubClientEventManager pubSubClientEventManager = new PubSubClientEventManager(messagesCluster, clientEventExecutor);
-    MessagesCache messagesCache = new MessagesCache(messagesCluster, keyspaceNotificationDispatchExecutor,
+    MessagesCache messagesCache = new MessagesCache(messagesCluster,
         messageDeliveryScheduler, messageDeletionExecutor, Clock.systemUTC(), dynamicConfigurationManager);
     ProfilesManager profilesManager = new ProfilesManager(profiles, cacheCluster);
     ReportMessageDynamoDb reportMessageDynamoDb = new ReportMessageDynamoDb(dynamoDbClient,
@@ -262,7 +260,6 @@ record CommandDependencies(
             Clock.systemUTC());
 
     environment.lifecycle().manage(apnSender);
-    environment.lifecycle().manage(messagesCache);
     environment.lifecycle().manage(pubSubClientEventManager);
     environment.lifecycle().manage(new ManagedAwsCrt());
 

service/src/main/java/org/whispersystems/textsecuregcm/workers/MessagePersisterServiceCommand.java

@@ -68,7 +68,6 @@ public class MessagePersisterServiceCommand extends ServerCommand<WhisperServerC
         Duration.ofMinutes(configuration.getMessageCacheConfiguration().getPersistDelayMinutes()),
         namespace.getInt(WORKER_COUNT));
 
-    environment.lifecycle().manage(deps.messagesCache());
     environment.lifecycle().manage(messagePersister);
 
     MetricsUtil.registerSystemResourceMetrics(environment);