Skip past un-scannable blob names so bloated named-blob containers still make progress

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

@@ -119,6 +119,16 @@ default CompletableFuture<PutResult> put(NamedBlobRecord record) {
    */
   CompletableFuture<StaleBlobsWithLatestBlobName> pullStaleBlobs(Container container, String blobName, int 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
+   * blob name whose stale-version scan repeatedly crosses the database long-transaction limit, so the rest of the
+   * container can still be cleaned.
+   * @param container the container to look in.
+   * @param blobNameFrom the inclusive lower-bound blob name.
+   */
+  CompletableFuture<String> getFirstBlobName(Container container, String blobNameFrom);
+
   /**
    * Cleanup the stale blobs records
    */

ambry-commons/src/main/java/com/github/ambry/commons/InMemNamedBlobDb.java

@@ -227,6 +227,23 @@ public CompletableFuture<StaleBlobsWithLatestBlobName> pullStaleBlobs(Container
     return pullStaleBlobs(container, latestBlob);
   }
 
+  @Override
+  public CompletableFuture<String> getFirstBlobName(Container container, String blobNameFrom) {
+    String containerName = container.getName();
+    String first = null;
+    for (String accountName : allRecords.keySet()) {
+      TreeMap<String, List<NamedBlobRow>> rowsPerContainer = allRecords.get(accountName).get(containerName);
+      if (rowsPerContainer == null) {
+        continue;
+      }
+      String candidate = rowsPerContainer.ceilingKey(blobNameFrom);
+      if (candidate != null && (first == null || candidate.compareTo(first) < 0)) {
+        first = candidate;
+      }
+    }
+    return CompletableFuture.completedFuture(first);
+  }
+
   @Override
   public CompletableFuture<StaleBlobsWithLatestBlobName> pullStaleBlobs(Container container, String latestBlob) {
     CompletableFuture<StaleBlobsWithLatestBlobName> future = new CompletableFuture<>();

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

@@ -78,6 +78,11 @@ public class NamedBlobsCleanupRunner implements Runnable {
   private final Counter containerCleanupFailedCount;
   /** Increments each time a stale-blob page scan is killed by the DB long-transaction limit (full-size or shrunk). */
   private final Counter pageScanKilledCount;
+  /**
+   * Increments each time the runner skips past a single blob name whose stale-version scan cannot complete under the
+   * database time limit even at the shrunk page size, so the rest of the container can still be cleaned.
+   */
+  private final Counter blobNameSkippedCount;
 
   /**
    * Maximum number of attempts for a single stale-blob page scan before the container is deferred to the next
@@ -94,6 +99,14 @@ public class NamedBlobsCleanupRunner implements Runnable {
    * database time limit, and the cursor keeps moving forward through the container.
    */
   private static final int SHRUNK_PAGE_SIZE = 50;
+  /**
+   * Maximum number of blob names the runner skips past in one container in a single run when even the shrunk page scan
+   * is killed. Bounds the database load (each skip follows a killed scan) while still letting the rest of the container
+   * make progress; any remaining skips happen on later runs, which resume from the saved cursor.
+   */
+  private static final int MAX_BLOB_NAME_SKIPS_PER_RUN = 3;
+  /** Client-side wait for the cheap indexed lookup that finds the next blob name to skip to. */
+  private static final long GET_FIRST_BLOB_NAME_TIMEOUT_SECONDS = 30L;
 
   public NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountService accountService) {
     this(router, namedBlobDb, accountService, 0);
@@ -141,6 +154,8 @@ public NamedBlobsCleanupRunner(Router router, NamedBlobDb namedBlobDb, AccountSe
         metricRegistry.counter(MetricRegistry.name(NamedBlobsCleanupRunner.class, "ContainerFailedCount"));
     this.pageScanKilledCount =
         metricRegistry.counter(MetricRegistry.name(NamedBlobsCleanupRunner.class, "PageScanKilledCount"));
+    this.blobNameSkippedCount =
+        metricRegistry.counter(MetricRegistry.name(NamedBlobsCleanupRunner.class, "BlobNameSkippedCount"));
     String cursorMapSizeGauge = MetricRegistry.name(NamedBlobsCleanupRunner.class, "CursorMapSize");
     metricRegistry.remove(cursorMapSizeGauge);
     metricRegistry.register(cursorMapSizeGauge, (Gauge<Integer>) containerCleanupCursors::size);
@@ -228,22 +243,43 @@ private void cleanupContainer(Container container) throws Exception {
       logger.info("Resuming cleanup of container {} from a saved cursor after a previous deferral", container.getId());
     }
     int pageSize = 0;  // 0 means the database default (full-size) page.
+    int blobNameSkips = 0;
     NamedBlobDb.StaleBlobsWithLatestBlobName staleBlobsWithLatestBlobName;
     do {
       try {
         staleBlobsWithLatestBlobName = pullStaleBlobsResilient(container, blobName, pageSize);
       } catch (PageKilledException e) {
-        if (pageSize != 0) {
-          // Already at the shrunk page size and still killed: cannot make progress on this container right now.
-          // Defer it; the next scheduled run resumes from this same cursor.
+        if (pageSize == 0) {
+          // First kill at full size: shrink and retry the same cursor so the scan reads fewer rows and the cursor can
+          // still move forward, rather than re-running the same heavy query (which only adds load).
+          pageSize = SHRUNK_PAGE_SIZE;
+          logger.warn("Stale-blob scan for container {} was killed at its cursor; shrinking the page to {} to make "
+              + "progress instead of retrying the same query", container.getId(), pageSize);
+          continue;
+        }
+        // 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
+        // bloated container cannot spin on many killed scans in a single run.
+        if (blobNameSkips >= MAX_BLOB_NAME_SKIPS_PER_RUN) {
+          throw new IllegalStateException("Stale-blob scan for container " + container.getId()
+              + " was killed even at the shrunk page size; deferring after " + blobNameSkips + " skips this run", e);
+        }
+        String stuckBlobName =
+            namedBlobDb.getFirstBlobName(container, blobName).get(GET_FIRST_BLOB_NAME_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+        if (stuckBlobName == null) {
+          // Nothing at or after the cursor to skip to: defer rather than drop the cursor, so the next run resumes here.
           throw new IllegalStateException(
               "Stale-blob scan for container " + container.getId() + " was killed even at the shrunk page size", e);
         }
-        // Shrink and retry the same cursor so the scan reads fewer rows and the cursor can still move forward,
-        // rather than re-running the same heavy query (which only adds load) or sitting on the page across runs.
-        pageSize = SHRUNK_PAGE_SIZE;
-        logger.warn("Stale-blob scan for container {} was killed at its cursor; shrinking the page to {} to make "
-            + "progress instead of retrying the same query", container.getId(), pageSize);
+        blobName = stuckBlobName + smallestASCII;  // advance strictly past the stuck blob name
+        blobNameSkips++;
+        blobNameSkippedCount.inc();
+        pageSize = 0;  // resume full-size scanning for the region after the skipped blob name
+        containerCleanupCursors.put(cursorKey, blobName);
+        logger.warn("Stale-blob scan for container {} was killed even at the shrunk page size; skipping blob name {} "
+            + "(skip {} of {}) and resuming after it", container.getId(), stuckBlobName, blobNameSkips,
+            MAX_BLOB_NAME_SKIPS_PER_RUN);
         continue;
       }
       List<StaleNamedBlob> batchStaleBlobs = staleBlobsWithLatestBlobName.getStaleBlobs();

ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobsCleanupRunnerTest.java

@@ -155,6 +155,34 @@ public void testKilledScanDefersContainerWithoutAbortingRunOrScheduler() {
     verify(namedBlobDb, times(1)).pullStaleBlobs(goodContainer, FIRST_BLOB_NAME);
   }
 
+  @Test
+  public void testKilledShrunkScanSkipsBloatedBlobNameAndContinues() {
+    // When even the shrunk page is killed, the runner looks up the offending blob name and skips past it, resuming
+    // after it so the rest of the container is still cleaned instead of making no progress at all.
+    Container container = mockContainer((short) 1, Container.NamedBlobMode.OPTIONAL, (short) 100, "container-a");
+    AccountService accountService = mock(AccountService.class);
+    when(accountService.getContainersByStatus(Container.ContainerStatus.ACTIVE)).thenReturn(
+        Collections.singleton(container));
+    when(accountService.getContainersByStatus(Container.ContainerStatus.INACTIVE)).thenReturn(Collections.emptySet());
+
+    NamedBlobDb namedBlobDb = mock(NamedBlobDb.class);
+    CompletableFuture<NamedBlobDb.StaleBlobsWithLatestBlobName> killedScan = new CompletableFuture<>();
+    killedScan.completeExceptionally(new SQLException("Query execution was interrupted", "70100", 1317));
+    // The blob name at the start of the container is fatally bloated: both the full and shrunk scans are killed.
+    when(namedBlobDb.pullStaleBlobs(eq(container), eq(FIRST_BLOB_NAME))).thenReturn(killedScan);
+    when(namedBlobDb.pullStaleBlobs(eq(container), eq(FIRST_BLOB_NAME), anyInt())).thenReturn(killedScan);
+    when(namedBlobDb.getFirstBlobName(eq(container), eq(FIRST_BLOB_NAME))).thenReturn(
+        CompletableFuture.completedFuture("bloated"));
+    // Everything after the skipped blob name scans cleanly and the container finishes.
+    when(namedBlobDb.pullStaleBlobs(eq(container), eq("bloated\u0000"))).thenReturn(CompletableFuture.completedFuture(
+        new NamedBlobDb.StaleBlobsWithLatestBlobName(Collections.emptyList(), null)));
+
+    new NamedBlobsCleanupRunner(mock(Router.class), namedBlobDb, accountService, 0, new MockTime()).run();
+
+    verify(namedBlobDb).getFirstBlobName(container, FIRST_BLOB_NAME);
+    verify(namedBlobDb).pullStaleBlobs(container, "bloated\u0000");
+  }
+
   @Test
   public void testResumesFromSavedCursorAfterKill() {
     // On the first run the container's first page succeeds (cursor advances to "cursor1"), then the page at "cursor1"

ambry-named-mysql/src/main/java/com/github/ambry/named/MySqlNamedBlobDb.java

@@ -167,6 +167,10 @@ public class MySqlNamedBlobDb implements NamedBlobDb {
           + "ORDER BY %s ASC, %s DESC " + "LIMIT ?", ACCOUNT_ID, CONTAINER_ID, BLOB_NAME, BLOB_ID, VERSION, BLOB_STATE,
       MODIFIED_TS, DELETED_TS, NAMED_BLOBS_V2, BLOB_NAME, VERSION);
 
+  private static final String GET_FIRST_BLOB_NAME = String.format(
+      "SELECT %s FROM %s WHERE %s = ? AND %s = ? AND %s >= ? ORDER BY %s ASC, %s ASC LIMIT 1", BLOB_NAME,
+      NAMED_BLOBS_V2, ACCOUNT_ID, CONTAINER_ID, BLOB_NAME, BLOB_NAME, VERSION);
+
   private final AccountService accountService;
   private final String localDatacenter;
   private final List<String> remoteDatacenters;
@@ -532,6 +536,25 @@ public CompletableFuture<StaleBlobsWithLatestBlobName> pullStaleBlobs(Container
     }, transactionStateTracker);
   }
 
+  @Override
+  public CompletableFuture<String> getFirstBlobName(Container container, String blobNameFrom) {
+    TransactionStateTracker transactionStateTracker =
+        new GetTransactionStateTracker(remoteDatacenters, localDatacenter);
+    return executeGenericTransactionAsync(true, (connection) -> {
+      try (PreparedStatement statement = connection.prepareStatement(GET_FIRST_BLOB_NAME)) {
+        statement.setInt(1, container.getParentAccountId());
+        statement.setInt(2, container.getId());
+        statement.setString(3, blobNameFrom);
+        try (ResultSet resultSet = statement.executeQuery()) {
+          if (resultSet.next()) {
+            return resultSet.getString(1);
+          }
+          return null;
+        }
+      }
+    }, transactionStateTracker);
+  }
+
   @Override
   public CompletableFuture<Integer> cleanupStaleData(List<StaleNamedBlob> staleRecords) {
     return executeGenericTransactionAsync(true, (connection) -> {