CAS gc with conditional batch deletes

docs/en/antalya/cas/architecture/garbage-collection.md

@@ -52,7 +52,7 @@ follower or a deferred round execution returns before that commit.
 | 8 | `fold_ref_intake` | fold | `GET` every new ref-log record and every referenced manifest, extracting blob source edges |
 | 9 | `fold_reduce` | fold | Merge prior edges, new deltas and the parent's condemned rows: spare, condemn, graduate or redelete each candidate; compute `suppress_destructive` |
 | 10 | `fold_seal_write` | fold | Write the new fold seal once, write-once deterministic, adopting a byte-identical replay instead of rewriting it |
-| 11 | `pending_deletes` | fold | The single blob-body delete site: exact-token delete of entries a *previous* round marked `delete_pending`, up to the redelete budget, plus the outcome-log writes |
+| 11 | `pending_deletes` | fold | The single blob-body delete site: `HEAD` each entry a *previous* round marked `delete_pending`, then exact-token conditional delete chunks where the mounted backend proved that capability, plus the outcome-log writes |
 | 12 | `meta_pool_wait` | fold | Drain the bounded pool of async `.meta` condemn-marker writes queued during the fold |
 | 13 | `round_commit` | fold | Retention-prune old generations, then publish the single `gc/state` `CAS` that adopts the whole round |
 | 14 | `handoff_reclaim` | post-`CAS` | Reclaim a generation a ref moved off during this round, which the ordinary retention prune already skipped and will not revisit |
@@ -400,16 +400,27 @@ forensic outcome logs.
 - **Reads:** one `HEAD` per `redelete` entry — the persisted condemned token cannot itself be a
   precondition, so the round observes the blob first, which also settles the absent case without
   spending a conditional delete
-- **Writes / deletes:** a `DELETE` conditional on the observed etag for each `redelete` entry whose
-  live etag matches the condemned token; one write-once outcome log per shard with settled entries
+- **Writes / deletes:** conditional delete chunks for entries whose live etag matches the condemned
+  token. A chunk is bounded by `cas_gc_bulk_delete_chunk_keys`, the backend conditional-batch limit,
+  and 1000 keys. An unprobed, unsupported, or `skip_access_check` mount uses one-key chunks. `.meta`
+  cleanup remains one per settled blob; it is not part of the conditional delete request. One
+  write-once outcome log is written per shard with settled entries
 - **Safety:** an entry is deletable only because a previously committed fold seal published it
   `delete_pending` — durable state from an earlier commit, safe at any leader staleness. Exact token
   means a stale leader cannot delete a fresh incarnation; `NotFound` and `TokenMismatch` are
   tolerated. This round's own commit outcome does not affect the delete's safety.
 - **Fails the round if:** a backend delete marker appears in a response — object versioning on a
   mis-provisioned pool (`LOGICAL_ERROR`)
 - **Observability:** phase row `pending_deletes`; metrics `deleted`, `absent`, `redeleted`,
-  `graduated`, `replaced`, `spared`, `outcome_logs_written`
+  `graduated`, `replaced`, `spared`, `outcome_logs_written`, `delete_batches`. `delete_batches`
+  counts logical conditional-delete chunks submitted by GC. `DiskS3DeleteObjects` counts physical S3
+  delete attempts, including request retries
+
+The project RustFS integration backend supports exact-token singleton deletes but does not implement
+AWS per-object `ETag` conditions in `DeleteObjects`. Its mount probe therefore keeps one-key chunks;
+integration coverage verifies this fallback and physical blob reclamation. Positive AWS batching is
+covered by deterministic GC and scripted S3 transport tests. A live AWS run is still required to
+validate the production service boundary end to end.
 
 Under `suppress_destructive` the `redelete` set is empty by construction, so nothing is deleted, and
 there is no graduation either (would-be graduates are carried unchanged); sparing and superseding
@@ -802,7 +813,7 @@ the chunk size reject `0`:
 | `cas_manifest_sweep_delete_budget_keys` | 100 | orphan-manifest sweep `DELETE` budget per round (phases 9, 18) |
 | `cas_gc_round_sweep_namespace_budget` | 20 | namespaces whose protection view the sweep may build per page (phase 9) |
 | `cas_gc_round_sweep_recovery_op_budget` | 5000 | committed-tail ref-log reads the sweep's recovery walk may spend (phase 9) |
-| `cas_gc_bulk_delete_chunk_keys` | 1000 | keys per batch `DELETE` request for write-once families (phases 15, 17); `1` to `1000` |
+| `cas_gc_bulk_delete_chunk_keys` | 1000 | keys per batch `DELETE` request for blob redelete and write-once families (phases 11, 15, 17); `1` to `1000` |
 | `cas_gc_meta_pool_size` | 16 | bounded pool for condemn-marker writes (phase 12) |
 | `cas_gc_io_concurrency` | 16 | bounded pool for the fold's read-ahead of checkpoints, ref logs, manifest bodies and zero-candidate `HEAD`s (phases 8, 9), the graduation `.meta` reads (phase 9), the orphan-sweep planning reads (phase 9), the rebuild read-ahead and the `pending_deletes` `HEAD` + conditional `DELETE` fan-out (phase 11); other GC requests run on the round thread; `1` runs the covered requests sequentially. `cas_gc_read_concurrency` is rejected without an alias; use `cas_gc_io_concurrency` instead |
 

docs/en/antalya/cas/bucket-requirements.md

@@ -131,7 +131,7 @@ establish acceptance of Google's multipart, native-copy, and exact-delete wire b
 | AWS S3 | ✓ | Native `ETag`-based conditional dialect for mutable objects and exact deletion; blob publication is unconditional |
 | Google Cloud Storage | implementation complete; release gate pending | Generation-token dialect for mutable objects/native-token `HEAD`/exact deletion, opted into via `http_client = gcs_hmac` or `gcp_oauth`; blob publication uses ordinary copy/multipart. Real credentialed GCS groups have not run yet |
 | Azure Blob Storage | probably | Azure's REST API documents the equivalent conditional headers, but ClickHouse's Azure object-storage backend does not yet wire up a `CAS` conditional dialect the way the S3 and GCS paths do — untested, not validated by the capability probe |
-| Other S3-compatible stores | only with enforced conditional operations | The capability probe is the actual gate: a store that silently ignores `If-None-Match`/`If-Match` (accepting and applying the write regardless) fails the probe and is refused. `RustFS` passes the full battery and is used as the project's test backend; `Garage` was evaluated and rejected because it silently ignores conditional operations |
+| Other S3-compatible stores | only with enforced conditional operations | The capability probe is the actual gate: a store that silently ignores `If-None-Match`/`If-Match` (accepting and applying the write regardless) fails the probe and is refused. `RustFS` passes the required singleton conditional-operation battery and is used as the project's test backend, but it does not implement AWS per-object `ETag` conditions in `DeleteObjects`; conditional GC deletion therefore stays on one-key chunks. `Garage` was evaluated and rejected because it silently ignores conditional operations |
 
 The full mechanics of the two dialects — how the backend detects which one a given endpoint speaks,
 what the capability probe actually checks, and how exact-token deletes map onto each provider's

docs/en/antalya/cas/configuration.md

@@ -106,7 +106,7 @@ entirely before release. Treat this table as a snapshot of the current build, no
 | `cas_part_folder_cache_max_entry_bytes` | 16 MiB | Oversized part-folder views bypass retention above this size |
 | `cas_manifest_decode_cache_bytes` | 128 MiB | Manifest decode cache byte budget (`0` disables) |
 | `cas_gc_meta_pool_size` | `16` | Bounded pool size for GC per-hash freshness-meta writes |
-| `cas_gc_io_concurrency` | `16` | Bounded pool size for GC object-storage requests that run in parallel: the fold's read-ahead (checkpoints, ref logs, manifests, zero-candidate HEADs), the orphan-manifest sweep planning reads, the `SYSTEM CAS GC REBUILD` read-ahead, the `pending_deletes` blob `HEAD` + conditional `DELETE` fan-out, and the namespace janitor's batch deletes of dead ref logs and snapshots. Not covered: meta writes (`cas_gc_meta_pool_size`) and all other GC requests, which run on the round thread. `1` runs the covered requests sequentially. `cas_gc_read_concurrency` is rejected without an alias; use `cas_gc_io_concurrency` instead |
+| `cas_gc_io_concurrency` | `16` | Bounded pool size for GC object-storage requests that run in parallel: the fold's read-ahead (checkpoints, ref logs, manifests, zero-candidate HEADs), the orphan-manifest sweep planning reads, the `SYSTEM CAS GC REBUILD` read-ahead, the `pending_deletes` blob `HEAD` + conditional `DELETE` chunks, and the namespace janitor's batch deletes of dead ref logs and snapshots. S3 multi-key conditional chunks are enabled only after the writable mount proves their per-object ETag semantics; unsupported storage and `skip_access_check` use one-key conditional deletes. Not covered: meta writes (`cas_gc_meta_pool_size`) and all other GC requests, which run on the round thread. `1` runs the covered requests sequentially. `cas_gc_read_concurrency` is rejected without an alias; use `cas_gc_io_concurrency` instead |
 | `cas_attempt_timeout_ms` | `5000` | Budget for one HTTP attempt of a writable Native mount's control-plane requests (read, head, list, remove, conditional write), at least 1. Together with the connect cap it forms the attempt envelope (`cas_attempt_timeout_ms + 2 × cap`; the cap is `cas_attempt_timeout_ms` itself when the disk's `connect_timeout_ms` is `0`, else `min(connect_timeout_ms, cas_attempt_timeout_ms)`) that the lease arithmetic reserves: one TCP connect and one TLS handshake under the cap each, send/receive bounded per socket operation by `cas_attempt_timeout_ms`. With background renewal the cadence check requires `cas_mount_renew_period_ms + 2 × envelope + cas_lease_safety_margin_ms < cas_mount_lease_ttl_ms`, which puts an effective ceiling on the frozen connect cap: under the defaults (TTL 30000, period 10000, margin 2000) the envelope must stay under 9000, so a disk `connect_timeout_ms` of 2000 ms or more refuses to open writable — lower the connect timeout or raise the TTL if you hit this |
 | `cas_lease_safety_margin_ms` | `2000` | Startup-only margin validated against the mount lease TTL: the attempt envelope + `cas_lease_safety_margin_ms` must be strictly less than the mount lease TTL, and `cas_mount_renew_period_ms` + 2 × envelope + `cas_lease_safety_margin_ms` too, or the disk refuses to open writable |
 | `cas_unsafe_remount_no_delay` | `0` | Reclaim a mount slot that carries this server's own uuid at once after a hard restart, without observing the slot's token for the lease TTL. Unsafe whenever two processes can hold the same `server_uuid` (a copied uuid file, a stalled predecessor). After such a reclaim the predecessor can still start conditional writes until its own cutoff (`confirmed deadline − cas_lease_safety_margin_ms − 2 × envelope`) or until its next renewal meets the token guard, and a request it already sent may still materialize later. That is not a data hazard: ref-log keys carry `(writer_epoch, sequence)` and creates are conditional, so two writers can never commit different bodies to one key, and recovery's epoch seal settles any straggler (recovery fails closed after 64 successive seal-create attempts displaced by newly materializing old-epoch transactions). The exposure is availability, not data. Intended for test stands and deployments that guarantee one process per uuid |

docs/en/operations/system-tables/cas_gc_log.md

@@ -81,7 +81,7 @@ The phases, in execution order:
 | `fold_ref_intake` | Read and fold every new ref log and the manifest bodies its edges name. | one `GET` per new log, one `GET` per manifest edge |
 | `fold_reduce` | The per-shard in-degree merge: condemn, spare, graduate. | prior-run streaming `GET`s, one `HEAD` per zero-transition candidate, run `PUT`s |
 | `fold_seal_write` | Publish the new fold seal. | one `PUT` |
-| `pending_deletes` | The single content-delete site: exact-token deletes of previously published `delete_pending` entries, plus the outcome logs. `phase_metrics` carries `jobs_scheduled` (entries sent to the GC I/O pool; `0` for no candidates, sequential execution, singleton batches, or refusal before the first submission) and `jobs_failed`. | one `HEAD` and at most one `DELETE` per entry, one outcome-log `PUT` per shard with rows |
+| `pending_deletes` | The single content-delete site: exact-token deletes of previously published `delete_pending` entries, plus the outcome logs. `phase_metrics` carries `jobs_scheduled` (entries submitted to the parallel observation wave; `0` for no candidates or sequential execution), `jobs_failed`, and `delete_batches` (logical conditional-delete chunks). Unsupported or unprobed storage uses one-key chunks. | one `HEAD` per entry, conditional delete chunks, one outcome-log `PUT` per shard with rows |
 | `meta_pool_wait` | Drain the round's per-hash freshness-meta writes. | none on this thread — see the caveat below |
 | `round_commit` | The generation-retention prune and the round's single `gc/state` compare-and-swap. | prune `LIST`s and deletes, one compare-and-swap |
 | `handoff_reclaim` | Wholesale-reclaim generations a moved run ref stranded below the retention cursor. | prefix `LIST`s and deletes |

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasBackend.h

@@ -20,6 +20,11 @@
 #include <variant>
 #include <vector>
 
+namespace DB::ErrorCodes
+{
+    extern const int NOT_IMPLEMENTED;
+}
+
 namespace DB::Cas
 {
 
@@ -169,6 +174,11 @@ class Backend
     /// `DeleteMarker` is a removal that did NOT reclaim: a versioned bucket archived a noncurrent
     /// version instead. Distinct from `Removed` because reclaiming the storage is the point.
     enum class RawRemoval : uint8_t { Removed, Gone, Mismatch, DeleteMarker };
+    struct RawConditionalRemove
+    {
+        String key;
+        String expected_token;
+    };
 
     /// Reads the whole object, or nullopt when the key is absent.
     virtual std::optional<Raw>     read  (const String & key, TransportAccess &) = 0;
@@ -201,10 +211,23 @@ class Backend
     /// absence is success.
     virtual void removeManyWriteOnce(const std::vector<WriteOnceKey> & keys, TransportAccess &) = 0;
 
+    virtual std::vector<RawRemoval> removeManyByConditional(
+        const std::vector<RawConditionalRemove> & removals, TransportAccess & access)
+    {
+        if (removals.size() > 1)
+            throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Conditional batch removal is not implemented by this backend");
+        if (removals.empty())
+            return {};
+        return {remove(removals.front().key, removals.front().expected_token, access)};
+    }
+
     /// The most keys one `removeManyWriteOnce` sends to the storage as one request; at least 1. A decorator
     /// must forward it: the default of 1 would turn every batch below it into one-key requests.
     virtual size_t bulkDeleteKeyLimit() const { return 1; }
 
+    virtual size_t conditionalBatchDeleteKeyLimit() const { return 1; }
+    virtual void pinConditionalBatchDeleteSupport(bool) {}
+
     /// Authoritative, cache-bypassing probe of one key -- see `ProbeOutcome`. DEFAULT (used by every
     /// backend without sharper raw-error evidence, e.g. `InMemoryBackend`): derived from `head`/`read`
     /// alone, so it can only distinguish `Present` from `KeyAbsent`, and ANY exception from either

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.cpp

@@ -326,6 +326,20 @@ void InMemoryBackend::removeManyWriteOnce(const std::vector<WriteOnceKey> & keys
     }
 }
 
+std::vector<Backend::RawRemoval> InMemoryBackend::removeManyByConditional(
+    const std::vector<RawConditionalRemove> & removals, TransportAccess &)
+{
+    for (const RawConditionalRemove & removal : removals)
+        checkExpectedValue(removal.key, removal.expected_token);
+
+    std::lock_guard lock(mutex_);
+    std::vector<RawRemoval> results;
+    results.reserve(removals.size());
+    for (const RawConditionalRemove & removal : removals)
+        results.push_back(applyDelete(removal.key, removal.expected_token));
+    return results;
+}
+
 void InMemoryBackend::failNextBulkRemoveWith(std::exception_ptr error)
 {
     std::lock_guard lock(mutex_);

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInMemoryBackend.h

@@ -51,6 +51,8 @@ class InMemoryBackend : public Backend
     /// `hold_deletes_` exactly as `remove` does: a held delete is queued and lands only on
     /// `landPendingDelete`.
     void removeManyWriteOnce(const std::vector<WriteOnceKey> & keys, TransportAccess & access) override;
+    std::vector<RawRemoval> removeManyByConditional(
+        const std::vector<RawConditionalRemove> & removals, TransportAccess & access) override;
     /// The next `removeManyWriteOnce` throws `error` instead of deleting anything; one-shot, like
     /// `failNextWriteWith`.
     void failNextBulkRemoveWith(std::exception_ptr error);
@@ -62,6 +64,8 @@ class InMemoryBackend : public Backend
 
     /// The emulation deletes keys one by one under its lock, so the CAS maximum is the only bound.
     size_t bulkDeleteKeyLimit() const override { return kBulkDeleteMaxKeys; }
+    size_t conditionalBatchDeleteKeyLimit() const override { return conditional_batch_delete_supported ? kBulkDeleteMaxKeys : 1; }
+    void pinConditionalBatchDeleteSupport(bool supported) override { conditional_batch_delete_supported = supported; }
 
     /// Creates the key when `expected_value` is empty, or replaces the incarnation it names. A
     /// refused precondition leaves the store unchanged. Value enforcement can be disabled with
@@ -227,6 +231,7 @@ class InMemoryBackend : public Backend
     std::vector<std::exception_ptr> armed_bulk_remove_failures_;
     std::function<void()> before_bulk_remove_hook_;
     size_t bulk_remove_calls_ = 0;
+    std::atomic_bool conditional_batch_delete_supported = false;
 };
 
 }

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.cpp

@@ -154,4 +154,13 @@ void InstrumentedBackend::removeManyWriteOnce(const std::vector<WriteOnceKey> &
         incrementCasEvent(classifyCasNs(key.str()), CasOp::Delete);
 }
 
+std::vector<Backend::RawRemoval> InstrumentedBackend::removeManyByConditional(
+    const std::vector<RawConditionalRemove> & removals, TransportAccess & access)
+{
+    std::vector<RawRemoval> results = inner->removeManyByConditional(removals, access);
+    for (const RawConditionalRemove & removal : removals)
+        incrementCasEvent(classifyCasNs(removal.key), CasOp::Delete);
+    return results;
+}
+
 }

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasInstrumentedBackend.h

@@ -130,6 +130,8 @@ class InstrumentedBackend final : public Backend
     /// carried, the request counter says how many requests it took. Out of line, like `publish`, so
     /// the header need not declare the `CASBulkDeleteRequests` extern.
     void removeManyWriteOnce(const std::vector<WriteOnceKey> & keys, TransportAccess & access) override;
+    std::vector<RawRemoval> removeManyByConditional(
+        const std::vector<RawConditionalRemove> & removals, TransportAccess & access) override;
 
     /// Count a create and a replacement separately, and each of them separately from its refusal:
     /// they cost the same one request, but a pool whose creates are mostly refused and one whose
@@ -162,6 +164,8 @@ class InstrumentedBackend final : public Backend
     uint64_t attemptTimeoutMs() const override { return inner->attemptTimeoutMs(); }
     uint64_t attemptEnvelopeMs() const override { return inner->attemptEnvelopeMs(); }
     size_t bulkDeleteKeyLimit() const override { return inner->bulkDeleteKeyLimit(); }
+    size_t conditionalBatchDeleteKeyLimit() const override { return inner->conditionalBatchDeleteKeyLimit(); }
+    void pinConditionalBatchDeleteSupport(bool supported) override { inner->pinConditionalBatchDeleteSupport(supported); }
     bool refreshCredentials() override { return inner->refreshCredentials(); }
 
 private: