Signal/Signal-Server
1
0
Fork 0
mirror of https://github.com/signalapp/Signal-Server synced 2026-04-21 06:08:05 +01:00
Code Issues Packages Projects Releases Wiki Activity

Ensure accounts are deleted after batch migration; store migration failures for later processing

Browse Source
This commit is contained in:
Chris Eager
2021-04-19 15:46:09 -05:00
committed by Chris Eager
parent a472774734
commit 5974328d9c
9 changed files with 396 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerConfiguration.java
View File

@@ -134,6 +134,16 @@ public class WhisperServerConfiguration extends Configuration {
@JsonProperty
private AccountsDynamoDbConfiguration accountsDynamoDb;
@Valid
@NotNull
@JsonProperty
private DynamoDbConfiguration migrationDeletedAccountsDynamoDb;
@Valid
@NotNull
@JsonProperty
private DynamoDbConfiguration migrationRetryAccountsDynamoDb;
@Valid
@NotNull
@JsonProperty
@@ -311,6 +321,14 @@ public class WhisperServerConfiguration extends Configuration {
return accountsDynamoDb;
}
public DynamoDbConfiguration getMigrationDeletedAccountsDynamoDbConfiguration() {
return migrationDeletedAccountsDynamoDb;
}
public DynamoDbConfiguration getMigrationRetryAccountsDynamoDbConfiguration() {
return migrationRetryAccountsDynamoDb;
}
public DatabaseConfiguration getAbuseDatabaseConfiguration() {
return abuseDatabase;
}

24
service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
View File

@@ -159,6 +159,8 @@ import org.whispersystems.textsecuregcm.storage.MessagePersister;
import org.whispersystems.textsecuregcm.storage.MessagesCache;
import org.whispersystems.textsecuregcm.storage.MessagesDynamoDb;
import org.whispersystems.textsecuregcm.storage.MessagesManager;
import org.whispersystems.textsecuregcm.storage.MigrationDeletedAccounts;
import org.whispersystems.textsecuregcm.storage.MigrationRetryAccounts;
import org.whispersystems.textsecuregcm.storage.PendingAccounts;
import org.whispersystems.textsecuregcm.storage.PendingAccountsManager;
import org.whispersystems.textsecuregcm.storage.PendingDevices;
@@ -301,14 +303,34 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
.withCredentials(accountsDynamoDbClientBuilder.getCredentials())
.withExecutorFactory(() -> accountsDynamoDbMigrationThreadPool);
AmazonDynamoDBClientBuilder migrationDeletedAccountsDynamoDbClientBuilder = AmazonDynamoDBClientBuilder
.standard()
.withRegion(config.getMigrationDeletedAccountsDynamoDbConfiguration().getRegion())
.withClientConfiguration(new ClientConfiguration().withClientExecutionTimeout(((int) config.getMigrationDeletedAccountsDynamoDbConfiguration().getClientExecutionTimeout().toMillis()))
.withRequestTimeout((int) config.getMigrationDeletedAccountsDynamoDbConfiguration().getClientRequestTimeout().toMillis()))
.withCredentials(InstanceProfileCredentialsProvider.getInstance());
AmazonDynamoDBClientBuilder migrationRetryAccountsDynamoDbClientBuilder = AmazonDynamoDBClientBuilder
.standard()
.withRegion(config.getMigrationRetryAccountsDynamoDbConfiguration().getRegion())
.withClientConfiguration(new ClientConfiguration().withClientExecutionTimeout(((int) config.getMigrationRetryAccountsDynamoDbConfiguration().getClientExecutionTimeout().toMillis()))
.withRequestTimeout((int) config.getMigrationRetryAccountsDynamoDbConfiguration().getClientRequestTimeout().toMillis()))
.withCredentials(InstanceProfileCredentialsProvider.getInstance());
DynamoDB messageDynamoDb = new DynamoDB(messageDynamoDbClientBuilder.build());
DynamoDB preKeyDynamoDb = new DynamoDB(keysDynamoDbClientBuilder.build());
AmazonDynamoDB accountsDynamoDbClient = accountsDynamoDbClientBuilder.build();
AmazonDynamoDBAsync accountsDynamodbAsyncClient = accountsDynamoDbAsyncClientBuilder.build();
DynamoDB recentlyDeletedAccountsDynamoDb = new DynamoDB(migrationDeletedAccountsDynamoDbClientBuilder.build());
DynamoDB migrationRetryAccountsDynamoDb = new DynamoDB(migrationRetryAccountsDynamoDbClientBuilder.build());
MigrationDeletedAccounts migrationDeletedAccounts = new MigrationDeletedAccounts(recentlyDeletedAccountsDynamoDb, config.getMigrationDeletedAccountsDynamoDbConfiguration().getTableName());
MigrationRetryAccounts migrationRetryAccounts = new MigrationRetryAccounts(migrationRetryAccountsDynamoDb, config.getMigrationRetryAccountsDynamoDbConfiguration().getTableName());
Accounts accounts = new Accounts(accountDatabase);
AccountsDynamoDb accountsDynamoDb = new AccountsDynamoDb(accountsDynamoDbClient, accountsDynamodbAsyncClient, accountsDynamoDbMigrationThreadPool, new DynamoDB(accountsDynamoDbClient), config.getAccountsDynamoDbConfiguration().getTableName(), config.getAccountsDynamoDbConfiguration().getPhoneNumberTableName());
AccountsDynamoDb accountsDynamoDb = new AccountsDynamoDb(accountsDynamoDbClient, accountsDynamodbAsyncClient, accountsDynamoDbMigrationThreadPool, new DynamoDB(accountsDynamoDbClient), config.getAccountsDynamoDbConfiguration().getTableName(), config.getAccountsDynamoDbConfiguration().getPhoneNumberTableName(), migrationDeletedAccounts, migrationRetryAccounts);
PendingAccounts pendingAccounts = new PendingAccounts(accountDatabase);
PendingDevices pendingDevices = new PendingDevices (accountDatabase);
Usernames usernames = new Usernames(accountDatabase);

110
service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsDynamoDb.java
View File

@@ -35,6 +35,8 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.whispersystems.textsecuregcm.util.SystemMapper;
import org.whispersystems.textsecuregcm.util.UUIDUtil;
@@ -55,6 +57,9 @@ public class AccountsDynamoDb extends AbstractDynamoDbStore implements AccountSt
private final ThreadPoolExecutor migrationThreadPool;
private final MigrationDeletedAccounts migrationDeletedAccounts;
private final MigrationRetryAccounts migrationRetryAccounts;
private final String phoneNumbersTableName;
private static final Timer CREATE_TIMER = Metrics.timer(name(AccountsDynamoDb.class, "create"));
@@ -63,7 +68,13 @@ public class AccountsDynamoDb extends AbstractDynamoDbStore implements AccountSt
private static final Timer GET_BY_UUID_TIMER = Metrics.timer(name(AccountsDynamoDb.class, "getByUuid"));
private static final Timer DELETE_TIMER = Metrics.timer(name(AccountsDynamoDb.class, "delete"));
public AccountsDynamoDb(AmazonDynamoDB client, AmazonDynamoDBAsync asyncClient, ThreadPoolExecutor migrationThreadPool, DynamoDB dynamoDb, String accountsTableName, String phoneNumbersTableName) {
private final Logger logger = LoggerFactory.getLogger(AccountsDynamoDb.class);
public AccountsDynamoDb(AmazonDynamoDB client, AmazonDynamoDBAsync asyncClient,
ThreadPoolExecutor migrationThreadPool, DynamoDB dynamoDb, String accountsTableName, String phoneNumbersTableName,
MigrationDeletedAccounts migrationDeletedAccounts,
MigrationRetryAccounts accountsMigrationErrors) {
super(dynamoDb);
this.client = client;
@@ -72,6 +83,9 @@ public class AccountsDynamoDb extends AbstractDynamoDbStore implements AccountSt
this.asyncClient = asyncClient;
this.migrationThreadPool = migrationThreadPool;
this.migrationDeletedAccounts = migrationDeletedAccounts;
this.migrationRetryAccounts = accountsMigrationErrors;
}
@Override
@@ -208,50 +222,72 @@ public class AccountsDynamoDb extends AbstractDynamoDbStore implements AccountSt
public void delete(UUID uuid) {
DELETE_TIMER.record(() -> {
Optional<Account> maybeAccount = get(uuid);
maybeAccount.ifPresent(account -> {
TransactWriteItem phoneNumberDelete = new TransactWriteItem()
.withDelete(new Delete()
.withTableName(phoneNumbersTableName)
.withKey(Map.of(ATTR_ACCOUNT_E164, new AttributeValue(account.getNumber()))));
TransactWriteItem accountDelete = new TransactWriteItem().withDelete(
new Delete()
.withTableName(accountsTable.getTableName())
.withKey(Map.of(KEY_ACCOUNT_UUID, new AttributeValue().withB(UUIDUtil.toByteBuffer(uuid)))));
TransactWriteItemsRequest request = new TransactWriteItemsRequest()
.withTransactItems(phoneNumberDelete, accountDelete);
client.transactWriteItems(request);
});
delete(uuid, true);
});
}
public CompletableFuture<Void> migrate(List<Account> accounts, int threads) {
private void delete(UUID uuid, boolean saveInDeletedAccountsTable) {
migrationThreadPool.setCorePoolSize(threads);
migrationThreadPool.setMaximumPoolSize(threads);
if (saveInDeletedAccountsTable) {
migrationDeletedAccounts.put(uuid);
}
final List<CompletableFuture<?>> futures = accounts.stream()
.map(this::migrate)
.map(f -> f.whenComplete((migrated, e) -> {
if (e == null) {
MIGRATED_COUNTER.increment(migrated ? 1 : 0);
} else {
ERROR_COUNTER.increment();
}
}))
.collect(Collectors.toList());
Optional<Account> maybeAccount = get(uuid);
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{}));
maybeAccount.ifPresent(account -> {
TransactWriteItem phoneNumberDelete = new TransactWriteItem()
.withDelete(new Delete()
.withTableName(phoneNumbersTableName)
.withKey(Map.of(ATTR_ACCOUNT_E164, new AttributeValue(account.getNumber()))));
TransactWriteItem accountDelete = new TransactWriteItem().withDelete(
new Delete()
.withTableName(accountsTable.getTableName())
.withKey(Map.of(KEY_ACCOUNT_UUID, new AttributeValue().withB(UUIDUtil.toByteBuffer(uuid)))));
TransactWriteItemsRequest request = new TransactWriteItemsRequest()
.withTransactItems(phoneNumberDelete, accountDelete);
client.transactWriteItems(request);
});
}
private static final Counter MIGRATED_COUNTER = Metrics.counter(name(AccountsDynamoDb.class, "migration", "count"));
private static final Counter ERROR_COUNTER = Metrics.counter(name(AccountsDynamoDb.class, "migration", "error"));
public CompletableFuture<Void> migrate(List<Account> accounts, int threads) {
migrationThreadPool.setCorePoolSize(threads);
migrationThreadPool.setMaximumPoolSize(threads);
final List<CompletableFuture<?>> futures = accounts.stream()
.map(this::migrate)
.map(f -> f.whenComplete((migrated, e) -> {
if (e == null) {
MIGRATED_COUNTER.increment(migrated ? 1 : 0);
} else {
ERROR_COUNTER.increment();
}
}))
.collect(Collectors.toList());
CompletableFuture<Void> migrationBatch = CompletableFuture.allOf(futures.toArray(new CompletableFuture[]{}));
return migrationBatch.whenComplete((result, exception) -> deleteRecentlyDeletedUuids());
}
public void deleteRecentlyDeletedUuids() {
final List<UUID> recentlyDeletedUuids = migrationDeletedAccounts.getRecentlyDeletedUuids();
for (UUID recentlyDeletedUuid : recentlyDeletedUuids) {
delete(recentlyDeletedUuid, false);
}
migrationDeletedAccounts.delete(recentlyDeletedUuids);
}
public CompletableFuture<Boolean> migrate(Account account) {
try {
TransactWriteItem phoneNumberConstraintPut = buildPutWriteItemForPhoneNumberConstraint(account, account.getUuid());
@@ -279,7 +315,11 @@ public class AccountsDynamoDb extends AbstractDynamoDbStore implements AccountSt
// account is already migrated
resultFuture.complete(false);
} else {
ERROR_COUNTER.increment();
try {
migrationRetryAccounts.put(account.getUuid());
} catch (final Exception e) {
logger.error("Could not store account {}", account.getUuid());
}
resultFuture.completeExceptionally(exception);
}
}

64
service/src/main/java/org/whispersystems/textsecuregcm/storage/MigrationDeletedAccounts.java Normal file
View File

@@ -0,0 +1,64 @@
package org.whispersystems.textsecuregcm.storage;
import com.amazonaws.services.dynamodbv2.document.DynamoDB;
import com.amazonaws.services.dynamodbv2.document.Item;
import com.amazonaws.services.dynamodbv2.document.PrimaryKey;
import com.amazonaws.services.dynamodbv2.document.Table;
import com.amazonaws.services.dynamodbv2.document.TableWriteItems;
import com.amazonaws.services.dynamodbv2.document.spec.ScanSpec;
import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import org.whispersystems.textsecuregcm.util.UUIDUtil;
public class MigrationDeletedAccounts extends AbstractDynamoDbStore {
private final Table table;
static final String KEY_UUID = "U";
public MigrationDeletedAccounts(DynamoDB dynamoDb, String tableName) {
super(dynamoDb);
table = dynamoDb.getTable(tableName);
}
public void put(UUID uuid) {
table.putItem(new Item()
.withPrimaryKey(primaryKey(uuid)));
}
public List<UUID> getRecentlyDeletedUuids() {
final List<UUID> uuids = new ArrayList<>();
for (Item item : table.scan(new ScanSpec()).firstPage()) {
// only process one page each time. If we have a significant backlog at the end of the migration
// we can handle it separately
uuids.add(UUIDUtil.fromByteBuffer(item.getByteBuffer(KEY_UUID)));
}
return uuids;
}
public void delete(List<UUID> uuids) {
writeInBatches(uuids, (batch) -> {
final TableWriteItems deleteItems = new TableWriteItems(table.getTableName());
for (UUID uuid : batch) {
deleteItems.addPrimaryKeyToDelete(primaryKey(uuid));
}
executeTableWriteItemsUntilComplete(deleteItems);
});
}
@VisibleForTesting
public static PrimaryKey primaryKey(UUID uuid) {
return new PrimaryKey(KEY_UUID, UUIDUtil.toBytes(uuid));
}
}

60
service/src/main/java/org/whispersystems/textsecuregcm/storage/MigrationRetryAccounts.java Normal file
View File

@@ -0,0 +1,60 @@
package org.whispersystems.textsecuregcm.storage;
import com.amazonaws.services.dynamodbv2.document.DynamoDB;
import com.amazonaws.services.dynamodbv2.document.Item;
import com.amazonaws.services.dynamodbv2.document.Page;
import com.amazonaws.services.dynamodbv2.document.PrimaryKey;
import com.amazonaws.services.dynamodbv2.document.ScanOutcome;
import com.amazonaws.services.dynamodbv2.document.Table;
import com.amazonaws.services.dynamodbv2.document.spec.ScanSpec;
import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import org.whispersystems.textsecuregcm.util.UUIDUtil;
public class MigrationRetryAccounts extends AbstractDynamoDbStore {
private final Table table;
static final String KEY_UUID = "U";
public MigrationRetryAccounts(DynamoDB dynamoDb, String tableName) {
super(dynamoDb);
table = dynamoDb.getTable(tableName);
}
public void put(UUID uuid) {
table.putItem(new Item()
.withPrimaryKey(primaryKey(uuid)));
}
public List<UUID> getUuids(int max) {
final List<UUID> uuids = new ArrayList<>();
for (Page<Item, ScanOutcome> page : table.scan(new ScanSpec()).pages()) {
for (Item item : page) {
uuids.add(UUIDUtil.fromByteBuffer(item.getByteBuffer(KEY_UUID)));
if (uuids.size() >= max) {
break;
}
}
if (uuids.size() >= max) {
break;
}
}
return uuids;
}
@VisibleForTesting
public static PrimaryKey primaryKey(UUID uuid) {
return new PrimaryKey(KEY_UUID, UUIDUtil.toBytes(uuid));
}
}

24
service/src/main/java/org/whispersystems/textsecuregcm/workers/DeleteUserCommand.java
View File

@@ -39,10 +39,12 @@ import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
import org.whispersystems.textsecuregcm.securestorage.SecureStorageClient;
import org.whispersystems.textsecuregcm.sqs.DirectoryQueue;
import org.whispersystems.textsecuregcm.storage.Account;
import org.whispersystems.textsecuregcm.storage.MigrationRetryAccounts;
import org.whispersystems.textsecuregcm.storage.Accounts;
import org.whispersystems.textsecuregcm.storage.AccountsDynamoDb;
import org.whispersystems.textsecuregcm.storage.AccountsManager;
import org.whispersystems.textsecuregcm.storage.AccountsManager.DeletionReason;
import org.whispersystems.textsecuregcm.storage.MigrationDeletedAccounts;
import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
import org.whispersystems.textsecuregcm.storage.FaultTolerantDatabase;
import org.whispersystems.textsecuregcm.storage.KeysDynamoDb;
@@ -126,6 +128,20 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
.withCredentials(accountsDynamoDbClientBuilder.getCredentials())
.withExecutorFactory(() -> accountsDynamoDbMigrationThreadPool);
AmazonDynamoDBClientBuilder migrationDeletedAccountsDynamoDbClientBuilder = AmazonDynamoDBClientBuilder
.standard()
.withRegion(configuration.getMigrationDeletedAccountsDynamoDbConfiguration().getRegion())
.withClientConfiguration(new ClientConfiguration().withClientExecutionTimeout(((int) configuration.getMigrationDeletedAccountsDynamoDbConfiguration().getClientExecutionTimeout().toMillis()))
.withRequestTimeout((int) configuration.getMigrationDeletedAccountsDynamoDbConfiguration().getClientRequestTimeout().toMillis()))
.withCredentials(InstanceProfileCredentialsProvider.getInstance());
AmazonDynamoDBClientBuilder migrationRetryAccountsDynamoDbClientBuilder = AmazonDynamoDBClientBuilder
.standard()
.withRegion(configuration.getMigrationRetryAccountsDynamoDbConfiguration().getRegion())
.withClientConfiguration(new ClientConfiguration().withClientExecutionTimeout(((int) configuration.getMigrationRetryAccountsDynamoDbConfiguration().getClientExecutionTimeout().toMillis()))
.withRequestTimeout((int) configuration.getMigrationRetryAccountsDynamoDbConfiguration().getClientRequestTimeout().toMillis()))
.withCredentials(InstanceProfileCredentialsProvider.getInstance());
DynamoDB messageDynamoDb = new DynamoDB(clientBuilder.build());
DynamoDB preKeysDynamoDb = new DynamoDB(keysDynamoDbClientBuilder.build());
@@ -146,8 +162,14 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
ExperimentEnrollmentManager experimentEnrollmentManager = new ExperimentEnrollmentManager(dynamicConfigurationManager);
DynamoDB migrationDeletedAccountsDynamoDb = new DynamoDB(migrationDeletedAccountsDynamoDbClientBuilder.build());
DynamoDB migrationRetryAccountsDynamoDb = new DynamoDB(migrationRetryAccountsDynamoDbClientBuilder.build());
MigrationDeletedAccounts migrationDeletedAccounts = new MigrationDeletedAccounts(migrationDeletedAccountsDynamoDb, configuration.getMigrationDeletedAccountsDynamoDbConfiguration().getTableName());
MigrationRetryAccounts migrationRetryAccounts = new MigrationRetryAccounts(migrationRetryAccountsDynamoDb, configuration.getMigrationRetryAccountsDynamoDbConfiguration().getTableName());
Accounts accounts = new Accounts(accountDatabase);
AccountsDynamoDb accountsDynamoDb = new AccountsDynamoDb(accountsDynamoDbClient, accountsDynamoDbAsyncClient, accountsDynamoDbMigrationThreadPool, new DynamoDB(accountsDynamoDbClient), configuration.getAccountsDynamoDbConfiguration().getTableName(), configuration.getAccountsDynamoDbConfiguration().getPhoneNumberTableName());
AccountsDynamoDb accountsDynamoDb = new AccountsDynamoDb(accountsDynamoDbClient, accountsDynamoDbAsyncClient, accountsDynamoDbMigrationThreadPool, new DynamoDB(accountsDynamoDbClient), configuration.getAccountsDynamoDbConfiguration().getTableName(), configuration.getAccountsDynamoDbConfiguration().getPhoneNumberTableName(), migrationDeletedAccounts, migrationRetryAccounts);
Usernames usernames = new Usernames(accountDatabase);
Profiles profiles = new Profiles(accountDatabase);
ReservedUsernames reservedUsernames = new ReservedUsernames(accountDatabase);