Standardize client tag version handling; add client version tags to delivery latency metrics

service/src/test/java/org/whispersystems/textsecuregcm/controllers/MessageControllerTest.java

@@ -41,6 +41,7 @@ import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.Arrays;
 import java.util.Base64;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
@@ -74,6 +75,8 @@ import org.signal.libsignal.protocol.ecc.ECKeyPair;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
 import org.whispersystems.textsecuregcm.auth.DisabledPermittedAuthenticatedAccount;
 import org.whispersystems.textsecuregcm.auth.OptionalAccess;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicDeliveryLatencyConfiguration;
 import org.whispersystems.textsecuregcm.entities.ECSignedPreKey;
 import org.whispersystems.textsecuregcm.entities.IncomingMessage;
 import org.whispersystems.textsecuregcm.entities.IncomingMessageList;
@@ -99,6 +102,7 @@ import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.DeletedAccounts;
 import org.whispersystems.textsecuregcm.storage.Device;
+import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
 import org.whispersystems.textsecuregcm.storage.ReportMessageManager;
 import org.whispersystems.textsecuregcm.tests.util.AccountsHelper;
@@ -149,6 +153,7 @@ class MessageControllerTest {
   private static final ReportMessageManager reportMessageManager = mock(ReportMessageManager.class);
   private static final ExecutorService multiRecipientMessageExecutor = mock(ExecutorService.class);
   private static final Scheduler messageDeliveryScheduler = Schedulers.newBoundedElastic(10, 10_000, "messageDelivery");
+  private static final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager = mock(DynamicConfigurationManager.class);
 
   private static final ResourceExtension resources = ResourceExtension.builder()
       .addProperty(ServerProperties.UNWRAP_COMPLETION_STAGE_IN_WRITER_ENABLE, Boolean.TRUE)
@@ -161,7 +166,7 @@ class MessageControllerTest {
       .addResource(
           new MessageController(rateLimiters, messageSender, receiptSender, accountsManager, deletedAccounts,
               messagesManager, pushNotificationManager, reportMessageManager, multiRecipientMessageExecutor,
-              messageDeliveryScheduler, ReportSpamTokenProvider.noop()))
+              messageDeliveryScheduler, ReportSpamTokenProvider.noop(), dynamicConfigurationManager))
       .build();
 
   @BeforeEach
@@ -191,6 +196,14 @@ class MessageControllerTest {
     when(accountsManager.getByPhoneNumberIdentifier(MULTI_DEVICE_PNI)).thenReturn(Optional.of(multiDeviceAccount));
     when(accountsManager.getByAccountIdentifier(INTERNATIONAL_UUID)).thenReturn(Optional.of(internationalAccount));
 
+    final DynamicDeliveryLatencyConfiguration deliveryLatencyConfiguration = mock(DynamicDeliveryLatencyConfiguration.class);
+    when(deliveryLatencyConfiguration.instrumentedVersions()).thenReturn(Collections.emptyMap());
+
+    final DynamicConfiguration dynamicConfiguration = mock(DynamicConfiguration.class);
+    when(dynamicConfiguration.getDeliveryLatencyConfiguration()).thenReturn(deliveryLatencyConfiguration);
+
+    when(dynamicConfigurationManager.getConfiguration()).thenReturn(dynamicConfiguration);
+
     when(rateLimiters.getMessagesLimiter()).thenReturn(rateLimiter);
   }
 

service/src/test/java/org/whispersystems/textsecuregcm/push/PushLatencyManagerTest.java

@@ -47,7 +47,7 @@ class PushLatencyManagerTest {
 
     when(dynamicConfigurationManager.getConfiguration()).thenReturn(dynamicConfiguration);
     when(dynamicConfiguration.getPushLatencyConfiguration()).thenReturn(dynamicPushLatencyConfiguration);
-    when(dynamicPushLatencyConfiguration.getInstrumentedVersions()).thenReturn(Collections.emptyMap());
+    when(dynamicPushLatencyConfiguration.instrumentedVersions()).thenReturn(Collections.emptyMap());
   }
 
   @ParameterizedTest

service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionIntegrationTest.java

@@ -25,6 +25,7 @@ import java.io.IOException;
 import java.time.Clock;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
@@ -44,12 +45,15 @@ import org.junit.jupiter.params.provider.CsvSource;
 import org.mockito.ArgumentCaptor;
 import org.mockito.stubbing.Answer;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicDeliveryLatencyConfiguration;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
 import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.Device;
+import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
 import org.whispersystems.textsecuregcm.storage.DynamoDbExtension;
 import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Tables;
 import org.whispersystems.textsecuregcm.storage.MessagesCache;
@@ -79,6 +83,7 @@ class WebSocketConnectionIntegrationTest {
   private Device device;
   private WebSocketClient webSocketClient;
   private Scheduler messageDeliveryScheduler;
+  private DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager;
 
   private long serialTimestamp = System.currentTimeMillis();
 
@@ -98,6 +103,15 @@ class WebSocketConnectionIntegrationTest {
     device = mock(Device.class);
     webSocketClient = mock(WebSocketClient.class);
 
+    final DynamicDeliveryLatencyConfiguration deliveryLatencyConfiguration = mock(DynamicDeliveryLatencyConfiguration.class);
+    when(deliveryLatencyConfiguration.instrumentedVersions()).thenReturn(Collections.emptyMap());
+
+    final DynamicConfiguration dynamicConfiguration = mock(DynamicConfiguration.class);
+    when(dynamicConfiguration.getDeliveryLatencyConfiguration()).thenReturn(deliveryLatencyConfiguration);
+
+    dynamicConfigurationManager = mock(DynamicConfigurationManager.class);
+    when(dynamicConfigurationManager.getConfiguration()).thenReturn(dynamicConfiguration);
+
     when(account.getNumber()).thenReturn("+18005551234");
     when(account.getUuid()).thenReturn(UUID.randomUUID());
     when(device.getId()).thenReturn(1L);
@@ -126,7 +140,7 @@ class WebSocketConnectionIntegrationTest {
         device,
         webSocketClient,
         scheduledExecutorService,
-        messageDeliveryScheduler);
+        messageDeliveryScheduler, dynamicConfigurationManager);
 
     final List<MessageProtos.Envelope> expectedMessages = new ArrayList<>(persistedMessageCount + cachedMessageCount);
 
@@ -210,7 +224,7 @@ class WebSocketConnectionIntegrationTest {
         device,
         webSocketClient,
         scheduledExecutorService,
-        messageDeliveryScheduler);
+        messageDeliveryScheduler, dynamicConfigurationManager);
 
     final int persistedMessageCount = 207;
     final int cachedMessageCount = 173;
@@ -276,7 +290,8 @@ class WebSocketConnectionIntegrationTest {
         webSocketClient,
         100, // use a very short timeout, so that this test completes quickly
         scheduledExecutorService,
-        messageDeliveryScheduler);
+        messageDeliveryScheduler,
+        dynamicConfigurationManager);
 
     final int persistedMessageCount = 207;
     final int cachedMessageCount = 173;

service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionTest.java

@@ -33,6 +33,7 @@ import io.lettuce.core.RedisException;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -52,12 +53,15 @@ import org.junit.jupiter.api.Test;
 import org.mockito.stubbing.Answer;
 import org.whispersystems.textsecuregcm.auth.AccountAuthenticator;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicDeliveryLatencyConfiguration;
 import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.push.PushNotificationManager;
 import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
+import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
 import org.whispersystems.textsecuregcm.util.Pair;
 import org.whispersystems.websocket.WebSocketClient;
@@ -88,6 +92,7 @@ class WebSocketConnectionTest {
   private ReceiptSender receiptSender;
   private ScheduledExecutorService retrySchedulingExecutor;
   private Scheduler messageDeliveryScheduler;
+  private DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager;
 
   @BeforeEach
   void setup() {
@@ -101,6 +106,15 @@ class WebSocketConnectionTest {
     receiptSender = mock(ReceiptSender.class);
     retrySchedulingExecutor = mock(ScheduledExecutorService.class);
     messageDeliveryScheduler = Schedulers.newBoundedElastic(10, 10_000, "messageDelivery");
+
+    final DynamicDeliveryLatencyConfiguration deliveryLatencyConfiguration = mock(DynamicDeliveryLatencyConfiguration.class);
+    when(deliveryLatencyConfiguration.instrumentedVersions()).thenReturn(Collections.emptyMap());
+
+    final DynamicConfiguration dynamicConfiguration = mock(DynamicConfiguration.class);
+    when(dynamicConfiguration.getDeliveryLatencyConfiguration()).thenReturn(deliveryLatencyConfiguration);
+
+    dynamicConfigurationManager = mock(DynamicConfigurationManager.class);
+    when(dynamicConfigurationManager.getConfiguration()).thenReturn(dynamicConfiguration);
   }
 
   @AfterEach
@@ -114,7 +128,7 @@ class WebSocketConnectionTest {
     WebSocketAccountAuthenticator webSocketAuthenticator = new WebSocketAccountAuthenticator(accountAuthenticator);
     AuthenticatedConnectListener connectListener = new AuthenticatedConnectListener(receiptSender, messagesManager,
         mock(PushNotificationManager.class), mock(ClientPresenceManager.class),
-        retrySchedulingExecutor, messageDeliveryScheduler);
+        retrySchedulingExecutor, messageDeliveryScheduler, dynamicConfigurationManager);
     WebSocketSessionContext sessionContext = mock(WebSocketSessionContext.class);
 
     when(accountAuthenticator.authenticate(eq(new BasicCredentials(VALID_USER, VALID_PASSWORD))))
@@ -194,7 +208,7 @@ class WebSocketConnectionTest {
         });
 
     WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager,
-        auth, device, client, retrySchedulingExecutor, Schedulers.immediate());
+        auth, device, client, retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     connection.start();
     verify(client, times(3)).sendRequest(eq("PUT"), eq("/api/v1/message"), any(List.class),
@@ -222,7 +236,7 @@ class WebSocketConnectionTest {
   public void testOnlineSend() {
     final WebSocketClient client = mock(WebSocketClient.class);
     final WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     final UUID accountUuid = UUID.randomUUID();
 
@@ -342,7 +356,7 @@ class WebSocketConnectionTest {
         });
 
     WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager,
-        auth, device, client, retrySchedulingExecutor, Schedulers.immediate());
+        auth, device, client, retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     connection.start();
 
@@ -366,7 +380,7 @@ class WebSocketConnectionTest {
   void testProcessStoredMessageConcurrency() {
     final WebSocketClient client = mock(WebSocketClient.class);
     final WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     when(account.getNumber()).thenReturn("+18005551234");
     when(account.getUuid()).thenReturn(UUID.randomUUID());
@@ -431,7 +445,7 @@ class WebSocketConnectionTest {
   void testProcessStoredMessagesMultiplePages() {
     final WebSocketClient client = mock(WebSocketClient.class);
     final WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     when(account.getNumber()).thenReturn("+18005551234");
     final UUID accountUuid = UUID.randomUUID();
@@ -483,7 +497,7 @@ class WebSocketConnectionTest {
   void testProcessStoredMessagesContainsSenderUuid() {
     final WebSocketClient client = mock(WebSocketClient.class);
     final WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     when(account.getNumber()).thenReturn("+18005551234");
     final UUID accountUuid = UUID.randomUUID();
@@ -545,7 +559,7 @@ class WebSocketConnectionTest {
   void testProcessStoredMessagesSingleEmptyCall() {
     final WebSocketClient client = mock(WebSocketClient.class);
     final WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     final UUID accountUuid = UUID.randomUUID();
 
@@ -574,7 +588,7 @@ class WebSocketConnectionTest {
   public void testRequeryOnStateMismatch() {
     final WebSocketClient client = mock(WebSocketClient.class);
     final WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
     final UUID accountUuid = UUID.randomUUID();
 
     when(account.getNumber()).thenReturn("+18005551234");
@@ -630,7 +644,7 @@ class WebSocketConnectionTest {
   void testProcessCachedMessagesOnly() {
     final WebSocketClient client = mock(WebSocketClient.class);
     final WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     final UUID accountUuid = UUID.randomUUID();
 
@@ -662,7 +676,7 @@ class WebSocketConnectionTest {
   void testProcessDatabaseMessagesAfterPersist() {
     final WebSocketClient client = mock(WebSocketClient.class);
     final WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     final UUID accountUuid = UUID.randomUUID();
 
@@ -709,7 +723,7 @@ class WebSocketConnectionTest {
     when(client.isOpen()).thenReturn(true);
 
     WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
     connection.start();
 
     verify(retrySchedulingExecutor, times(WebSocketConnection.MAX_CONSECUTIVE_RETRIES)).schedule(any(Runnable.class),
@@ -733,7 +747,7 @@ class WebSocketConnectionTest {
     when(client.isOpen()).thenReturn(false);
 
     WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
     connection.start();
 
     verify(retrySchedulingExecutor, never()).schedule(any(Runnable.class), anyLong(), any());
@@ -767,7 +781,7 @@ class WebSocketConnectionTest {
         CompletableFuture.completedFuture(Optional.empty()));
 
     WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, messageDeliveryScheduler);
+        retrySchedulingExecutor, messageDeliveryScheduler, dynamicConfigurationManager);
 
     connection.start();
 
@@ -824,7 +838,7 @@ class WebSocketConnectionTest {
         CompletableFuture.completedFuture(Optional.empty()));
 
     WebSocketConnection connection = new WebSocketConnection(receiptSender, messagesManager, auth, device, client,
-        retrySchedulingExecutor, Schedulers.immediate());
+        retrySchedulingExecutor, Schedulers.immediate(), dynamicConfigurationManager);
 
     connection.start();