From 1d4167043fe6875beb954425ac42d5559636c230 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 14:57:56 +0000 Subject: [PATCH 1/2] persistence follow-up gates: publication vs observed head, enforceable fences MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The five semantic gaps the post-#912 review surfaced, each closed with a falsifier rather than a doc comment. 1. A reconciled retry no longer claims a publication version. #912 named `Reconciled.current_head` carefully and then `seal_cycle` adopted it as `SealedCycle.version` anyway, so retrying cycle 1 at head V5 recorded cycle 1 as sealed into V5. `publication_version` is now Some ONLY for a fresh Committed; `observed_head` is Some only when a sink observed one (None on NoChange, which calls no sink). 2. `FleetRecovery::checkpoint_bound` turns the latecomer rule from an instruction in a doc comment into an API that caps the durable bound strictly below any foreign landing. 3. An ABI-malformed artifact is `CommitError::InvalidArtifact` — permanent. Reporting it as retryable `Io` meant "fail, regenerate identically, fail" without bound. 4. Store identity is lexical, so `x/./s.lance` cannot claim a second slot beside `x/s.lance`, and the registry claim is an RAII value taken before the first `.await` — a cancelled open can no longer leak a reservation. 5. Startup seeding streams frame rows to a max instead of materialising the whole timeline; the schema guard checks types and nullability, not just column names. The append/reopen/reconcile branch — which carries the whole no-rollback contract and cannot be made to fail on demand through real Lance — now has a `#[cfg(test)]` fault-injection seam and three falsifiers covering its unpublished, published-but-unacknowledged, and reconcile-unavailable arms. `scan_sealed` is payload-free by contract: stated on the trait and modelled by all seven fakes, which previously cloned payloads and so proved a property the real writer does not hold. Tests: cycle_sink 19, cycle_driver 26, persist_sink 22, supervisor integration probes green; fmt clean; clippy adds no new warnings. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01KCGhDYoQBXs3poaR7sFuqp --- .claude/board/EPIPHANIES.md | 70 ++ .claude/board/PR_ARC_INVENTORY.md | 33 + .../persistence-artifact-backed-commit-v1.md | 2 +- .../persistence-cycle-wal-bootstrap-v1.md | 14 +- .../examples/blw_fusion.rs | 7 +- .../examples/blw_tenant.rs | 7 +- .../lance-graph-planner/src/persist_sink.rs | 46 +- .../examples/measure_wal_curve.rs | 14 +- .../src/cycle_driver.rs | 241 ++++++- .../tests/d_ign_b_lenses.rs | 11 +- .../tests/probe_ignition.rs | 11 +- .../tests/probe_ignition_64k.rs | 11 +- crates/lance-graph/src/graph/cycle_sink.rs | 604 +++++++++++++++--- 13 files changed, 941 insertions(+), 130 deletions(-) diff --git a/.claude/board/EPIPHANIES.md b/.claude/board/EPIPHANIES.md index dcdcc840..5bc68a17 100644 --- a/.claude/board/EPIPHANIES.md +++ b/.claude/board/EPIPHANIES.md @@ -1,5 +1,74 @@ ## 2026-08-09 +## E-A-RECONCILED-HEAD-IS-NOT-A-PUBLICATION-1 (2026-08-09) + +**Status:** FINDING. **Confidence:** high — falsified both ways. + +Renaming a field is not the same as removing the confusion it names. +`CommitOutcome::Reconciled.current_head` was deliberately NOT called +`version` (#912) precisely because it is the store head AT RECONCILIATION +TIME. One layer up, `seal_cycle` then adopted it as `SealedCycle.version` +anyway — so retrying cycle 1 while the head stood at V5 recorded cycle 1 +as "sealed into V5". The careful name survived; the meaning did not. + +The repair is structural, not documentary: `SealedCycle` now carries +`publication_version: Option` (`Some` ONLY for a fresh +`Committed`) beside `observed_head: Option` (`Some` only +when a sink actually observed one — `None` on `NoChange`, which calls no +sink and whose `head` is the caller's asserted base). A publication +position that was never observed stays **unknown**; the durable identity +is `(cycle, batch_hash)` and the position is an audit-path read. + +**The general shape:** when a type distinguishes two things a consumer +will conflate, the distinction has to be *unrepresentable* at the +consumer, not merely *documented* at the producer. A doc comment warning +"this is not X" is evidence that the next layer will use it as X. + +Falsifier: `a_reconciled_retry_never_reports_the_current_head_as_publication` +(cycle 1 → V1, cycle 2 → V2, retry cycle 1 → `Reconciled`, observed head +V2, publication `None`, zero extra appends). + +## E-A-DOC-COMMENT-IS-NOT-AN-ENFORCEMENT-1 (2026-08-09) + +**Status:** FINDING. **Confidence:** high. + +`FleetRecovery::foreign_min_cycle` shipped with the correct rule written +in its doc comment: *"the caller must NEVER raise its durable +`after_cycle` bound to or past this cycle."* A rule stated in prose to a +caller who holds a plain `Option` is an instruction, not a +guard — the unsafe checkpoint stays one obvious line away. + +`FleetRecovery::checkpoint_bound(recovered_through)` is the same rule as +an API: it returns the bound the caller MAY store, capped strictly below +any foreign landing. The caller can still ignore it, but the safe path is +now the shortest one. + +Companion: the same session made the writer's single-owner claim +enforceable rather than narrated (lexical `store_identity` so +`x/./s.lance` cannot claim a second slot beside `x/s.lance`; an RAII +`WriterClaim` taken before the first `.await`, so a cancelled `open` +cannot leak a reservation). Both are the same move — *carry the +invariant in a value, not in a sentence.* + +## E-A-A-PERMANENT-FAULT-REPORTED-AS-RETRYABLE-IS-AN-INFINITE-LOOP-1 (2026-08-09) + +**Status:** FINDING. **Confidence:** high. + +A 511-byte artifact payload violates the writer's 512-byte ABI. It was +reported as `CommitError::Io` — the variant whose documented meaning is +"nothing published, safe to REGENERATE". A caller obeying that contract +regenerates the identical malformed batch and fails identically, forever: +the error classification, not the bug, is what makes it unbounded. + +`CommitError::InvalidArtifact { row, len }` is permanent by construction. +The taxonomy rule this instance teaches: **an error variant's retry +semantics are part of its contract**, so a producer-side defect must never +borrow a transport-side variant merely because both mean "did not +commit". + +Falsifier: `a_malformed_artifact_is_refused_permanently_not_as_retryable_io` +(refused twice, identically, store untouched). + ### E-THE-ARTIFACT-WRITE-DECIDES-WHAT-KANBAN-PROGRESS-BECOMES-DURABLE-1 **FINDING (operator-ruled, implemented Phase A).** The persistence question was @@ -15591,3 +15660,4 @@ Cross-ref: W2's sprint-2 deliverable (Tier-0 "what's shipped" index); `.claude/p Three review findings on the #629 doc arc, accepted and folded into the V3 docs (routing.md §1/§5, mailbox-kanban-model.md, INTEGRATION-PLAN W6a): (1) **Clustered-index caveat** — `NodeGuid` stores classid via `to_le_bytes`, so RAW key-byte prefixes order by the custom byte first; domain range-scans hold over the DECODED u32 (or an order-preserving big-endian rendering), never raw LE byte prefixes/tries (E-CLASSID-CANON-HIGH-IS-A-CLUSTERED-INDEX reads with this caveat). (2) **Corpus-proof scanner scope** — "old-form" = ALL THREE legacy shapes: `0x0000_DDCC`, `0x1000_DDCC`, AND `0xAAAA_DDCC` (legacy app/render prefix high, e.g. `0x0005_0901`) — exactly the set `classid_canon_compat` routes CanonLow; scanning only the first two can falsely prove the corpus clean (E-V3-MARKER-IS-A-MONITOR reads with this scope). (3) **ractor wording sharpened** — "compile-time ownership dummy / never a hot-path bus" scopes the DATA plane; the actor mailbox remains the runtime serialized single-writer CONTROL path (one-message Advance/MulAdvance serialization = the codex #578 atomicity mechanism). The operator ruling's meaning is unchanged; the wording now says both halves. Cross-ref: PR #629 review threads (2 codex P2 + 1 coderabbit); E-MAILBOX-KANBAN-NO-COLLAPSEGATE. + diff --git a/.claude/board/PR_ARC_INVENTORY.md b/.claude/board/PR_ARC_INVENTORY.md index f01d7392..b354800d 100644 --- a/.claude/board/PR_ARC_INVENTORY.md +++ b/.claude/board/PR_ARC_INVENTORY.md @@ -3157,3 +3157,36 @@ Removes `crate-ci/typos` spell-check job from `style.yml`; `cargo fmt --check` r - LATEST_STATE Current Contract Inventory annotated with `WitnessTable` + `MailboxSoA` additions. **Confidence (2026-05-28):** working — `cargo check -p cognitive-shader-driver -p lance-graph-contract` clean (only pre-existing ontology deprecation warnings); `cargo test -p cognitive-shader-driver -p lance-graph-contract --lib` 457 passed / 0 failed at merge time. Codex P1 (`f541b280`) addressed before merge. No consumer surfaces touched. + +## Follow-up gates after #912 (branch `claude/persistence-follow-up-gates`, 2026-08-09) + +**Added.** `SealedCycle.publication_version` / `observed_head` (both +`Option`, replacing the conflated `version`); `FleetRecovery::checkpoint_bound` +(the enforceable latecomer fence); `CommitError::InvalidArtifact` (permanent, +never retryable); `store_identity` + RAII `WriterClaim` (lexical store identity; +claim held before the first `.await`); `LanceCycleWriter::max_cycle` (streaming +O(1)-memory startup seed, replacing `timeline().max()`); full schema guard +(types + nullability, not names only); a `#[cfg(test)]` fault-injection seam for +the append/reopen/reconcile branch. + +**Falsifiers added (8).** Reconciled-retry-never-claims-publication; +checkpoint_bound both ways; permanent-InvalidArtifact (twice, identically); +alternate path spellings refused; failed open leaks no reservation; injected +unpublished append → Io + regenerable; injected published append → Reconciled, +exactly one durable frame; injected reconcile-read failure → Ambiguous, resolved +by re-submitting the same frozen batch. + +**Also.** `scan_sealed` is payload-free BY CONTRACT — documented on the trait +and modelled by all seven fakes (they previously proved a property the real +writer lacks); `persist_cycle`'s `NoChange.head` provenance documented; +`content_hash`'s inclusion of `base_version` documented as deliberate (a +re-derived frame is a DIFFERENT assertion and must fail closed, not launder). + +**Deferred (named, not silently dropped).** The typed +`IntentOnly | Artifact512` boundary (the gate still tests payload PRESENCE); +the first-Create ambiguity state machine (an unknown `Create` still treats a +later `NotFound` as absence); `run_cycle`'s borrow-over-`.await` prose; the +landing-row rollup. These are Phase-B/C/D work, tracked in the plan. + +**Confidence:** high on everything with a falsifier; the deferrals are the +honest remainder. diff --git a/.claude/plans/persistence-artifact-backed-commit-v1.md b/.claude/plans/persistence-artifact-backed-commit-v1.md index ece1359d..64fba240 100644 --- a/.claude/plans/persistence-artifact-backed-commit-v1.md +++ b/.claude/plans/persistence-artifact-backed-commit-v1.md @@ -119,7 +119,7 @@ competition.** ## 3. No rollback, no compensating delete — reconciliation is authoritative **Measured, not assumed:** Lance 9 has **no atomic expected-version fence for -Append`. The conflict rebase runs even on a single-attempt commit; strict +`Append`**. The conflict rebase runs even on a single-attempt commit; strict no-rebase mode exists only for `Overwrite` (`lance-9.0.0/src/io/commit.rs:914-950`). This is stated honestly rather than papered over with a read-check pretending to be compare-and-swap. diff --git a/.claude/plans/persistence-cycle-wal-bootstrap-v1.md b/.claude/plans/persistence-cycle-wal-bootstrap-v1.md index 3f255bd4..276ab47d 100644 --- a/.claude/plans/persistence-cycle-wal-bootstrap-v1.md +++ b/.claude/plans/persistence-cycle-wal-bootstrap-v1.md @@ -79,12 +79,14 @@ two-dimensional upgrade path is §3–§4; the accepted debts are §5. --- -## 2. A complete logical cycle is physically SPARSE (RATIFIED architecture, UNIMPLEMENTED in a concrete sink) +## 2. A complete logical cycle is physically SPARSE (RATIFIED architecture, IMPLEMENTED in `LanceCycleWriter`) -> **Status of this section:** the sparse-delta rule is **RATIFIED as -> architecture** and **UNIMPLEMENTED in a concrete Lance sink**. The #878 -> bootstrap remains SHIPPED; this section governs the *future* concrete sink, -> not the merged contract-probe. +> **Status of this section (updated 2026-08-09, #912):** the sparse-delta rule +> is **RATIFIED as architecture** and **IMPLEMENTED** in +> `lance_graph::graph::cycle_sink::LanceCycleWriter` (Phase A) — landing +> metadata rows + one coalesced image row per dirty row. The #878 bootstrap +> remains SHIPPED; the body text below predates the concrete sink and is kept +> append-only. **"One complete cycle image" must NEVER be read as serializing every row merely because every participant belonged to the cycle.** The load-bearing distinction: @@ -372,7 +374,7 @@ solves the final temporal model: | Shadow temporal-coherence correction pass | **PLANNED** (§4) | | `revision.rs` forward-correction mechanism | **PLANNED** (§4) | | Concrete Lance sink (real crash durability) | **DEFERRED** — gated on crash falsifiers (§5) | -| Sparse-delta storage rule (complete cycle ≠ full rewrite) | **RATIFIED architecture, UNIMPLEMENTED in a concrete sink** (§2) | +| Sparse-delta storage rule (complete cycle ≠ full rewrite) | **RATIFIED architecture, IMPLEMENTED in `LanceCycleWriter`** (Phase A #912, §2) | The bootstrap exists so the rest can be built on a running, durable seam. The scalar order is a load-bearing placeholder, and this document is the record that diff --git a/crates/lance-graph-planner/examples/blw_fusion.rs b/crates/lance-graph-planner/examples/blw_fusion.rs index 28271d74..4ea20203 100644 --- a/crates/lance-graph-planner/examples/blw_fusion.rs +++ b/crates/lance-graph-planner/examples/blw_fusion.rs @@ -477,7 +477,12 @@ impl WalSink for MemWal { .flat_map(|s| { s.landings.iter().map(|slot| LandedSlot { cycle: s.cycle, - slot: slot.clone(), + // `scan_sealed` is payload-free by contract; payloads + // live in the image read. + slot: SweepSlot { + payload: Vec::new(), + ..slot.clone() + }, }) }) .collect()) diff --git a/crates/lance-graph-planner/examples/blw_tenant.rs b/crates/lance-graph-planner/examples/blw_tenant.rs index 16dfd8e4..ac860f2e 100644 --- a/crates/lance-graph-planner/examples/blw_tenant.rs +++ b/crates/lance-graph-planner/examples/blw_tenant.rs @@ -503,7 +503,12 @@ impl WalSink for MemWal { .flat_map(|s| { s.landings.iter().map(|slot| LandedSlot { cycle: s.cycle, - slot: slot.clone(), + // `scan_sealed` is payload-free by contract; payloads + // live in the image read. + slot: SweepSlot { + payload: Vec::new(), + ..slot.clone() + }, }) }) .collect()) diff --git a/crates/lance-graph-planner/src/persist_sink.rs b/crates/lance-graph-planner/src/persist_sink.rs index 161decc1..1ddc9a95 100644 --- a/crates/lance-graph-planner/src/persist_sink.rs +++ b/crates/lance-graph-planner/src/persist_sink.rs @@ -67,8 +67,10 @@ //! real durability. **`compile+test green ≠ storage proven`** (the Ladybug //! lesson). The concrete sink is `lance_graph::graph::cycle_sink::LanceCycleWriter`. //! -//! ## The governing storage rule (operator-ruled 2026-08-09 — supersedes the -//! ## earlier "one version per cycle, empty cycles included" contract) +//! ## The governing storage rule +//! +//! Operator-ruled 2026-08-09; supersedes the earlier "one version per cycle, +//! empty cycles included" contract. //! //! **No artifact-backed semantic change → no write → no new [`DatasetVersion`].** //! @@ -270,6 +272,15 @@ pub enum CommitError { /// I/O failed with provably NOTHING published — safe to regenerate from /// the unchanged horizon. Io(WriteFailed), + /// An artifact payload violates the concrete writer's binary ABI (the + /// canonical witness row is exactly 512 bytes). **PERMANENT, not + /// retryable:** nothing was written, and re-submitting or regenerating + /// the same malformed batch can never succeed — classifying this as + /// [`Io`](CommitError::Io) sends the caller into an endless regenerate + /// loop. Fix the producer. (The [`persist_cycle`] artifact gate tests + /// payload PRESENCE only; the size ABI is enforced by the writer — the + /// typed `IntentOnly | Artifact512` split is the Phase-B refinement.) + InvalidArtifact { row: u64, len: usize }, /// The append's outcome could not be determined AND reconciliation itself /// failed. Re-submit the SAME frozen batch: `commit_cycle` reconciles /// first, so the retry cannot double-append. @@ -295,6 +306,11 @@ impl std::fmt::Display for CommitError { "cycle {cycle:?} durable with hash {stored_hash:#018x}, offered {offered_hash:#018x} — fail closed" ), Self::Io(e) => write!(f, "commit I/O (nothing published): {e}"), + Self::InvalidArtifact { row, len } => write!( + f, + "artifact payload for row {row} is {len} bytes, violating the writer's \ + ABI — permanent, fix the producer (nothing written, do NOT retry)" + ), Self::Ambiguous { cycle, batch_hash, @@ -374,6 +390,14 @@ impl DetachedCycleBatch { } /// FNV-1a 64 over the frame identity + canonical landing content. + /// + /// **The EXACT frame — `base_version` included — is part of the + /// idempotency identity, deliberately.** A retry after a lost + /// acknowledgement must resubmit the SAME frozen [`DetachedCycleBatch`], + /// never re-freeze from a re-derived frame: a caller that re-reads the + /// head and re-freezes has changed what it is asserting, and reconciling + /// that as "the same batch" would launder the divergence. Such a resubmit + /// fails closed ([`CommitError::HashConflict`] / `Fenced`) by design. fn content_hash(frame: CycleFrame, canonical: &[SweepSlot]) -> u64 { const OFFSET: u64 = 0xcbf2_9ce4_8422_2325; const PRIME: u64 = 0x0000_0100_0000_01b3; @@ -533,6 +557,14 @@ pub trait WalSink { /// canonical order — this seam does NOT sort. `after_cycle` bounds the read /// to cycles strictly after it (the recovery tail bound); implementations /// push the bound into the storage scan, never full-scan-and-filter. + /// + /// **Returned landings carry NO payload** (`SweepSlot::payload` is empty): + /// landing rows are transition METADATA; durable payloads live only in the + /// coalesced image read. Consequently an empty payload on a recovered + /// landing does NOT mean "intent-only" — that classification (the artifact + /// gate) applies to live casts in [`persist_cycle`], never to recovered + /// slots. Fakes must model this too, or they prove a property the real + /// writer does not hold. async fn scan_sealed( &self, after_cycle: Option, @@ -589,6 +621,9 @@ pub async fn persist_cycle( .filter(|c| !c.payload.is_empty()) .collect(); if artifacts.is_empty() { + // The sink is deliberately NOT called, so this head is the caller's + // asserted `frame.base_version`, never a fresh store read. No fence + // runs, which is sound only because nothing is written. return Ok(CommitOutcome::NoChange { head: frame.base_version, }); @@ -860,6 +895,8 @@ mod tests { after_cycle: Option, ) -> Result, WriteFailed> { // Returned in STORED order — no sort. (The order was fixed at seal.) + // Payload-free by contract: landing rows are metadata; the fake + // must model the production projection, not improve on it. Ok(self .sealed .lock() @@ -869,7 +906,10 @@ mod tests { .flat_map(|s| { s.landings.iter().map(|slot| LandedSlot { cycle: s.frame.cycle, - slot: slot.clone(), + slot: SweepSlot { + payload: Vec::new(), + ..slot.clone() + }, }) }) .collect()) diff --git a/crates/lance-graph-supervisor/examples/measure_wal_curve.rs b/crates/lance-graph-supervisor/examples/measure_wal_curve.rs index 0e7de05d..e70e7251 100644 --- a/crates/lance-graph-supervisor/examples/measure_wal_curve.rs +++ b/crates/lance-graph-supervisor/examples/measure_wal_curve.rs @@ -798,7 +798,8 @@ mod measure { cycle: frame.cycle, batch_hash: 0, }, - version: Some(version), + publication_version: Some(version), + observed_head: Some(version), transitions, next_position_base, } @@ -1454,12 +1455,13 @@ mod measure { batch: DetachedCycleBatch, ) -> Result { let mut sealed = self.sealed.lock().expect("MemWal poisoned"); + let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); // Reconciliation-first: an already-durable (cycle, hash) is success, // a matching cycle with a different hash fails closed. if let Some(rec) = sealed.iter().find(|s| s.frame.cycle == batch.frame.cycle) { return if rec.batch_hash == batch.batch_hash { Ok(CommitOutcome::Reconciled { - current_head: rec.version, + current_head: head, cycle: batch.frame.cycle, batch_hash: batch.batch_hash, }) @@ -1471,7 +1473,6 @@ mod measure { }) }; } - let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); if batch.frame.base_version != head { return Err(CommitError::Fenced { current_head: head }); } @@ -1504,7 +1505,12 @@ mod measure { .flat_map(|s| { s.landings.iter().map(|slot| LandedSlot { cycle: s.frame.cycle, - slot: slot.clone(), + // `scan_sealed` is payload-free by contract; payloads + // live in the image read. + slot: SweepSlot { + payload: Vec::new(), + ..slot.clone() + }, }) }) .collect()) diff --git a/crates/lance-graph-supervisor/src/cycle_driver.rs b/crates/lance-graph-supervisor/src/cycle_driver.rs index a0dbef9d..4b7d149b 100644 --- a/crates/lance-graph-supervisor/src/cycle_driver.rs +++ b/crates/lance-graph-supervisor/src/cycle_driver.rs @@ -131,12 +131,32 @@ pub struct SealedCycle { /// governing storage rule a cycle with zero artifact casts is /// [`CommitOutcome::NoChange`] and publishes nothing. pub outcome: CommitOutcome, - /// The version this cycle sealed into, or `None` for - /// [`CommitOutcome::NoChange`] (nothing published, head unchanged). - /// Derived from [`Self::outcome`] at construction — kept as a plain field - /// (not a method) so existing `sealed.version` call sites stay readable; - /// `Some` for [`CommitOutcome::Committed`] / [`CommitOutcome::Reconciled`]. - pub version: Option, + /// The PHYSICAL publication version — `Some` ONLY when THIS call + /// published (a fresh [`CommitOutcome::Committed`]). `None` on + /// [`CommitOutcome::NoChange`] (nothing written) **and on + /// [`CommitOutcome::Reconciled`]**: a reconciled batch was already + /// durable, its original publication position is not known from the + /// outcome, and it is never invented here — the durable identity is + /// `(cycle, batch_hash)`; the exact position is recoverable from the + /// version history on the audit path. (The previous field, `version`, + /// adopted `Reconciled.current_head` — so a retry of cycle 1 while the + /// head stood at V5 recorded cycle 1 as "sealed into V5", exactly the + /// audit corruption `current_head`'s naming warns against.) + pub publication_version: Option, + /// The store head actually OBSERVED by this outcome — `Some` only when + /// the sink was called and reported one: the publication head on + /// `Committed`, the reconciliation-time head on `Reconciled`. + /// + /// **`None` on [`CommitOutcome::NoChange`]**, deliberately: that outcome + /// calls NO sink, so its `head` is the caller's ASSERTED + /// `frame.base_version` and nothing observed it. Carrying it here as an + /// "observed" head would let a stale caller-assertion masquerade as a + /// store reading — the asserted base stays available on the frame, where + /// its provenance is legible. + /// + /// A read horizon in every case — NEVER an artifact's publication + /// position (that is [`Self::publication_version`]). + pub observed_head: Option, /// Only the owners that cast a `paired_move` — the sparse subset. /// With the pre-seal ≤1-per-owner partition, at most one per owner. /// **Computed over ALL collected casts, not just artifact ones** — an @@ -190,10 +210,12 @@ pub struct SealFailure { /// P4b output — the effect of applying a sealed cycle's sparse transition set. #[derive(Debug, Clone, PartialEq, Eq)] pub struct AppliedCycle { - /// The version whose sealed transitions were applied — mirrors - /// [`SealedCycle::version`]: `None` for a [`CommitOutcome::NoChange`] - /// cycle (an all-intent-only cycle can still apply sparse transitions to - /// the in-memory fleet even though nothing was published to the store). + /// The PHYSICAL publication version whose sealed transitions were applied + /// — mirrors [`SealedCycle::publication_version`]: `None` for a + /// [`CommitOutcome::NoChange`] cycle (an all-intent-only cycle can still + /// apply sparse transitions to the in-memory fleet even though nothing was + /// published) and for a [`CommitOutcome::Reconciled`] retry (already + /// durable; the position is never invented from the current head). pub version: Option, /// One move per **advanced** owner (distinct owners; ≤1 per cycle). pub applied: Vec, @@ -391,16 +413,21 @@ pub async fn seal_cycle( let frozen = casts.clone(); match persist_cycle(sink, frame, casts).await { Ok(outcome) => { - let version = match outcome { - CommitOutcome::NoChange { .. } => None, - CommitOutcome::Committed { version, .. } => Some(version), - // `current_head` is the store head AT RECONCILIATION TIME, not - // the publication version this cycle originally committed at. - CommitOutcome::Reconciled { current_head, .. } => Some(current_head), + // Publication is NEVER invented: only a fresh Committed carries + // one. Reconciled's `current_head` is the head NOW — adopting it + // as this cycle's version is the audit corruption the field's + // name warns against. + let (publication_version, observed_head) = match outcome { + // No sink call ⇒ nothing observed. The caller's asserted base + // stays on `frame`, never laundered into an observation. + CommitOutcome::NoChange { .. } => (None, None), + CommitOutcome::Committed { version, .. } => (Some(version), Some(version)), + CommitOutcome::Reconciled { current_head, .. } => (None, Some(current_head)), }; Ok(SealedCycle { outcome, - version, + publication_version, + observed_head, transitions, next_position_base, }) @@ -447,7 +474,7 @@ pub fn apply_sealed_transitions( let mut missing = 0usize; let partial = |applied: Vec, deferred, missing| AppliedCycle { - version: sealed.version, + version: sealed.publication_version, applied, deferred, missing, @@ -813,9 +840,39 @@ pub struct FleetRecovery { /// or past this cycle until those owners have recovered — advancing the /// global bound over an unrecovered latecomer's tail silences it /// permanently. `None` = no foreign landings in the scanned tail. + /// [`Self::checkpoint_bound`] is the enforceable form of this rule. pub foreign_min_cycle: Option, } +impl FleetRecovery { + /// The `after_cycle` bound the caller may DURABLY checkpoint after this + /// pass — the ENFORCEABLE form of the latecomer fence, replacing the + /// doc-comment instruction with an API the caller cannot mis-read. + /// + /// `recovered_through` is the highest cycle this pass fully recovered + /// (what the caller would naively store). With no foreign landings it + /// passes through unchanged. With a foreign landing first seen at cycle + /// `c`, the bound is capped strictly BELOW `c` (`c − 1`, or `None` when + /// `c` is the first cycle) — so the next `scan_sealed(bound)` still + /// returns the unrecovered latecomer's tail instead of silencing it. + #[must_use] + pub fn checkpoint_bound(&self, recovered_through: Option) -> Option { + match (recovered_through, self.foreign_min_cycle) { + (rt, None) => rt, + (None, Some(_)) => None, + (Some(rt), Some(f)) => { + if rt.0 < f.0 { + Some(rt) + } else if f.0 == 0 { + None + } else { + Some(CycleId(f.0 - 1)) + } + } + } + } +} + /// **P4e — COMMITTED-HISTORY recovery ONLY.** Valid solely when a commit /// SUCCEEDED (`Vn+1` exists as a sealed landing) and its application — or a /// restart — was interrupted. It is NEVER the path for an ordinary pre-commit @@ -1031,12 +1088,14 @@ mod tests { ))); } let mut sealed = self.sealed.lock().unwrap(); + let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); // Reconciliation-first: an already-durable (cycle, hash) is success, - // a matching cycle with a different hash fails closed. + // a matching cycle with a different hash fails closed. current_head + // is the store head NOW, never the original publication version. if let Some(rec) = sealed.iter().find(|s| s.frame.cycle == batch.frame.cycle) { return if rec.batch_hash == batch.batch_hash { Ok(CommitOutcome::Reconciled { - current_head: rec.version, + current_head: head, cycle: batch.frame.cycle, batch_hash: batch.batch_hash, }) @@ -1048,7 +1107,6 @@ mod tests { }) }; } - let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); if batch.frame.base_version != head { return Err(CommitError::Fenced { current_head: head }); } @@ -1072,6 +1130,8 @@ mod tests { after_cycle: Option, ) -> Result, WriteFailed> { self.reads.fetch_add(1, Ordering::SeqCst); + // Payload-free by contract: landing rows are metadata; the fake + // must model the production projection, not improve on it. Ok(self .sealed .lock() @@ -1081,7 +1141,10 @@ mod tests { .flat_map(|s| { s.landings.iter().map(|slot| LandedSlot { cycle: s.frame.cycle, - slot: slot.clone(), + slot: SweepSlot { + payload: Vec::new(), + ..slot.clone() + }, }) }) .collect()) @@ -1145,7 +1208,7 @@ mod tests { .unwrap(); assert_eq!(sink.wal_writes(), 1, "100 casts → exactly ONE WAL write"); assert_eq!( - sealed.version, + sealed.publication_version, Some(DatasetVersion(1)), "→ exactly one version" ); @@ -1260,7 +1323,7 @@ mod tests { // Exactly one Vn+1; exactly the represented owners advance once. assert_eq!(sink.wal_writes(), 1, "exactly one successful WAL write"); - assert_eq!(out.sealed.version, Some(DatasetVersion(1))); + assert_eq!(out.sealed.publication_version, Some(DatasetVersion(1))); assert_eq!(out.applied.applied.len(), 2); assert_eq!(fleet[&3].phase(), KanbanColumn::CognitiveWork); assert_eq!(fleet[&8].phase(), KanbanColumn::CognitiveWork); @@ -1303,12 +1366,24 @@ mod tests { let sealed = seal_cycle(&mut sink, failure.frame, failure.casts) .await .expect("retry succeeds"); - assert_eq!(sealed.version, Some(DatasetVersion(1))); + assert_eq!(sealed.publication_version, Some(DatasetVersion(1))); assert_eq!(sink.wal_writes(), 1, "one successful WAL write total"); - // Byte-identical: what landed is exactly the frozen set. + // What landed is exactly the frozen set's transition metadata. + // scan_sealed is payload-free by contract; payloads live in the + // image read — so the comparison is metadata-only. let landed = sink.scan_sealed(None).await.unwrap(); assert_eq!(landed.len(), 1); - assert_eq!(landed[0].slot, frozen_copy[0], "no cast lost or mutated"); + let (got, want) = (&landed[0].slot, &frozen_copy[0]); + assert_eq!( + (got.owner, got.stream_position, got.row, &got.paired_move), + ( + want.owner, + want.stream_position, + want.row, + &want.paired_move + ), + "no cast lost or mutated" + ); let applied = apply_sealed_transitions(&mut fleet, &sealed, &mut wm).unwrap(); assert_eq!(applied.applied.len(), 1, "owner advances exactly once"); @@ -1622,7 +1697,8 @@ mod tests { cycle: CycleId(1), batch_hash: 0, }, - version: Some(DatasetVersion(1)), + publication_version: Some(DatasetVersion(1)), + observed_head: Some(DatasetVersion(1)), transitions: vec![SealedTransition { stream_position: 0, owner: 7, @@ -1651,7 +1727,8 @@ mod tests { cycle: CycleId(1), batch_hash: 0, }, - version: Some(DatasetVersion(1)), + publication_version: Some(DatasetVersion(1)), + observed_head: Some(DatasetVersion(1)), transitions: vec![ SealedTransition { stream_position: 0, @@ -1690,7 +1767,8 @@ mod tests { cycle: CycleId(1), batch_hash: 0, }, - version: Some(DatasetVersion(1)), + publication_version: Some(DatasetVersion(1)), + observed_head: Some(DatasetVersion(1)), transitions: vec![SealedTransition { stream_position: 0, owner: 99, @@ -1727,7 +1805,7 @@ mod tests { .await .unwrap(); - assert_eq!(out.sealed.version, Some(DatasetVersion(1))); + assert_eq!(out.sealed.publication_version, Some(DatasetVersion(1))); assert_eq!( out.applied.applied.len(), 2, @@ -1807,7 +1885,7 @@ mod tests { ) .await .unwrap(); - assert_eq!(out2.sealed.version, Some(DatasetVersion(2))); + assert_eq!(out2.sealed.publication_version, Some(DatasetVersion(2))); assert_eq!( out2.applied.applied.len(), 1, @@ -2181,4 +2259,103 @@ mod tests { "a Hold is a reschedule, never a permanent strand" ); } + + // ── FALSIFIER (post-#912 review): a reconciled retry NEVER invents a + // publication version from the current head ────────────────────────── + // Cycle 1 → V1, cycle 2 → V2, then cycle 1's frozen batch is re-submitted + // (a lost acknowledgement). The retry reconciles at head V2 — and V2 must + // NEVER surface as cycle 1's publication. The durable identity is + // `(cycle, batch_hash)`; the position is an audit-path read, not a field. + #[tokio::test] + async fn a_reconciled_retry_never_reports_the_current_head_as_publication() { + let mut sink = FakeWalSink::new(); + let mut w1 = writer_with_moves(&[1]); + let c1 = collect_casts(&mut w1, CycleId(1), 0, u64::from); + let frozen_c1 = c1.slots.clone(); + let first = seal_cycle( + &mut sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + c1.slots, + ) + .await + .expect("cycle 1 seals"); + assert_eq!(first.publication_version, Some(DatasetVersion(1))); + + let mut w2 = writer_with_moves(&[2]); + let c2 = collect_casts(&mut w2, CycleId(2), first.next_position_base, u64::from); + let second = seal_cycle( + &mut sink, + CycleFrame::new(CycleId(2), DatasetVersion(1)), + c2.slots, + ) + .await + .expect("cycle 2 seals"); + assert_eq!(second.publication_version, Some(DatasetVersion(2))); + + // The lost-ack retry: same frozen cycle-1 batch, unchanged frame. + let retried = seal_cycle( + &mut sink, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + frozen_c1, + ) + .await + .expect("retry reconciles"); + let CommitOutcome::Reconciled { + current_head, + cycle, + .. + } = retried.outcome + else { + panic!("expected Reconciled, got {:?}", retried.outcome); + }; + assert_eq!(cycle, CycleId(1), "the durable identity names cycle 1"); + assert_eq!( + current_head, + DatasetVersion(2), + "the observed head is V2 (the head NOW)" + ); + assert_eq!(retried.observed_head, Some(DatasetVersion(2))); + assert_eq!( + retried.publication_version, None, + "V2 is the head at reconciliation time, NOT cycle 1's publication \ + — inventing one here is the audit corruption this field forbids" + ); + assert_eq!(sink.wal_writes(), 2, "the retry appended nothing"); + } + + // ── FALSIFIER (post-#912 review): checkpoint_bound enforces the + // latecomer fence, both ways ───────────────────────────────────────── + #[test] + fn checkpoint_bound_caps_below_foreign_and_passes_through_without() { + let clean = FleetRecovery { + total_applied: 3, + owners_recovered: 2, + foreign_landings: 0, + foreign_min_cycle: None, + }; + // Silence half: without foreign landings the naive bound passes. + assert_eq!( + clean.checkpoint_bound(Some(CycleId(7))), + Some(CycleId(7)), + "no foreign landings — the recovered-through bound is safe as-is" + ); + let fenced = FleetRecovery { + foreign_min_cycle: Some(CycleId(5)), + foreign_landings: 1, + ..clean + }; + // Firing half: recovered through 7, but a latecomer's landing sits at + // cycle 5 — checkpointing 7 (or 5) would exclude it from every future + // bounded scan. The bound is capped strictly below. + assert_eq!(fenced.checkpoint_bound(Some(CycleId(7))), Some(CycleId(4))); + // A bound already below the fence is untouched. + assert_eq!(fenced.checkpoint_bound(Some(CycleId(3))), Some(CycleId(3))); + // A foreign landing in the FIRST cycle leaves nothing safe to bound. + let at_zero = FleetRecovery { + foreign_min_cycle: Some(CycleId(0)), + ..fenced + }; + assert_eq!(at_zero.checkpoint_bound(Some(CycleId(7))), None); + assert_eq!(at_zero.checkpoint_bound(None), None); + } } diff --git a/crates/lance-graph-supervisor/tests/d_ign_b_lenses.rs b/crates/lance-graph-supervisor/tests/d_ign_b_lenses.rs index 319d13cd..f690d2ad 100644 --- a/crates/lance-graph-supervisor/tests/d_ign_b_lenses.rs +++ b/crates/lance-graph-supervisor/tests/d_ign_b_lenses.rs @@ -467,12 +467,13 @@ mod d_ign_b_lenses { batch: DetachedCycleBatch, ) -> Result { let mut sealed = self.sealed.lock().expect("MemWal poisoned"); + let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); // Reconciliation-first: an already-durable (cycle, hash) is success, // a matching cycle with a different hash fails closed. if let Some(rec) = sealed.iter().find(|s| s.frame.cycle == batch.frame.cycle) { return if rec.batch_hash == batch.batch_hash { Ok(CommitOutcome::Reconciled { - current_head: rec.version, + current_head: head, cycle: batch.frame.cycle, batch_hash: batch.batch_hash, }) @@ -484,7 +485,6 @@ mod d_ign_b_lenses { }) }; } - let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); if batch.frame.base_version != head { return Err(CommitError::Fenced { current_head: head }); } @@ -517,7 +517,12 @@ mod d_ign_b_lenses { .flat_map(|s| { s.landings.iter().map(|slot| LandedSlot { cycle: s.frame.cycle, - slot: slot.clone(), + // `scan_sealed` is payload-free by contract; payloads + // live in the image read. + slot: SweepSlot { + payload: Vec::new(), + ..slot.clone() + }, }) }) .collect()) diff --git a/crates/lance-graph-supervisor/tests/probe_ignition.rs b/crates/lance-graph-supervisor/tests/probe_ignition.rs index 06f3f7cf..5fb24351 100644 --- a/crates/lance-graph-supervisor/tests/probe_ignition.rs +++ b/crates/lance-graph-supervisor/tests/probe_ignition.rs @@ -373,12 +373,13 @@ mod probe_ignition { batch: DetachedCycleBatch, ) -> Result { let mut sealed = self.sealed.lock().expect("MemWal poisoned"); + let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); // Reconciliation-first: an already-durable (cycle, hash) is success, // a matching cycle with a different hash fails closed. if let Some(rec) = sealed.iter().find(|s| s.frame.cycle == batch.frame.cycle) { return if rec.batch_hash == batch.batch_hash { Ok(CommitOutcome::Reconciled { - current_head: rec.version, + current_head: head, cycle: batch.frame.cycle, batch_hash: batch.batch_hash, }) @@ -390,7 +391,6 @@ mod probe_ignition { }) }; } - let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); if batch.frame.base_version != head { return Err(CommitError::Fenced { current_head: head }); } @@ -424,7 +424,12 @@ mod probe_ignition { .flat_map(|s| { s.landings.iter().map(|slot| LandedSlot { cycle: s.frame.cycle, - slot: slot.clone(), + // `scan_sealed` is payload-free by contract; payloads + // live in the image read. + slot: SweepSlot { + payload: Vec::new(), + ..slot.clone() + }, }) }) .collect()) diff --git a/crates/lance-graph-supervisor/tests/probe_ignition_64k.rs b/crates/lance-graph-supervisor/tests/probe_ignition_64k.rs index 2c48d165..692acaa7 100644 --- a/crates/lance-graph-supervisor/tests/probe_ignition_64k.rs +++ b/crates/lance-graph-supervisor/tests/probe_ignition_64k.rs @@ -159,12 +159,13 @@ mod probe_ignition_64k { batch: DetachedCycleBatch, ) -> Result { let mut sealed = self.sealed.lock().expect("MemWal poisoned"); + let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); // Reconciliation-first: an already-durable (cycle, hash) is success, // a matching cycle with a different hash fails closed. if let Some(rec) = sealed.iter().find(|s| s.frame.cycle == batch.frame.cycle) { return if rec.batch_hash == batch.batch_hash { Ok(CommitOutcome::Reconciled { - current_head: rec.version, + current_head: head, cycle: batch.frame.cycle, batch_hash: batch.batch_hash, }) @@ -176,7 +177,6 @@ mod probe_ignition_64k { }) }; } - let head = sealed.last().map_or(DatasetVersion(0), |s| s.version); if batch.frame.base_version != head { return Err(CommitError::Fenced { current_head: head }); } @@ -209,7 +209,12 @@ mod probe_ignition_64k { .flat_map(|s| { s.landings.iter().map(|slot| LandedSlot { cycle: s.frame.cycle, - slot: slot.clone(), + // `scan_sealed` is payload-free by contract; payloads + // live in the image read. + slot: SweepSlot { + payload: Vec::new(), + ..slot.clone() + }, }) }) .collect()) diff --git a/crates/lance-graph/src/graph/cycle_sink.rs b/crates/lance-graph/src/graph/cycle_sink.rs index e711c923..edfe77a1 100644 --- a/crates/lance-graph/src/graph/cycle_sink.rs +++ b/crates/lance-graph/src/graph/cycle_sink.rs @@ -118,7 +118,7 @@ pub const EPISODIC_WITNESS_BYTES: usize = 512; /// |---|---|---|---| /// | `kind` | 0 | 1 | 2 | /// | `cycle` / `base_version` / `batch_hash` | ✓ | ✓ | ✓ | -/// | `stream_position` / `owner` / `row` | 0 | ✓ | winner's / 0 / row | +/// | `stream_position` / `owner` / `row` | 0 | ✓ | 0 / 0 / row | /// | `move_*` (nullable) | null | cast's move | null | /// | `payload` (`FixedSizeBinary(512)`, nullable) | null | null | final image | pub fn cycle_store_schema() -> SchemaRef { @@ -167,30 +167,116 @@ pub struct LanceCycleWriter { /// ambiguity resolution. Instrumented so "zero scans on the normal path" /// is measured, not asserted. reconcile_scans: AtomicU64, - /// The highest cycle known durable in THIS store (seeded at open from one - /// bounded frame scan; advanced in memory on every commit). Monotonic — - /// `cycle > committed_through` proves the cycle cannot already be durable, - /// which is what makes the scan-free fast path sound. + /// The highest cycle known durable in THIS store (seeded at open from a + /// frame-projected streaming fold — an O(#cycles) metadata scan with O(1) + /// memory, startup hydration, never a normal-path read; advanced in + /// memory on every commit). Monotonic — `cycle > committed_through` + /// proves the cycle cannot already be durable, which is what makes the + /// scan-free fast path sound. committed_through: Option, + /// The RAII registry claim. Held from BEFORE `open`'s first `.await`, so + /// a cancelled or failed `open` releases its slot through `Drop` — no + /// manual removal on any path, no leaked reservation. + /// + /// Never read: its `Drop` IS its behaviour, and the value must live + /// exactly as long as the writer. + #[allow(dead_code)] + claim: WriterClaim, + /// Test-only fault injection for the append / reopen / reconcile branch + /// (the branch that carries the whole no-rollback contract and cannot be + /// made to fail deterministically through real Lance). + #[cfg(test)] + fault: TestFaults, } /// The process-local single-writer registry: one LIVE [`LanceCycleWriter`] per -/// dataset path. `non-Clone + &mut self` serializes commits on one instance; +/// store IDENTITY. `non-Clone + &mut self` serializes commits on one instance; /// this registry closes the remaining in-process hole (a second `open` of the -/// same path is REFUSED while the first writer lives). Cross-PROCESS +/// same store is REFUSED while the first writer lives). Cross-PROCESS /// exclusivity remains a deployment lease this crate cannot enforce — stated, /// not implied away. static OPEN_WRITERS: std::sync::LazyLock>> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new())); -impl Drop for LanceCycleWriter { +/// The LEXICAL store identity the registry keys on: `.` segments, duplicate +/// separators and trailing slashes are collapsed, so `x/./cycles.lance`, +/// `x//cycles.lance` and `x/cycles.lance` claim ONE slot. A URI's +/// `scheme://authority` prefix is preserved verbatim. Deliberately NOT +/// filesystem canonicalization: `..` and symlinks are left alone (resolving +/// them needs I/O and still cannot cover object stores) — two spellings that +/// only a symlink makes equal remain the deployment lease's problem, same as +/// two processes. +fn store_identity(path: &str) -> String { + let (prefix, rest) = match path.find("://") { + Some(i) => { + let after_scheme = i + 3; + match path[after_scheme..].find('/') { + Some(j) => path.split_at(after_scheme + j), + None => (path, ""), + } + } + None => ("", path), + }; + let absolute = rest.starts_with('/'); + let parts: Vec<&str> = rest + .split('/') + .filter(|seg| !seg.is_empty() && *seg != ".") + .collect(); + let mut s = String::from(prefix); + if absolute { + s.push('/'); + } + s.push_str(&parts.join("/")); + s +} + +/// An RAII claim on [`OPEN_WRITERS`]. Acquired synchronously BEFORE the first +/// `.await` in [`LanceCycleWriter::open`]; released by `Drop` — which covers +/// the error paths, the writer's own drop, AND an `open` future that is +/// cancelled mid-`Dataset::open` (previously a leaked reservation, because no +/// RAII owner existed yet at that point). +#[derive(Debug)] +struct WriterClaim(String); + +impl WriterClaim { + fn acquire(identity: String) -> Result { + let mut set = OPEN_WRITERS + .lock() + .map_err(|_| WriteFailed("writer registry poisoned".into()))?; + if !set.insert(identity.clone()) { + return Err(WriteFailed(format!( + "a live LanceCycleWriter already owns {identity} in this process — \ + one logical writer per store (drop it first)" + ))); + } + Ok(Self(identity)) + } +} + +impl Drop for WriterClaim { fn drop(&mut self) { if let Ok(mut set) = OPEN_WRITERS.lock() { - set.remove(&self.dataset_path); + set.remove(&self.0); } } } +/// Deterministic failure injection for the ambiguous-append branch — +/// test-only, one-shot flags (each `swap(false)`s when consumed). +#[cfg(test)] +#[derive(Debug, Default)] +struct TestFaults { + /// Fail the next append WITHOUT publishing anything (the store is + /// untouched — models a pre-manifest I/O failure). + fail_append_unpublished: std::sync::atomic::AtomicBool, + /// Perform the next append for real, then report it failed (the manifest + /// IS durable — models a lost acknowledgement inside one attempt). + fail_append_published: std::sync::atomic::AtomicBool, + /// Fail the next reconciliation read (`find_frame`) — models storage + /// unavailable while resolving an ambiguous append. + fail_reconcile_read: std::sync::atomic::AtomicBool, +} + impl LanceCycleWriter { /// Open the writer over `path` (local path or object-store URI). Performs /// the ONE startup open (plus one bounded, frame-projected seed scan when @@ -201,48 +287,23 @@ impl LanceCycleWriter { /// store whose schema is not this writer's layout — a pre-Phase-A (#911) /// store is REJECTED loudly, never silently reinterpreted. pub async fn open(path: impl Into) -> Result { - let dataset_path = path.into(); - { - let mut set = OPEN_WRITERS - .lock() - .map_err(|_| WriteFailed("writer registry poisoned".into()))?; - if !set.insert(dataset_path.clone()) { - return Err(WriteFailed(format!( - "a live LanceCycleWriter already owns {dataset_path} in this process — \ - one logical writer per store (drop it first)" - ))); - } - } + // The claim is taken on the LEXICAL identity, synchronously, before + // the first await — errors and cancellation below release it via + // RAII, and `x/./cycles.lance` cannot claim a second slot beside + // `x/cycles.lance`. The normalized identity is also what we open: + // the two spellings resolve to the same store, so I/O and identity + // must not diverge. + let dataset_path = store_identity(&path.into()); + let claim = WriterClaim::acquire(dataset_path.clone())?; let opens = AtomicU64::new(0); let ds = match Dataset::open(&dataset_path).await { Ok(ds) => { opens.fetch_add(1, Ordering::Relaxed); - let expected = cycle_store_schema(); - let got = ds.schema(); - for field in expected.fields() { - if got.field(field.name()).is_none() { - OPEN_WRITERS - .lock() - .ok() - .map(|mut s| s.remove(&dataset_path)); - return Err(WriteFailed(format!( - "store at {dataset_path} is missing column `{}` — not this \ - writer's layout (a pre-Phase-A store is rejected, not \ - reinterpreted; migrate or discard it explicitly)", - field.name() - ))); - } - } + Self::guard_schema(&dataset_path, &ds)?; Some(ds) } Err(lance::Error::DatasetNotFound { .. }) => None, - Err(e) => { - OPEN_WRITERS - .lock() - .ok() - .map(|mut s| s.remove(&dataset_path)); - return Err(WriteFailed(format!("open {dataset_path}: {e}"))); - } + Err(e) => return Err(WriteFailed(format!("open {dataset_path}: {e}"))), }; let mut w = Self { dataset_path, @@ -250,16 +311,85 @@ impl LanceCycleWriter { opens, reconcile_scans: AtomicU64::new(0), committed_through: None, + claim, + #[cfg(test)] + fault: TestFaults::default(), }; if w.ds.is_some() { - // One bounded, projected seed read: the highest durable cycle. - // This is startup hydration (allowed), not a normal-path scan. - let frames = w.timeline().await?; - w.committed_through = frames.iter().map(|f| f.cycle).max(); + // Startup hydration: a frame-projected STREAMING fold to the + // highest durable cycle — O(#cycles) metadata rows scanned, + // O(1) memory (nothing materialized), never a normal-path read. + w.committed_through = w.max_cycle().await?; } Ok(w) } + /// Refuse a store whose schema is not this writer's layout — names AND + /// types AND nullability, so a pre-Phase-A (#911) store, a hand-altered + /// column, or a same-name/different-type drift is REJECTED loudly, never + /// silently reinterpreted (I-LEGACY-API-FEATURE-GATED). + fn guard_schema(dataset_path: &str, ds: &Dataset) -> Result<(), WriteFailed> { + let expected = cycle_store_schema(); + let got = ds.schema(); + for field in expected.fields() { + let Some(g) = got.field(field.name()) else { + return Err(WriteFailed(format!( + "store at {dataset_path} is missing column `{}` — not this \ + writer's layout (a pre-Phase-A store is rejected, not \ + reinterpreted; migrate or discard it explicitly)", + field.name() + ))); + }; + if g.data_type() != *field.data_type() || g.nullable != field.is_nullable() { + return Err(WriteFailed(format!( + "store at {dataset_path} column `{}` is {:?} (nullable={}) but this \ + writer's layout requires {:?} (nullable={}) — rejected, not \ + reinterpreted", + field.name(), + g.data_type(), + g.nullable, + field.data_type(), + field.is_nullable() + ))); + } + } + Ok(()) + } + + /// The highest durable cycle, as a streaming fold over frame rows + /// (projection: `cycle`; filter: `kind = 0`). The startup seed for + /// [`Self::commit_cycle`]'s fast-path watermark. + async fn max_cycle(&self) -> Result, WriteFailed> { + let Some(ds) = self.ds.as_ref() else { + return Ok(None); + }; + let mut scan = ds.scan(); + scan.filter(&format!("kind = {KIND_FRAME}")) + .map_err(|e| WriteFailed(format!("filter: {e}")))?; + scan.project(&["cycle"]) + .map_err(|e| WriteFailed(format!("project: {e}")))?; + let mut stream = scan + .try_into_stream() + .await + .map_err(|e| WriteFailed(format!("scan: {e}")))?; + let mut max: Option = None; + while let Some(b) = stream + .try_next() + .await + .map_err(|e| WriteFailed(format!("scan: {e}")))? + { + let cycle: &UInt64Array = b + .column_by_name("cycle") + .and_then(|c| c.as_any().downcast_ref()) + .ok_or_else(|| WriteFailed("missing column cycle".into()))?; + for i in 0..cycle.len() { + let v = cycle.value(i); + max = Some(max.map_or(v, |m: u64| m.max(v))); + } + } + Ok(max.map(CycleId)) + } + /// The store's current head version (`0` = empty store) — the in-memory /// token the normal path references instead of reloading state. #[must_use] @@ -322,12 +452,14 @@ impl LanceCycleWriter { // Landing metadata rows — the sparse transition set, payload NULL. for s in &batch.landings { + // PERMANENT refusal, never Io: an ABI-malformed artifact would + // "fail, regenerate identically, fail" forever if reported as + // retryable I/O. Nothing durable has happened at this point. if s.payload.len() != EPISODIC_WITNESS_BYTES { - return Err(CommitError::Io(WriteFailed(format!( - "artifact payload for row {} is {} bytes, ABI requires {EPISODIC_WITNESS_BYTES}", - s.row, - s.payload.len() - )))); + return Err(CommitError::InvalidArtifact { + row: s.row, + len: s.payload.len(), + }); } push_common(KIND_LANDING, s.stream_position, s.owner, s.row); match &s.paired_move { @@ -386,6 +518,16 @@ impl LanceCycleWriter { /// Look this cycle's durable frame up (projected `cycle` + `batch_hash` /// under a `kind = 0 AND cycle = …` predicate) — the reconciliation read. async fn find_frame(&self, cycle: CycleId) -> Result, WriteFailed> { + #[cfg(test)] + if self + .fault + .fail_reconcile_read + .swap(false, Ordering::Relaxed) + { + return Err(WriteFailed( + "injected: reconciliation read unavailable".into(), + )); + } let Some(ds) = self.ds.as_ref() else { return Ok(None); }; @@ -453,6 +595,36 @@ impl LanceCycleWriter { } } +impl LanceCycleWriter { + /// The one real store mutation: Create on the first-ever commit, Append + /// afterwards. Errors are carried as strings — the caller's ambiguity + /// branch only ever forwards the text, and the test fault-injection seam + /// needs a constructible error type. + async fn raw_append(&mut self, record_batch: RecordBatch) -> Result<(), String> { + let schema = cycle_store_schema(); + let reader = RecordBatchIterator::new(vec![Ok(record_batch)], schema); + match self.ds.as_mut() { + None => match Dataset::write( + reader, + &self.dataset_path, + Some(WriteParams { + mode: WriteMode::Create, + ..Default::default() + }), + ) + .await + { + Ok(ds) => { + self.ds = Some(ds); + Ok(()) + } + Err(e) => Err(e.to_string()), + }, + Some(ds) => ds.append(reader, None).await.map_err(|e| e.to_string()), + } + } +} + impl WalSink for LanceCycleWriter { /// THE single durable commit for a whole cycle — reconciliation-first, /// fence second, append third; the outcome is fully honored. @@ -520,27 +692,31 @@ impl WalSink for LanceCycleWriter { } // 3. The single atomic Lance MVCC commit. let record_batch = Self::build_batch(&batch)?; - let schema = cycle_store_schema(); - let reader = RecordBatchIterator::new(vec![Ok(record_batch)], schema); - let append_result = match self.ds.as_mut() { - None => match Dataset::write( - reader, - &self.dataset_path, - Some(WriteParams { - mode: WriteMode::Create, - ..Default::default() - }), - ) - .await + #[cfg(test)] + let append_result: Result<(), String> = { + if self + .fault + .fail_append_unpublished + .swap(false, Ordering::Relaxed) { - Ok(ds) => { - self.ds = Some(ds); - Ok(()) + // Nothing touched the store — models a pre-manifest failure. + Err("injected: append failed before publish".into()) + } else if self + .fault + .fail_append_published + .swap(false, Ordering::Relaxed) + { + // The manifest IS durable; only the acknowledgement is lost. + match self.raw_append(record_batch).await { + Ok(()) => Err("injected: acknowledgement lost after publish".into()), + Err(e) => Err(e), } - Err(e) => Err(e), - }, - Some(ds) => ds.append(reader, None).await, + } else { + self.raw_append(record_batch).await + } }; + #[cfg(not(test))] + let append_result: Result<(), String> = self.raw_append(record_batch).await; match append_result { Ok(()) => { self.committed_through = Some( @@ -556,12 +732,11 @@ impl WalSink for LanceCycleWriter { batch_hash: batch.batch_hash, }) } - Err(e) => { + Err(cause) => { // The commit's outcome is UNKNOWN (the manifest may or may not // have published before the failure). Reconcile from storage: // reopen (counted; NEVER degrades an existing handle), then // look for our durable identity. - let cause = e.to_string(); if let Err(re) = self.reopen().await { return Err(CommitError::Ambiguous { cycle: batch.frame.cycle, @@ -1343,4 +1518,287 @@ mod tests { assert!(r.is_err(), "511 bytes must be refused: {r:?}"); assert_eq!(w.head(), DatasetVersion(0), "nothing written"); } + + // ── FALSIFIER (post-#912): the ABI refusal is PERMANENT, never Io ──────── + // A 511-byte payload must surface as `InvalidArtifact` — reporting it as + // retryable Io would send the driver into fail → regenerate-identically → + // fail, forever. Nothing may touch the store. + #[tokio::test] + async fn a_malformed_artifact_is_refused_permanently_not_as_retryable_io() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let mut w = LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(); + let mut bad = artifact(1, 0, 42, 0); + bad.payload.truncate(511); + let err = persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![bad], + ) + .await + .expect_err("a malformed artifact must be refused"); + let lance_graph_planner::persist_sink::PersistError::Commit(commit) = err else { + panic!("expected a commit-layer refusal, got {err:?}"); + }; + assert_eq!( + commit, + CommitError::InvalidArtifact { row: 0, len: 511 }, + "the refusal is the PERMANENT variant, never CommitError::Io" + ); + assert_eq!(w.head(), DatasetVersion(0), "nothing was written"); + // The identical batch keeps failing identically — no retry loop exit. + let again = persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![{ + let mut b = artifact(1, 0, 42, 0); + b.payload.truncate(511); + b + }], + ) + .await + .expect_err("still refused"); + assert!(matches!( + again, + lance_graph_planner::persist_sink::PersistError::Commit( + CommitError::InvalidArtifact { .. } + ) + )); + } + + // ── FALSIFIER (post-#912): store identity is lexical, not string-equal ─── + #[tokio::test] + async fn a_second_spelling_of_the_same_store_is_refused() { + let dir = tempfile::tempdir().unwrap(); + let base = dir.path().to_str().unwrap().to_string(); + let w = LanceCycleWriter::open(format!("{base}/cycles.lance")) + .await + .unwrap(); + for spelling in [ + format!("{base}/./cycles.lance"), + format!("{base}//cycles.lance"), + format!("{base}/cycles.lance/"), + ] { + let err = LanceCycleWriter::open(spelling.clone()) + .await + .expect_err("an alternate spelling of a live store must be refused"); + assert!( + err.to_string().contains("already owns"), + "{spelling}: {err}" + ); + } + drop(w); + // The paired silence: a DIFFERENT store is never refused. + let _other = LanceCycleWriter::open(format!("{base}/other.lance")) + .await + .expect("a distinct store opens freely"); + } + + // ── FALSIFIER (post-#912): a failed open releases its claim (RAII) ─────── + // The claim is held from before the first await; an `open` that errors + // (here: a path that exists but is not a dataset and not NotFound-shaped) + // must not leave the slot reserved. + #[tokio::test] + async fn a_failed_open_leaves_no_leaked_reservation() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + // First open succeeds (empty store), then drops. + drop( + LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(), + ); + // Same path opens again — the drop released the claim. + drop( + LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(), + ); + // A DIFFERENT path whose open FAILS (a file where a directory is + // expected) must release too: retrying it is refused for the same + // I/O reason, never with "already owns". + let bogus = dir.path().join("not-a-dataset"); + std::fs::write(&bogus, b"junk").unwrap(); + let bogus_file = bogus.join("cycles.lance"); + let p = bogus_file.to_str().unwrap(); + let e1 = LanceCycleWriter::open(p).await; + if let Ok(w) = e1 { + // Environment-dependent: object-store may report NotFound here, + // making this a legal empty store — then the claim path is + // already covered by the success/drop halves above. + drop(w); + return; + } + let e2 = LanceCycleWriter::open(p) + .await + .expect_err("still the underlying I/O failure"); + assert!( + !e2.to_string().contains("already owns"), + "the failed first open must have released its claim: {e2}" + ); + } + + // ── FALSIFIERS (post-#912): the ambiguous-append branch, deterministically ─ + // The branch carries the whole no-rollback contract; real Lance cannot be + // made to fail on demand, so the TestFaults seam injects each arm. + // (The HashConflict arm of THIS branch — append fails while the same cycle + // is durable with different content — requires a competing writer between + // fence and append, unrepresentable under the registry; it shares its + // constructor with the normal-path HashConflict falsifier.) + + /// Arm `Ok(None)`: append failed with provably NOTHING published → + /// `Io("nothing published")`, and the SAME frozen batch then commits. + #[tokio::test] + async fn injected_unpublished_append_error_is_refused_io_and_regenerable() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let mut w = LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(); + // A durable first cycle so the store exists. + persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .unwrap(); + let head = w.head(); + + w.fault + .fail_append_unpublished + .store(true, std::sync::atomic::Ordering::Relaxed); + let err = persist_cycle( + &mut w, + CycleFrame::new(CycleId(2), head), + vec![artifact(2, 1, 42, 1)], + ) + .await + .expect_err("the injected append failure surfaces"); + let lance_graph_planner::persist_sink::PersistError::Commit(CommitError::Io(io)) = &err + else { + panic!("expected Io(nothing published), got {err:?}"); + }; + assert!( + io.to_string().contains("nothing published"), + "the reconcile proved absence: {io}" + ); + assert_eq!(w.head(), head, "the store is untouched"); + + // Safe to regenerate: the same cycle now commits cleanly. + let out = persist_cycle( + &mut w, + CycleFrame::new(CycleId(2), head), + vec![artifact(2, 1, 42, 1)], + ) + .await + .unwrap(); + assert!(matches!(out, CommitOutcome::Committed { .. })); + } + + /// Arm `Ok(Some(hash ==))`: the append PUBLISHED but the acknowledgement + /// was lost → the same call reconciles to the durable identity, appending + /// nothing twice. THE no-rollback falsifier. + #[tokio::test] + async fn injected_published_append_error_reconciles_to_the_durable_identity() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let mut w = LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(); + persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .unwrap(); + let head = w.head(); + + w.fault + .fail_append_published + .store(true, std::sync::atomic::Ordering::Relaxed); + let out = persist_cycle( + &mut w, + CycleFrame::new(CycleId(2), head), + vec![artifact(2, 1, 42, 1)], + ) + .await + .expect("the lost acknowledgement reconciles WITHIN the same call"); + let CommitOutcome::Reconciled { + cycle, + current_head, + .. + } = out + else { + panic!("expected Reconciled, got {out:?}"); + }; + assert_eq!(cycle, CycleId(2)); + assert_eq!(current_head, w.head()); + // Exactly one durable frame for cycle 2 — nothing double-appended. + let frames = w.timeline().await.unwrap(); + assert_eq!( + frames.iter().filter(|f| f.cycle == CycleId(2)).count(), + 1, + "no rollback, no duplicate — the publication stands once" + ); + } + + /// Arm `Err(reconcile)`: append outcome unknown AND the reconciliation + /// read fails → `Ambiguous`; the SAME frozen batch resolves it later. + #[tokio::test] + async fn injected_reconcile_read_failure_after_append_error_is_ambiguous() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let mut w = LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(); + persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .unwrap(); + let head = w.head(); + + w.fault + .fail_append_unpublished + .store(true, std::sync::atomic::Ordering::Relaxed); + w.fault + .fail_reconcile_read + .store(true, std::sync::atomic::Ordering::Relaxed); + let err = persist_cycle( + &mut w, + CycleFrame::new(CycleId(2), head), + vec![artifact(2, 1, 42, 1)], + ) + .await + .expect_err("append unknown + reconcile down = Ambiguous"); + let lance_graph_planner::persist_sink::PersistError::Commit(CommitError::Ambiguous { + cycle, + cause, + .. + }) = &err + else { + panic!("expected Ambiguous, got {err:?}"); + }; + assert_eq!(*cycle, CycleId(2)); + assert!( + cause.contains("append failed") && cause.contains("reconciliation failed"), + "both halves named: {cause}" + ); + // Resolution: the SAME frozen batch — reconciliation-first proves it + // absent (nothing had published) and the append lands once. + let out = persist_cycle( + &mut w, + CycleFrame::new(CycleId(2), head), + vec![artifact(2, 1, 42, 1)], + ) + .await + .expect("re-submission resolves the ambiguity"); + assert!(matches!(out, CommitOutcome::Committed { .. })); + } } From 000e9e91eb109e8f78066cd6967dc36352250ffa Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 15:28:11 +0000 Subject: [PATCH 2/2] close the first-Create hole and stop the taxonomy folding one layer up MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review findings, all real. **First-Create ambiguity (the symmetric half of the append path).** A failed Append leaves `ds == Some`, so a reconciliation read has a store to answer from. A failed Create does not: afterwards `ds == None` is exactly as consistent with "the manifest published and this reopen cannot see it yet" as with "nothing happened" — and the writer was reading that as absence, i.e. `Io/nothing published`, i.e. an invitation to regenerate, i.e. a SECOND Create against an unresolved first one. `create_unknown` makes the doubt explicit and sticky. While it is set, NOT-FOUND is never proof of absence (`Ambiguous`), and `commit_cycle` refuses to Create again. It clears only when storage proves the store exists — a successful commit, a reconciled batch, or any readable frame. `bootstrap()` is the sanctioned way out: explicit infrastructure creation, separated from semantic cycle publication. It publishes an empty dataset and so moves the head — a batch frozen against the pre-bootstrap horizon comes back `Fenced`, which is honest and safe (nothing of it is durable). Four falsifiers: published Create + lost ack + visible reopen → Reconciled, one frame; published Create + lost ack + NOT-FOUND reopen → Ambiguous, and the same frozen batch later reconciles without a second dataset; unpublished Create → Ambiguous, retry refuses to Create, bootstrap resolves it, exactly one frame across the whole episode; bootstrap on an existing store publishes nothing. The injection now models a lost Create ack faithfully — the write reaches storage and the HANDLE never comes back. **The taxonomy no longer folds at the supervisor boundary.** The outer caller sees only `CycleError::Seal`, whose doc said "nothing published, regenerate". That is true of two of the five commit errors and harmful for two others: regenerating an `Ambiguous` cycle risks a second publication, regenerating an `InvalidArtifact` one loops forever on an identical malformed batch. `SealFailure::recovery() -> SealRecovery` carries the decision out — Regenerate / ResubmitFrozen / Permanent / Escalate — with a falsifier that checks all five causes and that the classifier is not a constant. **`store_identity` no longer touches the I/O path.** Normalizing the string we hand to Lance was a silent behaviour change on backends where spelling is significant: `s3://bucket//x` is a different object key from `s3://bucket/x`, and a UNC path's leading `\\server\share` does not survive separator collapsing. The normalized form is the registry key only; `Dataset::open` gets the caller's string verbatim. The doc now states what the lexical claim does NOT cover (`..`, symlinks, `file://` vs bare, object-store URI equivalence) rather than implying it away. Also: the schema guard rejects unknown EXTRA columns, not just missing ones; `AppliedCycle.version` is renamed `publication_version` (a bare `version` beside a `SealedCycle` that now distinguishes the two is the same ambiguity one layer up); `run_cycle`'s two contradictory borrow paragraphs are down to the honest one; `max_cycle` is documented as O(1) memory but still O(history) I/O, not "bounded". Tests: cycle_sink 23, cycle_driver 27, persist_sink 22, supervisor probes — green; fmt clean; clippy adds no new warnings. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01KCGhDYoQBXs3poaR7sFuqp --- .../src/cycle_driver.rs | 160 +++++- crates/lance-graph/src/graph/cycle_sink.rs | 454 +++++++++++++++++- 2 files changed, 574 insertions(+), 40 deletions(-) diff --git a/crates/lance-graph-supervisor/src/cycle_driver.rs b/crates/lance-graph-supervisor/src/cycle_driver.rs index 4b7d149b..167efbcb 100644 --- a/crates/lance-graph-supervisor/src/cycle_driver.rs +++ b/crates/lance-graph-supervisor/src/cycle_driver.rs @@ -101,8 +101,8 @@ use lance_graph_contract::QualiaI4_16D; use lance_graph_planner::batch_writer::BatchWriter; use lance_graph_planner::owner_adapter::emit_bootstrap_intent; use lance_graph_planner::persist_sink::{ - persist_cycle, recover_and_apply, CommitOutcome, CycleFrame, CycleId, LandedSlot, PersistError, - SweepSlot, WalSink, + persist_cycle, recover_and_apply, CommitError, CommitOutcome, CycleFrame, CycleId, LandedSlot, + PersistError, SweepSlot, WalSink, }; use lance_graph_planner::traits::StrategyOutcome; @@ -207,6 +207,48 @@ pub struct SealFailure { pub cause: PersistError, } +/// What a caller must DO about a [`SealFailure`] — the commit taxonomy +/// preserved as a decision, rather than flattened into "it failed". +/// +/// Exists because the outer caller sees only [`CycleError::Seal`]; without +/// this the four commit errors collapse into one recovery, and two of the +/// four recoveries are then wrong (an endless retry of a permanently +/// malformed batch, or a regeneration that may double-publish). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum SealRecovery { + /// Nothing published, PROVEN. Regenerate the cycle (refresh the + /// `base_version` first if the cause was a fence). + Regenerate, + /// Publication is UNKNOWN. Re-submit the SAME frozen + /// [`SealFailure::casts`] under the SAME [`SealFailure::frame`] — + /// reconciliation runs first, so the retry cannot double-append. + ResubmitFrozen, + /// PERMANENT. The batch can never commit as-is (an ABI-malformed + /// artifact); retrying is an infinite loop. Fix the producer. + Permanent, + /// Fail closed. This cycle is durable with DIFFERENT content — never + /// promoted, never overwritten; escalate. + Escalate, +} + +impl SealFailure { + /// Classify [`Self::cause`] into the action a caller must take. + #[must_use] + pub fn recovery(&self) -> SealRecovery { + match &self.cause { + PersistError::Commit(CommitError::Fenced { .. } | CommitError::Io(_)) => { + SealRecovery::Regenerate + } + PersistError::Commit(CommitError::Ambiguous { .. }) => SealRecovery::ResubmitFrozen, + PersistError::Commit(CommitError::InvalidArtifact { .. }) => SealRecovery::Permanent, + PersistError::Commit(CommitError::HashConflict { .. }) => SealRecovery::Escalate, + // Pre-commit guards (owner mismatch, stale phase, …): the batch + // never reached storage, so nothing is published. + _ => SealRecovery::Regenerate, + } + } +} + /// P4b output — the effect of applying a sealed cycle's sparse transition set. #[derive(Debug, Clone, PartialEq, Eq)] pub struct AppliedCycle { @@ -216,7 +258,11 @@ pub struct AppliedCycle { /// apply sparse transitions to the in-memory fleet even though nothing was /// published) and for a [`CommitOutcome::Reconciled`] retry (already /// durable; the position is never invented from the current head). - pub version: Option, + /// + /// Named in full deliberately: a bare `version` beside a `SealedCycle` + /// that now distinguishes publication from observed head is exactly the + /// ambiguity this arc removed one layer down. + pub publication_version: Option, /// One move per **advanced** owner (distinct owners; ≤1 per cycle). pub applied: Vec, /// Defence-in-depth counter: same-owner extras in a sealed input NOT @@ -474,7 +520,7 @@ pub fn apply_sealed_transitions( let mut missing = 0usize; let partial = |applied: Vec, deferred, missing| AppliedCycle { - version: sealed.publication_version, + publication_version: sealed.publication_version, applied, deferred, missing, @@ -547,11 +593,29 @@ pub struct CycleOutcome { /// A [`run_cycle`] failure. #[derive(Debug)] pub enum CycleError { - /// The WAL commit failed — **nothing published, no owner mutated, no - /// watermark advanced**. Held intents from this pass were discarded. - /// Prescribed recovery: rerun the unchanged Kanban task from `Vn` - /// (deterministic regeneration, module docs § Failure semantics); the - /// boxed [`SealFailure`] is an optional retry cache only. + /// The WAL commit did not yield a durable outcome. **No owner was + /// mutated and no watermark advanced**; held intents from this pass were + /// discarded. + /// + /// **The recovery is NOT uniform — read [`SealFailure::cause`], never + /// this variant alone.** An earlier version of this doc said "nothing + /// published, regenerate", which is true of only two of the four commit + /// errors and actively harmful for the other two: regenerating an + /// `Ambiguous` cycle risks a second publication, and regenerating an + /// `InvalidArtifact` one loops forever on an identical malformed batch. + /// [`SealFailure::recovery`] classifies it; the four cases are: + /// + /// | cause | published? | do | + /// |---|---|---| + /// | `Fenced` | no | regenerate against the current head | + /// | `Io` | no (proven) | regenerate unchanged | + /// | `Ambiguous` | UNKNOWN | re-submit the SAME frozen batch | + /// | `HashConflict` | yes, differently | fail closed — escalate | + /// | `InvalidArtifact` | no | PERMANENT — fix the producer, never retry | + /// + /// The boxed [`SealFailure`] carries the frozen casts, which are the + /// re-submission payload for `Ambiguous` and an optional retry cache + /// otherwise. Seal(Box), /// A guard tripped mid-apply — the applied prefix (with its watermarks /// already advanced) is preserved; re-drive the tail via [`recover_fleet`]. @@ -585,11 +649,12 @@ pub enum CycleError { /// place alongside the phases. On [`CycleError::Seal`] the frozen cycle is /// retryable; on [`CycleError::Apply`] the applied prefix is preserved. /// -/// **Borrow note (operator-ruled): `fleet: &mut F` is not touched across the -/// seal's `.await`.** The parameter's lifetime spans the whole function, but -/// the body only reads/writes through it in [`apply_sealed_transitions`], -/// AFTER `seal_cycle`'s I/O has already completed — the exclusive fleet -/// borrow is effectively taken post-I/O, not held live across the WAL commit. +/// (An earlier note here claimed the fleet borrow "is effectively taken +/// post-I/O" because the body only dereferences it after the seal. That is +/// true of the BODY and false of the SIGNATURE, which is what a caller is +/// bound by — the two paragraphs contradicted each other and the honest one +/// above is the one that holds. Removed rather than reconciled: a reader who +/// believed the second would use this as the production path.) pub async fn run_cycle( sink: &mut S, fleet: &mut F, @@ -1514,7 +1579,7 @@ mod tests { assert_eq!(applied.applied.len(), 17, "exactly 17 owners advanced"); assert_eq!(applied.deferred, 0); assert_eq!(applied.missing, 0); - assert_eq!(applied.version, Some(DatasetVersion(1))); + assert_eq!(applied.publication_version, Some(DatasetVersion(1))); assert_eq!(wm.len(), 17, "exactly 17 watermarks advanced"); // Every represented owner is now at CognitiveWork (cycle bumped); every @@ -2358,4 +2423,69 @@ mod tests { assert_eq!(at_zero.checkpoint_bound(Some(CycleId(7))), None); assert_eq!(at_zero.checkpoint_bound(None), None); } + + // ── FALSIFIER (post-#912 review): the commit taxonomy survives the + // supervisor boundary ──────────────────────────────────────────────── + // The outer caller sees only `CycleError::Seal`. If that flattened the + // four commit errors into one recovery, two of the four would be wrong: + // regenerating an Ambiguous cycle risks a second publication, and + // regenerating an InvalidArtifact one loops forever. + #[test] + fn seal_failure_recovery_separates_all_four_commit_errors() { + let frame = CycleFrame::new(CycleId(1), DatasetVersion(0)); + let fail = |cause| SealFailure { + frame, + casts: Vec::new(), + cause: PersistError::Commit(cause), + }; + assert_eq!( + fail(CommitError::Fenced { + current_head: DatasetVersion(3) + }) + .recovery(), + SealRecovery::Regenerate + ); + assert_eq!( + fail(CommitError::Io( + lance_graph_planner::persist_sink::WriteFailed("nothing published".into()) + )) + .recovery(), + SealRecovery::Regenerate + ); + assert_eq!( + fail(CommitError::Ambiguous { + cycle: CycleId(1), + batch_hash: 7, + cause: "unknown".into() + }) + .recovery(), + SealRecovery::ResubmitFrozen, + "an unknown publication must NEVER be regenerated" + ); + assert_eq!( + fail(CommitError::InvalidArtifact { row: 0, len: 511 }).recovery(), + SealRecovery::Permanent, + "a malformed batch retried is an infinite loop" + ); + assert_eq!( + fail(CommitError::HashConflict { + cycle: CycleId(1), + stored_hash: 1, + offered_hash: 2 + }) + .recovery(), + SealRecovery::Escalate + ); + // Anti-vacuity: the classifier discriminates — it is not a constant. + let all = [ + SealRecovery::Regenerate, + SealRecovery::ResubmitFrozen, + SealRecovery::Permanent, + SealRecovery::Escalate, + ]; + assert_eq!( + all.iter().collect::>().len(), + 4 + ); + } } diff --git a/crates/lance-graph/src/graph/cycle_sink.rs b/crates/lance-graph/src/graph/cycle_sink.rs index edfe77a1..5613ff8a 100644 --- a/crates/lance-graph/src/graph/cycle_sink.rs +++ b/crates/lance-graph/src/graph/cycle_sink.rs @@ -174,6 +174,14 @@ pub struct LanceCycleWriter { /// proves the cycle cannot already be durable, which is what makes the /// scan-free fast path sound. committed_through: Option, + /// **The first-`Create` doubt.** Set when a dataset-CREATING attempt + /// returned an unknown outcome; cleared the moment storage proves the + /// store exists (a successful commit, a reconciled batch, or any readable + /// frame). While it is set, `ds == None` no longer means "absent" — the + /// dataset may have been published by that very attempt — so + /// [`WalSink::commit_cycle`] refuses to run a SECOND `Create` and + /// `DatasetNotFound` is never read as proof of absence. + create_unknown: bool, /// The RAII registry claim. Held from BEFORE `open`'s first `.await`, so /// a cancelled or failed `open` releases its slot through `Drop` — no /// manual removal on any path, no leaked reservation. @@ -201,11 +209,15 @@ static OPEN_WRITERS: std::sync::LazyLock String { let (prefix, rest) = match path.find("://") { Some(i) => { @@ -275,6 +287,11 @@ struct TestFaults { /// Fail the next reconciliation read (`find_frame`) — models storage /// unavailable while resolving an ambiguous append. fail_reconcile_read: std::sync::atomic::AtomicBool, + /// Report the next `reopen` as `DatasetNotFound` even when the dataset + /// EXISTS — models the eventual-consistency window in which a manifest + /// this writer just published is not yet visible. This is precisely the + /// state in which NOT-FOUND must not be read as proof of absence. + fail_reopen_notfound: std::sync::atomic::AtomicBool, } impl LanceCycleWriter { @@ -290,11 +307,17 @@ impl LanceCycleWriter { // The claim is taken on the LEXICAL identity, synchronously, before // the first await — errors and cancellation below release it via // RAII, and `x/./cycles.lance` cannot claim a second slot beside - // `x/cycles.lance`. The normalized identity is also what we open: - // the two spellings resolve to the same store, so I/O and identity - // must not diverge. - let dataset_path = store_identity(&path.into()); - let claim = WriterClaim::acquire(dataset_path.clone())?; + // `x/cycles.lance`. + // + // **The normalized form is the REGISTRY KEY ONLY; I/O uses the + // caller's string verbatim.** Rewriting the path we open would be a + // silent behaviour change on backends where the spelling is + // significant: `s3://bucket//x` is a different object key from + // `s3://bucket/x`, and a UNC path's leading `\\server\share` does not + // survive separator collapsing. Identity may be approximate and + // conservative; the path we hand to Lance may not be touched at all. + let dataset_path = path.into(); + let claim = WriterClaim::acquire(store_identity(&dataset_path))?; let opens = AtomicU64::new(0); let ds = match Dataset::open(&dataset_path).await { Ok(ds) => { @@ -311,14 +334,18 @@ impl LanceCycleWriter { opens, reconcile_scans: AtomicU64::new(0), committed_through: None, + create_unknown: false, claim, #[cfg(test)] fault: TestFaults::default(), }; if w.ds.is_some() { // Startup hydration: a frame-projected STREAMING fold to the - // highest durable cycle — O(#cycles) metadata rows scanned, - // O(1) memory (nothing materialized), never a normal-path read. + // highest durable cycle. **O(1) MEMORY, still O(history) I/O** — + // nothing is materialized, but every frame row is read. It is not + // "bounded"; it is a one-per-process startup cost that grows with + // the store, and shrinking it needs a Lance tail/aggregate + // mechanism, never a second head ledger. w.committed_through = w.max_cycle().await?; } Ok(w) @@ -353,6 +380,19 @@ impl LanceCycleWriter { ))); } } + // Reject EXTRA columns too: a store carrying a field this writer does + // not know is not this writer's layout either, and appending a batch + // built from OUR schema against it either fails deep in Lance or + // silently nulls a column somebody else depends on. + for got_field in &got.fields { + if expected.field_with_name(&got_field.name).is_err() { + return Err(WriteFailed(format!( + "store at {dataset_path} carries UNKNOWN column `{}` — not this \ + writer's layout (rejected, not reinterpreted)", + got_field.name + ))); + } + } Ok(()) } @@ -564,6 +604,24 @@ impl LanceCycleWriter { /// `Ambiguous`, and the next commit can never fall into `Create` over a /// store that has history. async fn reopen(&mut self) -> Result<(), WriteFailed> { + #[cfg(test)] + if self + .fault + .fail_reopen_notfound + .swap(false, Ordering::Relaxed) + { + self.opens.fetch_add(1, Ordering::Relaxed); + return if self.ds.is_some() { + Err(WriteFailed(format!( + "reopen {}: injected NOT FOUND while holding history", + self.dataset_path + ))) + } else { + // The handle stays None — indistinguishable, from here, from + // a store that was never created. That is the whole point. + Ok(()) + }; + } match Dataset::open(&self.dataset_path).await { Ok(ds) => { self.opens.fetch_add(1, Ordering::Relaxed); @@ -586,6 +644,51 @@ impl LanceCycleWriter { } } + /// Explicitly create the empty cycle store — INFRASTRUCTURE creation, + /// deliberately separated from semantic cycle publication. + /// + /// This is the sanctioned way out of an unresolved first-`Create`: after + /// a creating attempt with an unknown outcome, [`WalSink::commit_cycle`] + /// refuses to `Create` again (it might publish a second dataset over one + /// this writer already published and never saw), so the doubt is resolved + /// by an operator/deployment action rather than by a data write. + /// + /// Idempotent in the direction that matters: if the store turns out to + /// EXIST, that existence resolves the doubt and nothing is written — this + /// never overwrites and never publishes a cycle. The dataset it creates + /// carries zero rows, so a subsequent commit reconciles honestly (no + /// frame ⇒ nothing landed). + /// + /// **It does move the head**, because creating the empty dataset IS a + /// Lance version. A batch frozen against the pre-bootstrap horizon is + /// therefore genuinely stale and comes back [`CommitError::Fenced`] + /// (nothing written) — regenerate it against [`Self::head`]. That is the + /// honest classification, not a wart: nothing of that batch is durable, so + /// regeneration is exactly the safe move. + pub async fn bootstrap(&mut self) -> Result<(), WriteFailed> { + if self.ds.is_none() { + self.reopen().await?; + } + if self.ds.is_none() { + let schema = cycle_store_schema(); + let empty = RecordBatchIterator::new(Vec::new(), schema); + let ds = Dataset::write( + empty, + &self.dataset_path, + Some(WriteParams { + mode: WriteMode::Create, + ..Default::default() + }), + ) + .await + .map_err(|e| WriteFailed(format!("bootstrap {}: {e}", self.dataset_path)))?; + self.ds = Some(ds); + } + // Storage has now shown the store to exist, by either route. + self.create_unknown = false; + Ok(()) + } + /// How many reconciliation scans ([`find_frame`](Self::find_frame)) this /// writer has EVER run. Zero across a run of fresh monotonic commits — /// the "zero scans on the normal path" falsifier reads this. @@ -640,6 +743,35 @@ impl WalSink for LanceCycleWriter { head: DatasetVersion(batch.frame.base_version.0), }); } + // 0. THE UNRESOLVED-CREATE GUARD. An earlier dataset-creating attempt + // returned an unknown outcome, so it is not known whether the store + // exists. Try ONE reopen to resolve it; if storage still cannot + // show the dataset, refuse — running `Create` again here is the one + // move that could produce two datasets (or clobber a manifest this + // writer published and never saw). Re-submitting the SAME frozen + // batch is the resolution: reconciliation runs first, so the retry + // cannot double-append once the store becomes readable. + if self.create_unknown { + if self.ds.is_none() { + let _ = self.reopen().await; + } + if self.ds.is_none() { + return Err(CommitError::Ambiguous { + cycle: batch.frame.cycle, + batch_hash: batch.batch_hash, + cause: format!( + "a previous Create on {} has an UNRESOLVED outcome and the store \ + is still not readable — refusing to Create a second time; \ + re-submit the SAME frozen batch", + self.dataset_path + ), + }); + } + // The store is readable: the doubt is over. Whether OUR batch + // landed is now an ordinary reconciliation question, answered + // below by cycle/hash — never by the handle's existence. + self.create_unknown = false; + } // 1. The scan-free FAST PATH decision. A fresh monotonic cycle // (`cycle > committed_through`, seeded at open) provably cannot be // durable yet, and a matching fence proves the horizon — so the @@ -692,6 +824,11 @@ impl WalSink for LanceCycleWriter { } // 3. The single atomic Lance MVCC commit. let record_batch = Self::build_batch(&batch)?; + // Was THIS attempt the dataset-CREATING one? A failed Create is not + // symmetric with a failed Append: afterwards `ds == None` no longer + // proves the store is absent, so a later `DatasetNotFound` must not + // be read as "nothing published". + let was_create = self.ds.is_none(); #[cfg(test)] let append_result: Result<(), String> = { if self @@ -707,8 +844,18 @@ impl WalSink for LanceCycleWriter { .swap(false, Ordering::Relaxed) { // The manifest IS durable; only the acknowledgement is lost. + // On a CREATE that also means the HANDLE never came back — + // `Dataset::write` returning an error leaves `ds == None` + // even though the dataset now exists. Modelling that is the + // whole point: it is the state in which NOT-FOUND is not + // proof of absence. match self.raw_append(record_batch).await { - Ok(()) => Err("injected: acknowledgement lost after publish".into()), + Ok(()) => { + if was_create { + self.ds = None; + } + Err("injected: acknowledgement lost after publish".into()) + } Err(e) => Err(e), } } else { @@ -718,7 +865,12 @@ impl WalSink for LanceCycleWriter { #[cfg(not(test))] let append_result: Result<(), String> = self.raw_append(record_batch).await; match append_result { + // (`was_create` above records whether THIS attempt was the + // dataset-creating one — the asymmetry the arms below turn on.) Ok(()) => { + // A successful commit resolves any earlier Create doubt: the + // store demonstrably exists and this batch is in it. + self.create_unknown = false; self.committed_through = Some( self.committed_through .map_or(batch.frame.cycle, |ct| ct.max(batch.frame.cycle)), @@ -734,9 +886,15 @@ impl WalSink for LanceCycleWriter { } Err(cause) => { // The commit's outcome is UNKNOWN (the manifest may or may not - // have published before the failure). Reconcile from storage: - // reopen (counted; NEVER degrades an existing handle), then - // look for our durable identity. + // have published before the failure). If this attempt was the + // CREATE, the doubt is sticky: it is now unknown whether the + // dataset exists at all, and no second Create may ever run + // against that doubt. + if was_create { + self.create_unknown = true; + } + // Reconcile from storage: reopen (counted; NEVER degrades an + // existing handle), then look for our durable identity. if let Err(re) = self.reopen().await { return Err(CommitError::Ambiguous { cycle: batch.frame.cycle, @@ -747,6 +905,8 @@ impl WalSink for LanceCycleWriter { self.reconcile_scans.fetch_add(1, Ordering::Relaxed); match self.find_frame(batch.frame.cycle).await { Ok(Some(stored_hash)) if stored_hash == batch.batch_hash => { + // The store is proven to exist and to hold this batch. + self.create_unknown = false; self.committed_through = Some( self.committed_through .map_or(batch.frame.cycle, |ct| ct.max(batch.frame.cycle)), @@ -757,15 +917,42 @@ impl WalSink for LanceCycleWriter { batch_hash: batch.batch_hash, }) } - Ok(Some(stored_hash)) => Err(CommitError::HashConflict { - cycle: batch.frame.cycle, - stored_hash, - offered_hash: batch.batch_hash, - }), - // Proven absent: nothing landed — safe to regenerate. - Ok(None) => Err(CommitError::Io(WriteFailed(format!( - "append failed with nothing published: {cause}" - )))), + Ok(Some(stored_hash)) => { + // A readable frame proves the store exists. + self.create_unknown = false; + Err(CommitError::HashConflict { + cycle: batch.frame.cycle, + stored_hash, + offered_hash: batch.batch_hash, + }) + } + // `find_frame` said "not there" — but that is only PROOF + // OF ABSENCE when a store was actually read. After a + // failed CREATE the store itself may or may not exist, and + // `DatasetNotFound` is exactly as consistent with "the + // manifest published and this reopen cannot see it yet" as + // with "nothing happened". Reporting Io there would invite + // a regenerate — i.e. a SECOND Create against an unknown + // first one. + Ok(None) if self.create_unknown && self.ds.is_none() => { + Err(CommitError::Ambiguous { + cycle: batch.frame.cycle, + batch_hash: batch.batch_hash, + cause: format!( + "create failed ({cause}) and the store is still not \ + readable: NOT-FOUND after an unresolved Create is not \ + proof of absence — re-submit the SAME frozen batch" + ), + }) + } + // Proven absent: a readable store without our frame — + // nothing landed, safe to regenerate. + Ok(None) => { + self.create_unknown = false; + Err(CommitError::Io(WriteFailed(format!( + "append failed with nothing published: {cause}" + )))) + } Err(re) => Err(CommitError::Ambiguous { cycle: batch.frame.cycle, batch_hash: batch.batch_hash, @@ -1801,4 +1988,221 @@ mod tests { .expect("re-submission resolves the ambiguity"); assert!(matches!(out, CommitOutcome::Committed { .. })); } + + // ── FALSIFIERS (post-#912): the FIRST-CREATE path, the symmetric half ──── + // A failed Append leaves `ds == Some`, so a reconciliation read has a + // store to answer from. A failed CREATE does not: afterwards `ds == None` + // is exactly as consistent with "the manifest published and this reopen + // cannot see it yet" as with "nothing happened". These four prove the + // writer never resolves that doubt by guessing. + + /// Create publishes, the acknowledgement is lost, and the reopen CAN see + /// the store → the durable identity is found → `Reconciled`, one frame. + #[tokio::test] + async fn a_published_create_with_a_lost_ack_reconciles_within_the_same_call() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let mut w = LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(); + assert_eq!(w.head(), DatasetVersion(0), "nothing exists yet"); + + w.fault + .fail_append_published + .store(true, std::sync::atomic::Ordering::Relaxed); + let out = persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .expect("the lost ack on a CREATE reconciles"); + assert!( + matches!(out, CommitOutcome::Reconciled { cycle, .. } if cycle == CycleId(1)), + "got {out:?}" + ); + let frames = w.timeline().await.unwrap(); + assert_eq!(frames.len(), 1, "exactly one dataset, exactly one frame"); + } + + /// Create publishes, the acknowledgement is lost, AND the reopen reports + /// NOT-FOUND (the eventual-consistency window) → `Ambiguous`, never + /// `Io/nothing published`. Re-submitting the same frozen batch once the + /// store is visible reconciles it — no second Create, no duplicate. + #[tokio::test] + async fn a_published_create_invisible_to_reopen_is_ambiguous_never_absent() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let mut w = LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(); + + w.fault + .fail_append_published + .store(true, std::sync::atomic::Ordering::Relaxed); + w.fault + .fail_reopen_notfound + .store(true, std::sync::atomic::Ordering::Relaxed); + let err = persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .expect_err("an invisible published Create cannot be called absent"); + let lance_graph_planner::persist_sink::PersistError::Commit(CommitError::Ambiguous { + cause, + .. + }) = &err + else { + panic!("NOT-FOUND after an unresolved Create must be Ambiguous, got {err:?}"); + }; + assert!( + cause.contains("not proof of absence"), + "the reason is named: {cause}" + ); + + // The resolution: the SAME frozen batch, once storage is visible. + let out = persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .expect("re-submission resolves it"); + assert!( + matches!(out, CommitOutcome::Reconciled { .. }), + "the batch WAS durable all along: {out:?}" + ); + let frames = w.timeline().await.unwrap(); + assert_eq!( + frames.len(), + 1, + "one frame — the retry never created a second dataset" + ); + } + + /// Create definitely does NOT publish → still `Ambiguous` (storage cannot + /// prove absence from here), and a retry REFUSES a second Create rather + /// than guessing. `bootstrap` is the sanctioned resolution; afterwards the + /// same frozen batch commits exactly once. + #[tokio::test] + async fn an_unpublished_create_refuses_a_second_create_until_bootstrap() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let mut w = LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(); + + w.fault + .fail_append_unpublished + .store(true, std::sync::atomic::Ordering::Relaxed); + let first = persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .expect_err("unknown Create outcome"); + assert!( + matches!( + first, + lance_graph_planner::persist_sink::PersistError::Commit( + CommitError::Ambiguous { .. } + ) + ), + "an unresolved Create is never Io/nothing-published: {first:?}" + ); + + // A retry must NOT run Create again — that is the move that could + // publish a second dataset over one already published. + let second = persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .expect_err("still unresolved"); + let lance_graph_planner::persist_sink::PersistError::Commit(CommitError::Ambiguous { + cause, + .. + }) = &second + else { + panic!("expected a refusal, got {second:?}"); + }; + assert!( + cause.contains("refusing to Create a second time"), + "the refusal is explicit: {cause}" + ); + assert_eq!(w.head(), DatasetVersion(0), "nothing was written"); + + // The sanctioned way out: explicit infrastructure creation. It + // publishes the EMPTY dataset, so it moves the head — the frozen + // batch's V0 base is now genuinely stale and the writer says so. + w.bootstrap().await.expect("bootstrap creates the store"); + let stale = persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .expect_err("the pre-bootstrap horizon is stale, and that is honest"); + assert!( + matches!( + stale, + lance_graph_planner::persist_sink::PersistError::Commit(CommitError::Fenced { .. }) + ), + "Fenced means nothing written — regenerate against the new head: {stale:?}" + ); + + // Regenerated against the post-bootstrap head, it commits — once. + let head = w.head(); + let out = persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), head), + vec![artifact(1, 0, 42, 0)], + ) + .await + .expect("the doubt is resolved and the horizon is current"); + assert!( + matches!(out, CommitOutcome::Committed { .. }), + "no frame existed, so this is a genuine first publication: {out:?}" + ); + let frames = w.timeline().await.unwrap(); + assert_eq!( + frames.len(), + 1, + "exactly one frame across the whole episode" + ); + } + + /// The paired silence: bootstrap on a store that ALREADY exists writes + /// nothing and publishes no cycle — it only resolves the doubt. + #[tokio::test] + async fn bootstrap_on_an_existing_store_publishes_nothing() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cycles.lance"); + let mut w = LanceCycleWriter::open(path.to_str().unwrap()) + .await + .unwrap(); + persist_cycle( + &mut w, + CycleFrame::new(CycleId(1), DatasetVersion(0)), + vec![artifact(1, 0, 42, 0)], + ) + .await + .unwrap(); + let head = w.head(); + let frames_before = w.timeline().await.unwrap().len(); + + w.bootstrap() + .await + .expect("idempotent on an existing store"); + assert_eq!(w.head(), head, "no version published"); + assert_eq!( + w.timeline().await.unwrap().len(), + frames_before, + "no frame added" + ); + } }