Fix early manifest deletion in CAS GC

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/<br/>freeze candidates with exact GET"] --> B{"build-prefix eligible?<br/>durable watermark fact only"}
+    A["LIST one page of cas/manifests/<br/>read initial catalog K0"] --> B{"build-prefix eligible?<br/>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<br/>plus live precommits<br/>plus manifests with an unfolded minus-one"]
+    ELIG --> ROW{"namespace row in K0?"}
+    ROW -->|yes| C["protection view from K0:<br/>committed manifests, live precommits,<br/>and unfolded removal targets"]
+    ROW -->|no| T["tentative candidate"]
     C -->|"key protected"| SKIP2["skip"]
-    C -->|"not protected"| D{"open and decode<br/>frozen body"}
+    C -->|"not protected"| P{"deletion premise:<br/>fold coverage permits deletion?"}
+    P -->|no| SKIP2
+    P -->|yes| T
+    T --> K["read and validate one post-floor<br/>catalog K1 for all tentative candidates"]
+    K --> Q{"namespace row equal in K0 and K1<br/>or absent in both?"}
+    Q -->|no| SKIP2
+    Q -->|yes| GET["freeze candidate with exact GET"]
+    GET --> D{"open and decode<br/>frozen body"}
     D -->|"cannot open or decode"| U["retain; skipped++ and undecodable++<br/>log exact key; advance decision cursor<br/>continue to later candidates"]
     D -->|decoded| I{"body ref and namespace<br/>match key?"}
     I -->|no| BAD["CORRUPTED_DATA<br/>fail-closed round error"]

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<String, std::optional<MountLease>> & 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<ListedManifestObject> & 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<String>{});
         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)
     {

src/Disks/tests/cas_sweep_test_support.h

@@ -2,11 +2,58 @@
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Backend/CasRequests.h>
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasOrphanManifestSweep.h>
 #include <Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Pool/CasPool.h>
+#include "cas_test_helpers.h"
 #include <base/types.h>
+#include <functional>
+#include <mutex>
+#include <utility>
 
 namespace DB::Cas::tests
 {
 
+class SweepReadHookBackend final : public CountingBackend
+{
+public:
+    void arm(const String & key, uint64_t read_number, std::function<void()> 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<Raw> read(const String & key, TransportAccess & access) override
+    {
+        std::function<void()> 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<void()> 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