Make MessagesManager#clear asynchronous

service/src/main/java/org/whispersystems/textsecuregcm/controllers/DeviceController.java

@@ -142,10 +142,10 @@ public void removeDevice(@Auth AuthenticatedAccount auth, @PathParam("device_id"
 
     final CompletableFuture<Void> deleteKeysFuture = keys.delete(account.getUuid(), deviceId);
 
-    messages.clear(account.getUuid(), deviceId);
+    messages.clear(account.getUuid(), deviceId).join();
     account = accounts.update(account, a -> a.removeDevice(deviceId));
     // ensure any messages that came in after the first clear() are also removed
-    messages.clear(account.getUuid(), deviceId);
+    messages.clear(account.getUuid(), deviceId).join();
 
     deleteKeysFuture.join();
   }
@@ -419,7 +419,7 @@ private Pair<Account, Device> createDevice(final String password,
           keys.delete(a.getUuid(), device.getId()),
           keys.delete(a.getPhoneNumberIdentifier(), device.getId()));
 
-      messages.clear(a.getUuid(), device.getId());
+      messages.clear(a.getUuid(), device.getId()).join();
 
       deleteKeysFuture.join();
 

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

@@ -235,7 +235,7 @@ public Account create(final String number,
               keysManager.delete(actualUuid),
               keysManager.delete(account.getPhoneNumberIdentifier()));
 
-          messagesManager.clear(actualUuid);
+          messagesManager.clear(actualUuid).join();
           profilesManager.deleteAll(actualUuid);
 
           deleteKeysFuture.join();
@@ -877,12 +877,15 @@ private void delete(final Account account) {
         keysManager.delete(account.getUuid()),
         keysManager.delete(account.getPhoneNumberIdentifier()));
 
+    final CompletableFuture<Void> deleteMessagesFuture = CompletableFuture.allOf(
+        messagesManager.clear(account.getUuid()),
+        messagesManager.clear(account.getPhoneNumberIdentifier()));
+
     profilesManager.deleteAll(account.getUuid());
-    messagesManager.clear(account.getUuid());
-    messagesManager.clear(account.getPhoneNumberIdentifier());
     registrationRecoveryPasswordsManager.removeForNumber(account.getNumber());
 
     deleteKeysFuture.join();
+    deleteMessagesFuture.join();
     deleteStorageServiceDataFuture.join();
     deleteBackupServiceDataFuture.join();
     deleteSecureValueRecoveryServiceDataFuture.join();

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

@@ -102,14 +102,16 @@ private Publisher<Envelope> getMessagesForDevice(UUID destinationUuid, long dest
         .tap(Micrometer.metrics(Metrics.globalRegistry));
   }
 
-  public void clear(UUID destinationUuid) {
-    messagesCache.clear(destinationUuid).join();
-    messagesDynamoDb.deleteAllMessagesForAccount(destinationUuid).join();
+  public CompletableFuture<Void> clear(UUID destinationUuid) {
+    return CompletableFuture.allOf(
+        messagesCache.clear(destinationUuid),
+        messagesDynamoDb.deleteAllMessagesForAccount(destinationUuid));
   }
 
-  public void clear(UUID destinationUuid, long deviceId) {
-    messagesCache.clear(destinationUuid, deviceId).join();
-    messagesDynamoDb.deleteAllMessagesForDevice(destinationUuid, deviceId).join();
+  public CompletableFuture<Void> clear(UUID destinationUuid, long deviceId) {
+    return CompletableFuture.allOf(
+        messagesCache.clear(destinationUuid, deviceId),
+        messagesDynamoDb.deleteAllMessagesForDevice(destinationUuid, deviceId));
   }
 
   public CompletableFuture<Optional<Envelope>> delete(UUID destinationUuid, long destinationDeviceId, UUID guid,

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

@@ -68,7 +68,7 @@ protected void run(final Environment environment, final Namespace namespace,
     deps.keysManager().delete(account.getUuid(), deviceId).join();
 
     System.out.format("Clearing additional messages for %s::%d\n", aci, deviceId);
-    deps.messagesManager().clear(account.getUuid(), deviceId);
+    deps.messagesManager().clear(account.getUuid(), deviceId).join();
 
     System.out.format("Clearing presence state for %s::%d\n", aci, deviceId);
     deps.clientPresenceManager().disconnectPresence(aci, deviceId);

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

@@ -128,6 +128,8 @@ void setup() {
 
     when(keysManager.storePqLastResort(any(), any())).thenReturn(CompletableFuture.completedFuture(null));
     when(keysManager.delete(any(), anyLong())).thenReturn(CompletableFuture.completedFuture(null));
+
+    when(messagesManager.clear(any(), anyLong())).thenReturn(CompletableFuture.completedFuture(null));
   }
 
   @AfterEach

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

@@ -103,14 +103,17 @@ void setup() throws InterruptedException {
       final KeysManager keysManager = mock(KeysManager.class);
       when(keysManager.delete(any())).thenReturn(CompletableFuture.completedFuture(null));
 
+      final MessagesManager messagesManager = mock(MessagesManager.class);
+      when(messagesManager.clear(any())).thenReturn(CompletableFuture.completedFuture(null));
+
       accountsManager = new AccountsManager(
           accounts,
           phoneNumberIdentifiers,
           CACHE_CLUSTER_EXTENSION.getRedisCluster(),
           accountLockManager,
           deletedAccounts,
           keysManager,
-          mock(MessagesManager.class),
+          messagesManager,
           mock(ProfilesManager.class),
           secureStorageClient,
           secureBackupClient,

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

@@ -188,6 +188,7 @@ void setup() throws InterruptedException {
     }).when(accountLockManager).withLock(any(), any());
 
     when(keysManager.delete(any())).thenReturn(CompletableFuture.completedFuture(null));
+    when(messagesManager.clear(any())).thenReturn(CompletableFuture.completedFuture(null));
 
     accountsManager = new AccountsManager(
         accounts,