Add opt-in all-version purge to named-blob cleanup

README.md

@@ -125,3 +125,55 @@ mysql> USE AmbryRepairRequests; SOURCE ./ambry-mysql/src/main/resources/AmbryRep
 ```
 
 Then run `./gradlew build` to build and run all unit tests.
+
+## Named-blob cleaner purge (review proposal)
+
+Normal named-blob cleanup retains the latest version. The optional
+`NamedBlobsCleanupRunner` constructor accepting a `Predicate<Container> purgeEligibility`
+and `boolean purgeDryRun` can instead delete every unexpired, non-deleted mapping in
+selected containers, including latest READY versions and recent IN_PROGRESS versions.
+Existing constructors never enable purge. Container/account exclusions and disabled
+named-blob mode still take precedence; container lifecycle status is not changed.
+
+The caller owns the eligibility policy. It must require explicit per-container opt-in,
+verify that the source data is no longer needed, and prevent source reads and writes
+before returning `true`. For a migration, write routing alone is not proof of read
+cutover or destination completeness. Keep this policy live so removing approval
+acts as a kill switch; it is checked before every page and every blob deletion.
+A check cannot cancel an already-issued deletion, so it does not replace fencing
+source traffic. Start with `purgeDryRun=true`: this reports only the first bounded page
+per eligible container, not a full-container count, and performs no blob or metadata
+deletion for those containers. Other containers continue normal cleanup.
+
+`MySqlNamedBlobDb` scans purge candidates in ascending primary-key order rather than
+normal cleanup's descending version order. The last blob name remains an inclusive
+cursor so a name with more versions than a page is revisited. Purge waits for metadata
+soft-deletion before fetching another page and stops on deletion or metadata failure,
+or if the next page repeats versions that should have been removed;
+the next scheduled run restarts its purge sweep. Unlike normal cleanup, it never skips
+a name after repeated query kills. Primary-key ordering avoids the mixed-order filesort,
+but filtering already-deleted rows can still require scanning many index entries:
+this is not a fixed query-latency guarantee or a fix for normal-cleanup scans.
+
+Database implementations without purge support fail explicitly through the new default
+`NamedBlobDb.pullStaleBlobs` overload. Blob deletion continues to use the supplied
+router's existing offline `deleteBlob(blobId, serviceId)` path. This OSS change does not
+guarantee primary-only routing in a composite router; that must be verified by its
+integration. Migration-state/completeness checks, deployment configuration, composite
+database delegation, and any secondary metadata cleanup are deliberately outside this
+OSS proposal. Expired/already-deleted mappings remain under their existing cleanup
+mechanisms; metadata soft-deletion does not itself immediately reclaim storage bytes.
+
+`PurgePreviewBlobCount` counts dry-run candidates and `PurgedBlobCount` counts versions
+processed after metadata cleanup succeeds. Failures use the existing container failure
+counter and logs. These are cumulative counters, not distinct-blob inventories.
+
+Focused validation (the integration tests require the local MySQL setup above and the
+`AmbryNamedBlobs` schema from `ambry-named-mysql/src/main/resources/NamedBlobsSchema.ddl`):
+
+```bash
+./gradlew -PdisableShipkit :ambry-frontend:test --tests com.github.ambry.frontend.NamedBlobsCleanupRunnerTest \
+  :ambry-named-mysql:test --tests com.github.ambry.named.MySqlNamedBlobDbTest
+./gradlew -PdisableShipkit :ambry-named-mysql:intTest \
+  --tests 'com.github.ambry.named.MySqlNamedBlobDbIntegrationTest.testPurge*'
+```

ambry-api/src/main/java/com/github/ambry/named/NamedBlobDb.java

@@ -119,6 +119,27 @@ default CompletableFuture<PutResult> put(NamedBlobRecord record) {
    */
   CompletableFuture<StaleBlobsWithLatestBlobName> pullStaleBlobs(Container container, String blobName, int maxResults);
 
+  /**
+   * Pull a cleanup page, optionally treating every unexpired, non-deleted version (including the latest) as stale.
+   * Purge callers must prevent reads and writes to the source container and verify that its data is no longer needed.
+   * In purge mode the next blob name is inclusive: callers must finish deleting and soft-deleting a page before
+   * fetching the next one, and stop on failure, since a single name can span multiple pages.
+   * @param container the container to clean.
+   * @param blobName inclusive starting blob name.
+   * @param maxResults maximum rows to scan, or {@code 0} for the database default.
+   * @param purgeAllVersions whether to bypass latest-version and retention protections.
+   * @return a cleanup page; implementations without purge support fail explicitly when purge is requested.
+   */
+  default CompletableFuture<StaleBlobsWithLatestBlobName> pullStaleBlobs(Container container, String blobName,
+      int maxResults, boolean purgeAllVersions) {
+    if (purgeAllVersions) {
+      CompletableFuture<StaleBlobsWithLatestBlobName> future = new CompletableFuture<>();
+      future.completeExceptionally(new UnsupportedOperationException("Named blob purge is not supported"));
+      return future;
+    }
+    return maxResults == 0 ? pullStaleBlobs(container, blobName) : pullStaleBlobs(container, blobName, maxResults);
+  }
+
   /**
    * Returns the first (lowest) blob name in the container at or after {@code blobNameFrom}, or {@code null} if the
    * container has no such blob. A cheap index-only point lookup used by the cleanup runner to advance its cursor past a

ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobsCleanupRunner.java

@@ -37,12 +37,14 @@
 import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -69,6 +71,10 @@ public class NamedBlobsCleanupRunner implements Runnable {
   private final Set<String> excludedContainers;
   private final Set<String> excludedAccounts;
   private final Time time;
+  private final Predicate<Container> purgeEligibility;
+  private final boolean purgeDryRun;
+  private final Counter purgePreviewBlobCount;
+  private final Counter purgedBlobCount;
   /**
    * Per-container resume cursor keyed by "{accountId}_{containerId}", retained in memory across scheduled runs so a
    * container whose scan was killed and deferred resumes from its last processed page on the next run instead of
@@ -138,6 +144,19 @@ public NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountSe
         SystemTime.getInstance(), metricRegistry);
   }
 
+  /**
+   * Construct a cleaner with an opt-in policy for purging all versions, including live/latest versions.
+   * Exclusions take precedence. The policy is rechecked before each page and each delete and must verify that
+   * source reads/writes have stopped and the data is no longer needed. External migration/completeness checks
+   * belong in this policy, not in the OSS cleaner. A dry run previews only the first bounded page per container.
+   */
+  public NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountService accountService,
+      int containerDelaySeconds, Collection<String> excludedContainers, Collection<String> excludedAccounts,
+      Predicate<Container> purgeEligibility, boolean purgeDryRun, MetricRegistry metricRegistry) {
+    this(router, namedBlobDb, accountService, containerDelaySeconds, excludedContainers, excludedAccounts,
+        SystemTime.getInstance(), metricRegistry, purgeEligibility, purgeDryRun);
+  }
+
   NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountService accountService,
       int containerDelaySeconds, Time time) {
     this(router, namedBlobDb, accountService, containerDelaySeconds, Collections.emptySet(), time);
@@ -159,6 +178,13 @@ public NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountSe
   NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountService accountService,
       int containerDelaySeconds, Collection<String> excludedContainers, Collection<String> excludedAccounts, Time time,
       MetricRegistry metricRegistry) {
+    this(router, namedBlobDb, accountService, containerDelaySeconds, excludedContainers, excludedAccounts, time,
+        metricRegistry, container -> false, true);
+  }
+
+  NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountService accountService,
+      int containerDelaySeconds, Collection<String> excludedContainers, Collection<String> excludedAccounts, Time time,
+      MetricRegistry metricRegistry, Predicate<Container> purgeEligibility, boolean purgeDryRun) {
     if (containerDelaySeconds < 0) {
       throw new IllegalArgumentException("containerDelaySeconds must not be negative");
     }
@@ -170,6 +196,12 @@ public NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountSe
         excludedContainers == null ? Collections.emptySet() : new HashSet<>(excludedContainers);
     this.excludedAccounts = excludedAccounts == null ? Collections.emptySet() : new HashSet<>(excludedAccounts);
     this.time = time;
+    this.purgeEligibility = Objects.requireNonNull(purgeEligibility, "purgeEligibility");
+    this.purgeDryRun = purgeDryRun;
+    this.purgePreviewBlobCount =
+        metricRegistry.counter(MetricRegistry.name(NamedBlobsCleanupRunner.class, "PurgePreviewBlobCount"));
+    this.purgedBlobCount =
+        metricRegistry.counter(MetricRegistry.name(NamedBlobsCleanupRunner.class, "PurgedBlobCount"));
     this.containerCleanupFailedCount =
         metricRegistry.counter(MetricRegistry.name(NamedBlobsCleanupRunner.class, "ContainerFailedCount"));
     this.pageScanKilledCount =
@@ -254,6 +286,12 @@ public void run() {
    */
   private void cleanupContainer(Container container) throws Exception {
     String cursorKey = cursorKey(container);
+    boolean purgeAllVersions = purgeEligibility.test(container);
+    if (purgeAllVersions) {
+      // Purge and normal cleanup have different retention semantics; never reuse a normal cleanup cursor for purge.
+      containerCleanupCursors.remove(cursorKey);
+      logger.warn("All-version purge selected for container {} (dry run: {})", cursorKey, purgeDryRun);
+    }
     // Resume from the cursor saved by a previous run (if this container was deferred after a kill); otherwise start at
     // "\0", the lowest ASCII value, so the scan begins at the start of the container.
     String blobName = containerCleanupCursors.getOrDefault(cursorKey, smallestASCII);
@@ -264,10 +302,14 @@ private void cleanupContainer(Container container) throws Exception {
     }
     int pageSize = 0;  // 0 means the database default (full-size) page.
     int blobNameSkips = 0;
+    StaleNamedBlob lastPurgedBlob = null;
     NamedBlobDb.StaleBlobsWithLatestBlobName staleBlobsWithLatestBlobName;
     do {
+      if (purgeAllVersions && !purgeEligibility.test(container)) {
+        throw new IllegalStateException("Purge eligibility revoked for container " + cursorKey);
+      }
       try {
-        staleBlobsWithLatestBlobName = pullStaleBlobsResilient(container, blobName, pageSize);
+        staleBlobsWithLatestBlobName = pullStaleBlobsResilient(container, blobName, pageSize, purgeAllVersions);
       } catch (PageKilledException e) {
         if (pageSize == 0) {
           // First kill at full size: shrink and retry the same cursor so the scan reads fewer rows and the cursor can
@@ -277,6 +319,9 @@ private void cleanupContainer(Container container) throws Exception {
               + "progress instead of retrying the same query", container.getId(), pageSize);
           continue;
         }
+        if (purgeAllVersions) {
+          throw new IllegalStateException("Purge scan failed at the shrunk page size for container " + cursorKey, e);
+        }
         // Even the shrunk page was killed: the blob name at this cursor has too many live versions to scan under the
         // database time limit. Skip past it with a cheap indexed lookup so the rest of the container is still cleaned;
         // the skipped blob name is left to a targeted reap / TTL / version cap. Bound the skips per run so a broadly
@@ -312,11 +357,29 @@ private void cleanupContainer(Container container) throws Exception {
             MAX_BLOB_NAME_SKIPS_PER_RUN);
         continue;
       }
-      List<StaleNamedBlob> batchStaleBlobs = staleBlobsWithLatestBlobName.getStaleBlobs();
+      List<StaleNamedBlob> batchStaleBlobs = new ArrayList<>(staleBlobsWithLatestBlobName.getStaleBlobs());
+      if (purgeAllVersions && lastPurgedBlob != null && !batchStaleBlobs.isEmpty()) {
+        StaleNamedBlob firstBlob = batchStaleBlobs.get(0);
+        if (firstBlob.getBlobName().equals(lastPurgedBlob.getBlobName())
+            && firstBlob.getVersion() <= lastPurgedBlob.getVersion()) {
+          throw new IllegalStateException("Purge page did not advance after metadata cleanup for container " + cursorKey);
+        }
+      }
+      if (purgeAllVersions && purgeDryRun) {
+        purgePreviewBlobCount.inc(batchStaleBlobs.size());
+        logger.warn("Purge dry run for container {}: first page has {} candidates; more pages: {}", cursorKey,
+            batchStaleBlobs.size(), staleBlobsWithLatestBlobName.getLatestBlob() != null);
+        return;
+      }
       List<StaleNamedBlob> failedResults = new ArrayList<>();
       for (StaleNamedBlob staleBlob : batchStaleBlobs) {
+        if (purgeAllVersions && !purgeEligibility.test(container)) {
+          throw new IllegalStateException("Purge eligibility revoked for container " + cursorKey);
+        }
         try {
           router.deleteBlob(staleBlob.getBlobId(), "ambry-named-blobs-cleanup-runner").get();
+        } catch (InterruptedException e) {
+          throw e;
         } catch (Exception e) {
           if (e.getMessage() == null || !e.getMessage().contains(RouterErrorCode.BlobDoesNotExist.name())) {
             logger.error("Failed to cleanup named stale blob {}", staleBlob, e);
@@ -326,7 +389,18 @@ private void cleanupContainer(Container container) throws Exception {
       }
 
       batchStaleBlobs.removeAll(failedResults);
-      namedBlobDb.cleanupStaleData(batchStaleBlobs);
+      CompletableFuture<Integer> cleanupFuture = namedBlobDb.cleanupStaleData(batchStaleBlobs);
+      if (purgeAllVersions) {
+        // The next scan revisits the last name; it must see these rows soft-deleted before it can make progress.
+        cleanupFuture.get(PULL_STALE_BLOBS_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+        purgedBlobCount.inc(batchStaleBlobs.size());
+        if (!failedResults.isEmpty()) {
+          throw new IllegalStateException("Purge blob deletions failed for container " + cursorKey);
+        }
+        if (!batchStaleBlobs.isEmpty()) {
+          lastPurgedBlob = batchStaleBlobs.get(batchStaleBlobs.size() - 1);
+        }
+      }
 
       if (!batchStaleBlobs.isEmpty()) {
         logger.info("Named Blobs Cleanup Runner processed {} stale blobs ({} failed deletions)",
@@ -367,12 +441,13 @@ private static String cursorKey(Container container) {
    * @throws Exception if a retriable failure persists after {@link #MAX_PULL_ATTEMPTS} attempts.
    */
   private NamedBlobDb.StaleBlobsWithLatestBlobName pullStaleBlobsResilient(Container container, String blobName,
-      int pageSize) throws Exception {
+      int pageSize, boolean purgeAllVersions) throws Exception {
     Exception lastException = null;
     for (int attempt = 1; attempt <= MAX_PULL_ATTEMPTS; attempt++) {
       try {
         CompletableFuture<NamedBlobDb.StaleBlobsWithLatestBlobName> future =
-            pageSize > 0 ? namedBlobDb.pullStaleBlobs(container, blobName, pageSize)
+            purgeAllVersions ? namedBlobDb.pullStaleBlobs(container, blobName, pageSize, true)
+                : pageSize > 0 ? namedBlobDb.pullStaleBlobs(container, blobName, pageSize)
                 : namedBlobDb.pullStaleBlobs(container, blobName);
         return future.get(PULL_STALE_BLOBS_TIMEOUT_SECONDS, TimeUnit.SECONDS);
       } catch (InterruptedException e) {