From e7b45306ba3438ff15fe3fdfac6c8be3fd6bae10 Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 6 Nov 2026 03:04:28 +0100 Subject: [PATCH 1/2] Fix early manifest deletion in CAS GC Port of https://github.com/Altinity/ClickHouse/pull/2730 (antalya-26.8) to antalya-26.6. Conflicts: `Gc/CasOrphanManifestSweep.cpp` (`buildPrefixEligibleUnderFloor` keeps the `min_active_build_sequence == max` sentinel test, which is what `MountLease::isCleanRelease` means here; `sweepNamespace` keeps the antalya-26.6 `mountRequests` admission with the renamed eligibility helper), `Tools/CasDecommission.cpp` (comment only). (cherry picked from commit 856ca70a4ffc168a3899e62014159413337c0484) Co-Authored-By: Claude Fable 5.1 Signed-off-by: Mikhail Filimonov --- .../cas/architecture/manifests-and-refs.md | 38 ++- .../ContentAddressed/Gc/CasGc.cpp | 4 +- .../Gc/CasOrphanManifestSweep.cpp | 112 ++++++--- .../Gc/CasOrphanManifestSweep.h | 14 +- .../Tools/CasDecommission.cpp | 9 +- .../ContentAddressed/Tools/CasFsck.cpp | 2 +- src/Disks/tests/cas_sweep_test_support.h | 47 ++++ src/Disks/tests/cas_test_helpers.h | 2 +- .../tests/gtest_cas_orphan_manifest_sweep.cpp | 218 +++++++++++++++++- .../tests/gtest_cas_orphan_sweep_requests.cpp | 215 ++++++++++++++++- src/Disks/tests/gtest_cas_writer_duties.cpp | 2 +- 11 files changed, 596 insertions(+), 67 deletions(-) diff --git a/docs/en/antalya/cas/architecture/manifests-and-refs.md b/docs/en/antalya/cas/architecture/manifests-and-refs.md index 6e6b1ed667bfa..55e9137a67b89 100644 --- a/docs/en/antalya/cas/architecture/manifests-and-refs.md +++ b/docs/en/antalya/cas/architecture/manifests-and-refs.md @@ -90,22 +90,50 @@ a live precommit would clamp `GC`'s fold barrier forever. ### The orphan-manifest sweep {#orphan-sweep} The cursor-paced, budgeted sweep has two stages (`Gc/CasOrphanManifestSweep.cpp`). During fold -planning, it freezes candidates with exact `GET`s, opens and decodes their bodies, and derives the -state changes needed to make deletion safe. Only after the round `CAS` adopts those changes does +planning, it checks candidates against the mount floor, ownership, fold coverage, and catalog +lifecycle. It then freezes accepted candidates with exact `GET`s, opens and decodes their bodies, +and derives the state changes needed to make deletion safe. Only after the round `CAS` adopts those changes does phase 18 perform physical deletion. Eligibility comes **exclusively** from the durable watermark in the mount lease — there is no age threshold and no time-based grace period anywhere in this protocol. No mount lease for the `server_root_id` means no deletion authority means nothing is swept for that root. +The page reads an initial catalog cut `K0`. The mount floor applies to every candidate, including +manifests whose namespace is absent from `K0`: a writer can stage a manifest before its first +`precommitAdd` creates the catalog row. An active build, or a build above a lagging published floor, +must therefore be retained even when no row exists. + +Ownership recovery and fold coverage use the row from `K0`. After collecting tentative candidates, +the page reads one more catalog cut, `K1`, after all floor observations and before any candidate body +is hinted or read. A candidate passes only if its namespace row is absent in both cuts or the full +`CatalogEntry` is equal in both. A row that appears, disappears, changes incarnation, or changes +lifecycle state retains that candidate. A change to another namespace does not block it. Recovery +never switches to a newer row during this page. + +The second cut costs one extra catalog `GET` per page with tentative candidates; pages with none, +including list-only pages, do not read it. The floor is read once per namespace per page. Candidates +retained by `K1` increment `skipped` without refilling the nomination budget or rolling back the +decision cursor. A read or decode error, or an ambiguous life index in `K1`, fails planning before +body reads and before the new cursor can be adopted. + ```mermaid flowchart TD - A["LIST one page of cas/manifests/
freeze candidates with exact GET"] --> B{"build-prefix eligible?
durable watermark fact only"} + A["LIST one page of cas/manifests/
read initial catalog K0"] --> B{"build-prefix eligible?
mount floor, including rowless namespaces"} B -->|"epoch less than lease epoch"| ELIG["eligible, old-epoch debris"] B -->|"same epoch, min_active_build_sequence clears build_seq"| ELIG B -->|"no lease, or epoch ahead, or build may be live"| SKIP["skip"] - ELIG --> C["protection view: committed manifests
plus live precommits
plus manifests with an unfolded minus-one"] + ELIG --> ROW{"namespace row in K0?"} + ROW -->|yes| C["protection view from K0:
committed manifests, live precommits,
and unfolded removal targets"] + ROW -->|no| T["tentative candidate"] C -->|"key protected"| SKIP2["skip"] - C -->|"not protected"| D{"open and decode
frozen body"} + C -->|"not protected"| P{"deletion premise:
fold coverage permits deletion?"} + P -->|no| SKIP2 + P -->|yes| T + T --> K["read and validate one post-floor
catalog K1 for all tentative candidates"] + K --> Q{"namespace row equal in K0 and K1
or absent in both?"} + Q -->|no| SKIP2 + Q -->|yes| GET["freeze candidate with exact GET"] + GET --> D{"open and decode
frozen body"} D -->|"cannot open or decode"| U["retain; skipped++ and undecodable++
log exact key; advance decision cursor
continue to later candidates"] D -->|decoded| I{"body ref and namespace
match key?"} I -->|no| BAD["CORRUPTED_DATA
fail-closed round error"] diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp index 568c354dcf96a..26c4786fe6db5 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp @@ -2987,7 +2987,7 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional & /*state_etag*/, /// clamps every round with no terminal resolution. Fail-closed: no durable watermark (the /// build cannot be proven dead) keeps the barrier. if (edge.change > 0 && edge.owner_kind == RefOwnerKind::Precommit - && prefixEligible(*store, ns, + && buildPrefixEligibleUnderFloor(*store, ns, BuildPrefix{.writer_epoch = edge.manifest_id.ref.writer_epoch, .build_sequence = edge.manifest_id.ref.build_sequence})) { @@ -4844,7 +4844,7 @@ RebuildReport Gc::rebuildBaseline(bool force) if (!parsed) return true; /// foreign key shape — debris const ManifestRef & mref = parsed->ref; - if (prefixEligible(*store, ns, BuildPrefix{mref.writer_epoch, mref.build_sequence})) + if (buildPrefixEligibleUnderFloor(*store, ns, BuildPrefix{mref.writer_epoch, mref.build_sequence})) return true; /// provably dead — the orphan sweep's territory, never an edge const ManifestId id{ns, mref}; if (foldManifestEdges(reads, id, +1, deltas, mf_cleanup_unused, /*txn_ordinal=*/0)) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp index 73786c18e63b7..1ef3ff7000ff5 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp @@ -45,8 +45,8 @@ void onGcEnumerationPage() /// mount there is no deletion authority, so the caller must leave the prefix untouched. The mount's /// `writer_epoch` and `min_active_build_sequence` are the single durable epoch/floor pair used for eligibility, including /// across process replacement and the retired sentinel. -std::optional floorForNamespace(CasOperation & op, const Layout & layout, const RootNamespace & ns, - uint64_t * reads = nullptr) +std::optional mountLeaseForNamespace(CasOperation & op, const Layout & layout, const RootNamespace & ns, + uint64_t * reads = nullptr) { const String & value = ns.string(); size_t pos = value.size(); @@ -70,14 +70,38 @@ std::optional floorForNamespace(CasOperation & op, const Layout & la return std::nullopt; } +BuildFloor toBuildFloor(const MountLease & mount_lease) +{ + return BuildFloor{ + .writer_epoch = mount_lease.writer_epoch, + .min_active_build_sequence = mount_lease.min_active_build_sequence + }; +} + struct ListedManifestObject { RootNamespace ns; - BuildPrefix prefix; + BuildPrefix build_prefix; ManifestRef ref; String key; }; +bool isBuildUnderFloor( + CasOperation & op, + const Layout & layout, + const ListedManifestObject & manifest, + std::map> & mount_lease_by_ns, + ManifestSweepResult & result) +{ + auto [lease_it, lease_inserted] = mount_lease_by_ns.emplace(manifest.ns.string(), std::nullopt); + if (lease_inserted) + { + ++result.floor_lookups; + lease_it->second = mountLeaseForNamespace(op, layout, manifest.ns, &result.floor_reads); + } + return buildPrefixEligibleUnderFloor(lease_it->second.transform(toBuildFloor), manifest.build_prefix); +} + /// Delegates to the shared `Layout::parseManifestKey`, which validates the canonical manifest-key /// encoding. Keeping parsing in `Layout` avoids a second interpretation of the manifest path and ensures /// the sweep and filesystem checker derive the same namespace and build identity. @@ -89,7 +113,7 @@ std::optional parseListedManifestObject(const Layout & lay return ListedManifestObject{ .ns = parsed->root_namespace, - .prefix = BuildPrefix{.writer_epoch = parsed->ref.writer_epoch, .build_sequence = parsed->ref.build_sequence}, + .build_prefix = BuildPrefix{.writer_epoch = parsed->ref.writer_epoch, .build_sequence = parsed->ref.build_sequence}, .ref = parsed->ref, .key = key}; } @@ -142,6 +166,29 @@ const CatalogEntry * catalogEntryOf(const CasRefCatalog::Snapshot & catalog_cut, return it == catalog_cut.catalog.entries.end() ? nullptr : &*it; } +void excludeCandidatesWithChangedCatalogEntries( + CasOperation & op, + const Layout & layout, + const CasRefCatalog::Snapshot & catalog_cut, + std::vector & candidates, + ManifestSweepResult & result) +{ + if (candidates.empty()) + return; + + const CasRefCatalog::Snapshot rechecked_catalog = CasRefCatalog::read(op, layout); + rechecked_catalog.life_index.throwIfAmbiguous("CAS orphan manifest sweep post-floor catalog"); + std::erase_if(candidates, [&](const ListedManifestObject & candidate) + { + const CatalogEntry * before = catalogEntryOf(catalog_cut, candidate.ns); + const CatalogEntry * after = catalogEntryOf(rechecked_catalog, candidate.ns); + if ((!before && !after) || (before && after && *before == *after)) + return false; + ++result.skipped; + return true; + }); +} + /// The durable `last_folded_ref_id` of a coverage row, `{0, 0}` when there is none — as for a fresh /// pool. A manifest removed by a log above this cursor has not had its `-1` decrement folded, so its /// body remains load-bearing until the fold catches up. @@ -529,32 +576,32 @@ namespace /// do not replace that authority check with a frozen-sequence or judged-dead guess. Compare /// `writer_epoch` first, then `build_sequence`, so old-epoch /// debris drains after a process restart even when its build_sequence is above the current min_active_build_sequence. -bool prefixEligibleOn(CasOperation & op, const Layout & layout, const RootNamespace & ns, const BuildPrefix & prefix) +bool buildPrefixEligibleUnderFloorOn( + CasOperation & op, const Layout & layout, const RootNamespace & ns, const BuildPrefix & build_prefix) { - return prefixEligibleUnder(floorForNamespace(op, layout, ns), prefix); + return buildPrefixEligibleUnderFloor(mountLeaseForNamespace(op, layout, ns).transform(toBuildFloor), build_prefix); } } -bool prefixEligibleUnder(const std::optional & floor, const BuildPrefix & prefix) +bool buildPrefixEligibleUnderFloor(const std::optional & floor, const BuildPrefix & build_prefix) { if (!floor) return false; - const MountLease & w = *floor; - if (prefix.writer_epoch < w.writer_epoch) + if (floor->writer_epoch > build_prefix.writer_epoch) return true; - if (prefix.writer_epoch > w.writer_epoch) + if (floor->writer_epoch < build_prefix.writer_epoch) return false; - if (w.isCleanRelease()) + if (floor->min_active_build_sequence == std::numeric_limits::max()) return true; /// farewell/retired sentinel: every seq is retired - return w.min_active_build_sequence > prefix.build_sequence; + return floor->min_active_build_sequence > build_prefix.build_sequence; } -bool prefixEligible(Pool & store, const RootNamespace & ns, const BuildPrefix & prefix) +bool buildPrefixEligibleUnderFloor(Pool & store, const RootNamespace & ns, const BuildPrefix & build_prefix) { CasOperation op = store.openRequests().admit(); - return prefixEligibleOn(op, store.layout(), ns, prefix); + return buildPrefixEligibleUnderFloorOn(op, store.layout(), ns, build_prefix); } uint64_t sweepNamespace(Pool & store, const RootNamespace & ns, const BuildPrefix & prefix, @@ -563,7 +610,7 @@ uint64_t sweepNamespace(Pool & store, const RootNamespace & ns, const BuildPrefi const Layout & layout = store.layout(); /// The decommission drain runs under the admin's lease; a fenced admin must stop deleting. CasOperation op = store.mountRequests().admit(); - if (!prefixEligibleOn(op, layout, ns, prefix)) + if (!buildPrefixEligibleUnderFloorOn(op, layout, ns, prefix)) return 0; /// not eligible by the durable watermark fact — delete nothing (controls #8/#9) /// Build the protection view. A missing snapshot body, an invalid transaction, or an incomplete @@ -709,7 +756,7 @@ ManifestSweepResult planManifestCursorPage( /// One floor observation per namespace for the whole page. Every build of a namespace is judged /// against the same mount body; reading it per build prefix would cost three requests per listed /// key on a pool where every INSERT is its own build. - std::map> floor_by_ns; + std::map> mount_lease_by_ns; std::map view_by_ns; std::map> active_by_ns; std::set errored_namespaces; /// protection view unavailable => skip, never delete @@ -760,23 +807,18 @@ ManifestSweepResult planManifestCursorPage( continue; } - const CatalogEntry * catalog_entry = catalogEntryOf(catalog_cut, parsed->ns); - if (catalog_entry) + /// Within one `writer_epoch`, `build_sequence < min_active_build_sequence` means the build is no longer active. + /// It may have completed `promote`, been cancelled, or finished cleanup. This does not yet make its manifests + /// safe to delete: ownership and fold coverage must also be checked. + /// If `build_sequence >= min_active_build_sequence`, the build may still be active, so retain its manifests. + if (!isBuildUnderFloor(op, layout, *parsed, mount_lease_by_ns, result)) { - auto [floor_it, floor_inserted] = floor_by_ns.emplace(parsed->ns.string(), std::nullopt); - if (floor_inserted) - { - ++result.floor_lookups; - floor_it->second = floorForNamespace(op, layout, parsed->ns, &result.floor_reads); - } - if (!prefixEligibleUnder(floor_it->second, parsed->prefix)) - { - ++result.skipped; - decided_through = listed.key; - continue; - } + ++result.skipped; + decided_through = listed.key; + continue; } + const CatalogEntry * catalog_entry = catalogEntryOf(catalog_cut, parsed->ns); auto [view_it, view_inserted] = view_by_ns.emplace(parsed->ns.string(), NamespaceFoldView{}); auto [active_it, inserted] = active_by_ns.emplace(parsed->ns.string(), std::set{}); if (inserted) @@ -884,17 +926,13 @@ ManifestSweepResult planManifestCursorPage( continue; } - /// If the post-observation catalog cut has no row, this candidate was necessarily created by a - /// now-dead life: creation publishes its row before any life-owned object. No current life can - /// name it, and a concurrent later creation can only replace its token after this observation. - /// A catalog-named row instead needs the ordinary coverage/owner proof below. if (catalog_entry) { /// THE §6 SAFETY FLOOR, the same predicate `sweepNamespace` calls, under the same watermark /// eligibility. A refusal is recorded rather than silent (Constraint 10). String retain_reason; SweepRetainClass retain_class = SweepRetainClass::None; - if (!manifestDeletionPremise(view_it->second, ManifestKey{parsed->key, parsed->prefix}, + if (!manifestDeletionPremise(view_it->second, ManifestKey{parsed->key, parsed->build_prefix}, &retain_reason, &retain_class)) { /// The sentence goes to the debug log for whoever is chasing ONE object; the class goes to @@ -921,6 +959,10 @@ ManifestSweepResult planManifestCursorPage( decided_through = listed.key; } + /// A writer may create a namespace, commit a ref and advance the floor after `catalog_cut` was read. + /// Recheck after floor reads so the old catalog cut cannot make its committed manifest look unowned. + excludeCandidatesWithChangedCatalogEntries(op, layout, catalog_cut, candidates, result); + size_t next_body_hint = 0; for (const ListedManifestObject & candidate : candidates) { diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.h index e2d63f5053904..f96a20bef8cc1 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.h @@ -21,6 +21,12 @@ struct BuildPrefix uint64_t build_sequence = 0; }; +struct BuildFloor +{ + uint64_t writer_epoch = 0; + uint64_t min_active_build_sequence = 0; +}; + /// One manifest object the sweep is considering, carrying both halves the §6 deletion premise needs: /// the full object key (what the tail's removal targets are named by) and the build prefix it lives @@ -192,17 +198,17 @@ struct ManifestSweepResult uint64_t sweepNamespace(Pool & store, const RootNamespace & ns, const BuildPrefix & prefix, std::vector * warnings = nullptr); -/// Whether `prefix` is sweep-eligible by the durable watermark fact alone. The floor is read from the +/// Whether `build_prefix` is sweep-eligible by the durable watermark fact alone. The floor is read from the /// mount lease identified by the namespace's server-root prefix, not inferred from the manifest key or a /// judged-dead heuristic. A missing lease provides no deletion authority, so the prefix is not eligible. -bool prefixEligible(Pool & store, const RootNamespace & ns, const BuildPrefix & prefix); +bool buildPrefixEligibleUnderFloor(Pool & store, const RootNamespace & ns, const BuildPrefix & build_prefix); -/// The pure half of `prefixEligible`: whether `prefix` is retired under one observation of the mount +/// The pure half of `buildPrefixEligibleUnderFloor`: whether `build_prefix` is retired under one observation of the mount /// floor. `nullopt` (no mount body under any prefix of the namespace) admits nothing. Retirement is /// permanent -- the epoch and the acknowledgement floor only grow and the farewell is terminal -- so /// an admission derived from any observation stays true afterwards, which is what lets a page judge /// every build of a namespace against one read. -bool prefixEligibleUnder(const std::optional & floor, const BuildPrefix & prefix); +bool buildPrefixEligibleUnderFloor(const std::optional & floor, const BuildPrefix & build_prefix); /// Plan one cursor page without deleting. A key is decided from key-derived facts first; only a /// candidate's body is read, after the catalog cut, then decoded and identity-validated; its exact diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasDecommission.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasDecommission.cpp index 3cd6887aa4603..bb5123852df1c 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasDecommission.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasDecommission.cpp @@ -418,9 +418,12 @@ void drainVictim(Pool & admin, const CasRefCatalog::Snapshot & catalog_cut, cons }); } - /// The decommission claim has advanced the writer epoch, so every build prefix with - /// `prefix.writer_epoch < w.writer_epoch` is eligible here. Group the listed keys by namespace and - /// build prefix so each group can use the exact-token orphan sweep. + /// Manifest debris must be removed before the mount slot: deleting the mount body removes the + /// watermark authority, after which `mountLeaseForNamespace` returns no value and the ordinary orphan + /// sweep cannot prove that old-epoch debris is eligible. The decommission claim has advanced the + /// writer epoch, so every build prefix with `prefix.writer_epoch < w.writer_epoch` is eligible here. + /// Group the listed keys by namespace and build prefix so each group can use the exact-token orphan + /// sweep while the mount body still supplies its authority. { const String debris_prefix = admin.layout().casManifestsServerPrefix(victim_srid); std::set> groups; /// (namespace, writer epoch, build sequence) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasFsck.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasFsck.cpp index 8b67d00251a17..2576c74d3a29f 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasFsck.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasFsck.cpp @@ -1115,7 +1115,7 @@ void runFsckImpl(Pool & store, bool detail, const FsckProgress & on_progress, co o.kind = ObjectKind::Blob; o.size = sz; o.cls = FsckClass::Unreachable; - if (parsed && prefixEligible(store, ns, prefix)) + if (parsed && buildPrefixEligibleUnderFloor(store, ns, prefix)) o.reachable_from = {"reclaimable-pre-precommit"}; else o.reachable_from = {"in-flight-pre-precommit"}; diff --git a/src/Disks/tests/cas_sweep_test_support.h b/src/Disks/tests/cas_sweep_test_support.h index 6326dd7c0546c..db40639eb2aeb 100644 --- a/src/Disks/tests/cas_sweep_test_support.h +++ b/src/Disks/tests/cas_sweep_test_support.h @@ -2,11 +2,58 @@ #include #include #include +#include "cas_test_helpers.h" #include +#include +#include +#include namespace DB::Cas::tests { +class SweepReadHookBackend final : public CountingBackend +{ +public: + void arm(const String & key, uint64_t read_number, std::function callback) + { + std::lock_guard lock(hook_mutex); + hook_key = key; + remaining_reads = read_number; + hook = std::move(callback); + fired = false; + } + + bool didFire() const + { + std::lock_guard lock(hook_mutex); + return fired; + } + + std::optional read(const String & key, TransportAccess & access) override + { + std::function callback; + { + std::lock_guard lock(hook_mutex); + if (hook && key == hook_key && --remaining_reads == 0) + { + callback = std::move(hook); + hook = {}; + fired = true; + } + } + if (callback) + callback(); + return CountingBackend::read(key, access); + } + +private: + mutable std::mutex hook_mutex; + String hook_key; + uint64_t remaining_reads = 0; + std::function hook; + bool fired = false; +}; + /// TEST-ONLY variant of the cursor page: plans a page via the production `planManifestCursorPage` and /// then exact-token-deletes every nomination immediately, with no source-edge retirement and no /// `gc/state` adoption of the retirement. Production deletion always goes through `Gc::fold`'s diff --git a/src/Disks/tests/cas_test_helpers.h b/src/Disks/tests/cas_test_helpers.h index ef8391034263b..82a74f62dc298 100644 --- a/src/Disks/tests/cas_test_helpers.h +++ b/src/Disks/tests/cas_test_helpers.h @@ -1348,7 +1348,7 @@ inline uint64_t foldCursorOf( /// Set a server root's durable floor (so orphan-sweep eligibility can be driven). After the ack-floor /// merge the floor rides the mount lease body (`mountKey`), so this seeds a MountLease carrying -/// `{writer_epoch, min_active_build_sequence}` — exactly what `prefixEligible` reads. +/// `{writer_epoch, min_active_build_sequence}` — exactly what `buildPrefixEligibleUnderFloor` reads. inline void setWatermarkMinActive( DB::Cas::Backend & backend, const DB::Cas::Layout & layout, const String & server_root_id, uint64_t writer_epoch, uint64_t min_active_build_sequence) diff --git a/src/Disks/tests/gtest_cas_orphan_manifest_sweep.cpp b/src/Disks/tests/gtest_cas_orphan_manifest_sweep.cpp index 3d87eab377a7a..65b1e9388cec4 100644 --- a/src/Disks/tests/gtest_cas_orphan_manifest_sweep.cpp +++ b/src/Disks/tests/gtest_cas_orphan_manifest_sweep.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -438,23 +439,226 @@ TEST(CASOrphanManifestSweep, CursorPageRespectsDeleteBudget) EXPECT_NE(first_exists, second_exists); } -/// A physical manifest captured before a catalog cut which omits its name is dead-life debris: a live -/// creation cannot publish a life-owned object before its catalog row. It therefore has an eventual -/// page-sweep owner without trying to reconstruct a deleted incarnation from the key. -TEST(CASOrphanManifestSweep, CursorPageDeletesObservedBodyWhenCatalogOmitsNamespace) +TEST(CASOrphanManifestSweep, CursorPageRetainsRowlessBodyWithoutMountLease) { auto backend = std::make_shared(); auto store = openPoolForTest(backend); const RootNamespace ns{"00/catalog-absent-debris@cas@"}; const ManifestRef r = ref(5, 0xA9); writeManifestRaw(*backend, store->layout(), ns, r, {blobEntryFor("debris", DB::UInt128(9))}); - /// No mount lease/watermark exists: after legal catalog-row deletion there may be no server-root - /// state left to supply one. The post-observation absent row is the complete dead-life proof. const ManifestSweepResult result = sweepManifestCursorPageForTest(*store, "", /*list_budget=*/100, /*delete_budget=*/10); + EXPECT_EQ(result.floor_lookups, 1u); + EXPECT_TRUE(result.nominations.empty()); + EXPECT_EQ(result.deleted, 0u); + EXPECT_TRUE(headExists(*backend, store->layout().manifestKey(ManifestId{ns, r}))); +} + +TEST(CASOrphanManifestSweep, CursorPageRetainsStagedManifestUntilFirstPublication) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + const RootNamespace ns{store->poolConfig().server_root_id + "/staged-before-admission@cas@"}; + auto build = store->beginPartWrite(PartWriteInfo{ + .intended_ref = ns.string() + "/part", .intended_namespace = ns, .op = ProvenanceOp::Insert}); + ManifestEntry entry; + entry.path = "f"; + entry.placement = EntryPlacement::Inline; + entry.ref = BlobRef{BlobHashAlgo::CityHash128, BlobDigest::fromU128(u128Of("v"))}; + entry.blob_size = 1; + entry.inline_bytes = "v"; + const ManifestId id = build->stageManifest({entry}); + CasOperation op = store->openRequests().admit(); + ASSERT_FALSE(CasRefCatalog::lifeIfCataloged(op, store->layout(), ns)); + + const ManifestSweepResult result = sweepManifestCursorPageForTest(*store, "", 100, 10); + + ASSERT_TRUE(result.nominations.empty()); + EXPECT_EQ(result.deleted, 0u); + ASSERT_TRUE(headExists(*backend, store->layout().manifestKey(id))); + build->precommitAdd(ns, "part", id); + build->promote(ns, "part", build->buildId(), id); + const PartManifest manifest = store->readManifest(id); + ASSERT_EQ(manifest.entries.size(), 1u); + EXPECT_EQ(manifest.entries.front().inline_bytes, "v"); +} + +TEST(CASOrphanManifestSweep, CursorPageRetainsRowlessBodyOfActiveBuild) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + const RootNamespace ns{"00/active-before-admission@cas@"}; + const ManifestId id = writeManifestRaw(*backend, store->layout(), ns, ref(5, 1), {blobEntryFor("f", DB::UInt128(1))}); + setWatermarkMinActive(*backend, store->layout(), kServerRoot, kWriterEpoch, 5); + + const ManifestSweepResult result = sweepManifestCursorPageForTest(*store, "", 100, 10); + + EXPECT_TRUE(result.nominations.empty()); + EXPECT_EQ(result.deleted, 0u); + EXPECT_TRUE(headExists(*backend, store->layout().manifestKey(id))); +} + +TEST(CASOrphanManifestSweep, CursorPageDeletesRowlessBodyBelowFloor) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + const RootNamespace ns{"00/retired-before-admission@cas@"}; + const ManifestId id = writeManifestRaw(*backend, store->layout(), ns, ref(5, 1), {blobEntryFor("f", DB::UInt128(1))}); + setWatermarkMinActive(*backend, store->layout(), kServerRoot, kWriterEpoch, 6); + + const ManifestSweepResult result = sweepManifestCursorPageForTest(*store, "", 100, 10); + EXPECT_EQ(result.deleted, 1u); - EXPECT_FALSE(headExists(*backend, store->layout().manifestKey(ManifestId{ns, r}))); + EXPECT_FALSE(headExists(*backend, store->layout().manifestKey(id))); +} + +TEST(CASOrphanManifestSweep, CursorPageRetainsRowlessBodyWhenNamespaceAppearsAfterCatalogCut) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + const Layout & layout = store->layout(); + const RootNamespace ns{"00/admitted-after-cut@cas@"}; + const ManifestRef candidate = ref(5, 1); + const ManifestId id = writeManifestRaw(*backend, layout, ns, candidate, {blobEntryFor("f", DB::UInt128(1))}); + setWatermarkMinActive(*backend, layout, kServerRoot, kWriterEpoch, 5); + backend->arm(layout.mountKey(kServerRoot), 1, [&] + { + publishCommittedTransition(*backend, layout, ns, "part", std::nullopt, candidate); + setWatermarkMinActive(*backend, layout, kServerRoot, kWriterEpoch, 6); + }); + + const ManifestSweepResult result = sweepManifestCursorPageForTest(*store, "", 100, 10); + + EXPECT_TRUE(backend->didFire()); + EXPECT_TRUE(result.nominations.empty()); + EXPECT_TRUE(headExists(*backend, layout.manifestKey(id))); +} + +TEST(CASOrphanManifestSweep, CursorPageRetainsBodyWhenNamespaceIsRebornAfterCatalogCut) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + const Layout & layout = store->layout(); + const RootNamespace ns{"00/reborn-after-cut@cas@"}; + const ManifestRef candidate = ref(5, 1); + const ManifestId id = writeManifestRaw(*backend, layout, ns, candidate, {blobEntryFor("f", DB::UInt128(1))}); + registerNamespaceRaw(*backend, layout, ns); + seedConsumedSealCursor(*backend, layout, ns); + seedEmptyRecoveryAuthority(*backend, layout, ns); + setWatermarkMinActive(*backend, layout, kServerRoot, kWriterEpoch, 6); + backend->arm(layout.mountKey(kServerRoot), 1, [&] + { + OperationForTest op(*backend); + const auto before = CasRefCatalog::read(*op, layout); + RefCatalog next = before.catalog; + ASSERT_EQ(next.entries.size(), 1u); + next.entries.front().incarnation = fixture::fixtureLife(RootNamespace{"00/successor@cas@"}).incarnation; + ASSERT_TRUE(before.etag); + ASSERT_TRUE(std::holds_alternative( + (*op).replace(layout.refCatalogKey(), encodeRefCatalog(next), *before.etag, Retry::standard()))); + publishCommittedTransition(*backend, layout, ns, "part", std::nullopt, candidate); + }); + + const ManifestSweepResult result = sweepManifestCursorPageForTest(*store, "", 100, 10); + + EXPECT_TRUE(backend->didFire()); + EXPECT_TRUE(result.nominations.empty()); + EXPECT_EQ(result.deleted, 0u); + EXPECT_TRUE(headExists(*backend, layout.manifestKey(id))); +} + +TEST(CASOrphanManifestSweep, CursorPageRetainsBodyWhenCatalogEntryDisappearsOrChangesState) +{ + for (bool remove_entry : {false, true}) + { + SCOPED_TRACE(remove_entry); + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + const Layout & layout = store->layout(); + const RootNamespace ns{"00/lifecycle-after-cut@cas@"}; + const ManifestId id = writeManifestRaw(*backend, layout, ns, ref(5, 1), {blobEntryFor("f", DB::UInt128(1))}); + registerNamespaceRaw(*backend, layout, ns); + seedConsumedSealCursor(*backend, layout, ns); + seedEmptyRecoveryAuthority(*backend, layout, ns); + setWatermarkMinActive(*backend, layout, kServerRoot, kWriterEpoch, 6); + backend->arm(layout.refCatalogKey(), 2, [&] + { + OperationForTest op(*backend); + const auto before = CasRefCatalog::read(*op, layout); + RefCatalog next = before.catalog; + ASSERT_EQ(next.entries.size(), 1u); + if (remove_entry) + next.entries.clear(); + else + { + next.entries.front().state = NsState::Removing; + next.entries.front().removal_started_round = 1; + } + ASSERT_TRUE(before.etag); + ASSERT_TRUE(std::holds_alternative( + (*op).replace(layout.refCatalogKey(), encodeRefCatalog(next), *before.etag, Retry::standard()))); + }); + + const ManifestSweepResult result = sweepManifestCursorPageForTest(*store, "", 100, 10); + + EXPECT_TRUE(backend->didFire()); + EXPECT_TRUE(result.nominations.empty()); + EXPECT_TRUE(headExists(*backend, layout.manifestKey(id))); + } +} + +TEST(CASOrphanManifestSweep, CursorPageRejectsAmbiguousPostFloorCatalog) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + const Layout & layout = store->layout(); + const RootNamespace ns{"00/ambiguous-after-cut@cas@"}; + const ManifestId id = writeManifestRaw(*backend, layout, ns, ref(5, 1), {blobEntryFor("f", DB::UInt128(1))}); + registerNamespaceRaw(*backend, layout, ns); + seedConsumedSealCursor(*backend, layout, ns); + seedEmptyRecoveryAuthority(*backend, layout, ns); + setWatermarkMinActive(*backend, layout, kServerRoot, kWriterEpoch, 6); + backend->arm(layout.refCatalogKey(), 2, [&] + { + OperationForTest op(*backend); + const auto before = CasRefCatalog::read(*op, layout); + RefCatalog next = before.catalog; + ASSERT_EQ(next.entries.size(), 1u); + CatalogEntry duplicate = next.entries.front(); + duplicate.ns = RootNamespace{"00/ambiguous-after-cut-twin@cas@"}; + next.entries.push_back(duplicate); + std::sort(next.entries.begin(), next.entries.end(), + [](const CatalogEntry & lhs, const CatalogEntry & rhs) { return lhs.ns.string() < rhs.ns.string(); }); + ASSERT_TRUE(before.etag); + ASSERT_TRUE(std::holds_alternative( + (*op).replace(layout.refCatalogKey(), encodeRefCatalog(next), *before.etag, Retry::standard()))); + }); + backend->resetCounts(); + + EXPECT_THROW(sweepManifestCursorPageForTest(*store, "", 100, 10), DB::Exception); + + EXPECT_TRUE(backend->didFire()); + EXPECT_EQ(backend->getCount(layout.manifestKey(id)), 0u); + EXPECT_EQ(backend->deleteCount(layout.manifestKey(id)), 0u); +} + +TEST(CASOrphanManifestSweep, CursorPagePropagatesPostFloorCatalogReadFailure) +{ + auto backend = std::make_shared(); + auto store = openPoolForTest(backend); + const Layout & layout = store->layout(); + const RootNamespace ns{"00/catalog-read-failure@cas@"}; + const ManifestId id = writeManifestRaw(*backend, layout, ns, ref(5, 1), {blobEntryFor("f", DB::UInt128(1))}); + setWatermarkMinActive(*backend, layout, kServerRoot, kWriterEpoch, 6); + backend->arm(layout.refCatalogKey(), 2, [] { throw std::runtime_error("injected catalog read failure"); }); + backend->resetCounts(); + + EXPECT_ANY_THROW(sweepManifestCursorPageForTest(*store, "", 100, 10)); + + EXPECT_TRUE(backend->didFire()); + EXPECT_EQ(backend->getCount(layout.manifestKey(id)), 0u); + EXPECT_EQ(backend->deleteCount(layout.manifestKey(id)), 0u); } /// Manifest keys are write-once, so a same-key rewrite after the observation below never happens in diff --git a/src/Disks/tests/gtest_cas_orphan_sweep_requests.cpp b/src/Disks/tests/gtest_cas_orphan_sweep_requests.cpp index 34e87af705ed4..88e943d9317c0 100644 --- a/src/Disks/tests/gtest_cas_orphan_sweep_requests.cpp +++ b/src/Disks/tests/gtest_cas_orphan_sweep_requests.cpp @@ -5,6 +5,7 @@ #include #include #include +#include "cas_sweep_test_support.h" #include #include #include @@ -86,6 +87,204 @@ struct PageFixture } }; +struct RowlessPageFixture +{ + std::shared_ptr backend = std::make_shared(); + PoolPtr store = openPoolForTest(backend); + + RowlessPageFixture() + { + setWatermarkMinActive(*backend, store->layout(), "test", 1, 100); + } + + String add(uint64_t sequence, const RootNamespace & ns = kNs, uint64_t epoch = 1) + { + const ManifestRef ref{.writer_epoch = epoch, .build_sequence = sequence, .manifest_ordinal = 1}; + const ManifestId id = writeManifestRaw(*backend, store->layout(), ns, ref, {blobEntryFor("f", DB::UInt128(sequence))}); + return store->layout().manifestKey(id); + } + + void replaceCatalogOnSecondRead(String bytes) + { + OperationForTest op(*backend); + const auto before = CasRefCatalog::read(*op, store->layout()); + ASSERT_TRUE(before.etag); + backend->arm(store->layout().refCatalogKey(), 2, [this, catalog_bytes = std::move(bytes), token = *before.etag] + { + OperationForTest replacement(*backend); + ASSERT_TRUE(std::holds_alternative( + (*replacement).replace(store->layout().refCatalogKey(), catalog_bytes, token, Retry::standard()))); + }); + } + + ManifestSweepResult page(uint64_t list_budget = 100, uint64_t nomination_budget = 100) + { + backend->resetCounts(); + return planManifestCursorPage(*store, "", list_budget, nomination_budget, true, nullptr); + } +}; + +} + +TEST(CASOrphanSweepRequests, RowlessPageRetainsBuildsAbovePublishedFloor) +{ + RowlessPageFixture f; + const Layout & layout = f.store->layout(); + const std::vector keys{f.add(5), f.add(6), f.add(8)}; + setWatermarkMinActive(*f.backend, layout, "test", 1, 5); + + const ManifestSweepResult result = f.page(); + + EXPECT_TRUE(result.nominations.empty()); + EXPECT_EQ(result.skipped, 3u); + EXPECT_EQ(result.floor_lookups, 1u); + EXPECT_EQ(f.backend->getCount(layout.mountKey("test")), 1u); + EXPECT_EQ(f.backend->getCount(layout.refCatalogKey()), 1u); + for (const String & key : keys) + EXPECT_EQ(f.backend->getCount(key), 0u); +} + +TEST(CASOrphanSweepRequests, RowlessPageRespectsEpochAndRetiredFloor) +{ + struct Case + { + uint64_t epoch; + uint64_t floor; + size_t nominations; + }; + for (const Case & c : {Case{1, 5, 1}, Case{3, 5, 0}, Case{2, std::numeric_limits::max(), 1}}) + { + SCOPED_TRACE(c.epoch); + RowlessPageFixture f; + const String key = f.add(5, kNs, c.epoch); + setWatermarkMinActive(*f.backend, f.store->layout(), "test", 2, c.floor); + + const ManifestSweepResult result = f.page(); + + EXPECT_EQ(result.nominations.size(), c.nominations); + EXPECT_EQ(f.backend->getCount(key), c.nominations); + } +} + +TEST(CASOrphanSweepRequests, PostFloorCatalogFiltersOnlyChangedNamespaces) +{ + RowlessPageFixture f; + const Layout & layout = f.store->layout(); + const String changed = f.add(1); + const String stable = f.add(1, RootNamespace{"test/store/465/bb@cas@"}); + CatalogEntry entry; + entry.ns = kNs; + entry.state = NsState::Live; + entry.incarnation = fixture::fixtureLife(kNs).incarnation; + f.replaceCatalogOnSecondRead(encodeRefCatalog(RefCatalog{.entries = {entry}})); + + const ManifestSweepResult result = f.page(); + + EXPECT_TRUE(f.backend->didFire()); + ASSERT_EQ(result.nominations.size(), 1u); + EXPECT_EQ(result.nominations.front().key, stable); + EXPECT_EQ(result.skipped, 1u); + EXPECT_EQ(f.backend->getCount(changed), 0u); + EXPECT_EQ(f.backend->getCount(stable), 1u); + EXPECT_EQ(f.backend->getCount(layout.refCatalogKey()), 2u); +} + +TEST(CASOrphanSweepRequests, PostFloorRetentionDoesNotRefillNominationBudget) +{ + RowlessPageFixture f; + const Layout & layout = f.store->layout(); + const String first = f.add(1); + const String second = f.add(2); + ASSERT_LT(first, second); + CatalogEntry entry; + entry.ns = kNs; + entry.state = NsState::Live; + entry.incarnation = fixture::fixtureLife(kNs).incarnation; + f.replaceCatalogOnSecondRead(encodeRefCatalog(RefCatalog{.entries = {entry}})); + + const ManifestSweepResult result = f.page(100, 1); + + EXPECT_TRUE(f.backend->didFire()); + EXPECT_TRUE(result.nominations.empty()); + EXPECT_EQ(result.next_cursor, first); + EXPECT_FALSE(result.wrapped); + EXPECT_EQ(f.backend->getCount(first), 0u); + EXPECT_EQ(f.backend->getCount(second), 0u); + EXPECT_EQ(f.backend->listCount(layout.casManifestsPrefix()), 1u); +} + +TEST(CASOrphanSweepRequests, MalformedPostFloorCatalogFailsBeforeBodyReads) +{ + RowlessPageFixture f; + const String key = f.add(1); + f.replaceCatalogOnSecondRead("not a catalog"); + + EXPECT_THROW(f.page(), DB::Exception); + + EXPECT_TRUE(f.backend->didFire()); + EXPECT_EQ(f.backend->getCount(key), 0u); + EXPECT_EQ(f.backend->deleteCount(key), 0u); +} + +TEST(CASOrphanSweepRequests, CandidatePageReadsCatalogTwice) +{ + RowlessPageFixture f; + const Layout & layout = f.store->layout(); + const std::vector keys{f.add(1), f.add(2), f.add(3)}; + + const ManifestSweepResult result = f.page(); + + EXPECT_EQ(result.nominations.size(), 3u); + EXPECT_EQ(f.backend->getCount(layout.refCatalogKey()), 2u); + EXPECT_EQ(result.floor_lookups, 1u); + EXPECT_EQ(f.backend->getCount(layout.mountKey("test")), 1u); + EXPECT_EQ(f.backend->listCount(layout.casManifestsPrefix()), 1u); + for (const String & key : keys) + EXPECT_EQ(f.backend->getCount(key), 1u); +} + +TEST(CASOrphanSweepRequests, ListOnlyPageDoesNotReadPostFloorCatalog) +{ + RowlessPageFixture f; + const Layout & layout = f.store->layout(); + const String first = f.add(1); + const String second = f.add(2); + + const ManifestSweepResult result = f.page(1, 0); + + EXPECT_TRUE(result.nominations.empty()); + EXPECT_EQ(f.backend->getCount(layout.refCatalogKey()), 1u); + EXPECT_EQ(result.floor_lookups, 0u); + EXPECT_EQ(f.backend->getCount(first), 0u); + EXPECT_EQ(f.backend->getCount(second), 0u); + EXPECT_EQ(result.next_cursor, first); + EXPECT_FALSE(result.wrapped); +} + +TEST(CASOrphanSweepRequests, UnrelatedCatalogChangeDoesNotRetainStableOrphan) +{ + RowlessPageFixture f; + const Layout & layout = f.store->layout(); + const String key = f.add(1); + seedFoldCursorForTest(*f.backend, layout, kNs, RefTxnId{2, 1}); + writeRecoverableCkptForRawFixture(*f.backend, layout, kNs, RefCkpt{ + .life_epoch = 1, .committed_through = std::nullopt, + .checkpoint_snapshot_id = std::nullopt, .last_epoch_seal = std::nullopt}); + casAdmitEntry(*f.backend, layout, RootNamespace{"test/zz-unrelated@cas@"}); + OperationForTest op(*f.backend); + RefCatalog next = CasRefCatalog::read(*op, layout).catalog; + ASSERT_EQ(next.entries.size(), 2u); + next.entries.back().state = NsState::Removing; + next.entries.back().removal_started_round = 1; + f.replaceCatalogOnSecondRead(encodeRefCatalog(next)); + + const ManifestSweepResult result = f.page(); + + EXPECT_TRUE(f.backend->didFire()); + ASSERT_EQ(result.nominations.size(), 1u); + EXPECT_EQ(result.nominations.front().key, key); + EXPECT_EQ(f.backend->getCount(layout.refCatalogKey()), 2u); + EXPECT_EQ(f.backend->getCount(key), 1u); } TEST(CASOrphanSweepRequests, FloorIsReadOncePerNamespacePerPage) @@ -136,18 +335,18 @@ TEST(CASOrphanSweepRequests, RetainedKeysCostNoBodyRead) EXPECT_EQ(f.backend->getCount(layout.manifestKey(ManifestId{kNs, build(seq)})), 0u) << seq; } -TEST(CASOrphanSweepRequests, PrefixEligibleUnderIsTheFourComparisons) +TEST(CASOrphanSweepRequests, BuildPrefixEligibleUnderFloorIsTheFourComparisons) { - MountLease floor; + BuildFloor floor; floor.writer_epoch = 3; floor.min_active_build_sequence = 10; - EXPECT_TRUE(prefixEligibleUnder(floor, BuildPrefix{.writer_epoch = 2, .build_sequence = 999})); - EXPECT_FALSE(prefixEligibleUnder(floor, BuildPrefix{.writer_epoch = 4, .build_sequence = 1})); - EXPECT_TRUE(prefixEligibleUnder(floor, BuildPrefix{.writer_epoch = 3, .build_sequence = 9})); - EXPECT_FALSE(prefixEligibleUnder(floor, BuildPrefix{.writer_epoch = 3, .build_sequence = 10})); + EXPECT_TRUE(buildPrefixEligibleUnderFloor(floor, BuildPrefix{.writer_epoch = 2, .build_sequence = 999})); + EXPECT_FALSE(buildPrefixEligibleUnderFloor(floor, BuildPrefix{.writer_epoch = 4, .build_sequence = 1})); + EXPECT_TRUE(buildPrefixEligibleUnderFloor(floor, BuildPrefix{.writer_epoch = 3, .build_sequence = 9})); + EXPECT_FALSE(buildPrefixEligibleUnderFloor(floor, BuildPrefix{.writer_epoch = 3, .build_sequence = 10})); floor.min_active_build_sequence = std::numeric_limits::max(); - EXPECT_TRUE(prefixEligibleUnder(floor, BuildPrefix{.writer_epoch = 3, .build_sequence = 10})); - EXPECT_FALSE(prefixEligibleUnder(std::nullopt, BuildPrefix{.writer_epoch = 1, .build_sequence = 1})); + EXPECT_TRUE(buildPrefixEligibleUnderFloor(floor, BuildPrefix{.writer_epoch = 3, .build_sequence = 10})); + EXPECT_FALSE(buildPrefixEligibleUnderFloor(std::nullopt, BuildPrefix{.writer_epoch = 1, .build_sequence = 1})); } namespace diff --git a/src/Disks/tests/gtest_cas_writer_duties.cpp b/src/Disks/tests/gtest_cas_writer_duties.cpp index 45e24c1377fee..8a9d29cf12ba0 100644 --- a/src/Disks/tests/gtest_cas_writer_duties.cpp +++ b/src/Disks/tests/gtest_cas_writer_duties.cpp @@ -464,7 +464,7 @@ TEST(CASWriterDuties, RejectedAttemptBodyIsEventuallyNominatedAndSwept) /// field alone; pair the two so the mount lease's admission arithmetic sees what the budget claims. backend->setAttemptTimeoutMs(budget.attempt_timeout_ms); /// Rooted under the POOL's OWN `server_root_id` ("test", unlike this file's other fixtures, which - /// stay under "srv1" precisely because they never drive the orphan sweep): `prefixEligible`'s + /// stay under "srv1" precisely because they never drive the orphan sweep): `buildPrefixEligibleUnderFloor`'s /// watermark floor is looked up by walking the NAMESPACE's own prefix segments for a live mount /// lease, so a namespace rooted under any other server-root would find no floor and retain forever /// regardless of epoch/coverage. From 300a8c155c70608d77b2f2b71f66f7eea4159bee Mon Sep 17 00:00:00 2001 From: Mikhail Filimonov Date: Fri, 6 Nov 2026 03:04:28 +0100 Subject: [PATCH 2/2] CAS: Reject content part writes through encrypted CAS disks Port of https://github.com/Altinity/ClickHouse/pull/2773 (antalya-26.8) to antalya-26.6. (cherry picked from commit 86c6de4f742fbbf7ce404c9bce5bb4efac8a8b37) Co-Authored-By: Claude Fable 5.1 Signed-off-by: Mikhail Filimonov --- src/Disks/DiskEncryptedTransaction.cpp | 12 +++ .../test_cas_encrypted_disk/__init__.py | 0 .../configs/storage.xml | 56 +++++++++++++ .../test_cas_encrypted_disk/test.py | 78 +++++++++++++++++++ 4 files changed, 146 insertions(+) create mode 100644 tests/integration/test_cas_encrypted_disk/__init__.py create mode 100644 tests/integration/test_cas_encrypted_disk/configs/storage.xml create mode 100644 tests/integration/test_cas_encrypted_disk/test.py diff --git a/src/Disks/DiskEncryptedTransaction.cpp b/src/Disks/DiskEncryptedTransaction.cpp index 14fa16a3b607a..7981db2642b77 100644 --- a/src/Disks/DiskEncryptedTransaction.cpp +++ b/src/Disks/DiskEncryptedTransaction.cpp @@ -1,6 +1,8 @@ #include #if USE_SSL +#include +#include #include #include #include @@ -16,6 +18,7 @@ namespace DB namespace ErrorCodes { extern const int DATA_ENCRYPTION_ERROR; + extern const int NOT_IMPLEMENTED; } @@ -87,6 +90,15 @@ std::unique_ptr DiskEncryptedTransaction::writeFileImpl const WriteSettings & settings) { auto wrapped_path = wrappedPath(path); + if (delegate_disk->isContentAddressed() && Cas::isPartFilePath(wrapped_path)) + { + auto part_path = Cas::parsePartFilePath(wrapped_path); + if (!part_path || part_path->file.empty() || Cas::partFileMustStayBlob(part_path->file)) + throw Exception( + ErrorCodes::NOT_IMPLEMENTED, + "Writing content part files through an encrypted content-addressed disk is not supported"); + } + FileEncryption::Header header; String key; UInt64 old_file_size = 0; diff --git a/tests/integration/test_cas_encrypted_disk/__init__.py b/tests/integration/test_cas_encrypted_disk/__init__.py new file mode 100644 index 0000000000000..e69de29bb2d1d diff --git a/tests/integration/test_cas_encrypted_disk/configs/storage.xml b/tests/integration/test_cas_encrypted_disk/configs/storage.xml new file mode 100644 index 0000000000000..33a8d6cd78dfe --- /dev/null +++ b/tests/integration/test_cas_encrypted_disk/configs/storage.xml @@ -0,0 +1,56 @@ + + + + + object_storage + s3 + cas + itest-cas-encrypted-disk + http://rustfs1:11121/test/cas_encrypted_disk/ + clickhouse + clickhouse + + + encrypted + disk_cas + encrypted/ + 1234567812345678 + + + s3 + http://rustfs1:11121/test/plain_s3_encrypted_disk/ + clickhouse + clickhouse + + + encrypted + disk_s3 + encrypted/ + 1234567812345678 + + + + + +
+ disk_cas +
+ + + + +
+ disk_cas_encrypted +
+ + + + +
+ disk_s3_encrypted +
+ + + + + diff --git a/tests/integration/test_cas_encrypted_disk/test.py b/tests/integration/test_cas_encrypted_disk/test.py new file mode 100644 index 0000000000000..528533a62ccd2 --- /dev/null +++ b/tests/integration/test_cas_encrypted_disk/test.py @@ -0,0 +1,78 @@ +import pytest + +from helpers.cluster import ClickHouseCluster + +cluster = ClickHouseCluster(__file__) +node = cluster.add_instance( + "node", + main_configs=["configs/storage.xml"], + with_rustfs=True, +) + +TABLE = "encrypted_cas_insert" +INSERT = f"INSERT INTO {TABLE} SELECT number, toString(number) FROM numbers(1000)" + + +@pytest.fixture(scope="module", autouse=True) +def start_cluster(): + try: + cluster.start() + yield cluster + finally: + cluster.shutdown() + + +@pytest.fixture(autouse=True) +def cleanup_table(start_cluster): + try: + yield + finally: + node.query(f"DROP TABLE IF EXISTS {TABLE} SYNC") + + +def create_table(policy): + node.query( + f""" + CREATE TABLE {TABLE} (id UInt64, data String) + ENGINE = MergeTree + ORDER BY id + SETTINGS storage_policy = '{policy}', + min_bytes_for_wide_part = 0, + min_rows_for_wide_part = 0, + min_bytes_for_full_part_storage = 0, + min_rows_for_full_part_storage = 0, + min_level_for_full_part_storage = 0 + """ + ) + + +def test_insert_into_encrypted_cas_is_rejected(): + create_table("cas_encrypted") + + error = node.query_and_get_error(INSERT) + + assert "NOT_IMPLEMENTED" in error, error + assert ( + "Writing content part files through an encrypted content-addressed disk is not supported" + in error + ), error + assert node.query(f"SELECT count() FROM {TABLE}") == "0\n" + assert node.query( + f"SELECT count() FROM system.parts " + f"WHERE database = currentDatabase() AND table = '{TABLE}' AND active" + ) == "0\n" + + +@pytest.mark.parametrize("policy", ["cas", "s3_encrypted"]) +def test_insert_into_supported_disk(policy): + create_table(policy) + + node.query(INSERT) + + assert node.query( + f"SELECT count(), sum(id), countIf(data != toString(id)) FROM {TABLE}" + ) == "1000\t499500\t0\n" + assert node.query( + f"SELECT DISTINCT disk_name FROM system.parts " + f"WHERE database = currentDatabase() AND table = '{TABLE}' AND active" + ) == f"disk_{policy}\n"