From 8e0ec8de875929bed941cf2ed7c237bb0a157394 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Fri, 11 Sep 2026 13:30:07 +0200 Subject: [PATCH 1/7] cas: parallel delete blobs Signed-off-by: Konstantin Morozov --- .../cas/architecture/garbage-collection.md | 2 + docs/en/antalya/cas/configuration.md | 2 + docs/en/operations/storing-data.md | 7 + .../ContentAddressedMetadataStorage.cpp | 6 + .../ContentAddressedMetadataStorage.h | 2 + .../ContentAddressedSettings.cpp | 12 ++ .../ContentAddressed/Gc/CasGc.cpp | 188 ++++++++++++------ .../ContentAddressed/Gc/CasGc.h | 40 ++++ .../ContentAddressed/Pool/CasPool.h | 2 + .../gtest_cas_gc_redelete_concurrency.cpp | 131 ++++++++++++ src/Disks/tests/gtest_cas_settings.cpp | 42 ++++ 11 files changed, 378 insertions(+), 56 deletions(-) create mode 100644 src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp diff --git a/docs/en/antalya/cas/architecture/garbage-collection.md b/docs/en/antalya/cas/architecture/garbage-collection.md index a604104c7796..cbf085451e72 100644 --- a/docs/en/antalya/cas/architecture/garbage-collection.md +++ b/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} diff --git a/docs/en/antalya/cas/configuration.md b/docs/en/antalya/cas/configuration.md index bf0fe17f7dd3..96b73316aa8d 100644 --- a/docs/en/antalya/cas/configuration.md +++ b/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 | diff --git a/docs/en/operations/storing-data.md b/docs/en/operations/storing-data.md index a3503bc1eb2c..ac3ed64e7982 100644 --- a/docs/en/operations/storing-data.md +++ b/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. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp index 78666e0ad297..7da0e01c0774 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp +++ b/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; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h index bffb608688cc..b8dd4dc32933 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h +++ b/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 diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp index b2a3651b2472..654693a3528e 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp +++ b/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, diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index 336ab21fbefa..c3b13e5e2844 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -348,6 +348,11 @@ Gc::Gc(PoolPtr store_, UInt128 gc_id_, std::function now_ms_fn_, read_pool = std::make_unique( 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(1, store->poolConfig().gc_redelete_concurrency); + if (redelete_concurrency > 1) + redelete_pool = std::make_unique( + 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 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 & 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 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 on_lease_acquired, bool allow_steal, UniversePolicy policy, RoundReport * progress) { @@ -717,62 +848,7 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al static const std::vector kNothingToDelete; const std::vector & 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 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 diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h index bd0d65e2e573..5a6ae38f0b1d 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h @@ -726,6 +726,44 @@ class Gc /// every destructive-work family the round touches — see `GcRoundWorkBudget`. GcRoundWorkBudget & work_budget); + struct RedeleteIo + { + String blob_key; + Removal del = Removal::Gone; + std::exception_ptr error; + }; + + RedeleteIo performRedeleteIo(const RetiredEntry & entry, const Layout & layout, CasOperation & op); + + void applyRedeleteOutcome( + const RetiredEntry & entry, + const RedeleteIo & io, + uint64_t new_round, + uint64_t generation, + GcRoundWorkBudget & round_work_budget, + RoundReport & report, + OutcomeLog & outcome_log); + + void 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); + + void redeleteBlobs( + const std::vector & entries, + const Layout & layout, + CasOperation & op, + uint64_t new_round, + uint64_t generation, + GcRoundWorkBudget & round_work_budget, + RoundReport & report, + OutcomeLog & outcome_log); + /// The round's `_ckpt.checkpoint` witness per namespace — the SECOND, hint-independent witness the /// walk decides its absents against. ONE call site, in the fold, right where the hint is grouped. /// @@ -981,6 +1019,8 @@ class Gc /// constructor body has validated `store`. std::unique_ptr read_pool; + std::unique_ptr redelete_pool; + /// Probe B1's two numbers for the round: the ref-log POSITIONS the sealed coverage declares covered /// (counted arithmetically over each namespace's cut -- not by listed ids, which under arithmetic /// intake say nothing about what was applied), and the ref logs that actually folded. They are EQUAL diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index cbde11caa87c..e07c3de11e0e 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -179,6 +179,8 @@ struct PoolConfig /// only the fetch overlaps. `1` issues no read-ahead at all and is the sequential round, request /// for request. uint64_t gc_read_concurrency = 16; + uint64_t gc_redelete_concurrency = 1; + uint64_t gc_redelete_min_batch_size = 2; /// Tests drive `renewWatermarkOnce` explicitly; gates both persistent runtime workers. bool background_watermark = false; /// Installed on the pool before a writable mount can start its runtime-owned workers. diff --git a/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp new file mode 100644 index 000000000000..5f4547143322 --- /dev/null +++ b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp @@ -0,0 +1,131 @@ +#include + +#include +#include + +#include +#include +#include +#include "cas_test_helpers.h" + +using namespace DB::Cas; +using namespace DB::Cas::tests; + +namespace +{ + +const DB::UInt128 kGc = hexToU128("00000000000000000000000000000001"); +constexpr uint64_t kBlobs = 6; +constexpr uint64_t kFaultedBlob = 3; + +class RemoveFaultBackend : public InMemoryBackend +{ +public: + RawRemoval remove(const String & key, const String & expected_value, TransportAccess & access) override + { + if (key == faulted_key && armed.exchange(false)) + throw std::runtime_error("injected remove fault"); + return InMemoryBackend::remove(key, expected_value, access); + } + + String faulted_key; + std::atomic armed{false}; +}; + +template +PoolPtr openPoolWithRedeleteConcurrency(std::shared_ptr backend, uint64_t concurrency) +{ + PoolConfig config{.pool_prefix = "p", .server_root_id = "test"}; + config.gc_redelete_concurrency = concurrency; + return Pool::open(std::move(backend), std::move(config)); +} + +String blobKeyOf(const Pool & store, uint64_t blob) +{ + return store.layout().blobKey(BlobRef{BlobHashAlgo::CityHash128, BlobDigest::fromU128(DB::UInt128(blob))}); +} + +bool allBlobsAbsent(Backend & backend, const Pool & store) +{ + for (uint64_t b = 1; b <= kBlobs; ++b) + if (!blobAbsent(backend, store.layout(), DB::UInt128(b))) + return false; + return true; +} + +void publishThenDrop(Backend & backend, const PoolPtr & store, Gc & gc) +{ + const RootNamespace ns{"00/aa@cas@"}; + const ManifestRef r{.writer_epoch = 1, .build_sequence = 1, .manifest_ordinal = 0xAA}; + std::vector entries; + for (uint64_t b = 1; b <= kBlobs; ++b) + { + writeBlobBody(backend, store->layout(), DB::UInt128(b)); + entries.push_back(blobEntryFor("f" + std::to_string(b), DB::UInt128(b))); + } + writeManifestRaw(backend, store->layout(), ns, r, entries); + publishCommittedTransition(backend, store->layout(), ns, "tbl", std::nullopt, r); + + runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + ASSERT_FALSE(blobAbsent(backend, store->layout(), DB::UInt128(1))); + + dropRefTransition(backend, store->layout(), ns, "tbl", r); +} + +} + +TEST(CASGCRedeleteConcurrency, ParallelRedeleteReclaimsEveryBlob) +{ + auto backend = std::make_shared(); + auto store = openPoolWithRedeleteConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + + size_t redeleted = 0; + size_t deleted = 0; + for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) + { + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + redeleted += rep.redeleted; + deleted += rep.deleted; + } + + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); + EXPECT_EQ(redeleted, kBlobs); + EXPECT_EQ(deleted, kBlobs); +} + +TEST(CASGCRedeleteConcurrency, FailedRemoveKeepsSiblingOutcomesAndPoolAlive) +{ + auto backend = std::make_shared(); + auto store = openPoolWithRedeleteConcurrency(backend, 4); + backend->faulted_key = blobKeyOf(*store, kFaultedBlob); + backend->armed = true; + + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + + size_t failed_rounds = 0; + for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) + { + RoundReport progress; + try + { + gc.runRegularRound({}, /*allow_steal*/ true, UniversePolicy::Authoritative, &progress); + } + catch (const std::exception &) + { + ++failed_rounds; + EXPECT_EQ(progress.redeleted, kBlobs - 1); + for (uint64_t b = 1; b <= kBlobs; ++b) + EXPECT_EQ(blobAbsent(*backend, store->layout(), DB::UInt128(b)), b != kFaultedBlob) << "blob " << b; + } + store->renewWatermarkOnce(); + } + + EXPECT_EQ(failed_rounds, 1u); + EXPECT_FALSE(backend->armed.load()); + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); +} diff --git a/src/Disks/tests/gtest_cas_settings.cpp b/src/Disks/tests/gtest_cas_settings.cpp index 53b3b50b4fbb..554bff83b95c 100644 --- a/src/Disks/tests/gtest_cas_settings.cpp +++ b/src/Disks/tests/gtest_cas_settings.cpp @@ -26,6 +26,8 @@ namespace DB::ContentAddressedSetting extern const ContentAddressedSettingsUInt64 gc_shards; extern const ContentAddressedSettingsUInt64 gc_interval_sec; extern const ContentAddressedSettingsUInt64 gc_bulk_delete_chunk_keys; + extern const ContentAddressedSettingsUInt64 gc_redelete_concurrency; + extern const ContentAddressedSettingsUInt64 gc_redelete_min_batch_size; extern const ContentAddressedSettingsString scratch_path; extern const ContentAddressedSettingsBool unsafe_remount_no_delay; } @@ -209,6 +211,46 @@ TEST(CASSettings, BulkDeleteChunkKeysBoundsAreEnforced) EXPECT_EQ(s[ContentAddressedSetting::gc_bulk_delete_chunk_keys].value, 1u); } +TEST(CASSettings, RedeleteConcurrencyBoundsAreEnforced) +{ + expectLoadFailureWithExactMessage( + "srv1" + "0", + ErrorCodes::BAD_ARGUMENTS, + "content_addressed disk: cas_gc_redelete_concurrency must be >= 1 (got 0)"); + + { + auto cfg = makeConfig("srv1"); + ContentAddressedSettings s; + s.loadFromConfig(*cfg, "disk", "/scratch", "/scratch", identity_macros); + EXPECT_EQ(s[ContentAddressedSetting::gc_redelete_concurrency].value, 1u); + EXPECT_EQ(s[ContentAddressedSetting::gc_redelete_min_batch_size].value, 2u); + } + + auto cfg = makeConfig( + "srv1" + "8"); + ContentAddressedSettings s; + EXPECT_NO_THROW(s.loadFromConfig(*cfg, "disk", "/scratch", "/scratch", identity_macros)); + EXPECT_EQ(s[ContentAddressedSetting::gc_redelete_concurrency].value, 8u); +} + +TEST(CASSettings, RedeleteMinBatchSizeBoundsAreEnforced) +{ + expectLoadFailureWithExactMessage( + "srv1" + "0", + ErrorCodes::BAD_ARGUMENTS, + "content_addressed disk: cas_gc_redelete_min_batch_size must be >= 1 (got 0)"); + + auto cfg = makeConfig( + "srv1" + "7"); + ContentAddressedSettings s; + EXPECT_NO_THROW(s.loadFromConfig(*cfg, "disk", "/scratch", "/scratch", identity_macros)); + EXPECT_EQ(s[ContentAddressedSetting::gc_redelete_min_batch_size].value, 7u); +} + TEST(CASContentAddressedSettings, InvalidEnumDiagnosticsNameExternalConfigKeys) { expectLoadFailureWithExactMessage( From 2e62b8fffe3248deaef984e00ad2a924d9790e00 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 15 Sep 2026 11:07:20 +0200 Subject: [PATCH 2/7] log and profile all errors Signed-off-by: Konstantin Morozov --- src/Common/ProfileEvents.cpp | 1 + .../ContentAddressed/Gc/CasGc.cpp | 25 ++++++++++++++++++- .../gtest_cas_gc_redelete_concurrency.cpp | 9 +++++++ 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index 184d4941aa3c..a9986b5aa485 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -910,6 +910,7 @@ The server successfully detected this situation and will download merged part fr M(CASGCRetiredSparedByReref, "Number of CAS GC delete_pending entries spared because a fresh deduplicating adopt re-referenced them after graduation (an ordinary observe/condemn race, not a fail-closed abort). Subset of CASGCRetiredSpared.", ValueType::Number) \ M(CASGCRetiredGraduated, "Number of CAS GC retired entries that passed the safety floor and became pending deletion. Growth indicates cleanup progress.", ValueType::Number) \ M(CASGCRetiredRedeleted, "Number of CAS GC pending deletes executed with an exact object token. Growth indicates physical cleanup activity.", ValueType::Number) \ + M(CASGCRetiredRedeleteFailed, "Number of CAS GC pending blob deletes whose HEAD or exact-token DELETE failed. A non-zero value indicates object storage errors; the affected blobs are retried in the next round.", ValueType::Number) \ M(CASGCUnmatchedRemoveDeltas, "Number of CAS GC in-degree removal deltas that matched no existing source edge. The in-degree model is a set, not a counter, so this is a per-key no-op by design and never causes a false deletion — but a persistent nonzero rate means removal deltas are reaching the reducer without their matching activation, which is a correctness signal.", ValueType::Number) \ M(CASGCCondemnMarkerUnconfirmedCarry, "Number of CAS GC retirements delayed because a durable condemn marker could not be confirmed. A non-zero value indicates marker write or read failures and safely postpone deletion.", ValueType::Number) \ M(CASGCHeartbeatFenceOuts, "Number of CAS GC operations fenced out for expired mounts. A non-zero value indicates stale mounts or heartbeat delays.", ValueType::Number) \ diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index c3b13e5e2844..0fe7844c15f5 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -44,6 +44,7 @@ namespace ProfileEvents extern const Event CASGCRetiredSpared; extern const Event CASGCRetiredGraduated; extern const Event CASGCRetiredRedeleted; + extern const Event CASGCRetiredRedeleteFailed; extern const Event CASGCRetireReplaced; extern const Event CASGCCondemnMarkerUnconfirmedCarry; extern const Event CASGCHeartbeatFenceOuts; @@ -464,7 +465,20 @@ void Gc::redeleteBlob( RoundReport & report, OutcomeLog & outcome_log) { - const RedeleteIo io = performRedeleteIo(entry, layout, op); + RedeleteIo io; + try + { + io = performRedeleteIo(entry, layout, op); + } + catch (...) + { + ProfileEvents::increment(ProfileEvents::CASGCRetiredRedeleteFailed); + LOG_WARNING(logger, + "CAS gc: pending delete of blob {} (key `{}`, condemned at round {}) failed; the entry stays delete_pending " + "and is retried in the next round: {}", + blobIdOf(entry.ref), layout.blobKey(entry.ref), entry.condemn_round, getCurrentExceptionMessage(false)); + throw; + } applyRedeleteOutcome(entry, io, new_round, generation, round_work_budget, report, outcome_log); } @@ -504,6 +518,15 @@ void Gc::redeleteBlobs( catch (...) { io_results[i].error = std::current_exception(); + ProfileEvents::increment(ProfileEvents::CASGCRetiredRedeleteFailed); + LOG_WARNING( + logger, + "CAS gc: pending delete of blob {} (key `{}`, condemned at round {}) failed; the entry stays " + "delete_pending and is retried in the next round: {}", + blobIdOf(entries[i].ref), + layout.blobKey(entries[i].ref), + entries[i].condemn_round, + getCurrentExceptionMessage(false)); } }); } diff --git a/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp index 5f4547143322..b0fc35e3cd5b 100644 --- a/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp +++ b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp @@ -3,11 +3,18 @@ #include #include +#include + #include #include #include #include "cas_test_helpers.h" +namespace ProfileEvents +{ +extern const Event CASGCRetiredRedeleteFailed; +} + using namespace DB::Cas; using namespace DB::Cas::tests; @@ -107,6 +114,7 @@ TEST(CASGCRedeleteConcurrency, FailedRemoveKeepsSiblingOutcomesAndPoolAlive) Gc gc(store, kGc); publishThenDrop(*backend, store, gc); + const auto failed_before = ProfileEvents::global_counters[ProfileEvents::CASGCRetiredRedeleteFailed].load(); size_t failed_rounds = 0; for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) { @@ -126,6 +134,7 @@ TEST(CASGCRedeleteConcurrency, FailedRemoveKeepsSiblingOutcomesAndPoolAlive) } EXPECT_EQ(failed_rounds, 1u); + EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASGCRetiredRedeleteFailed].load() - failed_before, 1u); EXPECT_FALSE(backend->armed.load()); EXPECT_TRUE(allBlobsAbsent(*backend, *store)); } From 2afa76086ed538789e69bdb7e638ccdd3da95861 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 15 Sep 2026 11:45:36 +0200 Subject: [PATCH 3/7] add comments Signed-off-by: Konstantin Morozov --- .../ContentAddressed/Gc/CasGc.cpp | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index 0fe7844c15f5..635e85703545 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -408,6 +408,10 @@ Gc::RedeleteIo Gc::performRedeleteIo(const RetiredEntry & entry, const Layout & { RedeleteIo io; io.blob_key = layout.blobKey(entry.ref); + /// 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 std::optional 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; @@ -428,6 +432,8 @@ void Gc::applyRedeleteOutcome( : OutcomeKind::Replaced; OutcomeEntry outcome{.kind = entry.kind, .ref = entry.ref, .token = entry.token, .outcome = outcome_kind}; const String del_outcome{removalName(io.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) { @@ -441,6 +447,9 @@ void Gc::applyRedeleteOutcome( 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}}; }); + /// The audit row is observability only -- the delete in `performRedeleteIo` 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()) { outcome_log.entries.push_back(std::move(outcome)); @@ -448,10 +457,16 @@ void Gc::applyRedeleteOutcome( } ++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 published a fresh incarnation at this hash, and that writer's own + /// republication path (`PartWriteTxn::ensureBlobPresent`) reconciles the meta back to Clean + /// right after the publication; deleting it 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 (io.del == Removal::Removed || io.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); } From 0a0af4637729a5f96efb372ec8892d75414b25cc Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 15 Sep 2026 11:49:10 +0200 Subject: [PATCH 4/7] rename generations Signed-off-by: Konstantin Morozov --- .../ContentAddressed/Gc/CasGc.cpp | 18 +++++++++--------- .../ContentAddressed/Gc/CasGc.h | 6 +++--- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index 635e85703545..d3aa6e52f170 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -422,7 +422,7 @@ void Gc::applyRedeleteOutcome( const RetiredEntry & entry, const RedeleteIo & io, uint64_t new_round, - uint64_t generation, + uint64_t snap_generation, GcRoundWorkBudget & round_work_budget, RoundReport & report, OutcomeLog & outcome_log) @@ -442,7 +442,7 @@ void Gc::applyRedeleteOutcome( e.object_hash = blobIdOf(entry.ref); e.token = renderIncarnation(entry.token); e.round = new_round; - e.gen = generation; + e.gen = snap_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}}; @@ -475,7 +475,7 @@ void Gc::redeleteBlob( const Layout & layout, CasOperation & op, uint64_t new_round, - uint64_t generation, + uint64_t snap_generation, GcRoundWorkBudget & round_work_budget, RoundReport & report, OutcomeLog & outcome_log) @@ -494,7 +494,7 @@ void Gc::redeleteBlob( blobIdOf(entry.ref), layout.blobKey(entry.ref), entry.condemn_round, getCurrentExceptionMessage(false)); throw; } - applyRedeleteOutcome(entry, io, new_round, generation, round_work_budget, report, outcome_log); + applyRedeleteOutcome(entry, io, new_round, snap_generation, round_work_budget, report, outcome_log); } void Gc::redeleteBlobs( @@ -502,7 +502,7 @@ void Gc::redeleteBlobs( const Layout & layout, CasOperation & op, uint64_t new_round, - uint64_t generation, + uint64_t snap_generation, GcRoundWorkBudget & round_work_budget, RoundReport & report, OutcomeLog & outcome_log) @@ -510,12 +510,12 @@ void Gc::redeleteBlobs( 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); + redeleteBlob(entry, layout, op, new_round, snap_generation, round_work_budget, report, outcome_log); return; } std::vector io_results(entries.size()); - const uint64_t gen = op.generation(); + const uint64_t admitted_generation = op.generation(); size_t scheduled = 0; std::exception_ptr first_error; try @@ -527,7 +527,7 @@ void Gc::redeleteBlobs( { try { - CasOperation job_op = store->openRequests().resume(gen); + CasOperation job_op = store->openRequests().resume(admitted_generation); io_results[i] = performRedeleteIo(entries[i], layout, job_op); } catch (...) @@ -560,7 +560,7 @@ void Gc::redeleteBlobs( first_error = io_results[i].error; continue; } - applyRedeleteOutcome(entries[i], io_results[i], new_round, generation, round_work_budget, report, outcome_log); + applyRedeleteOutcome(entries[i], io_results[i], new_round, snap_generation, round_work_budget, report, outcome_log); } if (first_error) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h index 5a6ae38f0b1d..6ea13d39dfdc 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h @@ -739,7 +739,7 @@ class Gc const RetiredEntry & entry, const RedeleteIo & io, uint64_t new_round, - uint64_t generation, + uint64_t snap_generation, GcRoundWorkBudget & round_work_budget, RoundReport & report, OutcomeLog & outcome_log); @@ -749,7 +749,7 @@ class Gc const Layout & layout, CasOperation & op, uint64_t new_round, - uint64_t generation, + uint64_t snap_generation, GcRoundWorkBudget & round_work_budget, RoundReport & report, OutcomeLog & outcome_log); @@ -759,7 +759,7 @@ class Gc const Layout & layout, CasOperation & op, uint64_t new_round, - uint64_t generation, + uint64_t snap_generation, GcRoundWorkBudget & round_work_budget, RoundReport & report, OutcomeLog & outcome_log); From d4a9d55f2323ad3dcfd025f4c293f02cdfda7bc3 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 15 Sep 2026 12:54:11 +0200 Subject: [PATCH 5/7] reuse pool, change settings, add tests Signed-off-by: Konstantin Morozov --- .../cas/architecture/garbage-collection.md | 4 +- docs/en/antalya/cas/configuration.md | 4 +- docs/en/operations/storing-data.md | 23 +- .../ContentAddressedMetadataStorage.cpp | 12 +- .../ContentAddressedMetadataStorage.h | 6 +- .../ContentAddressedSettings.cpp | 20 +- .../ContentAddressed/Gc/CasGc.cpp | 131 ++++++---- .../ContentAddressed/Gc/CasGc.h | 15 +- .../ContentAddressed/Pool/CasPool.h | 13 +- src/Disks/tests/gtest_cas_gc_read_ahead.cpp | 6 +- .../gtest_cas_gc_redelete_concurrency.cpp | 234 +++++++++++++++++- src/Disks/tests/gtest_cas_ref_gc.cpp | 2 +- src/Disks/tests/gtest_cas_settings.cpp | 54 +--- 13 files changed, 350 insertions(+), 174 deletions(-) diff --git a/docs/en/antalya/cas/architecture/garbage-collection.md b/docs/en/antalya/cas/architecture/garbage-collection.md index cbf085451e72..4590b6b4a8cc 100644 --- a/docs/en/antalya/cas/architecture/garbage-collection.md +++ b/docs/en/antalya/cas/architecture/garbage-collection.md @@ -226,9 +226,7 @@ the user-facing configuration surface. | Setting | Default | Bounds | |---|---|---| | `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 | +| `cas_gc_io_concurrency` | 16 | bounded pool for the fold's read-ahead, the orphan-sweep planning reads, the rebuild read-ahead and the `pending_deletes` `HEAD` + conditional `DELETE` fan-out; other GC requests run on the round thread; `1` runs the covered requests sequentially | ## Observability {#observability} diff --git a/docs/en/antalya/cas/configuration.md b/docs/en/antalya/cas/configuration.md index 96b73316aa8d..2e282d38e54b 100644 --- a/docs/en/antalya/cas/configuration.md +++ b/docs/en/antalya/cas/configuration.md @@ -106,9 +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_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_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, and the `pending_deletes` blob `HEAD` + conditional `DELETE` fan-out. 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_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 | diff --git a/docs/en/operations/storing-data.md b/docs/en/operations/storing-data.md index ac3ed64e7982..0a6b733a9cc2 100644 --- a/docs/en/operations/storing-data.md +++ b/docs/en/operations/storing-data.md @@ -550,16 +550,19 @@ disk-level and server-level settings surface. - `cas_gc_meta_pool_size` — `16` by default. Bounded thread-pool size for the GC's per-hash freshness-meta writes (condemn/spare/delete), so a mass `DROP` condemning millions of blobs does not run fully sequentially. -- `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. +- `cas_gc_io_concurrency` — `16` by default. Bounded thread-pool size for the GC object-storage requests + that run in parallel. It covers: + - the fold's read-ahead of checkpoints, ref logs, manifest bodies and zero-candidate `HEAD`s; + - the planning reads of the orphan-manifest sweep and the read-ahead of `SYSTEM CAS GC REBUILD`; + - the `pending_deletes` phase, which runs one `HEAD` and one conditional `DELETE` (`If-Match`) per blob. + + It does not cover the per-hash freshness-meta writes (`cas_gc_meta_pool_size`) or any other GC request + (`LIST`, `gc/state` updates, manifest and ref-object batch deletes, generation pruning, the namespace + janitor, orphan-manifest deletes): those run on the round thread. Decisions, outcomes, events and the + audit log stay on the round thread in their original order; only the requests overlap. An entry is + recorded as deleted only if its own `HEAD` and `DELETE` ran; entries whose request failed, or that were + not submitted, stay pending and are retried in the next round. `1` runs all covered requests + sequentially on the round thread. - `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. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp index 7da0e01c0774..ebd28e1d0862 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp @@ -85,9 +85,7 @@ namespace ContentAddressedSetting extern const ContentAddressedSettingsUInt64 part_folder_cache_max_entry_bytes; 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_io_concurrency; extern const ContentAddressedSettingsUInt64 gc_bulk_delete_chunk_keys; extern const ContentAddressedSettingsUInt64 attempt_timeout_ms; extern const ContentAddressedSettingsUInt64 lease_safety_margin_ms; @@ -311,9 +309,7 @@ ContentAddressedMetadataStorage::ContentAddressedMetadataStorage( , cas_part_folder_cache_max_entry_bytes(settings_[ContentAddressedSetting::part_folder_cache_max_entry_bytes].value) , 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_io_concurrency(settings_[ContentAddressedSetting::gc_io_concurrency].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) @@ -804,9 +800,7 @@ ContentAddressedMetadataStorage::PoolView ContentAddressedMetadataStorage::openP pool_config.gc_round_handoff_prefix_wholesale_budget = gc_round_handoff_prefix_wholesale_budget; 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_io_concurrency = gc_io_concurrency; 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; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h index b8dd4dc32933..0e01c5738781 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h @@ -635,10 +635,8 @@ class ContentAddressedMetadataStorage final : public IMetadataStorage, public IC const uint64_t manifest_decode_cache_bytes; /// Bounded pool size for GC's per-hash freshness-metadata writes. 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; + /// Maximum number of threads in the GC I/O pool; 1 disables parallel GC I/O. + const uint64_t gc_io_concurrency; /// 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 diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp index 654693a3528e..7331d2bad29a 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp @@ -79,9 +79,7 @@ constexpr std::string_view CAS_KEY_PREFIX = "cas_"; DECLARE(UInt64, part_folder_cache_max_entry_bytes, 16ULL << 20, "Oversized part-folder views bypass retention above this size", 0) \ 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_io_concurrency, 16, "Maximum number of threads in the GC I/O pool. Used for fold and rebuild read-ahead, orphan-manifest sweep planning reads, and pending_deletes HEAD plus conditional DELETE. Per-hash meta writes use gc_meta_pool_size; other GC requests run on the round thread. 1 disables parallel GC I/O", 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) \ @@ -231,22 +229,12 @@ void ContentAddressedSettings::validate() auto & settings = *this; if (settings[ContentAddressedSetting::gc_interval_sec] == 0 || settings[ContentAddressedSetting::gc_shards] == 0 - || settings[ContentAddressedSetting::gc_read_concurrency] == 0) + || settings[ContentAddressedSetting::gc_io_concurrency] == 0) throw Exception(ErrorCodes::BAD_ARGUMENTS, - "content_addressed disk: cas_gc_interval_sec, cas_gc_shards and cas_gc_read_concurrency must be >= 1 " + "content_addressed disk: cas_gc_interval_sec, cas_gc_shards and cas_gc_io_concurrency must be >= 1 " "(got {}, {}, {})", 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); + settings[ContentAddressedSetting::gc_io_concurrency].value); if (settings[ContentAddressedSetting::gc_bulk_delete_chunk_keys] == 0 || settings[ContentAddressedSetting::gc_bulk_delete_chunk_keys] > 1000) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index d3aa6e52f170..520c0645ec8e 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -20,6 +20,8 @@ #include #include #include +#include +#include #include #include #include @@ -341,19 +343,16 @@ Gc::Gc(PoolPtr store_, UInt128 gc_id_, std::function now_ms_fn_, /// `store->poolConfig()` AFTER the null check above. meta_writer = std::make_unique( store, logger, static_cast(store->poolConfig().gc_meta_pool_size)); - /// The fold's read-ahead pool, built here for the same reason. The queue is UNBOUNDED because the - /// hinting sites throttle themselves against `GcReadAhead::window`; a bounded queue would only - /// move the throttle into `scheduleOrThrowOnError`, blocking the round thread instead of the - /// hint loop that already knows how much it wants in flight. - const size_t read_concurrency = std::max(1, store->poolConfig().gc_read_concurrency); - read_pool = std::make_unique( + /// The GC I/O pool (read-ahead and the `pending_deletes` fan-out), built here for the same reason. + /// The queue is UNBOUNDED because the hinting sites throttle themselves against + /// `GcReadAhead::window`; a bounded queue would only move the throttle into + /// `scheduleOrThrowOnError`, blocking the round thread instead of the hint loop that already knows + /// how much it wants in flight. + const size_t io_concurrency = std::max(1, store->poolConfig().gc_io_concurrency); + io_pool = std::make_unique( 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(1, store->poolConfig().gc_redelete_concurrency); - if (redelete_concurrency > 1) - redelete_pool = std::make_unique( - CurrentMetrics::LocalThread, CurrentMetrics::LocalThreadActive, CurrentMetrics::LocalThreadScheduled, - /*max_threads*/ redelete_concurrency, /*max_free_threads*/ redelete_concurrency, /*queue_size*/ 0); + /*max_threads*/ io_concurrency, /*max_free_threads*/ io_concurrency, /*queue_size*/ 0, + /*shutdown_on_exception*/ false); } void Gc::runNamespaceJanitorPage( @@ -499,6 +498,8 @@ void Gc::redeleteBlob( void Gc::redeleteBlobs( const std::vector & entries, + ThreadPool & pool, + size_t concurrency, const Layout & layout, CasOperation & op, uint64_t new_round, @@ -507,57 +508,83 @@ void Gc::redeleteBlobs( RoundReport & report, OutcomeLog & outcome_log) { - if (!redelete_pool || entries.size() < store->poolConfig().gc_redelete_min_batch_size) + if (concurrency <= 1 || entries.size() <= 1) { for (const RetiredEntry & entry : entries) redeleteBlob(entry, layout, op, new_round, snap_generation, round_work_budget, report, outcome_log); return; } + using RunnerTask = ThreadPoolCallbackRunnerLocal::Task; std::vector io_results(entries.size()); - const uint64_t admitted_generation = op.generation(); - size_t scheduled = 0; + std::vector errors(entries.size()); std::exception_ptr first_error; - try + size_t scheduled = 0; { - for (; scheduled < entries.size(); ++scheduled) + ThreadPoolCallbackRunnerLocal runner(pool, ThreadName::UNKNOWN); + std::vector> handles; + handles.reserve(entries.size()); + SCOPE_EXIT_SAFE({ ThreadPoolCallbackRunnerLocal::waitForAllToFinish(handles); }); + + const uint64_t admitted_generation = op.generation(); + try { - redelete_pool->scheduleOrThrowOnError( - [&, i = scheduled] - { - try - { - CasOperation job_op = store->openRequests().resume(admitted_generation); - io_results[i] = performRedeleteIo(entries[i], layout, job_op); - } - catch (...) + for (; scheduled < entries.size(); ++scheduled) + { + handles.emplace_back(runner.enqueueAndGiveOwnership( + [slot = &io_results[scheduled], + entry = entries[scheduled], + layout_ptr = &layout, + pool_store = store, + gc_logger = logger, + admitted_generation] { - io_results[i].error = std::current_exception(); - ProfileEvents::increment(ProfileEvents::CASGCRetiredRedeleteFailed); - LOG_WARNING( - logger, - "CAS gc: pending delete of blob {} (key `{}`, condemned at round {}) failed; the entry stays " - "delete_pending and is retried in the next round: {}", - blobIdOf(entries[i].ref), - layout.blobKey(entries[i].ref), - entries[i].condemn_round, - getCurrentExceptionMessage(false)); - } - }); + try + { + CasOperation job_op = pool_store->openRequests().resume(admitted_generation); + *slot = performRedeleteIo(entry, *layout_ptr, job_op); + } + catch (...) + { + ProfileEvents::increment(ProfileEvents::CASGCRetiredRedeleteFailed); + LOG_WARNING( + gc_logger, + "CAS gc: pending delete of blob {} (key `{}`, condemned at round {}) failed; the entry stays " + "delete_pending and is retried in the next round: {}", + blobIdOf(entry.ref), + layout_ptr->blobKey(entry.ref), + entry.condemn_round, + getCurrentExceptionMessage(false)); + throw; + } + })); + } + } + catch (...) + { + first_error = std::current_exception(); + } + + ThreadPoolCallbackRunnerLocal::waitForAllToFinish(handles); + for (size_t i = 0; i < handles.size(); ++i) + { + try + { + handles[i]->future.get(); + } + catch (...) + { + errors[i] = 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 (errors[i]) { if (!first_error) - first_error = io_results[i].error; + first_error = errors[i]; continue; } applyRedeleteOutcome(entries[i], io_results[i], new_round, snap_generation, round_work_budget, report, outcome_log); @@ -886,7 +913,9 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al static const std::vector kNothingToDelete; const std::vector & redelete_now = suppress_destructive ? kNothingToDelete : merge.redelete; - redeleteBlobs(redelete_now, layout, op, new_round, generation, round_work_budget, report, outcomes[shard]); + redeleteBlobs( + redelete_now, *io_pool, store->poolConfig().gc_io_concurrency, 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 @@ -1795,8 +1824,8 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, /// The fold's read-ahead. It fetches through `op`'s own admitted generation and hands every result /// back at the site that would otherwise have read inline, so the walk's order, its counters, its /// holds and its events are what they were; only the moment of the fetch moves. At - /// `gc_read_concurrency` 1 it hints nothing and every take IS the original inline read. - GcReadAhead reads(op, store->openRequests(), *read_pool, store->poolConfig().gc_read_concurrency); + /// `gc_io_concurrency` 1 it hints nothing and every take IS the original inline read. + GcReadAhead reads(op, store->openRequests(), *io_pool, store->poolConfig().gc_io_concurrency); FoldResult result; /// 1. Group the round's one enumeration of `cas/ns/stream/` (taken before the defer decision) into @@ -2169,7 +2198,7 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, /// position, this epoch's next positions, and a decoded log's manifest edges -- so the phase's round /// trips overlap instead of running strictly one after another. Every take happens where the inline /// read happened, in the same order, and increments the same counters, which is why this row's - /// semantic metrics are identical at any `gc_read_concurrency`. Its S3 VERB counts are not: a request + /// semantic metrics are identical at any `gc_io_concurrency`. Its S3 VERB counts are not: a request /// a worker performed lands on that worker's ProfileEvents, the same gap `meta_pool_wait` has always /// had. Read `CASGCReadAheadHit`/`Miss`/`Wasted` on this row for the read-ahead's own behaviour. std::optional intake_timer; @@ -3381,7 +3410,7 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, /// cut and `_ckpt` frontier the round's own universe came from -- which is exactly what an /// authoritative universe means, and is why this is the gate's term and not a separate one. universe_authoritative, - &work_budget, read_pool.get(), store->poolConfig().gc_read_concurrency); + &work_budget, io_pool.get(), store->poolConfig().gc_io_concurrency); for (const ManifestSweepResult::Nomination & nomination : result.orphan_sweep.nominations) orphan_source_retirements.insert( orphan_source_retirements.end(), @@ -4166,7 +4195,7 @@ RebuildReport Gc::rebuildBaseline(bool force) /// them one at a time, exactly as it did before: a rebuild walks a plan it already holds rather /// than discovering its next key from the body it just read, so a lookahead would have nothing to /// hide behind. - GcReadAhead reads(op, store->openRequests(), *read_pool, store->poolConfig().gc_read_concurrency); + GcReadAhead reads(op, store->openRequests(), *io_pool, store->poolConfig().gc_io_concurrency); /// Read bookkeeping health before the lease (the lease acquire on an absent state CREATES a /// bootstrap body, which must not make scenario (а) look healthy). A generation-0 ref-baseline diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h index 6ea13d39dfdc..286332bdc61d 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h @@ -730,10 +730,9 @@ class Gc { String blob_key; Removal del = Removal::Gone; - std::exception_ptr error; }; - RedeleteIo performRedeleteIo(const RetiredEntry & entry, const Layout & layout, CasOperation & op); + static RedeleteIo performRedeleteIo(const RetiredEntry & entry, const Layout & layout, CasOperation & op); void applyRedeleteOutcome( const RetiredEntry & entry, @@ -756,6 +755,8 @@ class Gc void redeleteBlobs( const std::vector & entries, + ThreadPool & pool, + size_t concurrency, const Layout & layout, CasOperation & op, uint64_t new_round, @@ -1014,12 +1015,10 @@ class Gc /// initialized before that check. std::unique_ptr meta_writer; - /// The fold's read-ahead pool, sized by `gc_read_concurrency`. A `unique_ptr` for the same reason - /// as `meta_writer`: the size comes from `store->poolConfig()`, which may only be read after the - /// constructor body has validated `store`. - std::unique_ptr read_pool; - - std::unique_ptr redelete_pool; + /// The GC I/O pool: the fold's and rebuild's read-ahead, the orphan-manifest sweep planning reads, + /// and the `pending_deletes` fan-out, sized by `gc_io_concurrency`. A `unique_ptr` for the same reason as `meta_writer`: the size comes from + /// `store->poolConfig()`, which may only be read after the constructor body has validated `store`. + std::unique_ptr io_pool; /// Probe B1's two numbers for the round: the ref-log POSITIONS the sealed coverage declares covered /// (counted arithmetically over each namespace's cut -- not by listed ids, which under arithmetic diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index e07c3de11e0e..20938991c155 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -174,13 +174,14 @@ struct PoolConfig /// feedback_ca_gc_never_throw_on_404) and `Gc::runRegularRound` waits for the round's whole batch /// before the round's single gc/state CAS, so the meta writes are durable before that CAS commits. uint64_t gc_meta_pool_size = 16; - /// Bounded pool size for the fold's read-ahead of checkpoints, ref logs, manifest bodies and - /// zero-candidate HEADs. Every decision stays on the round thread, in the order it always ran; - /// only the fetch overlaps. `1` issues no read-ahead at all and is the sequential round, request + /// Bounded pool size for the GC requests that overlap: the fold's read-ahead of checkpoints, ref + /// logs, manifest bodies and zero-candidate HEADs, the orphan-manifest sweep planning reads, the + /// rebuild read-ahead, and the `pending_deletes` HEAD + conditional DELETE fan-out. Meta writes + /// have their own pool (`gc_meta_pool_size`); every other GC request runs on the round thread. + /// Every decision stays on the round thread, in the order it always ran; only the requests + /// overlap. `1` issues no read-ahead and no fan-out at all and is the sequential round, request /// for request. - uint64_t gc_read_concurrency = 16; - uint64_t gc_redelete_concurrency = 1; - uint64_t gc_redelete_min_batch_size = 2; + uint64_t gc_io_concurrency = 16; /// Tests drive `renewWatermarkOnce` explicitly; gates both persistent runtime workers. bool background_watermark = false; /// Installed on the pool before a writable mount can start its runtime-owned workers. diff --git a/src/Disks/tests/gtest_cas_gc_read_ahead.cpp b/src/Disks/tests/gtest_cas_gc_read_ahead.cpp index cf8ad113a530..8a3cc408a8cf 100644 --- a/src/Disks/tests/gtest_cas_gc_read_ahead.cpp +++ b/src/Disks/tests/gtest_cas_gc_read_ahead.cpp @@ -316,7 +316,7 @@ void runFolds(uint64_t concurrency, size_t rounds, FoldRun & out) auto backend = std::make_shared(); auto store = Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test", - .gc_fold_max_defer_rounds = 0, .gc_read_concurrency = concurrency}); + .gc_fold_max_defer_rounds = 0, .gc_io_concurrency = concurrency}); populate(store); CasRequests requests = openRequestsForTest(backend); @@ -521,7 +521,7 @@ TEST(CASGCReadAhead, TheFoldsReadsActuallyOverlap) auto backend = std::make_shared(/*k_overlap*/ 2); auto store = Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test", - .gc_fold_max_defer_rounds = 0, .gc_read_concurrency = 8}); + .gc_fold_max_defer_rounds = 0, .gc_io_concurrency = 8}); populate(store); Gc gc(store, kGc); @@ -539,7 +539,7 @@ TEST(CASGCReadAhead, WorkerReadFaultFailsTheRoundAndTheNextRoundRecovers) auto backend = std::make_shared(); auto store = Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test", - .gc_fold_max_defer_rounds = 0, .gc_read_concurrency = 8}); + .gc_fold_max_defer_rounds = 0, .gc_io_concurrency = 8}); populate(store); Gc gc(store, kGc); diff --git a/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp index b0fc35e3cd5b..c2cbbb0b7206 100644 --- a/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp +++ b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp @@ -1,11 +1,22 @@ #include +#include #include +#include +#include +#include +#include +#include #include +#include +#include +#include #include #include +#include +#include #include #include #include "cas_test_helpers.h" @@ -25,25 +36,103 @@ const DB::UInt128 kGc = hexToU128("00000000000000000000000000000001"); constexpr uint64_t kBlobs = 6; constexpr uint64_t kFaultedBlob = 3; -class RemoveFaultBackend : public InMemoryBackend +class WorkerRemoveFaultBackend : public InMemoryBackend { public: + void armAgainstOtherThreads(String key) + { + faulted_key = std::move(key); + owner = std::this_thread::get_id(); + armed.store(true); + } + + bool fired() const { return !armed.load(); } + RawRemoval remove(const String & key, const String & expected_value, TransportAccess & access) override { - if (key == faulted_key && armed.exchange(false)) - throw std::runtime_error("injected remove fault"); + if (armed.load() && key == faulted_key && std::this_thread::get_id() != owner && armed.exchange(false)) + throw std::runtime_error("injected worker remove fault"); return InMemoryBackend::remove(key, expected_value, access); } +private: String faulted_key; + std::thread::id owner; std::atomic armed{false}; }; +class BlobRemoveWitnessBackend : public InMemoryBackend +{ +public: + void arm(std::set blob_keys_, size_t k_overlap_) + { + blob_keys = std::move(blob_keys_); + k_overlap = k_overlap_; + armed.store(true); + } + + bool sawOverlap() const { return saw_overlap.load(); } + + size_t peakInFlight() const + { + std::lock_guard lock(mutex); + return peak; + } + + size_t blobRemoves() const + { + std::lock_guard lock(mutex); + return total; + } + + RawRemoval remove(const String & key, const String & expected_value, TransportAccess & access) override + { + if (!armed.load() || !blob_keys.contains(key)) + return InMemoryBackend::remove(key, expected_value, access); + { + std::unique_lock lock(mutex); + ++total; + ++in_flight; + peak = std::max(peak, in_flight); + if (k_overlap > 1) + { + if (in_flight >= k_overlap) + { + saw_overlap.store(true); + gate.notify_all(); + } + else + { + gate.wait_for(lock, std::chrono::milliseconds(250), + [&] { return in_flight >= k_overlap || saw_overlap.load(); }); + } + } + } + const RawRemoval result = InMemoryBackend::remove(key, expected_value, access); + { + std::lock_guard lock(mutex); + --in_flight; + } + return result; + } + +private: + std::set blob_keys; + size_t k_overlap = 1; + std::atomic armed{false}; + mutable std::mutex mutex; + std::condition_variable gate; + size_t in_flight = 0; + size_t peak = 0; + size_t total = 0; + std::atomic saw_overlap{false}; +}; + template -PoolPtr openPoolWithRedeleteConcurrency(std::shared_ptr backend, uint64_t concurrency) +PoolPtr openPoolWithIoConcurrency(std::shared_ptr backend, uint64_t concurrency) { PoolConfig config{.pool_prefix = "p", .server_root_id = "test"}; - config.gc_redelete_concurrency = concurrency; + config.gc_io_concurrency = concurrency; return Pool::open(std::move(backend), std::move(config)); } @@ -52,6 +141,14 @@ String blobKeyOf(const Pool & store, uint64_t blob) return store.layout().blobKey(BlobRef{BlobHashAlgo::CityHash128, BlobDigest::fromU128(DB::UInt128(blob))}); } +std::set allBlobKeys(const Pool & store) +{ + std::set keys; + for (uint64_t b = 1; b <= kBlobs; ++b) + keys.insert(blobKeyOf(store, b)); + return keys; +} + bool allBlobsAbsent(Backend & backend, const Pool & store) { for (uint64_t b = 1; b <= kBlobs; ++b) @@ -80,12 +177,62 @@ void publishThenDrop(Backend & backend, const PoolPtr & store, Gc & gc) dropRefTransition(backend, store->layout(), ns, "tbl", r); } +using OutcomeRow = std::pair; + +struct RedeleteRun +{ + std::vector> reports; + std::map> outcome_logs; +}; + +void collectOutcomeLogs(Backend & backend, std::map> & out) +{ + OperationForTest op(backend); + String cursor; + while (true) + { + const ListPage page = (*op).list("", cursor, 1000, Retry::standard()); + for (const ListedKey & listed : page.keys) + { + if (listed.key.find("/outcomes/") == String::npos || out.contains(listed.key)) + continue; + const auto object = (*op).read(listed.key, Retry::standard()); + if (!object) + continue; + std::vector rows; + for (const OutcomeEntry & entry : decodeOutcomeLog(openObject(FormatId::GcOutcomes, object->bytes)).entries) + rows.emplace_back(entry.ref, entry.outcome); + out.emplace(listed.key, std::move(rows)); + } + if (page.next_cursor.empty()) + break; + cursor = page.next_cursor; + } +} + +void runRedeleteScenario(uint64_t concurrency, RedeleteRun & out) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, concurrency); + Gc gc(store, kGc); + ASSERT_NO_FATAL_FAILURE(publishThenDrop(*backend, store, gc)); + + for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) + { + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + out.reports.push_back({rep.redeleted, rep.deleted, rep.absent, rep.replaced, rep.spared}); + collectOutcomeLogs(*backend, out.outcome_logs); + } + ASSERT_TRUE(allBlobsAbsent(*backend, *store)); +} + } TEST(CASGCRedeleteConcurrency, ParallelRedeleteReclaimsEveryBlob) { auto backend = std::make_shared(); - auto store = openPoolWithRedeleteConcurrency(backend, 4); + auto store = openPoolWithIoConcurrency(backend, 4); Gc gc(store, kGc); publishThenDrop(*backend, store, gc); @@ -104,16 +251,55 @@ TEST(CASGCRedeleteConcurrency, ParallelRedeleteReclaimsEveryBlob) EXPECT_EQ(deleted, kBlobs); } -TEST(CASGCRedeleteConcurrency, FailedRemoveKeepsSiblingOutcomesAndPoolAlive) +TEST(CASGCRedeleteConcurrency, BlobDeletesActuallyOverlap) { - auto backend = std::make_shared(); - auto store = openPoolWithRedeleteConcurrency(backend, 4); - backend->faulted_key = blobKeyOf(*store, kFaultedBlob); - backend->armed = true; + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + + backend->arm(allBlobKeys(*store), /*k_overlap*/ 2); + for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) + { + runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + } + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); + EXPECT_EQ(backend->blobRemoves(), kBlobs); + EXPECT_TRUE(backend->sawOverlap()) + << "no two blob deletes were ever in the backend at the same time; peak in flight was " + << backend->peakInFlight(); + EXPECT_GT(backend->peakInFlight(), 1u); +} + +TEST(CASGCRedeleteConcurrency, ConcurrencyOneDeletesSequentially) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 1); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + + backend->arm(allBlobKeys(*store), /*k_overlap*/ 1); + for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) + { + runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + } + + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); + EXPECT_EQ(backend->blobRemoves(), kBlobs); + EXPECT_EQ(backend->peakInFlight(), 1u); +} + +TEST(CASGCRedeleteConcurrency, WorkerRemoveFaultKeepsSiblingOutcomesAndPoolAlive) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); Gc gc(store, kGc); publishThenDrop(*backend, store, gc); + backend->armAgainstOtherThreads(blobKeyOf(*store, kFaultedBlob)); const auto failed_before = ProfileEvents::global_counters[ProfileEvents::CASGCRetiredRedeleteFailed].load(); size_t failed_rounds = 0; for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) @@ -133,8 +319,32 @@ TEST(CASGCRedeleteConcurrency, FailedRemoveKeepsSiblingOutcomesAndPoolAlive) store->renewWatermarkOnce(); } + EXPECT_TRUE(backend->fired()) << "the faulted blob delete never ran on a pool worker"; EXPECT_EQ(failed_rounds, 1u); EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASGCRetiredRedeleteFailed].load() - failed_before, 1u); - EXPECT_FALSE(backend->armed.load()); EXPECT_TRUE(allBlobsAbsent(*backend, *store)); } + +TEST(CASGCRedeleteConcurrency, SameOutcomesAtConcurrencyOneAndFour) +{ + RedeleteRun one; + RedeleteRun four; + ASSERT_NO_FATAL_FAILURE(runRedeleteScenario(1, one)); + ASSERT_NO_FATAL_FAILURE(runRedeleteScenario(4, four)); + + EXPECT_EQ(one.reports, four.reports); + + std::vector> logs_one; + for (const auto & [key, rows] : one.outcome_logs) + logs_one.push_back(rows); + std::vector> logs_four; + for (const auto & [key, rows] : four.outcome_logs) + logs_four.push_back(rows); + EXPECT_EQ(logs_one, logs_four); + + size_t deleted_rows = 0; + for (const auto & rows : logs_one) + deleted_rows += static_cast( + std::count_if(rows.begin(), rows.end(), [](const OutcomeRow & row) { return row.second == OutcomeKind::Deleted; })); + EXPECT_EQ(deleted_rows, kBlobs) << "the scenario must delete every blob through the outcome log, or ordering is untested"; +} diff --git a/src/Disks/tests/gtest_cas_ref_gc.cpp b/src/Disks/tests/gtest_cas_ref_gc.cpp index c76f68d0489b..835ddb197cbe 100644 --- a/src/Disks/tests/gtest_cas_ref_gc.cpp +++ b/src/Disks/tests/gtest_cas_ref_gc.cpp @@ -197,7 +197,7 @@ class RefCleanupAuthorityRaceBackend : public CountingBackend } /// `read` also runs on the GC read-ahead pool's threads (`CasGcReadAhead.cpp` schedules - /// `CasOperation::read` there; `gc_read_concurrency` defaults to 16), concurrently with the round + /// `CasOperation::read` there; `gc_io_concurrency` defaults to 16), concurrently with the round /// thread's own reads -- `catalog_seam_armed` and `last_control_key_read` below are shared mutable /// state a pool thread's read can land between `authorityHolds`'s two reads, so both are read AND /// written only under `seam_mutex`. `last_control_key_read` tracks only the catalog and `gc/state` diff --git a/src/Disks/tests/gtest_cas_settings.cpp b/src/Disks/tests/gtest_cas_settings.cpp index 554bff83b95c..eb83a880bac9 100644 --- a/src/Disks/tests/gtest_cas_settings.cpp +++ b/src/Disks/tests/gtest_cas_settings.cpp @@ -26,8 +26,6 @@ namespace DB::ContentAddressedSetting extern const ContentAddressedSettingsUInt64 gc_shards; extern const ContentAddressedSettingsUInt64 gc_interval_sec; extern const ContentAddressedSettingsUInt64 gc_bulk_delete_chunk_keys; - extern const ContentAddressedSettingsUInt64 gc_redelete_concurrency; - extern const ContentAddressedSettingsUInt64 gc_redelete_min_batch_size; extern const ContentAddressedSettingsString scratch_path; extern const ContentAddressedSettingsBool unsafe_remount_no_delay; } @@ -178,15 +176,15 @@ TEST(CASContentAddressedSettings, InvalidBoundsDiagnosticNamesExternalConfigKeys expectLoadFailureWithExactMessage( "srv10", ErrorCodes::BAD_ARGUMENTS, - "content_addressed disk: cas_gc_interval_sec, cas_gc_shards and cas_gc_read_concurrency must be >= 1 " + "content_addressed disk: cas_gc_interval_sec, cas_gc_shards and cas_gc_io_concurrency must be >= 1 " "(got 60, 0, 16)"); - /// The fold's read-ahead pool is refused at zero for the same reason the shard count is: a zero - /// would be a silently disabled subsystem rather than a configuration the pool can honour. One is - /// the sequential fold and is the way to turn the read-ahead off. + /// The GC I/O pool is refused at zero for the same reason the shard count is: a zero would be a + /// silently disabled subsystem rather than a configuration the pool can honour. One is the way to + /// run the covered GC I/O sequentially. expectLoadFailureWithExactMessage( - "srv10", + "srv10", ErrorCodes::BAD_ARGUMENTS, - "content_addressed disk: cas_gc_interval_sec, cas_gc_shards and cas_gc_read_concurrency must be >= 1 " + "content_addressed disk: cas_gc_interval_sec, cas_gc_shards and cas_gc_io_concurrency must be >= 1 " "(got 60, 1, 0)"); } @@ -211,46 +209,6 @@ TEST(CASSettings, BulkDeleteChunkKeysBoundsAreEnforced) EXPECT_EQ(s[ContentAddressedSetting::gc_bulk_delete_chunk_keys].value, 1u); } -TEST(CASSettings, RedeleteConcurrencyBoundsAreEnforced) -{ - expectLoadFailureWithExactMessage( - "srv1" - "0", - ErrorCodes::BAD_ARGUMENTS, - "content_addressed disk: cas_gc_redelete_concurrency must be >= 1 (got 0)"); - - { - auto cfg = makeConfig("srv1"); - ContentAddressedSettings s; - s.loadFromConfig(*cfg, "disk", "/scratch", "/scratch", identity_macros); - EXPECT_EQ(s[ContentAddressedSetting::gc_redelete_concurrency].value, 1u); - EXPECT_EQ(s[ContentAddressedSetting::gc_redelete_min_batch_size].value, 2u); - } - - auto cfg = makeConfig( - "srv1" - "8"); - ContentAddressedSettings s; - EXPECT_NO_THROW(s.loadFromConfig(*cfg, "disk", "/scratch", "/scratch", identity_macros)); - EXPECT_EQ(s[ContentAddressedSetting::gc_redelete_concurrency].value, 8u); -} - -TEST(CASSettings, RedeleteMinBatchSizeBoundsAreEnforced) -{ - expectLoadFailureWithExactMessage( - "srv1" - "0", - ErrorCodes::BAD_ARGUMENTS, - "content_addressed disk: cas_gc_redelete_min_batch_size must be >= 1 (got 0)"); - - auto cfg = makeConfig( - "srv1" - "7"); - ContentAddressedSettings s; - EXPECT_NO_THROW(s.loadFromConfig(*cfg, "disk", "/scratch", "/scratch", identity_macros)); - EXPECT_EQ(s[ContentAddressedSetting::gc_redelete_min_batch_size].value, 7u); -} - TEST(CASContentAddressedSettings, InvalidEnumDiagnosticsNameExternalConfigKeys) { expectLoadFailureWithExactMessage( From 6ee7cf20d191891a0aec433531c047ace7790217 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 15 Sep 2026 13:45:24 +0200 Subject: [PATCH 6/7] add tests Signed-off-by: Konstantin Morozov --- .../ContentAddressed/Gc/CasGc.cpp | 2 + .../ContentAddressed/Pool/CasPool.h | 2 + .../gtest_cas_gc_redelete_concurrency.cpp | 632 +++++++++++++++++- 3 files changed, 609 insertions(+), 27 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index 520c0645ec8e..54058a0bb358 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -426,6 +426,8 @@ void Gc::applyRedeleteOutcome( RoundReport & report, OutcomeLog & outcome_log) { + if (const auto & hook = store->poolConfig().gc_redelete_apply_hook_for_test) + hook(entry.ref); const OutcomeKind outcome_kind = io.del == Removal::Removed ? OutcomeKind::Deleted : io.del == Removal::Gone ? OutcomeKind::Absent : OutcomeKind::Replaced; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index 20938991c155..92cf9bbaf49c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -213,6 +213,8 @@ struct PoolConfig std::function teardown_phase2_throw_for_test = {}; std::function teardown_phase3_throw_for_test = {}; + std::function gc_redelete_apply_hook_for_test = {}; + /// Mount-lease TTL: how long a freshly-renewed mount lease is valid. The local /// write fence's monotonic deadline is `renew_time + this`, so a superseded/paused writer is fenced /// once `this` elapses with no successful renew. The background renewer runs every diff --git a/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp index c2cbbb0b7206..ee5b3858b715 100644 --- a/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp +++ b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp @@ -10,14 +10,19 @@ #include #include #include +#include #include +#include #include +#include +#include #include #include #include #include +#include #include #include "cas_test_helpers.h" @@ -26,6 +31,11 @@ namespace ProfileEvents extern const Event CASGCRetiredRedeleteFailed; } +namespace DB::ErrorCodes +{ +extern const int CANNOT_SCHEDULE_TASK; +} + using namespace DB::Cas; using namespace DB::Cas::tests; @@ -35,30 +45,61 @@ namespace const DB::UInt128 kGc = hexToU128("00000000000000000000000000000001"); constexpr uint64_t kBlobs = 6; constexpr uint64_t kFaultedBlob = 3; +constexpr uint64_t kReplacedBlob = 5; +constexpr uint64_t kAbsentBlob = 4; +constexpr uint64_t kShards = 4; + +DB::UInt128 blobHash(uint64_t blob) +{ + const DB::UInt128 value(blob); + return (value << 64) | value; +} -class WorkerRemoveFaultBackend : public InMemoryBackend +class WorkerFaultBackend : public InMemoryBackend { public: - void armAgainstOtherThreads(String key) + void armRemoveFaults(std::set keys) { arm(remove_keys, std::move(keys)); } + + void armHeadFaults(std::set keys) { arm(head_keys, std::move(keys)); } + + bool allFired() const { - faulted_key = std::move(key); - owner = std::this_thread::get_id(); - armed.store(true); + std::lock_guard lock(mutex); + return remove_keys.empty() && head_keys.empty(); } - bool fired() const { return !armed.load(); } - RawRemoval remove(const String & key, const String & expected_value, TransportAccess & access) override { - if (armed.load() && key == faulted_key && std::this_thread::get_id() != owner && armed.exchange(false)) + if (takeFault(remove_keys, key)) throw std::runtime_error("injected worker remove fault"); return InMemoryBackend::remove(key, expected_value, access); } + std::optional head(const String & key, TransportAccess & access) override + { + if (takeFault(head_keys, key)) + throw std::runtime_error("injected worker head fault"); + return InMemoryBackend::head(key, access); + } + private: - String faulted_key; + void arm(std::set & target, std::set keys) + { + std::lock_guard lock(mutex); + owner = std::this_thread::get_id(); + target = std::move(keys); + } + + bool takeFault(std::set & keys, const String & key) + { + std::lock_guard lock(mutex); + return std::this_thread::get_id() != owner && keys.erase(key) > 0; + } + + mutable std::mutex mutex; std::thread::id owner; - std::atomic armed{false}; + std::set remove_keys; + std::set head_keys; }; class BlobRemoveWitnessBackend : public InMemoryBackend @@ -103,8 +144,7 @@ class BlobRemoveWitnessBackend : public InMemoryBackend } else { - gate.wait_for(lock, std::chrono::milliseconds(250), - [&] { return in_flight >= k_overlap || saw_overlap.load(); }); + gate.wait_for(lock, std::chrono::milliseconds(250), [&] { return in_flight >= k_overlap || saw_overlap.load(); }); } } } @@ -129,16 +169,21 @@ class BlobRemoveWitnessBackend : public InMemoryBackend }; template -PoolPtr openPoolWithIoConcurrency(std::shared_ptr backend, uint64_t concurrency) +PoolPtr openPoolWithIoConcurrency( + std::shared_ptr backend, uint64_t concurrency, uint64_t gc_shards = 1, uint64_t outcome_entry_budget = 0, + CasEventSink event_sink = {}) { PoolConfig config{.pool_prefix = "p", .server_root_id = "test"}; config.gc_io_concurrency = concurrency; + config.gc_shards = gc_shards; + config.gc_round_outcome_entry_budget = outcome_entry_budget; + config.event_sink = std::move(event_sink); return Pool::open(std::move(backend), std::move(config)); } String blobKeyOf(const Pool & store, uint64_t blob) { - return store.layout().blobKey(BlobRef{BlobHashAlgo::CityHash128, BlobDigest::fromU128(DB::UInt128(blob))}); + return store.layout().blobKey(BlobRef{BlobHashAlgo::CityHash128, BlobDigest::fromU128(blobHash(blob))}); } std::set allBlobKeys(const Pool & store) @@ -152,7 +197,7 @@ std::set allBlobKeys(const Pool & store) bool allBlobsAbsent(Backend & backend, const Pool & store) { for (uint64_t b = 1; b <= kBlobs; ++b) - if (!blobAbsent(backend, store.layout(), DB::UInt128(b))) + if (!blobAbsent(backend, store.layout(), blobHash(b))) return false; return true; } @@ -164,15 +209,15 @@ void publishThenDrop(Backend & backend, const PoolPtr & store, Gc & gc) std::vector entries; for (uint64_t b = 1; b <= kBlobs; ++b) { - writeBlobBody(backend, store->layout(), DB::UInt128(b)); - entries.push_back(blobEntryFor("f" + std::to_string(b), DB::UInt128(b))); + writeBlobBody(backend, store->layout(), blobHash(b)); + entries.push_back(blobEntryFor("f" + std::to_string(b), blobHash(b))); } writeManifestRaw(backend, store->layout(), ns, r, entries); publishCommittedTransition(backend, store->layout(), ns, "tbl", std::nullopt, r); runRegularRoundReclaiming(gc); store->renewWatermarkOnce(); - ASSERT_FALSE(blobAbsent(backend, store->layout(), DB::UInt128(1))); + ASSERT_FALSE(blobAbsent(backend, store->layout(), blobHash(1))); dropRefTransition(backend, store->layout(), ns, "tbl", r); } @@ -210,10 +255,10 @@ void collectOutcomeLogs(Backend & backend, std::map(); - auto store = openPoolWithIoConcurrency(backend, concurrency); + auto store = openPoolWithIoConcurrency(backend, concurrency, 1, outcome_entry_budget); Gc gc(store, kGc); ASSERT_NO_FATAL_FAILURE(publishThenDrop(*backend, store, gc)); @@ -267,9 +312,8 @@ TEST(CASGCRedeleteConcurrency, BlobDeletesActuallyOverlap) EXPECT_TRUE(allBlobsAbsent(*backend, *store)); EXPECT_EQ(backend->blobRemoves(), kBlobs); - EXPECT_TRUE(backend->sawOverlap()) - << "no two blob deletes were ever in the backend at the same time; peak in flight was " - << backend->peakInFlight(); + EXPECT_TRUE(backend->sawOverlap()) << "no two blob deletes were ever in the backend at the same time; peak in flight was " + << backend->peakInFlight(); EXPECT_GT(backend->peakInFlight(), 1u); } @@ -294,12 +338,12 @@ TEST(CASGCRedeleteConcurrency, ConcurrencyOneDeletesSequentially) TEST(CASGCRedeleteConcurrency, WorkerRemoveFaultKeepsSiblingOutcomesAndPoolAlive) { - auto backend = std::make_shared(); + auto backend = std::make_shared(); auto store = openPoolWithIoConcurrency(backend, 4); Gc gc(store, kGc); publishThenDrop(*backend, store, gc); - backend->armAgainstOtherThreads(blobKeyOf(*store, kFaultedBlob)); + backend->armRemoveFaults({blobKeyOf(*store, kFaultedBlob)}); const auto failed_before = ProfileEvents::global_counters[ProfileEvents::CASGCRetiredRedeleteFailed].load(); size_t failed_rounds = 0; for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) @@ -314,12 +358,12 @@ TEST(CASGCRedeleteConcurrency, WorkerRemoveFaultKeepsSiblingOutcomesAndPoolAlive ++failed_rounds; EXPECT_EQ(progress.redeleted, kBlobs - 1); for (uint64_t b = 1; b <= kBlobs; ++b) - EXPECT_EQ(blobAbsent(*backend, store->layout(), DB::UInt128(b)), b != kFaultedBlob) << "blob " << b; + EXPECT_EQ(blobAbsent(*backend, store->layout(), blobHash(b)), b != kFaultedBlob) << "blob " << b; } store->renewWatermarkOnce(); } - EXPECT_TRUE(backend->fired()) << "the faulted blob delete never ran on a pool worker"; + EXPECT_TRUE(backend->allFired()) << "the faulted blob delete never ran on a pool worker"; EXPECT_EQ(failed_rounds, 1u); EXPECT_EQ(ProfileEvents::global_counters[ProfileEvents::CASGCRetiredRedeleteFailed].load() - failed_before, 1u); EXPECT_TRUE(allBlobsAbsent(*backend, *store)); @@ -348,3 +392,537 @@ TEST(CASGCRedeleteConcurrency, SameOutcomesAtConcurrencyOneAndFour) std::count_if(rows.begin(), rows.end(), [](const OutcomeRow & row) { return row.second == OutcomeKind::Deleted; })); EXPECT_EQ(deleted_rows, kBlobs) << "the scenario must delete every blob through the outcome log, or ordering is untested"; } + +namespace +{ + +class RemoveRaceBackend : public InMemoryBackend +{ +public: + void armAgainstOtherThreads(String key) + { + raced_key = std::move(key); + owner = std::this_thread::get_id(); + armed.store(true); + } + + bool racedOnWorker() const { return raced_on_worker.load(); } + + bool replacementCommitted() const { return replacement_committed.load(); } + + RawRemoval remove(const String & key, const String & expected_value, TransportAccess & access) override + { + if (armed.load() && key == raced_key && armed.exchange(false)) + { + raced_on_worker.store(std::this_thread::get_id() != owner); + const std::optional current = InMemoryBackend::read(key, access); + if (current) + replacement_committed.store(InMemoryBackend::write(key, current->bytes, current->value, access).has_value()); + } + return InMemoryBackend::remove(key, expected_value, access); + } + +private: + String raced_key; + std::thread::id owner; + std::atomic armed{false}; + std::atomic raced_on_worker{false}; + std::atomic replacement_committed{false}; +}; + +BlobRef blobRefOf(uint64_t blob) +{ + return BlobRef{BlobHashAlgo::CityHash128, BlobDigest::fromU128(blobHash(blob))}; +} + +void driveUntilAllGraduated(Backend & backend, Gc & gc, const PoolPtr & store, uint64_t marked_blob) +{ + bool graduated = false; + for (int i = 0; i < 8 && !graduated; ++i) + { + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + ASSERT_EQ(rep.redeleted, 0u) << "no delete may run before the replacement is in place"; + graduated = rep.graduated == kBlobs; + } + ASSERT_TRUE(graduated) << "the six blobs never became delete_pending together"; + + OperationForTest op(backend); + ASSERT_TRUE((*op).head(store->layout().blobMetaKey(blobRefOf(marked_blob)), Retry::standard()).has_value()) + << "the condemn marker must exist before the redelete round, or the meta assertion is vacuous"; +} + +void expectOnlyReplacedBlobSurvives(Backend & backend, const Pool & store, const RoundReport & rep) +{ + EXPECT_EQ(rep.redeleted, kBlobs); + EXPECT_EQ(rep.deleted, kBlobs - 1); + EXPECT_EQ(rep.replaced, 1u); + + for (uint64_t b = 1; b <= kBlobs; ++b) + { + OperationForTest op(backend); + const bool body_present = (*op).head(blobKeyOf(store, b), Retry::standard()).has_value(); + const bool meta_present = (*op).head(store.layout().blobMetaKey(blobRefOf(b)), Retry::standard()).has_value(); + EXPECT_EQ(body_present, b == kReplacedBlob) << "blob " << b; + EXPECT_EQ(meta_present, b == kReplacedBlob) << "meta of blob " << b; + } + + std::map> logs; + collectOutcomeLogs(backend, logs); + size_t replaced_rows = 0; + size_t deleted_rows = 0; + for (const auto & [key, rows] : logs) + { + for (const auto & [ref, outcome] : rows) + { + if (outcome == OutcomeKind::Replaced) + { + EXPECT_EQ(ref, blobRefOf(kReplacedBlob)); + ++replaced_rows; + } + else if (outcome == OutcomeKind::Deleted) + { + EXPECT_NE(ref, blobRefOf(kReplacedBlob)); + ++deleted_rows; + } + } + } + EXPECT_EQ(replaced_rows, 1u); + EXPECT_EQ(deleted_rows, kBlobs - 1); +} + +} + +TEST(CASGCRedeleteConcurrency, ReplacedBlobInsideParallelBatchSurvivesWhileSiblingsDelete) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kReplacedBlob)); + + { + OperationForTest op(*backend); + const auto current = (*op).read(blobKeyOf(*store, kReplacedBlob), Retry::standard()); + ASSERT_TRUE(current); + ASSERT_TRUE(std::holds_alternative( + (*op).replace(blobKeyOf(*store, kReplacedBlob), current->bytes, current->etag, Retry::standard()))); + } + + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + expectOnlyReplacedBlobSurvives(*backend, *store, rep); +} + +TEST(CASGCRedeleteConcurrency, BlobReplacedBetweenHeadAndDeleteSurvivesWhileSiblingsDelete) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kReplacedBlob)); + + backend->armAgainstOtherThreads(blobKeyOf(*store, kReplacedBlob)); + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + + EXPECT_TRUE(backend->replacementCommitted()) << "the race never replaced the blob, so the 412 path is untested"; + EXPECT_TRUE(backend->racedOnWorker()) << "the raced delete must have run on a pool worker"; + expectOnlyReplacedBlobSurvives(*backend, *store, rep); +} + +TEST(CASGCRedeleteConcurrency, BlobAlreadyAbsentInsideParallelBatchIsRecordedAbsent) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kAbsentBlob)); + + { + OperationForTest op(*backend); + const auto current = (*op).head(blobKeyOf(*store, kAbsentBlob), Retry::standard()); + ASSERT_TRUE(current); + ASSERT_EQ((*op).remove(blobKeyOf(*store, kAbsentBlob), current->etag, Retry::standard()), Removal::Removed); + } + + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + EXPECT_EQ(rep.redeleted, kBlobs); + EXPECT_EQ(rep.deleted, kBlobs - 1); + EXPECT_EQ(rep.absent, 1u); + EXPECT_EQ(rep.replaced, 0u); + + for (uint64_t b = 1; b <= kBlobs; ++b) + { + OperationForTest op(*backend); + EXPECT_FALSE((*op).head(blobKeyOf(*store, b), Retry::standard()).has_value()) << "blob " << b; + EXPECT_FALSE((*op).head(store->layout().blobMetaKey(blobRefOf(b)), Retry::standard()).has_value()) << "meta of blob " << b; + } + + std::map> logs; + collectOutcomeLogs(*backend, logs); + size_t absent_rows = 0; + size_t deleted_rows = 0; + for (const auto & [key, rows] : logs) + { + for (const auto & [ref, outcome] : rows) + { + if (outcome == OutcomeKind::Absent) + { + EXPECT_EQ(ref, blobRefOf(kAbsentBlob)); + ++absent_rows; + } + else if (outcome == OutcomeKind::Deleted) + { + EXPECT_NE(ref, blobRefOf(kAbsentBlob)); + ++deleted_rows; + } + } + } + EXPECT_EQ(absent_rows, 1u); + EXPECT_EQ(deleted_rows, kBlobs - 1); +} + +namespace +{ + +uint64_t failedCounter() +{ + return ProfileEvents::global_counters[ProfileEvents::CASGCRetiredRedeleteFailed].load(); +} + +void expectBodiesPresentOnlyFor(Backend & backend, const Pool & store, const std::set & present) +{ + for (uint64_t b = 1; b <= kBlobs; ++b) + EXPECT_EQ(blobAbsent(backend, store.layout(), blobHash(b)), !present.contains(b)) << "blob " << b; +} + +struct ApplyHookState +{ + std::atomic armed{false}; + BlobRef target{}; + size_t applied = 0; +}; + +} + +TEST(CASGCRedeleteConcurrency, TwoWorkerFailuresAreEachCountedAndTheRoundThrowsOnce) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kFaultedBlob)); + + backend->armRemoveFaults({blobKeyOf(*store, kFaultedBlob), blobKeyOf(*store, kReplacedBlob)}); + const auto failed_before = failedCounter(); + RoundReport progress; + EXPECT_ANY_THROW(gc.runRegularRound({}, /*allow_steal*/ true, UniversePolicy::Authoritative, &progress)); + store->renewWatermarkOnce(); + + EXPECT_TRUE(backend->allFired()) << "both faulted deletes must have run on pool workers"; + EXPECT_EQ(failedCounter() - failed_before, 2u); + EXPECT_EQ(progress.redeleted, kBlobs - 2); + expectBodiesPresentOnlyFor(*backend, *store, {kFaultedBlob, kReplacedBlob}); + + const RoundReport next = runRegularRoundReclaiming(gc); + EXPECT_EQ(next.redeleted, kBlobs); + EXPECT_EQ(next.deleted, 2u); + EXPECT_EQ(next.absent, kBlobs - 2); + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); +} + +TEST(CASGCRedeleteConcurrency, WorkerHeadFaultIsCountedAndRetriedNextRound) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kFaultedBlob)); + + backend->armHeadFaults({blobKeyOf(*store, kFaultedBlob)}); + const auto failed_before = failedCounter(); + RoundReport progress; + EXPECT_ANY_THROW(gc.runRegularRound({}, /*allow_steal*/ true, UniversePolicy::Authoritative, &progress)); + store->renewWatermarkOnce(); + + EXPECT_TRUE(backend->allFired()) << "the faulted HEAD must have run on a pool worker"; + EXPECT_EQ(failedCounter() - failed_before, 1u); + EXPECT_EQ(progress.redeleted, kBlobs - 1); + expectBodiesPresentOnlyFor(*backend, *store, {kFaultedBlob}); + + const RoundReport next = runRegularRoundReclaiming(gc); + EXPECT_EQ(next.redeleted, kBlobs); + EXPECT_EQ(next.deleted, 1u); + EXPECT_EQ(next.absent, kBlobs - 1); + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); +} + +TEST(CASGCRedeleteConcurrency, SchedulingFailureAttemptsNothingAndTheNextRoundReclaims) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kFaultedBlob)); + + SCOPE_EXIT({ CannotAllocateThreadFaultInjector::setFaultProbability(0); }); + auto injected = std::make_shared>(false); + gc.setPhaseSink([injected](const GcPhaseRecord & rec) + { + if (rec.phase == "fold_seal_write") + { + CannotAllocateThreadFaultInjector::setFaultProbability(1.0); + injected->store(true); + } + else if (rec.phase == "pending_deletes") + { + CannotAllocateThreadFaultInjector::setFaultProbability(0); + } + }); + + const auto failed_before = failedCounter(); + RoundReport progress; + int code = 0; + try + { + gc.runRegularRound({}, /*allow_steal*/ true, UniversePolicy::Authoritative, &progress); + } + catch (const DB::Exception & e) + { + code = e.code(); + } + gc.setPhaseSink({}); + CannotAllocateThreadFaultInjector::setFaultProbability(0); + store->renewWatermarkOnce(); + + ASSERT_TRUE(injected->load()) << "the fault was never armed before pending_deletes"; + EXPECT_EQ(code, DB::ErrorCodes::CANNOT_SCHEDULE_TASK); + EXPECT_EQ(progress.redeleted, 0u); + EXPECT_EQ(failedCounter() - failed_before, 0u) << "an entry that was never submitted is not a failed delete"; + expectBodiesPresentOnlyFor(*backend, *store, {1, 2, 3, 4, 5, 6}); + + const RoundReport next = runRegularRoundReclaiming(gc); + EXPECT_EQ(next.redeleted, kBlobs); + EXPECT_EQ(next.deleted, kBlobs); + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); +} + +TEST(CASGCRedeleteConcurrency, ShardedPoolDeletesEveryShard) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4, kShards); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + + std::set expected_shards; + for (uint64_t b = 1; b <= kBlobs; ++b) + expected_shards.insert(blobShard(blobRefOf(b), kShards)); + ASSERT_GT(expected_shards.size(), 1u) << "the blobs must spread over several shards, or sharding is untested"; + + std::map> logs; + size_t deleted = 0; + for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) + { + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + deleted += rep.deleted; + collectOutcomeLogs(*backend, logs); + } + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); + EXPECT_EQ(deleted, kBlobs); + + std::set shards_with_rows; + size_t deleted_rows = 0; + for (const auto & [key, rows] : logs) + { + const size_t slash = key.rfind('/'); + const uint64_t log_shard = std::stoull(key.substr(slash + 1, key.find('.', slash) - slash - 1)); + for (const auto & [ref, outcome] : rows) + { + if (outcome != OutcomeKind::Deleted) + continue; + EXPECT_EQ(blobShard(ref, kShards), log_shard) << key; + shards_with_rows.insert(log_shard); + ++deleted_rows; + } + } + EXPECT_EQ(deleted_rows, kBlobs); + EXPECT_EQ(shards_with_rows, expected_shards); +} + +TEST(CASGCRedeleteConcurrency, ApplyPhaseFaultLeavesEveryEntryForTheNextRound) +{ + auto state = std::make_shared(); + auto backend = std::make_shared(); + PoolConfig config{.pool_prefix = "p", .server_root_id = "test"}; + config.gc_io_concurrency = 4; + config.gc_redelete_apply_hook_for_test = [state](const BlobRef & ref) + { + if (!state->armed.load()) + return; + if (ref == state->target) + { + state->armed.store(false); + throw std::runtime_error("injected apply fault"); + } + ++state->applied; + }; + auto store = Pool::open(backend, std::move(config)); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kFaultedBlob)); + + state->target = blobRefOf(kFaultedBlob); + state->armed.store(true); + const auto failed_before = failedCounter(); + RoundReport progress; + String message; + try + { + gc.runRegularRound({}, /*allow_steal*/ true, UniversePolicy::Authoritative, &progress); + } + catch (const std::exception & e) + { + message = e.what(); + } + store->renewWatermarkOnce(); + + EXPECT_NE(message.find("injected apply fault"), String::npos) << message; + EXPECT_EQ(progress.redeleted, state->applied); + EXPECT_EQ(failedCounter() - failed_before, 0u) << "a failed apply is not a failed delete"; + EXPECT_TRUE(allBlobsAbsent(*backend, *store)) << "every delete finished before the apply walk started"; + + const RoundReport next = runRegularRoundReclaiming(gc); + EXPECT_EQ(next.redeleted, kBlobs); + EXPECT_EQ(next.absent, kBlobs); + EXPECT_EQ(next.deleted, 0u); + for (uint64_t b = 1; b <= kBlobs; ++b) + { + OperationForTest op(*backend); + EXPECT_FALSE((*op).head(store->layout().blobMetaKey(blobRefOf(b)), Retry::standard()).has_value()) << "meta of blob " << b; + } +} + +namespace +{ + +struct BlobDeleteEvents +{ + std::mutex mutex; + std::map outcome_by_key; + size_t count = 0; + + CasEventSink sink() + { + return [this](CasEvent event) + { + if (event.type != CasEventType::BlobDelete) + return; + std::lock_guard lock(mutex); + outcome_by_key[event.detail["key"]] = event.outcome; + ++count; + }; + } + + std::pair, size_t> take() + { + std::lock_guard lock(mutex); + auto result = std::make_pair(std::move(outcome_by_key), count); + outcome_by_key.clear(); + count = 0; + return result; + } +}; + +} + +TEST(CASGCRedeleteConcurrency, BlobDeleteEventsMatchAppliedEntriesInAFailedRound) +{ + auto events = std::make_shared(); + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4, 1, 0, events->sink()); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + ASSERT_NO_FATAL_FAILURE(driveUntilAllGraduated(*backend, gc, store, kFaultedBlob)); + ASSERT_EQ(events->take().second, 0u) << "no blob delete may be reported before the redelete round"; + + backend->armRemoveFaults({blobKeyOf(*store, kFaultedBlob)}); + EXPECT_ANY_THROW(runRegularRoundReclaiming(gc)); + store->renewWatermarkOnce(); + ASSERT_TRUE(backend->allFired()); + + auto [failed_round, failed_round_count] = events->take(); + EXPECT_EQ(failed_round_count, kBlobs - 1); + EXPECT_FALSE(failed_round.contains(blobKeyOf(*store, kFaultedBlob))) << "a failed delete must not be reported"; + for (uint64_t b = 1; b <= kBlobs; ++b) + if (b != kFaultedBlob) + EXPECT_EQ(failed_round[blobKeyOf(*store, b)], "deleted") << "blob " << b; + + runRegularRoundReclaiming(gc); + auto [next_round, next_round_count] = events->take(); + EXPECT_EQ(next_round_count, kBlobs); + for (uint64_t b = 1; b <= kBlobs; ++b) + EXPECT_EQ(next_round[blobKeyOf(*store, b)], b == kFaultedBlob ? "deleted" : "absent") << "blob " << b; +} + +TEST(CASGCRedeleteConcurrency, OutcomeBudgetCapsAuditRowsButNotDeletes) +{ + constexpr uint64_t budget = 3; + RedeleteRun one; + RedeleteRun four; + ASSERT_NO_FATAL_FAILURE(runRedeleteScenario(1, one, budget)); + ASSERT_NO_FATAL_FAILURE(runRedeleteScenario(4, four, budget)); + + EXPECT_EQ(one.reports, four.reports); + + std::vector> logs_one; + for (const auto & [key, rows] : one.outcome_logs) + logs_one.push_back(rows); + std::vector> logs_four; + for (const auto & [key, rows] : four.outcome_logs) + logs_four.push_back(rows); + EXPECT_EQ(logs_one, logs_four) << "the capped rows must be the same entries in the same order at any concurrency"; + + size_t rows_total = 0; + for (const auto & rows : logs_four) + rows_total += rows.size(); + EXPECT_EQ(rows_total, budget); + + uint64_t redeleted_total = 0; + uint64_t deleted_total = 0; + for (const auto & report : four.reports) + { + redeleted_total += report[0]; + deleted_total += report[1]; + } + EXPECT_EQ(redeleted_total, kBlobs) << "the budget caps audit rows, never deletes"; + EXPECT_EQ(deleted_total, budget) << "`deleted` is tallied from the durable audit rows"; +} + +TEST(CASGCRedeleteConcurrency, RoundsWithoutRedeletesWriteNoOutcomeLog) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4, kShards); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + + std::map> logs; + collectOutcomeLogs(*backend, logs); + bool graduated = false; + for (int i = 0; i < 8 && !graduated; ++i) + { + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + ASSERT_EQ(rep.redeleted, 0u); + ASSERT_EQ(rep.spared, 0u); + graduated = rep.graduated == kBlobs; + collectOutcomeLogs(*backend, logs); + } + ASSERT_TRUE(graduated) << "the scenario must reach graduation, or the rounds under test did nothing"; + + for (const auto & [key, rows] : logs) + ADD_FAILURE() << "a round without redeletes or spares wrote the outcome log " << key << " with " << rows.size() << " row(s)"; + EXPECT_TRUE(logs.empty()); +} From 735691a1cb8fc40f2ed5e9abf269f6b791424a95 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 15 Sep 2026 13:59:05 +0200 Subject: [PATCH 7/7] add context and tests Signed-off-by: Konstantin Morozov --- .../ContentAddressed/Gc/CasGc.cpp | 56 +++++++------------ .../ContentAddressed/Gc/CasGc.h | 39 +++++-------- 2 files changed, 36 insertions(+), 59 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index 54058a0bb358..dbb6e0215988 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -417,14 +417,7 @@ Gc::RedeleteIo Gc::performRedeleteIo(const RetiredEntry & entry, const Layout & return io; } -void Gc::applyRedeleteOutcome( - const RetiredEntry & entry, - const RedeleteIo & io, - uint64_t new_round, - uint64_t snap_generation, - GcRoundWorkBudget & round_work_budget, - RoundReport & report, - OutcomeLog & outcome_log) +void Gc::applyRedeleteOutcome(RedeleteRoundContext & ctx, const RetiredEntry & entry, const RedeleteIo & io) { if (const auto & hook = store->poolConfig().gc_redelete_apply_hook_for_test) hook(entry.ref); @@ -442,8 +435,8 @@ void Gc::applyRedeleteOutcome( e.object_kind = CasEventObjectKind::Blob; e.object_hash = blobIdOf(entry.ref); e.token = renderIncarnation(entry.token); - e.round = new_round; - e.gen = snap_generation; + e.round = ctx.new_round; + e.gen = ctx.snap_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}}; @@ -451,12 +444,12 @@ void Gc::applyRedeleteOutcome( /// The audit row is observability only -- the delete in `performRedeleteIo` 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()) + if (ctx.round_work_budget.outcomeEntryAvailable()) { - outcome_log.entries.push_back(std::move(outcome)); - ++round_work_budget.outcome_entries_used; + ctx.outcomes[ctx.shard].entries.push_back(std::move(outcome)); + ++ctx.round_work_budget.outcome_entries_used; } - ++report.redeleted; + ++ctx.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 published a fresh incarnation at this hash, and that writer's own @@ -471,15 +464,7 @@ void Gc::applyRedeleteOutcome( meta_writer->forgetCondemnMarker(entry.ref, entry.token); } -void Gc::redeleteBlob( - const RetiredEntry & entry, - const Layout & layout, - CasOperation & op, - uint64_t new_round, - uint64_t snap_generation, - GcRoundWorkBudget & round_work_budget, - RoundReport & report, - OutcomeLog & outcome_log) +void Gc::redeleteBlob(RedeleteRoundContext & ctx, const RetiredEntry & entry, const Layout & layout, CasOperation & op) { RedeleteIo io; try @@ -495,25 +480,21 @@ void Gc::redeleteBlob( blobIdOf(entry.ref), layout.blobKey(entry.ref), entry.condemn_round, getCurrentExceptionMessage(false)); throw; } - applyRedeleteOutcome(entry, io, new_round, snap_generation, round_work_budget, report, outcome_log); + applyRedeleteOutcome(ctx, entry, io); } void Gc::redeleteBlobs( + RedeleteRoundContext & ctx, const std::vector & entries, ThreadPool & pool, size_t concurrency, const Layout & layout, - CasOperation & op, - uint64_t new_round, - uint64_t snap_generation, - GcRoundWorkBudget & round_work_budget, - RoundReport & report, - OutcomeLog & outcome_log) + CasOperation & op) { if (concurrency <= 1 || entries.size() <= 1) { for (const RetiredEntry & entry : entries) - redeleteBlob(entry, layout, op, new_round, snap_generation, round_work_budget, report, outcome_log); + redeleteBlob(ctx, entry, layout, op); return; } @@ -589,7 +570,7 @@ void Gc::redeleteBlobs( first_error = errors[i]; continue; } - applyRedeleteOutcome(entries[i], io_results[i], new_round, snap_generation, round_work_budget, report, outcome_log); + applyRedeleteOutcome(ctx, entries[i], io_results[i]); } if (first_error) @@ -915,9 +896,14 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al static const std::vector kNothingToDelete; const std::vector & redelete_now = suppress_destructive ? kNothingToDelete : merge.redelete; - redeleteBlobs( - redelete_now, *io_pool, store->poolConfig().gc_io_concurrency, layout, op, new_round, generation, - round_work_budget, report, outcomes[shard]); + RedeleteRoundContext redelete_ctx{ + .new_round = new_round, + .snap_generation = generation, + .shard = shard, + .round_work_budget = round_work_budget, + .report = report, + .outcomes = outcomes}; + redeleteBlobs(redelete_ctx, redelete_now, *io_pool, store->poolConfig().gc_io_concurrency, layout, op); for (const RetiredEntry & entry : merge.spared) { /// A fresh dedup-adopt raced the condemn (see the matching CasGcFold Debug log emitted diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h index 286332bdc61d..5b393ae98e95 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h @@ -732,38 +732,29 @@ class Gc Removal del = Removal::Gone; }; + struct RedeleteRoundContext + { + uint64_t new_round; + uint64_t snap_generation; + uint64_t shard; + GcRoundWorkBudget & round_work_budget; + RoundReport & report; + std::map & outcomes; + }; + static RedeleteIo performRedeleteIo(const RetiredEntry & entry, const Layout & layout, CasOperation & op); - void applyRedeleteOutcome( - const RetiredEntry & entry, - const RedeleteIo & io, - uint64_t new_round, - uint64_t snap_generation, - GcRoundWorkBudget & round_work_budget, - RoundReport & report, - OutcomeLog & outcome_log); - - void redeleteBlob( - const RetiredEntry & entry, - const Layout & layout, - CasOperation & op, - uint64_t new_round, - uint64_t snap_generation, - GcRoundWorkBudget & round_work_budget, - RoundReport & report, - OutcomeLog & outcome_log); + void applyRedeleteOutcome(RedeleteRoundContext & ctx, const RetiredEntry & entry, const RedeleteIo & io); + + void redeleteBlob(RedeleteRoundContext & ctx, const RetiredEntry & entry, const Layout & layout, CasOperation & op); void redeleteBlobs( + RedeleteRoundContext & ctx, const std::vector & entries, ThreadPool & pool, size_t concurrency, const Layout & layout, - CasOperation & op, - uint64_t new_round, - uint64_t snap_generation, - GcRoundWorkBudget & round_work_budget, - RoundReport & report, - OutcomeLog & outcome_log); + CasOperation & op); /// The round's `_ckpt.checkpoint` witness per namespace — the SECOND, hint-independent witness the /// walk decides its absents against. ONE call site, in the fold, right where the hint is grouped.