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/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.cpp index 28a471b673ff6..68f04b4da91d6 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 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. @@ -758,23 +797,14 @@ ManifestSweepResult planManifestCursorPage( continue; } - const CatalogEntry * catalog_entry = catalogEntryOf(catalog_cut, parsed->ns); - if (catalog_entry) + if (!isBuildBelowFloor(op, layout, *parsed, floor_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) @@ -882,10 +912,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 +945,8 @@ ManifestSweepResult planManifestCursorPage( decided_through = listed.key; } + excludeCandidatesWithChangedCatalogEntries(op, layout, catalog_cut, candidates, result); + 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 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)