CAS: parallel delete blobs

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

@@ -227,6 +227,8 @@ the user-facing configuration surface.
 |---|---|---|
 | `cas_gc_meta_pool_size` | 16 | bounded pool for condemn-marker writes |
 | `cas_gc_read_concurrency` | 16 | bounded pool for the fold's read-ahead; `1` disables |
+| `cas_gc_redelete_concurrency` | 1 | bounded pool for the `pending_deletes` `HEAD` + conditional `DELETE` fan-out; `1` keeps it sequential |
+| `cas_gc_redelete_min_batch_size` | 2 | minimum `pending_deletes` batch size required to enable the parallel fan-out |
 
 ## Observability {#observability}
 

docs/en/antalya/cas/configuration.md

@@ -107,6 +107,8 @@ entirely before release. Treat this table as a snapshot of the current build, no
 | `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_read_concurrency` | `16` | Bounded pool size for the GC fold's read-ahead of checkpoints, ref logs, manifests and zero-candidate HEADs; `1` disables |
+| `cas_gc_redelete_concurrency` | `1` | Bounded pool size for the GC `pending_deletes` phase: how many blob `HEAD` + conditional `DELETE` pairs run at once; `1` keeps the phase sequential |
+| `cas_gc_redelete_min_batch_size` | `2` | Minimum `pending_deletes` batch size required to enable parallel `HEAD` + conditional `DELETE`; smaller batches run sequentially |
 | `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/storing-data.md

@@ -553,6 +553,13 @@ disk-level and server-level settings surface.
 - `cas_gc_read_concurrency` — `16` by default. Bounded thread-pool size for the GC fold's read-ahead of
   checkpoints, ref logs, manifest bodies and zero-candidate `HEAD`s. The fold's decisions stay on the
   round thread in their original order; only the fetches overlap. `1` disables read-ahead.
+- `cas_gc_redelete_concurrency` — `1` by default. Bounded thread-pool size for the GC `pending_deletes`
+  phase, which runs one `HEAD` and one conditional `DELETE` (`If-Match`) per blob. Only these requests
+  run in parallel; outcomes, events and the audit log are applied on the round thread in their original
+  order. If one blob fails, the other blobs are still deleted and recorded, and then the round fails.
+  `1` keeps the phase sequential.
+- `cas_gc_redelete_min_batch_size` — `2` by default. Minimum `pending_deletes` batch size required to
+  enable the re-delete thread pool; smaller batches stay sequential.
 - `skip_access_check` — `false` by default. Skips the disk's `CAS` capability probe ("start now,
   fix later"). The server-level `skip_access_check` flag skips the generic disk access check;
   this disk key governs the `CAS` capability probe.

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp

@@ -86,6 +86,8 @@ namespace ContentAddressedSetting
     extern const ContentAddressedSettingsUInt64 manifest_decode_cache_bytes;
     extern const ContentAddressedSettingsUInt64 gc_meta_pool_size;
     extern const ContentAddressedSettingsUInt64 gc_read_concurrency;
+    extern const ContentAddressedSettingsUInt64 gc_redelete_concurrency;
+    extern const ContentAddressedSettingsUInt64 gc_redelete_min_batch_size;
     extern const ContentAddressedSettingsUInt64 gc_bulk_delete_chunk_keys;
     extern const ContentAddressedSettingsUInt64 attempt_timeout_ms;
     extern const ContentAddressedSettingsUInt64 lease_safety_margin_ms;
@@ -310,6 +312,8 @@ ContentAddressedMetadataStorage::ContentAddressedMetadataStorage(
     , manifest_decode_cache_bytes(settings_[ContentAddressedSetting::manifest_decode_cache_bytes].value)
     , gc_meta_pool_size(settings_[ContentAddressedSetting::gc_meta_pool_size].value)
     , gc_read_concurrency(settings_[ContentAddressedSetting::gc_read_concurrency].value)
+    , gc_redelete_concurrency(settings_[ContentAddressedSetting::gc_redelete_concurrency].value)
+    , gc_redelete_min_batch_size(settings_[ContentAddressedSetting::gc_redelete_min_batch_size].value)
     , gc_bulk_delete_chunk_keys(settings_[ContentAddressedSetting::gc_bulk_delete_chunk_keys].value)
     , cas_attempt_timeout_ms(settings_[ContentAddressedSetting::attempt_timeout_ms].value)
     , cas_lease_safety_margin_ms(settings_[ContentAddressedSetting::lease_safety_margin_ms].value)
@@ -801,6 +805,8 @@ ContentAddressedMetadataStorage::PoolView ContentAddressedMetadataStorage::openP
     pool_config.gc_round_outcome_entry_budget = gc_round_outcome_entry_budget;
     pool_config.gc_meta_pool_size = gc_meta_pool_size;
     pool_config.gc_read_concurrency = gc_read_concurrency;
+    pool_config.gc_redelete_concurrency = gc_redelete_concurrency;
+    pool_config.gc_redelete_min_batch_size = gc_redelete_min_batch_size;
     pool_config.gc_bulk_delete_chunk_keys = gc_bulk_delete_chunk_keys;
     pool_config.cas_request_budget.attempt_timeout_ms = cas_attempt_timeout_ms;
     pool_config.cas_request_budget.lease_safety_margin_ms = cas_lease_safety_margin_ms;

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h

@@ -637,6 +637,8 @@ class ContentAddressedMetadataStorage final : public IMetadataStorage, public IC
     const uint64_t gc_meta_pool_size;
     /// Bounded pool size for the GC fold's read-ahead; 1 disables it.
     const uint64_t gc_read_concurrency;
+    const uint64_t gc_redelete_concurrency;
+    const uint64_t gc_redelete_min_batch_size;
     /// Keys per batch delete request for the write-once families.
     const uint64_t gc_bulk_delete_chunk_keys;
     /// The budget for one HTTP attempt of a writable Native mount's control-plane requests; feeds

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp

@@ -80,6 +80,8 @@ constexpr std::string_view CAS_KEY_PREFIX = "cas_";
     DECLARE(UInt64, manifest_decode_cache_bytes, 128ULL << 20, "Manifest DECODE cache byte budget (0 disables)", 0) \
     DECLARE(UInt64, gc_meta_pool_size, 16, "Bounded pool size for GC per-hash freshness-meta writes", 0) \
     DECLARE(UInt64, gc_read_concurrency, 16, "Bounded pool size for the GC fold's read-ahead of checkpoints, ref logs, manifest bodies and zero-candidate HEADs; 1 disables read-ahead", 0) \
+    DECLARE(UInt64, gc_redelete_concurrency, 1, "Bounded pool size for pending_deletes' HEAD+conditional-DELETE fan-out; 1 keeps it sequential", 0) \
+    DECLARE(UInt64, gc_redelete_min_batch_size, 2, "Minimum pending_deletes batch size to enable parallel HEAD+conditional-DELETE fan-out", 0) \
     DECLARE(UInt64, gc_bulk_delete_chunk_keys, 1000, "Keys per batch delete request in GC's write-once families (owner-removed manifest bodies, covered ref logs and snapshots); 1 to 1000", 0) \
     DECLARE(UInt64, 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. With the connect cap it forms the attempt envelope the lease arithmetic reserves", 0) \
     DECLARE(UInt64, lease_safety_margin_ms, 2000, "Startup-only margin validated against the mount lease TTL: attempt envelope + this must be strictly less than the TTL, and renew period + 2 × envelope + this too", 0) \
@@ -236,6 +238,16 @@ void ContentAddressedSettings::validate()
             settings[ContentAddressedSetting::gc_interval_sec].value, settings[ContentAddressedSetting::gc_shards].value,
             settings[ContentAddressedSetting::gc_read_concurrency].value);
 
+    if (settings[ContentAddressedSetting::gc_redelete_concurrency] == 0)
+        throw Exception(ErrorCodes::BAD_ARGUMENTS,
+            "content_addressed disk: cas_gc_redelete_concurrency must be >= 1 (got {})",
+            settings[ContentAddressedSetting::gc_redelete_concurrency].value);
+
+    if (settings[ContentAddressedSetting::gc_redelete_min_batch_size] == 0)
+        throw Exception(ErrorCodes::BAD_ARGUMENTS,
+            "content_addressed disk: cas_gc_redelete_min_batch_size must be >= 1 (got {})",
+            settings[ContentAddressedSetting::gc_redelete_min_batch_size].value);
+
     if (settings[ContentAddressedSetting::gc_bulk_delete_chunk_keys] == 0
         || settings[ContentAddressedSetting::gc_bulk_delete_chunk_keys] > 1000)
         throw Exception(ErrorCodes::BAD_ARGUMENTS,

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp

@@ -348,6 +348,11 @@ Gc::Gc(PoolPtr store_, UInt128 gc_id_, std::function<uint64_t()> now_ms_fn_,
     read_pool = std::make_unique<ThreadPool>(
         CurrentMetrics::LocalThread, CurrentMetrics::LocalThreadActive, CurrentMetrics::LocalThreadScheduled,
         /*max_threads*/ read_concurrency, /*max_free_threads*/ read_concurrency, /*queue_size*/ 0);
+    const size_t redelete_concurrency = std::max<size_t>(1, store->poolConfig().gc_redelete_concurrency);
+    if (redelete_concurrency > 1)
+        redelete_pool = std::make_unique<ThreadPool>(
+            CurrentMetrics::LocalThread, CurrentMetrics::LocalThreadActive, CurrentMetrics::LocalThreadScheduled,
+            /*max_threads*/ redelete_concurrency, /*max_free_threads*/ redelete_concurrency, /*queue_size*/ 0);
 }
 
 void Gc::runNamespaceJanitorPage(
@@ -398,6 +403,132 @@ uint64_t removeChunkWriteOnceOrOneByOne(CasOperation & op, const std::vector<Wri
     }
 }
 
+Gc::RedeleteIo Gc::performRedeleteIo(const RetiredEntry & entry, const Layout & layout, CasOperation & op)
+{
+    RedeleteIo io;
+    io.blob_key = layout.blobKey(entry.ref);
+    const std::optional<Meta> observed = op.head(io.blob_key, Retry::standard());
+    if (observed)
+        io.del = entry.token.matches(observed->etag) ? op.remove(io.blob_key, observed->etag, Retry::standard()) : Removal::Mismatch;
+    return io;
+}
+
+void Gc::applyRedeleteOutcome(
+    const RetiredEntry & entry,
+    const RedeleteIo & io,
+    uint64_t new_round,
+    uint64_t generation,
+    GcRoundWorkBudget & round_work_budget,
+    RoundReport & report,
+    OutcomeLog & outcome_log)
+{
+    const OutcomeKind outcome_kind = io.del == Removal::Removed ? OutcomeKind::Deleted
+        : io.del == Removal::Gone                               ? OutcomeKind::Absent
+                                                                : OutcomeKind::Replaced;
+    OutcomeEntry outcome{.kind = entry.kind, .ref = entry.ref, .token = entry.token, .outcome = outcome_kind};
+    const String del_outcome{removalName(io.del)};
+    EventEmitter{*store}.emit(
+        [&](CasEvent & e)
+        {
+            e.type = CasEventType::BlobDelete;
+            e.object_kind = CasEventObjectKind::Blob;
+            e.object_hash = blobIdOf(entry.ref);
+            e.token = renderIncarnation(entry.token);
+            e.round = new_round;
+            e.gen = generation;
+            e.outcome = del_outcome;
+            e.reason = "delete_pending published by a prior pass; exact-incarnation delete (pre-CAS)";
+            e.detail = {{"condemn_round", std::to_string(entry.condemn_round)}, {"key", io.blob_key}};
+        });
+    if (round_work_budget.outcomeEntryAvailable())
+    {
+        outcome_log.entries.push_back(std::move(outcome));
+        ++round_work_budget.outcome_entries_used;
+    }
+    ++report.redeleted;
+    ProfileEvents::increment(ProfileEvents::CASGCRetiredRedeleted);
+    if (io.del == Removal::Removed || io.del == Removal::Gone)
+    {
+        meta_writer->scheduleConfirmedMetaDelete(entry.ref);
+    }
+    meta_writer->forgetCondemnMarker(entry.ref, entry.token);
+}
+
+void Gc::redeleteBlob(
+    const RetiredEntry & entry,
+    const Layout & layout,
+    CasOperation & op,
+    uint64_t new_round,
+    uint64_t generation,
+    GcRoundWorkBudget & round_work_budget,
+    RoundReport & report,
+    OutcomeLog & outcome_log)
+{
+    const RedeleteIo io = performRedeleteIo(entry, layout, op);
+    applyRedeleteOutcome(entry, io, new_round, generation, round_work_budget, report, outcome_log);
+}
+
+void Gc::redeleteBlobs(
+    const std::vector<RetiredEntry> & entries,
+    const Layout & layout,
+    CasOperation & op,
+    uint64_t new_round,
+    uint64_t generation,
+    GcRoundWorkBudget & round_work_budget,
+    RoundReport & report,
+    OutcomeLog & outcome_log)
+{
+    if (!redelete_pool || entries.size() < store->poolConfig().gc_redelete_min_batch_size)
+    {
+        for (const RetiredEntry & entry : entries)
+            redeleteBlob(entry, layout, op, new_round, generation, round_work_budget, report, outcome_log);
+        return;
+    }
+
+    std::vector<RedeleteIo> io_results(entries.size());
+    const uint64_t gen = op.generation();
+    size_t scheduled = 0;
+    std::exception_ptr first_error;
+    try
+    {
+        for (; scheduled < entries.size(); ++scheduled)
+        {
+            redelete_pool->scheduleOrThrowOnError(
+                [&, i = scheduled]
+                {
+                    try
+                    {
+                        CasOperation job_op = store->openRequests().resume(gen);
+                        io_results[i] = performRedeleteIo(entries[i], layout, job_op);
+                    }
+                    catch (...)
+                    {
+                        io_results[i].error = std::current_exception();
+                    }
+                });
+        }
+    }
+    catch (...)
+    {
+        first_error = std::current_exception();
+    }
+    redelete_pool->wait();
+
+    for (size_t i = 0; i < scheduled; ++i)
+    {
+        if (io_results[i].error)
+        {
+            if (!first_error)
+                first_error = io_results[i].error;
+            continue;
+        }
+        applyRedeleteOutcome(entries[i], io_results[i], new_round, generation, round_work_budget, report, outcome_log);
+    }
+
+    if (first_error)
+        std::rethrow_exception(first_error);
+}
+
 RoundReport Gc::runRegularRound(std::function<void()> on_lease_acquired, bool allow_steal, UniversePolicy policy,
                                 RoundReport * progress)
 {
@@ -717,62 +848,7 @@ RoundReport Gc::runRegularRound(std::function<void()> on_lease_acquired, bool al
         static const std::vector<RetiredEntry> kNothingToDelete;
         const std::vector<RetiredEntry> & redelete_now =
             suppress_destructive ? kNothingToDelete : merge.redelete;
-        for (const RetiredEntry & entry : redelete_now)
-        {
-            /// The condemned incarnation is a PERSISTED pair and cannot itself be a precondition, so
-            /// the round observes the blob and compares the two renderings. Observing first also
-            /// settles the absent case without spending a conditional delete against a key that is
-            /// already gone.
-            const String blob_key = layout.blobKey(entry.ref);
-            const std::optional<Meta> observed = op.head(blob_key, Retry::standard());
-            Removal del = Removal::Gone;
-            if (observed)
-                del = entry.token.matches(observed->etag)
-                    ? op.remove(blob_key, observed->etag, Retry::standard())
-                    : Removal::Mismatch;
-
-            const OutcomeKind outcome_kind = del == Removal::Removed ? OutcomeKind::Deleted
-                                            : del == Removal::Gone  ? OutcomeKind::Absent
-                                                                    : OutcomeKind::Replaced;
-            OutcomeEntry outcome{.kind = entry.kind, .ref = entry.ref, .token = entry.token, .outcome = outcome_kind};
-            const String del_outcome{removalName(del)};
-            /// The single content-delete site is attributable per row. A mismatch (a writer recreated
-            /// the incarnation) is terminal-OK: the fresh incarnation is a live object.
-            EventEmitter{*store}.emit([&](CasEvent & e)
-            {
-                e.type = CasEventType::BlobDelete;
-                e.object_kind = CasEventObjectKind::Blob;
-                e.object_hash = blobIdOf(entry.ref);
-                e.token = renderIncarnation(entry.token);
-                e.round = new_round;
-                e.gen = generation;
-                e.outcome = del_outcome;
-                e.reason = "delete_pending published by a prior pass; exact-incarnation delete (pre-CAS)";
-                e.detail = {{"condemn_round", std::to_string(entry.condemn_round)},
-                            {"key", blob_key}};
-            });
-            /// The audit row is observability only -- the delete above already executed regardless of
-            /// this cap. Skipping it here bounds the per-shard `GcOutcomes` body without skipping or
-            /// deferring any destructive work.
-            if (round_work_budget.outcomeEntryAvailable())
-            {
-                outcomes[shard].entries.push_back(std::move(outcome));
-                ++round_work_budget.outcome_entries_used;
-            }
-            ++report.redeleted;
-            ProfileEvents::increment(ProfileEvents::CASGCRetiredRedeleted);
-            /// Drop the per-hash meta only on a removal or a proven absence — a mismatch means a
-            /// writer already resurrected a fresh incarnation at this hash, and that writer's
-            /// own republication path already flipped the meta back to Clean; blindly deleting here
-            /// would race that legitimate Clean write for no reason (the meta is advisory, but there is
-            /// no reason to touch it on that path at all).
-            if (del == Removal::Removed || del == Removal::Gone)
-            {
-                meta_writer->scheduleConfirmedMetaDelete(entry.ref);
-            }
-            /// The entry left the pipeline — drop its in-process condemn-marker confirmation.
-            meta_writer->forgetCondemnMarker(entry.ref, entry.token);
-        }
+        redeleteBlobs(redelete_now, layout, op, new_round, generation, round_work_budget, report, outcomes[shard]);
         for (const RetiredEntry & entry : merge.spared)
         {
             /// A fresh dedup-adopt raced the condemn (see the matching CasGcFold Debug log emitted