Reconcile inactive and undiscoverable accounts when using v3 endpoints

service/src/main/java/org/whispersystems/textsecuregcm/sqs/DirectoryQueue.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2013-2021 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 package org.whispersystems.textsecuregcm.sqs;
@@ -95,12 +95,8 @@ public class DirectoryQueue implements Managed {
     sqs.close();
   }
 
-  public boolean isDiscoverable(final Account account) {
-    return account.isEnabled() && account.isDiscoverableByPhoneNumber();
-  }
-
   public void refreshAccount(final Account account) {
-    sendUpdateMessage(account, isDiscoverable(account) ? UpdateAction.ADD : UpdateAction.DELETE);
+    sendUpdateMessage(account, account.shouldBeVisibleInDirectory() ? UpdateAction.ADD : UpdateAction.DELETE);
   }
 
   public void deleteAccount(final Account account) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2013-2021 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 package org.whispersystems.textsecuregcm.storage;
@@ -327,7 +327,7 @@ public class Account implements Principal  {
 
     return new StoredRegistrationLock(Optional.ofNullable(registrationLock), Optional.ofNullable(registrationLockSalt), getLastSeen());
   }
-  
+
   public Optional<byte[]> getUnidentifiedAccessKey() {
     requireNotStale();
 
@@ -372,6 +372,12 @@ public class Account implements Principal  {
     this.discoverableByPhoneNumber = discoverableByPhoneNumber;
   }
 
+  public boolean shouldBeVisibleInDirectory() {
+    requireNotStale();
+
+    return isEnabled() && isDiscoverableByPhoneNumber();
+  }
+
   public int getVersion() {
     requireNotStale();
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2013-2021 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 package org.whispersystems.textsecuregcm.storage;
@@ -250,7 +250,7 @@ public class AccountsManager {
 
   public Account update(Account account, Consumer<Account> updater) {
 
-    final boolean wasDiscoverableBeforeUpdate = directoryQueue.isDiscoverable(account);
+    final boolean wasVisibleBeforeUpdate = account.shouldBeVisibleInDirectory();
 
     final Account updatedAccount;
 
@@ -293,9 +293,9 @@ public class AccountsManager {
       redisSet(updatedAccount);
     }
 
-    final boolean isDiscoverableAfterUpdate = directoryQueue.isDiscoverable(updatedAccount);
+    final boolean isVisibleAfterUpdate = updatedAccount.shouldBeVisibleInDirectory();
 
-    if (wasDiscoverableBeforeUpdate != isDiscoverableAfterUpdate) {
+    if (wasVisibleBeforeUpdate != isVisibleAfterUpdate) {
       directoryQueue.refreshAccount(updatedAccount);
     }
 

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

@@ -1,105 +1,134 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2013-2021 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 package org.whispersystems.textsecuregcm.storage;
 
 import static com.codahale.metrics.MetricRegistry.name;
 
-import com.codahale.metrics.Meter;
+import io.micrometer.core.instrument.Metrics;
-import com.codahale.metrics.MetricRegistry;
-import com.codahale.metrics.SharedMetricRegistries;
-import com.codahale.metrics.Timer;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.function.Function;
 import javax.ws.rs.ProcessingException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest;
 import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse;
-import org.whispersystems.textsecuregcm.util.Constants;
+import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse.Status;
 
 public class DirectoryReconciler extends AccountDatabaseCrawlerListener {
 
   private static final Logger logger = LoggerFactory.getLogger(DirectoryReconciler.class);
-  private static final MetricRegistry metricRegistry      = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
+  private static final String SEND_TIMER_NAME = name(DirectoryReconciler.class, "sendRequest");
 
+  private final String replicationName;
   private final DirectoryReconciliationClient reconciliationClient;
-  private final Timer                         sendChunkTimer;
-  private final Meter                         sendChunkErrorMeter;
 
   private boolean useV3Endpoints;
 
-  public DirectoryReconciler(String name, DirectoryReconciliationClient reconciliationClient) {
+  public DirectoryReconciler(String replicationName, DirectoryReconciliationClient reconciliationClient) {
     this.reconciliationClient = reconciliationClient;
-    sendChunkTimer            = metricRegistry.timer(name(DirectoryReconciler.class, name, "sendChunk"));
+    this.replicationName = replicationName;
-    sendChunkErrorMeter       = metricRegistry.meter(name(DirectoryReconciler.class, name, "sendChunkError"));
   }
 
   @Override
-  public void onCrawlStart() { }
+  public void onCrawlStart() {
+  }
 
   @Override
   public void onCrawlEnd(Optional<UUID> fromUuid) {
-    DirectoryReconciliationRequest  request  = new DirectoryReconciliationRequest(fromUuid.orElse(null), null, Collections.emptyList());
-    sendChunk(request);
 
     if (useV3Endpoints) {
       reconciliationClient.complete();
+    } else {
+      final DirectoryReconciliationRequest request = new DirectoryReconciliationRequest(fromUuid.orElse(null), null,
+          Collections.emptyList());
+      sendAdditions(request);
     }
   }
 
   @Override
-  protected void onCrawlChunk(Optional<UUID> fromUuid, List<Account> chunkAccounts) throws AccountDatabaseCrawlerRestartException {
+  protected void onCrawlChunk(final Optional<UUID> fromUuid, final List<Account> accounts)
-    DirectoryReconciliationRequest  request  = createChunkRequest(fromUuid, chunkAccounts);
+      throws AccountDatabaseCrawlerRestartException {
-    DirectoryReconciliationResponse response = sendChunk(request);
+
-    if (response.getStatus() == DirectoryReconciliationResponse.Status.MISSING) {
+    final DirectoryReconciliationRequest addUsersRequest;
+    final DirectoryReconciliationRequest deleteUsersRequest;
+    {
+      final List<DirectoryReconciliationRequest.User> addedUsers = new ArrayList<>(accounts.size());
+      final List<DirectoryReconciliationRequest.User> deletedUsers = new ArrayList<>(accounts.size());
+
+      accounts.forEach(account -> {
+        if (account.shouldBeVisibleInDirectory()) {
+          addedUsers.add(new DirectoryReconciliationRequest.User(account.getUuid(), account.getNumber()));
+        } else {
+          deletedUsers.add(new DirectoryReconciliationRequest.User(account.getUuid(), account.getNumber()));
+        }
+      });
+
+      final Optional<UUID> toUuid;
+      if (!accounts.isEmpty()) {
+        toUuid = Optional.of(accounts.get(accounts.size() - 1).getUuid());
+      } else {
+        toUuid = Optional.empty();
+      }
+
+      addUsersRequest = new DirectoryReconciliationRequest(fromUuid.orElse(null), toUuid.orElse(null), addedUsers);
+      deleteUsersRequest = new DirectoryReconciliationRequest(null, null, deletedUsers);
+    }
+
+    final DirectoryReconciliationResponse addUsersResponse = sendAdditions(addUsersRequest);
+    final DirectoryReconciliationResponse deleteUsersResponse = sendDeletes(deleteUsersRequest);
+
+    if (addUsersResponse.getStatus() == DirectoryReconciliationResponse.Status.MISSING
+        || deleteUsersResponse.getStatus() == Status.MISSING) {
+
       throw new AccountDatabaseCrawlerRestartException("directory reconciler missing");
     }
   }
 
-  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+  private DirectoryReconciliationResponse sendDeletes(final DirectoryReconciliationRequest request) {
-  private DirectoryReconciliationRequest createChunkRequest(Optional<UUID> fromUuid, List<Account> accounts) {
+    if (useV3Endpoints) {
-    List<DirectoryReconciliationRequest.User> users = new ArrayList<>(accounts.size());
+      return sendRequest(request, reconciliationClient::delete, "delete");
-    for (Account account : accounts) {
-      if (account.isEnabled() && account.isDiscoverableByPhoneNumber()) {
-        users.add(new DirectoryReconciliationRequest.User(account.getUuid(), account.getNumber()));
-      }
     }
 
-    Optional<UUID> toUuid = Optional.empty();
+    return new DirectoryReconciliationResponse(DirectoryReconciliationResponse.Status.OK);
-
-    if (!accounts.isEmpty()) {
-      toUuid = Optional.of(accounts.get(accounts.size() - 1).getUuid());
-    }
-
-    return new DirectoryReconciliationRequest(fromUuid.orElse(null), toUuid.orElse(null), users);
   }
 
-  private DirectoryReconciliationResponse sendChunk(DirectoryReconciliationRequest request) {
+  private DirectoryReconciliationResponse sendAdditions(final DirectoryReconciliationRequest request) {
-    try (Timer.Context timer = sendChunkTimer.time()) {
+
-      DirectoryReconciliationResponse response;
+    if (useV3Endpoints) {
-      if (useV3Endpoints) {
+      return sendRequest(request, reconciliationClient::sendChunkV3, "add");
-        response = reconciliationClient.sendChunkV3(request);
-      } else {
-        response = reconciliationClient.sendChunk(request);
-      }
-      if (response.getStatus() != DirectoryReconciliationResponse.Status.OK) {
-        sendChunkErrorMeter.mark();
-        logger.warn("reconciliation error: " + response.getStatus());
-      }
-      return response;
-    } catch (ProcessingException ex) {
-      sendChunkErrorMeter.mark();
-      logger.warn("request error: ", ex);
-      throw new ProcessingException(ex);
     }
+
+    return sendRequest(request, reconciliationClient::sendChunk, "add_v2");
+  }
+
+  private DirectoryReconciliationResponse sendRequest(final DirectoryReconciliationRequest request,
+      final Function<DirectoryReconciliationRequest, DirectoryReconciliationResponse> requestHandler,
+      final String context) {
+
+    return Metrics.timer(SEND_TIMER_NAME, "context", context, "replication", replicationName)
+        .record(() -> {
+          try {
+            final DirectoryReconciliationResponse response = requestHandler.apply(request);
+
+            if (response.getStatus() != DirectoryReconciliationResponse.Status.OK) {
+              logger.warn("reconciliation error: " + response.getStatus());
+            }
+            return response;
+          } catch (ProcessingException ex) {
+            logger.warn("request error: ", ex);
+            throw new ProcessingException(ex);
+          }
+        });
   }
 
   public void setUseV3Endpoints(final boolean useV3Endpoints) {
     this.useV3Endpoints = useV3Endpoints;
   }
+
 }

service/src/test/java/org/whispersystems/textsecuregcm/sqs/DirectoryQueueTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2013-2021 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
@@ -46,14 +46,13 @@ public class DirectoryQueueTest {
 
   @ParameterizedTest
   @MethodSource("argumentsForTestRefreshRegisteredUser")
-  void testRefreshRegisteredUser(final boolean accountEnabled, final boolean accountDiscoverableByPhoneNumber, final String expectedAction) {
+  void testRefreshRegisteredUser(final boolean shouldBeVisibleInDirectory, final String expectedAction) {
     final DirectoryQueue directoryQueue = new DirectoryQueue(List.of("sqs://test"), sqsAsyncClient);
 
     final Account account = mock(Account.class);
     when(account.getNumber()).thenReturn("+18005556543");
     when(account.getUuid()).thenReturn(UUID.randomUUID());
-    when(account.isEnabled()).thenReturn(accountEnabled);
+    when(account.shouldBeVisibleInDirectory()).thenReturn(shouldBeVisibleInDirectory);
-    when(account.isDiscoverableByPhoneNumber()).thenReturn(accountDiscoverableByPhoneNumber);
 
     directoryQueue.refreshAccount(account);
 
@@ -67,10 +66,8 @@ public class DirectoryQueueTest {
   @SuppressWarnings("unused")
   private static Stream<Arguments> argumentsForTestRefreshRegisteredUser() {
     return Stream.of(
-        Arguments.of(true, true, "add"),
+        Arguments.of(true, "add"),
-        Arguments.of(true, false, "delete"),
+        Arguments.of(false, "delete"));
-        Arguments.of(false, true, "delete"),
-        Arguments.of(false, false, "delete"));
   }
 
   @Test
@@ -80,8 +77,7 @@ public class DirectoryQueueTest {
     final Account account = mock(Account.class);
     when(account.getNumber()).thenReturn("+18005556543");
     when(account.getUuid()).thenReturn(UUID.randomUUID());
-    when(account.isEnabled()).thenReturn(true);
+    when(account.shouldBeVisibleInDirectory()).thenReturn(true);
-    when(account.isDiscoverableByPhoneNumber()).thenReturn(true);
 
     directoryQueue.refreshAccount(account);
 
@@ -104,8 +100,7 @@ public class DirectoryQueueTest {
     final Account account = mock(Account.class);
     when(account.getNumber()).thenReturn("+18005556543");
     when(account.getUuid()).thenReturn(UUID.randomUUID());
-    when(account.isEnabled()).thenReturn(true);
+    when(account.shouldBeVisibleInDirectory()).thenReturn(true);
-    when(account.isDiscoverableByPhoneNumber()).thenReturn(true);
 
     directoryQueue.refreshAccount(account);
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2013-2021 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
@@ -617,14 +617,19 @@ class AccountsManagerTest {
 
   @ParameterizedTest
   @MethodSource
-  void testUpdateDirectoryQueue(final boolean discoverableBeforeUpdate, final boolean discoverableAfterUpdate, final boolean expectRefresh) {
+  void testUpdateDirectoryQueue(final boolean visibleBeforeUpdate, final boolean visibleAfterUpdate,
+      final boolean expectRefresh) {
     final Account account = new Account("+14152222222", UUID.randomUUID(), new HashSet<>(), new byte[16]);
 
-    when(directoryQueue.isDiscoverable(any()))
+    // this sets up the appropriate result for Account#shouldBeVisibleInDirectory
-        .thenReturn(discoverableBeforeUpdate)
+    final Device device = new Device(Device.MASTER_ID, "device", "token", "salt", null, null, null, true, 1,
-        .thenReturn(discoverableAfterUpdate);
+        new SignedPreKey(1, "key", "sig"), 0, 0,
+        "OWT", 0, new DeviceCapabilities());
+    account.addDevice(device);
+    account.setDiscoverableByPhoneNumber(visibleBeforeUpdate);
 
-    final Account updatedAccount = accountsManager.update(account, a -> a.setProfileName("Hello I am a unit test"));
+    final Account updatedAccount = accountsManager.update(account,
+        a -> a.setDiscoverableByPhoneNumber(visibleAfterUpdate));
 
     verify(directoryQueue, times(expectRefresh ? 1 : 0)).refreshAccount(updatedAccount);
   }

service/src/test/java/org/whispersystems/textsecuregcm/tests/storage/DirectoryReconcilerTest.java

@@ -1,71 +1,92 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2013-2021 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
 package org.whispersystems.textsecuregcm.tests.storage;
 
-import org.junit.Before;
+import static org.assertj.core.api.Assertions.assertThat;
-import org.junit.Test;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.ArgumentCaptor;
 import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest;
+import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest.User;
 import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountDatabaseCrawlerRestartException;
 import org.whispersystems.textsecuregcm.storage.DirectoryReconciler;
 import org.whispersystems.textsecuregcm.storage.DirectoryReconciliationClient;
 
-import java.util.Arrays;
+class DirectoryReconcilerTest {
-import java.util.Optional;
-import java.util.UUID;
 
-import static org.assertj.core.api.Assertions.assertThat;
+  private static final UUID VALID_UUID = UUID.randomUUID();
-import static org.mockito.ArgumentMatchers.any;
+  private static final String VALID_NUMBER = "+14152222222";
-import static org.mockito.Mockito.*;
+  private static final UUID UNDISCOVERABLE_UUID = UUID.randomUUID();
-
-public class DirectoryReconcilerTest {
-  private static final UUID   VALID_UUID            = UUID.randomUUID();
-  private static final String VALID_NUMBERRR        = "+14152222222";
-  private static final UUID   INACTIVE_UUID         = UUID.randomUUID();
-  private static final String INACTIVE_NUMBERRR     = "+14151111111";
-  private static final UUID   UNDISCOVERABLE_UUID   = UUID.randomUUID();
   private static final String UNDISCOVERABLE_NUMBER = "+14153333333";
 
-  private final Account                       activeAccount         = mock(Account.class);
+  private final Account visibleAccount = mock(Account.class);
-  private final Account                       inactiveAccount       = mock(Account.class);
+  private final Account undiscoverableAccount = mock(Account.class);
-  private final Account                       undiscoverableAccount = mock(Account.class);
+  private final DirectoryReconciliationClient reconciliationClient = mock(DirectoryReconciliationClient.class);
-  private final DirectoryReconciliationClient reconciliationClient  = mock(DirectoryReconciliationClient.class);
+  private final DirectoryReconciler directoryReconciler = new DirectoryReconciler("test", reconciliationClient);
-  private final DirectoryReconciler           directoryReconciler   = new DirectoryReconciler("test", reconciliationClient);
 
   private final DirectoryReconciliationResponse successResponse = new DirectoryReconciliationResponse(DirectoryReconciliationResponse.Status.OK);
 
-  @Before
+  @BeforeEach
-  public void setup() {
+  void setup() {
-    when(activeAccount.getUuid()).thenReturn(VALID_UUID);
+    when(visibleAccount.getUuid()).thenReturn(VALID_UUID);
-    when(activeAccount.isEnabled()).thenReturn(true);
+    when(visibleAccount.getNumber()).thenReturn(VALID_NUMBER);
-    when(activeAccount.getNumber()).thenReturn(VALID_NUMBERRR);
+    when(visibleAccount.shouldBeVisibleInDirectory()).thenReturn(true);
-    when(activeAccount.isDiscoverableByPhoneNumber()).thenReturn(true);
-    when(inactiveAccount.getUuid()).thenReturn(INACTIVE_UUID);
-    when(inactiveAccount.getNumber()).thenReturn(INACTIVE_NUMBERRR);
-    when(inactiveAccount.isEnabled()).thenReturn(false);
-    when(inactiveAccount.isDiscoverableByPhoneNumber()).thenReturn(true);
     when(undiscoverableAccount.getUuid()).thenReturn(UNDISCOVERABLE_UUID);
     when(undiscoverableAccount.getNumber()).thenReturn(UNDISCOVERABLE_NUMBER);
-    when(undiscoverableAccount.isEnabled()).thenReturn(true);
+    when(undiscoverableAccount.shouldBeVisibleInDirectory()).thenReturn(false);
-    when(undiscoverableAccount.isDiscoverableByPhoneNumber()).thenReturn(false);
   }
 
-  @Test
+  @ParameterizedTest
-  public void testCrawlChunkValid() throws AccountDatabaseCrawlerRestartException {
+  @ValueSource(booleans = {true, false})
+  public void testCrawlChunkValid(final boolean useV3Endpoints) throws AccountDatabaseCrawlerRestartException {
+    directoryReconciler.setUseV3Endpoints(useV3Endpoints);
+
     when(reconciliationClient.sendChunk(any())).thenReturn(successResponse);
-    directoryReconciler.timeAndProcessCrawlChunk(Optional.of(VALID_UUID), Arrays.asList(activeAccount, inactiveAccount, undiscoverableAccount));
+    when(reconciliationClient.sendChunkV3(any())).thenReturn(successResponse);
+    when(reconciliationClient.delete(any())).thenReturn(successResponse);
 
-    ArgumentCaptor<DirectoryReconciliationRequest> request = ArgumentCaptor.forClass(DirectoryReconciliationRequest.class);
+    directoryReconciler.timeAndProcessCrawlChunk(Optional.of(VALID_UUID),
-    verify(reconciliationClient, times(1)).sendChunk(request.capture());
+        Arrays.asList(visibleAccount, undiscoverableAccount));
 
-    assertThat(request.getValue().getFromUuid()).isEqualTo(VALID_UUID);
+    ArgumentCaptor<DirectoryReconciliationRequest> chunkRequest = ArgumentCaptor.forClass(
-    assertThat(request.getValue().getToUuid()).isEqualTo(UNDISCOVERABLE_UUID);
+        DirectoryReconciliationRequest.class);
-    assertThat(request.getValue().getUsers()).isEqualTo(Arrays.asList(new DirectoryReconciliationRequest.User(VALID_UUID, VALID_NUMBERRR)));
+    if (useV3Endpoints) {
+      verify(reconciliationClient, times(1)).sendChunkV3(chunkRequest.capture());
+    } else {
+      verify(reconciliationClient, times(1)).sendChunk(chunkRequest.capture());
+    }
+
+    assertThat(chunkRequest.getValue().getFromUuid()).isEqualTo(VALID_UUID);
+    assertThat(chunkRequest.getValue().getToUuid()).isEqualTo(UNDISCOVERABLE_UUID);
+    assertThat(chunkRequest.getValue().getUsers()).isEqualTo(List.of(new User(VALID_UUID, VALID_NUMBER)));
+
+    if (useV3Endpoints) {
+
+      ArgumentCaptor<DirectoryReconciliationRequest> deletesRequest = ArgumentCaptor.forClass(DirectoryReconciliationRequest.class);
+      verify(reconciliationClient, times(1)).delete(deletesRequest.capture());
+
+      assertThat(deletesRequest.getValue().getUsers()).isEqualTo(
+          List.of(new User(UNDISCOVERABLE_UUID, UNDISCOVERABLE_NUMBER)));
+    }
+
+    verifyNoMoreInteractions(reconciliationClient);
   }
 
 }