Add dynamic configuration for using DynamoDB in AccountsDatabaseCrawler

2026-04-20 14:08:01 +01:00 · 2021-07-06 13:01:24 -05:00
parent 78819d5382
commit a824b5575d
14 changed files with 434 additions and 130 deletions
--- a/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
@@ -482,7 +482,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
    CurrencyConversionManager currencyManager = new CurrencyConversionManager(fixerClient, ftxClient, config.getPaymentsServiceConfiguration().getPaymentCurrencies());

    AccountDatabaseCrawlerCache accountDatabaseCrawlerCache = new AccountDatabaseCrawlerCache(cacheCluster);
-    AccountDatabaseCrawler      accountDatabaseCrawler      = new AccountDatabaseCrawler(accountsManager, accountDatabaseCrawlerCache, accountDatabaseCrawlerListeners, config.getAccountDatabaseCrawlerConfiguration().getChunkSize(), config.getAccountDatabaseCrawlerConfiguration().getChunkIntervalMs());
+    AccountDatabaseCrawler      accountDatabaseCrawler      = new AccountDatabaseCrawler(accountsManager, accountDatabaseCrawlerCache, accountDatabaseCrawlerListeners, config.getAccountDatabaseCrawlerConfiguration().getChunkSize(), config.getAccountDatabaseCrawlerConfiguration().getChunkIntervalMs(), dynamicConfigurationManager);

    DeletedAccountsTableCrawler deletedAccountsTableCrawler = new DeletedAccountsTableCrawler(deletedAccounts, deletedAccountsDirectoryReconcilers, cacheCluster, recurringJobExecutor);
    MigrationRetryAccountsTableCrawler migrationRetryAccountsTableCrawler = new MigrationRetryAccountsTableCrawler(migrationRetryAccounts, accountsManager, accountsDynamoDb, cacheCluster, recurringJobExecutor);
--- a/service/src/main/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicAccountsDynamoDbMigrationConfiguration.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicAccountsDynamoDbMigrationConfiguration.java
@@ -23,6 +23,12 @@ public class DynamicAccountsDynamoDbMigrationConfiguration {
  @JsonProperty
  boolean logMismatches;

+  @JsonProperty
+  boolean dynamoCrawlerEnabled;
+
+  @JsonProperty
+  int dynamoCrawlerScanPageSize = 10;
+
  public boolean isBackgroundMigrationEnabled() {
    return backgroundMigrationEnabled;
  }
@@ -59,4 +65,12 @@ public class DynamicAccountsDynamoDbMigrationConfiguration {
  public boolean isLogMismatches() {
    return logMismatches;
  }
+
+  public boolean isDynamoCrawlerEnabled() {
+    return dynamoCrawlerEnabled;
+  }
+
+  public int getDynamoCrawlerScanPageSize() {
+    return dynamoCrawlerScanPageSize;
+  }
 }
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountCrawlChunk.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountCrawlChunk.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2013-2021 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+package org.whispersystems.textsecuregcm.storage;
+
+import javax.annotation.Nullable;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+
+public class AccountCrawlChunk {
+
+  private final List<Account> accounts;
+  @Nullable
+  private final UUID lastUuid;
+
+  public AccountCrawlChunk(final List<Account> accounts, @Nullable final UUID lastUuid) {
+    this.accounts = accounts;
+    this.lastUuid = lastUuid;
+  }
+
+  public List<Account> getAccounts() {
+    return accounts;
+  }
+
+  public Optional<UUID> getLastUuid() {
+    return Optional.ofNullable(lastUuid);
+  }
+}
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawler.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawler.java
@@ -4,22 +4,21 @@
 */
 package org.whispersystems.textsecuregcm.storage;

+import static com.codahale.metrics.MetricRegistry.name;
+
 import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.SharedMetricRegistries;
 import com.codahale.metrics.Timer;
 import com.google.common.annotations.VisibleForTesting;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.whispersystems.textsecuregcm.util.Constants;
-import org.whispersystems.textsecuregcm.util.Util;
-
+import io.dropwizard.lifecycle.Managed;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
-
-import static com.codahale.metrics.MetricRegistry.name;
-import io.dropwizard.lifecycle.Managed;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.util.Constants;
+import org.whispersystems.textsecuregcm.util.Util;

@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
 public class AccountDatabaseCrawler implements Managed, Runnable {
@@ -38,6 +37,8 @@ public class AccountDatabaseCrawler implements Managed, Runnable {
  private final AccountDatabaseCrawlerCache          cache;
  private final List<AccountDatabaseCrawlerListener> listeners;

+  private final DynamicConfigurationManager dynamicConfigurationManager;
+
  private AtomicBoolean running = new AtomicBoolean(false);
  private boolean finished;

@@ -45,7 +46,8 @@ public class AccountDatabaseCrawler implements Managed, Runnable {
                                AccountDatabaseCrawlerCache cache,
                                List<AccountDatabaseCrawlerListener> listeners,
                                int chunkSize,
-                                long chunkIntervalMs)
+                                long chunkIntervalMs,
+                                DynamicConfigurationManager dynamicConfigurationManager)
  {
    this.accounts             = accounts;
    this.chunkSize            = chunkSize;
@@ -53,6 +55,8 @@ public class AccountDatabaseCrawler implements Managed, Runnable {
    this.workerId             = UUID.randomUUID().toString();
    this.cache                = cache;
    this.listeners            = listeners;
+
+    this.dynamicConfigurationManager = dynamicConfigurationManager;
  }

  @Override
@@ -93,14 +97,15 @@ public class AccountDatabaseCrawler implements Managed, Runnable {
  @VisibleForTesting
  public boolean doPeriodicWork() {
    if (cache.claimActiveWork(workerId, WORKER_TTL_MS)) {
+
      try {
-        long startTimeMs = System.currentTimeMillis();
+        final long startTimeMs = System.currentTimeMillis();
        processChunk();
        if (cache.isAccelerated()) {
          return true;
        }
-        long endTimeMs = System.currentTimeMillis();
-        long sleepIntervalMs = chunkIntervalMs - (endTimeMs - startTimeMs);
+        final long endTimeMs = System.currentTimeMillis();
+        final long sleepIntervalMs = chunkIntervalMs - (endTimeMs - startTimeMs);
        if (sleepIntervalMs > 0) sleepWhileRunning(sleepIntervalMs);
      } finally {
        cache.releaseActiveWork(workerId);
@@ -110,42 +115,67 @@ public class AccountDatabaseCrawler implements Managed, Runnable {
  }

  private void processChunk() {
-    Optional<UUID> fromUuid = cache.getLastUuid();
+    final boolean useDynamo = dynamicConfigurationManager.getConfiguration()
+        .getAccountsDynamoDbMigrationConfiguration()
+        .isDynamoCrawlerEnabled();

-    if (!fromUuid.isPresent()) {
+    listeners.stream().filter(listener -> listener instanceof DirectoryReconciler)
+        .forEach(reconciler -> ((DirectoryReconciler) reconciler).setUseV3Endpoints(useDynamo));
+
+    final Optional<UUID> fromUuid = getLastUuid(useDynamo);
+
+    if (fromUuid.isEmpty()) {
      listeners.forEach(AccountDatabaseCrawlerListener::onCrawlStart);
    }

-    List<Account> chunkAccounts = readChunk(fromUuid, chunkSize);
+    final AccountCrawlChunk chunkAccounts = readChunk(fromUuid, chunkSize, useDynamo);

-    if (chunkAccounts.isEmpty()) {
+    if (chunkAccounts.getAccounts().isEmpty()) {
      logger.info("Finished crawl");
      listeners.forEach(listener -> listener.onCrawlEnd(fromUuid));
-      cache.setLastUuid(Optional.empty());
+      cacheLastUuid(Optional.empty(), useDynamo);
      cache.setAccelerated(false);
    } else {
      try {
        for (AccountDatabaseCrawlerListener listener : listeners) {
-          listener.timeAndProcessCrawlChunk(fromUuid, chunkAccounts);
+          listener.timeAndProcessCrawlChunk(fromUuid, chunkAccounts.getAccounts());
        }
-        cache.setLastUuid(Optional.of(chunkAccounts.get(chunkAccounts.size() - 1).getUuid()));
+        cacheLastUuid(chunkAccounts.getLastUuid(), useDynamo);
      } catch (AccountDatabaseCrawlerRestartException e) {
-        cache.setLastUuid(Optional.empty());
+        cacheLastUuid(Optional.empty(), useDynamo);
        cache.setAccelerated(false);
      }
-
    }
-
  }

-  private List<Account> readChunk(Optional<UUID> fromUuid, int chunkSize) {
+  private AccountCrawlChunk readChunk(Optional<UUID> fromUuid, int chunkSize, boolean useDynamo) {
    try (Timer.Context timer = readChunkTimer.time()) {

      if (fromUuid.isPresent()) {
-        return accounts.getAllFrom(fromUuid.get(), chunkSize);
+        return useDynamo
+            ? accounts.getAllFromDynamo(fromUuid.get(), chunkSize)
+            : accounts.getAllFrom(fromUuid.get(), chunkSize);
      }

-      return accounts.getAllFrom(chunkSize);
+      return useDynamo
+          ? accounts.getAllFromDynamo(chunkSize)
+          : accounts.getAllFrom(chunkSize);
+    }
+  }
+
+  private Optional<UUID> getLastUuid(final boolean useDynamo) {
+    if (useDynamo) {
+      return cache.getLastUuidDynamo();
+    } else {
+      return cache.getLastUuid();
+    }
+  }
+
+  private void cacheLastUuid(final Optional<UUID> lastUuid, final boolean useDynamo) {
+    if (useDynamo) {
+      cache.setLastUuidDynamo(lastUuid);
+    } else {
+      cache.setLastUuid(lastUuid);
    }
  }

--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawlerCache.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawlerCache.java
@@ -21,6 +21,8 @@ public class AccountDatabaseCrawlerCache {
  private static final String LAST_UUID_KEY     = "account_database_crawler_cache_last_uuid";
  private static final String ACCELERATE_KEY    = "account_database_crawler_cache_accelerate";

+  private static final String LAST_UUID_DYNAMO_KEY  = "account_database_crawler_cache_last_uuid_dynamo";
+
  private static final long LAST_NUMBER_TTL_MS  = 86400_000L;

  private final FaultTolerantRedisCluster cacheCluster;
@@ -66,4 +68,19 @@ public class AccountDatabaseCrawlerCache {
    }
  }

+  public Optional<UUID> getLastUuidDynamo() {
+    final String lastUuidString = cacheCluster.withCluster(connection -> connection.sync().get(LAST_UUID_DYNAMO_KEY));
+
+    if (lastUuidString == null) return Optional.empty();
+    else                        return Optional.of(UUID.fromString(lastUuidString));
+  }
+
+  public void setLastUuidDynamo(Optional<UUID> lastUuid) {
+    if (lastUuid.isPresent()) {
+      cacheCluster.useCluster(connection -> connection.sync().psetex(LAST_UUID_DYNAMO_KEY, LAST_NUMBER_TTL_MS, lastUuid.get().toString()));
+    } else {
+      cacheCluster.useCluster(connection -> connection.sync().del(LAST_UUID_DYNAMO_KEY));
+    }
+  }
+
 }
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/Accounts.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/Accounts.java
@@ -9,6 +9,7 @@ import static com.codahale.metrics.MetricRegistry.name;
 import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.SharedMetricRegistries;
 import com.codahale.metrics.Timer;
+import com.codahale.metrics.Timer.Context;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.List;
@@ -103,20 +104,21 @@ public class Accounts implements AccountStore {
    }));
  }

-  public List<Account> getAllFrom(UUID from, int length) {
-    return database.with(jdbi -> jdbi.withHandle(handle -> {
-      try (Timer.Context ignored = getAllFromOffsetTimer.time()) {
+  public AccountCrawlChunk getAllFrom(UUID from, int length) {
+    final List<Account> accounts = database.with(jdbi -> jdbi.withHandle(handle -> {
+      try (Context ignored = getAllFromOffsetTimer.time()) {
        return handle.createQuery("SELECT * FROM accounts WHERE " + UID + " > :from ORDER BY " + UID + " LIMIT :limit")
-                     .bind("from", from)
-                     .bind("limit", length)
-                     .mapTo(Account.class)
-                     .list();
+            .bind("from", from)
+            .bind("limit", length)
+            .mapTo(Account.class)
+            .list();
      }
    }));
+    return buildChunkForAccounts(accounts);
  }

-  public List<Account> getAllFrom(int length) {
-    return database.with(jdbi -> jdbi.withHandle(handle -> {
+  public AccountCrawlChunk getAllFrom(int length) {
+    final List<Account> accounts = database.with(jdbi -> jdbi.withHandle(handle -> {
      try (Timer.Context ignored = getAllFromTimer.time()) {
        return handle.createQuery("SELECT * FROM accounts ORDER BY " + UID + " LIMIT :limit")
                     .bind("limit", length)
@@ -124,6 +126,12 @@ public class Accounts implements AccountStore {
                     .list();
      }
    }));
+
+    return buildChunkForAccounts(accounts);
+  }
+
+  private AccountCrawlChunk buildChunkForAccounts(final List<Account> accounts) {
+    return new AccountCrawlChunk(accounts, accounts.isEmpty() ? null : accounts.get(accounts.size() - 1).getUuid());
  }

  @Override
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsDynamoDb.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsDynamoDb.java
@@ -1,3 +1,7 @@
+/*
+ * 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;
@@ -31,6 +35,7 @@ import software.amazon.awssdk.services.dynamodb.model.GetItemRequest;
 import software.amazon.awssdk.services.dynamodb.model.GetItemResponse;
 import software.amazon.awssdk.services.dynamodb.model.Put;
 import software.amazon.awssdk.services.dynamodb.model.ReturnValuesOnConditionCheckFailure;
+import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
 import software.amazon.awssdk.services.dynamodb.model.TransactWriteItem;
 import software.amazon.awssdk.services.dynamodb.model.TransactWriteItemsRequest;
 import software.amazon.awssdk.services.dynamodb.model.TransactionCanceledException;
@@ -62,6 +67,8 @@ public class AccountsDynamoDb extends AbstractDynamoDbStore implements AccountSt
  private static final Timer UPDATE_TIMER = Metrics.timer(name(AccountsDynamoDb.class, "update"));
  private static final Timer GET_BY_NUMBER_TIMER = Metrics.timer(name(AccountsDynamoDb.class, "getByNumber"));
  private static final Timer GET_BY_UUID_TIMER = Metrics.timer(name(AccountsDynamoDb.class, "getByUuid"));
+  private static final Timer GET_ALL_FROM_START_TIMER = Metrics.timer(name(AccountsDynamoDb.class, "getAllFrom"));
+  private static final Timer GET_ALL_FROM_OFFSET_TIMER = Metrics.timer(name(AccountsDynamoDb.class, "getAllFromOffset"));
  private static final Timer DELETE_TIMER = Metrics.timer(name(AccountsDynamoDb.class, "delete"));

  private final Logger logger = LoggerFactory.getLogger(AccountsDynamoDb.class);
@@ -230,6 +237,33 @@ public class AccountsDynamoDb extends AbstractDynamoDbStore implements AccountSt
    });
  }

+  public AccountCrawlChunk getAllFrom(final UUID from, final int maxCount, final int pageSize) {
+    final ScanRequest.Builder scanRequestBuilder = ScanRequest.builder()
+        .limit(pageSize)
+        .exclusiveStartKey(Map.of(KEY_ACCOUNT_UUID, AttributeValues.fromUUID(from)));
+
+    return scanForChunk(scanRequestBuilder, maxCount, GET_ALL_FROM_OFFSET_TIMER);
+  }
+
+  public AccountCrawlChunk getAllFromStart(final int maxCount, final int pageSize) {
+    final ScanRequest.Builder scanRequestBuilder = ScanRequest.builder()
+        .limit(pageSize);
+
+    return scanForChunk(scanRequestBuilder, maxCount, GET_ALL_FROM_START_TIMER);
+  }
+
+  private AccountCrawlChunk scanForChunk(final ScanRequest.Builder scanRequestBuilder, final int maxCount, final Timer timer) {
+
+    scanRequestBuilder.tableName(accountsTableName);
+
+    final List<Account> accounts = timer.record(() -> scan(scanRequestBuilder.build(), maxCount)
+        .stream()
+        .map(AccountsDynamoDb::fromItem)
+        .collect(Collectors.toList()));
+
+    return new AccountCrawlChunk(accounts, accounts.size() > 0 ? accounts.get(accounts.size() - 1).getUuid() : null);
+  }
+
  private void delete(UUID uuid, boolean saveInDeletedAccountsTable) {

    if (saveInDeletedAccountsTable) {
@@ -332,7 +366,7 @@ public class AccountsDynamoDb extends AbstractDynamoDbStore implements AccountSt
          return;
        }
        try {
-          migrationRetryAccounts.put(account.getUuid()); 
+          migrationRetryAccounts.put(account.getUuid());
        } catch (final Exception e) {
          logger.error("Could not store account {}", account.getUuid());
        }
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
@@ -19,7 +19,6 @@ import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Metrics;
 import java.io.IOException;
 import java.util.Arrays;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -234,14 +233,26 @@ public class AccountsManager {
  }


-  public List<Account> getAllFrom(int length) {
+  public AccountCrawlChunk getAllFrom(int length) {
    return accounts.getAllFrom(length);
  }

-  public List<Account> getAllFrom(UUID uuid, int length) {
+  public AccountCrawlChunk getAllFrom(UUID uuid, int length) {
    return accounts.getAllFrom(uuid, length);
  }

+  public AccountCrawlChunk getAllFromDynamo(int length) {
+    final int maxPageSize = dynamicConfigurationManager.getConfiguration().getAccountsDynamoDbMigrationConfiguration()
+        .getDynamoCrawlerScanPageSize();
+    return accountsDynamoDb.getAllFromStart(length, maxPageSize);
+  }
+
+  public AccountCrawlChunk getAllFromDynamo(UUID uuid, int length) {
+    final int maxPageSize = dynamicConfigurationManager.getConfiguration().getAccountsDynamoDbMigrationConfiguration()
+        .getDynamoCrawlerScanPageSize();
+    return accountsDynamoDb.getAllFrom(uuid, length, maxPageSize);
+  }
+
  public void delete(final Account account, final DeletionReason deletionReason) {
    try (final Timer.Context ignored = deleteTimer.time()) {
      final CompletableFuture<Void> deleteStorageServiceDataFuture = secureStorageClient.deleteStoredData(account.getUuid());
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciler.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciler.java
@@ -4,24 +4,23 @@
 */
 package org.whispersystems.textsecuregcm.storage;

+import static com.codahale.metrics.MetricRegistry.name;
+
 import com.codahale.metrics.Meter;
 import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.SharedMetricRegistries;
 import com.codahale.metrics.Timer;
-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 javax.ws.rs.ProcessingException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
-
-import static com.codahale.metrics.MetricRegistry.name;
+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;

 public class DirectoryReconciler extends AccountDatabaseCrawlerListener {

@@ -32,6 +31,8 @@ public class DirectoryReconciler extends AccountDatabaseCrawlerListener {
  private final Timer                         sendChunkTimer;
  private final Meter                         sendChunkErrorMeter;

+  private boolean useV3Endpoints;
+
  public DirectoryReconciler(String name, DirectoryReconciliationClient reconciliationClient) {
    this.reconciliationClient = reconciliationClient;
    sendChunkTimer            = metricRegistry.timer(name(DirectoryReconciler.class, name, "sendChunk"));
@@ -45,6 +46,10 @@ public class DirectoryReconciler extends AccountDatabaseCrawlerListener {
  public void onCrawlEnd(Optional<UUID> fromUuid) {
    DirectoryReconciliationRequest  request  = new DirectoryReconciliationRequest(fromUuid.orElse(null), null, Collections.emptyList());
    sendChunk(request);
+
+    if (useV3Endpoints) {
+      reconciliationClient.complete();
+    }
  }

  @Override
@@ -76,7 +81,12 @@ public class DirectoryReconciler extends AccountDatabaseCrawlerListener {

  private DirectoryReconciliationResponse sendChunk(DirectoryReconciliationRequest request) {
    try (Timer.Context timer = sendChunkTimer.time()) {
-      DirectoryReconciliationResponse response = reconciliationClient.sendChunk(request);
+      DirectoryReconciliationResponse response;
+      if (useV3Endpoints) {
+        response = reconciliationClient.sendChunkV3(request);
+      } else {
+        response = reconciliationClient.sendChunk(request);
+      }
      if (response.getStatus() != DirectoryReconciliationResponse.Status.OK) {
        sendChunkErrorMeter.mark();
        logger.warn("reconciliation error: " + response.getStatus());
@@ -89,4 +99,7 @@ public class DirectoryReconciler extends AccountDatabaseCrawlerListener {
    }
  }

+  public void setUseV3Endpoints(final boolean useV3Endpoints) {
+    this.useV3Endpoints = useV3Endpoints;
+  }
 }
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciliationClient.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciliationClient.java
@@ -46,6 +46,13 @@ public class DirectoryReconciliationClient {
                 .put(Entity.json(request), DirectoryReconciliationResponse.class);
  }

+  public DirectoryReconciliationResponse sendChunkV3(DirectoryReconciliationRequest request) {
+    return client.target(replicationUrl)
+                 .path("/v3/directory/exists")
+                 .request(MediaType.APPLICATION_JSON_TYPE)
+                 .put(Entity.json(request), DirectoryReconciliationResponse.class);
+  }
+
  public DirectoryReconciliationResponse delete(DirectoryReconciliationRequest request) {
    return client.target(replicationUrl)
        .path("/v3/directory/deletes")
@@ -53,6 +60,13 @@ public class DirectoryReconciliationClient {
        .put(Entity.json(request), DirectoryReconciliationResponse.class);
  }

+  public DirectoryReconciliationResponse complete() {
+    return client.target(replicationUrl)
+        .path("/v3/directory/complete")
+        .request(MediaType.APPLICATION_JSON_TYPE)
+        .post(null, DirectoryReconciliationResponse.class);
+  }
+
  private static Client initializeClient(DirectoryServerConfiguration directoryServerConfiguration)
      throws CertificateException
  {