diff --git a/docs/en/antalya/cas/architecture/blob-protocol.md b/docs/en/antalya/cas/architecture/blob-protocol.md index 7ff302cfe9bef..54f6399d073a7 100644 --- a/docs/en/antalya/cas/architecture/blob-protocol.md +++ b/docs/en/antalya/cas/architecture/blob-protocol.md @@ -143,15 +143,15 @@ sequenceDiagram end Note over GC: round n+1 -- graduation, only if still zero - GC->>S3: re-verify in-degree, requires confirmed durable Condemned evidence for hash+token t1 - Note over GC: publishes delete_pending + GC->>S3: re-verify in-degree, then GET .meta + Note over GC: Condemned at round n or newer -- publishes delete_pending Note over GC: round n+2 -- the single content-delete site GC->>S3: deleteExact(blob, t1) alt writer republished S3-->>GC: TokenMismatch -- nothing deleted, blob is live at t2 else genuinely dead - S3-->>GC: Deleted -- then drop the .meta + S3-->>GC: Deleted -- then drop the .meta only while it is still Condemned at round n or older end ``` @@ -170,9 +170,13 @@ Why this closes the race in both directions: `TokenMismatch` and reclaims nothing — the delete names an exact incarnation, never "the object at this key". - The delete lags condemnation by at least two full rounds, and publishing the one edge that - authorizes an irreversible delete requires confirmed durable `Condemned` evidence for that - exact `(hash, token)` pair. Without it `GC` never throws — it carries the entry and retries the - marker write on the next round. + authorizes an irreversible delete requires reading the blob's `.meta` and finding it `Condemned` + at the entry's round or newer. A `Clean`, absent, older-round or unreadable marker never + graduates the entry: `GC` does not throw, it carries the entry and rewrites the marker at the + current round. +- The delete job removes the `.meta` only while it is still `Condemned` at the entry's round or + older, and conditionally on the version it read. A marker rewritten `Clean` by a writer, or + condemned again at a later round, stays. Both directions degrade to a spurious re-upload or a no-op delete. Neither can lose data or leave a dangling manifest entry. diff --git a/docs/en/antalya/cas/architecture/correctness.md b/docs/en/antalya/cas/architecture/correctness.md index 3a1e076ae127e..8260ca02f9b47 100644 --- a/docs/en/antalya/cas/architecture/correctness.md +++ b/docs/en/antalya/cas/architecture/correctness.md @@ -35,7 +35,7 @@ The full model index (source `.tla` files and proof-run records) lives at | `CaGcAckFloorZombie.tla` | `INV_NO_DANGLE` under two fully-interleaved GC leaders | `sab_eagerdelete`: a leader deleting its own fresh (not-yet-pending) graduations — the pre-amendment single-phase behavior — dangles when a deposed leader's pass overlaps a live one | | `CaGcRoundDeferCore.tla` | `NoOverDelete` — a deferred round may skip a rebuild only when nothing destructive is pending | `sab_graduate_on_stale`: dropping the "an unfolded delta covers this blob" guard lets a deferred round delete a blob its own unread history still protects | | `CaEdgeBeforeObserve.tla` | The writer/`GC` publish order is safe to simplify | `sab_late_edge`: allowing adoption before the precommit closure is durable (the pre-fix order) dangles | -| `CaGcCondemnMarkerGate.tla` | Graduation requires confirmed durable `Condemned` evidence | Swallowing a failed asynchronous condemn-marker write let a writer adopt a token a later graduation was about to delete | +| `CaGcCondemnMarkerGate.tla` | Graduation reads the marker and requires `Condemned` at the entry's round or later | Swallowing a failed asynchronous condemn-marker write let a writer adopt a token a later graduation was about to delete | | `CaRefTableSnapshotLogCore.tla` | Dense, per-life ref ids with an in-band `_ckpt` recovery frontier | `sab_scanistruth`: trusting a listing as the source of truth for "acked" reproduces the real production incident where a `LIST` omitted two already-durable, already-acknowledged ref entries | ## Soak and chaos: the empirical oracle {#soak-and-chaos} diff --git a/docs/en/antalya/cas/architecture/garbage-collection.md b/docs/en/antalya/cas/architecture/garbage-collection.md index 21953307f9d15..46b34600296d5 100644 --- a/docs/en/antalya/cas/architecture/garbage-collection.md +++ b/docs/en/antalya/cas/architecture/garbage-collection.md @@ -340,7 +340,8 @@ Recomputes the per-shard in-degree snapshot and computes the round's single dest - **Runs on:** fold path only - **Reads:** streaming `GET` of each referenced parent run segment; one `HEAD` per zero-in-degree - candidate; one `.meta` `GET` per graduation candidate lacking in-process marker confirmation. When + candidate; one `.meta` `GET` per graduation candidate the round's graduation budget admits, read ahead + when `cas_gc_io_concurrency` is above 1. When orphan-sweep planning runs: a `LIST` page of `cas/manifests/`, a `GET` per candidate, plus `gc/state`, the adopted seal, the catalog, and per-namespace `_ckpt` / tail `_log`. - **Writes / deletes:** one `PUT` per rewritten run segment; schedules the async `.meta` condemn @@ -673,24 +674,31 @@ flowchart LR B --> C["retired with condemn_round = n"] C --> D{"round n+1: re-verify"} D -->|"in-degree recovered"| S["SPARED -- recovery wins, even past the floor"] - D -->|"still zero, confirmed durable Condemned evidence for hash and t"| G["GRADUATED -- delete_pending"] - D -->|"still zero, evidence unconfirmed"| C2["carried unchanged, retry the marker, never throw"] + D -->|"still zero, reads .meta: Condemned at round n or newer"| G["GRADUATED -- delete_pending"] + D -->|"still zero, .meta absent, Clean, older or unreadable"| C2["carried, .meta rewritten at this round, never throw"] D -->|"current token not equal to t"| SUP["SUPERSEDED -- a writer resurrected, re-condemn the CURRENT token"] G --> E["round n+2, pre-CAS: exact-token DELETE of blob at t"] - E -->|"Deleted or Absent"| F["then drop the .meta"] + E -->|"Deleted or Absent"| F["then drop the .meta if it is still Condemned at round n or older"] E -->|TokenMismatch| H["nothing deleted -- live at a newer token, leave the .meta alone"] ``` -The `.meta` sidecar carries **no token** — it is a per-hash hint. The exact incarnation token lives -in the condemned sentinel row inside the run, together with the condemn round and two flags, -`delete_pending` and `marker_confirmed`. `GC`'s marker is add-only: `Clean → Condemned` yes, the -reverse never, not even when sparing — only a writer that has already displaced the body may clear -it. A blob whose in-degree reaches zero in round `n` is retired with `condemn_round = n`; it can -graduate to `delete_pending` in round `n+1` at the earliest and be deleted in round `n+2`, so a -minimum of two full rounds separate condemnation from deletion. `delete_pending` is never cleared in -place, but it authorizes a delete only while in-degree stays zero: a fresh edge folded in a later -round spares the entry and removes it from the retired pipeline (recovery wins, even past the -floor). +The `.meta` sidecar carries **no token** — it is a per-hash hint. The exact incarnation token lives in the +condemned sentinel row inside the run, together with the condemn round and two flags, `delete_pending` and +`marker_confirmed`; `GC` writes the second with the first and never reads it. Every `GC` marker write stamps the +round that issues it and replaces an absent marker, `Clean`, or an older-round `Condemned`; `GC` never writes +`Clean`, not even when sparing — only a writer that has already displaced the body clears it. + +Graduation reads the marker in the round that graduates and accepts only `Condemned` at the entry's round or +newer: an older round may belong to another incarnation of the same content. The marker delete after a redelete +removes only `Condemned` at the entry's round or older, conditional on the version it read, so a delayed delete +never removes the marker of a later incarnation. It can leave an orphan `Condemned` marker over an absent body; +the next writer of that hash republishes and clears it, and `cas-fsck` counts it in `meta_without_body`. + +A blob whose in-degree reaches zero in round `n` is retired with `condemn_round = n`; it can graduate to +`delete_pending` in round `n+1` at the earliest and be deleted in round `n+2`, so a minimum of two full rounds +separate condemnation from deletion. `delete_pending` is never cleared in place, but it authorizes a delete only +while in-degree stays zero: a fresh edge folded in a later round spares the entry and removes it from the retired +pipeline (recovery wins, even past the floor). ## Sharding {#sharding} @@ -742,6 +750,8 @@ logs: | `GET` manifests | 1 per folded owner (manifest) edge — a manifest emits many blob edges but is read once per edge event; no manifest-body cache within a round | | `PUT` run segments | 1 per non-pure-carry shard, plus 1 fold seal | | `HEAD` blobs | 1 per newly condemned | +| `GET` blob `.meta` at graduation | 1 per entry the graduation budget admits, plus 1 `GET` and 1 `PUT` per refusal (the rewrite job re-reads the marker before its conditional `PUT`) | +| `GET` + conditional `DELETE` blob `.meta` after a redelete | 1 `GET` per `Deleted` or `Absent` outcome; the `DELETE` only while the marker is `Condemned` at the entry's round or older | | Blob `HEAD` + conditional `DELETE` | 1 `HEAD` per `redelete` entry — an entry that graduated in an *earlier* round, not the current one — up to `cas_gc_round_redelete_budget`; a `DELETE` only when the body is present at the condemned token | | Successful lease `CAS gc/state` | 1 | | Commit `CAS gc/state` | 1 | @@ -781,7 +791,7 @@ the chunk size reject `0`: | `cas_gc_round_sweep_recovery_op_budget` | 5000 | committed-tail ref-log reads the sweep's recovery walk may spend (phase 9) | | `cas_gc_bulk_delete_chunk_keys` | 1000 | keys per batch `DELETE` request for write-once families (phases 15, 17); `1` to `1000` | | `cas_gc_meta_pool_size` | 16 | bounded pool for condemn-marker writes (phase 12) | -| `cas_gc_io_concurrency` | 16 | bounded pool for the fold's read-ahead of checkpoints, ref logs, manifest bodies and zero-candidate `HEAD`s (phases 8, 9), the orphan-sweep planning reads (phase 9), the rebuild read-ahead and the `pending_deletes` `HEAD` + conditional `DELETE` fan-out (phase 11); other GC requests run on the round thread; `1` runs the covered requests sequentially. `cas_gc_read_concurrency` is rejected without an alias; use `cas_gc_io_concurrency` instead | +| `cas_gc_io_concurrency` | 16 | bounded pool for the fold's read-ahead of checkpoints, ref logs, manifest bodies and zero-candidate `HEAD`s (phases 8, 9), the graduation `.meta` reads (phase 9), the orphan-sweep planning reads (phase 9), the rebuild read-ahead and the `pending_deletes` `HEAD` + conditional `DELETE` fan-out (phase 11); other GC requests run on the round thread; `1` runs the covered requests sequentially. `cas_gc_read_concurrency` is rejected without an alias; use `cas_gc_io_concurrency` instead | The fold-batching controls `gc_fold_threshold` (default 1), `gc_fold_max_defer_rounds` (default 8) and `gc_frontier_probe_budget` (default unbounded) are internal `PoolConfig` fields with no disk @@ -877,7 +887,7 @@ No writes. |---|---|---:| | referenced parent run segments | streaming `GET` | one per referenced run | | `/blobs/...` | `HEAD` | one per zero-in-degree candidate, plus one peek per carried entry that reached zero again | -| blob `.meta` | `GET` | one per graduation candidate with no in-process marker confirmation | +| blob `.meta` | `GET` | one per graduation candidate the graduation budget admits | | new run segments | `PUT` | one per written run | | `/cas/manifests/` | `LIST` | one bounded page, only when orphan planning runs | | manifest candidate body | `GET` | one per nominated candidate (≤ `cas_manifest_sweep_delete_budget_keys`), through the read-ahead; keys decided from their name alone are never read; only when orphan planning runs | diff --git a/docs/en/antalya/cas/operations/debugging.md b/docs/en/antalya/cas/operations/debugging.md index d7c2efa9f6682..4e9d4b335b7f9 100644 --- a/docs/en/antalya/cas/operations/debugging.md +++ b/docs/en/antalya/cas/operations/debugging.md @@ -242,7 +242,7 @@ mount. | `cas-fsck [--detail] [--timeout N] [--namespace PREFIX] [--partial]` | The same reachability scan as `SYSTEM CAS FSCK`, offline. `--detail` adds a per-object `\t\t` listing (`reachable`, `dangling`, `unreachable`, `pending-gc`, `awaiting-gc`, `unaccounted`, `stale-edge`, `corrupted-run`, `chain-broken`, `unchecked`, `lifeless-key`, `janitor-pending`) — the only way to get per-object, not just per-pool, findings. `--timeout`/`--partial` bound a scan on a large pool | | `cas-gc-dryrun` | Previews the next round's deletes, read-only, no lease. Over-reports away from quiescence (does not fold new owner events) — a diagnostic only, never a delete source | | `cas-inspect ''` | Decodes one raw object-storage key (as printed by `cas-fsck`/`cas-gc-dryrun`) straight to JSON | -| `cas-gc-rebuild [--force]` | Disaster recovery: rebuilds a `gc/state` baseline from raw owner state after the GC guard has refused every regular round. `--force` bypasses only the healthy-state refusal, never a competing leader or a failed `CAS`. See [`SYSTEM CAS GC REBUILD`](/sql-reference/statements/system#system-cas-gc-rebuild) for the destructive-tool caveats | +| `cas-gc-rebuild [--force]` | Disaster recovery: rebuilds a `gc/state` baseline from raw owner state after the GC guard has refused every regular round. `--force` bypasses only the healthy-state refusal, never a competing leader or a failed `CAS`. See [`SYSTEM CAS GC REBUILD`](/sql-reference/statements/system#system-cas-gc-rebuild) for the destructive-tool caveats. Before rebuilding a lost `gc/state`, stop or restart every server of the pool; a request sent before a server stopped can still reach the object store afterwards, and this rule does not cover it. | ```bash clickhouse-disks -C config.xml --disk cas cas-fsck --detail @@ -255,6 +255,19 @@ clickhouse-disks -C config.xml --disk cas cas-gc-rebuild --force [migration page](/antalya/cas/operations/migration#decommission) alongside the SQL form, since decommissioning a pool member is a migration/scale-down operation, not an incident-time tool. +## Upgrading a pool with GC enabled {#upgrade-gc} + +`cas_gc_enabled` is read once, when the disk is constructed: only a restart applies a change. + +1. On every server still on the old version, set `false` in each CAS disk's + configuration and restart the server. It then runs no GC and refuses `SYSTEM CAS GC RUN`, `START` and `REBUILD`. +2. Only after step 1 has finished on every server, upgrade the servers, removing the setting on each upgraded one. +3. Never run `SYSTEM CAS GC` commands on, or re-enable GC on, a server still on the old version. + +A delete request sent before a restart can still reach the object store later and remove a marker GC relies on, +so a blob can be lost; nothing bounds when such requests stop. Entries the old version already scheduled for +deletion are deleted by the upgraded GC without a fresh marker read. Downgrades follow the same procedure. + ## The CLICKHOUSE_USER_FILES gotcha when reproducing a test manually {#user-files-gotcha} Running a `CAS` stateless test directly with `tests/clickhouse-test` against a manually started diff --git a/docs/en/antalya/cas/operations/monitoring.md b/docs/en/antalya/cas/operations/monitoring.md index 8bde03328575a..52ea010fb2f41 100644 --- a/docs/en/antalya/cas/operations/monitoring.md +++ b/docs/en/antalya/cas/operations/monitoring.md @@ -43,7 +43,7 @@ subset for a first health pass; the full list groups by object class (`CASBlob*` | `CASRefAppendSealRejected` | Occasional (a deposed writer losing a race is the protocol working); sustained growth is not | A writer keeps retrying after losing its mount and does not yet know it | | `CASGCHeartbeatFenceOuts` | Zero on a healthy pool | GC fenced an expired mount; check `system.cas_mounts` for a member that should have cleanly unmounted | | `CASGCUnmatchedRemoveDeltas` | Occasional (benign per-key no-op by design) | A persistent nonzero rate means removal deltas are reaching the reducer without their matching activation — a correctness signal worth a look, not an automatic false deletion | -| `CASGCCondemnMarkerUnconfirmedCarry` | Zero | A durable condemn marker could not be confirmed; deletion is safely postponed but investigate marker write/read failures | +| `CASGCCondemnMarkerUnconfirmedCarry` | Zero | The graduation read found no `Condemned` marker at the entry's round or later (absent, `Clean`, older round, or unreadable); deletion is postponed and the marker is rewritten, but investigate repeated carries | | `CASGCMetaWriteAnomaly` | Zero | The bounded GC metadata pool failed an operation; backend or pool pressure may delay metadata convergence | | `CASRefRollbackBestEffortDropFailed` | Zero | A rollback cleanup drop hit a backend failure; refs may remain live and GC may be delayed on that namespace | diff --git a/docs/en/sql-reference/statements/system.md b/docs/en/sql-reference/statements/system.md index 55be366183468..a6e3233fa5ae9 100644 --- a/docs/en/sql-reference/statements/system.md +++ b/docs/en/sql-reference/statements/system.md @@ -517,6 +517,9 @@ baseline — on a pool that has ever completed a GC round this means the object `adopted_seal_generation` names which generation's fold seal the rebuild carried holds from; `0` when it carried none. +Before rebuilding a lost `gc/state`, stop or restart every server of the pool. A request a server sent before it +stopped can still reach the object store afterwards; this rule does not cover it. + ### SYSTEM CAS GC STOP / SYSTEM CAS GC START {#system-cas-gc-stop-start} Pause or resume the background GC scheduler on one content-addressed disk, without affecting reads diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index a9986b5aa4858..05546f4867afc 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -912,7 +912,7 @@ The server successfully detected this situation and will download merged part fr 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(CASGCCondemnMarkerUnconfirmedCarry, "Number of CAS GC retirements delayed because the condemn marker did not confirm: it could not be read or written, it was a readable `Clean`, or it was `Condemned` at an older round than the entry. Deletion is postponed safely.", 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) \ M(CASGCMetaWriteAnomaly, "Number of CAS GC metadata operations that failed on the bounded metadata pool. A non-zero value indicates backend or pool pressure and may delay metadata convergence.", ValueType::Number) \ M(CASMetaPut, "Number of conditional CAS blob-metadata create attempts, including lost races. Growth indicates metadata reconciliation or write contention.", ValueType::Number) \ diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasRecordStreamFormat.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasRecordStreamFormat.h index 5741a61fc9b71..1707d8f2ad077 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasRecordStreamFormat.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Formats/CasRecordStreamFormat.h @@ -90,7 +90,7 @@ struct SourceEdgeRecord PersistedEtag token{}; uint64_t size = 0; uint64_t condemn_round = 0; - bool marker_confirmed = false; /// durable Condemned meta confirmed for this entry (graduation gate) + bool marker_confirmed = false; /// written with `delete_pending`; the graduation gate reads the marker, not this bit }; /// The header-line `kind` word for the only live `cas_run` kind. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasBlobInDegree.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasBlobInDegree.cpp index 3170b197968dd..f80e9f4976786 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasBlobInDegree.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasBlobInDegree.cpp @@ -17,6 +17,7 @@ namespace ProfileEvents #include #include #include +#include #include #include @@ -38,9 +39,10 @@ namespace const UInt128 kZeroSourceId{0}; -/// Streams a shard's prior source-edge run at O(one block) resident memory: chains the run SEGMENTS the -/// caller resolved from the parent seal (`blob_target_runs` filtered to one shard) and exposes a one-row -/// lookahead for the fold merge. The prior run carries +/// Streams a shard's prior source-edge run with one block resident, plus up to `kGraduationLookaheadRows` +/// decoded rows while a graduation scan is buffering: chains the run SEGMENTS the caller resolved from the +/// parent seal (`blob_target_runs` filtered to one shard) and exposes the next row to the fold merge. The +/// prior run carries /// BOTH surviving edges (`RunMarker::Edge`) AND the retired `RunMarker::Condemned` sentinel rows at the zero source id, /// so the cursor stops at edges AND at condemned rows (exposing the type via `rowType`), while zero-marker /// sentinels are dropped on carry (per-generation, never carried forward). Row/key invariants are enforced @@ -61,64 +63,35 @@ class PriorEdgeCursor } bool valid() const { return has_current; } - const String & key() const { return current_key; } + const String & key() const { return current.key; } /// The value byte of the current row: `RunMarker::Edge` (a surviving edge) or `RunMarker::Condemned` (a retired /// sentinel row). Zero markers are never surfaced (dropped on carry). - RunMarker rowType() const { return current_type; } + RunMarker rowType() const { return current.type; } /// The decoded retired sentinel for the current row (only valid when `rowType() == RunMarker::Condemned`). - const CondemnedRow & condemnedRow() const { return current_condemned; } + const CondemnedRow & condemnedRow() const { return current.condemned; } /// Advance to the next surviving edge OR retired sentinel, dropping zero markers, enforcing the /// row/key invariants, and crossing segment boundaries. void advance() { + if (!lookahead.empty()) + { + current = std::move(lookahead.front()); + lookahead.pop_front(); + current_offered = offered_ahead > 0; + if (offered_ahead > 0) + --offered_ahead; + has_current = true; + return; + } + current_offered = false; while (true) { - /// Pull rows from the open segment until a surviving edge, retired sentinel, or the segment ends. if (reader) { - String k; - String p; - while (reader->next(k, p)) + if (auto row = pullRow()) { - BlobRef bh; - UInt128 sid; - /// `parse` throws CORRUPTED_DATA on a malformed size / NOT_IMPLEMENTED on an - /// unknown algo byte (fail-closed). - SourceEdgeKeyCodec::parse(k, bh, sid); - if (p.empty()) - throw Exception(ErrorCodes::CORRUPTED_DATA, "CAS source-edge run: empty row payload"); - const RunMarker v = runMarkerFromByte(p[0], "CAS source-edge run"); - const bool sentinel_key = (sid == kZeroSourceId); - - if (sentinel_key) - { - /// A sentinel key carries exactly one row per blob and never an edge. - if (v == RunMarker::Edge) - throw Exception(ErrorCodes::CORRUPTED_DATA, - "CAS source-edge run: active edge at the reserved sentinel source_id 0"); - if (have_sentinel_blob && sentinel_blob == bh) - throw Exception(ErrorCodes::CORRUPTED_DATA, - "CAS source-edge run: duplicate sentinel row for one blob"); - have_sentinel_blob = true; - sentinel_blob = bh; - if (v == RunMarker::Zero) - continue; // A zero marker is per-generation and is dropped on carry. - /// A retired sentinel: decode and surface it (settled at close-out, not an edge). - current_condemned = decodeCondemnedRow(p); - current_key = k; - current_type = RunMarker::Condemned; - has_current = true; - return; - } - - /// A non-sentinel key must carry a surviving edge and nothing else. - if (v != RunMarker::Edge) - throw Exception(ErrorCodes::CORRUPTED_DATA, - "CAS source-edge run: sentinel row type 0x{:02x} at a non-sentinel key", - static_cast(v)); - current_key = k; - current_type = RunMarker::Edge; + current = std::move(*row); has_current = true; return; } @@ -139,19 +112,119 @@ class PriorEdgeCursor /// Typed open validates the NDJSON header before any row is consumed. Each row carries its /// own algorithm byte, so no separate width gate is needed. reader = openSourceEdgeRun(op, segments[seg_idx].key); + reader_drained = false; + } + } + + /// Offers the condemned rows from the current row onward, in order, each until `offer` accepts it: rows + /// of the OPEN segment are decoded into a buffer of at most `max_rows`. Stops when `offer` returns false + /// (that row is offered again next time), the buffer is full, or the open segment ends: opening the next segment, + /// and verifying the checksum of the one that ends, stay with `advance`. Rows are validated as `advance` + /// validates them, so a corrupt row throws here a few rows early. + void scanAhead(size_t max_rows, const std::function & offer) + { + if (has_current && !current_offered) + { + if (!offerRow(current, offer)) + return; + current_offered = true; + } + while (offered_ahead < lookahead.size()) + { + if (!offerRow(lookahead[offered_ahead], offer)) + return; + ++offered_ahead; + } + while (reader && lookahead.size() < max_rows) + { + auto row = pullRow(); + if (!row) + return; + lookahead.push_back(std::move(*row)); + if (!offerRow(lookahead.back(), offer)) + return; + ++offered_ahead; } } private: + struct Row + { + String key; + RunMarker type = RunMarker::Edge; + CondemnedRow condemned; + }; + + static bool offerRow(const Row & row, const std::function & offer) + { + if (row.type != RunMarker::Condemned) + return true; + BlobRef ref; + UInt128 source_id; + SourceEdgeKeyCodec::parse(row.key, ref, source_id); + return offer(ref, row.condemned); + } + + /// The next surviving edge or retired sentinel of the open segment, validated; nullopt once it is drained. + std::optional pullRow() + { + if (reader_drained) + return std::nullopt; + String k; + String p; + while (reader->next(k, p)) + { + BlobRef bh; + UInt128 sid; + /// `parse` throws CORRUPTED_DATA on a malformed size / NOT_IMPLEMENTED on an + /// unknown algo byte (fail-closed). + SourceEdgeKeyCodec::parse(k, bh, sid); + if (p.empty()) + throw Exception(ErrorCodes::CORRUPTED_DATA, "CAS source-edge run: empty row payload"); + const RunMarker v = runMarkerFromByte(p[0], "CAS source-edge run"); + const bool sentinel_key = (sid == kZeroSourceId); + + if (sentinel_key) + { + /// A sentinel key carries exactly one row per blob and never an edge. + if (v == RunMarker::Edge) + throw Exception(ErrorCodes::CORRUPTED_DATA, + "CAS source-edge run: active edge at the reserved sentinel source_id 0"); + if (have_sentinel_blob && sentinel_blob == bh) + throw Exception(ErrorCodes::CORRUPTED_DATA, + "CAS source-edge run: duplicate sentinel row for one blob"); + have_sentinel_blob = true; + sentinel_blob = bh; + if (v == RunMarker::Zero) + continue; // A zero marker is per-generation and is dropped on carry. + /// A retired sentinel: decode and surface it (settled at close-out, not an edge). + return Row{.key = k, .type = RunMarker::Condemned, .condemned = decodeCondemnedRow(p)}; + } + + /// A non-sentinel key must carry a surviving edge and nothing else. + if (v != RunMarker::Edge) + throw Exception(ErrorCodes::CORRUPTED_DATA, + "CAS source-edge run: sentinel row type 0x{:02x} at a non-sentinel key", + static_cast(v)); + return Row{.key = k, .type = RunMarker::Edge, .condemned = {}}; + } + reader_drained = true; + return std::nullopt; + } + CasOperation & op; const std::vector & segments; size_t seg_idx = 0; std::optional reader; - String current_key; - RunMarker current_type = RunMarker::Edge; - CondemnedRow current_condemned; + /// The open segment returned its last row; `advance` verifies it and opens the next. + bool reader_drained = false; + Row current; bool has_current = false; + bool current_offered = false; + /// Rows decoded ahead of `current`; the first `offered_ahead` of them were offered. + std::deque lookahead; + size_t offered_ahead = 0; /// Duplicate-sentinel guard: the last blob for which a sentinel row was seen (across skipped zero /// markers too). Rows are globally sorted, so two sentinels for one blob are adjacent. @@ -368,7 +441,8 @@ void foldDeltasIntoGeneration(CasOperation & op, const Layout & layout, bool suppress_destructive, std::vector * out_applied_by_txn_ordinal, std::vector source_retirements, - GcRoundWorkBudget * work_budget) + GcRoundWorkBudget * work_budget, + const GraduationReadHints * graduation_reads) { RetiredMergeResult sink; RetiredMergeResult & rmr = out_retired ? *out_retired : sink; @@ -484,20 +558,18 @@ void foldDeltasIntoGeneration(CasOperation & op, const Layout & layout, } else if (!suppress_destructive && e.condemn_round < current_round) { - /// Graduation gate (triage 2026-07-17 §3.4): publishing delete_pending is the one edge that - /// authorizes an irreversible delete, and it requires CONFIRMED durable Condemned evidence - /// for this exact (hash, token) — the marker is the writer's adopt gate, so an entry whose - /// marker write was swallowed could be same-token adopted invisibly to this fold's cut. - /// Unconfirmed => carry unchanged (fail-safe delay; the gate callback retries the marker so a - /// later pass can confirm). This gates a DELETE on missing evidence; it never throws. - if (e.marker_confirmed || !confirm_condemned_marker || confirm_condemned_marker(e)) + /// Budget before the gate: a carried entry must not cost a marker read every round. A refusal + /// consumes no slot, so refusals cannot crowd out graduations. + const bool admitted = graduation_reads ? graduation_reads->admit(e.ref) + : (!work_budget || work_budget->graduationAvailable()); + if (!admitted) + rmr.still_retired.push_back(e); + else { - /// Excess past the round's graduation budget carries the floor-passed entry unchanged - /// (still condemned, not yet delete_pending) — it re-evaluates the floor next round and - /// graduates then; nothing is lost, only delayed. - if (work_budget && !work_budget->graduationAvailable()) - rmr.still_retired.push_back(e); - else + /// After the admission: a look-ahead hint must never take the settling entry's slot. + if (graduation_reads) + cursor.scanAhead(kGraduationLookaheadRows, graduation_reads->offer); + if (!confirm_condemned_marker || confirm_condemned_marker(e)) { if (work_budget) ++work_budget->graduations_used; @@ -507,9 +579,9 @@ void foldDeltasIntoGeneration(CasOperation & op, const Layout & layout, rmr.graduated.push_back(pending); rmr.still_retired.push_back(std::move(pending)); } + else + rmr.still_retired.push_back(e); /// the gate refused: carried; it has rewritten the marker } - else - rmr.still_retired.push_back(e); /// no durable condemn-marker evidence yet — carried } else rmr.still_retired.push_back(e); /// carried unchanged until the floor passes it diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasBlobInDegree.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasBlobInDegree.h index db6f66b52f92b..049fd89713860 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasBlobInDegree.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasBlobInDegree.h @@ -36,12 +36,8 @@ struct RetiredEntry /// delete (pre-CAS, safe at any leader staleness) and drops the entry. /// Terminal: a pending entry is never un-pended (writers keep seeing /// it condemned and recreate). - bool marker_confirmed = false; /// Durable `Condemned` meta CONFIRMED for this entry — the - /// graduation gate (triage 2026-07-17 §3.4): the per-hash condemn - /// marker is the writer's adopt gate, so graduation to - /// delete_pending requires confirmed durable evidence; an - /// unconfirmed entry is CARRIED, never fail-open deleted. Set at - /// graduation (delete_pending rows always carry it). + bool marker_confirmed = false; /// Written with `delete_pending`, as released builds write it. The + /// graduation gate never reads it: it reads the marker itself. }; /// The fold's HEAD hooks (`head_blob`, `peek_head`): the caller issues the request on its own admitted @@ -82,7 +78,7 @@ struct CondemnedRow PersistedEtag token; // the incarnation the exact-token delete must re-observe uint64_t size = 0; uint64_t condemn_round = 0; - bool marker_confirmed = false; // durable Condemned meta confirmed (graduation gate) + bool marker_confirmed = false; // written with `delete_pending`; the graduation gate reads the marker, not this bit /// Spelled out rather than defaulted because `PersistedEtag` carries no equality of its /// own. A field added above belongs here too. bool operator==(const CondemnedRow & o) const @@ -102,6 +98,22 @@ String encodeCondemnedRow(const CondemnedRow & row); /// throw `CORRUPTED_DATA`. CondemnedRow decodeCondemnedRow(std::string_view payload); +/// Look-ahead for the graduation gate's marker read, supplied by the GC round. The gate still decides on the +/// value it takes; only the moment of the fetch moves. +struct GraduationReadHints +{ + /// Reserves a budget slot for the entry at the gate and hints its read if the window has room. False: no + /// budget left; the entry is carried unread. Idempotent for an entry that already holds a slot. + std::function admit; + /// Reserves and hints a later condemned row, or skips a row that will not reach the gate. False: the window + /// or the budget is full, or the row's blob was touched this round and must be decided inline; the scan + /// stops and offers the row again later. + std::function offer; +}; + +/// Rows the merge decodes ahead of its position to find graduation candidates. Bounds memory, not requests. +inline constexpr size_t kGraduationLookaheadRows = 16384; + /// Bridges the backend-free `Formats/CasRecordStreamFormat` NDJSON reader to the /// `(key, payload)` BYTE interface the fold / `zeroInDegree` / `previewDeletes` / `fsck` consumers use: /// `next` reconstructs the packed `SourceEdgeKeyCodec` key and the original payload bytes (a single @@ -353,10 +365,9 @@ struct GcRoundWorkBudget /// fail-closed abort; /// d > 0 -> spared (recovery wins even past the floor); /// d = 0 and condemn_round < current_round -> graduated: REPUBLISHED as delete_pending (two-phase -/// graduation) — deleted the NEXT pass. GATED on a -/// confirmed durable condemn marker (see -/// `confirm_condemned_marker` below): an unconfirmed -/// entry is carried unchanged instead; +/// graduation) — deleted the NEXT pass. Only while the +/// round's graduation budget has a slot, and only if +/// `confirm_condemned_marker` accepts; otherwise carried; /// d = 0 otherwise -> still_retired, carried byte-unchanged. /// A carried `RunMarker::Condemned` row is SETTLEMENT-ONLY: it never sets the blob's `cur_touched` bit, so a /// generation that only carries the row emits no zero-marker and pays no `peek_head` HEAD. The surviving @@ -386,15 +397,16 @@ struct GcRoundWorkBudget /// `peek_head` is a plain HEAD with no events and no counters. No supersede detection happens if /// `peek_head` is unset (default `{}`), independent of whether `head_blob` is set. /// -/// `confirm_condemned_marker` is the GRADUATION GATE (triage 2026-07-17 §3.4): graduation to -/// `delete_pending` is the one edge that authorizes an irreversible delete, and the per-hash condemn -/// marker (`writeCondemnedMeta`) is the writer's adopt gate — an entry whose marker write was silently -/// swallowed can be same-token adopted by a writer invisible to this fold's cut, so it must NOT -/// graduate. The callback returns whether durable `Condemned` evidence is confirmed for the entry's -/// exact (hash, token); on false the entry is carried unchanged (fail-safe delay — the caller is -/// expected to retry the marker so a later pass can confirm). An entry whose `marker_confirmed` bit is -/// already set skips the callback. Unset (default `{}`) means UNGATED — the pre-gate merge semantics, -/// for merge-mechanics unit tests only; the real GC round always passes the gate. +/// `confirm_condemned_marker` is the GRADUATION GATE: graduation to `delete_pending` is the one edge that +/// authorizes an irreversible delete, and the per-hash marker is the writer's adopt gate. It is called only +/// for an entry the graduation budget admits, and returns whether the entry may graduate; on false the entry +/// is carried unchanged. Unset (default `{}`) means UNGATED — merge-mechanics unit tests only; the real GC +/// round always passes the gate. +/// +/// `graduation_reads`, when set, admits each graduation candidate against the round's budget before the gate and +/// offers later condemned rows of the open run segment so their marker reads can start early. Null reads every +/// admitted marker at the gate. +/// /// The merge comparator is exactly `(ref.algo, ref.digest, source_id)` (that is, `BlobRef::operator<` /// followed by `source_id`), which is also the raw key order produced by `SourceEdgeKeyCodec`. void foldDeltasIntoGeneration(CasOperation & op, const Layout & layout, @@ -418,7 +430,8 @@ void foldDeltasIntoGeneration(CasOperation & op, const Layout & layout, std::vector source_retirements = {}, /// Shared by reference across every shard's call within one round; /// `nullptr` (default) is unbounded, matching every existing caller. - GcRoundWorkBudget * work_budget = nullptr); + GcRoundWorkBudget * work_budget = nullptr, + const GraduationReadHints * graduation_reads = nullptr); /// Stream the sealed in-degree runs named by `runs` (the current seal's `blob_target_runs` filtered to one /// shard) and return every blob written at in-degree 0 (the candidates that transitioned to zero). An diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index f871182c8edb4..272540db3b1cc 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -463,10 +463,8 @@ void Gc::applyRedeleteOutcome(RedeleteRoundContext & ctx, const RetiredEntry & e /// 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); + meta_writer->scheduleConfirmedMetaDelete(entry.ref, entry.condemn_round); } - /// The entry left the pipeline — drop its in-process condemn-marker confirmation. - meta_writer->forgetCondemnMarker(entry.ref, entry.token); } void Gc::reportRedeleteFailure( @@ -969,12 +967,10 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al /// `PartWriteTxn::ensureBlobPresent`) — the SOLE /// `Condemned -> Clean` transition. Clearing here on a /// deposed leader that then lost its round CAS would strand a stray-`Clean` over a still-live - /// condemned token and lose the reuse to a stale exact-token redelete (INV_NO_LOSS); see + /// condemned token and lose the reuse to a stale exact-token redelete (INV_NO_LOSS). /// The next `putBlob` self-heals /// the marker: `PartWriteTxn::ensureBlobPresent` refuses adoption on `Condemned` and /// publishes the writer's source under a fresh envelope. - /// The entry left the pipeline — drop its in-process condemn-marker confirmation. - meta_writer->forgetCondemnMarker(entry.ref, entry.token); } for (const RetiredEntry & entry : merge.graduated) { @@ -1016,14 +1012,9 @@ RoundReport Gc::runRegularRound(std::function on_lease_acquired, bool al "incarnation; superseded the stale entry and re-condemned the current token"; e.detail = {{"superseded_token", renderIncarnation(replaced.old_token)}}; }); - /// The supersede is ALSO a blob entering the retired set fresh (a re-condemn of the - /// CURRENT token) — write the meta Condemned exactly like a fresh `head_blob` condemn would, - /// so a NEXT writer's point-read gate sees it (and the graduation gate gets its (hash, token) - /// confirmation on success). `peek_head` itself stays side-effect-free (it runs once per - /// closed candidate, not just on a real supersede — see its own comment). The SUPERSEDED - /// (stale) token's in-process confirmation is dropped — that entry left the pipeline. - meta_writer->scheduleCondemnMarkerWrite(entry.ref, entry.token, entry.condemn_round, entry.size); - meta_writer->forgetCondemnMarker(entry.ref, replaced.old_token); + /// The supersede is also a blob entering the retired set: write its marker like a fresh + /// `head_blob` condemn, so a writer's point-read sees it. `peek_head` stays side-effect-free. + meta_writer->scheduleCondemnMarkerWrite(entry.ref, entry.condemn_round, entry.size); } } @@ -1980,10 +1971,11 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, std::vector> head_candidates(state.gc_shards); size_t head_hint_shard = 0; size_t next_head_hint = 0; + /// Head hints count only heads, so outstanding graduation reads cannot starve them. const auto topUpHeadHints = [&] { const std::vector & shard_candidates = head_candidates[head_hint_shard]; - while (next_head_hint < shard_candidates.size() && reads.pending() < reads.window()) + while (next_head_hint < shard_candidates.size() && reads.pendingHeads() < reads.window()) reads.hintHead(layout.blobKey(shard_candidates[next_head_hint++])); }; @@ -2037,14 +2029,10 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, }); Meta adjusted = *observed; adjusted.size = retiredLogicalSize(ObjectKind::Blob, observed->size, store->poolMeta().blob_header_len); - /// This candidate unconditionally becomes a fresh `RetiredEntry` in `closeBlob` (the ONLY - /// caller of `head_blob`) whenever this lambda returns a value — so this is exactly the round's - /// side-effecting condemn site. Write the meta Condemned so the writer's point-read gate - /// sees it; a successful write records the in-process (hash, token) confirmation the graduation - /// gate consumes (`scheduleCondemnMarkerWrite` captures everything BY VALUE — never by reference - /// to `cur_blob`, which the fold's tight streaming loop mutates while the job is queued). - meta_writer->scheduleCondemnMarkerWrite(ref, PersistedEtag::capture(observed->etag), - condemn_round, adjusted.size); + /// This candidate always becomes a fresh `RetiredEntry` in `closeBlob` (the only caller of + /// `head_blob`), so this is the round's side-effecting condemn site. The job captures everything by + /// value: the fold mutates `cur_blob` while the job is queued. + meta_writer->scheduleCondemnMarkerWrite(ref, condemn_round, adjusted.size); return adjusted; }; @@ -2072,43 +2060,37 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, return hr; }; - /// Graduation gate (triage 2026-07-17 §3.4): the merge consults this before publishing an entry - /// delete_pending. Confirmation sources, in order: the in-process (hash, token) record left by a - /// successful `writeCondemnedMeta` completion, then ONE synchronous `loadMeta` re-check — a durable - /// `Condemned` meta observed NOW is sufficient evidence, because a writer that same-token adopted - /// must have observed a non-Condemned meta EARLIER, its edge (EDGE-BEFORE-OBSERVE) landed before the - /// meta turned Condemned, and the redelete only fires from a LATER fold whose cut postdates this - /// round — that fold sees the edge and spares. (`BlobMeta` carries no token, so the re-check is - /// per-hash by design; the two-phase pipeline + the exact-token delete carry the rest.) No durable - /// evidence => count the carry, RETRY the marker write (liveness: a swallowed write would otherwise - /// carry forever), and refuse — never throw (an unreadable meta is missing evidence, not a wedge). + /// Marker keys admitted to a graduation read and not yet taken: every outstanding hint and the entry at the + /// gate. With `graduations_used` it never exceeds the graduation budget. + std::unordered_set graduation_reserved; + + /// Graduation gate. Publishing `delete_pending` authorizes an irreversible delete, and the marker is the + /// writer's adopt gate, so the marker is read every time, after the round's cut. A writer that adopted saw + /// a non-`Condemned` marker before this read; its edge is then in the delete round's cut and spares the + /// entry. A refusal carries the entry and rewrites the marker; a failed read is missing evidence and + /// never wedges the round. const auto confirm_condemned_marker = [&](const RetiredEntry & entry) -> bool { - if (meta_writer->condemnMarkerConfirmedInProcess(entry.ref, entry.token)) - return true; + const String key = layout.blobMetaKey(entry.ref); + graduation_reserved.erase(key); try { - if (const auto lm = loadMeta(op, layout, entry.ref); lm && lm->meta.state == MetaState::Condemned) - { - meta_writer->noteCondemnMarkerDurable(entry.ref, entry.token); /// memoize for a round-CAS-abort replay + /// An older round may be another incarnation's marker, which that incarnation's delete job may remove. + if (const auto lm = decodeLoadedMeta(reads.takeRead(key)); + lm && lm->meta.state == MetaState::Condemned && lm->meta.condemn_round >= entry.condemn_round) return true; - } } catch (...) { ProfileEvents::increment(ProfileEvents::CASGCMetaWriteAnomaly); tryLogCurrentException(logger, - "CAS gc: condemn-marker re-check failed to read the meta (treated as missing evidence; " - "the entry is carried, never wedges the round)"); + "CAS gc: the graduation gate failed to read the condemn marker (the entry is carried; " + "never wedges the round)"); } ProfileEvents::increment(ProfileEvents::CASGCCondemnMarkerUnconfirmedCarry); - /// Accepted race: this retry writes `Condemned` per-hash, with no token check. If a writer - /// resurrected this exact hash under a FRESH token between the original swallowed write and this - /// retry, the retry stamps `Condemned` over that writer's live, uncondemned incarnation. This is - /// never destructive -- the eventual exact-token delete is a no-op against the fresh token - /// (it finds a different incarnation, or none) -- worst case the resurrecting writer's later - /// same-token adopter sees stale `Condemned` metadata and republishes once unnecessarily. - meta_writer->scheduleCondemnMarkerWrite(entry.ref, entry.token, entry.condemn_round, entry.size); + /// This round, not the entry's: a publisher's late `Clean If-Match` names the bytes it read, and on S3 + /// an entry-round rewrite recreates exactly those bytes. + meta_writer->scheduleCondemnMarkerWrite(entry.ref, condemn_round, entry.size); return false; }; @@ -3450,6 +3432,7 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, /// the sweep, and a retirement is a removal like any other. The round's cut is frozen well before /// this point -- intake has finished and `deltas` has taken its final form -- so no HEAD is issued /// before the round knows what it folded. + std::set touched_blobs; { std::map, bool> last_verdict_is_remove; for (const BlobDelta & delta : deltas) @@ -3460,12 +3443,60 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, std::set removed; std::set surviving_add; for (const auto & [edge, is_remove] : last_verdict_is_remove) + { + /// Only the graduation read-ahead reads the set; the sequential setting has none. + if (reads.window() > 0) + touched_blobs.insert(edge.first); (is_remove ? removed : surviving_add).insert(edge.first); + } for (const BlobRef & ref : removed) if (!surviving_add.contains(ref)) head_candidates[blobShard(ref, state.gc_shards)].push_back(ref); } + /// Graduation read-ahead. A carried row that no delta or retirement touches this round has no edge and is + /// not superseded, so it reaches the gate; its read is hinted once the budget admits it. Hints are issued in + /// the merge, after intake, and the fold writes no marker of a hinted hash before the gate takes the read. + const auto graduationBudgetFull = [&] + { + return work_budget.max_graduations != 0 + && work_budget.graduations_used + graduation_reserved.size() >= work_budget.max_graduations; + }; + const GraduationReadHints graduation_hints{ + .admit = [&](const BlobRef & ref) + { + const String key = layout.blobMetaKey(ref); + if (graduation_reserved.contains(key)) + return true; + if (graduationBudgetFull()) + return false; + if (graduation_reserved.size() < reads.window()) + reads.hintRead(key); + graduation_reserved.insert(key); + return true; + }, + .offer = [&](const BlobRef & ref, const CondemnedRow & row) + { + if (row.delete_pending || row.condemn_round >= current_round || suppress_destructive) + return true; + /// A touched row may still reach the gate; reserving past it would let later rows take its budget slot. + if (touched_blobs.contains(ref)) + return false; + if (graduationBudgetFull() || graduation_reserved.size() >= reads.window()) + return false; + if (const String key = layout.blobMetaKey(ref); graduation_reserved.insert(key).second) + reads.hintRead(key); + return true; + }}; + const GraduationReadHints * graduation_reads = reads.window() > 0 ? &graduation_hints : nullptr; + /// A reservation left after a shard's merge belongs to no gate. + const auto discardGraduationReads = [&] + { + for (const String & key : graduation_reserved) + reads.discardRead(key); + graduation_reserved.clear(); + }; + if (state.gc_shards == 1) { /// SINGLE-SHARD PATH (gc_shards == 1). Every blob routes to shard 0, so the entire delta stream @@ -3490,7 +3521,8 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, confirm_condemned_marker, result.retired_merge.data(), suppress_destructive, &ledger.applied, std::move(orphan_source_retirements), - &work_budget); + &work_budget, graduation_reads); + discardGraduationReads(); result.fold_seal.condemned_summary[0] = summarize(result.retired_merge[0].still_retired); } } @@ -3534,7 +3566,8 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, confirm_condemned_marker, &result.retired_merge[shard], suppress_destructive, &ledger.applied, std::move(retirement_buckets[shard]), - &work_budget); + &work_budget, graduation_reads); + discardGraduationReads(); for (RunRef & r : shard_runs) result.fold_seal.blob_target_runs.push_back(std::move(r)); result.fold_seal.condemned_summary[shard] = summarize(result.retired_merge[shard].still_retired); diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.cpp index cd36d866f7762..62f49d74df883 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.cpp @@ -26,67 +26,39 @@ namespace DB::Cas namespace { -/// The registry key for one condemned incarnation: the persisted pair rendered the way a live -/// `Etag` renders itself, so two dialects can never collide on a shared value. -String condemnMarkerKey(const PersistedEtag & token) -{ - return token.dialect + ":" + token.value; -} - -/// The per-hash freshness-meta operations GC schedules on the bounded pool are -/// best-effort/idempotent by design. The meta is only a point-read freshness marker for the writer/ -/// promote gate; the ledger retired-set + the exact-token body delete remain the actual safety -/// core. A lost CAS here is never a correctness problem, only a (rare, -/// self-healing) staleness window for the NEXT point-reader — with ONE exception: the CONDEMN marker -/// is load-bearing for the delete edge. The exact-token delete argument -/// below assumes the marker was durably written before the delete fires; a swallowed condemn-marker -/// write lets a writer observe absent/Clean meta and adopt the SAME token the graduated entry later -/// deletes (a dangling manifest). Graduation to `delete_pending` is therefore GATED on confirmed -/// durable Condemned evidence for the exact (hash, token) — recorded in-process when the scheduled -/// `writeCondemnedMeta` reports success, or re-established by a synchronous `loadMeta` re-check at -/// graduation time. An unconfirmed entry is CARRIED (fail-safe delay) and its marker write retried; -/// the delete itself and every other meta op stay async/advisory. +/// The per-hash marker operations GC schedules on the bounded pool. The marker is the writer's adopt gate: +/// a writer that reads `Condemned` republishes instead of adopting the incarnation GC will delete. The ledger +/// and the exact-token body delete stay the safety core; round rules keep a marker from standing for the +/// wrong incarnation: +/// - every GC write stamps the round of the attempt that issues it, and replaces an absent marker, `Clean` +/// or an older-round `Condemned`; +/// - graduation accepts only `Condemned` at the entry's round or newer, read in that round (`Gc::fold`); +/// - a delete job removes only `Condemned` at its entry's round or older, `If-Match` the etag it read. +/// While `gc/state` is readable, a later incarnation that can own a job is condemned at a larger round, so no job +/// removes its marker; a rebuild that regresses rounds breaks this. /// -/// GC freshness meta is ADD-ONLY: GC may publish `Condemned`, and may REMOVE the meta once the exact body -/// token is confirmed deleted/absent (`deleteConfirmedMeta`), but it NEVER transitions `Condemned -> -/// Clean` on a spare. The SOLE `-> Clean` transition is a WRITER that has already displaced the body with -/// a fresh incarnation token (`PartWriteTxn::ensureBlobPresent` publication + metadata reconciliation). Rationale: a deposed leader that cleared a -/// spare's meta then lost its round CAS would leave a durable stray-`Clean` over a still-condemned body; -/// a writer reading `Clean` would reuse the exact condemned token, which a stale pre-CAS exact-token -/// redelete then deletes -- live-blob data loss (INV_NO_LOSS). Removing the clear restores the exact-token -/// delete argument in full: once a hash is `Condemned`, observing `Clean` means EITHER the condemned body -/// is absent OR a writer already changed its incarnation, so every stale exact-incarnation delete of the -/// condemned incarnation finds the body absent or holding a different one. - -/// Write the per-hash meta to Condemned: a blob newly entering the retired set this round (either the -/// fresh zero-in-degree condemn, or a republication-supersede re-condemn of the current token). Absent meta -/// is created fresh; an already-Condemned meta (a racing condemn, or a replay of this same round) is left -/// alone rather than clobbering a possibly-newer condemn_round. -/// -/// Returns whether durable Condemned evidence exists after the call: the conditional write committed, -/// or an already-Condemned meta was observed. Any other write outcome reports false and writes nothing -/// further (the loser re-reads next time); a thrown backend error propagates (the scheduling wrapper -/// swallows it) — either way the entry stays UNCONFIRMED and the graduation gate carries it. -bool writeCondemnedMeta(CasOperation & op, const Layout & layout, const BlobRef & ref, +/// GC never writes `Clean`, not even on a spare: a deposed leader that cleared a spare's marker and then lost +/// its round CAS would leave `Clean` over a still-condemned body, and a stale exact-token redelete would delete +/// what a writer adopted. Only a writer that displaced the body writes `Clean` (`PartWriteTxn::ensureBlobPresent`). +void writeCondemnedMeta(CasOperation & op, const Layout & layout, const BlobRef & ref, uint64_t condemn_round, uint64_t size) { const auto lm = loadMeta(op, layout, ref); const BlobMeta desired{.state = MetaState::Condemned, .condemn_round = condemn_round, .size = size}; + /// An older round may be another incarnation's marker, which that incarnation's delete job may remove. if (!lm) - return std::holds_alternative(putMetaIfAbsent(op, layout, ref, desired)); - if (lm->meta.state != MetaState::Condemned) - return std::holds_alternative(casMeta(op, layout, ref, lm->etag, desired)); - return true; + putMetaIfAbsent(op, layout, ref, desired); + else if (lm->meta.state != MetaState::Condemned || lm->meta.condemn_round < condemn_round) + casMeta(op, layout, ref, lm->etag, desired); } -/// Drop the meta after its body was physically deleted (or already found absent) by the round's -/// exact-token delete. NO tombstone -- an absent meta reads exactly like a Clean one (absent -/// means not condemned"). Idempotent: an already-absent meta, or one a racing writer/GC pass already -/// moved, is a silent no-op. -void deleteConfirmedMeta(CasOperation & op, const Layout & layout, const BlobRef & ref) +/// Rounds only grow while `gc/state` is readable, so a marker at a newer round guards a later incarnation +/// and `Clean` means a writer displaced the body: both are left alone. `If-Match` the etag read here, so a +/// marker changed since the read is left alone too. +void deleteConfirmedMeta(CasOperation & op, const Layout & layout, const BlobRef & ref, uint64_t condemn_round) { const auto lm = loadMeta(op, layout, ref); - if (!lm) + if (!lm || lm->meta.state != MetaState::Condemned || lm->meta.condemn_round > condemn_round) return; deleteMetaExact(op, layout, ref, lm->etag); } @@ -144,25 +116,23 @@ void GcMetaWriter::submit(std::function op) } } -void GcMetaWriter::scheduleCondemnMarkerWrite(const BlobRef & ref, const PersistedEtag & token, - uint64_t condemn_round, uint64_t size) +void GcMetaWriter::scheduleCondemnMarkerWrite(const BlobRef & ref, uint64_t condemn_round, uint64_t size) { /// The job admits its OWN operation: a `CasOperation` carries per-call state and belongs to one /// task, while several of these run concurrently on the pool. - submit([st = state, ref, token, condemn_round, size]() + submit([st = state, ref, condemn_round, size]() { CasOperation op = st->store->openRequests().admit(); - if (writeCondemnedMeta(op, st->store->layout(), ref, condemn_round, size)) - st->noteCondemnMarkerDurable(ref, token); + writeCondemnedMeta(op, st->store->layout(), ref, condemn_round, size); }); } -void GcMetaWriter::scheduleConfirmedMetaDelete(const BlobRef & ref) +void GcMetaWriter::scheduleConfirmedMetaDelete(const BlobRef & ref, uint64_t condemn_round) { - submit([st = state, ref]() + submit([st = state, ref, condemn_round]() { CasOperation op = st->store->openRequests().admit(); - deleteConfirmedMeta(op, st->store->layout(), ref); + deleteConfirmedMeta(op, st->store->layout(), ref, condemn_round); }); } @@ -201,37 +171,4 @@ uint64_t GcMetaWriter::completed() const return state->completed.load(std::memory_order_relaxed); } -void GcMetaWriter::State::noteCondemnMarkerDurable(const BlobRef & ref, const PersistedEtag & token) -{ - std::lock_guard lock(condemn_marker_mutex); - condemn_markers_confirmed.emplace(ref, condemnMarkerKey(token)); -} - -bool GcMetaWriter::State::condemnMarkerConfirmedInProcess(const BlobRef & ref, const PersistedEtag & token) -{ - std::lock_guard lock(condemn_marker_mutex); - return condemn_markers_confirmed.contains({ref, condemnMarkerKey(token)}); -} - -void GcMetaWriter::State::forgetCondemnMarker(const BlobRef & ref, const PersistedEtag & token) -{ - std::lock_guard lock(condemn_marker_mutex); - condemn_markers_confirmed.erase({ref, condemnMarkerKey(token)}); -} - -void GcMetaWriter::noteCondemnMarkerDurable(const BlobRef & ref, const PersistedEtag & token) -{ - state->noteCondemnMarkerDurable(ref, token); -} - -bool GcMetaWriter::condemnMarkerConfirmedInProcess(const BlobRef & ref, const PersistedEtag & token) -{ - return state->condemnMarkerConfirmedInProcess(ref, token); -} - -void GcMetaWriter::forgetCondemnMarker(const BlobRef & ref, const PersistedEtag & token) -{ - state->forgetCondemnMarker(ref, token); -} - } diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.h index 9392da61c9e61..b57fbe46427b3 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcMetaWriter.h @@ -9,7 +9,6 @@ #include #include #include -#include namespace DB::Cas { @@ -29,15 +28,15 @@ class GcMetaWriter GcMetaWriter(const GcMetaWriter &) = delete; GcMetaWriter & operator=(const GcMetaWriter &) = delete; - /// Publish durable Condemned evidence for one (blob, exact incarnation) pair. On success the - /// pair is recorded in the in-process confirmation registry, which the graduation gate reads. A - /// refused write or a thrown error leaves the pair UNCONFIRMED: the gate then carries the entry and - /// a later round retries the write. - void scheduleCondemnMarkerWrite(const BlobRef & ref, const PersistedEtag & token, - uint64_t condemn_round, uint64_t size); + /// Write `Condemned` at `condemn_round` for one blob: create it over an absent marker, replace `Clean` + /// or an older-round `Condemned`, leave one at `condemn_round` or newer. A thrown write is counted and + /// logged, a write that lost its condition is dropped silently; either way the graduation gate reads the + /// marker itself and schedules the write again. + void scheduleCondemnMarkerWrite(const BlobRef & ref, uint64_t condemn_round, uint64_t size); - /// Drop the freshness meta of a blob whose body is confirmed deleted or absent. - void scheduleConfirmedMetaDelete(const BlobRef & ref); + /// Drop the marker of a blob whose body is confirmed deleted or absent, but only while it is `Condemned` + /// at `condemn_round` or older: a newer marker, or `Clean`, belongs to a later incarnation. + void scheduleConfirmedMetaDelete(const BlobRef & ref, uint64_t condemn_round); /// Successful-path protocol barrier. Wait for every job scheduled so far, propagating any /// pool/framework exception recorded by `ThreadPool`; per-hash operation exceptions are caught @@ -51,15 +50,6 @@ class GcMetaWriter uint64_t scheduled() const; uint64_t completed() const; - /// The in-process condemn-marker confirmation registry, keyed (blob, rendered incarnation). It is - /// keyed by the PERSISTED pair because every entry that consults it arrives from a durable - /// condemned row; a live observation enters through `PersistedEtag::capture`. Pool - /// completions insert concurrently with the round thread's reads, and the round thread also - /// inserts directly when it re-checks a marker synchronously. - void noteCondemnMarkerDurable(const BlobRef & ref, const PersistedEtag & token); - bool condemnMarkerConfirmedInProcess(const BlobRef & ref, const PersistedEtag & token); - void forgetCondemnMarker(const BlobRef & ref, const PersistedEtag & token); - private: /// Everything a job reaches. Held by `shared_ptr` and captured by value into every job. struct State @@ -68,12 +58,6 @@ class GcMetaWriter LoggerPtr logger; std::atomic scheduled{0}; std::atomic completed{0}; - std::mutex condemn_marker_mutex; - std::set> condemn_markers_confirmed; - - void noteCondemnMarkerDurable(const BlobRef & ref, const PersistedEtag & token); - bool condemnMarkerConfirmedInProcess(const BlobRef & ref, const PersistedEtag & token); - void forgetCondemnMarker(const BlobRef & ref, const PersistedEtag & token); }; /// Catch each meta-operation exception, count the job, and put it on the pool -- running it diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcReadAhead.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcReadAhead.h index 19b9df5a24c89..139f3c0ba1963 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcReadAhead.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGcReadAhead.h @@ -30,6 +30,8 @@ namespace DB::Cas /// an epoch's closing seal is (it learns that only by decoding the log at that position), so it may /// hint ids past the seal, inside the SAME epoch, that turn out not to exist. Those are discarded /// rather than taken, overshooting by at most one window per epoch crossing, and counted wasted. +/// A blob marker is not write-once: the fold hints it only after the round's cut, which is all the +/// graduation gate needs. /// /// Memory is the CALLER's to bound: `pending` counts hinted-but-untaken slots and `window` is how /// many a hinting site keeps in flight. A key hinted twice is one request. Results never taken are @@ -61,6 +63,8 @@ class GcReadAhead /// Hinted but not yet taken, both verbs together: what a hinting site throttles itself against. size_t pending() const { return reads.size() + heads.size(); } + /// Hinted HEADs not yet taken: the head hinting site throttles on its own count. + size_t pendingHeads() const { return heads.size(); } /// How many requests a hinting site should keep in flight. Zero at concurrency 1, which is what /// makes every `while (pending() < window())` loop hint nothing at all on the sequential setting. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasBlobMeta.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasBlobMeta.cpp index a6ded95e363ce..f7a2ee8375df7 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasBlobMeta.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasBlobMeta.cpp @@ -12,13 +12,17 @@ namespace ProfileEvents namespace DB::Cas { +std::optional decodeLoadedMeta(std::optional object) +{ + if (!object) + return std::nullopt; + return LoadedMeta{.meta = decodeBlobMeta(object->bytes), .etag = std::move(object->etag)}; +} + std::optional loadMeta(CasOperation & op, const Layout & layout, const BlobRef & ref, const Retry & policy) { - auto got = op.read(layout.blobMetaKey(ref), policy); - if (!got) - return std::nullopt; - return LoadedMeta{.meta = decodeBlobMeta(got->bytes), .etag = std::move(got->etag)}; + return decodeLoadedMeta(op.read(layout.blobMetaKey(ref), policy)); } WriteResult putMetaIfAbsent(CasOperation & op, const Layout & layout, const BlobRef & ref, diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasBlobMeta.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasBlobMeta.h index df84aee4d9f2c..ea4ead430d77c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasBlobMeta.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasBlobMeta.h @@ -38,6 +38,9 @@ struct LoadedMeta std::optional loadMeta(CasOperation & op, const Layout & layout, const BlobRef & ref, const Retry & policy = Retry::standard()); +/// The decode half of `loadMeta`, for a read issued elsewhere (the GC read-ahead). Decoding errors propagate. +std::optional decodeLoadedMeta(std::optional object); + /// Creates the marker only when its key is absent, on the plane `op` belongs to -- like its siblings, /// so one caller's decision cannot end up split across two fences. Anything at the key that this call /// did not itself write -- a stale `Condemned` marker still present when a vanished body is freshly diff --git a/src/Disks/tests/cas_gc_marker_test_support.h b/src/Disks/tests/cas_gc_marker_test_support.h new file mode 100644 index 0000000000000..efba25b9d7acb --- /dev/null +++ b/src/Disks/tests/cas_gc_marker_test_support.h @@ -0,0 +1,372 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace ProfileEvents +{ +extern const Event CASGCCondemnMarkerUnconfirmedCarry; +extern const Event CASGCMetaWriteAnomaly; +extern const Event CASGCReadAheadWasted; +} + +/// Fixtures for the condemn-marker gate: real GC rounds over a raw ref log, so a test decides which edges each +/// round folds; real writers where adoption is the subject; backends that act at one chosen request. +namespace DB::Cas::tests +{ + +inline const RootNamespace kGateNs{"00/aa@cas@"}; +inline const UInt128 kGateLeaderA = hexToU128("000000000000000000000000000000a1"); +inline const UInt128 kGateLeaderB = hexToU128("000000000000000000000000000000b2"); + +inline BlobRef gateRef(const UInt128 & hash) +{ + return BlobRef{BlobHashAlgo::CityHash128, BlobDigest::fromU128(hash)}; +} + +/// A pool that folds every round, so a round's number is the number of rounds run. +template +PoolPtr openGatePool(std::shared_ptr backend, uint64_t io_concurrency = 1, uint64_t graduation_budget = 0, + std::function apply_hook = {}, uint64_t gc_shards = 1) +{ + PoolConfig config{.pool_prefix = "p", .server_root_id = "test", + .gc_fold_max_defer_rounds = 0, .gc_io_concurrency = io_concurrency}; + config.gc_shards = gc_shards; + config.gc_round_graduation_budget = graduation_budget; + config.gc_redelete_apply_hook_for_test = std::move(apply_hook); + return Pool::open(std::move(backend), std::move(config)); +} + +/// Publishes `ref_name` in `kGateNs` owning every blob of `hashes`, writing each missing body first. +/// `build_sequence` names the manifest, so it must differ between publications. +inline ManifestRef publishBlobs(Backend & backend, const Layout & layout, const String & ref_name, + uint64_t build_sequence, const std::vector & hashes) +{ + std::vector entries; + for (size_t i = 0; i < hashes.size(); ++i) + { + bool present = false; + { + OperationForTest op(backend); + present = (*op).head(layout.blobKey(gateRef(hashes[i])), Retry::standard()).has_value(); + } + if (!present) + writeBlobBody(backend, layout, hashes[i]); + entries.push_back(blobEntryFor("f" + std::to_string(i), hashes[i])); + } + const ManifestRef manifest{.writer_epoch = 1, .build_sequence = build_sequence, .manifest_ordinal = 1}; + writeManifestRaw(backend, layout, kGateNs, manifest, entries); + publishCommittedTransition(backend, layout, kGateNs, ref_name, std::nullopt, manifest); + return manifest; +} + +inline void dropBlobs(Backend & backend, const Layout & layout, const String & ref_name, const ManifestRef & manifest) +{ + dropRefTransition(backend, layout, kGateNs, ref_name, manifest); +} + +inline uint64_t committedGcRound(Backend & backend, const Layout & layout) +{ + OperationForTest op(backend); + const auto got = (*op).read(layout.gcStateKey(), Retry::standard()); + return got ? decodeGcState(got->bytes).round : 0; +} + +/// The retired entry of `ref` in the adopted seal, looked up in the blob's own GC shard. +inline std::optional retiredEntryFor(Backend & backend, const Layout & layout, const BlobRef & ref, + uint64_t gc_shards = 1) +{ + for (const RetiredEntry & entry : currentRetiredSet(backend, layout, blobShard(ref, gc_shards))) + if (entry.ref == ref) + return entry; + return std::nullopt; +} + +inline std::optional markerOf(Backend & backend, const Layout & layout, const BlobRef & ref) +{ + OperationForTest op(backend); + return loadMeta(*op, layout, ref); +} + +inline BlobMeta condemnedAt(uint64_t round, uint64_t size = 1) +{ + return BlobMeta{.state = MetaState::Condemned, .condemn_round = round, .size = size}; +} + +inline BlobMeta cleanMarker(uint64_t size = 1) +{ + return BlobMeta{.state = MetaState::Clean, .condemn_round = 0, .size = size}; +} + +/// Replaces the marker of `ref` with `meta`, or creates it when absent. +inline void setMarker(Backend & backend, const Layout & layout, const BlobRef & ref, const BlobMeta & meta) +{ + OperationForTest op(backend); + const auto current = loadMeta(*op, layout, ref); + const WriteResult written = current ? casMeta(*op, layout, ref, current->etag, meta) + : putMetaIfAbsent(*op, layout, ref, meta); + ASSERT_TRUE(std::holds_alternative(written)) << layout.blobMetaKey(ref); +} + +inline void removeMarker(Backend & backend, const Layout & layout, const BlobRef & ref) +{ + OperationForTest op(backend); + const auto current = loadMeta(*op, layout, ref); + ASSERT_TRUE(current.has_value()) << layout.blobMetaKey(ref); + ASSERT_EQ(deleteMetaExact(*op, layout, ref, current->etag), Removal::Removed); +} + +inline bool blobPresent(Backend & backend, const Layout & layout, const BlobRef & ref) +{ + OperationForTest op(backend); + return (*op).head(layout.blobKey(ref), Retry::standard()).has_value(); +} + +/// Moves the GC lease to `owner` as a steal would, without the observation window. +inline void handGcLeaseTo(Backend & backend, const Layout & layout, const UInt128 & owner) +{ + OperationForTest op(backend); + const auto got = (*op).read(layout.gcStateKey(), Retry::standard()); + ASSERT_TRUE(got.has_value()); + GcState state = decodeGcState(got->bytes); + state.lease.owner = owner; + ++state.lease.seq; + ASSERT_TRUE(std::holds_alternative( + (*op).replace(layout.gcStateKey(), encodeGcState(state), got->etag, Retry::once()))); +} + +inline RoundReport runGateRound(const PoolPtr & store, Gc & gc, UniversePolicy policy = UniversePolicy::Authoritative) +{ + const RoundReport report = gc.runRegularRound({}, /*allow_steal*/ true, policy); + store->renewWatermarkOnce(); + return report; +} + +/// Runs rounds until `done` holds after one of them; fails the test after `max_rounds`. +inline void runRoundsUntil(const PoolPtr & store, Gc & gc, const std::function & done, size_t max_rounds = 12) +{ + for (size_t i = 0; i < max_rounds; ++i) + { + runGateRound(store, gc); + if (done()) + return; + } + FAIL() << "the condition did not hold within " << max_rounds << " rounds"; +} + +/// Publishes `hashes`, folds their edges, drops the ref and runs rounds until every blob is condemned. All of +/// them are condemned in one round, written to `condemn_round`. +inline void condemnBlobs(const PoolPtr & store, Gc & gc, Backend & backend, const String & ref_name, + uint64_t build_sequence, const std::vector & hashes, uint64_t & condemn_round) +{ + const Layout & layout = store->layout(); + const uint64_t shards = store->poolConfig().gc_shards; + const ManifestRef manifest = publishBlobs(backend, layout, ref_name, build_sequence, hashes); + store->renewWatermarkOnce(); + runGateRound(store, gc); + dropBlobs(backend, layout, ref_name, manifest); + store->renewWatermarkOnce(); + ASSERT_NO_FATAL_FAILURE(runRoundsUntil(store, gc, [&] + { + for (const UInt128 & hash : hashes) + if (!retiredEntryFor(backend, layout, gateRef(hash), shards)) + return false; + return true; + })); + condemn_round = retiredEntryFor(backend, layout, gateRef(hashes.front()), shards)->condemn_round; +} + +/// Writes one part holding `payload` through the writer. `Observed` means the writer adopted the stored +/// incarnation; `Published` means it uploaded a fresh one. +inline BlobMaterializationAction writeOnePart(const PoolPtr & store, const String & ns, const String & ref_name, + const String & payload) +{ + const RootNamespace root{ns}; + PartWriteInfo info; + info.intended_ref = ns + "/" + ref_name; + auto build = store->beginPartWrite(info); + ManifestEntry entry; + entry.path = "data.bin"; + entry.placement = EntryPlacement::Blob; + entry.ref = idOf(payload); + entry.blob_size = payload.size(); + const ManifestId id = build->stageManifest({entry}); + build->precommitAdd(root, ref_name, id); + const BlobUploadResult uploaded = build->uploadBlobDetached(BlobUploadRequest{ + .ref = entry.ref, .source = BlobSource::fromString(payload), .declared_size = payload.size()}); + build->mergeBlobUploadResults(std::span(&uploaded, 1)); + build->promote(root, ref_name, build->buildId(), id); + return uploaded.diagnostics.action; +} + +/// A bounded spin that fails the test instead of hanging. +inline void awaitCondition(const std::function & condition) +{ + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while (!condition()) + { + ASSERT_LT(std::chrono::steady_clock::now(), deadline) << "the awaited condition never held"; + std::this_thread::yield(); + } +} + +inline uint64_t unconfirmedCarries() +{ + return ProfileEvents::global_counters[ProfileEvents::CASGCCondemnMarkerUnconfirmedCarry].load(); +} + +inline uint64_t metaWriteAnomalies() +{ + return ProfileEvents::global_counters[ProfileEvents::CASGCMetaWriteAnomaly].load(); +} + +inline uint64_t readAheadWasted() +{ + return ProfileEvents::global_counters[ProfileEvents::CASGCReadAheadWasted].load(); +} + +/// Runs a hook once, on the first HEAD of one key after `armOnHead`, before that HEAD is served and with no +/// backend lock held. A writer that commits there lands after the round's cut and before the redelete +/// observes the blob. +class HeadHookBackend : public CountingBackend +{ +public: + void armOnHead(const String & key, std::function hook) + { + std::lock_guard lock(hook_mutex); + hooked_key = key; + on_head = std::move(hook); + } + + std::optional head(const String & key, TransportAccess & access) override + { + std::function fire; + { + std::lock_guard lock(hook_mutex); + if (on_head && key == hooked_key) + fire = std::exchange(on_head, {}); + } + if (fire) + fire(); + return CountingBackend::head(key, access); + } + +private: + std::mutex hook_mutex; + String hooked_key; + std::function on_head; +}; + +/// Parks the first redelete apply of `target` until released. The parked leader has removed the body and has +/// not yet scheduled its marker delete: the window a successor's rounds run in. +struct ApplyParking +{ + std::atomic armed{false}; + BlobRef target{}; + ManualBarrier barrier; +}; + +/// `target` is set before `armed`, and the hook reads `armed` first, so the two never race. +inline std::function parkingHook(const std::shared_ptr & parking) +{ + return [parking](const BlobRef & ref) + { + if (parking->armed.load() && ref == parking->target && parking->armed.exchange(false)) + parking->barrier.arriveAndWait(); + }; +} + + +/// Marker etags derived from the bytes, as S3 derives them: equal bytes carry equal etags, so a replaced +/// marker's etag comes back when the same bytes are written again. Every other key keeps minted values. +class ContentEtagBackend : public HeadHookBackend +{ +public: + std::optional read(const String & key, TransportAccess & access) override + { + auto raw = HeadHookBackend::read(key, access); + if (raw && isMarker(key)) + raw->value = contentValue(raw->bytes); + return raw; + } + + std::optional head(const String & key, TransportAccess & access) override + { + auto meta = HeadHookBackend::head(key, access); + if (meta && isMarker(key)) + if (const auto raw = InMemoryBackend::read(key, access)) + meta->value = contentValue(raw->bytes); + return meta; + } + + RawListPage list(const String & prefix, const String & cursor, size_t limit, TransportAccess & access) override + { + RawListPage page = HeadHookBackend::list(prefix, cursor, limit, access); + for (RawListedKey & listed : page.keys) + if (listed.value && isMarker(listed.key)) + if (const auto raw = InMemoryBackend::read(listed.key, access)) + listed.value = contentValue(raw->bytes); + return page; + } + + RawRemoval remove(const String & key, const String & expected_value, TransportAccess & access) override + { + if (!isMarker(key)) + return HeadHookBackend::remove(key, expected_value, access); + std::lock_guard lock(marker_mutex); + const auto current = InMemoryBackend::read(key, access); + if (!current) + return RawRemoval::Gone; + if (contentValue(current->bytes) != expected_value) + return RawRemoval::Mismatch; + return HeadHookBackend::remove(key, current->value, access); + } + + std::expected write(const String & key, const String & bytes, + const std::optional & expected_value, + TransportAccess & access) override + { + if (!isMarker(key)) + return HeadHookBackend::write(key, bytes, expected_value, access); + std::lock_guard lock(marker_mutex); + std::optional minted_expected; + if (expected_value) + { + const auto current = InMemoryBackend::read(key, access); + if (!current || contentValue(current->bytes) != *expected_value) + return std::unexpected(RawConflict{}); + minted_expected = current->value; + } + const auto written = HeadHookBackend::write(key, bytes, minted_expected, access); + if (!written) + return written; + return contentValue(bytes); + } + +private: + static bool isMarker(const String & key) { return key.ends_with(".meta"); } + static String contentValue(const String & bytes) { return "content-" + hexOf(bytes); } + + std::mutex marker_mutex; +}; + +} diff --git a/src/Disks/tests/gtest_cas_blob_indegree.cpp b/src/Disks/tests/gtest_cas_blob_indegree.cpp index d1eb99b05cc23..2177563cae6ef 100644 --- a/src/Disks/tests/gtest_cas_blob_indegree.cpp +++ b/src/Disks/tests/gtest_cas_blob_indegree.cpp @@ -14,6 +14,8 @@ #include #include +#include + namespace DB::ErrorCodes { extern const int ABORTED; extern const int CORRUPTED_DATA; extern const int NOT_IMPLEMENTED; } using namespace DB::Cas; @@ -207,7 +209,8 @@ SourceEdgeRecord condemnedRec(UInt128 h, const CondemnedRow & row) return SourceEdgeRecord{.ref = BlobRef{BlobHashAlgo::CityHash128, BlobDigest::fromU128(h)}, .source_id = UInt128{0}, .marker = RunMarker::Condemned, .delete_pending = row.delete_pending, .token = row.token, - .size = row.size, .condemn_round = row.condemn_round}; + .size = row.size, .condemn_round = row.condemn_round, + .marker_confirmed = row.marker_confirmed}; } /// An active-edge record (`RunMarker::Edge`) for `h` at source `sid`. @@ -1121,3 +1124,236 @@ TEST(CASThreeCursorMerge, RedeleteBudgetDrainsCohortToFixpointOverRounds) EXPECT_EQ(total_redeleted, 10u) << "no entry lost to the cap across the whole drain"; EXPECT_EQ(rounds, 4u) << "ceil(10 / 3) rounds to fully drain"; } + +TEST(CASThreeCursorMerge, GraduationBudgetIsCheckedBeforeTheGate) +{ + for (const size_t refusals : {size_t{0}, size_t{2}}) + { + SCOPED_TRACE(refusals); + InMemoryBackend backend; + DB::Cas::tests::OperationForTest backend_req(backend); + Layout layout{"pool"}; + const RunRef gen1 = writeSourceEdgeRun(backend, layout, 1, 0, 0, condemnedCohort(10, /*condemn_round*/1, false)); + + GcRoundWorkBudget budget; + budget.max_graduations = 3; + size_t gate_calls = 0; + const std::function gate = [&](const RetiredEntry &) { return ++gate_calls > refusals; }; + + std::vector runs2; + RetiredMergeResult rmr; + foldDeltasIntoGeneration(*backend_req, layout, /*prior_runs*/{gen1}, 2, 0, 0, {}, runs2, + /*current_round*/5, /*condemn_round*/6, /*head_blob*/{}, /*peek_head*/{}, gate, + &rmr, /*suppress_destructive*/false, /*out_applied_by_txn_ordinal*/nullptr, + /*source_retirements*/{}, &budget); + + EXPECT_EQ(gate_calls, 3 + refusals) << "the gate read entries the budget had no slot for"; + EXPECT_EQ(rmr.graduated.size(), 3u); + EXPECT_EQ(budget.graduations_used, 3u); + EXPECT_EQ(rmr.still_retired.size(), 10u); + } +} + +TEST(CASThreeCursorMerge, MarkerConfirmedBitDoesNotSkipTheGate) +{ + InMemoryBackend backend; + DB::Cas::tests::OperationForTest backend_req(backend); + Layout layout{"pool"}; + CondemnedRow row = condemnedRowFor(/*condemn_round*/1, "t1"); + row.marker_confirmed = true; + const RunRef gen1 = writeSourceEdgeRun(backend, layout, 1, 0, 0, {{b(1), row}}); + + size_t gate_calls = 0; + const std::function refuse = [&](const RetiredEntry &) { ++gate_calls; return false; }; + std::vector runs2; + RetiredMergeResult rmr; + foldDeltasIntoGeneration(*backend_req, layout, /*prior_runs*/{gen1}, 2, 0, 0, {}, runs2, + /*current_round*/5, /*condemn_round*/6, /*head_blob*/{}, /*peek_head*/{}, refuse, &rmr); + + EXPECT_EQ(gate_calls, 1u) << "a non-pending row's bit skipped the gate"; + EXPECT_TRUE(rmr.graduated.empty()); + ASSERT_EQ(rmr.still_retired.size(), 1u); + EXPECT_FALSE(rmr.still_retired.front().delete_pending); +} + +namespace +{ + +GraduationReadHints acceptAllHints(std::vector & offered) +{ + return GraduationReadHints{ + .admit = [](const BlobRef &) { return true; }, + .offer = [&offered](const BlobRef & ref, const CondemnedRow &) + { + offered.push_back(ref); + return true; + }}; +} + +std::vector> condemnedRowsOf(const std::vector & refs) +{ + std::vector> rows; + for (size_t i = 0; i < refs.size(); ++i) + rows.push_back({refs[i].digest.toU128(), condemnedRowFor(1, "t" + std::to_string(i))}); + return rows; +} + +} + +TEST(CASThreeCursorMerge, ScanAheadStopsAtTheRowCap) +{ + InMemoryBackend backend; + DB::Cas::tests::OperationForTest backend_req(backend); + Layout layout{"pool"}; + const uint64_t n = kGraduationLookaheadRows + 100; + const RunRef gen1 = writeSourceEdgeRun(backend, layout, 1, 0, 0, condemnedCohort(n, 1, false)); + + std::vector offered; + const GraduationReadHints hints = acceptAllHints(offered); + std::optional offers_at_first_gate; + const std::function gate = [&](const RetiredEntry &) + { + if (!offers_at_first_gate) + offers_at_first_gate = offered.size(); + return true; + }; + + std::vector with_scan; + RetiredMergeResult scanned; + foldDeltasIntoGeneration(*backend_req, layout, {gen1}, 2, 0, 0, {}, with_scan, /*current_round*/5, + /*condemn_round*/6, {}, {}, gate, &scanned, false, nullptr, {}, nullptr, &hints); + + ASSERT_TRUE(offers_at_first_gate.has_value()); + EXPECT_EQ(*offers_at_first_gate, kGraduationLookaheadRows + 1) << "the current row and a full buffer, no more"; + EXPECT_EQ(offered.size(), n - 1) << "every row after the first is offered"; + EXPECT_EQ(std::set(offered.begin(), offered.end()).size(), n - 1) << "a row was offered twice"; + + /// Same output key: a different run would fail the write-once replay with `CORRUPTED_DATA`. + std::vector without_scan; + RetiredMergeResult plain; + const std::function accept = [](const RetiredEntry &) { return true; }; + foldDeltasIntoGeneration(*backend_req, layout, {gen1}, 2, 0, 0, {}, without_scan, 5, 6, {}, {}, accept, &plain); + ASSERT_EQ(with_scan.size(), 1u); + ASSERT_EQ(without_scan.size(), 1u); + EXPECT_EQ(with_scan.front().checksum, without_scan.front().checksum); + EXPECT_EQ(scanned.graduated.size(), plain.graduated.size()); +} + +TEST(CASThreeCursorMerge, ScanAheadStopsAtTheSegmentEnd) +{ + DB::Cas::tests::CountingBackend backend; + DB::Cas::tests::OperationForTest backend_req(backend); + Layout layout{"pool"}; + std::vector refs; + for (uint64_t i = 1; i <= 20; ++i) + refs.push_back(bh(i)); + std::sort(refs.begin(), refs.end()); + const std::vector first(refs.begin(), refs.begin() + 10); + const std::vector second(refs.begin() + 10, refs.end()); + const RunRef seg1 = writeSourceEdgeRun(backend, layout, 1, 0, 0, condemnedRowsOf(first)); + const RunRef seg2 = writeSourceEdgeRun(backend, layout, 1, 1, 0, condemnedRowsOf(second)); + const std::set second_set(second.begin(), second.end()); + + size_t early_offers = 0; + std::vector offered; + const GraduationReadHints hints{ + .admit = [](const BlobRef &) { return true; }, + .offer = [&](const BlobRef & ref, const CondemnedRow &) + { + if (second_set.contains(ref) && backend.getStreamCount(seg2.key) == 0) + ++early_offers; + offered.push_back(ref); + return true; + }}; + std::vector gated; + size_t segment_two_opens_at_first_gate = 0; + const std::function gate = [&](const RetiredEntry & e) + { + /// The first gate runs right after the scan that offered segment 1's rows, which fit the row cap. + if (gated.empty()) + segment_two_opens_at_first_gate = backend.getStreamCount(seg2.key); + gated.push_back(e.ref); + return true; + }; + + std::vector with_scan; + foldDeltasIntoGeneration(*backend_req, layout, {seg1, seg2}, 2, 0, 0, {}, with_scan, 5, 6, {}, {}, gate, + nullptr, false, nullptr, {}, nullptr, &hints); + + EXPECT_EQ(segment_two_opens_at_first_gate, 0u) << "the scan opened segment 2 before the merge reached it"; + EXPECT_EQ(early_offers, 0u) << "a row of segment 2 was offered before the merge opened it"; + EXPECT_EQ(offered.size(), 19u) << "offers resume once the next segment is open"; + ASSERT_EQ(gated.size(), 20u); + EXPECT_EQ(gated[9], first.back()) << "the last entry of segment 1 is still admitted and read"; + + std::vector without_scan; + const std::function accept = [](const RetiredEntry &) { return true; }; + foldDeltasIntoGeneration(*backend_req, layout, {seg1, seg2}, 2, 0, 0, {}, without_scan, 5, 6, {}, {}, accept); + EXPECT_EQ(with_scan.front().checksum, without_scan.front().checksum); +} + +TEST(CASThreeCursorMerge, ScanAheadKeepsRowAndChecksumValidation) +{ + Layout layout{"pool"}; + std::vector unused; + const GraduationReadHints hints = acceptAllHints(unused); + const std::function accept = [](const RetiredEntry &) { return true; }; + + for (const bool with_scan : {false, true}) + { + SCOPED_TRACE(with_scan); + /// A duplicate sentinel in the middle of a condemned cohort. + { + InMemoryBackend backend; + DB::Cas::tests::OperationForTest op(backend); + std::vector recs; + for (uint64_t i = 1; i <= 6; ++i) + recs.push_back(condemnedRec(b(i), condemnedRowFor(1, "t" + std::to_string(i)))); + std::sort(recs.begin(), recs.end(), [](const SourceEdgeRecord & l, const SourceEdgeRecord & r) { return l.ref < r.ref; }); + const SourceEdgeRecord duplicate = recs[2]; + recs.insert(recs.begin() + 3, duplicate); + DB::WriteBufferFromOwnString out; + SourceEdgeRunWriter writer(out); + for (const SourceEdgeRecord & rec : recs) + writer.append(rec); + writer.finish(); + out.finalize(); + const String bytes = out.str(); + const RunRef bad{.key = layout.blobTargetRunKey(1, 0, 0, 0), .checksum = sourceEdgeRunChecksum(bytes), + .shard = 0, .key_generation = 1}; + (*op).create(bad.key, bytes, Retry::once()); + + std::vector out_runs; + DB::Cas::tests::expectThrowsCode(DB::ErrorCodes::CORRUPTED_DATA, [&] + { + foldDeltasIntoGeneration(*op, layout, {bad}, 2, 0, 0, {}, out_runs, 5, 6, {}, {}, accept, nullptr, + false, nullptr, {}, nullptr, with_scan ? &hints : nullptr); + }); + EXPECT_TRUE(out_runs.empty()); + EXPECT_FALSE((*op).read(layout.blobTargetRunKey(2, 0, 0, 0), Retry::once()).has_value()) << "a successor run was written"; + } + /// A bad seal checksum on segment 2. + { + InMemoryBackend backend; + DB::Cas::tests::OperationForTest op(backend); + std::vector refs; + for (uint64_t i = 1; i <= 8; ++i) + refs.push_back(bh(i)); + std::sort(refs.begin(), refs.end()); + const RunRef seg1 = writeSourceEdgeRun(backend, layout, 1, 0, 0, + condemnedRowsOf(std::vector(refs.begin(), refs.begin() + 4))); + RunRef seg2 = writeSourceEdgeRun(backend, layout, 1, 1, 0, + condemnedRowsOf(std::vector(refs.begin() + 4, refs.end()))); + seg2.checksum = seg2.checksum + 1; + + std::vector out_runs; + DB::Cas::tests::expectThrowsCode(DB::ErrorCodes::CORRUPTED_DATA, [&] + { + foldDeltasIntoGeneration(*op, layout, {seg1, seg2}, 2, 0, 0, {}, out_runs, 5, 6, {}, {}, accept, + nullptr, false, nullptr, {}, nullptr, with_scan ? &hints : nullptr); + }); + EXPECT_TRUE(out_runs.empty()); + EXPECT_FALSE((*op).read(layout.blobTargetRunKey(2, 0, 0, 0), Retry::once()).has_value()) << "a successor run was written"; + } + } +} diff --git a/src/Disks/tests/gtest_cas_gc_ack_floor.cpp b/src/Disks/tests/gtest_cas_gc_ack_floor.cpp index ea63268a8fe36..ba4259709f740 100644 --- a/src/Disks/tests/gtest_cas_gc_ack_floor.cpp +++ b/src/Disks/tests/gtest_cas_gc_ack_floor.cpp @@ -1361,11 +1361,16 @@ TEST(CASGCCondemnMarker, SwallowedMarkerWriteCarriesEntryInsteadOfDeleting) << "every refused graduation must count one unconfirmed carry"; /// Heal the backend: the carry-time retry publishes the marker, the entry confirms + graduates, and - /// the pipeline reclaims the blob and drops the meta. + /// the pipeline reclaims the blob. The retry stamped a round above the entry's, so the entry's marker + /// delete leaves it: an orphan over the absent body. + const uint64_t entry_round = e->condemn_round; backend->fail_meta_writes.store(false); EXPECT_TRUE(runRoundsUntilAbsent(store, gc, *backend, store->layout(), blob)); EXPECT_FALSE(currentEntryFor(*backend, store->layout(), blob).has_value()); - EXPECT_FALSE(loadMetaForTest(*backend, store->layout(), blob).has_value()); + const auto orphan = loadMetaForTest(*backend, store->layout(), blob); + ASSERT_TRUE(orphan.has_value()); + EXPECT_EQ(orphan->meta.state, MetaState::Condemned); + EXPECT_GT(orphan->meta.condemn_round, entry_round); } /// The healthy-path counterpart: with the condemn-time marker write landing normally, the gate must not @@ -1409,14 +1414,9 @@ TEST(CASGCCondemnMarker, DurableMarkerKeepsCanonicalGraduationSchedule) } } -/// The leader-restart path: `condemn_markers_confirmed` is a process-local registry on `Gc`, lost on a -/// GC leader restart. Same idiom as the crash-replay tests above (`Gc gc2(store, kGc)` -- a fresh `Gc` -/// object under the SAME identity models a process restart that resumes its own lease, not a steal by a -/// different owner). The fresh instance must still confirm graduation via the ONE synchronous `loadMeta` -/// re-check: the durable `Condemned` meta observed now is sufficient evidence on its own, with no -/// in-process (hash, token) confirmation available at all. This proves the fallback branch -- not just -/// the in-process registry -- authorizes the delete. -TEST(CASGCCondemnMarker, LoadMetaFallbackConfirmsGraduationAfterLeaderRestart) +/// The leader-restart path: a fresh `Gc` under the same identity models a process restart that resumes its own +/// lease. It never observed the condemn round, and still graduates on the durable `Condemned` marker it reads. +TEST(CASGCCondemnMarker, MarkerReadConfirmsGraduationAfterLeaderRestart) { auto backend = std::make_shared(); auto store = openPoolForTest(backend); @@ -1429,7 +1429,7 @@ TEST(CASGCCondemnMarker, LoadMetaFallbackConfirmsGraduationAfterLeaderRestart) { /// The first (soon-to-be-gone) leader: seeds the blob, condemns it, and lets the marker write - /// land on the healthy backend. Its `condemn_markers_confirmed` registry dies with it. + /// land on the healthy backend. Nothing it observed survives it. Gc gc(store, kGc); runRegularRoundReclaiming(gc); dropRefTransition(*backend, store->layout(), ns, "tbl", r); @@ -1442,15 +1442,11 @@ TEST(CASGCCondemnMarker, LoadMetaFallbackConfirmsGraduationAfterLeaderRestart) EXPECT_EQ(lm->meta.state, MetaState::Condemned) << "precondition: the durable marker must be on disk before the simulated restart"; - /// A brand-new `Gc` object under the SAME identity -- an empty `condemn_markers_confirmed`, exactly - /// as after a process restart that resumes its own lease. It never observed the condemn round above, - /// so the in-process confirmation path (`condemnMarkerConfirmedInProcess`) has nothing to return true - /// for; only the `loadMeta` fallback can authorize graduation. + /// A brand-new `Gc` under the same identity: only the marker read can authorize graduation. Gc gc2(store, kGc); const RoundReport rep = runRegularRoundReclaiming(gc2); EXPECT_EQ(rep.graduated, 1u) - << "the loadMeta fallback (leader-restart path) must authorize graduation from durable evidence " - "alone"; + << "after a leader restart the marker read alone must authorize graduation"; const auto e = currentEntryFor(*backend, store->layout(), blob); ASSERT_TRUE(e.has_value()); EXPECT_TRUE(e->delete_pending); diff --git a/src/Disks/tests/gtest_cas_gc_meta_writer.cpp b/src/Disks/tests/gtest_cas_gc_meta_writer.cpp index c51b006be9bfa..2c1f23ceaacd4 100644 --- a/src/Disks/tests/gtest_cas_gc_meta_writer.cpp +++ b/src/Disks/tests/gtest_cas_gc_meta_writer.cpp @@ -4,15 +4,20 @@ #include #include #include +#include #include +#include + #include +#include #include #include #include #include using namespace DB::Cas; +using namespace DB::Cas::tests; using DB::Cas::tests::MetaWriteLatchBackend; using DB::Cas::tests::awaitLatchEntered; using DB::Cas::tests::openRequestsForTest; @@ -38,11 +43,9 @@ TEST(CASGcMetaWriter, RealCondemnMarkerJobCompletesAcrossOwnerDestruction) auto store = Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); const BlobRef ref = DB::Cas::tests::idOf("1"); - const PersistedEtag token{"emulated", "tok-1"}; - auto gc = std::make_unique(store, DB::Cas::tests::u128Of(kGcId)); backend->arm(); - gc->metaWriterForTest().scheduleCondemnMarkerWrite(ref, token, /*condemn_round=*/1, /*size=*/128); + gc->metaWriterForTest().scheduleCondemnMarkerWrite(ref, /*condemn_round=*/1, /*size=*/128); awaitLatchEntered(*backend); @@ -58,30 +61,9 @@ TEST(CASGcMetaWriter, RealCondemnMarkerJobCompletesAcrossOwnerDestruction) EXPECT_EQ(meta->meta.condemn_round, 1u); } -/// The confirmation registry is written by the pool thread and read by the graduation gate. Assert it -/// on a `Gc` that is still alive, so the read is possible at all: after destruction there is no -/// registry left to consult, which is the documented behaviour a fresh leader relies on. -TEST(CASGcMetaWriter, CondemnMarkerConfirmationIsVisibleAfterDrain) -{ - auto backend = std::make_shared(); - auto store = Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); - - const BlobRef ref = DB::Cas::tests::idOf("1"); - const PersistedEtag token{"emulated", "tok-1"}; - - Gc gc(store, DB::Cas::tests::u128Of(kGcId)); - EXPECT_FALSE(gc.metaWriterForTest().condemnMarkerConfirmedInProcess(ref, token)); - - gc.metaWriterForTest().scheduleCondemnMarkerWrite(ref, token, /*condemn_round=*/1, /*size=*/128); - gc.metaWriterForTest().drain(); - - EXPECT_TRUE(gc.metaWriterForTest().condemnMarkerConfirmedInProcess(ref, token)); - EXPECT_EQ(gc.metaWriterForTest().scheduled(), gc.metaWriterForTest().completed()); -} - -/// Same lifetime property for the other production job. `deleteConfirmedMeta` RETURNS IMMEDIATELY when -/// no meta object exists (`Gc/CasGcMetaWriter.cpp`), so the meta must be seeded first -- otherwise the -/// job never reaches the latch and the wait above is waiting for something that will never happen. +/// Same lifetime property for the other production job. `deleteConfirmedMeta` returns before the latch +/// unless the marker is `Condemned` at a round not newer than the job's, so the meta must be seeded that +/// way first -- otherwise the wait above is waiting for something that will never happen. TEST(CASGcMetaWriter, RealConfirmedMetaDeleteCompletesAcrossOwnerDestruction) { auto backend = std::make_shared(); @@ -96,7 +78,7 @@ TEST(CASGcMetaWriter, RealConfirmedMetaDeleteCompletesAcrossOwnerDestruction) auto gc = std::make_unique(store, DB::Cas::tests::u128Of(kGcId)); backend->arm(); - gc->metaWriterForTest().scheduleConfirmedMetaDelete(ref); + gc->metaWriterForTest().scheduleConfirmedMetaDelete(ref, /*condemn_round=*/1); awaitLatchEntered(*backend); @@ -109,7 +91,7 @@ TEST(CASGcMetaWriter, RealConfirmedMetaDeleteCompletesAcrossOwnerDestruction) } /// A round that throws must not leave its meta jobs running into the next round: their effects would -/// land in the registry the next round's graduation gate reads, and inside its counter deltas. +/// land inside the next round's marker reads and counter deltas. /// /// The round is made to throw at its outcome-log write, with the confirmed-meta delete it scheduled a /// few lines earlier held inside the backend. The round must then BLOCK, draining, until that job is @@ -192,3 +174,604 @@ TEST(CASGcMetaWriter, ThrowingRoundDrainsBeforeReturning) EXPECT_EQ(gc.metaWriterForTest().scheduled(), gc.metaWriterForTest().completed()); } + +namespace +{ + +/// The delete job, scheduled the way `Gc::applyRedeleteOutcome` schedules it, and drained. +void runConfirmedMetaDelete(Gc & gc, const BlobRef & ref, uint64_t condemn_round) +{ + gc.metaWriterForTest().scheduleConfirmedMetaDelete(ref, condemn_round); + gc.metaWriterForTest().drain(); +} + +PoolPtr openPlainPool(const std::shared_ptr & backend) +{ + return Pool::open(backend, PoolConfig{.pool_prefix = "p", .server_root_id = "test"}); +} + +} + +TEST(CASGcMetaWriter, ConfirmedMetaDeleteSparesANewerRound) +{ + auto backend = std::make_shared(); + auto store = openPlainPool(backend); + const BlobRef ref = idOf("u1"); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, store->layout(), ref, condemnedAt(5))); + const auto before = markerOf(*backend, store->layout(), ref); + ASSERT_TRUE(before.has_value()); + + Gc gc(store, u128Of(kGcId)); + runConfirmedMetaDelete(gc, ref, 3); + + const auto after = markerOf(*backend, store->layout(), ref); + ASSERT_TRUE(after.has_value()) << "a job for round 3 deleted a Condemned(5) marker"; + EXPECT_EQ(after->meta.condemn_round, 5u); + EXPECT_EQ(after->etag, before->etag); +} + +TEST(CASGcMetaWriter, ConfirmedMetaDeleteSparesACleanMarker) +{ + auto backend = std::make_shared(); + auto store = openPlainPool(backend); + const BlobRef ref = idOf("u2"); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, store->layout(), ref, cleanMarker())); + + Gc gc(store, u128Of(kGcId)); + runConfirmedMetaDelete(gc, ref, 3); + + const auto after = markerOf(*backend, store->layout(), ref); + ASSERT_TRUE(after.has_value()) << "a job deleted a Clean marker"; + EXPECT_EQ(after->meta.state, MetaState::Clean); +} + +TEST(CASGcMetaWriter, ConfirmedMetaDeleteRemovesItsOwnOrAnOlderRound) +{ + auto backend = std::make_shared(); + auto store = openPlainPool(backend); + const BlobRef own = idOf("u3-own"); + const BlobRef older = idOf("u3-older"); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, store->layout(), own, condemnedAt(3))); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, store->layout(), older, condemnedAt(2))); + + Gc gc(store, u128Of(kGcId)); + runConfirmedMetaDelete(gc, own, 3); + runConfirmedMetaDelete(gc, older, 3); + + EXPECT_FALSE(markerOf(*backend, store->layout(), own).has_value()); + EXPECT_FALSE(markerOf(*backend, store->layout(), older).has_value()); +} + +TEST(CASGcMetaWriter, ConfirmedMetaDeleteRefusesAMarkerChangedAfterItsRead) +{ + auto backend = std::make_shared(); + auto store = openPlainPool(backend); + const BlobRef ref = idOf("u6"); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, store->layout(), ref, condemnedAt(3))); + + backend->setHoldDeletes(true); + Gc gc(store, u128Of(kGcId)); + runConfirmedMetaDelete(gc, ref, 3); + ASSERT_EQ(backend->pendingDeletes(), 1u) << "the job read Condemned(3) and sent its delete"; + + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, store->layout(), ref, condemnedAt(5))); + EXPECT_EQ(backend->landPendingDelete(0), Backend::RawRemoval::Mismatch); + + const auto after = markerOf(*backend, store->layout(), ref); + ASSERT_TRUE(after.has_value()); + EXPECT_EQ(after->meta.condemn_round, 5u); +} + +TEST(CASGcMetaWriter, ConfirmedMetaDeleteLeavesAnUndecodableMarker) +{ + auto backend = std::make_shared(); + auto store = openPlainPool(backend); + const BlobRef ref = idOf("u7"); + const String key = store->layout().blobMetaKey(ref); + { + OperationForTest op(*backend); + ASSERT_TRUE(std::holds_alternative((*op).create(key, "not a marker", Retry::once()))); + } + + const uint64_t anomalies_before = metaWriteAnomalies(); + Gc gc(store, u128Of(kGcId)); + ASSERT_NO_THROW(runConfirmedMetaDelete(gc, ref, 3)); + + EXPECT_EQ(metaWriteAnomalies() - anomalies_before, 1u); + OperationForTest op(*backend); + const auto raw = (*op).read(key, Retry::standard()); + ASSERT_TRUE(raw.has_value()) << "the job deleted a marker it could not read"; + EXPECT_EQ(raw->bytes, "not a marker"); +} + +TEST(CASGcMetaWriter, NewLeaderDeletesAGraduatedEntry) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x5005); + const BlobRef x = gateRef(hash); + { + Gc gc_a(store, kGateLeaderA); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc_a, *backend, "tbl", 1, {hash}, c)); + ASSERT_EQ(runGateRound(store, gc_a).graduated, 1u); + ASSERT_TRUE(retiredEntryFor(*backend, layout, x)->delete_pending); + } + + ASSERT_NO_FATAL_FAILURE(handGcLeaseTo(*backend, layout, kGateLeaderB)); + Gc gc_b(store, kGateLeaderB); + const RoundReport report = runGateRound(store, gc_b); + + EXPECT_EQ(report.deleted, 1u); + EXPECT_FALSE(blobPresent(*backend, layout, x)); + EXPECT_FALSE(markerOf(*backend, layout, x).has_value()) << "the new leader's job left an orphan marker"; +} + +TEST(CASGcMetaWriter, GraduationAcceptsANewerRoundMarker) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x9009); + const BlobRef x = gateRef(hash); + Gc gc(store, kGateLeaderA); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, c)); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, layout, x, condemnedAt(c + 3))); + + const uint64_t carries_before = unconfirmedCarries(); + backend->resetCounts(); + EXPECT_EQ(runGateRound(store, gc).graduated, 1u); + EXPECT_EQ(unconfirmedCarries() - carries_before, 0u); + EXPECT_EQ(backend->writeCount(layout.blobMetaKey(x)), 0u) << "an accepted gate writes nothing"; + + EXPECT_EQ(runGateRound(store, gc).redeleted, 1u); + EXPECT_FALSE(blobPresent(*backend, layout, x)); + const auto marker = markerOf(*backend, layout, x); + ASSERT_TRUE(marker.has_value()) << "the job for round " << c << " deleted a newer marker"; + EXPECT_EQ(marker->meta.condemn_round, c + 3); +} + +namespace +{ + +enum class ReleaseLeaderA +{ + BeforeGraduation, + AfterGraduation, +}; + +/// A deposed leader removes incarnation T0 and parks before its marker delete. Its successor finishes T0, +/// condemns the next incarnation T1 of the same content, and graduates it. The deposed leader's delayed job +/// must leave T1's marker, so a writer arriving after the delete round's cut republishes instead of adopting. +void staleMetaDeleteSparesSuccessorMarker(ReleaseLeaderA release) +{ + auto parking = std::make_shared(); + auto backend = std::make_shared(); + auto store = openGatePool(backend, /*io_concurrency*/ 1, /*graduation_budget*/ 0, parkingHook(parking)); + const Layout & layout = store->layout(); + const String ns = "test/tbl"; + const String payload = "stale-meta-delete-payload"; + const BlobRef x = idOf(payload); + parking->target = x; + + ASSERT_EQ(writeOnePart(store, ns, "p0", payload), BlobMaterializationAction::Published); + store->dropRef(RootNamespace{ns}, "p0"); + store->renewWatermarkOnce(); + Gc gc_a(store, kGateLeaderA); + ASSERT_NO_FATAL_FAILURE(runRoundsUntil(store, gc_a, [&] + { + const auto entry = retiredEntryFor(*backend, layout, x); + return entry && entry->delete_pending; + })); + const uint64_t c0 = retiredEntryFor(*backend, layout, x)->condemn_round; + + /// A's commit `CAS` loses once B has the lease; whether its round throws or returns, only the committed + /// round is asserted. + parking->armed.store(true); + std::exception_ptr leader_a_failure; + std::thread leader_a([&] + { + try + { + runGateRound(store, gc_a); + } + catch (...) + { + leader_a_failure = std::current_exception(); + } + }); + SCOPE_EXIT({ + parking->barrier.release(); + if (leader_a.joinable()) + leader_a.join(); + }); + parking->barrier.waitUntilArrived(); + ASSERT_FALSE(blobPresent(*backend, layout, x)) << "A removed T0 before parking"; + + ASSERT_NO_FATAL_FAILURE(handGcLeaseTo(*backend, layout, kGateLeaderB)); + Gc gc_b(store, kGateLeaderB); + runGateRound(store, gc_b); + ASSERT_FALSE(retiredEntryFor(*backend, layout, x).has_value()); + ASSERT_FALSE(markerOf(*backend, layout, x).has_value()) << "B's own job removes T0's marker"; + + ASSERT_EQ(writeOnePart(store, ns, "p1", payload), BlobMaterializationAction::Published); + store->dropRef(RootNamespace{ns}, "p1"); + store->renewWatermarkOnce(); + ASSERT_NO_FATAL_FAILURE(runRoundsUntil(store, gc_b, [&] { return retiredEntryFor(*backend, layout, x).has_value(); })); + const uint64_t c1 = retiredEntryFor(*backend, layout, x)->condemn_round; + ASSERT_GE(c1, c0 + 2); + + const auto releaseLeaderA = [&] + { + parking->barrier.release(); + leader_a.join(); + }; + if (release == ReleaseLeaderA::BeforeGraduation) + releaseLeaderA(); + ASSERT_EQ(runGateRound(store, gc_b).graduated, 1u); + if (release == ReleaseLeaderA::AfterGraduation) + releaseLeaderA(); + ASSERT_EQ(committedGcRound(*backend, layout), c1 + 1) << "A's attempt committed after losing its lease"; + + const auto marker = markerOf(*backend, layout, x); + ASSERT_TRUE(marker.has_value()) << "A's delayed job removed the marker of T1"; + EXPECT_EQ(marker->meta.state, MetaState::Condemned); + EXPECT_EQ(marker->meta.condemn_round, c1); + + std::optional w2; + backend->armOnHead(layout.blobKey(x), [&] { w2 = writeOnePart(store, ns, "p2", payload); }); + const RoundReport report = runGateRound(store, gc_b); + ASSERT_TRUE(w2.has_value()) << "the redelete never observed the blob"; + EXPECT_EQ(*w2, BlobMaterializationAction::Published) << "W2 adopted the incarnation B deletes"; + EXPECT_EQ(report.replaced, 1u); + ASSERT_TRUE(blobPresent(*backend, layout, x)); + OperationForTest op(*backend); + const auto body = (*op).read(layout.blobKey(x), Retry::standard()); + ASSERT_TRUE(body.has_value()); + EXPECT_TRUE(body->bytes.ends_with(payload)) << "W2's part names a blob that no longer holds its content"; +} + +} + +TEST(CASGcMetaWriter, StaleMetaDeleteSparesSuccessorMarkerBeforeGraduation) +{ + staleMetaDeleteSparesSuccessorMarker(ReleaseLeaderA::BeforeGraduation); +} + +TEST(CASGcMetaWriter, StaleMetaDeleteSparesSuccessorMarkerAfterGraduation) +{ + staleMetaDeleteSparesSuccessorMarker(ReleaseLeaderA::AfterGraduation); +} + +TEST(CASGcMetaWriter, GraduationReadsStayWithinTheBudget) +{ + for (const uint64_t concurrency : std::initializer_list{1, 8}) + { + SCOPED_TRACE(concurrency); + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency, /*graduation_budget*/ 1); + const Layout & layout = store->layout(); + const std::vector hashes{DB::UInt128(0x2001), DB::UInt128(0x2002), DB::UInt128(0x2003)}; + { + Gc gc(store, kGateLeaderA); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, hashes, c)); + } + + /// A fresh `Gc`: nothing in process can stand in for a read. + Gc gc(store, kGateLeaderA); + backend->resetCounts(); + const uint64_t wasted_before = readAheadWasted(); + EXPECT_EQ(runGateRound(store, gc).graduated, 1u); + + uint64_t gets = 0; + for (const UInt128 & hash : hashes) + gets += backend->getCount(layout.blobMetaKey(gateRef(hash))); + EXPECT_EQ(gets, 1u) << "the gate read entries the budget did not admit"; + EXPECT_EQ(readAheadWasted() - wasted_before, 0u); + } +} + +TEST(CASGcMetaWriter, GraduationRereadsTheMarkerOwnMarker) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x1001); + const BlobRef x = gateRef(hash); + Gc gc(store, kGateLeaderA); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, c)); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, layout, x, cleanMarker())); + + const uint64_t carries_before = unconfirmedCarries(); + EXPECT_EQ(runGateRound(store, gc).graduated, 0u) << "graduated over a Clean marker"; + EXPECT_EQ(unconfirmedCarries() - carries_before, 1u); + const auto marker = markerOf(*backend, layout, x); + ASSERT_TRUE(marker.has_value()); + EXPECT_EQ(marker->meta.state, MetaState::Condemned); + EXPECT_EQ(marker->meta.condemn_round, c + 1) << "the retry stamps the round that refused"; + + EXPECT_EQ(runGateRound(store, gc).graduated, 1u); +} + +TEST(CASGcMetaWriter, GraduationRereadsTheMarkerLostAttemptMarker) +{ + auto parking = std::make_shared(); + auto backend = std::make_shared(); + auto store = openGatePool(backend, /*io_concurrency*/ 1, /*graduation_budget*/ 0, parkingHook(parking)); + const Layout & layout = store->layout(); + const UInt128 x_hash = DB::UInt128(0x1101); + const UInt128 y_hash = DB::UInt128(0x1102); + const BlobRef x = gateRef(x_hash); + parking->target = gateRef(y_hash); + + /// X is owned throughout; Y is condemned and graduated, so A's next round both redeletes Y (and parks on it) + /// and condemns X. + Gc gc_a(store, kGateLeaderA); + const ManifestRef x_manifest = publishBlobs(*backend, layout, "x", 1, {x_hash}); + uint64_t y_round = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc_a, *backend, "y", 2, {y_hash}, y_round)); + ASSERT_EQ(runGateRound(store, gc_a).graduated, 1u); + dropBlobs(*backend, layout, "x", x_manifest); + store->renewWatermarkOnce(); + const uint64_t c = committedGcRound(*backend, layout) + 1; + + parking->armed.store(true); + std::exception_ptr leader_a_failure; + std::thread leader_a([&] + { + try + { + runGateRound(store, gc_a); + } + catch (...) + { + leader_a_failure = std::current_exception(); + } + }); + SCOPE_EXIT({ + parking->barrier.release(); + if (leader_a.joinable()) + leader_a.join(); + }); + parking->barrier.waitUntilArrived(); + ASSERT_NO_FATAL_FAILURE(awaitCondition([&] + { + const auto marker = markerOf(*backend, layout, x); + return marker && marker->meta.state == MetaState::Condemned; + })); + + /// B commits round c on its own condemnation of X, and the marker A's losing attempt wrote goes. + ASSERT_NO_FATAL_FAILURE(handGcLeaseTo(*backend, layout, kGateLeaderB)); + { + Gc gc_b(store, kGateLeaderB); + runGateRound(store, gc_b); + } + ASSERT_EQ(committedGcRound(*backend, layout), c); + ASSERT_EQ(retiredEntryFor(*backend, layout, x)->condemn_round, c); + ASSERT_NO_FATAL_FAILURE(removeMarker(*backend, layout, x)); + + parking->barrier.release(); + leader_a.join(); + ASSERT_EQ(committedGcRound(*backend, layout), c) << "A's losing attempt committed"; + + ASSERT_NO_FATAL_FAILURE(handGcLeaseTo(*backend, layout, kGateLeaderA)); + const uint64_t carries_before = unconfirmedCarries(); + EXPECT_EQ(runGateRound(store, gc_a).graduated, 0u) << "A graduated on its own attempt's marker, which is gone"; + EXPECT_EQ(unconfirmedCarries() - carries_before, 1u); + const auto marker = markerOf(*backend, layout, x); + ASSERT_TRUE(marker.has_value()); + EXPECT_EQ(marker->meta.condemn_round, c + 1); + EXPECT_EQ(runGateRound(store, gc_a).graduated, 1u); +} + +TEST(CASGcMetaWriter, GraduationAndDeleteRequestCounts) +{ + for (const uint64_t concurrency : std::initializer_list{1, 8}) + { + SCOPED_TRACE(concurrency); + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x4004); + const String meta_key = layout.blobMetaKey(gateRef(hash)); + Gc gc(store, kGateLeaderA); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, c)); + + backend->resetCounts(); + ASSERT_EQ(runGateRound(store, gc).graduated, 1u); + EXPECT_EQ(backend->getCount(meta_key), 1u) << "graduation reads the marker"; + EXPECT_EQ(backend->writeCount(meta_key), 0u); + EXPECT_EQ(backend->deleteCount(meta_key), 0u); + + backend->resetCounts(); + ASSERT_EQ(runGateRound(store, gc).redeleted, 1u); + EXPECT_EQ(backend->getCount(meta_key), 1u) << "a pending entry passes no gate; only its job reads"; + EXPECT_EQ(backend->deleteCount(meta_key), 1u); + EXPECT_FALSE(markerOf(*backend, layout, gateRef(hash)).has_value()); + } +} + +TEST(CASGcMetaWriter, LatePublisherCleanMissesTheRetriedMarker) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend); + const Layout & layout = store->layout(); + const String ns = "test/tbl"; + const String payload = "late-publisher-clean"; + const BlobRef x = idOf(payload); + + ASSERT_EQ(writeOnePart(store, ns, "p1", payload), BlobMaterializationAction::Published); + store->dropRef(RootNamespace{ns}, "p1"); + store->renewWatermarkOnce(); + Gc gc(store, kGateLeaderA); + ASSERT_NO_FATAL_FAILURE(runRoundsUntil(store, gc, [&] { return retiredEntryFor(*backend, layout, x).has_value(); })); + const uint64_t c = retiredEntryFor(*backend, layout, x)->condemn_round; + const auto condemned = markerOf(*backend, layout, x); + ASSERT_TRUE(condemned && condemned->meta.state == MetaState::Condemned); + const Etag m = condemned->etag; + const BlobMeta clean{.state = MetaState::Clean, .condemn_round = 0, .size = condemned->meta.size}; + + /// The publisher's reissued `Clean If-Match M` lands; its first attempt is still in flight. + { + OperationForTest op(*backend); + ASSERT_TRUE(std::holds_alternative(casMeta(*op, layout, x, m, clean))); + } + + const uint64_t carries_before = unconfirmedCarries(); + EXPECT_EQ(runGateRound(store, gc).graduated, 0u) << "graduated over the publisher's Clean"; + EXPECT_EQ(unconfirmedCarries() - carries_before, 1u); + const auto retried = markerOf(*backend, layout, x); + ASSERT_TRUE(retried.has_value()); + EXPECT_EQ(retried->meta.condemn_round, c + 1); + EXPECT_NE(retried->etag, m) << "the retry recreated the bytes the late request names"; + EXPECT_EQ(runGateRound(store, gc).graduated, 1u); + + /// The first attempt lands now: its precondition names bytes no marker carries. + { + OperationForTest op(*backend); + EXPECT_FALSE(std::holds_alternative(casMeta(*op, layout, x, m, clean))); + } + + std::optional w2; + backend->armOnHead(layout.blobKey(x), [&] { w2 = writeOnePart(store, ns, "p2", payload); }); + const RoundReport report = runGateRound(store, gc); + ASSERT_TRUE(w2.has_value()) << "the redelete never observed the blob"; + EXPECT_EQ(*w2, BlobMaterializationAction::Published) << "W2 adopted the incarnation GC deletes"; + EXPECT_EQ(report.replaced, 1u); + EXPECT_TRUE(blobPresent(*backend, layout, x)); +} + +TEST(CASGcMetaWriter, GraduationCarriesAnUndecodableMarker) +{ + for (const uint64_t concurrency : std::initializer_list{1, 8}) + { + SCOPED_TRACE(concurrency); + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x1010); + const BlobRef x = gateRef(hash); + Gc gc(store, kGateLeaderA); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, c)); + { + OperationForTest op(*backend); + const String key = layout.blobMetaKey(x); + const auto current = (*op).read(key, Retry::standard()); + ASSERT_TRUE(current.has_value()); + ASSERT_TRUE(std::holds_alternative((*op).replace(key, "not a marker", current->etag, Retry::once()))); + } + + const uint64_t anomalies_before = metaWriteAnomalies(); + const uint64_t carries_before = unconfirmedCarries(); + RoundReport report; + ASSERT_NO_THROW(report = runGateRound(store, gc)); + EXPECT_EQ(report.graduated, 0u) << "graduated on a marker nobody could read"; + EXPECT_EQ(metaWriteAnomalies() - anomalies_before, 2u) << "the gate's read and the retry's read"; + EXPECT_EQ(unconfirmedCarries() - carries_before, 1u); + EXPECT_EQ(committedGcRound(*backend, layout), c + 1) << "the round committed"; + EXPECT_FALSE(retiredEntryFor(*backend, layout, x)->delete_pending); + EXPECT_TRUE(blobPresent(*backend, layout, x)); + } +} + +TEST(CASGcMetaWriter, CondemnMarkerWriteRestampsAnOlderRound) +{ + auto backend = std::make_shared(); + auto store = openPlainPool(backend); + const BlobRef ref = idOf("u4"); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, store->layout(), ref, condemnedAt(3, 64))); + + Gc gc(store, u128Of(kGcId)); + gc.metaWriterForTest().scheduleCondemnMarkerWrite(ref, /*condemn_round=*/5, /*size=*/64); + gc.metaWriterForTest().drain(); + + const auto after = markerOf(*backend, store->layout(), ref); + ASSERT_TRUE(after.has_value()); + EXPECT_EQ(after->meta.state, MetaState::Condemned); + EXPECT_EQ(after->meta.condemn_round, 5u) << "an older-round marker may be another incarnation's"; +} + +TEST(CASGcMetaWriter, CondemnMarkerWriteLeavesAnEqualOrNewerRound) +{ + auto backend = std::make_shared(); + auto store = openPlainPool(backend); + const BlobRef equal = idOf("u5-equal"); + const BlobRef newer = idOf("u5-newer"); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, store->layout(), equal, condemnedAt(5, 64))); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, store->layout(), newer, condemnedAt(7, 64))); + const auto equal_before = markerOf(*backend, store->layout(), equal); + const auto newer_before = markerOf(*backend, store->layout(), newer); + + Gc gc(store, u128Of(kGcId)); + gc.metaWriterForTest().scheduleCondemnMarkerWrite(equal, /*condemn_round=*/5, /*size=*/64); + gc.metaWriterForTest().scheduleCondemnMarkerWrite(newer, /*condemn_round=*/5, /*size=*/64); + gc.metaWriterForTest().drain(); + + EXPECT_EQ(markerOf(*backend, store->layout(), equal)->etag, equal_before->etag); + EXPECT_EQ(markerOf(*backend, store->layout(), newer)->etag, newer_before->etag); + EXPECT_EQ(markerOf(*backend, store->layout(), newer)->meta.condemn_round, 7u); +} + +TEST(CASGcMetaWriter, GraduationRefusesAnOlderRoundMarker) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x2202); + const BlobRef x = gateRef(hash); + Gc gc(store, kGateLeaderA); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, c)); + ASSERT_GE(c, 2u); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, layout, x, condemnedAt(c - 1))); + + EXPECT_EQ(runGateRound(store, gc).graduated, 0u) << "accepted an older-round marker"; + const auto restamped = markerOf(*backend, layout, x); + ASSERT_TRUE(restamped.has_value()); + EXPECT_EQ(restamped->meta.condemn_round, c + 1) << "the retry left the older round in place"; + + EXPECT_EQ(runGateRound(store, gc).graduated, 1u); + EXPECT_EQ(runGateRound(store, gc).redeleted, 1u); + EXPECT_FALSE(blobPresent(*backend, layout, x)); + const auto orphan = markerOf(*backend, layout, x); + ASSERT_TRUE(orphan.has_value()) << "the job for round " << c << " removed a newer marker"; + EXPECT_EQ(orphan->meta.condemn_round, c + 1); +} + +TEST(CASGcMetaWriter, SparedThenRecondemnedBlobGraduates) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x3303); + const BlobRef x = gateRef(hash); + Gc gc(store, kGateLeaderA); + uint64_t r1 = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, r1)); + + /// A new owner spares the entry; GC never clears a marker, so `Condemned(r1)` stays. + const ManifestRef again = publishBlobs(*backend, layout, "tbl2", 2, {hash}); + store->renewWatermarkOnce(); + runGateRound(store, gc); + ASSERT_FALSE(retiredEntryFor(*backend, layout, x).has_value()); + ASSERT_EQ(markerOf(*backend, layout, x)->meta.condemn_round, r1); + + dropBlobs(*backend, layout, "tbl2", again); + store->renewWatermarkOnce(); + runGateRound(store, gc); + const auto recondemned = retiredEntryFor(*backend, layout, x); + ASSERT_TRUE(recondemned.has_value()); + ASSERT_GT(recondemned->condemn_round, r1); + + ASSERT_NO_FATAL_FAILURE(runRoundsUntil(store, gc, [&] + { + const auto entry = retiredEntryFor(*backend, layout, x); + return entry && entry->delete_pending; + }, /*max_rounds*/ 2)) << "a spared, re-condemned blob never graduates"; +} diff --git a/src/Disks/tests/gtest_cas_gc_read_ahead.cpp b/src/Disks/tests/gtest_cas_gc_read_ahead.cpp index 8a3cc408a8cf0..7bfeb46ad03ad 100644 --- a/src/Disks/tests/gtest_cas_gc_read_ahead.cpp +++ b/src/Disks/tests/gtest_cas_gc_read_ahead.cpp @@ -6,22 +6,27 @@ #include #include #include +#include #include #include #include #include +#include #include #include #include +#include #include #include #include #include #include +#include #include #include +#include namespace CurrentMetrics { @@ -35,6 +40,7 @@ using DB::Cas::tests::CountingBackend; using DB::Cas::tests::idOf; using DB::Cas::tests::openRequestsForTest; using DB::Cas::tests::u128Of; +using namespace DB::Cas::tests; namespace { @@ -551,3 +557,698 @@ TEST(CASGCReadAhead, WorkerReadFaultFailsTheRoundAndTheNextRoundRecovers) const RoundReport recovered = gc.runRegularRound(); EXPECT_TRUE(recovered.acquired_lease); } + +namespace +{ + +/// What one graduation round read and wrote for a cohort of ten condemned blobs. +struct GraduationRoundCounts +{ + uint64_t graduated = 0; + uint64_t gets = 0; /// `.meta` GETs of the cohort: the gate's and the retries' + uint64_t writes = 0; /// `.meta` writes of the cohort: the retries' + uint64_t wasted = 0; + std::map gets_by_key; +}; + +enum class CohortTwist +{ + None, + TwoClean, /// the two smallest refs read `Clean` at the gate + OneReReferenced, /// the smallest ref gains an owner before the round + Suppressed, /// the round's universe is not authoritative +}; + +/// Condemns ten blobs, rebuilds the `Gc` so nothing in process stands in for a read, applies `twist`, and +/// measures the round due to graduate them under a graduation budget of three. +void measureGraduationRound(uint64_t concurrency, CohortTwist twist, GraduationRoundCounts & out) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency, /*graduation_budget*/ 3); + const Layout & layout = store->layout(); + std::vector hashes; + for (uint64_t i = 1; i <= 10; ++i) + hashes.push_back(DB::UInt128(0x5000 + i)); + { + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "cohort", 1, hashes, c)); + } + + std::vector refs; + for (const UInt128 & hash : hashes) + refs.push_back(gateRef(hash)); + std::sort(refs.begin(), refs.end()); + if (twist == CohortTwist::TwoClean) + for (size_t i = 0; i < 2; ++i) + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, layout, refs[i], cleanMarker())); + if (twist == CohortTwist::OneReReferenced) + publishBlobs(*backend, layout, "again", 2, {refs[0].digest.toU128()}); + store->renewWatermarkOnce(); + + Gc gc(store, kGc); + backend->resetCounts(); + const uint64_t wasted_before = readAheadWasted(); + const RoundReport report = runGateRound(store, gc, + twist == CohortTwist::Suppressed ? UniversePolicy::StageA_Suppressed : UniversePolicy::Authoritative); + + out.graduated = report.graduated; + out.wasted = readAheadWasted() - wasted_before; + for (const BlobRef & ref : refs) + { + const String key = layout.blobMetaKey(ref); + out.gets += backend->getCount(key); + out.writes += backend->writeCount(key); + out.gets_by_key[key] = backend->getCount(key); + } +} + +} + +TEST(CASGCReadAhead, GraduationReadAheadRequestCounts) +{ + struct Expected + { + CohortTwist twist; + uint64_t graduated; + uint64_t gets; + uint64_t writes; + }; + for (const Expected & expected : {Expected{CohortTwist::None, 3, 3, 0}, + Expected{CohortTwist::TwoClean, 3, 7, 2}, + Expected{CohortTwist::OneReReferenced, 3, 3, 0}, + Expected{CohortTwist::Suppressed, 0, 0, 0}}) + { + SCOPED_TRACE(static_cast(expected.twist)); + GraduationRoundCounts sequential; + GraduationRoundCounts ahead; + ASSERT_NO_FATAL_FAILURE(measureGraduationRound(1, expected.twist, sequential)); + ASSERT_NO_FATAL_FAILURE(measureGraduationRound(8, expected.twist, ahead)); + + EXPECT_EQ(ahead.graduated, expected.graduated); + EXPECT_EQ(ahead.gets, expected.gets) << "5 gate reads and 2 retry reads when two refuse; 3 otherwise"; + EXPECT_EQ(ahead.writes, expected.writes); + EXPECT_EQ(ahead.wasted, 0u); + EXPECT_EQ(ahead.gets_by_key, sequential.gets_by_key); + EXPECT_EQ(ahead.graduated, sequential.graduated); + EXPECT_EQ(ahead.writes, sequential.writes); + } +} + +namespace +{ + +/// Journals every GET in arrival order, so a test can place the gate's marker read after the round's cut. +class GetJournalBackend : public CountingBackend +{ +public: + std::optional read(const String & key, TransportAccess & access) override + { + { + std::lock_guard lock(journal_mutex); + journal.push_back(key); + } + return CountingBackend::read(key, access); + } + + std::vector takeJournal() + { + std::lock_guard lock(journal_mutex); + return std::exchange(journal, {}); + } + +private: + std::mutex journal_mutex; + std::vector journal; +}; + +/// Once armed, rewrites `key` with `replacement` right after a GET of it returns, with no backend lock held: +/// the value the reader got is stale by the time it arrives. +class ReadCompletionHookBackend : public CountingBackend +{ +public: + void armAfterRead(const String & key, String replacement) + { + std::lock_guard lock(hook_mutex); + hooked_key = key; + replacement_bytes = std::move(replacement); + armed = true; + } + + std::optional read(const String & key, TransportAccess & access) override + { + auto raw = CountingBackend::read(key, access); + std::optional rewrite; + { + std::lock_guard lock(hook_mutex); + if (armed && raw && key == hooked_key) + { + armed = false; + rewrite = replacement_bytes; + } + } + if (rewrite) + (void)InMemoryBackend::write(key, *rewrite, raw->value, access); + return raw; + } + +private: + std::mutex hook_mutex; + String hooked_key; + String replacement_bytes; + bool armed = false; +}; + +} + +TEST(CASGCReadAhead, GraduationReadAheadReadsEachRound) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend, /*io_concurrency*/ 8); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x3001); + const BlobRef x = gateRef(hash); + const String meta_key = layout.blobMetaKey(x); + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, c)); + ASSERT_NO_FATAL_FAILURE(setMarker(*backend, layout, x, cleanMarker())); + + /// An unrelated publication gives each measured round ref-log reads to order the gate's read against. + uint64_t noise = 100; + const auto roundAfterPublication = [&] + { + ++noise; + publishBlobs(*backend, layout, "noise" + std::to_string(noise), noise, {DB::UInt128(0x3100 + noise)}); + store->renewWatermarkOnce(); + backend->resetCounts(); + backend->takeJournal(); + const RoundReport report = runGateRound(store, gc); + return std::make_pair(report, backend->takeJournal()); + }; + const auto expectReadAfterCut = [&](const std::vector & journal) + { + const auto meta_at = std::find(journal.begin(), journal.end(), meta_key); + ASSERT_NE(meta_at, journal.end()); + const auto last_log = std::find_if(journal.rbegin(), journal.rend(), + [](const String & key) { return key.contains("/_log/"); }); + ASSERT_NE(last_log, journal.rend()) << "the round read no ref log"; + const size_t last_log_at = static_cast(journal.rend() - last_log) - 1; + EXPECT_LT(last_log_at, static_cast(meta_at - journal.begin())) + << "the gate read the marker before its round's cut"; + }; + + const auto [refused, refused_journal] = roundAfterPublication(); + EXPECT_EQ(refused.graduated, 0u); + EXPECT_EQ(backend->getCount(meta_key), 2u) << "the gate's read and the retry's read"; + EXPECT_EQ(backend->writeCount(meta_key), 1u); + expectReadAfterCut(refused_journal); + const auto marker = markerOf(*backend, layout, x); + ASSERT_TRUE(marker.has_value()); + EXPECT_EQ(marker->meta.condemn_round, c + 1); + + const auto [accepted, accepted_journal] = roundAfterPublication(); + EXPECT_EQ(accepted.graduated, 1u); + EXPECT_EQ(backend->getCount(meta_key), 1u) << "every round reads the marker again"; + expectReadAfterCut(accepted_journal); +} + +TEST(CASGCReadAhead, GraduationReadAheadDecidesOnTheFetchedValue) +{ + for (const uint64_t concurrency : std::initializer_list{1, 8}) + { + SCOPED_TRACE(concurrency); + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x3101); + const BlobRef x = gateRef(hash); + const String meta_key = layout.blobMetaKey(x); + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, c)); + + backend->resetCounts(); + backend->armAfterRead(meta_key, encodeBlobMeta(cleanMarker())); + EXPECT_EQ(runGateRound(store, gc).graduated, 1u) << "the gate did not decide on the value it fetched"; + EXPECT_EQ(backend->getCount(meta_key), 1u); + const auto after = markerOf(*backend, layout, x); + ASSERT_TRUE(after.has_value()); + EXPECT_EQ(after->meta.state, MetaState::Clean) << "the hook never fired"; + } +} + +TEST(CASGCReadAhead, GraduationReadAheadSurfacesAReadError) +{ + for (const uint64_t concurrency : std::initializer_list{1, 8}) + { + SCOPED_TRACE(concurrency); + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency); + const Layout & layout = store->layout(); + const UInt128 hash = DB::UInt128(0x3201); + const String meta_key = layout.blobMetaKey(gateRef(hash)); + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "tbl", 1, {hash}, c)); + + backend->failNextReadWith(meta_key, std::make_exception_ptr(std::runtime_error("injected marker read fault"))); + const uint64_t anomalies_before = metaWriteAnomalies(); + const uint64_t carries_before = unconfirmedCarries(); + RoundReport report; + ASSERT_NO_THROW(report = runGateRound(store, gc)); + EXPECT_EQ(report.graduated, 0u) << "graduated without reading the marker"; + EXPECT_EQ(metaWriteAnomalies() - anomalies_before, 1u); + EXPECT_EQ(unconfirmedCarries() - carries_before, 1u); + EXPECT_EQ(committedGcRound(*backend, layout), c + 1) << "a failed marker read failed the round"; + EXPECT_EQ(runGateRound(store, gc).graduated, 1u); + } +} + +namespace +{ + +/// Parks the first GET of a watched key until `quorum` more GETs of watched keys arrive, or a bounded wait ends. +/// Serial reads never release it; overlapping ones do. +class ParkFirstWatchedReadBackend : public CountingBackend +{ +public: + void watch(std::set keys, size_t quorum_) + { + std::lock_guard lock(park_mutex); + watched = std::move(keys); + quorum = quorum_; + armed = true; + } + + bool timedOut() const + { + std::lock_guard lock(park_mutex); + return timed_out; + } + + std::optional read(const String & key, TransportAccess & access) override + { + { + std::unique_lock lock(park_mutex); + if (armed && watched.contains(key)) + { + if (!parked) + { + parked = true; + if (!arrived.wait_for(lock, std::chrono::seconds(10), [&] { return arrivals >= quorum; })) + timed_out = true; + armed = false; + } + else + { + ++arrivals; + arrived.notify_all(); + } + } + } + return CountingBackend::read(key, access); + } + +private: + mutable std::mutex park_mutex; + std::condition_variable arrived; + std::set watched; + size_t quorum = 0; + size_t arrivals = 0; + bool armed = false; + bool parked = false; + bool timed_out = false; +}; + +std::vector hashRange(uint64_t base, uint64_t count) +{ + std::vector out; + for (uint64_t i = 1; i <= count; ++i) + out.push_back(DB::UInt128(base + i)); + return out; +} + +/// `.meta` GETs per key over `hashes`. +std::map markerGets(const CountingBackend & backend, const Layout & layout, const std::vector & hashes) +{ + std::map out; + for (const UInt128 & hash : hashes) + out[layout.blobMetaKey(gateRef(hash))] = backend.getCount(layout.blobMetaKey(gateRef(hash))); + return out; +} + +struct SupersedeRun +{ + std::vector graduated; + std::map superseded_gets; + uint64_t wasted = 0; +}; + +/// Twenty condemned blobs; before each of three rounds a writer republishes ten of them under a new token and +/// drops them again, so those ten are superseded every round. +void runSupersedeRounds(uint64_t concurrency, SupersedeRun & out) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency); + const Layout & layout = store->layout(); + const std::vector cohort = hashRange(0x6000, 20); + const std::vector superseded(cohort.begin() + 10, cohort.end()); + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "cohort", 1, cohort, c)); + + backend->resetCounts(); + const uint64_t wasted_before = readAheadWasted(); + for (uint64_t round = 0; round < 3; ++round) + { + for (const UInt128 & hash : superseded) + displaceBlobToken(*backend, layout, gateRef(hash)); + const String ref_name = "touch" + std::to_string(round); + const ManifestRef touch = publishBlobs(*backend, layout, ref_name, 10 + round, superseded); + dropBlobs(*backend, layout, ref_name, touch); + store->renewWatermarkOnce(); + out.graduated.push_back(runGateRound(store, gc).graduated); + } + out.wasted = readAheadWasted() - wasted_before; + out.superseded_gets = markerGets(*backend, layout, superseded); +} + +} + +TEST(CASGCReadAhead, GraduationReadAheadSkipsSupersededEntries) +{ + SupersedeRun sequential; + SupersedeRun ahead; + ASSERT_NO_FATAL_FAILURE(runSupersedeRounds(1, sequential)); + ASSERT_NO_FATAL_FAILURE(runSupersedeRounds(8, ahead)); + + EXPECT_EQ(ahead.superseded_gets, sequential.superseded_gets) << "a superseded entry's marker was read ahead"; + EXPECT_EQ(ahead.wasted, 0u); + EXPECT_EQ(ahead.graduated, sequential.graduated); + EXPECT_EQ(sequential.graduated.front(), 10u); +} + +TEST(CASGCReadAhead, GraduationReadAheadIgnoresUnusedHeadHints) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend, /*io_concurrency*/ 8); + const Layout & layout = store->layout(); + + /// Roles by key order: the smallest blob is condemned fresh in the measured round, so its HEAD tops the head + /// hints up with `kept` (which keep an edge, so nobody takes them) before any graduation candidate settles. + std::vector refs; + for (const UInt128 & hash : hashRange(0x7000, 61)) + refs.push_back(gateRef(hash)); + std::sort(refs.begin(), refs.end()); + const auto hashesOf = [&](size_t from, size_t to) + { + std::vector out; + for (size_t i = from; i < to; ++i) + out.push_back(refs[i].digest.toU128()); + return out; + }; + const std::vector fresh = hashesOf(0, 1); + const std::vector graduating = hashesOf(1, 21); + const std::vector kept = hashesOf(21, 61); + + const ManifestRef kept_a = publishBlobs(*backend, layout, "kept_a", 1, kept); + publishBlobs(*backend, layout, "kept_b", 2, kept); + { + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "graduating", 3, graduating, c)); + } + const ManifestRef fresh_manifest = publishBlobs(*backend, layout, "fresh", 4, fresh); + dropBlobs(*backend, layout, "fresh", fresh_manifest); + dropBlobs(*backend, layout, "kept_a", kept_a); + store->renewWatermarkOnce(); + + std::set graduation_keys; + for (const UInt128 & hash : graduating) + graduation_keys.insert(layout.blobMetaKey(gateRef(hash))); + backend->watch(graduation_keys, /*quorum*/ 8); + + Gc gc(store, kGc); + const uint64_t wasted_before = readAheadWasted(); + const RoundReport report = runGateRound(store, gc); + EXPECT_FALSE(backend->timedOut()) << "the graduation reads never overlapped"; + EXPECT_EQ(report.graduated, 20u); + EXPECT_EQ(report.condemned, 1u); + EXPECT_GE(readAheadWasted() - wasted_before, 31u) << "the fixture must leave the head window full of untaken hints"; +} + +TEST(CASGCReadAhead, GraduationReadAheadAcrossShards) +{ + std::vector> gets; + for (const uint64_t concurrency : std::initializer_list{1, 8}) + { + SCOPED_TRACE(concurrency); + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency, /*graduation_budget*/ 10, {}, /*gc_shards*/ 4); + const Layout & layout = store->layout(); + /// The shard is chosen by the digest's high 64 bits, so spread the cohort over them. + std::vector cohort = hashRange(0x8000, 40); + for (size_t i = 0; i < cohort.size(); ++i) + cohort[i] = (UInt128(i) << 64) + cohort[i]; + std::set shards; + for (const UInt128 & hash : cohort) + shards.insert(blobShard(gateRef(hash), 4)); + ASSERT_GT(shards.size(), 1u) << "the cohort must span shards"; + { + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "cohort", 1, cohort, c)); + } + + Gc gc(store, kGc); + backend->resetCounts(); + const uint64_t wasted_before = readAheadWasted(); + EXPECT_EQ(runGateRound(store, gc).graduated, 10u); + const auto by_key = markerGets(*backend, layout, cohort); + uint64_t total = 0; + for (const auto & [key, n] : by_key) + total += n; + EXPECT_EQ(total, 10u) << "the shards together read past the shared budget"; + EXPECT_EQ(readAheadWasted() - wasted_before, 0u); + gets.push_back(by_key); + } + EXPECT_EQ(gets[0], gets[1]); +} + +TEST(CASGCReadAhead, GraduationReadAheadReadsALargeCohortOnce) +{ + std::vector> gets; + for (const uint64_t concurrency : std::initializer_list{1, 8}) + { + SCOPED_TRACE(concurrency); + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency); + const Layout & layout = store->layout(); + const std::vector cohort = hashRange(0x8800, 100); + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "cohort", 1, cohort, c)); + + backend->resetCounts(); + const uint64_t wasted_before = readAheadWasted(); + EXPECT_EQ(runGateRound(store, gc).graduated, 100u); + const auto by_key = markerGets(*backend, layout, cohort); + for (const auto & [key, n] : by_key) + EXPECT_EQ(n, 1u) << key; + EXPECT_EQ(readAheadWasted() - wasted_before, 0u); + gets.push_back(by_key); + } + EXPECT_EQ(gets[0], gets[1]); +} + +namespace +{ + +/// Samples, at the start of every watched read, how many jobs the process's thread pools hold queued or +/// running beyond `baseline`: the read-ahead's outstanding hints. The pool has fewer threads than the window, +/// so requests inside the backend cannot show the bound; each read takes `latency` so the scan has issued +/// its hints before the first reads finish. +class OutstandingHintsBackend : public CountingBackend +{ +public: + void watch(std::set keys, std::chrono::microseconds latency_) + { + std::lock_guard lock(sample_mutex); + watched = std::move(keys); + latency = latency_; + baseline = CurrentMetrics::get(CurrentMetrics::LocalThreadScheduled); + armed = true; + } + + int64_t peakOutstanding() const + { + std::lock_guard lock(sample_mutex); + return peak; + } + + std::optional read(const String & key, TransportAccess & access) override + { + std::chrono::microseconds delay{0}; + { + std::lock_guard lock(sample_mutex); + if (armed && watched.contains(key)) + { + peak = std::max(peak, CurrentMetrics::get(CurrentMetrics::LocalThreadScheduled) - baseline); + delay = latency; + } + } + std::this_thread::sleep_for(delay); + return CountingBackend::read(key, access); + } + +private: + mutable std::mutex sample_mutex; + std::set watched; + std::chrono::microseconds latency{0}; + int64_t baseline = 0; + int64_t peak = 0; + bool armed = false; +}; + +} + +TEST(CASGCReadAhead, GraduationReadAheadHintsAtMostAWindowAhead) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend, /*io_concurrency*/ 8); + const Layout & layout = store->layout(); + const std::vector cohort = hashRange(0x8900, 200); + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "cohort", 1, cohort, c)); + + std::set meta_keys; + for (const UInt128 & hash : cohort) + meta_keys.insert(layout.blobMetaKey(gateRef(hash))); + backend->watch(meta_keys, std::chrono::milliseconds(2)); + + const uint64_t io_concurrency = 8; + const uint64_t window = 4 * io_concurrency; + EXPECT_EQ(runGateRound(store, gc).graduated, cohort.size()); + EXPECT_GT(backend->peakOutstanding(), 1) << "no read was ever hinted ahead of its gate"; + /// A worker still counts as scheduled between its set_value and its exit, after take() freed the slot and the + /// round hinted a replacement, so the sample may run one past the window per worker. + EXPECT_LE(backend->peakOutstanding(), static_cast(window + io_concurrency)) << "hints ran past the read-ahead window"; +} + +TEST(CASGCReadAhead, GraduationReadAheadNeverHintsPendingRows) +{ + std::vector> gets; + for (const uint64_t concurrency : std::initializer_list{1, 8}) + { + SCOPED_TRACE(concurrency); + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency); + const Layout & layout = store->layout(); + const std::vector eligible = hashRange(0x9100, 10); + const std::vector pending = hashRange(0x9200, 10); + const ManifestRef eligible_manifest = publishBlobs(*backend, layout, "eligible", 1, eligible); + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "pending", 2, pending, c)); + dropBlobs(*backend, layout, "eligible", eligible_manifest); + store->renewWatermarkOnce(); + const RoundReport setup = runGateRound(store, gc); + ASSERT_EQ(setup.graduated, 10u); + ASSERT_EQ(setup.condemned, 10u); + + backend->resetCounts(); + const uint64_t wasted_before = readAheadWasted(); + const RoundReport report = runGateRound(store, gc); + EXPECT_EQ(report.redeleted, 10u); + EXPECT_EQ(report.graduated, 10u); + EXPECT_EQ(readAheadWasted() - wasted_before, 0u) << "a pending row was hinted and never taken"; + auto by_key = markerGets(*backend, layout, pending); + by_key.merge(markerGets(*backend, layout, eligible)); + gets.push_back(by_key); + } + EXPECT_EQ(gets[0], gets[1]); +} + +TEST(CASGCReadAhead, GraduationReadAheadTouchedRowKeepsItsBudgetSlot) +{ + std::vector> gets; + for (const uint64_t concurrency : std::initializer_list{1, 8}) + { + SCOPED_TRACE(concurrency); + auto backend = std::make_shared(); + auto store = openGatePool(backend, concurrency, /*graduation_budget*/ 3); + const Layout & layout = store->layout(); + + /// A, T, B, C in key order. T is published and dropped again within one round under its own token: + /// net-zero deltas, so T is touched yet not superseded, and still reaches the gate. + std::vector refs; + for (const UInt128 & hash : hashRange(0x9300, 4)) + refs.push_back(gateRef(hash)); + std::sort(refs.begin(), refs.end()); + std::vector cohort; + for (const BlobRef & ref : refs) + cohort.push_back(ref.digest.toU128()); + + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "cohort", 1, cohort, c)); + const std::vector touched{cohort[1]}; + const ManifestRef touch = publishBlobs(*backend, layout, "touch", 10, touched); + dropBlobs(*backend, layout, "touch", touch); + store->renewWatermarkOnce(); + + backend->resetCounts(); + EXPECT_EQ(runGateRound(store, gc).graduated, 3u); + const auto by_key = markerGets(*backend, layout, cohort); + EXPECT_EQ(by_key.at(layout.blobMetaKey(refs[0])), 1u) << "A"; + EXPECT_EQ(by_key.at(layout.blobMetaKey(refs[1])), 1u) << "T"; + EXPECT_EQ(by_key.at(layout.blobMetaKey(refs[2])), 1u) << "B"; + EXPECT_EQ(by_key.at(layout.blobMetaKey(refs[3])), 0u) << "C"; + gets.push_back(by_key); + } + EXPECT_EQ(gets[0], gets[1]); +} + +TEST(CASGCReadAhead, OutstandingGraduationReadsDoNotStopHeadHints) +{ + auto backend = std::make_shared(); + auto store = openGatePool(backend, /*io_concurrency*/ 8); + const Layout & layout = store->layout(); + + /// Key order: five graduating rows, the fresh condemn, forty `kept` blobs (head candidates nobody takes), + /// then thirty-five more graduating rows. When the fresh HEAD is taken, graduation reads fill the window, + /// so a head throttle that counted them would hint nothing past the fresh blob. + std::vector refs; + for (const UInt128 & hash : hashRange(0xA000, 81)) + refs.push_back(gateRef(hash)); + std::sort(refs.begin(), refs.end()); + std::vector graduating; + std::vector fresh; + std::vector kept; + for (size_t i = 0; i < refs.size(); ++i) + { + const UInt128 hash = refs[i].digest.toU128(); + (i == 5 ? fresh : (i < 46 && i > 5 ? kept : graduating)).push_back(hash); + } + + const ManifestRef kept_a = publishBlobs(*backend, layout, "kept_a", 1, kept); + publishBlobs(*backend, layout, "kept_b", 2, kept); + { + Gc gc(store, kGc); + uint64_t c = 0; + ASSERT_NO_FATAL_FAILURE(condemnBlobs(store, gc, *backend, "graduating", 3, graduating, c)); + } + const ManifestRef fresh_manifest = publishBlobs(*backend, layout, "fresh", 4, fresh); + dropBlobs(*backend, layout, "fresh", fresh_manifest); + dropBlobs(*backend, layout, "kept_a", kept_a); + store->renewWatermarkOnce(); + + Gc gc(store, kGc); + const uint64_t wasted_before = readAheadWasted(); + const RoundReport report = runGateRound(store, gc); + EXPECT_EQ(report.graduated, 40u); + EXPECT_EQ(report.condemned, 1u); + EXPECT_GE(readAheadWasted() - wasted_before, 20u) << "outstanding graduation reads kept the head hints from being issued"; +} diff --git a/src/Disks/tests/gtest_cas_ref_writer.cpp b/src/Disks/tests/gtest_cas_ref_writer.cpp index 159b12fb99e98..bb34f6b198448 100644 --- a/src/Disks/tests/gtest_cas_ref_writer.cpp +++ b/src/Disks/tests/gtest_cas_ref_writer.cpp @@ -3114,7 +3114,6 @@ TEST(CASRefWriterSnapshotPublish, C4LatchBoundedUnderSustainedNonCommittedPublis << "the one dispatch above must itself have reissued more than once against the saturated " "backend before giving up at its retry window -- a single attempt would not distinguish " "this from a non-retrying policy"; - EXPECT_GT(clock->longestPause(), 0u) << "at least one of those reissues must have paced with a real backoff"; const auto dispatched_before = global_counters[ProfileEvents::CASRefSnapshotPublishDispatched].load(); for (int i = 0; i < 30; ++i) @@ -3257,7 +3256,6 @@ TEST(CASRefWriterSnapshotPublish, C4BackoffDefersThenRetriesAndPublishes) EXPECT_GT(clock->pauseCount(), 1u) << "the failing dispatch above must itself have reissued more than once before giving up and " "arming the backoff -- a single attempt would not distinguish this from a non-retrying policy"; - EXPECT_GT(clock->longestPause(), 0u) << "at least one of those reissues must have paced with a real backoff"; /// A read within the backoff window (frozen clock) must not re-dispatch. const auto d1 = global_counters[ProfileEvents::CASRefSnapshotPublishDispatched].load(); diff --git a/src/Disks/tests/gtest_cas_writer_duties.cpp b/src/Disks/tests/gtest_cas_writer_duties.cpp index 6706df7e0109d..45e24c1377fee 100644 --- a/src/Disks/tests/gtest_cas_writer_duties.cpp +++ b/src/Disks/tests/gtest_cas_writer_duties.cpp @@ -125,7 +125,6 @@ void driveToNetworkErrorGiveUp(LatchedChunkFaultBackend & backend, DB::Cas::test EXPECT_GE(clock.pauseCount() - pauses_before, 1u) << "a give-up after a single attempt cannot distinguish a retrying `standard` policy from one " << "that never reissues at all"; - EXPECT_GT(clock.longestPause(), 0u) << "at least one of the retry's pauses must be a real, nonzero backoff"; backend.disarm(); }