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/CasGc.cpp

@@ -2989,7 +2989,7 @@ Gc::FoldResult Gc::fold(GcState & state, std::optional<Etag> & /*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}))
                 {
@@ -4846,7 +4846,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))

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<MountLease> floorForNamespace(CasOperation & op, const Layout & layout, const RootNamespace & ns,
-                                            uint64_t * reads = nullptr)
+std::optional<MountLease> 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<MountLease> 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<String, std::optional<MountLease>> & 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<ListedManifestObject> 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<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.
@@ -529,40 +576,40 @@ 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<MountLease> & floor, const BuildPrefix & prefix)
+bool buildPrefixEligibleUnderFloor(const std::optional<BuildFloor> & 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.min_active_build_sequence == std::numeric_limits<uint64_t>::max())
+    if (floor->min_active_build_sequence == std::numeric_limits<uint64_t>::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,
                         std::vector<String> * warnings)
 {
     const Layout & layout = store.layout();
     CasOperation op = store.openRequests().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
@@ -707,7 +754,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<String, std::optional<MountLease>> floor_by_ns;
+    std::map<String, std::optional<MountLease>> mount_lease_by_ns;
     std::map<String, NamespaceFoldView> view_by_ns;
     std::map<String, std::set<String>> active_by_ns;
     std::set<String> errored_namespaces;   /// protection view unavailable => skip, never delete
@@ -758,23 +805,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<String>{});
         if (inserted)
@@ -882,17 +924,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
@@ -919,6 +957,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)
     {

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
@@ -190,17 +196,17 @@ struct ManifestSweepResult
 uint64_t sweepNamespace(Pool & store, const RootNamespace & ns, const BuildPrefix & prefix,
                         std::vector<String> * 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<MountLease> & floor, const BuildPrefix & prefix);
+bool buildPrefixEligibleUnderFloor(const std::optional<BuildFloor> & 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

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Tools/CasDecommission.cpp

@@ -264,7 +264,7 @@ DecommissionReport decommissionPoolMember(BackendPtr backend, PoolConfig config,
     }
 
     /// Manifest debris must be removed before the mount slot: deleting the mount body removes the
-    /// watermark authority, after which `floorForNamespace` returns no value and the ordinary orphan
+    /// 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

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"};