queueUrls, final SqsAsyncClient sqs) {
- this.queueUrls = queueUrls;
- this.sqs = sqs;
- }
-
- @Override
- public void start() throws Exception {
- }
-
- @Override
- public void stop() throws Exception {
- synchronized (outstandingRequests) {
- while (outstandingRequests.get() > 0) {
- outstandingRequests.wait();
- }
- }
-
- sqs.close();
- }
-
- public void refreshAccount(final Account account) {
- sendUpdateMessage(account.getUuid(), account.getNumber(),
- account.shouldBeVisibleInDirectory() ? UpdateAction.ADD : UpdateAction.DELETE);
- }
-
- public void deleteAccount(final Account account) {
- sendUpdateMessage(account.getUuid(), account.getNumber(), UpdateAction.DELETE);
- }
-
- public void changePhoneNumber(final Account account, final String originalNumber, final String newNumber) {
- sendUpdateMessage(account.getUuid(), originalNumber, UpdateAction.DELETE);
- sendUpdateMessage(account.getUuid(), newNumber, account.shouldBeVisibleInDirectory() ? UpdateAction.ADD : UpdateAction.DELETE);
- }
-
- private void sendUpdateMessage(final UUID uuid, final String number, final UpdateAction action) {
- for (final String queueUrl : queueUrls) {
- final Timer.Context timerContext = sendMessageBatchTimer.time();
-
- final SendMessageRequest request = SendMessageRequest.builder()
- .queueUrl(queueUrl)
- .messageBody("-")
- .messageDeduplicationId(UUID.randomUUID().toString())
- .messageGroupId(number)
- .messageAttributes(Map.of(
- "id", MessageAttributeValue.builder().dataType("String").stringValue(number).build(),
- "uuid", MessageAttributeValue.builder().dataType("String").stringValue(uuid.toString()).build(),
- "action", action.toMessageAttributeValue()
- ))
- .build();
-
- synchronized (outstandingRequests) {
- outstandingRequests.incrementAndGet();
- }
-
- sqs.sendMessage(request).whenComplete((response, cause) -> {
- try {
- if (cause instanceof SdkServiceException) {
- serviceErrorMeter.mark();
- logger.warn("sqs service error", cause);
- } else if (cause instanceof SdkClientException) {
- clientErrorMeter.mark();
- logger.warn("sqs client error", cause);
- } else if (cause != null) {
- logger.warn("sqs unexpected error", cause);
- }
- } finally {
- synchronized (outstandingRequests) {
- outstandingRequests.decrementAndGet();
- outstandingRequests.notifyAll();
- }
-
- timerContext.close();
- }
- });
- }
- }
-}
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawlerCache.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawlerCache.java
index 100da674f..284c936dc 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawlerCache.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawlerCache.java
@@ -17,7 +17,6 @@ import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
public class AccountDatabaseCrawlerCache {
public static final String GENERAL_PURPOSE_PREFIX = "";
- public static final String DIRECTORY_RECONCILER_PREFIX = "directory-reconciler";
public static final String ACCOUNT_CLEANER_PREFIX = "account-cleaner";
private static final String ACTIVE_WORKER_KEY = "account_database_crawler_cache_active_worker";
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
index b18fc3bb5..d120d99ad 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
@@ -52,7 +52,6 @@ import org.whispersystems.textsecuregcm.redis.RedisOperation;
import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
import org.whispersystems.textsecuregcm.securevaluerecovery.SecureValueRecovery2Client;
-import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
import org.whispersystems.textsecuregcm.util.Constants;
import org.whispersystems.textsecuregcm.util.DestinationDeviceValidator;
import org.whispersystems.textsecuregcm.util.SystemMapper;
@@ -89,7 +88,6 @@ public class AccountsManager {
private final PhoneNumberIdentifiers phoneNumberIdentifiers;
private final FaultTolerantRedisCluster cacheCluster;
private final DeletedAccountsManager deletedAccountsManager;
- private final DirectoryQueue directoryQueue;
private final Keys keys;
private final MessagesManager messagesManager;
private final ProfilesManager profilesManager;
@@ -133,7 +131,6 @@ public class AccountsManager {
final PhoneNumberIdentifiers phoneNumberIdentifiers,
final FaultTolerantRedisCluster cacheCluster,
final DeletedAccountsManager deletedAccountsManager,
- final DirectoryQueue directoryQueue,
final Keys keys,
final MessagesManager messagesManager,
final ProfilesManager profilesManager,
@@ -149,7 +146,6 @@ public class AccountsManager {
this.phoneNumberIdentifiers = phoneNumberIdentifiers;
this.cacheCluster = cacheCluster;
this.deletedAccountsManager = deletedAccountsManager;
- this.directoryQueue = directoryQueue;
this.keys = keys;
this.messagesManager = messagesManager;
this.profilesManager = profilesManager;
@@ -237,11 +233,6 @@ public class AccountsManager {
Metrics.counter(CREATE_COUNTER_NAME, tags).increment();
- if (!account.isDiscoverableByPhoneNumber()) {
- // The newly-created account has explicitly opted out of discoverability
- directoryQueue.deleteAccount(account);
- }
-
accountAttributes.recoveryPassword().ifPresent(registrationRecoveryPassword ->
registrationRecoveryPasswordsManager.storeForCurrentNumber(account.getNumber(), registrationRecoveryPassword));
});
@@ -277,7 +268,6 @@ public class AccountsManager {
if (maybeExistingAccount.isPresent()) {
delete(maybeExistingAccount.get());
- directoryQueue.deleteAccount(maybeExistingAccount.get());
displacedUuid = maybeExistingAccount.map(Account::getUuid);
} else {
displacedUuid = deletedAci;
@@ -296,7 +286,6 @@ public class AccountsManager {
AccountChangeValidator.NUMBER_CHANGE_VALIDATOR);
updatedAccount.set(numberChangedAccount);
- directoryQueue.changePhoneNumber(numberChangedAccount, originalNumber, number);
keys.delete(phoneNumberIdentifier);
keys.delete(originalPhoneNumberIdentifier);
@@ -363,7 +352,7 @@ public class AccountsManager {
/**
* Reserve a username hash so that no other accounts may take it.
- *
+ *
* The reserved hash can later be set with {@link #confirmReservedUsernameHash(Account, byte[])}. The reservation
* will eventually expire, after which point confirmReservedUsernameHash may fail if another account has taken the
* username hash.
@@ -409,7 +398,7 @@ public class AccountsManager {
}
/**
- * Set a username hash previously reserved with {@link #reserveUsernameHash(Account, List)}
+ * Set a username hash previously reserved with {@link #reserveUsernameHash(Account, List)}
*
* @param account the account to update
* @param reservedUsernameHash the previously reserved username hash
@@ -500,8 +489,6 @@ public class AccountsManager {
*/
private Account update(Account account, Function updater) {
- final boolean wasVisibleBeforeUpdate = account.shouldBeVisibleInDirectory();
-
final Account updatedAccount;
try (Timer.Context ignored = updateTimer.time()) {
@@ -519,12 +506,6 @@ public class AccountsManager {
redisSet(updatedAccount);
}
- final boolean isVisibleAfterUpdate = updatedAccount.shouldBeVisibleInDirectory();
-
- if (wasVisibleBeforeUpdate != isVisibleAfterUpdate) {
- directoryQueue.refreshAccount(updatedAccount);
- }
-
return updatedAccount;
}
@@ -653,10 +634,6 @@ public class AccountsManager {
}
}
- public Optional getNumberForPhoneNumberIdentifier(UUID pni) {
- return phoneNumberIdentifiers.getPhoneNumber(pni);
- }
-
public UUID getPhoneNumberIdentifier(String e164) {
return phoneNumberIdentifiers.getPhoneNumberIdentifier(e164);
}
@@ -673,7 +650,6 @@ public class AccountsManager {
try (final Timer.Context ignored = deleteTimer.time()) {
deletedAccountsManager.lockAndPut(account.getNumber(), () -> {
delete(account);
- directoryQueue.deleteAccount(account);
return account.getUuid();
});
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccounts.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccounts.java
index 72a0cf46f..b1834688f 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccounts.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccounts.java
@@ -6,33 +6,17 @@ package org.whispersystems.textsecuregcm.storage;
import java.time.Duration;
import java.time.Instant;
-import java.util.ArrayDeque;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.Queue;
-import java.util.Set;
import java.util.UUID;
-import java.util.stream.Collectors;
import org.whispersystems.textsecuregcm.util.AttributeValues;
-import org.whispersystems.textsecuregcm.util.Pair;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
-import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
-import software.amazon.awssdk.services.dynamodb.model.BatchGetItemRequest;
-import software.amazon.awssdk.services.dynamodb.model.BatchGetItemResponse;
import software.amazon.awssdk.services.dynamodb.model.DeleteItemRequest;
import software.amazon.awssdk.services.dynamodb.model.GetItemRequest;
import software.amazon.awssdk.services.dynamodb.model.GetItemResponse;
-import software.amazon.awssdk.services.dynamodb.model.KeysAndAttributes;
import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
import software.amazon.awssdk.services.dynamodb.model.QueryRequest;
import software.amazon.awssdk.services.dynamodb.model.QueryResponse;
-import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
-import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest;
public class DeletedAccounts extends AbstractDynamoDbStore {
@@ -40,7 +24,6 @@ public class DeletedAccounts extends AbstractDynamoDbStore {
static final String KEY_ACCOUNT_E164 = "P";
static final String ATTR_ACCOUNT_UUID = "U";
static final String ATTR_EXPIRES = "E";
- static final String ATTR_NEEDS_CDS_RECONCILIATION = "R";
static final String UUID_TO_E164_INDEX_NAME = "u_to_p";
@@ -50,23 +33,20 @@ public class DeletedAccounts extends AbstractDynamoDbStore {
static final int GET_BATCH_SIZE = 100;
private final String tableName;
- private final String needsReconciliationIndexName;
- public DeletedAccounts(final DynamoDbClient dynamoDb, final String tableName, final String needsReconciliationIndexName) {
+ public DeletedAccounts(final DynamoDbClient dynamoDb, final String tableName) {
super(dynamoDb);
this.tableName = tableName;
- this.needsReconciliationIndexName = needsReconciliationIndexName;
}
- void put(UUID uuid, String e164, boolean needsReconciliation) {
+ void put(UUID uuid, String e164) {
db().putItem(PutItemRequest.builder()
.tableName(tableName)
.item(Map.of(
KEY_ACCOUNT_E164, AttributeValues.fromString(e164),
ATTR_ACCOUNT_UUID, AttributeValues.fromUUID(uuid),
- ATTR_EXPIRES, AttributeValues.fromLong(Instant.now().plus(TIME_TO_LIVE).getEpochSecond()),
- ATTR_NEEDS_CDS_RECONCILIATION, AttributeValues.fromInt(needsReconciliation ? 1 : 0)))
+ ATTR_EXPIRES, AttributeValues.fromLong(Instant.now().plus(TIME_TO_LIVE).getEpochSecond())))
.build());
}
@@ -108,72 +88,4 @@ public class DeletedAccounts extends AbstractDynamoDbStore {
.key(Map.of(KEY_ACCOUNT_E164, AttributeValues.fromString(e164)))
.build());
}
-
- List> listAccountsToReconcile(final int max) {
-
- final ScanRequest scanRequest = ScanRequest.builder()
- .tableName(tableName)
- .indexName(needsReconciliationIndexName)
- .limit(max)
- .build();
-
- return scan(scanRequest, max)
- .stream()
- .map(item -> new Pair<>(
- AttributeValues.getUUID(item, ATTR_ACCOUNT_UUID, null),
- AttributeValues.getString(item, KEY_ACCOUNT_E164, null)))
- .collect(Collectors.toList());
- }
-
- Set getAccountsNeedingReconciliation(final Collection e164s) {
- final Queue> pendingKeys = e164s.stream()
- .map(e164 -> Map.of(KEY_ACCOUNT_E164, AttributeValues.fromString(e164)))
- .collect(Collectors.toCollection(() -> new ArrayDeque<>(e164s.size())));
-
- final Set accountsNeedingReconciliation = new HashSet<>(e164s.size());
- final List> batchKeys = new ArrayList<>(GET_BATCH_SIZE);
-
- while (!pendingKeys.isEmpty()) {
- batchKeys.clear();
-
- for (int i = 0; i < GET_BATCH_SIZE && !pendingKeys.isEmpty(); i++) {
- batchKeys.add(pendingKeys.remove());
- }
-
- final BatchGetItemResponse response = db().batchGetItem(BatchGetItemRequest.builder()
- .requestItems(Map.of(tableName, KeysAndAttributes.builder()
- .consistentRead(true)
- .keys(batchKeys)
- .build()))
- .build());
-
- response.responses().getOrDefault(tableName, Collections.emptyList()).stream()
- .filter(attributes -> AttributeValues.getInt(attributes, ATTR_NEEDS_CDS_RECONCILIATION, 0) == 1)
- .map(attributes -> AttributeValues.getString(attributes, KEY_ACCOUNT_E164, null))
- .forEach(accountsNeedingReconciliation::add);
-
- if (response.hasUnprocessedKeys() && response.unprocessedKeys().containsKey(tableName)) {
- pendingKeys.addAll(response.unprocessedKeys().get(tableName).keys());
- }
- }
-
- return accountsNeedingReconciliation;
- }
-
- void markReconciled(final Collection phoneNumbersReconciled) {
-
- phoneNumbersReconciled.forEach(number -> db().updateItem(
- UpdateItemRequest.builder()
- .tableName(tableName)
- .key(Map.of(
- KEY_ACCOUNT_E164, AttributeValues.fromString(number)
- ))
- .updateExpression("REMOVE #needs_reconciliation")
- .expressionAttributeNames(Map.of(
- "#needs_reconciliation", ATTR_NEEDS_CDS_RECONCILIATION
- ))
- .build()
- ));
- }
-
}
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsDirectoryReconciler.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsDirectoryReconciler.java
deleted file mode 100644
index 944f41160..000000000
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsDirectoryReconciler.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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 io.micrometer.core.instrument.Counter;
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.Timer;
-import java.util.List;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest;
-import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest.User;
-import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse;
-
-public class DeletedAccountsDirectoryReconciler {
-
- private final Logger logger = LoggerFactory.getLogger(DeletedAccountsDirectoryReconciler.class);
-
- private final DirectoryReconciliationClient directoryReconciliationClient;
-
- private final Timer deleteTimer;
- private final Counter errorCounter;
-
- public DeletedAccountsDirectoryReconciler(
- final String replicationName,
- final DirectoryReconciliationClient directoryReconciliationClient) {
- this.directoryReconciliationClient = directoryReconciliationClient;
-
- deleteTimer = Timer.builder(name(DeletedAccountsDirectoryReconciler.class, "delete"))
- .tag("replicationName", replicationName)
- .register(Metrics.globalRegistry);
-
- errorCounter = Counter.builder(name(DeletedAccountsDirectoryReconciler.class, "error"))
- .tag("replicationName", replicationName)
- .register(Metrics.globalRegistry);
- }
-
- public void onCrawlChunk(final List deletedUsers) throws ChunkProcessingFailedException {
-
- try {
- deleteTimer.recordCallable(() -> {
- try {
- final DirectoryReconciliationResponse response = directoryReconciliationClient.delete(
- new DirectoryReconciliationRequest(deletedUsers));
-
- if (response.getStatus() != DirectoryReconciliationResponse.Status.OK) {
- errorCounter.increment();
- throw new ChunkProcessingFailedException("Response status: " + response.getStatus());
- }
- } catch (final Exception e) {
- errorCounter.increment();
- throw new ChunkProcessingFailedException(e);
- }
-
- return null;
- });
- } catch (final ChunkProcessingFailedException e) {
- throw e;
- } catch (final Exception e) {
- logger.warn("Unexpected exception", e);
- throw new RuntimeException(e);
- }
- }
-
-}
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManager.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManager.java
index 1dbdc104e..43c139251 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManager.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManager.java
@@ -11,21 +11,16 @@ import com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClientOptions;
import com.amazonaws.services.dynamodbv2.LockItem;
import com.amazonaws.services.dynamodbv2.ReleaseLockOptions;
-import com.amazonaws.services.dynamodbv2.model.LockCurrentlyUnavailableException;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.List;
import java.util.Optional;
-import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Supplier;
-import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.whispersystems.textsecuregcm.util.Pair;
public class DeletedAccountsManager {
@@ -35,20 +30,6 @@ public class DeletedAccountsManager {
private static final Logger log = LoggerFactory.getLogger(DeletedAccountsManager.class);
- @FunctionalInterface
- public interface DeletedAccountReconciliationConsumer {
-
- /**
- * Reconcile a list of deleted account records.
- *
- * @param deletedAccounts the account records to reconcile
- * @return a list of account records that were successfully reconciled; accounts that were not successfully
- * reconciled may be retried later
- * @throws ChunkProcessingFailedException in the event of an error while processing the batch of account records
- */
- Collection reconcile(List> deletedAccounts) throws ChunkProcessingFailedException;
- }
-
public DeletedAccountsManager(final DeletedAccounts deletedAccounts, final AmazonDynamoDB lockDynamoDb, final String lockTableName) {
this.deletedAccounts = deletedAccounts;
@@ -98,7 +79,7 @@ public class DeletedAccountsManager {
public void lockAndPut(final String e164, final Supplier supplier) throws InterruptedException {
withLock(List.of(e164), ignored -> {
try {
- deletedAccounts.put(supplier.get(), e164, true);
+ deletedAccounts.put(supplier.get(), e164);
} catch (final Exception e) {
log.warn("Supplier threw an exception while holding lock on a deleted account record", e);
throw new RuntimeException(e);
@@ -123,7 +104,7 @@ public class DeletedAccountsManager {
withLock(List.of(original, target), acis -> {
try {
- function.apply(acis.get(0), acis.get(1)).ifPresent(aci -> deletedAccounts.put(aci, original, true));
+ function.apply(acis.get(0), acis.get(1)).ifPresent(aci -> deletedAccounts.put(aci, original));
} catch (final Exception e) {
log.warn("Supplier threw an exception while holding lock on a deleted account record", e);
throw new RuntimeException(e);
@@ -154,48 +135,6 @@ public class DeletedAccountsManager {
}
}
- public void lockAndReconcileAccounts(final int max, final DeletedAccountReconciliationConsumer consumer) throws ChunkProcessingFailedException {
- final List lockItems = new ArrayList<>();
- try {
- final List> reconciliationCandidates = deletedAccounts.listAccountsToReconcile(max).stream()
- .filter(pair -> {
- boolean lockAcquired = false;
-
- try {
- lockItems.add(lockClient.acquireLock(AcquireLockOptions.builder(pair.second())
- .withAcquireReleasedLocksConsistently(true)
- .withShouldSkipBlockingWait(true)
- .build()));
-
- lockAcquired = true;
- } catch (final InterruptedException e) {
- log.warn("Interrupted while acquiring lock for reconciliation", e);
- } catch (final LockCurrentlyUnavailableException ignored) {
- }
-
- return lockAcquired;
- }).toList();
-
- assert lockItems.size() == reconciliationCandidates.size();
-
- // A deleted account's status may have changed in the time between getting a list of candidates and acquiring a lock
- // on the candidate records. Now that we hold the lock, check which of the candidates still need to be reconciled.
- final Set numbersNeedingReconciliationAfterLock =
- deletedAccounts.getAccountsNeedingReconciliation(reconciliationCandidates.stream()
- .map(Pair::second)
- .collect(Collectors.toList()));
-
- final List> accountsToReconcile = reconciliationCandidates.stream()
- .filter(candidate -> numbersNeedingReconciliationAfterLock.contains(candidate.second()))
- .collect(Collectors.toList());
-
- deletedAccounts.markReconciled(consumer.reconcile(accountsToReconcile));
- } finally {
- lockItems.forEach(
- lockItem -> lockClient.releaseLock(ReleaseLockOptions.builder(lockItem).withBestEffort(true).build()));
- }
- }
-
public Optional findDeletedAccountAci(final String e164) {
return deletedAccounts.findUuid(e164);
}
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsTableCrawler.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsTableCrawler.java
deleted file mode 100644
index c6e83b702..000000000
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsTableCrawler.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Copyright 2013-2020 Signal Messenger, LLC
- * SPDX-License-Identifier: AGPL-3.0-only
- */
-package org.whispersystems.textsecuregcm.storage;
-
-import static com.codahale.metrics.MetricRegistry.name;
-
-import io.micrometer.core.instrument.Metrics;
-import java.io.IOException;
-import java.time.Duration;
-import java.util.List;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.stream.Collectors;
-import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest.User;
-import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
-import org.whispersystems.textsecuregcm.util.Pair;
-
-public class DeletedAccountsTableCrawler extends ManagedPeriodicWork {
-
- private static final Duration WORKER_TTL = Duration.ofMinutes(2);
- private static final Duration RUN_INTERVAL = Duration.ofMinutes(15);
- private static final String ACTIVE_WORKER_KEY = "deleted_accounts_crawler_cache_active_worker";
-
- private static final int MAX_BATCH_SIZE = 5_000;
- private static final String BATCH_SIZE_DISTRIBUTION_NAME = name(DeletedAccountsTableCrawler.class, "batchSize");
-
- private final DeletedAccountsManager deletedAccountsManager;
- private final List reconcilers;
-
- public DeletedAccountsTableCrawler(
- final DeletedAccountsManager deletedAccountsManager,
- final List reconcilers,
- final FaultTolerantRedisCluster cluster,
- final ScheduledExecutorService executorService) throws IOException {
-
- super(new ManagedPeriodicWorkLock(ACTIVE_WORKER_KEY, cluster), WORKER_TTL, RUN_INTERVAL, executorService);
-
- this.deletedAccountsManager = deletedAccountsManager;
- this.reconcilers = reconcilers;
- }
-
- @Override
- public void doPeriodicWork() throws Exception {
-
- deletedAccountsManager.lockAndReconcileAccounts(MAX_BATCH_SIZE, deletedAccounts -> {
- final List deletedUsers = deletedAccounts.stream()
- .map(pair -> new User(pair.first(), pair.second()))
- .collect(Collectors.toList());
-
- for (DeletedAccountsDirectoryReconciler reconciler : reconcilers) {
- reconciler.onCrawlChunk(deletedUsers);
- }
-
- final List reconciledPhoneNumbers = deletedAccounts.stream()
- .map(Pair::second)
- .collect(Collectors.toList());
-
- Metrics.summary(BATCH_SIZE_DISTRIBUTION_NAME).record(reconciledPhoneNumbers.size());
-
- return reconciledPhoneNumbers;
- });
- }
-
-}
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciler.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciler.java
deleted file mode 100644
index 5d645e9d5..000000000
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciler.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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 io.micrometer.core.instrument.Metrics;
-import java.util.ArrayList;
-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.configuration.dynamic.DynamicConfiguration;
-import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest;
-import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse;
-import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse.Status;
-
-public class DirectoryReconciler extends AccountDatabaseCrawlerListener {
-
- private static final Logger logger = LoggerFactory.getLogger(DirectoryReconciler.class);
- private static final String SEND_TIMER_NAME = name(DirectoryReconciler.class, "sendRequest");
-
- private final String replicationName;
- private final DirectoryReconciliationClient reconciliationClient;
- private final DynamicConfigurationManager dynamicConfigurationManager;
-
- public DirectoryReconciler(String replicationName, DirectoryReconciliationClient reconciliationClient,
- DynamicConfigurationManager dynamicConfigurationManager) {
- this.reconciliationClient = reconciliationClient;
- this.replicationName = replicationName;
- this.dynamicConfigurationManager = dynamicConfigurationManager;
- }
-
- @Override
- public void onCrawlStart() {
- }
-
- @Override
- public void onCrawlEnd(Optional fromUuid) {
- if (!dynamicConfigurationManager.getConfiguration().getDirectoryReconcilerConfiguration().isEnabled()) {
- return;
- }
-
- reconciliationClient.complete();
- }
-
- @Override
- protected void onCrawlChunk(final Optional fromUuid, final List accounts)
- throws AccountDatabaseCrawlerRestartException {
-
- if (!dynamicConfigurationManager.getConfiguration().getDirectoryReconcilerConfiguration().isEnabled()) {
- return;
- }
-
- final DirectoryReconciliationRequest addUsersRequest;
- final DirectoryReconciliationRequest deleteUsersRequest;
- {
- final List addedUsers = new ArrayList<>(accounts.size());
- final List 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()));
- }
- });
-
- addUsersRequest = new DirectoryReconciliationRequest(addedUsers);
- deleteUsersRequest = new DirectoryReconciliationRequest(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");
- }
- }
-
- private DirectoryReconciliationResponse sendDeletes(final DirectoryReconciliationRequest request) {
- return sendRequest(request, reconciliationClient::delete, "delete");
-
- }
-
- private DirectoryReconciliationResponse sendAdditions(final DirectoryReconciliationRequest request) {
- return sendRequest(request, reconciliationClient::add, "add");
- }
-
- private DirectoryReconciliationResponse sendRequest(final DirectoryReconciliationRequest request,
- final Function 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(), context);
- }
- return response;
- } catch (ProcessingException ex) {
- logger.warn("request error: ", ex);
- throw new ProcessingException(ex);
- }
- });
- }
-
-}
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciliationClient.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciliationClient.java
deleted file mode 100644
index ce0fc4318..000000000
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciliationClient.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Copyright 2013-2020 Signal Messenger, LLC
- * SPDX-License-Identifier: AGPL-3.0-only
- */
-package org.whispersystems.textsecuregcm.storage;
-
-import java.security.KeyStore;
-import java.security.cert.CertificateException;
-import javax.net.ssl.SSLContext;
-import javax.ws.rs.client.Client;
-import javax.ws.rs.client.ClientBuilder;
-import javax.ws.rs.client.Entity;
-import javax.ws.rs.core.MediaType;
-import org.glassfish.jersey.SslConfigurator;
-import org.glassfish.jersey.client.authentication.HttpAuthenticationFeature;
-import org.whispersystems.textsecuregcm.configuration.DirectoryServerConfiguration;
-import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationRequest;
-import org.whispersystems.textsecuregcm.entities.DirectoryReconciliationResponse;
-import org.whispersystems.textsecuregcm.util.CertificateUtil;
-
-public class DirectoryReconciliationClient {
-
- private final String replicationUrl;
- private final Client client;
-
- public DirectoryReconciliationClient(DirectoryServerConfiguration directoryServerConfiguration)
- throws CertificateException
- {
- this.replicationUrl = directoryServerConfiguration.getReplicationUrl();
- this.client = initializeClient(directoryServerConfiguration);
- }
-
- public DirectoryReconciliationResponse add(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")
- .request(MediaType.APPLICATION_JSON_TYPE)
- .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 {
- KeyStore trustStore = CertificateUtil.buildKeyStoreForPem(
- directoryServerConfiguration.getReplicationCaCertificates().toArray(new String[0]));
- SSLContext sslContext = SslConfigurator.newInstance()
- .securityProtocol("TLSv1.2")
- .trustStore(trustStore)
- .createSSLContext();
-
- return ClientBuilder.newBuilder()
- .register(
- HttpAuthenticationFeature.basic("signal", directoryServerConfiguration.getReplicationPassword().getBytes()))
- .sslContext(sslContext)
- .build();
- }
-}
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/workers/AssignUsernameCommand.java b/service/src/main/java/org/whispersystems/textsecuregcm/workers/AssignUsernameCommand.java
index 4e867782d..058417d39 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/workers/AssignUsernameCommand.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/workers/AssignUsernameCommand.java
@@ -36,7 +36,6 @@ import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
import org.whispersystems.textsecuregcm.securevaluerecovery.SecureValueRecovery2Client;
-import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
import org.whispersystems.textsecuregcm.storage.Account;
import org.whispersystems.textsecuregcm.storage.Accounts;
import org.whispersystems.textsecuregcm.storage.AccountsManager;
@@ -150,8 +149,7 @@ public class AssignUsernameCommand extends EnvironmentCommand requestCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
- verify(sqsAsyncClient).sendMessage(requestCaptor.capture());
-
- assertEquals(MessageAttributeValue.builder().dataType("String").stringValue(expectedAction).build(),
- requestCaptor.getValue().messageAttributes().get("action"));
- }
-
- @SuppressWarnings("unused")
- private static Stream argumentsForTestRefreshRegisteredUser() {
- return Stream.of(
- Arguments.of(true, "add"),
- Arguments.of(false, "delete"));
- }
-
- @Test
- void testSendMessageMultipleQueues() {
- final DirectoryQueue directoryQueue = new DirectoryQueue(List.of("sqs://first", "sqs://second"), sqsAsyncClient);
-
- final Account account = mock(Account.class);
- when(account.getNumber()).thenReturn("+18005556543");
- when(account.getUuid()).thenReturn(UUID.randomUUID());
- when(account.shouldBeVisibleInDirectory()).thenReturn(true);
-
- directoryQueue.refreshAccount(account);
-
- final ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(SendMessageRequest.class);
- verify(sqsAsyncClient, times(2)).sendMessage(requestCaptor.capture());
-
- for (final SendMessageRequest sendMessageRequest : requestCaptor.getAllValues()) {
- assertEquals(MessageAttributeValue.builder().dataType("String").stringValue("add").build(),
- sendMessageRequest.messageAttributes().get("action"));
- }
- }
-
- @Test
- void testStop() {
- final CompletableFuture sendMessageFuture = new CompletableFuture<>();
- when(sqsAsyncClient.sendMessage(any(SendMessageRequest.class))).thenReturn(sendMessageFuture);
-
- 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.shouldBeVisibleInDirectory()).thenReturn(true);
-
- directoryQueue.refreshAccount(account);
-
- final CompletableFuture stopFuture = CompletableFuture.supplyAsync(() -> {
- try {
- directoryQueue.stop();
- return true;
- } catch (final Exception e) {
- return false;
- }
- });
-
- assertThrows(TimeoutException.class, () -> stopFuture.get(1, TimeUnit.SECONDS),
- "Directory queue should not finish shutting down until all outstanding requests are resolved");
-
- sendMessageFuture.complete(SendMessageResponse.builder().build());
- assertTrue(stopFuture.join());
- }
-}
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerChangeNumberIntegrationTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerChangeNumberIntegrationTest.java
index c81166f5c..9d17176c5 100644
--- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerChangeNumberIntegrationTest.java
+++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerChangeNumberIntegrationTest.java
@@ -33,15 +33,10 @@ import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
import org.whispersystems.textsecuregcm.securevaluerecovery.SecureValueRecovery2Client;
-import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
-import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Indexes;
import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Tables;
class AccountsManagerChangeNumberIntegrationTest {
- private static final String NUMBERS_TABLE_NAME = "numbers_test";
- private static final String PNI_ASSIGNMENT_TABLE_NAME = "pni_assignment_test";
- private static final String USERNAMES_TABLE_NAME = "usernames_test";
private static final int SCAN_PAGE_SIZE = 1;
@RegisterExtension
@@ -82,8 +77,7 @@ class AccountsManagerChangeNumberIntegrationTest {
SCAN_PAGE_SIZE);
deletedAccounts = new DeletedAccounts(DYNAMO_DB_EXTENSION.getDynamoDbClient(),
- Tables.DELETED_ACCOUNTS.tableName(),
- Indexes.DELETED_ACCOUNTS_NEEDS_RECONCILIATION.indexName());
+ Tables.DELETED_ACCOUNTS.tableName());
final DeletedAccountsManager deletedAccountsManager = new DeletedAccountsManager(deletedAccounts,
DYNAMO_DB_EXTENSION.getLegacyDynamoClient(),
@@ -108,7 +102,6 @@ class AccountsManagerChangeNumberIntegrationTest {
phoneNumberIdentifiers,
CACHE_CLUSTER_EXTENSION.getRedisCluster(),
deletedAccountsManager,
- mock(DirectoryQueue.class),
mock(Keys.class),
mock(MessagesManager.class),
mock(ProfilesManager.class),
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerConcurrentModificationIntegrationTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerConcurrentModificationIntegrationTest.java
index 320ad88c6..3bb3a428c 100644
--- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerConcurrentModificationIntegrationTest.java
+++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerConcurrentModificationIntegrationTest.java
@@ -47,7 +47,6 @@ import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
import org.whispersystems.textsecuregcm.securevaluerecovery.SecureValueRecovery2Client;
-import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Tables;
import org.whispersystems.textsecuregcm.tests.util.DevicesHelper;
import org.whispersystems.textsecuregcm.tests.util.JsonHelpers;
@@ -111,7 +110,6 @@ class AccountsManagerConcurrentModificationIntegrationTest {
phoneNumberIdentifiers,
RedisClusterHelper.builder().stringCommands(commands).build(),
deletedAccountsManager,
- mock(DirectoryQueue.class),
mock(Keys.class),
mock(MessagesManager.class),
mock(ProfilesManager.class),
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerTest.java
index 4a8af3d29..4fc755e19 100644
--- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerTest.java
+++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerTest.java
@@ -59,7 +59,6 @@ import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
import org.whispersystems.textsecuregcm.securevaluerecovery.SecureValueRecovery2Client;
-import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
import org.whispersystems.textsecuregcm.storage.Device.DeviceCapabilities;
import org.whispersystems.textsecuregcm.tests.util.AccountsHelper;
import org.whispersystems.textsecuregcm.tests.util.DevicesHelper;
@@ -73,7 +72,6 @@ class AccountsManagerTest {
private Accounts accounts;
private DeletedAccountsManager deletedAccountsManager;
- private DirectoryQueue directoryQueue;
private Keys keys;
private MessagesManager messagesManager;
private ProfilesManager profilesManager;
@@ -96,7 +94,6 @@ class AccountsManagerTest {
void setup() throws InterruptedException {
accounts = mock(Accounts.class);
deletedAccountsManager = mock(DeletedAccountsManager.class);
- directoryQueue = mock(DirectoryQueue.class);
keys = mock(Keys.class);
messagesManager = mock(MessagesManager.class);
profilesManager = mock(ProfilesManager.class);
@@ -153,7 +150,6 @@ class AccountsManagerTest {
phoneNumberIdentifiers,
RedisClusterHelper.builder().stringCommands(commands).build(),
deletedAccountsManager,
- directoryQueue,
keys,
messagesManager,
profilesManager,
@@ -598,10 +594,6 @@ class AccountsManagerTest {
final Account account = accountsManager.create("+18005550123", "password", null, attributes, new ArrayList<>());
assertEquals(discoverable, account.isDiscoverableByPhoneNumber());
-
- if (!discoverable) {
- verify(directoryQueue).deleteAccount(account);
- }
}
@ParameterizedTest
@@ -615,32 +607,6 @@ class AccountsManagerTest {
assertEquals(hasStorage, account.isStorageSupported());
}
- @ParameterizedTest
- @MethodSource
- void testUpdateDirectoryQueue(final boolean visibleBeforeUpdate, final boolean visibleAfterUpdate,
- final boolean expectRefresh) {
- final Account account = AccountsHelper.generateTestAccount("+14152222222", UUID.randomUUID(), UUID.randomUUID(), new ArrayList<>(), new byte[16]);
-
- // this sets up the appropriate result for Account#shouldBeVisibleInDirectory
- final Device device = generateTestDevice(0);
- account.addDevice(device);
- account.setDiscoverableByPhoneNumber(visibleBeforeUpdate);
-
- final Account updatedAccount = accountsManager.update(account,
- a -> a.setDiscoverableByPhoneNumber(visibleAfterUpdate));
-
- verify(directoryQueue, times(expectRefresh ? 1 : 0)).refreshAccount(updatedAccount);
- }
-
- @SuppressWarnings("unused")
- private static Stream testUpdateDirectoryQueue() {
- return Stream.of(
- Arguments.of(false, false, false),
- Arguments.of(true, true, false),
- Arguments.of(false, true, true),
- Arguments.of(true, false, true));
- }
-
@ParameterizedTest
@MethodSource
void testUpdateDeviceLastSeen(final boolean expectUpdate, final long initialLastSeen, final long updatedLastSeen) {
@@ -680,7 +646,6 @@ class AccountsManagerTest {
assertTrue(phoneNumberIdentifiersByE164.containsKey(targetNumber));
- verify(directoryQueue).changePhoneNumber(argThat(a -> a.getUuid().equals(uuid)), eq(originalNumber), eq(targetNumber));
verify(keys).delete(originalPni);
verify(keys).delete(phoneNumberIdentifiersByE164.get(targetNumber));
}
@@ -694,7 +659,6 @@ class AccountsManagerTest {
assertEquals(number, account.getNumber());
verify(deletedAccountsManager, never()).lockAndPut(anyString(), anyString(), any());
- verify(directoryQueue, never()).changePhoneNumber(any(), any(), any());
verify(keys, never()).delete(any());
}
@@ -736,8 +700,6 @@ class AccountsManagerTest {
assertTrue(phoneNumberIdentifiersByE164.containsKey(targetNumber));
- verify(directoryQueue).changePhoneNumber(argThat(a -> a.getUuid().equals(uuid)), eq(originalNumber), eq(targetNumber));
- verify(directoryQueue).deleteAccount(existingAccount);
verify(keys).delete(originalPni);
verify(keys).delete(targetPni);
}
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerUsernameIntegrationTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerUsernameIntegrationTest.java
index c00a94fec..40afcb800 100644
--- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerUsernameIntegrationTest.java
+++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/AccountsManagerUsernameIntegrationTest.java
@@ -44,7 +44,6 @@ import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
import org.whispersystems.textsecuregcm.securevaluerecovery.SecureValueRecovery2Client;
-import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Tables;
import org.whispersystems.textsecuregcm.util.AttributeValues;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
@@ -114,7 +113,6 @@ class AccountsManagerUsernameIntegrationTest {
phoneNumberIdentifiers,
CACHE_CLUSTER_EXTENSION.getRedisCluster(),
deletedAccountsManager,
- mock(DirectoryQueue.class),
mock(Keys.class),
mock(MessagesManager.class),
mock(ProfilesManager.class),
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManagerTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManagerTest.java
index c0cd7c54e..4ff809dc9 100644
--- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManagerTest.java
+++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManagerTest.java
@@ -5,21 +5,14 @@
package org.whispersystems.textsecuregcm.storage;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import java.lang.Thread.State;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.junit.jupiter.api.function.Executable;
-import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Indexes;
import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Tables;
class DeletedAccountsManagerTest {
@@ -34,8 +27,7 @@ class DeletedAccountsManagerTest {
@BeforeEach
void setUp() {
deletedAccounts = new DeletedAccounts(DYNAMO_DB_EXTENSION.getDynamoDbClient(),
- Tables.DELETED_ACCOUNTS.tableName(),
- Indexes.DELETED_ACCOUNTS_NEEDS_RECONCILIATION.indexName());
+ Tables.DELETED_ACCOUNTS.tableName());
deletedAccountsManager = new DeletedAccountsManager(deletedAccounts,
DYNAMO_DB_EXTENSION.getLegacyDynamoClient(),
@@ -47,7 +39,7 @@ class DeletedAccountsManagerTest {
final UUID uuid = UUID.randomUUID();
final String e164 = "+18005551234";
- deletedAccounts.put(uuid, e164, true);
+ deletedAccounts.put(uuid, e164);
deletedAccountsManager.lockAndTake(e164, maybeUuid -> assertEquals(Optional.of(uuid), maybeUuid));
assertEquals(Optional.empty(), deletedAccounts.findUuid(e164));
}
@@ -57,7 +49,7 @@ class DeletedAccountsManagerTest {
final UUID uuid = UUID.randomUUID();
final String e164 = "+18005551234";
- deletedAccounts.put(uuid, e164, true);
+ deletedAccounts.put(uuid, e164);
assertThrows(RuntimeException.class, () -> deletedAccountsManager.lockAndTake(e164, maybeUuid -> {
assertEquals(Optional.of(uuid), maybeUuid);
@@ -66,73 +58,4 @@ class DeletedAccountsManagerTest {
assertEquals(Optional.of(uuid), deletedAccounts.findUuid(e164));
}
-
- @Test
- void testReconciliationLockContention() throws ChunkProcessingFailedException {
-
- final UUID[] uuids = new UUID[3];
- final String[] e164s = new String[uuids.length];
-
- for (int i = 0; i < uuids.length; i++) {
- uuids[i] = UUID.randomUUID();
- e164s[i] = String.format("+1800555%04d", i);
- }
-
- final Map expectedReconciledAccounts = new HashMap<>();
-
- for (int i = 0; i < uuids.length; i++) {
- deletedAccounts.put(uuids[i], e164s[i], true);
- expectedReconciledAccounts.put(e164s[i], uuids[i]);
- }
-
- final UUID replacedUUID = UUID.randomUUID();
- final Map reconciledAccounts = new HashMap<>();
-
- final Thread putThread = new Thread(() -> {
- try {
- deletedAccountsManager.lockAndPut(e164s[0], () -> replacedUUID);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- },
- getClass().getSimpleName() + "-put");
-
- final Thread reconcileThread = new Thread(() -> {
- try {
- deletedAccountsManager.lockAndReconcileAccounts(uuids.length, deletedAccounts -> {
- // We hold the lock for the first account, so a thread trying to operate on that first count should block
- // waiting for the lock.
- putThread.start();
-
- // Make sure the other thread really does actually block at some point
- while (putThread.getState() != State.TIMED_WAITING) {
- Thread.yield();
- }
-
- deletedAccounts.forEach(pair -> reconciledAccounts.put(pair.second(), pair.first()));
- return reconciledAccounts.keySet();
- });
- } catch (ChunkProcessingFailedException e) {
- throw new AssertionError(e);
- }
- }, getClass().getSimpleName() + "-reconcile");
-
- reconcileThread.start();
-
- assertDoesNotThrow((Executable) reconcileThread::join);
- assertDoesNotThrow((Executable) putThread::join);
-
- assertEquals(expectedReconciledAccounts, reconciledAccounts);
-
- // The "put" thread should have completed after the reconciliation thread wrapped up. We can verify that's true by
- // reconciling again; the updated account (and only that account) should appear in the "needs reconciliation" list.
- deletedAccountsManager.lockAndReconcileAccounts(uuids.length, deletedAccounts -> {
- assertEquals(1, deletedAccounts.size());
- assertEquals(replacedUUID, deletedAccounts.get(0).first());
- assertEquals(e164s[0], deletedAccounts.get(0).second());
-
- return List.of(deletedAccounts.get(0).second());
- });
- }
-
}
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsTest.java
index 8b75d0cfe..60954e7a9 100644
--- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsTest.java
+++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsTest.java
@@ -5,20 +5,13 @@
package org.whispersystems.textsecuregcm.storage;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
import java.util.Optional;
-import java.util.Set;
import java.util.UUID;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Indexes;
import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Tables;
-import org.whispersystems.textsecuregcm.util.Pair;
class DeletedAccountsTest {
@@ -29,41 +22,7 @@ class DeletedAccountsTest {
@BeforeEach
void setUp() {
- deletedAccounts = new DeletedAccounts(DYNAMO_DB_EXTENSION.getDynamoDbClient(),
- Tables.DELETED_ACCOUNTS.tableName(),
- Indexes.DELETED_ACCOUNTS_NEEDS_RECONCILIATION.indexName());
- }
-
- @Test
- void testPutList() {
- UUID firstUuid = UUID.randomUUID();
- UUID secondUuid = UUID.randomUUID();
- UUID thirdUuid = UUID.randomUUID();
-
- String firstNumber = "+14152221234";
- String secondNumber = "+14152225678";
- String thirdNumber = "+14159998765";
-
- assertTrue(deletedAccounts.listAccountsToReconcile(1).isEmpty());
-
- deletedAccounts.put(firstUuid, firstNumber, true);
- deletedAccounts.put(secondUuid, secondNumber, true);
- deletedAccounts.put(thirdUuid, thirdNumber, true);
-
- assertEquals(1, deletedAccounts.listAccountsToReconcile(1).size());
-
- assertTrue(deletedAccounts.listAccountsToReconcile(10).containsAll(
- List.of(
- new Pair<>(firstUuid, firstNumber),
- new Pair<>(secondUuid, secondNumber))));
-
- deletedAccounts.markReconciled(List.of(firstNumber, secondNumber));
-
- assertEquals(List.of(new Pair<>(thirdUuid, thirdNumber)), deletedAccounts.listAccountsToReconcile(10));
-
- deletedAccounts.markReconciled(List.of(thirdNumber));
-
- assertTrue(deletedAccounts.listAccountsToReconcile(1).isEmpty());
+ deletedAccounts = new DeletedAccounts(DYNAMO_DB_EXTENSION.getDynamoDbClient(), Tables.DELETED_ACCOUNTS.tableName());
}
@Test
@@ -73,7 +32,7 @@ class DeletedAccountsTest {
assertEquals(Optional.empty(), deletedAccounts.findUuid(e164));
- deletedAccounts.put(uuid, e164, true);
+ deletedAccounts.put(uuid, e164);
assertEquals(Optional.of(uuid), deletedAccounts.findUuid(e164));
}
@@ -85,7 +44,7 @@ class DeletedAccountsTest {
assertEquals(Optional.empty(), deletedAccounts.findUuid(e164));
- deletedAccounts.put(uuid, e164, true);
+ deletedAccounts.put(uuid, e164);
assertEquals(Optional.of(uuid), deletedAccounts.findUuid(e164));
@@ -94,45 +53,6 @@ class DeletedAccountsTest {
assertEquals(Optional.empty(), deletedAccounts.findUuid(e164));
}
- @Test
- void testGetAccountsNeedingReconciliation() {
- final UUID firstUuid = UUID.randomUUID();
- final UUID secondUuid = UUID.randomUUID();
-
- final String firstNumber = "+14152221234";
- final String secondNumber = "+14152225678";
- final String thirdNumber = "+14159998765";
-
- assertEquals(Collections.emptySet(),
- deletedAccounts.getAccountsNeedingReconciliation(List.of(firstNumber, secondNumber, thirdNumber)));
-
- deletedAccounts.put(firstUuid, firstNumber, true);
- deletedAccounts.put(secondUuid, secondNumber, true);
-
- assertEquals(Set.of(firstNumber, secondNumber),
- deletedAccounts.getAccountsNeedingReconciliation(List.of(firstNumber, secondNumber, thirdNumber)));
- }
-
- @Test
- void testGetAccountsNeedingReconciliationLargeBatch() {
- final int itemCount = (DeletedAccounts.GET_BATCH_SIZE * 3) + 1;
-
- final Set expectedAccountsNeedingReconciliation = new HashSet<>(itemCount);
-
- for (int i = 0; i < itemCount; i++) {
- final String e164 = String.format("+18000555%04d", i);
-
- deletedAccounts.put(UUID.randomUUID(), e164, true);
- expectedAccountsNeedingReconciliation.add(e164);
- }
-
- final Set accountsNeedingReconciliation =
- deletedAccounts.getAccountsNeedingReconciliation(expectedAccountsNeedingReconciliation);
-
- assertEquals(expectedAccountsNeedingReconciliation, accountsNeedingReconciliation);
- }
-
-
@Test
void testFindE164() {
assertEquals(Optional.empty(), deletedAccounts.findE164(UUID.randomUUID()));
@@ -140,7 +60,7 @@ class DeletedAccountsTest {
final UUID uuid = UUID.randomUUID();
final String e164 = "+18005551234";
- deletedAccounts.put(uuid, e164, true);
+ deletedAccounts.put(uuid, e164);
assertEquals(Optional.of(e164), deletedAccounts.findE164(uuid));
}
@@ -153,7 +73,7 @@ class DeletedAccountsTest {
final UUID uuid = UUID.randomUUID();
- deletedAccounts.put(uuid, e164, true);
+ deletedAccounts.put(uuid, e164);
assertEquals(Optional.of(uuid), deletedAccounts.findUuid(e164));
}
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/DynamoDbExtensionSchema.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/DynamoDbExtensionSchema.java
index 2eb595738..a4b5fea3a 100644
--- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/DynamoDbExtensionSchema.java
+++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/DynamoDbExtensionSchema.java
@@ -18,20 +18,6 @@ import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType;
public final class DynamoDbExtensionSchema {
- public enum Indexes {
-
- DELETED_ACCOUNTS_NEEDS_RECONCILIATION("needs_reconciliation_test");
-
- private final String name;
-
- public String indexName() {
- return name;
- }
-
- Indexes(final String name) { this.name = name; }
-
- }
-
public enum Tables implements DynamoDbExtension.TableSchema {
ACCOUNTS("accounts_test",
@@ -50,22 +36,11 @@ public final class DynamoDbExtensionSchema {
AttributeDefinition.builder()
.attributeName(DeletedAccounts.KEY_ACCOUNT_E164)
.attributeType(ScalarAttributeType.S).build(),
- AttributeDefinition.builder()
- .attributeName(DeletedAccounts.ATTR_NEEDS_CDS_RECONCILIATION)
- .attributeType(ScalarAttributeType.N)
- .build(),
AttributeDefinition.builder()
.attributeName(DeletedAccounts.ATTR_ACCOUNT_UUID)
.attributeType(ScalarAttributeType.B)
.build()),
List.of(
- GlobalSecondaryIndex.builder()
- .indexName(Indexes.DELETED_ACCOUNTS_NEEDS_RECONCILIATION.indexName())
- .keySchema(KeySchemaElement.builder().attributeName(DeletedAccounts.KEY_ACCOUNT_E164).keyType(KeyType.HASH).build(),
- KeySchemaElement.builder().attributeName(DeletedAccounts.ATTR_NEEDS_CDS_RECONCILIATION).keyType(KeyType.RANGE).build())
- .projection(Projection.builder().projectionType(ProjectionType.INCLUDE).nonKeyAttributes(DeletedAccounts.ATTR_ACCOUNT_UUID).build())
- .provisionedThroughput(ProvisionedThroughput.builder().readCapacityUnits(10L).writeCapacityUnits(10L).build())
- .build(),
GlobalSecondaryIndex.builder()
.indexName(DeletedAccounts.UUID_TO_E164_INDEX_NAME)
.keySchema(
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/tests/controllers/DirectoryControllerTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/tests/controllers/DirectoryControllerTest.java
deleted file mode 100644
index a0a08055a..000000000
--- a/service/src/test/java/org/whispersystems/textsecuregcm/tests/controllers/DirectoryControllerTest.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- * Copyright 2013 Signal Messenger, LLC
- * SPDX-License-Identifier: AGPL-3.0-only
- */
-
-package org.whispersystems.textsecuregcm.tests.controllers;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import com.google.common.collect.ImmutableSet;
-import com.google.common.net.HttpHeaders;
-import io.dropwizard.auth.PolymorphicAuthValueFactoryProvider;
-import io.dropwizard.testing.junit5.DropwizardExtensionsSupport;
-import io.dropwizard.testing.junit5.ResourceExtension;
-import java.util.Collections;
-import javax.ws.rs.client.Entity;
-import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.Response;
-import javax.ws.rs.core.Response.Status.Family;
-import org.glassfish.jersey.test.grizzly.GrizzlyWebTestContainerFactory;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
-import org.whispersystems.textsecuregcm.auth.DisabledPermittedAuthenticatedAccount;
-import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentials;
-import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialsGenerator;
-import org.whispersystems.textsecuregcm.controllers.DirectoryController;
-import org.whispersystems.textsecuregcm.tests.util.AuthHelper;
-
-@ExtendWith(DropwizardExtensionsSupport.class)
-class DirectoryControllerTest {
-
- private static final ExternalServiceCredentialsGenerator directoryCredentialsGenerator = mock(ExternalServiceCredentialsGenerator.class);
- private static final ExternalServiceCredentials validCredentials = new ExternalServiceCredentials("username", "password");
-
- private static final ResourceExtension resources = ResourceExtension.builder()
- .addProvider(AuthHelper.getAuthFilter())
- .addProvider(new PolymorphicAuthValueFactoryProvider.Binder<>(
- ImmutableSet.of(AuthenticatedAccount.class, DisabledPermittedAuthenticatedAccount.class)))
- .setTestContainerFactory(new GrizzlyWebTestContainerFactory())
- .addResource(new DirectoryController(directoryCredentialsGenerator))
- .build();
-
- @BeforeEach
- void setup() {
- when(directoryCredentialsGenerator.generateFor(eq(AuthHelper.VALID_NUMBER))).thenReturn(validCredentials);
- }
-
- @Test
- void testFeedbackOk() {
- Response response =
- resources.getJerseyTest()
- .target("/v1/directory/feedback-v3/ok")
- .request()
- .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_UUID, AuthHelper.VALID_PASSWORD))
- .put(Entity.json("{\"reason\": \"test reason\"}"));
- assertThat(response.getStatusInfo().getFamily()).isEqualTo(Family.SUCCESSFUL);
- }
-
- @Test
- void testGetAuthToken() {
- ExternalServiceCredentials token =
- resources.getJerseyTest()
- .target("/v1/directory/auth")
- .request()
- .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_UUID, AuthHelper.VALID_PASSWORD))
- .get(ExternalServiceCredentials.class);
- assertThat(token.username()).isEqualTo(validCredentials.username());
- assertThat(token.password()).isEqualTo(validCredentials.password());
- }
-
- @Test
- void testDisabledGetAuthToken() {
- Response response =
- resources.getJerseyTest()
- .target("/v1/directory/auth")
- .request()
- .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.DISABLED_UUID, AuthHelper.DISABLED_PASSWORD))
- .get();
- assertThat(response.getStatus()).isEqualTo(401);
- }
-
-
- @Test
- void testContactIntersection() {
- Response response =
- resources.getJerseyTest()
- .target("/v1/directory/tokens/")
- .request()
- .header("Authorization",
- AuthHelper.getAuthHeader(AuthHelper.VALID_UUID,
- AuthHelper.VALID_PASSWORD))
- .header(HttpHeaders.X_FORWARDED_FOR, "192.168.1.1, 1.1.1.1")
- .put(Entity.entity(Collections.emptyMap(), MediaType.APPLICATION_JSON_TYPE));
-
-
- assertThat(response.getStatus()).isEqualTo(429);
- }
-}
diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/tests/storage/DirectoryReconcilerTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/tests/storage/DirectoryReconcilerTest.java
deleted file mode 100644
index 35f5564b7..000000000
--- a/service/src/test/java/org/whispersystems/textsecuregcm/tests/storage/DirectoryReconcilerTest.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * Copyright 2013-2021 Signal Messenger, LLC
- * SPDX-License-Identifier: AGPL-3.0-only
- */
-
-package org.whispersystems.textsecuregcm.tests.storage;
-
-import static org.assertj.core.api.Assertions.assertThat;
-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.api.Test;
-import org.mockito.ArgumentCaptor;
-import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
-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 org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
-
-class DirectoryReconcilerTest {
-
- private static final UUID VALID_UUID = UUID.randomUUID();
- private static final String VALID_NUMBER = "+14152222222";
- private static final UUID UNDISCOVERABLE_UUID = UUID.randomUUID();
- private static final String UNDISCOVERABLE_NUMBER = "+14153333333";
-
- private final Account visibleAccount = mock(Account.class);
- private final Account undiscoverableAccount = mock(Account.class);
- private final DirectoryReconciliationClient reconciliationClient = mock(DirectoryReconciliationClient.class);
- private final DynamicConfigurationManager dynamicConfigurationManager = mock(DynamicConfigurationManager.class);
- private final DirectoryReconciler directoryReconciler = new DirectoryReconciler("test", reconciliationClient,
- dynamicConfigurationManager);
-
- private final DirectoryReconciliationResponse successResponse = new DirectoryReconciliationResponse(
- DirectoryReconciliationResponse.Status.OK);
-
- @BeforeEach
- void setup() {
- when(dynamicConfigurationManager.getConfiguration()).thenReturn(new DynamicConfiguration());
-
- when(visibleAccount.getUuid()).thenReturn(VALID_UUID);
- when(visibleAccount.getNumber()).thenReturn(VALID_NUMBER);
- when(visibleAccount.shouldBeVisibleInDirectory()).thenReturn(true);
- when(undiscoverableAccount.getUuid()).thenReturn(UNDISCOVERABLE_UUID);
- when(undiscoverableAccount.getNumber()).thenReturn(UNDISCOVERABLE_NUMBER);
- when(undiscoverableAccount.shouldBeVisibleInDirectory()).thenReturn(false);
- }
-
- @Test
- void testCrawlChunkValid() throws AccountDatabaseCrawlerRestartException {
-
- when(reconciliationClient.add(any())).thenReturn(successResponse);
- when(reconciliationClient.delete(any())).thenReturn(successResponse);
-
- directoryReconciler.timeAndProcessCrawlChunk(Optional.of(VALID_UUID),
- Arrays.asList(visibleAccount, undiscoverableAccount));
-
- ArgumentCaptor chunkRequest = ArgumentCaptor.forClass(
- DirectoryReconciliationRequest.class);
- verify(reconciliationClient, times(1)).add(chunkRequest.capture());
-
- assertThat(chunkRequest.getValue().getUsers()).isEqualTo(List.of(new User(VALID_UUID, VALID_NUMBER)));
-
- ArgumentCaptor 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);
- }
-
-}