Discard old chunk-based account crawler machinery

This commit is contained in:
Jon Chambers
2023-10-19 12:21:17 -04:00
committed by Jon Chambers
parent 9d47a6f41f
commit 744eb58071
23 changed files with 18 additions and 1024 deletions

View File

@@ -17,11 +17,15 @@ import static org.mockito.Mockito.when;
import com.google.common.net.HttpHeaders;
import com.google.protobuf.ByteString;
import com.vdurmont.semver4j.Semver;
import io.grpc.ManagedChannel;
import io.grpc.Server;
import io.grpc.StatusRuntimeException;
import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.inprocess.InProcessServerBuilder;
import java.io.IOException;
import java.util.EnumMap;
import java.util.Set;
import java.util.stream.Stream;
import javax.servlet.FilterChain;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
@@ -30,8 +34,8 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.signal.chat.rpc.EchoServiceGrpc;
import org.signal.chat.rpc.EchoRequest;
import org.signal.chat.rpc.EchoServiceGrpc;
import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicRemoteDeprecationConfiguration;
import org.whispersystems.textsecuregcm.grpc.EchoServiceImpl;
@@ -40,15 +44,6 @@ import org.whispersystems.textsecuregcm.grpc.UserAgentInterceptor;
import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
import org.whispersystems.textsecuregcm.util.ua.ClientPlatform;
import io.grpc.Metadata;
import io.grpc.ManagedChannel;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.StatusRuntimeException;
import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.stub.MetadataUtils;
class RemoteDeprecationFilterTest {
@Test

View File

@@ -1,78 +0,0 @@
/*
* Copyright 2021 Signal Messenger, LLC
* SPDX-License-Identifier: AGPL-3.0-only
*/
package org.whispersystems.textsecuregcm.storage;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Collections;
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.junit.jupiter.api.extension.RegisterExtension;
import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
class AccountDatabaseCrawlerIntegrationTest {
@RegisterExtension
static final RedisClusterExtension REDIS_CLUSTER_EXTENSION = RedisClusterExtension.builder().build();
private static final UUID FIRST_UUID = UUID.fromString("82339e80-81cd-48e2-9ed2-ccd5dd262ad9");
private static final UUID SECOND_UUID = UUID.fromString("cc705c84-33cf-456b-8239-a6a34e2f561a");
private Account firstAccount;
private Account secondAccount;
private AccountsManager accountsManager;
private AccountDatabaseCrawlerListener listener;
private AccountDatabaseCrawler accountDatabaseCrawler;
private static final int CHUNK_SIZE = 1;
@BeforeEach
void setUp() throws Exception {
firstAccount = mock(Account.class);
secondAccount = mock(Account.class);
accountsManager = mock(AccountsManager.class);
listener = mock(AccountDatabaseCrawlerListener.class);
when(firstAccount.getUuid()).thenReturn(FIRST_UUID);
when(secondAccount.getUuid()).thenReturn(SECOND_UUID);
when(accountsManager.getAllFromDynamo(CHUNK_SIZE)).thenReturn(
new AccountCrawlChunk(List.of(firstAccount), FIRST_UUID));
when(accountsManager.getAllFromDynamo(any(UUID.class), eq(CHUNK_SIZE)))
.thenReturn(new AccountCrawlChunk(List.of(secondAccount), SECOND_UUID))
.thenReturn(new AccountCrawlChunk(Collections.emptyList(), null));
final AccountDatabaseCrawlerCache crawlerCache = new AccountDatabaseCrawlerCache(
REDIS_CLUSTER_EXTENSION.getRedisCluster(), "test");
accountDatabaseCrawler = new AccountDatabaseCrawler("test", accountsManager, crawlerCache, List.of(listener),
CHUNK_SIZE);
}
@Test
void testCrawlAllAccounts() throws Exception {
accountDatabaseCrawler.crawlAllAccounts();
verify(accountsManager).getAllFromDynamo(CHUNK_SIZE);
verify(accountsManager).getAllFromDynamo(FIRST_UUID, CHUNK_SIZE);
verify(accountsManager).getAllFromDynamo(SECOND_UUID, CHUNK_SIZE);
verify(listener).onCrawlStart();
verify(listener).timeAndProcessCrawlChunk(Optional.empty(), List.of(firstAccount));
verify(listener).timeAndProcessCrawlChunk(Optional.of(FIRST_UUID), List.of(secondAccount));
verify(listener).onCrawlEnd();
}
}

View File

@@ -1,82 +0,0 @@
/*
* Copyright 2013 Signal Messenger, LLC
* SPDX-License-Identifier: AGPL-3.0-only
*/
package org.whispersystems.textsecuregcm.storage;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
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.Collections;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
class AccountDatabaseCrawlerTest {
private static final UUID ACCOUNT1 = UUID.randomUUID();
private static final UUID ACCOUNT2 = UUID.randomUUID();
private static final int CHUNK_SIZE = 1000;
private final Account account1 = mock(Account.class);
private final Account account2 = mock(Account.class);
private final AccountsManager accounts = mock(AccountsManager.class);
private final AccountDatabaseCrawlerListener listener = mock(AccountDatabaseCrawlerListener.class);
private final AccountDatabaseCrawlerCache cache = mock(AccountDatabaseCrawlerCache.class);
private final AccountDatabaseCrawler crawler =
new AccountDatabaseCrawler("test", accounts, cache, List.of(listener), CHUNK_SIZE);
@BeforeEach
void setup() {
when(account1.getUuid()).thenReturn(ACCOUNT1);
when(account2.getUuid()).thenReturn(ACCOUNT2);
when(accounts.getAllFromDynamo(anyInt())).thenReturn(
new AccountCrawlChunk(List.of(account1, account2), ACCOUNT2));
when(accounts.getAllFromDynamo(eq(ACCOUNT1), anyInt())).thenReturn(
new AccountCrawlChunk(List.of(account2), ACCOUNT2));
when(accounts.getAllFromDynamo(eq(ACCOUNT2), anyInt())).thenReturn(
new AccountCrawlChunk(Collections.emptyList(), null));
when(cache.claimActiveWork(any(), anyLong())).thenReturn(true);
}
@Test
void testCrawlAllAccounts() {
when(cache.getLastUuid())
.thenReturn(Optional.empty());
crawler.crawlAllAccounts();
verify(cache, times(1)).claimActiveWork(any(String.class), anyLong());
verify(cache, times(1)).getLastUuid();
verify(listener, times(1)).onCrawlStart();
verify(accounts, times(1)).getAllFromDynamo(eq(CHUNK_SIZE));
verify(accounts, times(1)).getAllFromDynamo(eq(ACCOUNT2), eq(CHUNK_SIZE));
verify(listener, times(1)).timeAndProcessCrawlChunk(Optional.empty(), List.of(account1, account2));
verify(listener, times(1)).timeAndProcessCrawlChunk(Optional.of(ACCOUNT2), Collections.emptyList());
verify(listener, times(1)).onCrawlEnd();
verify(cache, times(1)).setLastUuid(eq(Optional.of(ACCOUNT2)));
// times(2) because empty() will get cached on the last run of loop and then again at the end
verify(cache, times(1)).setLastUuid(eq(Optional.empty()));
verify(cache, times(1)).releaseActiveWork(any(String.class));
verifyNoMoreInteractions(accounts);
verifyNoMoreInteractions(listener);
verifyNoMoreInteractions(cache);
}
}

View File

@@ -45,8 +45,6 @@ import org.whispersystems.textsecuregcm.tests.util.KeysHelper;
class AccountsManagerChangeNumberIntegrationTest {
private static final int SCAN_PAGE_SIZE = 1;
@RegisterExtension
static final DynamoDbExtension DYNAMO_DB_EXTENSION = new DynamoDbExtension(
Tables.ACCOUNTS,
@@ -82,8 +80,7 @@ class AccountsManagerChangeNumberIntegrationTest {
Tables.NUMBERS.tableName(),
Tables.PNI_ASSIGNMENTS.tableName(),
Tables.USERNAMES.tableName(),
Tables.DELETED_ACCOUNTS.tableName(),
SCAN_PAGE_SIZE);
Tables.DELETED_ACCOUNTS.tableName());
accountLockExecutor = Executors.newSingleThreadExecutor();

View File

@@ -57,8 +57,6 @@ import org.whispersystems.textsecuregcm.util.Pair;
class AccountsManagerConcurrentModificationIntegrationTest {
private static final int SCAN_PAGE_SIZE = 1;
@RegisterExtension
static final DynamoDbExtension DYNAMO_DB_EXTENSION = new DynamoDbExtension(
Tables.ACCOUNTS,
@@ -89,8 +87,7 @@ class AccountsManagerConcurrentModificationIntegrationTest {
Tables.NUMBERS.tableName(),
Tables.PNI_ASSIGNMENTS.tableName(),
Tables.USERNAMES.tableName(),
Tables.DELETED_ACCOUNTS.tableName(),
SCAN_PAGE_SIZE);
Tables.DELETED_ACCOUNTS.tableName());
{
//noinspection unchecked

View File

@@ -58,7 +58,6 @@ class AccountsManagerUsernameIntegrationTest {
private static final String BASE_64_URL_USERNAME_HASH_2 = "NLUom-CHwtemcdvOTTXdmXmzRIV7F05leS8lwkVK_vc";
private static final String BASE_64_URL_ENCRYPTED_USERNAME_1 = "md1votbj9r794DsqTNrBqA";
private static final String BASE_64_URL_ENCRYPTED_USERNAME_2 = "9hrqVLy59bzgPse-S9NUsA";
private static final int SCAN_PAGE_SIZE = 1;
private static final byte[] USERNAME_HASH_1 = Base64.getUrlDecoder().decode(BASE_64_URL_USERNAME_HASH_1);
private static final byte[] USERNAME_HASH_2 = Base64.getUrlDecoder().decode(BASE_64_URL_USERNAME_HASH_2);
private static final byte[] ENCRYPTED_USERNAME_1 = Base64.getUrlDecoder().decode(BASE_64_URL_ENCRYPTED_USERNAME_1);
@@ -99,8 +98,7 @@ class AccountsManagerUsernameIntegrationTest {
Tables.NUMBERS.tableName(),
Tables.PNI_ASSIGNMENTS.tableName(),
Tables.USERNAMES.tableName(),
Tables.DELETED_ACCOUNTS.tableName(),
SCAN_PAGE_SIZE));
Tables.DELETED_ACCOUNTS.tableName()));
final AccountLockManager accountLockManager = mock(AccountLockManager.class);

View File

@@ -19,7 +19,6 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.uuid.UUIDComparator;
import java.nio.charset.StandardCharsets;
import java.security.SecureRandom;
import java.time.Duration;
@@ -85,8 +84,6 @@ class AccountsTest {
private static final byte[] ENCRYPTED_USERNAME_1 = Base64.getUrlDecoder().decode(BASE_64_URL_ENCRYPTED_USERNAME_1);
private static final byte[] ENCRYPTED_USERNAME_2 = Base64.getUrlDecoder().decode(BASE_64_URL_ENCRYPTED_USERNAME_2);
private static final int SCAN_PAGE_SIZE = 1;
private static final AtomicInteger ACCOUNT_COUNTER = new AtomicInteger(1);
@@ -119,8 +116,7 @@ class AccountsTest {
Tables.NUMBERS.tableName(),
Tables.PNI_ASSIGNMENTS.tableName(),
Tables.USERNAMES.tableName(),
Tables.DELETED_ACCOUNTS.tableName(),
SCAN_PAGE_SIZE);
Tables.DELETED_ACCOUNTS.tableName());
}
@Test
@@ -423,7 +419,7 @@ class AccountsTest {
accounts = new Accounts(mock(DynamoDbClient.class),
dynamoDbAsyncClient, Tables.ACCOUNTS.tableName(),
Tables.NUMBERS.tableName(), Tables.PNI_ASSIGNMENTS.tableName(), Tables.USERNAMES.tableName(),
Tables.DELETED_ACCOUNTS.tableName(), SCAN_PAGE_SIZE);
Tables.DELETED_ACCOUNTS.tableName());
Exception e = TransactionConflictException.builder().build();
e = wrapException ? new CompletionException(e) : e;
@@ -436,55 +432,6 @@ class AccountsTest {
assertThatThrownBy(() -> accounts.update(account)).isInstanceOfAny(ContestedOptimisticLockException.class);
}
@Test
void testRetrieveFrom() {
List<Account> users = new ArrayList<>();
for (int i = 1; i <= 100; i++) {
Account account = generateAccount("+1" + String.format("%03d", i), UUID.randomUUID(), UUID.randomUUID());
users.add(account);
accounts.create(account);
}
users.sort((account, t1) -> UUIDComparator.staticCompare(account.getUuid(), t1.getUuid()));
AccountCrawlChunk retrieved = accounts.getAllFromStart(10);
assertThat(retrieved.getAccounts().size()).isEqualTo(10);
for (int i = 0; i < retrieved.getAccounts().size(); i++) {
final Account retrievedAccount = retrieved.getAccounts().get(i);
final Account expectedAccount = users.stream()
.filter(account -> account.getUuid().equals(retrievedAccount.getUuid()))
.findAny()
.orElseThrow();
verifyStoredState(expectedAccount.getNumber(), expectedAccount.getUuid(), expectedAccount.getPhoneNumberIdentifier(), null, retrievedAccount, expectedAccount);
users.remove(expectedAccount);
}
for (int j = 0; j < 9; j++) {
retrieved = accounts.getAllFrom(retrieved.getLastUuid().orElseThrow(), 10);
assertThat(retrieved.getAccounts().size()).isEqualTo(10);
for (int i = 0; i < retrieved.getAccounts().size(); i++) {
final Account retrievedAccount = retrieved.getAccounts().get(i);
final Account expectedAccount = users.stream()
.filter(account -> account.getUuid().equals(retrievedAccount.getUuid()))
.findAny()
.orElseThrow();
verifyStoredState(expectedAccount.getNumber(), expectedAccount.getUuid(), expectedAccount.getPhoneNumberIdentifier(), null, retrievedAccount, expectedAccount);
users.remove(expectedAccount);
}
}
assertThat(users).isEmpty();
}
@Test
void testGetAll() {
final List<Account> expectedAccounts = new ArrayList<>();

View File

@@ -1,159 +0,0 @@
/*
* Copyright 2013 Signal Messenger, LLC
* SPDX-License-Identifier: AGPL-3.0-only
*/
package org.whispersystems.textsecuregcm.storage;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyBoolean;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.isNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import static org.whispersystems.textsecuregcm.tests.util.AccountsHelper.eqUuid;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.whispersystems.textsecuregcm.tests.util.AccountsHelper;
import org.whispersystems.textsecuregcm.util.Util;
class PushFeedbackProcessorTest {
private final AccountsManager accountsManager = mock(AccountsManager.class);
private Account uninstalledAccount = mock(Account.class);
private Account mixedAccount = mock(Account.class);
private Account freshAccount = mock(Account.class);
private Account cleanAccount = mock(Account.class);
private Account stillActiveAccount = mock(Account.class);
private Device uninstalledDevice = mock(Device.class);
private Device uninstalledDeviceTwo = mock(Device.class);
private Device installedDevice = mock(Device.class);
private Device installedDeviceTwo = mock(Device.class);
private Device recentUninstalledDevice = mock(Device.class);
private Device stillActiveDevice = mock(Device.class);
@BeforeEach
void setup() {
AccountsHelper.setupMockUpdate(accountsManager);
when(uninstalledDevice.getUninstalledFeedbackTimestamp()).thenReturn(
Util.todayInMillis() - TimeUnit.DAYS.toMillis(2));
when(uninstalledDevice.getLastSeen()).thenReturn(Util.todayInMillis() - TimeUnit.DAYS.toMillis(2));
when(uninstalledDevice.isEnabled()).thenReturn(true);
when(uninstalledDeviceTwo.getUninstalledFeedbackTimestamp()).thenReturn(
Util.todayInMillis() - TimeUnit.DAYS.toMillis(3));
when(uninstalledDeviceTwo.getLastSeen()).thenReturn(Util.todayInMillis() - TimeUnit.DAYS.toMillis(3));
when(uninstalledDeviceTwo.isEnabled()).thenReturn(true);
when(installedDevice.getUninstalledFeedbackTimestamp()).thenReturn(0L);
when(installedDevice.isEnabled()).thenReturn(true);
when(installedDeviceTwo.getUninstalledFeedbackTimestamp()).thenReturn(0L);
when(installedDeviceTwo.isEnabled()).thenReturn(true);
when(recentUninstalledDevice.getUninstalledFeedbackTimestamp()).thenReturn(
Util.todayInMillis() - TimeUnit.DAYS.toMillis(1));
when(recentUninstalledDevice.getLastSeen()).thenReturn(Util.todayInMillis());
when(recentUninstalledDevice.isEnabled()).thenReturn(true);
when(stillActiveDevice.getUninstalledFeedbackTimestamp()).thenReturn(
Util.todayInMillis() - TimeUnit.DAYS.toMillis(2));
when(stillActiveDevice.getLastSeen()).thenReturn(Util.todayInMillis());
when(stillActiveDevice.isEnabled()).thenReturn(true);
when(uninstalledAccount.getDevices()).thenReturn(List.of(uninstalledDevice));
when(mixedAccount.getDevices()).thenReturn(List.of(installedDevice, uninstalledDeviceTwo));
when(freshAccount.getDevices()).thenReturn(List.of(recentUninstalledDevice));
when(cleanAccount.getDevices()).thenReturn(List.of(installedDeviceTwo));
when(stillActiveAccount.getDevices()).thenReturn(List.of(stillActiveDevice));
when(mixedAccount.getUuid()).thenReturn(UUID.randomUUID());
when(freshAccount.getUuid()).thenReturn(UUID.randomUUID());
when(cleanAccount.getUuid()).thenReturn(UUID.randomUUID());
when(stillActiveAccount.getUuid()).thenReturn(UUID.randomUUID());
when(uninstalledAccount.isEnabled()).thenReturn(true);
when(uninstalledAccount.isDiscoverableByPhoneNumber()).thenReturn(true);
when(uninstalledAccount.getUuid()).thenReturn(UUID.randomUUID());
when(uninstalledAccount.getNumber()).thenReturn("+18005551234");
AccountsHelper.setupMockGet(accountsManager,
Set.of(uninstalledAccount, mixedAccount, freshAccount, cleanAccount, stillActiveAccount));
}
@Test
void testEmpty() {
PushFeedbackProcessor processor = new PushFeedbackProcessor(accountsManager, Executors.newSingleThreadExecutor());
processor.timeAndProcessCrawlChunk(Optional.of(UUID.randomUUID()), Collections.emptyList());
verifyNoInteractions(accountsManager);
}
@Test
void testUpdate() {
PushFeedbackProcessor processor = new PushFeedbackProcessor(accountsManager, Executors.newSingleThreadExecutor());
processor.timeAndProcessCrawlChunk(Optional.of(UUID.randomUUID()),
List.of(uninstalledAccount, mixedAccount, stillActiveAccount, freshAccount, cleanAccount));
verify(uninstalledDevice).setApnId(isNull());
verify(uninstalledDevice).setGcmId(isNull());
verify(uninstalledDevice).setFetchesMessages(eq(false));
when(uninstalledDevice.isEnabled()).thenReturn(false);
verify(accountsManager).update(eqUuid(uninstalledAccount), any());
verify(uninstalledDeviceTwo).setApnId(isNull());
verify(uninstalledDeviceTwo).setGcmId(isNull());
verify(uninstalledDeviceTwo).setFetchesMessages(eq(false));
when(uninstalledDeviceTwo.isEnabled()).thenReturn(false);
verify(installedDevice, never()).setApnId(any());
verify(installedDevice, never()).setGcmId(any());
verify(installedDevice, never()).setFetchesMessages(anyBoolean());
verify(accountsManager).update(eqUuid(mixedAccount), any());
verify(recentUninstalledDevice, never()).setApnId(any());
verify(recentUninstalledDevice, never()).setGcmId(any());
verify(recentUninstalledDevice, never()).setFetchesMessages(anyBoolean());
verify(accountsManager, never()).update(eqUuid(freshAccount), any());
verify(installedDeviceTwo, never()).setApnId(any());
verify(installedDeviceTwo, never()).setGcmId(any());
verify(installedDeviceTwo, never()).setFetchesMessages(anyBoolean());
verify(accountsManager, never()).update(eqUuid(cleanAccount), any());
verify(stillActiveDevice).setUninstalledFeedbackTimestamp(eq(0L));
verify(stillActiveDevice, never()).setApnId(any());
verify(stillActiveDevice, never()).setGcmId(any());
verify(stillActiveDevice, never()).setFetchesMessages(anyBoolean());
when(stillActiveDevice.getUninstalledFeedbackTimestamp()).thenReturn(0L);
verify(accountsManager).update(eqUuid(stillActiveAccount), any());
// there are un-verified calls to updateDevice
clearInvocations(accountsManager);
// a second crawl should not make any further updates
processor.timeAndProcessCrawlChunk(Optional.of(UUID.randomUUID()),
List.of(uninstalledAccount, mixedAccount, stillActiveAccount, freshAccount, cleanAccount));
verify(accountsManager, never()).update(any(Account.class), any());
}
}