fix(graph): verify sealed retirement and preserve retry state

crates/tracedecay-graph-db/src/generation.rs

@@ -59,7 +59,10 @@ pub use replay::{
 pub(crate) use replay::{
     checked_decode_replay_source, metadata_manifest_from_source, validate_supplied_rows_binding,
 };
-pub(crate) use spill::{ENTITIES_FILE, read_spilled_entity};
+pub(crate) use spill::{
+    AbandonedRowSpill, ENTITIES_FILE, MAX_ROW_SPILL_OWNER_RECEIPT_BYTES,
+    ROW_SPILL_OWNER_RECEIPT_FILE, ROW_SPILL_PREFIX, read_spilled_entity, row_spill_owner,
+};
 pub use spill::{
     GRAPH_ROW_SPILL_RUN_BYTES, GraphGenerationRowSpill, GraphGenerationRows,
     GraphSpillRowFootprint, SpilledGraphGeneration,

crates/tracedecay-graph-db/src/generation_runtime.rs

@@ -1840,6 +1840,11 @@ impl GraphDb {
             self.retire_sealed_generation_store(locator)?;
             return Ok(GenerationContentsDeletion::RetentionPending);
         }
+        // Native row deletion may already have committed, but the generation
+        // is not collected until its independently durable sealed artifact is
+        // gone. If unlink refuses, keep the in-memory retirement state retryable
+        // instead of advertising collection while bytes still exist on disk.
+        self.retire_sealed_generation_store(locator)?;
         let mut state = self.wait_verified_generations_write()?;
         state.known.remove(locator);
         state.quarantined.remove(locator);
@@ -1858,7 +1863,6 @@ impl GraphDb {
             state.heads.remove(&locator.projection);
         }
         drop(state);
-        self.retire_sealed_generation_store(locator)?;
         Ok(GenerationContentsDeletion::Deleted)
     }
 
@@ -4001,20 +4005,73 @@ mod tests {
             .delete_generation_contents(&locator, &|| Ok(()))
             .expect_err("a failed sealed unlink must not complete retention");
         assert!(
-            error
-                .to_string()
-                .contains("sealed directory removal failed")
+            matches!(
+                &error,
+                GraphDbError::Unavailable { message }
+                    if message.starts_with("sealed generation store sealed directory removal failed:")
+                        || message.starts_with("sealed generation store sealed generation retirement refused:")
+            ),
+            "an unremovable artifact must fail at the sealed-directory unlink boundary: {error}"
         );
         assert_eq!(std::fs::read(&directory).unwrap(), b"not a directory");
+        assert!(
+            !database
+                .wait_verified_generations_write()
+                .unwrap()
+                .collected
+                .contains(&locator),
+            "a failed sealed unlink must not mark its generation collected"
+        );
 
         std::fs::remove_file(&directory).unwrap();
         std::fs::create_dir(&directory).unwrap();
+        // Retirement requires an exact durable identity from the directory
+        // being removed. A receipt-free synthetic directory must never confer
+        // deletion authority when the strict collector is integrated.
+        let identity = generation.identity();
+        let receipt = serde_json::json!({
+            "version": 1,
+            "form": "compact",
+            "namespace": locator.projection.namespace.as_str(),
+            "projection": locator.projection.projection.as_str(),
+            "generation": locator.generation.as_str(),
+            "physical_namespace": namespace.as_str(),
+            "recovered_digest": sealed_digest(&generation).as_str(),
+            "entities": 0,
+            "relations": 0,
+            "graph_format": GraphFormatVersion::current().get(),
+            "row_sum": null,
+            "base": null,
+            "source_generation": identity.source_generation.as_str(),
+            "watermark": identity.watermark.as_str()
+        });
+        std::fs::write(
+            directory.join("sealed.json"),
+            serde_json::to_vec(&receipt).unwrap(),
+        )
+        .unwrap();
         std::fs::write(directory.join("derived-rows"), b"retired").unwrap();
+        let unrelated = directory.parent().unwrap().join("unrelated-entry");
+        std::fs::create_dir(&unrelated).unwrap();
+        std::fs::write(unrelated.join("sentinel"), b"untouched").unwrap();
         assert!(matches!(
             database.delete_generation_contents(&locator, &|| Ok(())),
             Ok(GenerationContentsDeletion::Deleted)
         ));
+        assert_eq!(
+            std::fs::read(unrelated.join("sentinel")).unwrap(),
+            b"untouched",
+            "retiring one sealed generation must not traverse another child"
+        );
         assert!(!directory.exists());
+        assert!(
+            database
+                .wait_verified_generations_write()
+                .unwrap()
+                .collected
+                .contains(&locator),
+            "only successful physical retirement may mark the generation collected"
+        );
         // Idempotent replay after the directory is gone still succeeds.
         assert!(matches!(
             database.delete_generation_contents(&locator, &|| Ok(())),

crates/tracedecay-graph-db/src/registry.rs

@@ -2806,7 +2806,9 @@ mod tests {
         registry
             .retain_verification_fault_for_lease(&lease, &fault)
             .unwrap();
-        let path = corrupt_registration.canonical_path().to_path_buf();
+        // TempDir may live under a symlinked temporary root. The registry
+        // reports the physical canonical path selected at admission.
+        let path = std::fs::canonicalize(corrupt_registration.canonical_path()).unwrap();
         let bytes = std::fs::read(&path).unwrap();
 
         for _ in 0..2 {

crates/tracedecay-graph-db/src/runtime.rs

@@ -1015,8 +1015,8 @@ impl GraphDb {
     /// management, still checkpoints in full before this returns.
     #[tracing::instrument(name = "graph_db.runtime.close", level = "trace", skip_all)]
     pub(crate) fn close(&self) -> Result<(), GraphDbError> {
-        let engine_is_open = match self.inner.database.read() {
-            Ok(database) => database.is_some(),
+        let engine = match self.inner.database.read() {
+            Ok(database) => database,
             Err(_) => {
                 self.inner.closed.store(true, Ordering::Release);
                 self.inner.poisoned.store(true, Ordering::Release);
@@ -1027,12 +1027,23 @@ impl GraphDb {
                 });
             }
         };
-        if !engine_is_open {
-            if self.inner.closed.swap(true, Ordering::AcqRel) {
-                return Err(GraphDbError::Closed);
+        if engine.is_none() {
+            // Keep the absence observation locked until close is published.
+            // A lazy opener must not install an engine in that interval.
+            let previously_closed = self.inner.closed.swap(true, Ordering::AcqRel);
+            // A failed physical close can already have consumed the engine.
+            // Its absent handle must not turn an uncertain result into a
+            // successful close (or an idempotent Closed on the next retry).
+            if self.inner.poisoned.load(Ordering::Acquire) {
+                return Err(durability_uncertain());
             }
-            return Ok(());
+            return if previously_closed {
+                Err(GraphDbError::Closed)
+            } else {
+                Ok(())
+            };
         }
+        drop(engine);
         let _snapshot_gate = self.wait_snapshot_gate_write();
         let mut guard =
             match crate::observe::wait_lock(crate::observe::LOCK_WAIT_DATABASE_WRITE, || {
@@ -1713,11 +1724,16 @@ impl GraphDb {
             return Ok(());
         }
         self.ensure_available()?;
+        #[cfg(test)]
+        test_seams::fire(test_seams::Seam::LazyOpenAdmission);
         let mut database =
             crate::observe::wait_lock(crate::observe::LOCK_WAIT_DATABASE_WRITE, || {
                 self.inner.database.write()
             })
             .map_err(|_| GraphDbError::unavailable("graph database write lock is poisoned"))?;
+        // Close or a failed hibernation may have won while this opener
+        // waited. Revalidate before publishing another native engine.
+        self.ensure_available()?;
         if database.is_some() {
             return Ok(());
         }
@@ -1965,6 +1981,8 @@ pub(crate) mod test_seams {
 
     #[derive(Clone, Copy, Debug, Eq, PartialEq)]
     pub(crate) enum Seam {
+        /// After lazy-open admission, before acquiring the database write lock.
+        LazyOpenAdmission,
         /// Immediately before grafeo opens the container at `site`.
         EngineOpen(GraphEngineOpenSite),
         /// After the engine has closed and synced the container, immediately

crates/tracedecay-graph-db/src/runtime/tests.rs

@@ -6,6 +6,7 @@ use std::time::Duration;
 
 use grafeo_common::types::Value;
 
+use super::test_seams::{self, Seam};
 use super::{GraphDb, sync_wal};
 use crate::location::PersistentGraphStoreState;
 use crate::recovery::set_projection_quarantine;
@@ -29,6 +30,78 @@ fn memory_db() -> GraphDbLeaseV1 {
     .unwrap()
 }
 
+#[test]
+fn lazy_open_rechecks_close_before_materializing_an_engine() {
+    let directory = tempfile::tempdir().unwrap();
+    let path = directory.path().join("graph.grafeo");
+    let options = || GraphDbOpenOptions {
+        location: GraphDbLocation::Persistent(path.clone()),
+        expected_format: GraphFormatVersion::current(),
+        durability: GraphDurability::WalSync,
+        cancellation: Arc::new(NeverCancelled),
+    };
+    let seed = GraphDb::open(options()).unwrap();
+    seed.apply_unverified(scalar_batch("retained")).unwrap();
+    seed.close().unwrap();
+    drop(seed);
+
+    // Both a never-materialized mount and a hibernated engine can be absent
+    // when close races a lazy reader's admission.
+    for hibernated in [false, true] {
+        let database =
+            GraphDb::open_lazy_with_store_state(options(), PersistentGraphStoreState::Existing)
+                .unwrap();
+        if hibernated {
+            database.ensure_opened().unwrap();
+            database.hibernate_if_lazy().unwrap();
+        }
+        assert!(database.inner.database.read().unwrap().is_none());
+
+        let closing = Arc::clone(&database);
+        let admissions = Arc::new(AtomicUsize::new(0));
+        let observed = Arc::clone(&admissions);
+        let hook = test_seams::install(move |seam| {
+            if seam == Seam::LazyOpenAdmission {
+                observed.fetch_add(1, Ordering::Relaxed);
+                // The opener has already observed Ready. Complete the real
+                // close before it takes its database write lock.
+                closing.close().unwrap();
+            }
+        });
+        let result = database.ensure_opened();
+        drop(hook);
+
+        assert_eq!(admissions.load(Ordering::Relaxed), 1);
+        assert_eq!(result, Err(GraphDbError::Closed));
+        assert_eq!(database.runtime_state(), GraphDbRuntimeState::Closed);
+        assert!(
+            database.inner.database.read().unwrap().is_none(),
+            "a completed close must not be followed by a resurrected engine"
+        );
+    }
+
+    // Refusing the stale opener must not corrupt the durable store or block
+    // a new, independently authorized mount from reading the committed row.
+    let reopened = GraphDb::open(options()).unwrap();
+    let snapshot = reopened.snapshot().unwrap();
+    let entity = snapshot
+        .entity(
+            &GraphNamespace::new("project").unwrap(),
+            &GraphEntityId::new("a").unwrap(),
+            Arc::new(NeverCancelled),
+        )
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        entity
+            .properties
+            .get(&GraphPropertyName::new("name").unwrap()),
+        Some(&GraphProperty::String("retained".to_owned()))
+    );
+    drop(snapshot);
+    reopened.close().unwrap();
+}
+
 #[test]
 fn runtime_state_distinguishes_ready_closed_and_durability_uncertain() {
     let ready = memory_db();
@@ -63,10 +136,17 @@ fn owner_close_releases_the_physical_database_after_durability_uncertainty() {
                 .to_owned(),
         })
     );
+    assert!(
+        !owner.engine_is_open().unwrap(),
+        "the poisoned close must release the physical database"
+    );
     assert_eq!(
         owner.close(),
-        Err(GraphDbError::Closed),
-        "the poisoned close must release the physical database before a second close"
+        Err(GraphDbError::DurabilityUncertain {
+            message: "the handle was poisoned after an observed post-commit persistence failure"
+                .to_owned(),
+        }),
+        "a retry after a poisoned close must stay uncertain, never idempotent Closed"
     );
 }