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(