Mirror disconnection requests to `DisconnectionRequestManager`

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

@@ -81,6 +81,7 @@ import org.whispersystems.textsecuregcm.auth.AccountAuthenticator;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedDevice;
 import org.whispersystems.textsecuregcm.auth.CertificateGenerator;
 import org.whispersystems.textsecuregcm.auth.CloudflareTurnCredentialsManager;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialsGenerator;
 import org.whispersystems.textsecuregcm.auth.PhoneVerificationTokenManager;
 import org.whispersystems.textsecuregcm.auth.RegistrationLockVerificationManager;
@@ -549,6 +550,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         .virtualExecutorService(name(getClass(), "appleAppStore-%d"));
     ExecutorService clientEventExecutor = environment.lifecycle()
         .virtualExecutorService(name(getClass(), "clientEvent-%d"));
+    ExecutorService disconnectionRequestListenerExecutor = environment.lifecycle()
+        .virtualExecutorService(name(getClass(), "disconnectionRequest-%d"));
 
     ScheduledExecutorService appleAppStoreRetryExecutor = environment.lifecycle()
         .scheduledExecutorService(name(getClass(), "appleAppStoreRetry-%d")).threads(1).build();
@@ -597,6 +600,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         secureValueRecoveryServiceExecutor, secureValueRecoveryServiceRetryExecutor, config.getSvr2Configuration());
     SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
         storageServiceExecutor, storageServiceRetryExecutor, config.getSecureStorageServiceConfiguration());
+    DisconnectionRequestManager disconnectionRequestManager = new DisconnectionRequestManager(pubsubClient, disconnectionRequestListenerExecutor);
     WebSocketConnectionEventManager webSocketConnectionEventManager = new WebSocketConnectionEventManager(messagesCluster, clientEventExecutor);
     ProfilesManager profilesManager = new ProfilesManager(profiles, cacheCluster);
     MessagesCache messagesCache = new MessagesCache(messagesCluster, messageDeliveryScheduler,
@@ -615,7 +619,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         new ClientPublicKeysManager(clientPublicKeys, accountLockManager, accountLockExecutor);
     AccountsManager accountsManager = new AccountsManager(accounts, phoneNumberIdentifiers, cacheCluster,
         pubsubClient, accountLockManager, keysManager, messagesManager, profilesManager,
-        secureStorageClient, secureValueRecovery2Client, webSocketConnectionEventManager,
+        secureStorageClient, secureValueRecovery2Client, disconnectionRequestManager, webSocketConnectionEventManager,
         registrationRecoveryPasswordsManager, clientPublicKeysManager, accountLockExecutor,
         clock, config.getLinkDeviceSecretConfiguration().secret().value(), dynamicConfigurationManager);
     RemoteConfigsManager remoteConfigsManager = new RemoteConfigsManager(remoteConfigs);
@@ -645,7 +649,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         new MessageDeliveryLoopMonitor(rateLimitersCluster);
 
     final RegistrationLockVerificationManager registrationLockVerificationManager = new RegistrationLockVerificationManager(
-        accountsManager, webSocketConnectionEventManager, svr2CredentialsGenerator, svr3CredentialsGenerator,
+        accountsManager, disconnectionRequestManager, webSocketConnectionEventManager, svr2CredentialsGenerator, svr3CredentialsGenerator,
         registrationRecoveryPasswordsManager, pushNotificationManager, rateLimiters);
 
     final ReportedMessageMetricsListener reportedMessageMetricsListener = new ReportedMessageMetricsListener(
@@ -974,7 +978,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     environment.jersey().register(new AuthDynamicFeature(accountAuthFilter));
     environment.jersey().register(new AuthValueFactoryProvider.Binder<>(AuthenticatedDevice.class));
     environment.jersey().register(new WebsocketRefreshApplicationEventListener(accountsManager,
-        webSocketConnectionEventManager));
+        disconnectionRequestManager, webSocketConnectionEventManager));
     environment.jersey().register(new TimestampResponseFilter());
 
     ///
@@ -987,7 +991,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
             pushNotificationScheduler, webSocketConnectionEventManager, websocketScheduledExecutor,
             messageDeliveryScheduler, clientReleaseManager, messageDeliveryLoopMonitor));
     webSocketEnvironment.jersey()
-        .register(new WebsocketRefreshApplicationEventListener(accountsManager, webSocketConnectionEventManager));
+        .register(new WebsocketRefreshApplicationEventListener(accountsManager, disconnectionRequestManager, webSocketConnectionEventManager));
     webSocketEnvironment.jersey().register(new RateLimitByIpFilter(rateLimiters));
     webSocketEnvironment.jersey().register(new RequestStatisticsFilter(TrafficSource.WEBSOCKET));
     webSocketEnvironment.jersey().register(MultiRecipientMessageProvider.class);
@@ -1130,7 +1134,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     WebSocketEnvironment<AuthenticatedDevice> provisioningEnvironment = new WebSocketEnvironment<>(environment,
         webSocketEnvironment.getRequestLog(), Duration.ofMillis(60000));
     provisioningEnvironment.jersey().register(new WebsocketRefreshApplicationEventListener(accountsManager,
-        webSocketConnectionEventManager));
+        disconnectionRequestManager, webSocketConnectionEventManager));
     provisioningEnvironment.setConnectListener(new ProvisioningConnectListener(provisioningManager));
     provisioningEnvironment.jersey().register(new MetricsApplicationEventListener(TrafficSource.WEBSOCKET, clientReleaseManager));
     provisioningEnvironment.jersey().register(new KeepAliveController(webSocketConnectionEventManager));

service/src/main/java/org/whispersystems/textsecuregcm/auth/RegistrationLockVerificationManager.java

@@ -54,6 +54,7 @@ public class RegistrationLockVerificationManager {
   private static final String PHONE_VERIFICATION_TYPE_TAG_NAME = "phoneVerificationType";
 
   private final AccountsManager accounts;
+  private final DisconnectionRequestManager disconnectionRequestManager;
   private final WebSocketConnectionEventManager webSocketConnectionEventManager;
   private final ExternalServiceCredentialsGenerator svr2CredentialGenerator;
   private final ExternalServiceCredentialsGenerator svr3CredentialGenerator;
@@ -63,6 +64,7 @@ public class RegistrationLockVerificationManager {
 
   public RegistrationLockVerificationManager(
       final AccountsManager accounts,
+      final DisconnectionRequestManager disconnectionRequestManager,
       final WebSocketConnectionEventManager webSocketConnectionEventManager,
       final ExternalServiceCredentialsGenerator svr2CredentialGenerator,
       final ExternalServiceCredentialsGenerator svr3CredentialGenerator,
@@ -70,6 +72,7 @@ public class RegistrationLockVerificationManager {
       final PushNotificationManager pushNotificationManager,
       final RateLimiters rateLimiters) {
     this.accounts = accounts;
+    this.disconnectionRequestManager = disconnectionRequestManager;
     this.webSocketConnectionEventManager = webSocketConnectionEventManager;
     this.svr2CredentialGenerator = svr2CredentialGenerator;
     this.svr3CredentialGenerator = svr3CredentialGenerator;
@@ -162,6 +165,7 @@ public class RegistrationLockVerificationManager {
 
       final List<Byte> deviceIds = updatedAccount.getDevices().stream().map(Device::getId).toList();
       webSocketConnectionEventManager.requestDisconnection(updatedAccount.getUuid(), deviceIds);
+      disconnectionRequestManager.requestDisconnection(updatedAccount.getUuid(), deviceIds);
 
       try {
         // Send a push notification that prompts the client to attempt login and fail due to locked credentials

service/src/main/java/org/whispersystems/textsecuregcm/auth/WebsocketRefreshApplicationEventListener.java

@@ -20,9 +20,11 @@ public class WebsocketRefreshApplicationEventListener implements ApplicationEven
   private final WebsocketRefreshRequestEventListener websocketRefreshRequestEventListener;
 
   public WebsocketRefreshApplicationEventListener(final AccountsManager accountsManager,
+      final DisconnectionRequestManager disconnectionRequestManager,
       final WebSocketConnectionEventManager webSocketConnectionEventManager) {
 
     this.websocketRefreshRequestEventListener = new WebsocketRefreshRequestEventListener(
+        disconnectionRequestManager,
         webSocketConnectionEventManager,
         new LinkedDeviceRefreshRequirementProvider(accountsManager),
         new PhoneNumberChangeRefreshRequirementProvider(accountsManager));

service/src/main/java/org/whispersystems/textsecuregcm/auth/WebsocketRefreshRequestEventListener.java

@@ -23,6 +23,7 @@ import org.whispersystems.textsecuregcm.push.WebSocketConnectionEventManager;
 
 public class WebsocketRefreshRequestEventListener implements RequestEventListener {
 
+  private final DisconnectionRequestManager disconnectionRequestManager;
   private final WebSocketConnectionEventManager webSocketConnectionEventManager;
   private final WebsocketRefreshRequirementProvider[] providers;
 
@@ -35,9 +36,11 @@ public class WebsocketRefreshRequestEventListener implements RequestEventListene
   private static final Logger logger = LoggerFactory.getLogger(WebsocketRefreshRequestEventListener.class);
 
   public WebsocketRefreshRequestEventListener(
+      final DisconnectionRequestManager disconnectionRequestManager,
       final WebSocketConnectionEventManager webSocketConnectionEventManager,
       final WebsocketRefreshRequirementProvider... providers) {
 
+    this.disconnectionRequestManager = disconnectionRequestManager;
     this.webSocketConnectionEventManager = webSocketConnectionEventManager;
     this.providers = providers;
   }
@@ -61,6 +64,7 @@ public class WebsocketRefreshRequestEventListener implements RequestEventListene
             try {
               displacedDevices.incrementAndGet();
               webSocketConnectionEventManager.requestDisconnection(pair.first(), List.of(pair.second()));
+              disconnectionRequestManager.requestDisconnection(pair.first(), List.of(pair.second()));
             } catch (final Exception e) {
               logger.error("Could not displace device presence", e);
             }

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

@@ -63,6 +63,7 @@ import org.apache.commons.lang3.StringUtils;
 import org.signal.libsignal.protocol.IdentityKey;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.auth.SaltedTokenHash;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.controllers.MismatchedDevicesException;
@@ -124,6 +125,7 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
   private final ProfilesManager profilesManager;
   private final SecureStorageClient secureStorageClient;
   private final SecureValueRecovery2Client secureValueRecovery2Client;
+  private final DisconnectionRequestManager disconnectionRequestManager;
   private final WebSocketConnectionEventManager webSocketConnectionEventManager;
   private final RegistrationRecoveryPasswordsManager registrationRecoveryPasswordsManager;
   private final ClientPublicKeysManager clientPublicKeysManager;
@@ -202,6 +204,7 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
       final ProfilesManager profilesManager,
       final SecureStorageClient secureStorageClient,
       final SecureValueRecovery2Client secureValueRecovery2Client,
+      final DisconnectionRequestManager disconnectionRequestManager,
       final WebSocketConnectionEventManager webSocketConnectionEventManager,
       final RegistrationRecoveryPasswordsManager registrationRecoveryPasswordsManager,
       final ClientPublicKeysManager clientPublicKeysManager,
@@ -219,6 +222,7 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
     this.profilesManager = profilesManager;
     this.secureStorageClient = secureStorageClient;
     this.secureValueRecovery2Client = secureValueRecovery2Client;
+    this.disconnectionRequestManager = disconnectionRequestManager;
     this.webSocketConnectionEventManager = webSocketConnectionEventManager;
     this.registrationRecoveryPasswordsManager = requireNonNull(registrationRecoveryPasswordsManager);
     this.clientPublicKeysManager = clientPublicKeysManager;
@@ -326,6 +330,7 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
                   messagesManager.clear(aci),
                   profilesManager.deleteAll(aci))
               .thenCompose(ignored -> webSocketConnectionEventManager.requestDisconnection(aci))
+              .thenCompose(ignored -> disconnectionRequestManager.requestDisconnection(aci))
               .thenCompose(ignored -> accounts.reclaimAccount(e.getExistingAccount(), account, additionalWriteItems))
               .thenCompose(ignored -> {
                 // We should have cleared all messages before overwriting the old account, but more may have arrived
@@ -590,6 +595,7 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
         .whenComplete((ignored, throwable) -> {
           if (throwable == null) {
             webSocketConnectionEventManager.requestDisconnection(accountIdentifier, List.of(deviceId));
+            disconnectionRequestManager.requestDisconnection(accountIdentifier, List.of(deviceId));
           }
         });
   }
@@ -1236,7 +1242,10 @@ public class AccountsManager extends RedisPubSubAdapter<String, String> implemen
             registrationRecoveryPasswordsManager.removeForNumber(account.getNumber()))
         .thenCompose(ignored -> accounts.delete(account.getUuid(), additionalWriteItems))
         .thenCompose(ignored -> redisDeleteAsync(account))
-        .thenRun(() -> webSocketConnectionEventManager.requestDisconnection(account.getUuid()));
+        .thenRun(() -> {
+          webSocketConnectionEventManager.requestDisconnection(account.getUuid());
+          disconnectionRequestManager.requestDisconnection(account.getUuid());
+        });
   }
 
   private String getAccountMapKey(String key) {

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

@@ -23,6 +23,7 @@ import org.signal.libsignal.zkgroup.GenericServerSecretParams;
 import org.signal.libsignal.zkgroup.InvalidInputException;
 import org.whispersystems.textsecuregcm.WhisperServerConfiguration;
 import org.whispersystems.textsecuregcm.attachments.TusAttachmentGenerator;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialsGenerator;
 import org.whispersystems.textsecuregcm.backup.BackupManager;
 import org.whispersystems.textsecuregcm.backup.BackupsDb;
@@ -136,6 +137,8 @@ record CommandDependencies(
         .maxThreads(16).minThreads(16).build();
     ExecutorService clientEventExecutor = environment.lifecycle()
         .virtualExecutorService(name(name, "clientEvent-%d"));
+    ExecutorService disconnectionRequestListenerExecutor = environment.lifecycle()
+        .virtualExecutorService(name(name, "disconnectionRequest-%d"));
 
     ScheduledExecutorService secureValueRecoveryServiceRetryExecutor = environment.lifecycle()
         .scheduledExecutorService(name(name, "secureValueRecoveryServiceRetry-%d")).threads(1).build();
@@ -205,6 +208,7 @@ record CommandDependencies(
         configuration.getSvr2Configuration());
     SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
         storageServiceExecutor, storageServiceRetryExecutor, configuration.getSecureStorageServiceConfiguration());
+    DisconnectionRequestManager disconnectionRequestManager = new DisconnectionRequestManager(pubsubClient, disconnectionRequestListenerExecutor);
     WebSocketConnectionEventManager webSocketConnectionEventManager = new WebSocketConnectionEventManager(messagesCluster, clientEventExecutor);
     MessagesCache messagesCache = new MessagesCache(messagesCluster,
         messageDeliveryScheduler, messageDeletionExecutor, Clock.systemUTC(), dynamicConfigurationManager);
@@ -222,7 +226,7 @@ record CommandDependencies(
         new ClientPublicKeysManager(clientPublicKeys, accountLockManager, accountLockExecutor);
     AccountsManager accountsManager = new AccountsManager(accounts, phoneNumberIdentifiers, cacheCluster,
         pubsubClient, accountLockManager, keys, messagesManager, profilesManager,
-        secureStorageClient, secureValueRecovery2Client, webSocketConnectionEventManager,
+        secureStorageClient, secureValueRecovery2Client, disconnectionRequestManager, webSocketConnectionEventManager,
         registrationRecoveryPasswordsManager, clientPublicKeysManager, accountLockExecutor,
         clock, configuration.getLinkDeviceSecretConfiguration().secret().value(), dynamicConfigurationManager);
     RateLimiters rateLimiters = RateLimiters.createAndValidate(configuration.getLimitsConfiguration(),

service/src/test/java/org/whispersystems/textsecuregcm/auth/LinkedDeviceRefreshRequirementProviderTest.java

@@ -95,6 +95,7 @@ class LinkedDeviceRefreshRequirementProviderTest {
       .build();
 
   private AccountsManager accountsManager;
+  private DisconnectionRequestManager disconnectionRequestManager;
   private WebSocketConnectionEventManager webSocketConnectionEventManager;
 
   private LinkedDeviceRefreshRequirementProvider provider;
@@ -102,12 +103,13 @@ class LinkedDeviceRefreshRequirementProviderTest {
   @BeforeEach
   void setup() {
     accountsManager = mock(AccountsManager.class);
+    disconnectionRequestManager = mock(DisconnectionRequestManager.class);
     webSocketConnectionEventManager = mock(WebSocketConnectionEventManager.class);
 
     provider = new LinkedDeviceRefreshRequirementProvider(accountsManager);
 
     final WebsocketRefreshRequestEventListener listener =
-        new WebsocketRefreshRequestEventListener(webSocketConnectionEventManager, provider);
+        new WebsocketRefreshRequestEventListener(disconnectionRequestManager, webSocketConnectionEventManager, provider);
 
     when(applicationEventListener.onRequest(any())).thenReturn(listener);
 
@@ -142,6 +144,10 @@ class LinkedDeviceRefreshRequirementProviderTest {
     verify(webSocketConnectionEventManager).requestDisconnection(account.getUuid(), List.of((byte) 1));
     verify(webSocketConnectionEventManager).requestDisconnection(account.getUuid(), List.of((byte) 2));
     verify(webSocketConnectionEventManager).requestDisconnection(account.getUuid(), List.of((byte) 3));
+
+    verify(disconnectionRequestManager).requestDisconnection(account.getUuid(), List.of((byte) 1));
+    verify(disconnectionRequestManager).requestDisconnection(account.getUuid(), List.of((byte) 2));
+    verify(disconnectionRequestManager).requestDisconnection(account.getUuid(), List.of((byte) 3));
   }
 
   @ParameterizedTest
@@ -170,6 +176,7 @@ class LinkedDeviceRefreshRequirementProviderTest {
     assertEquals(200, response.getStatus());
 
     initialDeviceIds.forEach(deviceId -> {
+      verify(disconnectionRequestManager).requestDisconnection(account.getUuid(), List.of(deviceId));
       verify(webSocketConnectionEventManager).requestDisconnection(account.getUuid(), List.of(deviceId));
     });
 

service/src/test/java/org/whispersystems/textsecuregcm/auth/PhoneNumberChangeRefreshRequirementProviderTest.java

@@ -74,7 +74,11 @@ class PhoneNumberChangeRefreshRequirementProviderTest {
 
   private static final AccountAuthenticator AUTHENTICATOR = mock(AccountAuthenticator.class);
   private static final AccountsManager ACCOUNTS_MANAGER = mock(AccountsManager.class);
-  private static final WebSocketConnectionEventManager PUBSUB_CLIENT_PRESENCE = mock(WebSocketConnectionEventManager.class);
+  private static final DisconnectionRequestManager DISCONNECTION_REQUEST_MANAGER =
+      mock(DisconnectionRequestManager.class);
+
+  private static final WebSocketConnectionEventManager WEB_SOCKET_CONNECTION_EVENT_MANAGER =
+      mock(WebSocketConnectionEventManager.class);
 
   private WebSocketClient client;
   private final Account account1 = new Account();
@@ -84,7 +88,7 @@ class PhoneNumberChangeRefreshRequirementProviderTest {
 
   @BeforeEach
   void setUp() throws Exception {
-    reset(AUTHENTICATOR, ACCOUNTS_MANAGER, PUBSUB_CLIENT_PRESENCE);
+    reset(AUTHENTICATOR, ACCOUNTS_MANAGER, WEB_SOCKET_CONNECTION_EVENT_MANAGER);
     client = new WebSocketClient();
     client.start();
 
@@ -123,9 +127,9 @@ class PhoneNumberChangeRefreshRequirementProviderTest {
           .addMappingForUrlPatterns(EnumSet.of(DispatcherType.REQUEST), false, "/*");
       webSocketEnvironment.jersey().register(new RemoteAddressFilter());
       webSocketEnvironment.jersey()
-          .register(new WebsocketRefreshApplicationEventListener(ACCOUNTS_MANAGER, PUBSUB_CLIENT_PRESENCE));
+          .register(new WebsocketRefreshApplicationEventListener(ACCOUNTS_MANAGER, DISCONNECTION_REQUEST_MANAGER, WEB_SOCKET_CONNECTION_EVENT_MANAGER));
       environment.jersey()
-          .register(new WebsocketRefreshApplicationEventListener(ACCOUNTS_MANAGER, PUBSUB_CLIENT_PRESENCE));
+          .register(new WebsocketRefreshApplicationEventListener(ACCOUNTS_MANAGER, DISCONNECTION_REQUEST_MANAGER, WEB_SOCKET_CONNECTION_EVENT_MANAGER));
       webSocketEnvironment.setConnectListener(webSocketSessionContext -> {
       });
 
@@ -199,7 +203,7 @@ class PhoneNumberChangeRefreshRequirementProviderTest {
 
     // Event listeners can fire after responses are sent
     verify(ACCOUNTS_MANAGER, timeout(5000).times(1)).getByAccountIdentifier(eq(account1.getUuid()));
-    verifyNoMoreInteractions(PUBSUB_CLIENT_PRESENCE);
+    verifyNoMoreInteractions(WEB_SOCKET_CONNECTION_EVENT_MANAGER);
     verifyNoMoreInteractions(ACCOUNTS_MANAGER);
   }
 
@@ -213,9 +217,14 @@ class PhoneNumberChangeRefreshRequirementProviderTest {
 
     // Make sure we disconnect the account if the account has changed numbers. Event listeners can fire after responses
     // are sent, so use a timeout.
-    verify(PUBSUB_CLIENT_PRESENCE, timeout(5000))
+    verify(DISCONNECTION_REQUEST_MANAGER, timeout(5000))
         .requestDisconnection(account1.getUuid(), List.of(authenticatedDevice.getId()));
-    verifyNoMoreInteractions(PUBSUB_CLIENT_PRESENCE);
+    verifyNoMoreInteractions(DISCONNECTION_REQUEST_MANAGER);
+
+
+    verify(WEB_SOCKET_CONNECTION_EVENT_MANAGER, timeout(5000))
+        .requestDisconnection(account1.getUuid(), List.of(authenticatedDevice.getId()));
+    verifyNoMoreInteractions(WEB_SOCKET_CONNECTION_EVENT_MANAGER);
   }
 
   @Test
@@ -229,9 +238,13 @@ class PhoneNumberChangeRefreshRequirementProviderTest {
 
     // Make sure we disconnect the account if the account has changed numbers. Event listeners can fire after responses
     // are sent, so use a timeout.
-    verify(PUBSUB_CLIENT_PRESENCE, timeout(5000))
+    verify(DISCONNECTION_REQUEST_MANAGER, timeout(5000))
         .requestDisconnection(account1.getUuid(), List.of(authenticatedDevice.getId()));
-    verifyNoMoreInteractions(PUBSUB_CLIENT_PRESENCE);
+    verifyNoMoreInteractions(DISCONNECTION_REQUEST_MANAGER);
+
+    verify(WEB_SOCKET_CONNECTION_EVENT_MANAGER, timeout(5000))
+        .requestDisconnection(account1.getUuid(), List.of(authenticatedDevice.getId()));
+    verifyNoMoreInteractions(WEB_SOCKET_CONNECTION_EVENT_MANAGER);
   }
 
   @ParameterizedTest

service/src/test/java/org/whispersystems/textsecuregcm/auth/RegistrationLockVerificationManagerTest.java

@@ -46,6 +46,7 @@ import org.whispersystems.textsecuregcm.util.Pair;
 class RegistrationLockVerificationManagerTest {
 
   private final AccountsManager accountsManager = mock(AccountsManager.class);
+  private final DisconnectionRequestManager disconnectionRequestManager = mock(DisconnectionRequestManager.class);
   private final WebSocketConnectionEventManager webSocketConnectionEventManager = mock(WebSocketConnectionEventManager.class);
   private final ExternalServiceCredentialsGenerator svr2CredentialsGenerator = mock(
       ExternalServiceCredentialsGenerator.class);
@@ -56,8 +57,8 @@ class RegistrationLockVerificationManagerTest {
   private static PushNotificationManager pushNotificationManager = mock(PushNotificationManager.class);
   private final RateLimiters rateLimiters = mock(RateLimiters.class);
   private final RegistrationLockVerificationManager registrationLockVerificationManager = new RegistrationLockVerificationManager(
-      accountsManager, webSocketConnectionEventManager, svr2CredentialsGenerator, svr3CredentialsGenerator,
+      accountsManager, disconnectionRequestManager, webSocketConnectionEventManager, svr2CredentialsGenerator,
-      registrationRecoveryPasswordsManager, pushNotificationManager, rateLimiters);
+      svr3CredentialsGenerator, registrationRecoveryPasswordsManager, pushNotificationManager, rateLimiters);
 
   private final RateLimiter pinLimiter = mock(RateLimiter.class);
 
@@ -107,6 +108,7 @@ class RegistrationLockVerificationManagerTest {
             } else {
               verify(registrationRecoveryPasswordsManager, never()).removeForNumber(account.getNumber());
             }
+            verify(disconnectionRequestManager).requestDisconnection(account.getUuid(), List.of(Device.PRIMARY_ID));
             verify(webSocketConnectionEventManager).requestDisconnection(account.getUuid(), List.of(Device.PRIMARY_ID));
             try {
               verify(pushNotificationManager).sendAttemptLoginNotification(any(), eq("failedRegistrationLock"));
@@ -130,6 +132,7 @@ class RegistrationLockVerificationManagerTest {
             verify(pushNotificationManager, never()).sendAttemptLoginNotification(any(), eq("failedRegistrationLock"));
           } catch (NotPushRegisteredException npre) {}
           verify(registrationRecoveryPasswordsManager, never()).removeForNumber(account.getNumber());
+          verify(disconnectionRequestManager, never()).requestDisconnection(any(), any());
           verify(webSocketConnectionEventManager, never()).requestDisconnection(any(), any());
         });
       }
@@ -168,6 +171,7 @@ class RegistrationLockVerificationManagerTest {
 
     verify(account, never()).lockAuthTokenHash();
     verify(registrationRecoveryPasswordsManager, never()).removeForNumber(account.getNumber());
+    verify(disconnectionRequestManager, never()).requestDisconnection(any(), any());
     verify(webSocketConnectionEventManager, never()).requestDisconnection(any(), any());
   }
 

service/src/test/java/org/whispersystems/textsecuregcm/controllers/DeviceControllerTest.java

@@ -60,6 +60,7 @@ import org.signal.libsignal.protocol.IdentityKey;
 import org.signal.libsignal.protocol.ecc.Curve;
 import org.signal.libsignal.protocol.ecc.ECKeyPair;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedDevice;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.auth.WebsocketRefreshApplicationEventListener;
 import org.whispersystems.textsecuregcm.entities.AccountAttributes;
 import org.whispersystems.textsecuregcm.entities.ApnRegistrationId;
@@ -110,6 +111,7 @@ class DeviceControllerTest {
   private static final Account account = mock(Account.class);
   private static final Account maxedAccount = mock(Account.class);
   private static final Device primaryDevice = mock(Device.class);
+  private static final DisconnectionRequestManager disconnectionRequestManager = mock(DisconnectionRequestManager.class);
   private static final WebSocketConnectionEventManager webSocketConnectionEventManager = mock(WebSocketConnectionEventManager.class);
   private static final Map<String, Integer> deviceConfiguration = new HashMap<>();
   private static final TestClock testClock = TestClock.now();
@@ -131,7 +133,7 @@ class DeviceControllerTest {
       .addProvider(new AuthValueFactoryProvider.Binder<>(AuthenticatedDevice.class))
       .addProvider(new RateLimitExceededExceptionMapper())
       .setTestContainerFactory(new GrizzlyWebTestContainerFactory())
-      .addProvider(new WebsocketRefreshApplicationEventListener(accountsManager, webSocketConnectionEventManager))
+      .addProvider(new WebsocketRefreshApplicationEventListener(accountsManager, disconnectionRequestManager, webSocketConnectionEventManager))
       .addProvider(new DeviceLimitExceededExceptionMapper())
       .addResource(deviceController)
       .build();

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

@@ -7,6 +7,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.google.i18n.phonenumbers.PhoneNumberUtil;
@@ -36,6 +37,7 @@ import org.junitpioneer.jupiter.cartesian.CartesianTest;
 import org.signal.libsignal.protocol.IdentityKey;
 import org.signal.libsignal.protocol.ecc.Curve;
 import org.signal.libsignal.protocol.ecc.ECKeyPair;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.entities.AccountAttributes;
 import org.whispersystems.textsecuregcm.entities.ApnRegistrationId;
@@ -77,6 +79,8 @@ public class AccountCreationDeletionIntegrationTest {
   private AccountsManager accountsManager;
   private KeysManager keysManager;
   private ClientPublicKeysManager clientPublicKeysManager;
+  private WebSocketConnectionEventManager webSocketConnectionEventManager;
+  private DisconnectionRequestManager disconnectionRequestManager;
 
   record DeliveryChannels(boolean fetchesMessages, String apnsToken, String fcmToken) {}
 
@@ -138,10 +142,13 @@ public class AccountCreationDeletionIntegrationTest {
     when(registrationRecoveryPasswordsManager.removeForNumber(any()))
         .thenReturn(CompletableFuture.completedFuture(null));
 
-    final WebSocketConnectionEventManager webSocketConnectionEventManager = mock(WebSocketConnectionEventManager.class);
+    webSocketConnectionEventManager = mock(WebSocketConnectionEventManager.class);
     when(webSocketConnectionEventManager.requestDisconnection(any()))
         .thenReturn(CompletableFuture.completedFuture(null));
 
+    disconnectionRequestManager = mock(DisconnectionRequestManager.class);
+    when(disconnectionRequestManager.requestDisconnection(any())).thenReturn(CompletableFuture.completedFuture(null));
+
     accountsManager = new AccountsManager(
         accounts,
         phoneNumberIdentifiers,
@@ -153,6 +160,7 @@ public class AccountCreationDeletionIntegrationTest {
         profilesManager,
         secureStorageClient,
         svr2Client,
+        disconnectionRequestManager,
         webSocketConnectionEventManager,
         registrationRecoveryPasswordsManager,
         clientPublicKeysManager,
@@ -399,6 +407,9 @@ public class AccountCreationDeletionIntegrationTest {
         pniPqLastResortPreKey);
 
     assertEquals(existingAccountUuid, reregisteredAccount.getUuid());
+
+    verify(webSocketConnectionEventManager).requestDisconnection(existingAccountUuid);
+    verify(disconnectionRequestManager).requestDisconnection(existingAccountUuid);
   }
 
   @Test
@@ -472,6 +483,9 @@ public class AccountCreationDeletionIntegrationTest {
     assertFalse(keysManager.getLastResort(account.getUuid(), Device.PRIMARY_ID).join().isPresent());
     assertFalse(keysManager.getLastResort(account.getPhoneNumberIdentifier(), Device.PRIMARY_ID).join().isPresent());
     assertFalse(clientPublicKeysManager.findPublicKey(account.getUuid(), Device.PRIMARY_ID).join().isPresent());
+
+    verify(webSocketConnectionEventManager).requestDisconnection(aci);
+    verify(disconnectionRequestManager).requestDisconnection(aci);
   }
 
   @SuppressWarnings("OptionalUsedAsFieldOrParameterType")

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

@@ -31,6 +31,7 @@ import org.junit.jupiter.api.extension.RegisterExtension;
 import org.signal.libsignal.protocol.IdentityKey;
 import org.signal.libsignal.protocol.ecc.Curve;
 import org.signal.libsignal.protocol.ecc.ECKeyPair;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.controllers.MismatchedDevicesException;
 import org.whispersystems.textsecuregcm.entities.AccountAttributes;
@@ -66,6 +67,7 @@ class AccountsManagerChangeNumberIntegrationTest {
   static final RedisClusterExtension CACHE_CLUSTER_EXTENSION = RedisClusterExtension.builder().build();
 
   private KeysManager keysManager;
+  private DisconnectionRequestManager disconnectionRequestManager;
   private WebSocketConnectionEventManager webSocketConnectionEventManager;
   private ExecutorService accountLockExecutor;
 
@@ -117,6 +119,7 @@ class AccountsManagerChangeNumberIntegrationTest {
       when(svr2Client.deleteBackups(any())).thenReturn(CompletableFuture.completedFuture(null));
 
       webSocketConnectionEventManager = mock(WebSocketConnectionEventManager.class);
+      disconnectionRequestManager = mock(DisconnectionRequestManager.class);
 
       final PhoneNumberIdentifiers phoneNumberIdentifiers =
           new PhoneNumberIdentifiers(DYNAMO_DB_EXTENSION.getDynamoDbClient(), Tables.PNI.tableName());
@@ -144,6 +147,7 @@ class AccountsManagerChangeNumberIntegrationTest {
           profilesManager,
           secureStorageClient,
           svr2Client,
+          disconnectionRequestManager,
           webSocketConnectionEventManager,
           registrationRecoveryPasswordsManager,
           clientPublicKeysManager,
@@ -275,6 +279,7 @@ class AccountsManagerChangeNumberIntegrationTest {
     assertEquals(secondNumber, accountsManager.getByAccountIdentifier(originalUuid).map(Account::getNumber).orElseThrow());
 
     verify(webSocketConnectionEventManager).requestDisconnection(existingAccountUuid);
+    verify(disconnectionRequestManager).requestDisconnection(existingAccountUuid);
 
     assertEquals(Optional.of(existingAccountUuid), accountsManager.findRecentlyDeletedAccountIdentifier(originalNumber));
     assertEquals(Optional.empty(), accountsManager.findRecentlyDeletedAccountIdentifier(secondNumber));

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

@@ -43,6 +43,7 @@ import org.mockito.stubbing.Answer;
 import org.signal.libsignal.protocol.IdentityKey;
 import org.signal.libsignal.protocol.ecc.Curve;
 import org.signal.libsignal.protocol.ecc.ECKeyPair;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.auth.SaltedTokenHash;
 import org.whispersystems.textsecuregcm.auth.UnidentifiedAccessUtil;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
@@ -133,6 +134,7 @@ class AccountsManagerConcurrentModificationIntegrationTest {
           mock(ProfilesManager.class),
           mock(SecureStorageClient.class),
           mock(SecureValueRecovery2Client.class),
+          mock(DisconnectionRequestManager.class),
           mock(WebSocketConnectionEventManager.class),
           mock(RegistrationRecoveryPasswordsManager.class),
           mock(ClientPublicKeysManager.class),

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

@@ -11,6 +11,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.entities.RestoreAccountRequest;
 import org.whispersystems.textsecuregcm.entities.RemoteAttachment;
 import org.whispersystems.textsecuregcm.identity.IdentityType;
@@ -62,6 +63,7 @@ public class AccountsManagerDeviceTransferIntegrationTest {
         mock(ProfilesManager.class),
         mock(SecureStorageClient.class),
         mock(SecureValueRecovery2Client.class),
+        mock(DisconnectionRequestManager.class),
         mock(WebSocketConnectionEventManager.class),
         mock(RegistrationRecoveryPasswordsManager.class),
         mock(ClientPublicKeysManager.class),

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

@@ -70,6 +70,7 @@ import org.mockito.stubbing.Answer;
 import org.signal.libsignal.protocol.IdentityKey;
 import org.signal.libsignal.protocol.ecc.Curve;
 import org.signal.libsignal.protocol.ecc.ECKeyPair;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.auth.UnidentifiedAccessUtil;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.controllers.MismatchedDevicesException;
@@ -117,6 +118,7 @@ class AccountsManagerTest {
   private KeysManager keysManager;
   private MessagesManager messagesManager;
   private ProfilesManager profilesManager;
+  private DisconnectionRequestManager disconnectionRequestManager;
   private WebSocketConnectionEventManager webSocketConnectionEventManager;
   private ClientPublicKeysManager clientPublicKeysManager;
 
@@ -152,6 +154,7 @@ class AccountsManagerTest {
     keysManager = mock(KeysManager.class);
     messagesManager = mock(MessagesManager.class);
     profilesManager = mock(ProfilesManager.class);
+    disconnectionRequestManager = mock(DisconnectionRequestManager.class);
     webSocketConnectionEventManager = mock(WebSocketConnectionEventManager.class);
     clientPublicKeysManager = mock(ClientPublicKeysManager.class);
     dynamicConfiguration = mock(DynamicConfiguration.class);
@@ -241,6 +244,8 @@ class AccountsManagerTest {
     when(webSocketConnectionEventManager.requestDisconnection(any()))
         .thenReturn(CompletableFuture.completedFuture(null));
 
+    when(disconnectionRequestManager.requestDisconnection(any())).thenReturn(CompletableFuture.completedFuture(null));
+
     accountsManager = new AccountsManager(
         accounts,
         phoneNumberIdentifiers,
@@ -252,6 +257,7 @@ class AccountsManagerTest {
         profilesManager,
         storageClient,
         svr2Client,
+        disconnectionRequestManager,
         webSocketConnectionEventManager,
         registrationRecoveryPasswordsManager,
         clientPublicKeysManager,
@@ -879,6 +885,7 @@ class AccountsManagerTest {
     verify(messagesManager, times(2)).clear(existingUuid);
     verify(profilesManager, times(2)).deleteAll(existingUuid);
     verify(webSocketConnectionEventManager).requestDisconnection(existingUuid);
+    verify(disconnectionRequestManager).requestDisconnection(existingUuid);
   }
 
   @Test

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

@@ -34,6 +34,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.mockito.Mockito;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.push.WebSocketConnectionEventManager;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisClient;
@@ -137,6 +138,9 @@ class AccountsManagerUsernameIntegrationTest {
     final WebSocketConnectionEventManager webSocketConnectionEventManager = mock(WebSocketConnectionEventManager.class);
     when(webSocketConnectionEventManager.requestDisconnection(any())).thenReturn(CompletableFuture.completedFuture(null));
 
+    final DisconnectionRequestManager disconnectionRequestManager = mock(DisconnectionRequestManager.class);
+    when(disconnectionRequestManager.requestDisconnection(any())).thenReturn(CompletableFuture.completedFuture(null));
+
     accountsManager = new AccountsManager(
         accounts,
         phoneNumberIdentifiers,
@@ -148,6 +152,7 @@ class AccountsManagerUsernameIntegrationTest {
         profileManager,
         mock(SecureStorageClient.class),
         mock(SecureValueRecovery2Client.class),
+        disconnectionRequestManager,
         webSocketConnectionEventManager,
         mock(RegistrationRecoveryPasswordsManager.class),
         mock(ClientPublicKeysManager.class),

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

@@ -30,6 +30,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.signal.libsignal.protocol.ecc.Curve;
 import org.signal.libsignal.protocol.ecc.ECKeyPair;
+import org.whispersystems.textsecuregcm.auth.DisconnectionRequestManager;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.entities.DeviceInfo;
 import org.whispersystems.textsecuregcm.identity.IdentityType;
@@ -149,6 +150,7 @@ public class AddRemoveDeviceIntegrationTest {
         profilesManager,
         secureStorageClient,
         svr2Client,
+        mock(DisconnectionRequestManager.class),
         mock(WebSocketConnectionEventManager.class),
         registrationRecoveryPasswordsManager,
         clientPublicKeysManager,