Remove MessagePersister from WhisperServerService environment

Persistence is now exclusively done by a separate command.

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -163,7 +163,6 @@
 import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
 import org.whispersystems.textsecuregcm.storage.IssuedReceiptsManager;
 import org.whispersystems.textsecuregcm.storage.KeysManager;
-import org.whispersystems.textsecuregcm.storage.MessagePersister;
 import org.whispersystems.textsecuregcm.storage.MessagesCache;
 import org.whispersystems.textsecuregcm.storage.MessagesDynamoDb;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
@@ -547,9 +546,6 @@ public void run(WhisperServerConfiguration config, Environment environment) thro
     RateLimitChallengeManager rateLimitChallengeManager = new RateLimitChallengeManager(pushChallengeManager,
         captchaChecker, rateLimiters);
 
-    MessagePersister messagePersister = new MessagePersister(messagesCache, messagesManager, accountsManager,
-        dynamicConfigurationManager, Duration.ofMinutes(config.getMessageCacheConfiguration().getPersistDelayMinutes()),
-        Optional.empty());
     ChangeNumberManager changeNumberManager = new ChangeNumberManager(messageSender, accountsManager);
 
     HttpClient currencyClient = HttpClient.newBuilder().version(HttpClient.Version.HTTP_2).connectTimeout(Duration.ofSeconds(10)).build();
@@ -562,7 +558,6 @@ public void run(WhisperServerConfiguration config, Environment environment) thro
     environment.lifecycle().manage(apnPushNotificationScheduler);
     environment.lifecycle().manage(provisioningManager);
     environment.lifecycle().manage(messagesCache);
-    environment.lifecycle().manage(messagePersister);
     environment.lifecycle().manage(clientPresenceManager);
     environment.lifecycle().manage(currencyManager);
     environment.lifecycle().manage(registrationServiceClient);

service/src/main/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicMessagePersisterConfiguration.java

@@ -10,16 +10,9 @@
 public class DynamicMessagePersisterConfiguration {
 
   @JsonProperty
-  private boolean serverPersistenceEnabled = true;
+  private boolean persistenceEnabled = true;
 
-  @JsonProperty
-  private boolean dedicatedProcessEnabled = false;
-
-  public boolean isServerPersistenceEnabled() {
-    return serverPersistenceEnabled;
-  }
-
-  public boolean isDedicatedProcessEnabled() {
-    return dedicatedProcessEnabled;
+  public boolean isPersistenceEnabled() {
+    return persistenceEnabled;
   }
 }

service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagePersister.java

@@ -22,7 +22,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
-import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicMessagePersisterConfiguration;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
 import org.whispersystems.textsecuregcm.util.Constants;
 import org.whispersystems.textsecuregcm.util.Util;
@@ -52,8 +51,6 @@ public class MessagePersister implements Managed {
 
   private static final long EXCEPTION_PAUSE_MILLIS = Duration.ofSeconds(3).toMillis();
 
-  private static final int DEFAULT_WORKER_THREAD_COUNT = 4;
-
   private static final int CONSECUTIVE_EMPTY_CACHE_REMOVAL_LIMIT = 3;
 
   private static final Logger logger = LoggerFactory.getLogger(MessagePersister.class);
@@ -62,19 +59,19 @@ public MessagePersister(final MessagesCache messagesCache, final MessagesManager
       final AccountsManager accountsManager,
       final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager,
       final Duration persistDelay,
-      final Optional<Integer> dedicatedProcessWorkerThreadCount) {
+      final int dedicatedProcessWorkerThreadCount) {
     this.messagesCache = messagesCache;
     this.messagesManager = messagesManager;
     this.accountsManager = accountsManager;
     this.persistDelay = persistDelay;
-    this.workerThreads = dedicatedProcessWorkerThreadCount.map(Thread[]::new)
-        .orElseGet(() -> new Thread[DEFAULT_WORKER_THREAD_COUNT]);
-    this.dedicatedProcess = dedicatedProcessWorkerThreadCount.isPresent();
+    this.workerThreads = new Thread[dedicatedProcessWorkerThreadCount];
+    this.dedicatedProcess = true;
 
     for (int i = 0; i < workerThreads.length; i++) {
       workerThreads[i] = new Thread(() -> {
         while (running) {
-          if (enabled(dynamicConfigurationManager)) {
+          if (dynamicConfigurationManager.getConfiguration().getMessagePersisterConfiguration()
+              .isPersistenceEnabled()) {
             try {
               final int queuesPersisted = persistNextQueues(Instant.now());
               queueCountHistogram.update(queuesPersisted);
@@ -94,17 +91,6 @@ public MessagePersister(final MessagesCache messagesCache, final MessagesManager
     }
   }
 
-  @VisibleForTesting
-  boolean enabled(final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager) {
-    final DynamicMessagePersisterConfiguration messagePersisterConfiguration = dynamicConfigurationManager.getConfiguration()
-        .getMessagePersisterConfiguration();
-    if (dedicatedProcess) {
-      return messagePersisterConfiguration.isDedicatedProcessEnabled();
-    }
-
-    return messagePersisterConfiguration.isServerPersistenceEnabled();
-  }
-
   @VisibleForTesting
   Duration getPersistDelay() {
     return persistDelay;

service/src/main/java/org/whispersystems/textsecuregcm/workers/MessagePersisterServiceCommand.java

@@ -9,7 +9,6 @@
 import io.dropwizard.cli.ServerCommand;
 import io.dropwizard.setup.Environment;
 import java.time.Duration;
-import java.util.Optional;
 import net.sourceforge.argparse4j.inf.Namespace;
 import net.sourceforge.argparse4j.inf.Subparser;
 import org.whispersystems.textsecuregcm.WhisperServerConfiguration;
@@ -65,7 +64,7 @@ protected void run(Environment environment, Namespace namespace, WhisperServerCo
         deps.accountsManager(),
         dynamicConfigurationManager,
         Duration.ofMinutes(configuration.getMessageCacheConfiguration().getPersistDelayMinutes()),
-        Optional.of(namespace.getInt(WORKER_COUNT)));
+        namespace.getInt(WORKER_COUNT));
 
     environment.lifecycle().manage(deps.messagesCache());
     environment.lifecycle().manage(messagePersister);

service/src/test/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicConfigurationTest.java

@@ -359,37 +359,34 @@ void testMessagePersister() throws JsonProcessingException {
       final DynamicConfiguration emptyConfig =
           DynamicConfigurationManager.parseConfiguration(emptyConfigYaml, DynamicConfiguration.class).orElseThrow();
 
-      assertTrue(emptyConfig.getMessagePersisterConfiguration().isServerPersistenceEnabled());
-      assertFalse(emptyConfig.getMessagePersisterConfiguration().isDedicatedProcessEnabled());
+      assertTrue(emptyConfig.getMessagePersisterConfiguration().isPersistenceEnabled());
     }
 
     {
       final String messagePersisterEnabledYaml = REQUIRED_CONFIG.concat("""
           messagePersister:
-            serverPersistenceEnabled: true
+            persistenceEnabled: true
             dedicatedProcessEnabled: true
           """);
 
       final DynamicConfiguration config =
           DynamicConfigurationManager.parseConfiguration(messagePersisterEnabledYaml, DynamicConfiguration.class)
               .orElseThrow();
 
-      assertTrue(config.getMessagePersisterConfiguration().isServerPersistenceEnabled());
-      assertTrue(config.getMessagePersisterConfiguration().isDedicatedProcessEnabled());
+      assertTrue(config.getMessagePersisterConfiguration().isPersistenceEnabled());
     }
 
     {
       final String messagePersisterDisabledYaml = REQUIRED_CONFIG.concat("""
           messagePersister:
-            serverPersistenceEnabled: false
+            persistenceEnabled: false
           """);
 
       final DynamicConfiguration config =
           DynamicConfigurationManager.parseConfiguration(messagePersisterDisabledYaml, DynamicConfiguration.class)
               .orElseThrow();
 
-      assertFalse(config.getMessagePersisterConfiguration().isServerPersistenceEnabled());
-      assertFalse(config.getMessagePersisterConfiguration().isDedicatedProcessEnabled());
+      assertFalse(config.getMessagePersisterConfiguration().isPersistenceEnabled());
     }
   }
 

service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterIntegrationTest.java

@@ -83,7 +83,7 @@ void setUp() throws Exception {
     messagesManager = new MessagesManager(messagesDynamoDb, messagesCache, mock(ReportMessageManager.class),
         messageDeletionExecutorService);
     messagePersister = new MessagePersister(messagesCache, messagesManager, accountsManager,
-        dynamicConfigurationManager, PERSIST_DELAY, Optional.empty());
+        dynamicConfigurationManager, PERSIST_DELAY, 1);
 
     account = mock(Account.class);
 

service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterTest.java

@@ -37,12 +37,9 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.CsvSource;
 import org.mockito.ArgumentCaptor;
 import org.mockito.stubbing.Answer;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
-import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicMessagePersisterConfiguration;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
 import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
 import reactor.core.scheduler.Scheduler;
@@ -91,7 +88,7 @@ void setUp() throws Exception {
         REDIS_CLUSTER_EXTENSION.getRedisCluster(), sharedExecutorService, messageDeliveryScheduler,
         sharedExecutorService, Clock.systemUTC());
     messagePersister = new MessagePersister(messagesCache, messagesManager, accountsManager,
-        dynamicConfigurationManager, PERSIST_DELAY, Optional.empty());
+        dynamicConfigurationManager, PERSIST_DELAY, 1);
 
     doAnswer(invocation -> {
       final UUID destinationUuid = invocation.getArgument(0);
@@ -229,31 +226,6 @@ void testPersistQueueRetryLoop() {
             () -> messagePersister.persistQueue(DESTINATION_ACCOUNT_UUID, DESTINATION_DEVICE_ID)));
   }
 
-  @ParameterizedTest
-  @CsvSource({
-      "true, true, false, false",
-      "true, false, true, true",
-      "false, true, false, true",
-      "false, false, true, false",
-  })
-  void testEnabled(final boolean dedicatedProcess, final boolean serverPersistenceEnabled,
-      final boolean dedicatedProcessEnabled, final boolean expectEnabled) {
-    final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager = mock(
-        DynamicConfigurationManager.class);
-    final DynamicConfiguration dynamicConfiguration = mock(DynamicConfiguration.class);
-    when(dynamicConfigurationManager.getConfiguration()).thenReturn(dynamicConfiguration);
-
-    final DynamicMessagePersisterConfiguration dynamicMessagePersisterConfiguration = mock(
-        DynamicMessagePersisterConfiguration.class);
-    when(dynamicConfiguration.getMessagePersisterConfiguration()).thenReturn(dynamicMessagePersisterConfiguration);
-    when(dynamicMessagePersisterConfiguration.isDedicatedProcessEnabled()).thenReturn(dedicatedProcessEnabled);
-    when(dynamicMessagePersisterConfiguration.isServerPersistenceEnabled()).thenReturn(serverPersistenceEnabled);
-
-    messagePersister = new MessagePersister(messagesCache, messagesManager, accountsManager,
-        dynamicConfigurationManager, PERSIST_DELAY, dedicatedProcess ? Optional.of(4) : Optional.empty());
-    assertEquals(expectEnabled, messagePersister.enabled(dynamicConfigurationManager));
-  }
-
   @SuppressWarnings("SameParameterValue")
   private static String generateRandomQueueNameForSlot(final int slot) {
     final UUID uuid = UUID.randomUUID();