From 268c8382ee7bd71b3c0a472494a8120df31ee997 Mon Sep 17 00:00:00 2001
From: Ravi Khadiwala <ravi@signal.org>
Date: Tue, 2 Apr 2024 16:06:39 -0500
Subject: [PATCH] Fix Backup expiration purge time

---
 .../workers/RemoveExpiredBackupsCommand.java  | 35 ++++++++++---------
 1 file changed, 19 insertions(+), 16 deletions(-)

diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/workers/RemoveExpiredBackupsCommand.java b/service/src/main/java/org/whispersystems/textsecuregcm/workers/RemoveExpiredBackupsCommand.java
index 765aace05..1822d85e6 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/workers/RemoveExpiredBackupsCommand.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/workers/RemoveExpiredBackupsCommand.java
@@ -15,6 +15,7 @@ import java.time.Duration;
 import java.util.HexFormat;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 import net.sourceforge.argparse4j.inf.Namespace;
 import net.sourceforge.argparse4j.inf.Subparser;
 import org.slf4j.Logger;
@@ -115,15 +116,14 @@ public class RemoveExpiredBackupsCommand extends EnvironmentCommand<WhisperServe
           throw new RuntimeException(e);
         }
       });
-      final AtomicLong backupsExpired = new AtomicLong();
       final BackupManager backupManager = commandDependencies.backupManager();
-      backupManager
-          .getExpiredBackups(segments, Schedulers.parallel(), clock.instant().plus(gracePeriod))
+      final long backupsExpired = backupManager
+          .getExpiredBackups(segments, Schedulers.parallel(), clock.instant().minus(gracePeriod))
           .flatMap(expiredBackup -> removeExpiredBackup(backupManager, expiredBackup, dryRun), concurrency)
-          .doOnNext(ignored -> backupsExpired.incrementAndGet())
-          .then()
+          .filter(Boolean.TRUE::equals)
+          .count()
           .block();
-      logger.info("Expired {} backups", backupsExpired.get());
+      logger.info("Expired {} backups", backupsExpired);
     } finally {
       environment.lifecycle().getManagedObjects().forEach(managedObject -> {
         try {
@@ -135,28 +135,31 @@ public class RemoveExpiredBackupsCommand extends EnvironmentCommand<WhisperServe
     }
   }
 
-  private Mono<Void> removeExpiredBackup(
+  private Mono<Boolean> removeExpiredBackup(
       final BackupManager backupManager, final ExpiredBackup expiredBackup,
       final boolean dryRun) {
 
-    final Mono<Void> mono;
+    final Mono<Boolean> mono;
     if (dryRun) {
-      mono = Mono.empty();
+      mono = Mono.just(true);
     } else {
-      mono = Mono.fromCompletionStage(() -> backupManager.expireBackup(expiredBackup));
+      mono = Mono.fromCompletionStage(() -> backupManager.expireBackup(expiredBackup)).map(ignore -> true);
     }
 
     return mono
-        .doOnSuccess(ignored -> Metrics
-            .counter(EXPIRED_BACKUPS_COUNTER_NAME,
-                "tier", expiredBackup.expirationType().name(),
-                "dryRun", String.valueOf(dryRun))
-            .increment())
+        .doOnSuccess(ignored -> {
+          logger.info("incrementing metric for {}", HexFormat.of().formatHex(expiredBackup.hashedBackupId()));
+          Metrics
+              .counter(EXPIRED_BACKUPS_COUNTER_NAME,
+                  "tier", expiredBackup.expirationType().name(),
+                  "dryRun", String.valueOf(dryRun))
+              .increment();
+        })
         .onErrorResume(throwable -> {
           logger.warn("Failed to remove tier {} for backup {}",
               expiredBackup.expirationType(),
               HexFormat.of().formatHex(expiredBackup.hashedBackupId()));
-          return Mono.empty();
+          return Mono.just(false);
         });
   }