kglite 0.16.1

//! D2 Phase 2 — rollback while a reader holds the base

use super::unique_claims::{seeded_with_unique_name, unique_fingerprint};
use super::*;

/// A failed statement run while a reader is holding the graph must leave
/// **both** graphs exactly as they were.
///
/// This is D2 risk R3, and it is the one unforgivable failure mode in the whole
/// programme: every `UndoEntry` is keyed on a `NodeIndex`/`EdgeIndex` and is
/// reversed through `GraphWrite`. On a `Forked` backend that reversal must land
/// in the *overlay*. If any of it reached the shared base instead, the reader's
/// snapshot would silently acquire a rolled-back write — no error, no crash,
/// and no other test in this file would see it, because every other test owns
/// its graph outright.
///
/// The reader side is the golden snapshot: `fingerprint` is deliberately
/// over-specified (petgraph slot identity, inverted-index bucket *order*, schema
/// metadata, master column rows), so *any* base mutation shows up here, not just
/// one that changes a value a query would return.
///
/// The **unique** arm covers what `fingerprint` structurally cannot: unique
/// claims live in `unique_indices`, which the fingerprint does not read at all
/// (it is parked by `swap_data_scale` and restored by a per-type rebuild, not
/// by the journal). Its rollback path is therefore a *different* mechanism, and
/// until 2026-08-15 it had never been run on a forked backend — where the
/// journal's inverse ops replay through the overlay while the rebuild reads the
/// types those ops reported. A phantom or lost claim here rejects or admits
/// writes forever after, and no other assertion in this file can see it.
#[test]
fn a_rollback_while_a_reader_is_held_touches_neither_graph() {
    use crate::graph::handle::make_dir_graph_mut;
    use std::sync::Arc;

    for (name, build) in [
        ("plain", seeded as fn() -> DirGraph),
        ("columnar", seeded_columnar as fn() -> DirGraph),
        ("indexed", seeded_indexed as fn() -> DirGraph),
        ("unique", seeded_with_unique_name as fn() -> DirGraph),
    ] {
        let mut writer = Arc::new(build());
        let reader = Arc::clone(&writer);

        let claims_before = unique_fingerprint(&reader);
        assert_eq!(
            !claims_before.is_empty(),
            name == "unique",
            "{name}: only the unique arm may hold declared constraints, and it must — \
             otherwise the claim assertions below are vacuous"
        );

        // Fingerprinting needs `&mut`, and the reader is shared — so read
        // through a clone of it. The clone is a copy-on-write overlay over the
        // same base, so its content *is* the reader's content.
        let reader_before = fingerprint(&mut (*reader).clone());

        // One `make_dir_graph_mut` for both the fingerprint and the statement:
        // it is what bumps `version`, and `version` is part of the fingerprint
        // *on purpose* (a rolled-back statement must restore it), so calling it
        // twice would fail on a field the statement never touched.
        let writer_before = {
            let graph = make_dir_graph_mut(&mut writer);
            assert!(
                graph.graph.is_forked(),
                "{name}: precondition — a held reader must produce an overlay"
            );
            let before = fingerprint(&mut graph.clone());
            // Fails after its first write: row 1 commits, row 2 violates the
            // write scope.
            expect_failure(
                graph,
                "CREATE (:Item {id: 4000, name: 'first'}), (:Blocked {id: 4001, name: 'second'})",
                Some(&["Item"]),
            );
            before
        };

        assert_eq!(
            fingerprint(&mut (*reader).clone()),
            reader_before,
            "{name}: the reader's graph must be untouched by a write it never \
             asked for — a difference here means the undo journal reversed into \
             the shared base instead of the overlay (D2 R3)"
        );
        assert_eq!(
            fingerprint(&mut (*writer).clone()),
            writer_before,
            "{name}: the writer's failed statement must roll back exactly, \
             overlay or not"
        );
        assert_eq!(
            unique_fingerprint(&writer),
            claims_before,
            "{name}: the failed statement claimed 'first' and must have released it — \
             a claim surviving a rollback rejects that value forever"
        );
        assert_eq!(
            unique_fingerprint(&reader),
            claims_before,
            "{name}: the reader's occupancy map must be untouched by the writer's \
             constraint bookkeeping"
        );
    }
}

/// The **clone fallback**: a graph whose free lists are non-empty cannot be
/// forked, so a write taken while a reader holds it must deep-copy — and be
/// correct.
///
/// `forked::can_fork` is the predicate, and its unit tests cover the predicate
/// alone. Nothing covered what the `DirGraph` above it then *does*: every other
/// held-reader test in this file seeds a graph that has never deleted anything,
/// so they all take the overlay path and the fallback branch of
/// `GraphBackend::clone` was reached by no test at all. It is the branch that
/// has to stay right when the fast one cannot run — a deep copy that shared the
/// base instead (the `g.clone()`-instead-of-`deep_clone()` slip the code's own
/// comment warns about, one character wide) would let every later write mutate
/// the reader's snapshot in place.
///
/// Both directions are pinned: the fallback must fire *and* cost exactly one
/// whole-graph copy, so neither "it forked after a delete" (slot identity
/// broken) nor "it copied per statement" (the cliff D2 removed) can pass.
#[test]
fn a_write_under_a_held_reader_after_a_delete_takes_the_clone_path() {
    use crate::graph::handle::make_dir_graph_mut;
    use crate::graph::storage::backend::{backend_clone_nodes, reset_backend_clone_count};
    use std::sync::Arc;

    let mut writer = Arc::new(seeded());
    // A delete is what puts a slot on petgraph's free list; from here the
    // overlay cannot reproduce append indices, so `can_fork` refuses.
    run(
        Arc::make_mut(&mut writer),
        "MATCH (n:Item {id: 3}) DETACH DELETE n",
    );
    let live_nodes = writer.graph.node_count();

    let reader = Arc::clone(&writer);
    let reader_before = fingerprint(&mut (*reader).clone());

    reset_backend_clone_count();
    let graph = make_dir_graph_mut(&mut writer);
    assert!(
        !graph.graph.is_forked(),
        "a graph with a non-empty free list must NOT fork — the overlay hands out \
         append indices petgraph would reuse from the free list, and the fold-back \
         would then mis-key every DirGraph index recorded against them"
    );
    assert_eq!(
        backend_clone_nodes(),
        live_nodes,
        "the fallback is a genuine deep copy of the base, not an Arc share"
    );

    // Correct, not merely separate: the write lands, the delete stays deleted,
    // and a re-created id is found by the id index rather than the tombstone.
    run(graph, "CREATE (:Item {id: 4, name: 'd', qty: 40})");
    run(graph, "CREATE (:Item {id: 3, name: 'c-again', qty: 33})");
    assert_eq!(item_prop(graph, 4, "qty"), Some(Value::Int64(40)));
    assert_eq!(item_prop(graph, 3, "qty"), Some(Value::Int64(33)));
    assert_eq!(
        graph.graph.node_count(),
        live_nodes + 2,
        "both creates must be live nodes"
    );

    reset_backend_clone_count();
    run(graph, "CREATE (:Item {id: 5, name: 'e', qty: 50})");
    assert_eq!(
        backend_clone_nodes(),
        0,
        "the copy is paid once at the fork point — a per-statement copy here is the \
         cliff the overlay exists to remove, wearing the fallback's hat"
    );

    assert_eq!(
        fingerprint(&mut (*reader).clone()),
        reader_before,
        "the reader's snapshot must be untouched by every one of those writes"
    );
    let mut reader_now = (*reader).clone();
    assert_eq!(
        reader_now.lookup_by_id("Item", &Value::Int64(4)),
        None,
        "none of the writer's three creates may be visible through the reader — the \
         observable half of the deep copy, in the direction a shared backend breaks"
    );
    assert_eq!(
        reader_now.lookup_by_id("Item", &Value::Int64(3)),
        None,
        "and the id the writer re-created must still read as deleted here"
    );
}

/// The forked backend must take the **journal** path, not the clone checkpoint
/// — and the one write it cannot express must cost exactly one copy, not one
/// per statement.
///
/// D2 risk R2: `journal_covers` has exactly one term left,
/// `supports_undo_journal()`. If `Forked` answered `false` there, every
/// statement taken while a view is held would open a
/// `StatementCheckpoint::Clone` — an O(V+E) copy *per statement* instead of the
/// one-off fork this phase removed, i.e. the fix introducing a cliff worse than
/// the defect. The zeros below are what would break.
///
/// The middle assertion is the honest half. An overlay cannot express an
/// adjacency edit (`storage/forked.rs` module doc), so the edge `CREATE`
/// flattens the overlay — one deep copy, the pre-D2 cost. What matters is that
/// it happens **once**: the backend is a plain `Memory` afterwards, so every
/// later statement is back to mutating in place. A per-statement copy here
/// would be the R2 accident wearing a different hat.
#[test]
fn forked_statements_copy_zero_nodes_except_one_flatten() {
    use crate::graph::handle::make_dir_graph_mut;
    use crate::graph::storage::backend::{backend_clone_nodes, reset_backend_clone_count};
    use std::sync::Arc;

    /// Overlay-expressible: node adds and weight writes.
    const OVERLAY_QUERIES: &[&str] = &[
        "CREATE (:Item {id: 2000, name: 'x'})",
        "MATCH (n:Item {id: 1}) SET n.qty = 11, n.name = 'renamed'",
        "MATCH (n:Item {id: 2000}) SET n:Featured",
        "MERGE (n:Item {id: 2001}) ON CREATE SET n.name = 'merged'",
    ];
    /// Rewrites existing nodes' petgraph adjacency, so it flattens first.
    const ADJACENCY_QUERY: &str =
        "MATCH (a:Item {id: 1}), (b:Item {id: 3}) CREATE (a)-[:LINKS {weight: 2}]->(b)";

    let mut writer = Arc::new(seeded());
    let reader = Arc::clone(&writer);
    let fixture_nodes = reader.graph.node_count();

    let graph = make_dir_graph_mut(&mut writer);
    assert!(graph.graph.is_forked(), "precondition: the write forked");

    for &query in OVERLAY_QUERIES {
        reset_backend_clone_count();
        run(graph, query);
        assert_eq!(
            backend_clone_nodes(),
            0,
            "an overlay-expressible statement on a forked backend must copy no node: {query}"
        );
        assert!(
            graph.graph.is_forked(),
            "...and must leave the backend forked: {query}"
        );
    }

    reset_backend_clone_count();
    run(graph, ADJACENCY_QUERY);
    assert_eq!(
        backend_clone_nodes(),
        fixture_nodes,
        "the adjacency write flattens the overlay — exactly one copy of the base"
    );
    assert!(
        !graph.graph.is_forked(),
        "flattening must leave a plain backend, so the copy is paid once"
    );

    reset_backend_clone_count();
    run(graph, "MATCH (n:Item {id: 2000}) DETACH DELETE n");
    run(graph, "CREATE (:Item {id: 2002, name: 'after'})");
    assert_eq!(
        backend_clone_nodes(),
        0,
        "after flattening, later statements mutate in place — one copy per fork, \
         not one per statement"
    );

    // The reader is still holding the pre-fork base and must be untouched by
    // any of it.
    assert_eq!(reader.graph.node_count(), fixture_nodes);
}

/// A replace-all write on a forked columnar row must roll back the cells it
/// **nulled**, not just the ones it wrote.
///
/// `GraphWrite::replace_node_properties` is the only write whose pre-image
/// spans keys the caller never mentioned: it clears the row first, so a journal
/// that captured the incoming keys alone would restore the batch's own columns
/// and leave every omitted property permanently null — a rolled-back statement
/// deleting data. Reached through the trait rather than through Cypher because
/// its one caller is `add_nodes` replace-mode (`mutation::batch`), which is not
/// a statement and opens no checkpoint of its own; the checkpoint here is what
/// a caller running it inside one would get.
///
/// The forked fixture is the point: the overlay's arm is a separate
/// implementation from the two heap writers, and until 2026-08-13 it was not
/// replace-all at all, so nothing had ever journalled a `ReplaceRow` on it.
#[test]
fn a_replace_write_on_a_forked_columnar_row_rolls_back_the_cells_it_nulled() {
    use crate::graph::dir_graph::rollback::StatementCheckpoint;
    use crate::graph::handle::make_dir_graph_mut;
    use crate::graph::storage::GraphWrite;
    use std::sync::Arc;

    let mut writer = Arc::new(seeded_columnar());
    let _reader = Arc::clone(&writer);
    let graph = make_dir_graph_mut(&mut writer);
    assert!(
        graph.graph.is_forked(),
        "precondition: a held view must fork the writer"
    );

    let idx = graph
        .lookup_by_id("Item", &Value::Int64(1))
        .expect("the fixture must carry Item id 1");
    // Interned before the fingerprint so the write, not the interning, is what
    // the comparison is about.
    let name = graph.interner.get_or_intern("name");
    let fresh = graph.interner.get_or_intern("fresh");

    let before = fingerprint(&mut graph.clone());

    let checkpoint = StatementCheckpoint::open(graph);
    GraphWrite::replace_node_properties(
        &mut graph.graph,
        idx,
        vec![
            (name, Value::String("replaced".into())),
            (fresh, Value::Int64(42)),
        ],
    );

    // Non-vacuity: the write has to have nulled `qty` and grown the schema,
    // or the rollback below is restoring nothing.
    let mid = fingerprint(&mut graph.clone());
    assert_ne!(before, mid, "the replace must have changed the graph");
    let row = graph
        .graph
        .node_weight(idx)
        .and_then(|node| match &node.properties {
            PropertyStorage::Columnar(row) => Some(row.row_id()),
            _ => None,
        })
        .expect("the fixture node must be a columnar row");
    let qty = graph.interner.get_or_intern("qty");
    let store = graph.column_store("Item").expect("master store");
    assert!(
        store.get(row, qty).is_none(),
        "precondition: the replace must have dropped the property it omitted, \
         or this test cannot see whether the rollback restores it"
    );

    checkpoint.rollback(graph);

    assert_eq!(
        fingerprint(&mut graph.clone()),
        before,
        "a rolled-back replace must restore the whole row — the cells it wrote \
         and the ones it nulled"
    );
}