From c4efb260fa15c2e68001db649f1d20765b1eec0b Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Wed, 28 Oct 2026 17:47:19 +0100 Subject: [PATCH 1/4] Fix CAS orphan manifest sweep publication races Apply the mount floor to candidates without a catalog row. Recheck each candidate's full catalog row after floor reads and before reading bodies. Retain candidates whose namespace appears, disappears, or changes life. Add deterministic regressions for staged publication and catalog races. Verification: 8/9 regressions failed before the fix; 9/9 pass after it. https://gist.github.com/filimonov/1b96ea358804b1c6e22c45380cc65469 https://gist.github.com/k-morozov/13a60ab4424b1b20ba54e4d89b1c0917 Signed-off-by: Konstantin Morozov --- .../Gc/CasOrphanManifestSweep.cpp | 44 ++-- src/Disks/tests/cas_sweep_test_support.h | 47 ++++ .../tests/gtest_cas_orphan_manifest_sweep.cpp | 218 +++++++++++++++++- 3 files changed, 284 insertions(+), 25 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp index 28a471b673ff6..626370e847c4d 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp @@ -758,23 +758,20 @@ ManifestSweepResult planManifestCursorPage( continue; } - const CatalogEntry * catalog_entry = catalogEntryOf(catalog_cut, parsed->ns); - if (catalog_entry) + auto [floor_it, floor_inserted] = floor_by_ns.emplace(parsed->ns.string(), std::nullopt); + if (floor_inserted) { - 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.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; } + 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) @@ -882,10 +879,6 @@ 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 @@ -919,6 +912,21 @@ ManifestSweepResult planManifestCursorPage( decided_through = listed.key; } + if (!candidates.empty()) + { + const CasRefCatalog::Snapshot post_floor_catalog = CasRefCatalog::read(op, layout); + post_floor_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(post_floor_catalog, candidate.ns); + if ((!before && !after) || (before && after && *before == *after)) + return false; + ++result.skipped; + return true; + }); + } + size_t next_body_hint = 0; for (const ListedManifestObject & candidate : candidates) { 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/gtest_cas_orphan_manifest_sweep.cpp b/src/Disks/tests/gtest_cas_orphan_manifest_sweep.cpp index 2c68bf8492fd0..41bc8258a47d8 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 @@ -370,23 +371,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 From d1ee3e46a682c729342a77ab307a3f77c62afe0f Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Wed, 28 Oct 2026 17:51:03 +0100 Subject: [PATCH 2/4] Test CAS orphan sweep catalog recheck bounds Cover rowless mount floors, epoch boundaries, per-namespace catalog changes, cursor progress after retention, and catalog request counts. Document the floor and catalog checks before candidate body reads. Verification: unit_tests_dbms built successfully; 53 tests from CASOrphanManifestSweep, CASOrphanSweepRequests and CASSweepDeletionPremise pass. Scoped whitespace checks and documentation anchors pass. https://gist.github.com/filimonov/1b96ea358804b1c6e22c45380cc65469 https://gist.github.com/k-morozov/13a60ab4424b1b20ba54e4d89b1c0917 https://gist.github.com/k-morozov/5c81c35c5ea432c36f4467ece81dc2ce Signed-off-by: Konstantin Morozov --- .../cas/architecture/manifests-and-refs.md | 38 +++- .../tests/gtest_cas_orphan_sweep_requests.cpp | 199 ++++++++++++++++++ 2 files changed, 232 insertions(+), 5 deletions(-) diff --git a/docs/en/antalya/cas/architecture/manifests-and-refs.md b/docs/en/antalya/cas/architecture/manifests-and-refs.md index 2fc93c4c238e1..96f2b9114cd96 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/tests/gtest_cas_orphan_sweep_requests.cpp b/src/Disks/tests/gtest_cas_orphan_sweep_requests.cpp index 168d09683d3b7..1e7abff751294 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) From 706e7facf9c77d67ce675b3ef2699b7afe459be9 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Wed, 28 Oct 2026 18:08:45 +0100 Subject: [PATCH 3/4] Extract CAS orphan sweep checks Move mount-floor eligibility into `isBuildBelowFloor` and the second catalog cut into `filterCandidatesByCatalog`. Keep cursor progress in `planManifestCursorPage` and preserve skip accounting. Preserve floor caching, request counts and the order of observations. https://gist.github.com/filimonov/1b96ea358804b1c6e22c45380cc65469 https://gist.github.com/k-morozov/13a60ab4424b1b20ba54e4d89b1c0917 Signed-off-by: Konstantin Morozov --- .../Gc/CasOrphanManifestSweep.cpp | 62 ++++++++++++------- 1 file changed, 41 insertions(+), 21 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp index 626370e847c4d..9733c8a1b741e 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp @@ -78,6 +78,22 @@ struct ListedManifestObject String key; }; +bool isBuildBelowFloor( + CasOperation & op, + const Layout & layout, + const ListedManifestObject & manifest, + std::map> & floor_by_ns, + ManifestSweepResult & result) +{ + auto [floor_it, floor_inserted] = floor_by_ns.emplace(manifest.ns.string(), std::nullopt); + if (floor_inserted) + { + ++result.floor_lookups; + floor_it->second = floorForNamespace(op, layout, manifest.ns, &result.floor_reads); + } + return prefixEligibleUnder(floor_it->second, manifest.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. @@ -142,6 +158,29 @@ const CatalogEntry * catalogEntryOf(const CasRefCatalog::Snapshot & catalog_cut, return it == catalog_cut.catalog.entries.end() ? nullptr : &*it; } +void filterCandidatesByCatalog( + CasOperation & op, + const Layout & layout, + const CasRefCatalog::Snapshot & catalog_cut, + std::vector & candidates, + ManifestSweepResult & result) +{ + if (candidates.empty()) + return; + + const CasRefCatalog::Snapshot post_floor_catalog = CasRefCatalog::read(op, layout); + post_floor_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(post_floor_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. @@ -758,13 +797,7 @@ ManifestSweepResult planManifestCursorPage( continue; } - 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)) + if (!isBuildBelowFloor(op, layout, *parsed, floor_by_ns, result)) { ++result.skipped; decided_through = listed.key; @@ -912,20 +945,7 @@ ManifestSweepResult planManifestCursorPage( decided_through = listed.key; } - if (!candidates.empty()) - { - const CasRefCatalog::Snapshot post_floor_catalog = CasRefCatalog::read(op, layout); - post_floor_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(post_floor_catalog, candidate.ns); - if ((!before && !after) || (before && after && *before == *after)) - return false; - ++result.skipped; - return true; - }); - } + filterCandidatesByCatalog(op, layout, catalog_cut, candidates, result); size_t next_body_hint = 0; for (const ListedManifestObject & candidate : candidates) From fa0e5fe85da073788da7c904b264518ac2e87c47 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Wed, 28 Oct 2026 18:13:18 +0100 Subject: [PATCH 4/4] Clarify CAS orphan sweep catalog check names Use `excludeCandidatesWithChangedCatalogEntries` and `rechecked_catalog` to describe the catalog recheck and its candidate filtering. https://gist.github.com/filimonov/1b96ea358804b1c6e22c45380cc65469 https://gist.github.com/k-morozov/13a60ab4424b1b20ba54e4d89b1c0917 Signed-off-by: Konstantin Morozov --- .../ContentAddressed/Gc/CasOrphanManifestSweep.cpp | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp index 9733c8a1b741e..68f04b4da91d6 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp +++ b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp @@ -158,7 +158,7 @@ const CatalogEntry * catalogEntryOf(const CasRefCatalog::Snapshot & catalog_cut, return it == catalog_cut.catalog.entries.end() ? nullptr : &*it; } -void filterCandidatesByCatalog( +void excludeCandidatesWithChangedCatalogEntries( CasOperation & op, const Layout & layout, const CasRefCatalog::Snapshot & catalog_cut, @@ -168,12 +168,12 @@ void filterCandidatesByCatalog( if (candidates.empty()) return; - const CasRefCatalog::Snapshot post_floor_catalog = CasRefCatalog::read(op, layout); - post_floor_catalog.life_index.throwIfAmbiguous("CAS orphan manifest sweep post-floor catalog"); + 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(post_floor_catalog, candidate.ns); + const CatalogEntry * after = catalogEntryOf(rechecked_catalog, candidate.ns); if ((!before && !after) || (before && after && *before == *after)) return false; ++result.skipped; @@ -945,7 +945,7 @@ ManifestSweepResult planManifestCursorPage( decided_through = listed.key; } - filterCandidatesByCatalog(op, layout, catalog_cut, candidates, result); + excludeCandidatesWithChangedCatalogEntries(op, layout, catalog_cut, candidates, result); size_t next_body_hint = 0; for (const ListedManifestObject & candidate : candidates)