diff --git a/docs/en/antalya/cas/architecture/garbage-collection.md b/docs/en/antalya/cas/architecture/garbage-collection.md index a604104c7796..4590b6b4a8cc 100644 --- a/docs/en/antalya/cas/architecture/garbage-collection.md +++ b/docs/en/antalya/cas/architecture/garbage-collection.md @@ -226,7 +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_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 bf0fe17f7dd3..2e282d38e54b 100644 --- a/docs/en/antalya/cas/configuration.md +++ b/docs/en/antalya/cas/configuration.md @@ -106,7 +106,7 @@ entirely before release. Treat this table as a snapshot of the current build, no | `cas_part_folder_cache_max_entry_bytes` | 16 MiB | Oversized part-folder views bypass retention above this size | | `cas_manifest_decode_cache_bytes` | 128 MiB | Manifest decode cache byte budget (`0` disables) | | `cas_gc_meta_pool_size` | `16` | Bounded pool size for GC per-hash freshness-meta writes | -| `cas_gc_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_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/antalya/cas/operations/debugging.md b/docs/en/antalya/cas/operations/debugging.md index cb2e48f4f3b3..d7c2efa9f668 100644 --- a/docs/en/antalya/cas/operations/debugging.md +++ b/docs/en/antalya/cas/operations/debugging.md @@ -56,7 +56,7 @@ SELECT server_root_id, is_leader, state, last_success_age_seconds, pending_recla FROM system.cas_mounts WHERE disk = 'cas'; SELECT event_time, outcome, candidates_marked, entries_condemned, entries_graduated, - entries_redeleted, anomalies + entries_redeleted, entries_redelete_failed, anomalies FROM system.cas_gc_log WHERE event_type = 'Finish' AND disk_name = 'cas' ORDER BY event_time DESC LIMIT 10; @@ -193,8 +193,8 @@ SYSTEM CAS GC RUN cas; One row per disk it ran on: `disk`, `acquired_lease`, `deferred`, `round`, `candidates_marked`, `objects_deleted`, `objects_absent`, `objects_replaced`, `objects_spared`, `manifests_deleted`, -`entries_condemned`, `entries_graduated`, `entries_redeleted`, `fence_outs`, `anomalies`, -`pending_candidates`, `pending_condemned`, `pending_retired`. Omitting +`entries_condemned`, `entries_graduated`, `entries_redeleted`, `entries_redelete_failed`, `fence_outs`, +`anomalies`, `pending_candidates`, `pending_condemned`, `pending_retired`. Omitting the disk name runs one round on every content-addressed disk on the node. A manual run executes regardless of `SYSTEM CAS GC STOP` — `STOP` pauses only the background scheduler. diff --git a/docs/en/operations/storing-data.md b/docs/en/operations/storing-data.md index a3503bc1eb2c..0a6b733a9cc2 100644 --- a/docs/en/operations/storing-data.md +++ b/docs/en/operations/storing-data.md @@ -550,9 +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_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/docs/en/operations/system-tables/cas_gc_log.md b/docs/en/operations/system-tables/cas_gc_log.md index e92c2a6e3bd6..c8a56168d53f 100644 --- a/docs/en/operations/system-tables/cas_gc_log.md +++ b/docs/en/operations/system-tables/cas_gc_log.md @@ -48,6 +48,7 @@ specified (it is enabled by default in the shipped `config.xml`). - `entries_condemned` ([UInt64](/sql-reference/data-types/int-uint)) — Retired entries newly condemned this round (retired-cursor pipeline stage 1). - `entries_graduated` ([UInt64](/sql-reference/data-types/int-uint)) — Retired entries newly floor-passed and republished `delete_pending` this round (pipeline stage 2; deleted the next round). - `entries_redeleted` ([UInt64](/sql-reference/data-types/int-uint)) — Pending exact-token blob deletes executed this round (pipeline stage 3). +- `entries_redelete_failed` ([UInt64](/sql-reference/data-types/int-uint)) — Pending blob deletes whose `HEAD` or exact-token `DELETE` failed this round. Each such entry stays `delete_pending` and is retried in the next round; a non-zero value fails the round. - `fence_outs` ([UInt64](/sql-reference/data-types/int-uint)) — Expired mounts fenced out by this round's heartbeat floor. - `anomalies` ([UInt64](/sql-reference/data-types/int-uint)) — Fold clamps surfaced (and survived) this round. A steady non-zero value warrants a look at the round log details. - `duration_ms` ([UInt64](/sql-reference/data-types/int-uint)) — The round wall-clock duration (on a `Finish` row). @@ -80,7 +81,7 @@ The phases, in execution order: | `fold_ref_intake` | Read and fold every new ref log and the manifest bodies its edges name. | one `GET` per new log, one `GET` per manifest edge | | `fold_reduce` | The per-shard in-degree merge: condemn, spare, graduate. | prior-run streaming `GET`s, one `HEAD` per zero-transition candidate, run `PUT`s | | `fold_seal_write` | Publish the new fold seal. | one `PUT` | -| `pending_deletes` | The single content-delete site: exact-token deletes of previously published `delete_pending` entries, plus the outcome logs. | one `DELETE` per entry, one outcome-log `PUT` per shard | +| `pending_deletes` | The single content-delete site: exact-token deletes of previously published `delete_pending` entries, plus the outcome logs. `phase_metrics` carries `jobs_scheduled` (entries sent to the GC I/O pool; `0` when the phase ran sequentially) and `jobs_failed`. | one `HEAD` and at most one `DELETE` per entry, one outcome-log `PUT` per shard with rows | | `meta_pool_wait` | Drain the round's per-hash freshness-meta writes. | none on this thread — see the caveat below | | `round_commit` | The generation-retention prune and the round's single `gc/state` compare-and-swap. | prune `LIST`s and deletes, one compare-and-swap | | `handoff_reclaim` | Wholesale-reclaim generations a moved run ref stranded below the retention cursor. | prefix `LIST`s and deletes | @@ -128,6 +129,10 @@ Two caveats when reading these rows: - Work scheduled onto the GC meta pool runs on other threads, so the `meta_pool_wait` row's `ProfileEvents` delta is **empty by construction**. Read its `phase_metrics` `jobs_scheduled` / `jobs_completed` next to its duration instead: they distinguish a deep queue from a slow endpoint. +- Requests issued by GC I/O pool workers — the `pending_deletes` fan-out and the fold read-ahead of + `fold_ref_intake` and `fold_reduce` — also run on other threads and are missing from those phase + rows' `ProfileEvents`. They still count in `system.events`. On `pending_deletes`, read + `phase_metrics` `jobs_scheduled` / `jobs_failed` next to `redeleted`. - Phase durations do not sum to the round's `duration_ms`. The round also performs untimed bookkeeping between phases, and the `Finish` row's `duration_ms` remains the authority on total round time. diff --git a/docs/en/sql-reference/statements/system.md b/docs/en/sql-reference/statements/system.md index 389f87ca49ed..55be36618346 100644 --- a/docs/en/sql-reference/statements/system.md +++ b/docs/en/sql-reference/statements/system.md @@ -479,7 +479,7 @@ When `disk_name` is given, the round runs on that content-addressed disk only; t Each round is recorded in [`system.cas_gc_log`](/operations/system-tables/cas_gc_log) as a `Start` and a `Finish` row (with `trigger = 'Manual'`). -The command returns one row per disk it ran on (multiple rows when `disk_name` is omitted), with columns `disk`, `acquired_lease`, `deferred`, `round`, `candidates_marked`, `objects_deleted`, `objects_absent`, `objects_replaced`, `objects_spared`, `manifests_deleted`, `entries_condemned`, `entries_graduated`, `entries_redeleted`, `fence_outs`, `anomalies`, `pending_candidates`, `pending_condemned`, and `pending_retired`, describing the outcome of that round. The `pending_*` columns are the retire pipeline's remaining backlog sizes read from the `gc/state` this round's own commit just published (not this round's own delta, unlike the columns before them) — `0` on a non-authoritative row (`acquired_lease = 0` or `deferred = 1`), same as every other counter. +The command returns one row per disk it ran on (multiple rows when `disk_name` is omitted), with columns `disk`, `acquired_lease`, `deferred`, `round`, `candidates_marked`, `objects_deleted`, `objects_absent`, `objects_replaced`, `objects_spared`, `manifests_deleted`, `entries_condemned`, `entries_graduated`, `entries_redeleted`, `entries_redelete_failed`, `fence_outs`, `anomalies`, `pending_candidates`, `pending_condemned`, and `pending_retired`, describing the outcome of that round. The `pending_*` columns are the retire pipeline's remaining backlog sizes read from the `gc/state` this round's own commit just published (not this round's own delta, unlike the columns before them) — `0` on a non-authoritative row (`acquired_lease = 0` or `deferred = 1`), same as every other counter. A manual run always executes, regardless of [`SYSTEM CAS GC STOP`](#system-cas-gc-stop-start): `STOP` pauses only the background scheduler on that disk. 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/ContentAddressedMetadataStorage.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp index 78666e0ad297..374ee4a64efa 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp @@ -85,7 +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_io_concurrency; extern const ContentAddressedSettingsUInt64 gc_bulk_delete_chunk_keys; extern const ContentAddressedSettingsUInt64 attempt_timeout_ms; extern const ContentAddressedSettingsUInt64 lease_safety_margin_ms; @@ -309,7 +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_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) @@ -552,6 +552,7 @@ Cas::GcRoundLogger ContentAddressedMetadataStorage::makeGcRoundLogger() const e.entries_condemned = r.entries_condemned; e.entries_graduated = r.entries_graduated; e.entries_redeleted = r.entries_redeleted; + e.entries_redelete_failed = r.entries_redelete_failed; e.fence_outs = r.fence_outs; e.anomalies = r.anomalies; e.duration_ms = r.duration_ms; @@ -800,7 +801,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_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 bffb608688cc..0e01c5738781 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h @@ -635,8 +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; + /// 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 b2a3651b2472..7331d2bad29a 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp @@ -79,7 +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_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) \ @@ -229,12 +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); + 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 336ab21fbefa..504a563d5080 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 @@ -44,6 +46,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; @@ -340,14 +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); + /*max_threads*/ io_concurrency, /*max_free_threads*/ io_concurrency, /*queue_size*/ 0, + /*shutdown_on_exception*/ false); } void Gc::runNamespaceJanitorPage( @@ -398,6 +403,186 @@ 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(RedeleteRoundContext & ctx, const RetiredEntry & entry, const RedeleteIo & io) +{ + 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; + 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) + { + e.type = CasEventType::BlobDelete; + e.object_kind = CasEventObjectKind::Blob; + e.object_hash = blobIdOf(entry.ref); + e.token = renderIncarnation(entry.token); + 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}}; + }); + /// 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 (ctx.round_work_budget.outcomeEntryAvailable()) + { + ctx.outcomes[ctx.shard].entries.push_back(std::move(outcome)); + ++ctx.round_work_budget.outcome_entries_used; + } + ++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 + /// 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); +} + +void Gc::redeleteBlob(RedeleteRoundContext & ctx, const RetiredEntry & entry, const Layout & layout, CasOperation & 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)); + ++ctx.report.redelete_failed; + throw; + } + applyRedeleteOutcome(ctx, entry, io); +} + +void Gc::redeleteBlobs( + RedeleteRoundContext & ctx, + const std::vector & entries, + ThreadPool & pool, + size_t concurrency, + const Layout & layout, + CasOperation & op) +{ + if (concurrency <= 1 || entries.size() <= 1) + { + for (const RetiredEntry & entry : entries) + redeleteBlob(ctx, entry, layout, op); + return; + } + + using RunnerTask = ThreadPoolCallbackRunnerLocal::Task; + std::vector io_results(entries.size()); + std::vector errors(entries.size()); + std::exception_ptr first_error; + size_t scheduled = 0; + { + 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 + { + 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] + { + 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(); + } + ctx.jobs_scheduled += handles.size(); + + ThreadPoolCallbackRunnerLocal::waitForAllToFinish(handles); + for (size_t i = 0; i < handles.size(); ++i) + { + try + { + handles[i]->future.get(); + } + catch (...) + { + errors[i] = std::current_exception(); + } + } + } + + /// Every delete of the batch has already run. If `applyRedeleteOutcome` throws, the batch stays deleted + /// but unrecorded: the round fails before its outcome logs and `gc/state` commit, every entry stays + /// `delete_pending`, and the next round records the deleted ones as `Absent`. + for (size_t i = 0; i < scheduled; ++i) + { + if (errors[i]) + { + ++ctx.report.redelete_failed; + if (!first_error) + first_error = errors[i]; + continue; + } + applyRedeleteOutcome(ctx, entries[i], io_results[i]); + } + + if (first_error) + std::rethrow_exception(first_error); +} + RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool allow_steal, UniversePolicy policy, RoundReport * progress) { @@ -696,6 +881,17 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al pending_deletes_timer.emplace(phase_sink, "pending_deletes"); const uint64_t redeleted_before = report.redeleted; const uint64_t graduated_before = report.graduated; + const uint64_t redelete_failed_before = report.redelete_failed; + uint64_t redelete_jobs_scheduled = 0; + const auto record_redelete_job_metrics = [&] + { + pending_deletes_timer->metric("jobs_scheduled", redelete_jobs_scheduled); + pending_deletes_timer->metric("jobs_failed", report.redelete_failed - redelete_failed_before); + }; + SCOPE_EXIT_SAFE({ + if (pending_deletes_timer) + record_redelete_job_metrics(); + }); std::map outcomes; for (uint64_t shard = 0; shard < folded.retired_merge.size(); ++shard) { @@ -717,62 +913,15 @@ 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); - } + RedeleteRoundContext redelete_ctx{ + .new_round = new_round, + .snap_generation = generation, + .shard = shard, + .round_work_budget = round_work_budget, + .report = report, + .outcomes = outcomes, + .jobs_scheduled = redelete_jobs_scheduled}; + 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 @@ -921,6 +1070,7 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al pending_deletes_timer->metric("replaced", report.replaced); pending_deletes_timer->metric("spared", report.spared); pending_deletes_timer->metric("outcome_logs_written", outcomes.size()); + record_redelete_job_metrics(); pending_deletes_timer.reset(); /// emits the `pending_deletes` row /// Wait for the round's whole batch of per-hash freshness-meta writes (condemned during the @@ -1681,8 +1831,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 @@ -2055,7 +2205,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; @@ -3267,7 +3417,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(), @@ -4052,7 +4202,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 bd0d65e2e573..7d8988836f1f 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h @@ -164,6 +164,7 @@ struct RoundReport size_t condemned = 0; /// entries newly condemned into the retired list this round size_t graduated = 0; /// entries newly floor-passed (published delete_pending) this round size_t redeleted = 0; /// pending deletes executed this round (exact-token blob deletes) + size_t redelete_failed = 0; size_t fence_outs = 0; /// expired mounts fenced-out by the round's heartbeat floor std::vector anomalies; /// fold clamps surfaced this round (never wedge the round) @@ -726,6 +727,37 @@ class Gc /// every destructive-work family the round touches — see `GcRoundWorkBudget`. GcRoundWorkBudget & work_budget); + struct RedeleteIo + { + String blob_key; + 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; + uint64_t & jobs_scheduled; + }; + + static RedeleteIo performRedeleteIo(const RetiredEntry & entry, const Layout & layout, CasOperation & op); + + 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); + /// 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. /// @@ -976,10 +1008,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; + /// 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/Gc/CasGcScheduler.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp index 610d13879779..6dcfe4e37d01 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.cpp @@ -235,6 +235,7 @@ Cas::RoundReport CasGcScheduler::runRoundLogged(Cas::Gc & round_gc, GcRoundLogRe fin.entries_condemned = r.condemned; fin.entries_graduated = r.graduated; fin.entries_redeleted = r.redeleted; + fin.entries_redelete_failed = r.redelete_failed; fin.fence_outs = r.fence_outs; fin.anomalies = r.anomalies.size(); }; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h index 80b50f2fc448..9bd0296e567b 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcScheduler.h @@ -55,6 +55,7 @@ struct GcRoundLogRecord UInt64 entries_condemned = 0; /// entries newly condemned into the retired list this round UInt64 entries_graduated = 0; /// entries newly floor-passed (published delete_pending) this round UInt64 entries_redeleted = 0; /// pending exact-token blob deletes executed this round + UInt64 entries_redelete_failed = 0; UInt64 fence_outs = 0; /// expired mounts fenced-out by the round's heartbeat floor UInt64 anomalies = 0; /// fold clamps surfaced (never wedging) this round UInt64 duration_ms = 0; diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h index cbde11caa87c..92cf9bbaf49c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h @@ -174,11 +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_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. @@ -210,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_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 new file mode 100644 index 000000000000..ab41d517a6ee --- /dev/null +++ b/src/Disks/tests/gtest_cas_gc_redelete_concurrency.cpp @@ -0,0 +1,985 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include "cas_test_helpers.h" + +namespace ProfileEvents +{ +extern const Event CASGCRetiredRedeleteFailed; +} + +namespace DB::ErrorCodes +{ +extern const int CANNOT_SCHEDULE_TASK; +} + +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; +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 WorkerFaultBackend : public InMemoryBackend +{ +public: + 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 + { + std::lock_guard lock(mutex); + return remove_keys.empty() && head_keys.empty(); + } + + RawRemoval remove(const String & key, const String & expected_value, TransportAccess & access) override + { + 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: + 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::set remove_keys; + std::set head_keys; +}; + +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 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(blobHash(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) + if (!blobAbsent(backend, store.layout(), blobHash(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(), 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(), blobHash(1))); + + 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, uint64_t outcome_entry_budget = 0) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, concurrency, 1, outcome_entry_budget); + 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)); +} + +struct PendingDeletesRows +{ + std::vector> rows; + + UInt64 total(const String & key) const + { + UInt64 sum = 0; + for (const auto & row : rows) + if (const auto it = row.find(key); it != row.end()) + sum += it->second; + return sum; + } + + bool lastRowHas(const String & key) const { return !rows.empty() && rows.back().contains(key); } +}; + +GcPhaseSink recordPendingDeletes(std::shared_ptr out, GcPhaseSink next = {}) +{ + return [out = std::move(out), next = std::move(next)](const GcPhaseRecord & rec) + { + if (rec.phase == "pending_deletes") + out->rows.push_back(rec.metrics); + if (next) + next(rec); + }; +} + +} + +TEST(CASGCRedeleteConcurrency, ParallelRedeleteReclaimsEveryBlob) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + auto rows = std::make_shared(); + gc.setPhaseSink(recordPendingDeletes(rows)); + + size_t redeleted = 0; + size_t deleted = 0; + size_t redelete_failed = 0; + for (int i = 0; i < 8 && !allBlobsAbsent(*backend, *store); ++i) + { + const RoundReport rep = runRegularRoundReclaiming(gc); + store->renewWatermarkOnce(); + redeleted += rep.redeleted; + deleted += rep.deleted; + redelete_failed += rep.redelete_failed; + } + + EXPECT_TRUE(allBlobsAbsent(*backend, *store)); + EXPECT_EQ(redeleted, kBlobs); + EXPECT_EQ(deleted, kBlobs); + EXPECT_EQ(redelete_failed, 0u); + EXPECT_EQ(rows->total("jobs_scheduled"), kBlobs); + EXPECT_EQ(rows->total("jobs_failed"), 0u); +} + +TEST(CASGCRedeleteConcurrency, BlobDeletesActuallyOverlap) +{ + 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); + auto rows = std::make_shared(); + gc.setPhaseSink(recordPendingDeletes(rows)); + + 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); + ASSERT_TRUE(rows->lastRowHas("jobs_scheduled")); + EXPECT_EQ(rows->total("jobs_scheduled"), 0u) << "concurrency 1 must not use the pool"; +} + +TEST(CASGCRedeleteConcurrency, WorkerRemoveFaultKeepsSiblingOutcomesAndPoolAlive) +{ + auto backend = std::make_shared(); + auto store = openPoolWithIoConcurrency(backend, 4); + Gc gc(store, kGc); + publishThenDrop(*backend, store, gc); + + 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) + { + 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(), blobHash(b)), b != kFaultedBlob) << "blob " << b; + } + store->renewWatermarkOnce(); + } + + 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)); +} + +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"; +} + +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)}); + auto rows = std::make_shared(); + gc.setPhaseSink(recordPendingDeletes(rows)); + 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); + EXPECT_EQ(progress.redelete_failed, 2u); + ASSERT_EQ(rows->rows.size(), 1u); + ASSERT_TRUE(rows->lastRowHas("jobs_failed")) << "the phase metrics must survive the round's exception"; + EXPECT_EQ(rows->total("jobs_scheduled"), kBlobs); + EXPECT_EQ(rows->total("jobs_failed"), 2u); + 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)}); + auto rows = std::make_shared(); + gc.setPhaseSink(recordPendingDeletes(rows)); + 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); + EXPECT_EQ(progress.redelete_failed, 1u); + ASSERT_EQ(rows->rows.size(), 1u); + EXPECT_EQ(rows->total("jobs_scheduled"), kBlobs); + EXPECT_EQ(rows->total("jobs_failed"), 1u); + 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); + auto rows = std::make_shared(); + gc.setPhaseSink(recordPendingDeletes(rows, [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(progress.redelete_failed, 0u); + EXPECT_EQ(failedCounter() - failed_before, 0u) << "an entry that was never submitted is not a failed delete"; + ASSERT_EQ(rows->rows.size(), 1u); + ASSERT_TRUE(rows->lastRowHas("jobs_scheduled")); + EXPECT_EQ(rows->total("jobs_scheduled"), 0u); + EXPECT_EQ(rows->total("jobs_failed"), 0u); + 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()); +} 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 53b3b50b4fbb..eb83a880bac9 100644 --- a/src/Disks/tests/gtest_cas_settings.cpp +++ b/src/Disks/tests/gtest_cas_settings.cpp @@ -176,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)"); } diff --git a/src/Interpreters/ContentAddressedGarbageCollectionLog.cpp b/src/Interpreters/ContentAddressedGarbageCollectionLog.cpp index 356a6eed523f..8babfe9f6d80 100644 --- a/src/Interpreters/ContentAddressedGarbageCollectionLog.cpp +++ b/src/Interpreters/ContentAddressedGarbageCollectionLog.cpp @@ -49,13 +49,14 @@ ColumnsDescription ContentAddressedGarbageCollectionLogElement::getColumnsDescri {"entries_condemned", std::make_shared(), "Retired entries newly condemned this round (retired-cursor pipeline stage 1)."}, {"entries_graduated", std::make_shared(), "Retired entries newly floor-passed and republished delete_pending this round (stage 2; deleted the NEXT round)."}, {"entries_redeleted", std::make_shared(), "Pending exact-token blob deletes executed this round (stage 3)."}, + {"entries_redelete_failed", std::make_shared(), "Pending blob deletes whose HEAD or exact-token DELETE failed this round. Each one stays delete_pending and is retried next round; a non-zero value fails the round."}, {"fence_outs", std::make_shared(), "Expired mounts fenced out by this round's heartbeat floor."}, {"anomalies", std::make_shared(), "Fold clamps surfaced (and survived) this round; steady >0 warrants a look at the round log details."}, {"duration_ms", std::make_shared(), "Round wall-clock duration (Finish)."}, {"error", std::make_shared(), "Exception text when outcome = Aborted, Stopped or Error. On a Stopped row it names the engine\'s refusal, not the teardown."}, {"error_code", std::make_shared(), "Exception code when outcome = Aborted, Stopped or Error; 0 otherwise. The structured twin of `error`: key monitoring on this column, not on message text."}, {"ProfileEvents", std::make_shared(lc_string, std::make_shared()), - "On a Start/Finish row: the per-round ProfileEvents delta (the Cas* counters and S3 events for this round). On a Phase row: THAT PHASE's delta, so `GROUP BY phase` over `ProfileEvents['S3ListObjects']` attributes the round's LIST budget to the phase that spent it. Empty on the `meta_pool_wait` row by construction — that phase's work runs on other threads (read its `phase_metrics` instead)."}, + "On a Start/Finish row: the per-round ProfileEvents delta (the Cas* counters and S3 events for this round). On a Phase row: THAT PHASE's delta, so `GROUP BY phase` over `ProfileEvents['S3ListObjects']` attributes the round's LIST budget to the phase that spent it. Empty on the `meta_pool_wait` row by construction — that phase's work runs on other threads (read its `phase_metrics` instead). Requests issued by GC I/O pool workers (the `pending_deletes` fan-out and the fold read-ahead) are likewise missing from their phase rows; they still count in `system.events`."}, {"round_id", std::make_shared(), "Correlator for every row of one round attempt (its Start, each Phase, and its Finish). Minted per attempt; unlike `round` it exists even for a round that never committed and for a round that never led. Group by this column to reconstruct one round."}, {"phase", lc_string, @@ -63,7 +64,7 @@ ColumnsDescription ContentAddressedGarbageCollectionLogElement::getColumnsDescri {"phase_duration_microseconds", std::make_shared(), "Wall-clock duration of this phase in microseconds (Phase rows only). Microseconds because several phases are routinely sub-millisecond and the point is to see when they are not. Phase durations do not sum to the round's `duration_ms`: the round also does untimed bookkeeping between phases."}, {"phase_metrics", std::make_shared(lc_string, std::make_shared()), - "Phase-specific semantic counts a phase computes for itself and no ProfileEvent can supply (Phase rows only) — for example `changed_shards` on defer_decision, `logs_accounted`/`logs_applied` on fold_ref_intake, `transactions_unapplied` on fold_reduce, `jobs_scheduled`/`jobs_completed` on meta_pool_wait. The verb counts ride the `ProfileEvents` column of the same row."}, + "Phase-specific semantic counts a phase computes for itself and no ProfileEvent can supply (Phase rows only) — for example `changed_shards` on defer_decision, `logs_accounted`/`logs_applied` on fold_ref_intake, `transactions_unapplied` on fold_reduce, `jobs_scheduled`/`jobs_completed` on meta_pool_wait, `jobs_scheduled`/`jobs_failed` on pending_deletes. The verb counts ride the `ProfileEvents` column of the same row."}, }; } @@ -90,6 +91,7 @@ void ContentAddressedGarbageCollectionLogElement::appendToBlock(MutableColumns & columns[i++]->insert(entries_condemned); columns[i++]->insert(entries_graduated); columns[i++]->insert(entries_redeleted); + columns[i++]->insert(entries_redelete_failed); columns[i++]->insert(fence_outs); columns[i++]->insert(anomalies); columns[i++]->insert(duration_ms); diff --git a/src/Interpreters/ContentAddressedGarbageCollectionLog.h b/src/Interpreters/ContentAddressedGarbageCollectionLog.h index 65c231b76eac..852a729ed3cf 100644 --- a/src/Interpreters/ContentAddressedGarbageCollectionLog.h +++ b/src/Interpreters/ContentAddressedGarbageCollectionLog.h @@ -45,6 +45,7 @@ struct ContentAddressedGarbageCollectionLogElement UInt64 entries_condemned = 0; /// retired-cursor pipeline: entries newly condemned this round UInt64 entries_graduated = 0; /// retired-cursor pipeline: entries newly round-passed (delete_pending) this round UInt64 entries_redeleted = 0; /// retired-cursor pipeline: pending exact-token blob deletes executed this round + UInt64 entries_redelete_failed = 0; UInt64 fence_outs = 0; /// expired mounts fenced-out by the round's heartbeat floor UInt64 anomalies = 0; /// fold clamps surfaced this round UInt64 duration_ms = 0; diff --git a/src/Interpreters/InterpreterSystemQuery.cpp b/src/Interpreters/InterpreterSystemQuery.cpp index 7ebc130fb53c..dc4295ff4993 100644 --- a/src/Interpreters/InterpreterSystemQuery.cpp +++ b/src/Interpreters/InterpreterSystemQuery.cpp @@ -2372,6 +2372,7 @@ ColumnsDescription contentAddressedGcRoundColumns() {"entries_condemned", std::make_shared()}, {"entries_graduated", std::make_shared()}, {"entries_redeleted", std::make_shared()}, + {"entries_redelete_failed", std::make_shared()}, {"fence_outs", std::make_shared()}, {"anomalies", std::make_shared()}, /// Task 7: the retire pipeline's REMAINING (not this-round-delta) sizes, read from the gc/state @@ -2399,6 +2400,7 @@ void appendContentAddressedGcRoundRow(MutableColumns & res_columns, const String res_columns[i++]->insert(rep.condemned); res_columns[i++]->insert(rep.graduated); res_columns[i++]->insert(rep.redeleted); + res_columns[i++]->insert(rep.redelete_failed); res_columns[i++]->insert(rep.fence_outs); res_columns[i++]->insert(rep.anomalies.size()); res_columns[i++]->insert(rep.pending_candidates);