fix(inkless:storage): drain object-store deletes monotonically under throttling

docs/inkless/metrics.rst

@@ -218,14 +218,15 @@ FileCleaner metrics
 io.aiven.inkless.delete:type=FileCleaner
 ----------------------------------------
 
-=====================  =========================================================
-Attribute name         Description                                              
-=====================  =========================================================
-FileCleanerErrorRate   Total number of file cleaning errors                     
-FileCleanerFilesRate   Total number of files cleaned                            
-FileCleanerRate        Total number of file cleaning cycles started             
-FileCleanerTotalTime   Total time spent on a file cleaning cycle in milliseconds
-=====================  =========================================================
+===========================  =================================================================================================================================
+Attribute name               Description                                                                                                                      
+===========================  =================================================================================================================================
+FileCleanerErrorRate         Total number of file cleaning errors                                                                                             
+FileCleanerFilesFailedRate   Total number of files the storage backend did not confirm deleted; they stay marked for deletion and are retried on a later cycle
+FileCleanerFilesRate         Total number of files cleaned                                                                                                    
+FileCleanerRate              Total number of file cleaning cycles started                                                                                     
+FileCleanerTotalTime         Total time spent on a file cleaning cycle in milliseconds                                                                        
+===========================  =================================================================================================================================
 
 
 RetentionEnforcer metrics

storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java

@@ -108,22 +108,10 @@ public void run() {
             } else {
                 LOGGER.info("Running file cleaner: deleting {} of {} marked files", objectKeyPaths.size(), filesToDelete.size());
                 metrics.recordFileCleanerStart();
-                // 1-element holder to carry the duration out of the (synchronous, same-thread) callback
-                // for the log line below; a plain local cannot be assigned from the lambda.
-                final long[] durationMs = {0};
-                TimeUtils.measureDurationMs(time, () -> {
-                    try {
-                        cleanFiles(objectKeyPaths);
-                    } catch (StorageBackendException e) {
-                        LOGGER.error("Error while cleaning files", e);
-                        throw new RuntimeException(e);
-                    }
-                }, duration -> {
-                    durationMs[0] = duration;
-                    metrics.recordFileCleanerTotalTime(duration);
-                });
-                metrics.recordFileCleanerCompleted(objectKeyPaths.size());
-                LOGGER.info("File cleaner deleted {} files in {} ms", objectKeyPaths.size(), durationMs[0]);
+                final int deletedCount = TimeUtils.measureDurationMs(time,
+                    () -> cleanFiles(objectKeyPaths),
+                    metrics::recordFileCleanerTotalTime);
+                LOGGER.info("File cleaner deleted {} of {} files", deletedCount, objectKeyPaths.size());
             }
 
             attempts.set(0);
@@ -135,15 +123,30 @@ public void run() {
         }
     }
 
-    private void cleanFiles(Set<String> objectKeyPaths) throws StorageBackendException {
+    private int cleanFiles(Set<String> objectKeyPaths) throws StorageBackendException {
         final Set<ObjectKey> objectKeys = objectKeyPaths.stream()
             .map(objectKeyCreator::from)
             .collect(Collectors.toSet());
-        // delete files from storage backend
-        storage.delete(objectKeys);
+        // Delete files from the storage backend. Deletion may be partial (e.g. under S3 throttling):
+        // only the keys the backend confirmed deleted are dereferenced in the control plane, so the
+        // remaining keys stay marked for deletion and are retried on the next cycle instead of being
+        // re-attempted after already being deleted.
+        final Set<ObjectKey> deletedKeys = storage.delete(objectKeys);
+        metrics.recordFileCleanerFilesFailed(objectKeyPaths.size() - deletedKeys.size());
+        if (deletedKeys.isEmpty()) {
+            LOGGER.warn("No files deleted from storage out of {} candidates; retrying next cycle",
+                objectKeyPaths.size());
+            return 0;
+        }
+        final Set<String> deletedPaths = deletedKeys.stream()
+            .map(ObjectKey::value)
+            .collect(Collectors.toSet());
         // update control plane
-        final DeleteFilesRequest request = new DeleteFilesRequest(objectKeyPaths);
+        final DeleteFilesRequest request = new DeleteFilesRequest(deletedPaths);
         controlPlane.deleteFiles(request);
+
+        metrics.recordFileCleanerCompleted(deletedPaths.size());
+        return deletedPaths.size();
     }
 
     @Override

storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleanerMetrics.java

@@ -37,6 +37,9 @@ public class FileCleanerMetrics {
     private static final String FILE_CLEANER_FILES_RATE_DOC = "Total number of files cleaned";
     static final String FILE_CLEANER_ERROR_RATE = "FileCleanerErrorRate";
     private static final String FILE_CLEANER_ERROR_RATE_DOC = "Total number of file cleaning errors";
+    static final String FILE_CLEANER_FILES_FAILED_RATE = "FileCleanerFilesFailedRate";
+    private static final String FILE_CLEANER_FILES_FAILED_RATE_DOC = "Total number of files the storage backend did "
+        + "not confirm deleted; they stay marked for deletion and are retried on a later cycle";
 
     /**
      * This method returns a list of all the metric name templates for the FileCleanerMetrics class.
@@ -47,7 +50,8 @@ public static List<MetricNameTemplate> all() {
             new MetricNameTemplate(FILE_CLEANER_TOTAL_TIME, GROUP, FILE_CLEANER_TOTAL_TIME_DOC),
             new MetricNameTemplate(FILE_CLEANER_RATE, GROUP, FILE_CLEANER_RATE_DOC),
             new MetricNameTemplate(FILE_CLEANER_FILES_RATE, GROUP, FILE_CLEANER_FILES_RATE_DOC),
-            new MetricNameTemplate(FILE_CLEANER_ERROR_RATE, GROUP, FILE_CLEANER_ERROR_RATE_DOC)
+            new MetricNameTemplate(FILE_CLEANER_ERROR_RATE, GROUP, FILE_CLEANER_ERROR_RATE_DOC),
+            new MetricNameTemplate(FILE_CLEANER_FILES_FAILED_RATE, GROUP, FILE_CLEANER_FILES_FAILED_RATE_DOC)
         );
     }
 
@@ -57,12 +61,15 @@ public static List<MetricNameTemplate> all() {
     private final LongAdder fileCleanerRate = new LongAdder();
     private final LongAdder fileCleanerFiles = new LongAdder();
     private final LongAdder fileCleanerErrorRate = new LongAdder();
+    // package-private for tests, following ClientAzAwarenessMetrics
+    final LongAdder fileCleanerFilesFailed = new LongAdder();
 
     public FileCleanerMetrics() {
         fileCleanerTotalTime = metricsGroup.newHistogram(FILE_CLEANER_TOTAL_TIME, true, Map.of());
         metricsGroup.newGauge(FILE_CLEANER_RATE, fileCleanerRate::intValue);
         metricsGroup.newGauge(FILE_CLEANER_FILES_RATE, fileCleanerFiles::intValue);
         metricsGroup.newGauge(FILE_CLEANER_ERROR_RATE, fileCleanerErrorRate::intValue);
+        metricsGroup.newGauge(FILE_CLEANER_FILES_FAILED_RATE, fileCleanerFilesFailed::intValue);
     }
 
     public void recordFileCleanerStart() {
@@ -81,10 +88,15 @@ public void recordFileCleanerCompleted(int filesSize) {
         fileCleanerFiles.add(filesSize);
     }
 
+    public void recordFileCleanerFilesFailed(int filesSize) {
+        fileCleanerFilesFailed.add(filesSize);
+    }
+
     public void close() {
         metricsGroup.removeMetric(FILE_CLEANER_TOTAL_TIME);
         metricsGroup.removeMetric(FILE_CLEANER_RATE);
         metricsGroup.removeMetric(FILE_CLEANER_FILES_RATE);
         metricsGroup.removeMetric(FILE_CLEANER_ERROR_RATE);
+        metricsGroup.removeMetric(FILE_CLEANER_FILES_FAILED_RATE);
     }
 }

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/azure/AzureBlobStorage.java

@@ -32,12 +32,16 @@
 import com.azure.storage.common.StorageSharedKeyCredential;
 import com.groupcdg.pitest.annotations.CoverageIgnore;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.BufferedOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -52,6 +56,8 @@
 
 @CoverageIgnore // tested on integration level
 public final class AzureBlobStorage extends StorageBackend {
+    private static final Logger LOGGER = LoggerFactory.getLogger(AzureBlobStorage.class);
+
     private AzureBlobStorageConfig config;
     private BlobContainerClient blobContainerClient;
     private MetricCollector.MetricsPolicy policy;
@@ -195,16 +201,23 @@ public void delete(final ObjectKey key) throws StorageBackendException {
     }
 
     @Override
-    public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
-        try {
-            for (ObjectKey key : keys) {
+    public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
+        // Deleting one blob at a time (there is no Azure batch-delete dependency here), so a failure
+        // on one key must not abandon the rest: accumulate the keys that were removed and report the
+        // failed ones as not deleted. deleteIfExists() returns true if the blob was deleted and false
+        // if it was already absent; both mean the key is gone (idempotent).
+        final Set<ObjectKey> deleted = new HashSet<>();
+        for (final ObjectKey key : keys) {
+            try {
                 blobContainerClient.getBlobClient(key.value()).deleteIfExists();
+                deleted.add(key);
+            } catch (final BlobStorageException e) {
+                LOGGER.warn("Failed to delete {}; leaving it for the next cycle", key, e);
+            } catch (final RuntimeException e) {
+                LOGGER.warn("Failed to delete {}; leaving it for the next cycle", key, Exceptions.unwrap(e));
             }
-        } catch (final BlobStorageException e) {
-            throw new StorageBackendException("Failed to delete " + keys, e);
-        } catch (final RuntimeException e) {
-            throw unwrapReactorExceptions(e, "Failed to delete " + keys);
         }
+        return deleted;
     }
 
     private StorageBackendException unwrapReactorExceptions(final RuntimeException e, final String message) {

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectDeleter.java

@@ -34,6 +34,13 @@ public interface ObjectDeleter extends Closeable {
      * Delete objects from a set of keys.
      *
      * <p>If the object doesn't exist, the operation still succeeds as it is idempotent.
+     *
+     * <p>Deletion may be partial: implementations return the subset of {@code keys} that were
+     * confirmed deleted (which includes keys that were already absent). Keys omitted from the
+     * returned set were not deleted this round (e.g. throttled) and are safe to retry, since
+     * deletion is idempotent. Implementations may still throw for a total/unexpected failure.
+     *
+     * @return the subset of {@code keys} confirmed deleted.
      */
-    void delete(Set<ObjectKey> keys) throws StorageBackendException;
+    Set<ObjectKey> delete(Set<ObjectKey> keys) throws StorageBackendException;
 }

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/gcs/GcsStorage.java

@@ -124,13 +124,19 @@ public void delete(final ObjectKey key) throws StorageBackendException {
     }
 
     @Override
-    public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
+    public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
         try {
             final Set<BlobId> ids = keys.stream()
                     .map(k -> BlobId.of(this.bucketName,k.value()))
                     .collect(Collectors.toSet());
 
+            // storage.delete returns a List<Boolean> of deleted-vs-already-absent, but a genuine
+            // failure surfaces as a thrown BaseServiceException rather than a per-blob flag, so we
+            // cannot extract a confirmed-deleted subset the way the S3 backend does. This stays
+            // all-or-nothing: on success every key is gone (idempotent), and on failure we delete
+            // nothing and let the FileCleaner cycle retry the whole set.
             storage.delete(ids);
+            return Set.copyOf(keys);
         } catch (final BaseServiceException e) {
             throw new StorageBackendException("Failed to delete " + keys, e);
         }

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/in_memory/InMemoryStorage.java

@@ -110,9 +110,10 @@ public void delete(final ObjectKey key) throws StorageBackendException {
     }
 
     @Override
-    public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
+    public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
         Objects.requireNonNull(keys, "keys cannot be null");
         keys.forEach(storage::remove);
+        return Set.copyOf(keys);
     }
 
     @Override

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java

@@ -22,11 +22,15 @@
 
 import com.groupcdg.pitest.annotations.CoverageIgnore;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -54,11 +58,20 @@
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.S3Error;
 
 @CoverageIgnore  // tested on integration level
 public final class S3Storage extends StorageBackend {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(S3Storage.class);
+
     public static final int MAX_DELETE_KEYS_LIMIT = 1000;
+
+    // Per-key S3 error codes that indicate throttling rather than a hard, non-transient failure. Used
+    // only to log throttling distinctly; neither kind is retried in-call.
+    private static final Set<String> THROTTLE_ERROR_CODES =
+        Set.of("SlowDown", "ServiceUnavailable", "RequestLimitExceeded");
+
     private S3Client s3Client;
     private String bucketName;
 
@@ -155,38 +168,70 @@ public void delete(final ObjectKey key) throws StorageBackendException {
     }
 
     @Override
-    public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
+    public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
         final List<ObjectKey> objectKeys = new ArrayList<>(keys);
-        List<ObjectKey> batch = null;
-        try {
-            for (int i = 0; i < objectKeys.size(); i += MAX_DELETE_KEYS_LIMIT) {
-                batch = objectKeys.subList(
-                    i,
-                    Math.min(i + MAX_DELETE_KEYS_LIMIT, objectKeys.size())
-                );
-
-                final Set<ObjectIdentifier> ids = batch.stream()
-                    .map(k -> ObjectIdentifier.builder().key(k.value()).build())
-                    .collect(Collectors.toSet());
-                final Delete delete = Delete.builder().objects(ids).build();
-                final DeleteObjectsRequest deleteObjectsRequest = DeleteObjectsRequest.builder()
-                    .bucket(bucketName)
-                    .delete(delete)
-                    .build();
-                final DeleteObjectsResponse response = s3Client.deleteObjects(deleteObjectsRequest);
-
-                if (!response.errors().isEmpty()) {
-                    final var errors = response.errors().stream()
-                        .map(e -> String.format("Error %s: %s (%s)", e.key(), e.message(), e.code()))
-                        .collect(Collectors.joining(", "));
-                    throw new StorageBackendException("Failed to delete keys " + batch + ": " + errors);
+        final Set<ObjectKey> deleted = new HashSet<>();
+        for (int i = 0; i < objectKeys.size(); i += MAX_DELETE_KEYS_LIMIT) {
+            final Set<ObjectKey> batch = new HashSet<>(objectKeys.subList(
+                i,
+                Math.min(i + MAX_DELETE_KEYS_LIMIT, objectKeys.size())
+            ));
+            final Map<String, ObjectKey> byValue = batch.stream()
+                .collect(Collectors.toMap(ObjectKey::value, k -> k, (a, b) -> a));
+            final DeleteObjectsResponse response;
+            try {
+                response = deleteObjectsOnce(batch);
+            } catch (final SdkException e) {
+                // Whole-request failure, including a 503 the SDK's adaptive retry already exhausted and
+                // timeouts. Stop this pass and report the remaining keys as not deleted; deletion is
+                // idempotent, so re-attempting them later is safe.
+                LOGGER.warn("DeleteObjects request failed; {} keys not deleted",
+                    objectKeys.size() - deleted.size(), e);
+                break;
+            }
+
+            for (final var deletedObject : response.deleted()) {
+                final ObjectKey key = byValue.get(deletedObject.key());
+                if (key != null) {
+                    deleted.add(key);
                 }
             }
-        } catch (final ApiCallTimeoutException | ApiCallAttemptTimeoutException e) {
-            throw new StorageBackendTimeoutException("Failed to delete keys " + batch, e);
-        } catch (final SdkException e) {
-            throw new StorageBackendException("Failed to delete keys " + batch, e);
+            logDeleteErrors(response.errors());
         }
+        return deleted;
+    }
+
+    /**
+     * Logs per-key delete errors, distinguishing throttling (expected under load, aggregated) from
+     * hard errors (logged individually). No retry happens here: keys that were not deleted stay marked
+     * for deletion and are retried on the next FileCleaner cycle, while request-rate backoff is left to
+     * the S3 client's adaptive retry strategy.
+     */
+    private void logDeleteErrors(final List<S3Error> errors) {
+        int throttled = 0;
+        for (final var error : errors) {
+            if (THROTTLE_ERROR_CODES.contains(error.code())) {
+                throttled++;
+            } else {
+                LOGGER.warn("Failed to delete {}: {} ({}); leaving it for the next cycle",
+                    error.key(), error.message(), error.code());
+            }
+        }
+        if (throttled > 0) {
+            LOGGER.info("{} keys throttled by S3; leaving them for the next cycle", throttled);
+        }
+    }
+
+    private DeleteObjectsResponse deleteObjectsOnce(final Set<ObjectKey> keys) {
+        final Set<ObjectIdentifier> ids = keys.stream()
+            .map(k -> ObjectIdentifier.builder().key(k.value()).build())
+            .collect(Collectors.toSet());
+        final Delete delete = Delete.builder().objects(ids).build();
+        final DeleteObjectsRequest deleteObjectsRequest = DeleteObjectsRequest.builder()
+            .bucket(bucketName)
+            .delete(delete)
+            .build();
+        return s3Client.deleteObjects(deleteObjectsRequest);
     }
 
     @Override

storage/inkless/src/test/java/io/aiven/inkless/config/ConfigTestStorageBackend.java

@@ -52,7 +52,8 @@ public void delete(ObjectKey key) throws StorageBackendException {
     }
 
     @Override
-    public void delete(Set<ObjectKey> keys) throws StorageBackendException {
+    public Set<ObjectKey> delete(Set<ObjectKey> keys) throws StorageBackendException {
+        return Set.copyOf(keys);
     }
 
     @Override