zeppelin-embed 0.8.0

//! ZE-52 slice D1: one writer lease that admits both a read view and a
//! writer overlay.
//!
//! The structured-write path admits a writer lease, builds an admitted base
//! from the caller's request list, stages it and commits. A query-driven
//! mutation needs the same lease to also carry a live `GraphReadView`, so a
//! MATCH clause can discover its targets before the overlay stages them.
//! These tests pin the second admission shape, `Store::with_native_mutation`,
//! and pin that the structured path it was extracted from did not move.

#![cfg(test)]
#![allow(
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic,
    clippy::unwrap_used
)]

use super::super::base::NativeAdmittedBase;
use super::super::mutate::{NativeMutationConsumer, NativeMutationError, NativeMutationReport};
use super::*;
use crate::lifecycle::durability::{CommitTier, DurabilityMode};
use crate::property_graph::query::runtime::{NativeExecutionError, RuntimeError, WorkKind};
use crate::property_graph::staging::{BatchEntityRef, GraphBatchReadView, StatementImages};
use crate::property_graph::storage::NativePreparationSource;
use crate::property_graph::{BatchDisposition, GraphDeleteMode};

const DOCUMENT_TEXT: &str = "mutation admission text";

fn fixture_options() -> OpenOptions {
    OpenOptions::new()
        .with_durability(DurabilityMode::Durable, CommitTier::Durable)
        .with_max_resident_bytes(256 * 1024 * 1024)
}

fn control() -> QueryControl {
    QueryControl::Cancel(CancelToken::new())
}

/// Three committed `Document` nodes, exactly slice B's shape.
struct MutationFixture {
    directory: super::tempfile::TempDir,
    path: std::path::PathBuf,
    store: Store,
    nodes: [NodeId; 3],
}

impl MutationFixture {
    fn commit() -> Self {
        let directory = super::tempfile::tempdir().expect("temporary parent");
        let path = directory.path().join("mutation-admission");
        let store = Store::create_native_graph(&path, fixture_options(), None)
            .expect("create native store");
        let first = commit_node(&store, "first");
        let second = commit_node(&store, "second");
        let third = commit_node(&store, "third");
        assert_ne!(first, second);
        assert_ne!(second, third);
        Self {
            directory,
            path,
            store,
            nodes: [first, second, third],
        }
    }

    /// Closes the fixture store and reopens the same directory, keeping the
    /// temporary directory alive alongside the new handle.
    fn reopen(self) -> (super::tempfile::TempDir, Store) {
        let Self {
            directory,
            path,
            store,
            nodes: _,
        } = self;
        store.close().expect("close native store");
        let reopened =
            Store::open_native_graph(&path, fixture_options(), None).expect("reopen native store");
        (directory, reopened)
    }
}

/// One committed node carrying the shared property fixture and its own text.
fn commit_node(store: &Store, key: &str) -> NodeId {
    let mut labels = [GraphName::new("Document").expect("label")];
    let mut properties = super::publication::property_fixture();
    let image = CanonicalContents::node(&mut labels, &mut properties, Some(DOCUMENT_TEXT), None)
        .expect("node image");
    let receipts = store
        .apply_native_graph(
            &[StructuredWrite {
                key: ApplicationKey::new(EntityKind::Node, "app", key).expect("node key"),
                revision: GraphRevision::new(1).expect("node revision"),
                operation: StructuredOperation::Create,
                image: Some(WriteImage::Node(&image)),
            }],
            &control(),
        )
        .unwrap_or_else(|error| panic!("commit {key}: {error:?}"));
    match receipts[0].entity {
        EntityId::Node(node) => node,
        EntityId::Relationship(_) => panic!("commit {key} returned a relationship"),
    }
}

/// The current published graph generation.
fn published_generation(store: &Store) -> GraphGeneration {
    store
        .admit_native_read()
        .expect("native read lease")
        .bundle()
        .base()
        .generation
}

/// Document and graph commits share the real Store clock and WAL.
#[test]
fn burns_three_store_generations_between_two_graph_commits() {
    use crate::ingest::{DocId, DocumentVersion, IngestBatch, IngestDocument, Revision};
    let directory = super::tempfile::tempdir().expect("temporary parent");
    let path = directory.path().join("unified");
    let graph = Store::create_native_graph(&path, fixture_options(), None).expect("graph store");
    let first = commit_node(&graph, "before-gap");
    assert_eq!(
        super::recovery::observe_node(&graph, first),
        Some((2, 1, 1))
    );
    let ingest = |id: u128| {
        graph
            .ingest(IngestBatch::new(vec![IngestDocument::new(
                DocumentVersion::new(DocId::new(id), Revision::new(1)),
                vec![1.0, 0.0],
            )]))
            .expect("burn store generation")
            .generation()
    };
    assert_eq!([ingest(101), ingest(102), ingest(103)], [3, 4, 5]);
    assert_eq!(published_generation(&graph).get(), 2);
    let second = commit_node(&graph, "after-gap");
    for node in [first, second] {
        assert_eq!(super::recovery::observe_node(&graph, node), Some((6, 2, 1)));
    }
    assert_eq!(graph.count_documents(None, None).unwrap().count, 3);
    graph.close().expect("close graph");
    let recovered = Store::open_native_graph(&path, fixture_options(), None).expect("recover gap");
    for node in [first, second] {
        assert_eq!(
            super::recovery::observe_node(&recovered, node),
            Some((6, 2, 1))
        );
    }
    assert_eq!(recovered.count_documents(None, None).unwrap().count, 3);
    recovered.close().expect("close recovered graph");
}

/// Every live node the admitted read view reports, in scan order.
struct ScanNodes;

impl NativeReadConsumer<Vec<NodeId>> for ScanNodes {
    fn consume<'s, 'lease, 'm, 'g>(
        &mut self,
        view: &GraphReadView<'s, 'lease, 'm, 'g>,
        runtime: &mut RuntimeContext<'lease, 'm, 'g>,
    ) -> Result<Vec<NodeId>, TreeError> {
        scan_all(view, runtime)
    }
}

/// Drains one label-free node cursor to exhaustion.
fn scan_all<'s, 'lease, 'm, 'g>(
    view: &GraphReadView<'s, 'lease, 'm, 'g>,
    runtime: &mut RuntimeContext<'lease, 'm, 'g>,
) -> Result<Vec<NodeId>, TreeError> {
    let placeholder = NodeId::new(1).map_err(|_| TreeError::Invalid("placeholder node"))?;
    let mut cursor = view.node_cursor(LabelSelection::All, runtime)?;
    let mut observed = Vec::new();
    loop {
        let mut batch = [placeholder; 4];
        let (count, state) = view.scan_nodes(&mut cursor, &mut batch, runtime)?;
        observed.extend_from_slice(batch.get(..count).unwrap_or(&[]));
        if state == CursorState::Done {
            break;
        }
    }
    observed.sort_by_key(|node| node.get());
    Ok(observed)
}

/// One statement: scan every node through the read view, then optionally
/// stage one deletion through the writer overlay. Nothing here builds an
/// image; D1 admits the two halves, and D2 onward supplies `Mutate`.
///
/// `invocations` is borrowed from the caller because the admission consumes
/// the consumer, and a retried or rejected attempt must still report how many
/// times it ran.
struct ScanThenStage<'a> {
    /// The node to tombstone, or `None` to stage nothing at all.
    target: Option<NodeId>,
    /// Cancelled after the scan, before anything is staged.
    cancel: Option<&'a CancelToken>,
    /// How many times the admission actually ran this consumer.
    invocations: &'a std::cell::Cell<u64>,
}

impl<'a> ScanThenStage<'a> {
    const fn reading(invocations: &'a std::cell::Cell<u64>) -> Self {
        Self {
            target: None,
            cancel: None,
            invocations,
        }
    }

    const fn deleting(invocations: &'a std::cell::Cell<u64>, target: NodeId) -> Self {
        Self {
            target: Some(target),
            cancel: None,
            invocations,
        }
    }

    const fn cancelling(invocations: &'a std::cell::Cell<u64>, token: &'a CancelToken) -> Self {
        Self {
            target: None,
            cancel: Some(token),
            invocations,
        }
    }
}

impl NativeMutationConsumer<Vec<NodeId>> for ScanThenStage<'_> {
    fn consume<'lease, 'm, 'g, 'w, 'i>(
        &mut self,
        view: &'w GraphReadView<'w, 'lease, 'm, 'g>,
        runtime: &mut RuntimeContext<'lease, 'm, 'g>,
        mut overlay: GraphBatchReadView<'w, 'static>,
        _images: &'w StatementImages<'i>,
        control: &mut WriteControl<'_>,
    ) -> Result<(Vec<NodeId>, GraphBatchReadView<'w, 'static>), NativeExecutionError> {
        self.invocations.set(self.invocations.get() + 1);
        // The read view has to be live under the writer lease: this is a real
        // cursor scan against the admitted generation, not a replay of a
        // caller-supplied request list.
        let observed = scan_all(view, runtime)?;
        if let Some(token) = self.cancel {
            token.cancel();
            // A cancelled statement stops here, before it stages anything.
            runtime.checkpoint()?;
        }
        if let Some(target) = self.target {
            overlay.delete(
                BatchEntityRef::Node(NodeRef::Existing(target)),
                GraphDeleteMode::Restrict,
                control,
            )?;
        }
        Ok((observed, overlay))
    }
}

/// Runs one consumer through the admission under this module's defaults.
#[allow(
    clippy::result_large_err,
    reason = "test keeps typed failure path allocation-free"
)]
fn admit(
    store: &Store,
    control: &QueryControl,
    consumer: ScanThenStage<'_>,
) -> Result<(Vec<NodeId>, NativeMutationReport), NativeMutationError> {
    store.with_native_mutation(
        control,
        RuntimeLimits::default(),
        4 * 1024 * 1024,
        16,
        8,
        8,
        8,
        consumer,
    )
}

/// Sorted live node ids as the public read path reports them.
fn live_nodes(store: &Store) -> Vec<NodeId> {
    store
        .with_native_read(
            &control(),
            RuntimeLimits::default(),
            4 * 1024 * 1024,
            16,
            ScanNodes,
        )
        .expect("scoped native read")
}

/// Builds a read-only lazy admitted base over `store` and asks it whether
/// `node` is still a live entity. This never goes through the writer, so it
/// answers from the published roots alone.
fn lazily_admitted_entity_exists(store: &Store, node: NodeId) -> bool {
    let lease = store.admit_native_read().expect("native read lease");
    let shared = GraphResources::from_store(store).expect("graph resources");
    let writer = WriteMemory::new(&shared, WriteLimits::default()).expect("write memory");
    let control = control();
    let storage = StorageMemory::new(&writer, &control, 32 * 1024 * 1024).expect("storage memory");
    let source = NativePreparationSource::new(&lease, &storage, 32).expect("preparation source");
    let mut resources = source.resources(64 * 1024 * 1024).expect("tree resources");
    let cell = std::cell::RefCell::new(&mut resources);
    let error = std::cell::Cell::new(None);
    let base =
        NativeAdmittedBase::with_lazy_targets(&lease, &source, &storage, &[], &cell, &error, 4)
            .expect("lazy admitted base");
    base.entity(EntityId::Node(node), &mut |_| Ok(()))
        .expect("lazy entity lookup")
        .is_some()
}

/// A query-driven statement scans the live graph under the writer lease,
/// stages one deletion through the overlay, and reaches durability through
/// the same commit tail the structured path uses.
#[test]
fn ze52_slice_d1_query_admission_deletes_a_scanned_node_and_reopens() {
    let fixture = MutationFixture::commit();
    let [first, second, third] = fixture.nodes;
    let admitted_before = published_generation(&fixture.store);
    assert_eq!(admitted_before.get(), 4);

    let invocations = std::cell::Cell::new(0);
    let (scanned, report) = admit(
        &fixture.store,
        &control(),
        ScanThenStage::deleting(&invocations, first),
    )
    .expect("query-driven mutation");

    // The read view was live: the consumer saw every committed node, not an
    // empty base.
    assert_eq!(invocations.get(), 1);
    assert_eq!(scanned, sorted([first, second, third]));
    assert_eq!(report.disposition, BatchDisposition::Changed);
    assert_eq!(report.admitted, admitted_before);
    assert_eq!(
        report.changed.map(GraphGeneration::get),
        Some(admitted_before.get() + 1)
    );
    // The scan is real work under the writer lease, not a replayed list.
    assert!(report.counters.get(WorkKind::CopiedBytes) > 0);

    // The deletion is publicly visible before the reopen.
    assert_eq!(live_nodes(&fixture.store), sorted([second, third]));

    let (_directory, reopened) = fixture.reopen();
    assert_eq!(published_generation(&reopened).get(), 5);
    assert_eq!(live_nodes(&reopened), sorted([second, third]));
    assert!(!lazily_admitted_entity_exists(&reopened, first));
    assert!(lazily_admitted_entity_exists(&reopened, second));
    reopened.close().expect("close reopened store");
}

/// A statement that stages nothing is a successful NoOp: no generation, no
/// WAL envelope, and every capacity the admission charged is released.
#[test]
fn ze52_slice_d1_query_admission_without_changes_is_noop() {
    let fixture = MutationFixture::commit();
    let [first, second, third] = fixture.nodes;
    let admitted_before = published_generation(&fixture.store);
    let shared = GraphResources::from_store(&fixture.store).expect("graph resources");
    let baseline_reserved = shared.reserved_bytes().expect("baseline reservation");
    let baseline_queries = fixture
        .store
        .active_queries
        .load(std::sync::atomic::Ordering::Relaxed);

    let invocations = std::cell::Cell::new(0);
    let (scanned, report) = admit(
        &fixture.store,
        &control(),
        ScanThenStage::reading(&invocations),
    )
    .expect("read-only query-driven mutation");

    assert_eq!(invocations.get(), 1);
    assert_eq!(scanned, sorted([first, second, third]));
    assert_eq!(report.disposition, BatchDisposition::NoOp);
    assert_eq!(report.admitted, admitted_before);
    assert_eq!(report.changed, None);

    assert_eq!(published_generation(&fixture.store), admitted_before);
    assert_eq!(live_nodes(&fixture.store), sorted([first, second, third]));

    // Every admission-scoped charge returned to its baseline.
    assert_eq!(
        shared
            .reserved_bytes()
            .expect("reservation after the admission"),
        baseline_reserved
    );
    assert_eq!(
        fixture
            .store
            .active_queries
            .load(std::sync::atomic::Ordering::Relaxed),
        baseline_queries
    );

    let (_directory, reopened) = fixture.reopen();
    assert_eq!(published_generation(&reopened), admitted_before);
    assert_eq!(live_nodes(&reopened), sorted([first, second, third]));
    reopened.close().expect("close reopened store");
}

/// Cancellation inside the consumer rejects the whole statement before the
/// commit tail runs: no generation, no value, and no retained charge.
#[test]
fn ze52_slice_d1_query_admission_rejects_before_commit_on_cancel() {
    let fixture = MutationFixture::commit();
    let [first, second, third] = fixture.nodes;
    let admitted_before = published_generation(&fixture.store);
    let shared = GraphResources::from_store(&fixture.store).expect("graph resources");
    let baseline_reserved = shared.reserved_bytes().expect("baseline reservation");
    let baseline_queries = fixture
        .store
        .active_queries
        .load(std::sync::atomic::Ordering::Relaxed);

    let token = CancelToken::new();
    let control = QueryControl::Cancel(token.clone());
    let invocations = std::cell::Cell::new(0);
    let rejected = admit(
        &fixture.store,
        &control,
        ScanThenStage::cancelling(&invocations, &token),
    );

    assert_eq!(invocations.get(), 1);
    match rejected {
        Err(NativeMutationError::Execution(NativeExecutionError::Runtime(
            RuntimeError::Value(crate::property_graph::query::QueryError::Cancelled),
        ))) => {}
        Ok((value, report)) => panic!(
            "cancelled mutation committed {value:?} at {:?}",
            report.changed
        ),
        Err(other) => panic!("cancelled mutation: {other:?}"),
    }

    // Nothing was published and nothing stayed charged.
    assert_eq!(published_generation(&fixture.store), admitted_before);
    assert_eq!(
        shared
            .reserved_bytes()
            .expect("reservation after the rejection"),
        baseline_reserved
    );
    assert_eq!(
        fixture
            .store
            .active_queries
            .load(std::sync::atomic::Ordering::Relaxed),
        baseline_queries
    );

    // A fresh control still works, so the rejection left the writer usable.
    assert_eq!(live_nodes(&fixture.store), sorted([first, second, third]));
    let (_directory, reopened) = fixture.reopen();
    assert_eq!(published_generation(&reopened), admitted_before);
    assert_eq!(live_nodes(&reopened), sorted([first, second, third]));
    reopened.close().expect("close reopened store");
}

/// The commit tail may checkpoint instead of committing. The admission then
/// rebuilds the whole attempt, so the consumer runs a second time and exactly
/// one deletion reaches durability.
#[test]
fn ze52_slice_d1_query_admission_reruns_consumer_after_checkpoint() {
    let fixture = MutationFixture::commit();
    fixture
        .store
        .set_native_graph_maintenance_policy(crate::property_graph::GraphMaintenancePolicy {
            automatic: false,
            ..crate::property_graph::GraphMaintenancePolicy::default()
        })
        .unwrap();
    let [first, second, third] = fixture.nodes;
    // Three creates already happened; drive the writer to exactly the 64
    // complete envelopes that make the next commit checkpoint first.
    for revision in 2..=62_u64 {
        let mut labels = [GraphName::new("Document").expect("label")];
        let mut properties = super::publication::property_fixture();
        let text = format!("revision {revision}");
        let image = CanonicalContents::node(&mut labels, &mut properties, Some(&text), None)
            .expect("checkpoint filler image");
        let receipt = fixture
            .store
            .apply_native_graph(
                &[StructuredWrite {
                    key: ApplicationKey::new(EntityKind::Node, "app", "third").expect("filler key"),
                    revision: GraphRevision::new(revision).expect("filler revision"),
                    operation: StructuredOperation::Put(EntityId::Node(third)),
                    image: Some(WriteImage::Node(&image)),
                }],
                &control(),
            )
            .expect("filler commit");
        assert!(!receipt[0].replayed);
    }
    let admitted_before = published_generation(&fixture.store);
    assert_eq!(admitted_before.get(), 65);

    let invocations = std::cell::Cell::new(0);
    let (scanned, report) = admit(
        &fixture.store,
        &control(),
        ScanThenStage::deleting(&invocations, first),
    )
    .expect("query-driven mutation across a checkpoint");

    // The first attempt checkpointed instead of committing, so the whole
    // statement, including its scan, ran again.
    assert_eq!(invocations.get(), 2);
    assert_eq!(scanned, sorted([first, second, third]));
    assert_eq!(report.disposition, BatchDisposition::Changed);
    assert_eq!(report.admitted, admitted_before);
    assert_eq!(report.changed.map(GraphGeneration::get), Some(67));

    // Exactly one deletion committed.
    assert_eq!(live_nodes(&fixture.store), sorted([second, third]));
    let (_directory, reopened) = fixture.reopen();
    assert_eq!(published_generation(&reopened).get(), 67);
    assert_eq!(live_nodes(&reopened), sorted([second, third]));
    assert!(!lazily_admitted_entity_exists(&reopened, first));
    reopened.close().expect("close reopened store");
}

/// PIN TEST. The structured-write path is the code slice D1 splits its commit
/// tail out of. This test never touches the new entry point: it drives
/// `apply_native_graph` through create, put, exact retry, delete and a
/// no-change statement, and pins the exact receipt shape and generation
/// sequence each one returns, plus the reopened public state.
///
/// Entity identities come from OS entropy, so the pin is over revisions,
/// generations, replay classification and the identities the create receipts
/// themselves reported, never over absolute id values.
#[test]
fn ze52_slice_d1_structured_path_is_unchanged() {
    let fixture = MutationFixture::commit();
    let [first, second, third] = fixture.nodes;
    let store = &fixture.store;

    // Three creates, one generation each.
    assert_eq!(published_generation(store).get(), 4);
    assert_eq!(live_nodes(store), sorted([first, second, third]));

    let key = |name: &'static str| {
        ApplicationKey::new(EntityKind::Node, "app", name).expect("structured key")
    };
    let mut labels = [GraphName::new("Document").expect("label")];
    let mut properties = super::publication::property_fixture();
    let replacement =
        CanonicalContents::node(&mut labels, &mut properties, Some("replaced text"), None)
            .expect("replacement image");

    // A changed PUT advances one generation and is not a replay.
    let put = [StructuredWrite {
        key: key("first"),
        revision: GraphRevision::new(2).expect("revision"),
        operation: StructuredOperation::Put(EntityId::Node(first)),
        image: Some(WriteImage::Node(&replacement)),
    }];
    let changed = store.apply_native_graph(&put, &control()).expect("put");
    assert_eq!(changed.len(), 1);
    assert_eq!(changed[0].entity, EntityId::Node(first));
    assert_eq!(changed[0].revision.get(), 2);
    assert_eq!(changed[0].generation.get(), 5);
    assert!(!changed[0].replayed);

    // The identical PUT is an exact retry: same generation, replayed, and no
    // new published generation.
    let retried = store
        .apply_native_graph(&put, &control())
        .expect("exact retry");
    assert_eq!(retried.len(), 1);
    assert_eq!(retried[0].entity, EntityId::Node(first));
    assert_eq!(retried[0].revision.get(), 2);
    assert_eq!(retried[0].generation.get(), 5);
    assert!(retried[0].replayed);
    assert_eq!(published_generation(store).get(), 5);

    // A DELETE advances one generation and removes the node publicly.
    let delete = [StructuredWrite {
        key: key("second"),
        revision: GraphRevision::new(2).expect("revision"),
        operation: StructuredOperation::Delete(EntityId::Node(second), GraphDeleteMode::Restrict),
        image: None,
    }];
    let removed = store
        .apply_native_graph(&delete, &control())
        .expect("delete");
    assert_eq!(removed.len(), 1);
    assert_eq!(removed[0].entity, EntityId::Node(second));
    assert_eq!(removed[0].revision.get(), 2);
    assert_eq!(removed[0].generation.get(), 6);
    assert!(!removed[0].replayed);

    // An empty statement is a no-op with no receipts and no generation.
    assert!(
        store
            .apply_native_graph(&[], &control())
            .expect("empty statement")
            .is_empty()
    );
    assert_eq!(published_generation(store).get(), 6);
    assert_eq!(live_nodes(store), sorted([first, third]));

    // The same state survives a reopen.
    let (_directory, reopened) = fixture.reopen();
    assert_eq!(published_generation(&reopened).get(), 6);
    assert_eq!(live_nodes(&reopened), sorted([first, third]));
    reopened.close().expect("close reopened store");
}

/// One committed node whose stored text is large enough that reading its
/// canonical image back dominates the work one statement charges.
const WIDE_TEXT_BYTES: usize = 64 * 1024;

/// Five measured budgets inside the warmed canonical comparison after ZE-436
/// artifact-proof reuse: reads span 466,101..597,986 work units. The successful
/// replacement below warms the proofs before these injections.
const CANONICAL_READ_WORK_BUDGETS: [u64; 5] = [480_000, 505_000, 530_000, 555_000, 580_000];

/// A budget the same statement completes under, proving the window above is a
/// storage failure and not an unconditional rejection.
const SUFFICIENT_WORK_BUDGET: u64 = 4 * 1024 * 1024;

/// A node image owned for the rest of the process. The admission hands its
/// consumer an overlay whose lifetime is chosen by the caller, so a replacement
/// image has to outlive every possible choice.
fn retained_wide_image() -> WriteImage<'static, 'static> {
    let text: &'static str = Box::leak("wide".repeat(WIDE_TEXT_BYTES / 4).into_boxed_str());
    let labels: &'static mut [GraphName<'static>] =
        Box::leak(Box::new([GraphName::new("Document").expect("wide label")]));
    let properties: &'static mut [GraphProperty<'static>] =
        Box::leak(super::publication::property_fixture().into_boxed_slice());
    let contents: &'static CanonicalContents<'static> = Box::leak(Box::new(
        CanonicalContents::node(labels, properties, Some(text), None).expect("wide node image"),
    ));
    WriteImage::Node(contents)
}

/// Stages one full-image replacement of an already committed node. The final
/// image equals the committed one, so finalization compares the two canonical
/// streams byte for byte and therefore reads the admitted base's stored image
/// back through `CachedCanonical::read_at`.
struct ReplaceWithSameImage {
    target: NodeId,
    image: WriteImage<'static, 'static>,
}

impl NativeMutationConsumer<()> for ReplaceWithSameImage {
    fn consume<'lease, 'm, 'g, 'w, 'i>(
        &mut self,
        _view: &'w GraphReadView<'w, 'lease, 'm, 'g>,
        _runtime: &mut RuntimeContext<'lease, 'm, 'g>,
        mut overlay: GraphBatchReadView<'w, 'static>,
        _images: &'w StatementImages<'i>,
        control: &mut WriteControl<'_>,
    ) -> Result<((), GraphBatchReadView<'w, 'static>), NativeExecutionError> {
        overlay.replace(
            BatchEntityRef::Node(NodeRef::Existing(self.target)),
            self.image,
            control,
        )?;
        Ok(((), overlay))
    }
}

/// A canonical read that fails inside the admitted base surfaces as the typed
/// lifecycle storage error, not as the opaque canonical I/O rejection it
/// caused.
///
/// `CachedCanonical::read_at` is the only writer of the base's error stash: it
/// records the real `TreeError` and hands its caller `io::ErrorKind::Other`,
/// which staging classifies as a canonical-comparison failure. The admission
/// must therefore read the stash before applying `?` to the call it guards.
/// Checking afterwards is unreachable for exactly the case the stash exists
/// for, and reports `Stage(Lifecycle(Canonical(Io(Other))))` with the root
/// cause discarded.
#[test]
fn ze52_slice_d1_canonical_read_failure_surfaces_as_the_typed_storage_error() {
    let directory = super::tempfile::tempdir().expect("temporary parent");
    let path = directory.path().join("canonical-read-failure");
    let store =
        Store::create_native_graph(&path, fixture_options(), None).expect("create native store");
    let image = retained_wide_image();
    let WriteImage::Node(contents) = image else {
        panic!("wide image is a node image");
    };
    let receipts = store
        .apply_native_graph(
            &[StructuredWrite {
                key: ApplicationKey::new(EntityKind::Node, "app", "wide").expect("wide key"),
                revision: GraphRevision::new(1).expect("wide revision"),
                operation: StructuredOperation::Create,
                image: Some(WriteImage::Node(contents)),
            }],
            &control(),
        )
        .expect("commit the wide node");
    let target = match receipts[0].entity {
        EntityId::Node(node) => node,
        EntityId::Relationship(_) => panic!("wide commit returned a relationship"),
    };
    let committed = published_generation(&store);

    let replace = |work_limit: u64| {
        let _schedule =
            crate::lifecycle::native_graph::preparation_limits::install(None, Some(work_limit));
        store.with_native_mutation(
            &control(),
            RuntimeLimits::default(),
            4 * 1024 * 1024,
            16,
            8,
            8,
            8,
            ReplaceWithSameImage { target, image },
        )
    };

    // The same statement is a successful NoOp when the base can be read, so
    // the window below fails on storage and on nothing else.
    let (value, report) = replace(SUFFICIENT_WORK_BUDGET).expect("unbudgeted replacement");
    assert_eq!(value, ());
    assert_eq!(report.disposition, BatchDisposition::NoOp);
    assert_eq!(report.changed, None);

    for work_limit in CANONICAL_READ_WORK_BUDGETS {
        let rejected = replace(work_limit);
        match rejected {
            Err(NativeMutationError::Graph(NativeGraphError::Stage(
                StageError::NativeStorage(TreeError::Work),
            ))) => {}
            Ok((_, report)) => panic!(
                "a canonical read at {work_limit} units committed {:?}",
                report.changed
            ),
            Err(other) => panic!("a canonical read at {work_limit} units: {other:?}"),
        }
        // Every rejection leaves the writer usable and publishes nothing.
        assert_eq!(published_generation(&store), committed);
    }

    assert_eq!(live_nodes(&store), vec![target]);
    store.close().expect("close native store");
}

fn sorted<const N: usize>(mut nodes: [NodeId; N]) -> Vec<NodeId> {
    nodes.sort_by_key(|node| node.get());
    nodes.to_vec()
}

// A real failed write leaves a complete object at its assigned generation.
fn failed_generation_five_preparation() -> (super::tempfile::TempDir, PathBuf, NodeId, PathBuf) {
    use super::publication::{FaultPoint, RecordingVfs};
    let directory = super::tempfile::tempdir().expect("temporary parent");
    let path = directory.path().join("gapped-orphan");
    let vfs = Arc::new(RecordingVfs::default());
    let infrastructure: Arc<dyn crate::vfs::Vfs> = vfs.clone();
    let graph = Store::create_native_graph_with_infrastructure(
        &path,
        fixture_options(),
        None,
        infrastructure,
        Arc::new(crate::lifecycle::SystemMonotonicClock),
        &mut crate::property_graph::storage::allocation::OsEntropy,
    )
    .expect("graph store");
    let first = commit_node(&graph, "before-gap");
    vfs.clear_events();
    graph
        .native_graph
        .assigned_generation
        .store(5, std::sync::atomic::Ordering::Release);
    vfs.arm_fault(FaultPoint::ObjectSync);
    let mut labels = [];
    let mut properties = [];
    let image = CanonicalContents::node(&mut labels, &mut properties, Some("after-gap"), None)
        .expect("node image");
    let failed = graph.apply_native_graph(
        &[StructuredWrite {
            key: ApplicationKey::new(EntityKind::Node, "app", "after-gap").expect("key"),
            revision: GraphRevision::new(1).expect("revision"),
            operation: StructuredOperation::Create,
            image: Some(WriteImage::Node(&image)),
        }],
        &control(),
    );
    assert!(failed.is_err());
    vfs.assert_fired_once();
    let orphan = vfs
        .take()
        .into_iter()
        .find_map(|event| match event {
            super::publication::DurabilityEvent::Create(path)
                if path
                    .extension()
                    .is_some_and(|extension| extension == "zgraph") =>
            {
                Some(path)
            }
            _ => None,
        })
        .expect("new native object from failed preparation");
    let bytes = std::fs::read(&orphan).expect("failed preparation object");
    assert_eq!(
        u64::from_le_bytes(bytes[64..72].try_into().expect("generation field")),
        5
    );
    assert_eq!(published_generation(&graph).get(), 2);
    graph.close().expect("close rejected graph write");
    (directory, path, first, orphan)
}

#[test]
fn assigned_generation_precommit_orphan_reopens() {
    let (_directory, path, first, orphan) = failed_generation_five_preparation();
    let before = super::recovery::file_snapshot(&path);
    let reader = Store::open_native_graph(
        &path,
        OpenOptions::read_only().with_max_resident_bytes(256 * 1024 * 1024),
        None,
    )
    .expect("read-only recovery ignores unreferenced preparation");
    assert_eq!(
        super::recovery::observe_node(&reader, first),
        Some((2, 1, 1))
    );
    reader.close().expect("close read-only graph");
    assert_eq!(super::recovery::file_snapshot(&path), before);
    assert!(orphan.exists());
    let reopened = Store::open_native_graph(&path, fixture_options(), None)
        .expect("valid gapped preparation orphan must not prevent recovery");
    assert_eq!(
        super::recovery::observe_node(&reopened, first),
        Some((2, 1, 1))
    );
    reopened.close().expect("close recovered graph");
}

#[test]
fn future_generation_orphan_with_bad_header_or_checksum_is_refused() {
    let (_directory, path, first, orphan) = failed_generation_five_preparation();
    let original = std::fs::read(&orphan).expect("complete preparation object");
    for corruption in ["magic", "family", "store", "checksum"] {
        let mut bytes = original.clone();
        match corruption {
            "magic" => bytes[0] ^= 1,
            "family" => bytes[8..10].copy_from_slice(&99_u16.to_le_bytes()),
            "store" => bytes[32..48].fill(0),
            "checksum" => *bytes.last_mut().expect("checksum trailer") ^= 1,
            _ => panic!("unknown corruption"),
        }
        std::fs::write(&orphan, &bytes).expect("hostile recognized native object");
        let error = match Store::open_native_graph(&path, fixture_options(), None) {
            Err(error) => error,
            Ok(store) => {
                store.close().expect("close incorrectly accepted graph");
                panic!("accepted {corruption} corruption");
            }
        };
        let expected = if corruption == "checksum" {
            "corrupt recognized native artifact"
        } else {
            "corrupt recognized native artifact header"
        };
        assert!(
            matches!(&error, NativeGraphError::Store(crate::lifecycle::StoreError::Manifest(crate::manifest::ManifestError::Decode(message))) if message.contains(expected)),
            "wrong refusal for {corruption}: {error:?}"
        );
        assert_eq!(
            std::fs::read(&orphan).expect("rejected object remains"),
            bytes
        );
        std::fs::write(&orphan, &original).expect("restore valid orphan");
    }
    let reopened =
        Store::open_native_graph(&path, fixture_options(), None).expect("valid control reopens");
    assert_eq!(
        super::recovery::observe_node(&reopened, first),
        Some((2, 1, 1))
    );
    reopened.close().expect("close valid control");
}