From 394badbead5e75a1f872071c073e983cb4458838 Mon Sep 17 00:00:00 2001 From: Zicheng Liu Date: Thu, 20 Aug 2026 14:20:42 -0700 Subject: [PATCH] Skip past un-scannable blob names so bloated containers still make progress When a container's stale-blob page scan is killed by the DB long-transaction limit even after shrinking the page, the offending blob name has more live versions than can be sorted under the limit, and shrinking the page does not help: the scan filesorts the whole [cursor,end] range before LIMIT, so its cost is independent of page size. Previously the container was deferred at the same cursor and made zero forward progress across runs. Add NamedBlobDb.getFirstBlobName(container, from): a cheap index-only lookup (blob_name >= ? ORDER BY blob_name, version LIMIT 1). On a shrunk-page kill the runner uses it to advance the cursor just past the stuck blob name and resumes after it, so the rest of the container is still cleaned. The skipped blob name is left to a targeted reap / TTL / version cap. Skips per container per run are bounded (MAX_BLOB_NAME_SKIPS_PER_RUN) so a broadly bloated container cannot spin on many killed scans in one run; remaining skips happen on later runs, which resume from the saved cursor. Adds a BlobNameSkippedCount metric and a test. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../com/github/ambry/named/NamedBlobDb.java | 10 ++++ .../ambry/commons/InMemNamedBlobDb.java | 17 ++++++ .../frontend/NamedBlobsCleanupRunner.java | 52 ++++++++++++++++--- .../frontend/NamedBlobsCleanupRunnerTest.java | 28 ++++++++++ .../github/ambry/named/MySqlNamedBlobDb.java | 23 ++++++++ 5 files changed, 122 insertions(+), 8 deletions(-) diff --git a/ambry-api/src/main/java/com/github/ambry/named/NamedBlobDb.java b/ambry-api/src/main/java/com/github/ambry/named/NamedBlobDb.java index 4459409be4..af6fe4c8a8 100644 --- a/ambry-api/src/main/java/com/github/ambry/named/NamedBlobDb.java +++ b/ambry-api/src/main/java/com/github/ambry/named/NamedBlobDb.java @@ -119,6 +119,16 @@ default CompletableFuture put(NamedBlobRecord record) { */ CompletableFuture 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 getFirstBlobName(Container container, String blobNameFrom); + /** * Cleanup the stale blobs records */ diff --git a/ambry-commons/src/main/java/com/github/ambry/commons/InMemNamedBlobDb.java b/ambry-commons/src/main/java/com/github/ambry/commons/InMemNamedBlobDb.java index 050015091c..9dec7bf508 100644 --- a/ambry-commons/src/main/java/com/github/ambry/commons/InMemNamedBlobDb.java +++ b/ambry-commons/src/main/java/com/github/ambry/commons/InMemNamedBlobDb.java @@ -227,6 +227,23 @@ public CompletableFuture pullStaleBlobs(Container return pullStaleBlobs(container, latestBlob); } + @Override + public CompletableFuture getFirstBlobName(Container container, String blobNameFrom) { + String containerName = container.getName(); + String first = null; + for (String accountName : allRecords.keySet()) { + TreeMap> 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 pullStaleBlobs(Container container, String latestBlob) { CompletableFuture future = new CompletableFuture<>(); diff --git a/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobsCleanupRunner.java b/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobsCleanupRunner.java index 184f94bc74..b1ecd3216e 100644 --- a/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobsCleanupRunner.java +++ b/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) 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 batchStaleBlobs = staleBlobsWithLatestBlobName.getStaleBlobs(); diff --git a/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobsCleanupRunnerTest.java b/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobsCleanupRunnerTest.java index eef8720f47..aa47ead3ba 100644 --- a/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobsCleanupRunnerTest.java +++ b/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 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" diff --git a/ambry-named-mysql/src/main/java/com/github/ambry/named/MySqlNamedBlobDb.java b/ambry-named-mysql/src/main/java/com/github/ambry/named/MySqlNamedBlobDb.java index ea45d7f2f8..0f20113989 100644 --- a/ambry-named-mysql/src/main/java/com/github/ambry/named/MySqlNamedBlobDb.java +++ b/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 remoteDatacenters; @@ -532,6 +536,25 @@ public CompletableFuture pullStaleBlobs(Container }, transactionStateTracker); } + @Override + public CompletableFuture 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 cleanupStaleData(List staleRecords) { return executeGenericTransactionAsync(true, (connection) -> {