issundb-core 0.1.0-alpha.26

use std::{
    any::{Any, TypeId as StdTypeId},
    collections::HashMap,
    path::Path,
    sync::Arc,
};

use parking_lot::ReentrantMutex;
use serde::Serialize;
use tracing::instrument;
use zerocopy::{FromBytes, IntoBytes};

use ahash::{AHashMap, AHashSet};

use crate::{
    csr::{CsrCache, CsrSnapshot},
    error::Error,
    schema::{
        AdjEntry, DirectedNeighborEntry, EdgeId, EdgeRecord, LabelId, Language, NeighborEntry,
        NodeId, NodeRecord, PropKeyId, PropValue, TypeId, WeightedPath,
    },
    storage::{
        Storage, fts,
        ids::{
            adjust_label_count, adjust_type_count, alloc_edge_id, alloc_node_id, get_label,
            get_or_create_label, get_or_create_prop_key, get_or_create_type, get_prop_key,
            get_prop_key_name, get_type,
        },
        props,
    },
};

pub mod algo;
pub mod edge;
pub mod fts_mod;
pub mod index;
pub mod kernels;
pub mod node;
pub mod stats;
pub mod txn;
pub mod vector;

/// The direction of edges to count for degree centrality.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum DegreeDirection {
    /// Count incoming edges only.
    In,
    /// Count outgoing edges only.
    Out,
    /// Count both incoming and outgoing edges.
    Both,
}

/// Which score [`Graph::link_prediction_score`] computes for a pair of nodes.
///
/// All five read the graph as undirected over distinct neighbors, the same
/// neighborhood [`Graph::clustering_coefficient`] uses, so a pair joined by several
/// edges is one neighbor and direction never matters. A higher score means the pair
/// is more likely to become connected.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum LinkPredictionMetric {
    /// How many neighbors the two nodes share.
    CommonNeighbors,
    /// Shared neighbors over the size of the combined neighborhood, so a pair of
    /// low-degree nodes is not penalized against a pair of hubs. Zero when neither
    /// node has a neighbor.
    Jaccard,
    /// Shared neighbors weighted by `1 / ln(degree)`, so a neighbor that everyone
    /// shares counts for little. A shared neighbor of degree one contributes nothing,
    /// since `ln(1)` is zero and the term is undefined rather than large.
    AdamicAdar,
    /// Shared neighbors weighted by `1 / degree`, which penalizes popular neighbors
    /// harder than Adamic-Adar does.
    ResourceAllocation,
    /// The product of the two degrees, on the theory that busy nodes attract more
    /// edges. This one ignores shared neighbors entirely, so it scores pairs that
    /// have nothing in common.
    PreferentialAttachment,
}

/// Describes the pattern [`Graph::count_triangle_cycles`] counts, the directed
/// cycle `(a)-[t1]->(b)-[t2]->(c)-[t3]->(a)` with an optional relationship
/// type per hop and an optional label per node variable. `None` means
/// unconstrained.
#[derive(Debug, Clone, Default)]
pub struct TriangleCountSpec<'a> {
    /// Relationship types for the hops `a -> b`, `b -> c`, and `c -> a`.
    pub rel_types: [Option<&'a str>; 3],
    /// Labels required on `a`, `b`, and `c`.
    pub labels: [Option<&'a str>; 3],
}

/// Describes the pattern [`Graph::count_linear_paths`] counts, an open directed
/// path of one or two hops, `(v0)-[t1]->(v1)` or
/// `(v0)-[t1]->(v1)-[t2]->(v2)`, with an optional relationship type per hop
/// and an optional label per node variable. `None` means unconstrained.
///
/// `rel_types.len()` is the hop count (1 or 2); `labels.len()` is the node
/// count (hop count plus one). The two-hop count follows Cypher MATCH
/// relationship-uniqueness semantics: the two relationships must be distinct,
/// which only constrains self-loop assignments where one edge could fill both
/// hops.
#[derive(Debug, Clone, Default)]
pub struct PathCountSpec<'a> {
    /// Relationship type per hop, in path order. Length 1 or 2.
    pub rel_types: Vec<Option<&'a str>>,
    /// Label per node variable, in path order. Length is `rel_types.len() + 1`.
    pub labels: Vec<Option<&'a str>>,
    /// Optional explicit allow-set of node ids per variable, in path order. A
    /// `Some(ids)` entry restricts that variable to `ids` (intersected with its
    /// label, if any); `None` leaves it unconstrained beyond the label. The
    /// caller resolves these sets by pushing per-vertex property predicates down
    /// into index lookups, so a filtered path count stays a kernel call instead
    /// of materializing rows. An empty vector (the default) means no variable is
    /// constrained, identical to the unfiltered path count.
    pub vertex_allow: Vec<Option<Vec<NodeId>>>,
}

/// Describes the pattern [`Graph::grouped_edge_counts`] counts, typed edges
/// grouped by one endpoint. With `group_is_dst`, edges are grouped by their
/// destination and the source is the counted endpoint (in-degree per
/// destination); otherwise edges are grouped by their source and the
/// destination is counted (out-degree per source). `group_label` and
/// `counted_label` optionally constrain each endpoint (`None` is
/// unconstrained). `counted_nonnull_prop` counts an edge only when the counted
/// endpoint's property is non-null (the semantics of `count(v.prop)` over the
/// expansion); `None` counts every qualifying edge (the semantics of
/// `count(*)` or `count(v)`, where a bound node variable is never null).
#[derive(Debug, Clone, Default)]
pub struct GroupedDegreeSpec<'a> {
    /// Relationship type to count, or `None` for any type.
    pub rel_type: Option<&'a str>,
    /// Group by the edge destination (count incoming) when true; by the edge
    /// source (count outgoing) when false.
    pub group_is_dst: bool,
    /// Label required on the group endpoint.
    pub group_label: Option<&'a str>,
    /// Label required on the counted endpoint.
    pub counted_label: Option<&'a str>,
    /// Explicit allow-set the counted endpoint must belong to, intersected with
    /// `counted_label`; `None` leaves it unconstrained beyond the label. The
    /// caller resolves this set by pushing a per-vertex property predicate down
    /// into index lookups, as [`PathCountSpec::vertex_allow`] does, so a filtered
    /// grouped count stays a kernel call. An empty slice counts zero.
    pub counted_allow: Option<&'a [NodeId]>,
    /// Property that must be non-null on the counted endpoint for an edge to
    /// count; `None` counts every qualifying edge.
    pub counted_nonnull_prop: Option<&'a str>,
}

/// Describes the pattern [`Graph::typed_neighbor_counts`] counts, the typed
/// neighbors of each source across one hop. `incoming` follows incoming edges instead
/// of outgoing ones. A neighbor qualifies when it carries every label in
/// `neighbor_labels` (an empty slice is unconstrained) and, when
/// `neighbor_allow` is present, is a member of that set; it adds to the counted
/// total only when `neighbor_nonnull_prop` is absent or non-null on it (the
/// semantics of `count(v.prop)` over the expansion, against `count(*)`).
#[derive(Debug, Clone, Default)]
pub struct NeighborCountSpec<'a> {
    /// Relationship type to follow, or `None` for any type.
    pub rel_type: Option<&'a str>,
    /// Follow incoming edges (neighbors are edge sources) instead of outgoing.
    pub incoming: bool,
    /// Labels a neighbor must all carry to qualify.
    pub neighbor_labels: &'a [&'a str],
    /// Explicit allow-set a neighbor must belong to, intersected with the labels
    /// above; `None` leaves the neighbor unconstrained beyond its labels. The
    /// caller resolves this set by evaluating per-neighbor property predicates
    /// itself, so a filtered count stays a kernel call instead of materializing
    /// one entry per traversed edge, exactly as
    /// [`PathCountSpec::vertex_allow`] does for the path count. An empty slice
    /// admits no neighbor and counts zero.
    pub neighbor_allow: Option<&'a [NodeId]>,
    /// Property that must be non-null on a qualifying neighbor for it to add to
    /// the counted total; `None` counts every qualifying neighbor.
    pub neighbor_nonnull_prop: Option<&'a str>,
}

/// Builds a 12-byte composite key `(prefix u32 BE, id u64 BE)` for secondary index lookups.
/// Decided `schema_has_edge` verdicts and the write generation they were decided
/// under. A `None` value is a remembered "undecided", which is worth keeping so the
/// probe budget is not respent to reach the same non-answer.
pub(super) type SchemaProbeMemo = (u64, AHashMap<(LabelId, TypeId, LabelId), Option<bool>>);

pub(super) fn composite_key(prefix: u32, id: u64) -> [u8; 12] {
    let mut key = [0u8; 12];
    key[..4].copy_from_slice(&prefix.to_be_bytes());
    key[4..].copy_from_slice(&id.to_be_bytes());
    key
}

/// Type tag for a null value in the sortable property encoding.
pub(super) const ENCODED_NULL: u8 = 0x00;

/// Sign bit mask used to make IEEE-754 `f64` bit patterns and two's-complement
/// `i64` values sort in ascending numeric order as big-endian bytes.
const SORT_SIGN_BIT: u64 = 0x8000_0000_0000_0000;

/// Maximum string length (in bytes) that can be auto-indexed. The property
/// index key is `(label_id, prop_key_id, encoded_val, node_id)`, so it carries
/// 16 bytes of fixed fields plus the 2-byte string-encoding frame (`0x04` tag
/// and `0x00` terminator) around the value. LMDB's default maximum key size is
/// 511 bytes; a string longer than this would overflow that limit and cannot be
/// indexed, so `encode_property_value` declines it and the value is left
/// unindexed (equality lookups fall back to a scan, and long text belongs in a
/// full-text index anyway). The bound is conservative to leave headroom.
pub(super) const MAX_INDEXED_STRING_LEN: usize = 480;

/// Encodes a JSON property value into a sortable byte representation for the index.
///
/// Numbers use a fixed 17-byte encoding: a `0x03` tag, then 8 bytes of the
/// order-preserving `f64` bit pattern (the primary numeric sort key), then 8
/// bytes of an integer disambiguator. The disambiguator makes the encoding
/// lossless for `i64` values: two integers that round to the same `f64` (any
/// pair beyond 2^53) still produce distinct keys, while an integer and a float
/// of the same real value (e.g. `30` and `30.0`) produce identical keys so they
/// continue to compare equal. Keeping every numeric encoding the same length is
/// required because property lookups match by key prefix; a variable-length
/// encoding where one value is a prefix of another would yield false matches.
pub(super) fn encode_property_value(val: &serde_json::Value) -> Option<Vec<u8>> {
    match val {
        serde_json::Value::Null => Some(vec![ENCODED_NULL]),
        serde_json::Value::Bool(false) => Some(vec![0x01]),
        serde_json::Value::Bool(true) => Some(vec![0x02]),
        serde_json::Value::Number(num) => {
            let float_val = num.as_f64()?;
            let bits = float_val.to_bits();
            let masked = if (bits & SORT_SIGN_BIT) != 0 {
                !bits
            } else {
                bits ^ SORT_SIGN_BIT
            };
            // Integer disambiguator: for any number whose exact real value is an
            // integer in `i64` range, store that integer in sign-flipped
            // big-endian order so distinct large integers never collide. All
            // other numbers (non-integers, out-of-range) get a fixed sentinel;
            // they already have a unique `f64` bit pattern in the primary key,
            // so the sentinel value cannot affect ordering or equality.
            let int_disambig: u64 = if let Some(i) = num.as_i64() {
                (i as u64) ^ SORT_SIGN_BIT
            } else if float_val.fract() == 0.0
                && float_val >= i64::MIN as f64
                && float_val <= i64::MAX as f64
            {
                ((float_val as i64) as u64) ^ SORT_SIGN_BIT
            } else {
                0
            };
            let mut buf = Vec::with_capacity(17);
            buf.push(0x03);
            buf.extend_from_slice(&masked.to_be_bytes());
            buf.extend_from_slice(&int_disambig.to_be_bytes());
            Some(buf)
        }
        serde_json::Value::String(s) => {
            // A string too long to fit an LMDB key cannot be indexed; decline it
            // so the property is left unindexed rather than crashing the write.
            if s.len() > MAX_INDEXED_STRING_LEN {
                return None;
            }
            let mut buf = Vec::with_capacity(1 + s.len() + 1);
            buf.push(0x04);
            buf.extend_from_slice(s.as_bytes());
            buf.push(0x00);
            Some(buf)
        }
        _ => None, // Skip arrays and objects
    }
}

/// Comparable-type family of an encoded property value's leading type tag.
/// Booleans span two tags (`0x01` false, `0x02` true) but form one comparable
/// family; every other tag is its own family. Range scans compare only values
/// within the bound's family, because under openCypher a value of one type
/// never satisfies a range bound of another (a string is not comparable to a
/// numeric bound), even though the tagged encoding orders them globally.
pub(super) fn encoded_tag_family(tag: u8) -> u8 {
    match tag {
        0x02 => 0x01,
        t => t,
    }
}

/// Decodes a sortable byte representation back into a JSON property value.
#[allow(dead_code)]
pub(super) fn decode_property_value(bytes: &[u8]) -> Option<serde_json::Value> {
    if bytes.is_empty() {
        return None;
    }
    match bytes[0] {
        0x00 => Some(serde_json::Value::Null),
        0x01 => Some(serde_json::Value::Bool(false)),
        0x02 => Some(serde_json::Value::Bool(true)),
        0x03 => {
            // Numbers are `tag + 8-byte f64 sort key + 8-byte int disambiguator`.
            if bytes.len() < 17 {
                return None;
            }
            // Prefer the lossless integer disambiguator when it round-trips,
            // so large integers decode exactly rather than through `f64`.
            let mut int_arr = [0u8; 8];
            int_arr.copy_from_slice(&bytes[9..17]);
            let int_val = (u64::from_be_bytes(int_arr) ^ SORT_SIGN_BIT) as i64;

            let mut arr = [0u8; 8];
            arr.copy_from_slice(&bytes[1..9]);
            let masked = u64::from_be_bytes(arr);
            let bits = if (masked & SORT_SIGN_BIT) == 0 {
                !masked
            } else {
                masked ^ SORT_SIGN_BIT
            };
            let float_val = f64::from_bits(bits);

            // If the disambiguator's integer equals the float key, the value was
            // an integer (or integer-valued float): return it losslessly as an
            // integer. Non-integers store a sentinel whose sign-flipped form is
            // `i64::MIN`, which never matches a non-integer float key.
            if (int_val as f64) == float_val {
                Some(serde_json::Value::Number(int_val.into()))
            } else {
                serde_json::Number::from_f64(float_val).map(serde_json::Value::Number)
            }
        }
        0x04 => {
            let str_bytes = if bytes.ends_with(&[0x00]) {
                &bytes[1..bytes.len() - 1]
            } else {
                &bytes[1..]
            };
            String::from_utf8(str_bytes.to_vec())
                .ok()
                .map(serde_json::Value::String)
        }
        _ => None,
    }
}

/// Builds a composite key `(label_id, prop_key_id, encoded_val, node_id)` for node property index.
pub(super) fn node_prop_index_key(
    label_id: LabelId,
    prop_key_id: PropKeyId,
    encoded_val: &[u8],
    node_id: NodeId,
) -> Vec<u8> {
    let mut key = Vec::with_capacity(4 + 4 + encoded_val.len() + 8);
    key.extend_from_slice(&label_id.to_be_bytes());
    key.extend_from_slice(&prop_key_id.to_be_bytes());
    key.extend_from_slice(encoded_val);
    key.extend_from_slice(&node_id.to_be_bytes());
    key
}

/// Builds a composite key `(type_id, prop_key_id, encoded_val, edge_id)` for edge property index.
pub(super) fn edge_prop_index_key(
    type_id: TypeId,
    prop_key_id: PropKeyId,
    encoded_val: &[u8],
    edge_id: EdgeId,
) -> Vec<u8> {
    let mut key = Vec::with_capacity(4 + 4 + encoded_val.len() + 8);
    key.extend_from_slice(&type_id.to_be_bytes());
    key.extend_from_slice(&prop_key_id.to_be_bytes());
    key.extend_from_slice(encoded_val);
    key.extend_from_slice(&edge_id.to_be_bytes());
    key
}

/// Returns the trailing 8-byte id from a property-index key, but only when the
/// key's encoded-value segment equals `encoded` exactly.
///
/// A property-index key is `(prefix u32, prop_key_id u32, encoded_val, id u64)`,
/// so the value segment is `key[8 .. len - 8]`. A prefix scan on
/// `(prefix, prop_key_id, encoded)` also matches keys whose value merely *starts*
/// with `encoded`: for the NUL-terminated string encoding, a stored `"a\0"`
/// (encoded `04 61 00 00`) is matched by a lookup for `"a"` (encoded `04 61 00`),
/// because a small id has leading zero bytes. Requiring the value segment to
/// equal `encoded` exactly rejects those collisions so equality lookups and
/// unique-constraint checks never conflate distinct string values. Fixed-width
/// encodings (numbers, bools, null) are already exact, so this never rejects a
/// genuine match. Returns `None` when the key is too short or the value differs.
pub(super) fn exact_prop_index_id(key: &[u8], encoded: &[u8]) -> Option<NodeId> {
    if key.len() < 8 + 8 {
        return None;
    }
    if &key[8..key.len() - 8] != encoded {
        return None;
    }
    let id_bytes: [u8; 8] = key[key.len() - 8..].try_into().ok()?;
    Some(u64::from_be_bytes(id_bytes))
}

/// Whether a stored string lies within the range `[lo, hi]` (or the open
/// variants), using byte-wise comparison, which is the same order the
/// order-preserving index encoding reproduces (`"a" < "a\0" < "ab"`). A `None`
/// bound is unbounded on that side. Backs the string-range label-scan fallback
/// for values too long to index.
pub(super) fn str_in_range(
    s: &str,
    lo: Option<&str>,
    lo_inclusive: bool,
    hi: Option<&str>,
    hi_inclusive: bool,
) -> bool {
    if let Some(lo) = lo {
        if lo_inclusive {
            if s < lo {
                return false;
            }
        } else if s <= lo {
            return false;
        }
    }
    if let Some(hi) = hi {
        if hi_inclusive {
            if s > hi {
                return false;
            }
        } else if s >= hi {
            return false;
        }
    }
    true
}

/// Builds a composite key `(label_id, prop_key_id, term)` for FTS postings.
pub(super) fn fts_postings_key(label_id: LabelId, prop_key_id: PropKeyId, term: &str) -> Vec<u8> {
    let mut key = Vec::with_capacity(8 + term.len());
    key.extend_from_slice(&label_id.to_be_bytes());
    key.extend_from_slice(&prop_key_id.to_be_bytes());
    key.extend_from_slice(term.as_bytes());
    key
}

/// Builds a 12-byte FTS posting value `(node_id, frequency)`.
pub(super) fn fts_posting_val(node_id: NodeId, frequency: u32) -> [u8; 12] {
    let mut val = [0u8; 12];
    val[0..8].copy_from_slice(&node_id.to_be_bytes());
    val[8..12].copy_from_slice(&frequency.to_be_bytes());
    val
}

/// Parses a 12-byte FTS posting value into `(node_id, frequency)`.
pub(super) fn parse_fts_posting_val(bytes: &[u8]) -> Result<(NodeId, u32), Error> {
    if bytes.len() != 12 {
        return Err(Error::Corrupt("fts posting value must be 12 bytes"));
    }
    let node_id = NodeId::from_be_bytes(
        bytes[0..8]
            .try_into()
            .map_err(|_| Error::Corrupt("fts posting: node_id slice wrong size"))?,
    );
    let frequency = u32::from_be_bytes(
        bytes[8..12]
            .try_into()
            .map_err(|_| Error::Corrupt("fts posting: frequency slice wrong size"))?,
    );
    Ok((node_id, frequency))
}

/// Builds a 16-byte FTS doc key `(label_id, prop_key_id, node_id)`.
pub(super) fn fts_doc_key(label_id: LabelId, prop_key_id: PropKeyId, node_id: NodeId) -> [u8; 16] {
    let mut key = [0u8; 16];
    key[0..4].copy_from_slice(&label_id.to_be_bytes());
    key[4..8].copy_from_slice(&prop_key_id.to_be_bytes());
    key[8..16].copy_from_slice(&node_id.to_be_bytes());
    key
}

/// Parses a 4-byte doc length value.
pub(super) fn parse_fts_doc_val(bytes: &[u8]) -> Result<u32, Error> {
    if bytes.len() != 4 {
        return Err(Error::Corrupt("fts doc val must be 4 bytes"));
    }
    Ok(u32::from_be_bytes(bytes.try_into().map_err(|_| {
        Error::Corrupt("fts doc val: slice wrong size")
    })?))
}

pub(super) fn fts_stats_n_key(label_id: LabelId, prop_key_id: PropKeyId) -> String {
    format!("fts_stats:node:l:{label_id}:p:{prop_key_id}:N")
}

pub(super) fn fts_stats_sum_dl_key(label_id: LabelId, prop_key_id: PropKeyId) -> String {
    format!("fts_stats:node:l:{label_id}:p:{prop_key_id}:sum_dl")
}

/// The graph database handle. It is cheap to clone, since all state is behind `Arc`.
#[derive(Clone)]
pub struct Graph {
    pub(super) storage: Arc<Storage>,
    pub(super) _write_lock: Arc<ReentrantMutex<()>>,
    pub(super) csr_cache: Arc<CsrCache>,
    pub(super) prop_columns: Arc<crate::columns::ColumnsCache<crate::columns::NodeSource>>,
    pub(super) edge_columns: Arc<crate::columns::ColumnsCache<crate::columns::EdgeSource>>,
    /// Per-`(label, type)` edge frequencies backing the optimizer's per-source-label
    /// expand-ratio estimate. Never built as a side effect of a query; see
    /// [`crate::graph::stats`] for which reader tolerates which staleness.
    pub(super) edge_fanout: Arc<parking_lot::Mutex<Option<crate::graph::stats::EdgeFanout>>>,
    /// Decided `schema_has_edge` verdicts for one write generation, keyed by
    /// `(src_label, type, dst_label)`. The type-inference pass asks the same questions
    /// on every execution because there is no plan cache, and answering without the
    /// statistics table means walking the graph, so a decided verdict is remembered
    /// until a write invalidates the generation. See [`crate::graph::stats`].
    pub(super) schema_probes: Arc<parking_lot::Mutex<SchemaProbeMemo>>,
    /// Cached id-indexed group codes, one shared array per grouped property,
    /// valid for exactly one write generation, which is what lets a grouped
    /// bulk aggregation read one array cell per row instead of interning one
    /// value per row per query. See [`crate::columns::IdGroupCodes`].
    pub(super) group_codes_by_id: Arc<parking_lot::Mutex<crate::columns::IdGroupCodesCache>>,
    /// Cached full label scans for the committed-read path, one shared sorted id
    /// vector per label, valid for exactly one write generation. Filters, the
    /// vectorized executor, and the counting kernels each enumerate a whole
    /// label per query, and with no plan cache the same label is rescanned
    /// through LMDB on every execution; this pins that scan until a committed
    /// write moves the generation. Transaction-scoped label reads bypass it,
    /// because an open write transaction must see its own uncommitted labels.
    pub(super) label_scans: Arc<parking_lot::Mutex<index::LabelScanCache>>,
    pub(super) n_threads: Arc<std::sync::atomic::AtomicI32>,
    /// Type-erased extension cache. Higher-level crates attach caches (e.g. the
    /// HNSW vector index) to a Graph without creating a circular dependency,
    /// through the `get_extension`, `set_extension`, and
    /// `get_or_init_extension_with` methods. Keys are `std::any::TypeId`; values
    /// are `Arc<dyn Any + Send + Sync>`.
    pub(crate) extensions: Arc<parking_lot::Mutex<AHashMap<StdTypeId, Box<dyn Any + Send + Sync>>>>,
    /// Test-only injection points; see [`TestHooks`]. Never compiled into a
    /// release build.
    #[cfg(test)]
    pub(super) test_hooks: Arc<TestHooks>,
}

/// One test-only injection point: a closure the test installs, fired at most
/// once at its call site.
#[cfg(test)]
pub(super) type HookSlot = parking_lot::Mutex<Option<Box<dyn Fn() + Send>>>;

/// Test-only injection points for the race-condition tests. Instance-scoped,
/// so parallel tests over their own `TempDir` graphs cannot interfere. Each
/// hook fires at most once: [`TestHooks::fire`] takes the closure out before
/// calling it, so a hook that writes back into the graph cannot re-trigger
/// itself, and later passes through the same site run unhooked.
#[cfg(test)]
#[derive(Default)]
pub(super) struct TestHooks {
    /// Fires inside [`Graph::update`] after `commit_and_publish` and before
    /// the column bookkeeping, while the write lock is still held. This is
    /// the window the columns stamp race needs: the persisted generation has
    /// moved, and the touched ids are not yet in the pending buffer.
    pub(super) after_commit_before_column_bookkeeping: HookSlot,
    /// Fires inside [`Graph::schema_has_edge`] after the probe computes its
    /// verdict and before `memoize_schema_probe`. This is the window the memo
    /// race needs: a write committing here makes the verdict describe
    /// pre-commit state.
    pub(super) before_schema_memoize: HookSlot,
}

#[cfg(test)]
impl TestHooks {
    pub(super) fn fire(slot: &HookSlot) {
        let hook = slot.lock().take();
        if let Some(hook) = hook {
            hook();
        }
    }
}

/// A read-only transaction on the graph.
pub struct ReadTxn<'a> {
    pub(super) graph: &'a Graph,
    pub(super) rtxn: crate::storage::OwnedRoTxn<'a>,
}

/// A read-write transaction on the graph.
pub struct WriteTxn<'a> {
    pub(super) graph: &'a Graph,
    pub(super) wtxn: crate::storage::RwTxn<'a>,
    pub(super) mutations_count: usize,
    /// Structural mutations staged during this transaction, flushed to the
    /// `CsrCache` only on commit so an aborted transaction records nothing.
    pub(super) delta: crate::csr::GraphDelta,
    /// Per-transaction memo for work that is identical across the records of one
    /// batch. See [`WriteBatchCache`].
    pub(super) cache: WriteBatchCache,
}

/// Holds the answers that stay true for the whole of one write transaction, so
/// that a bulk write pays for them once instead of once per record.
///
/// A batch of a million edges asked the same three questions a million times:
/// what integer is this relationship type (a `format!` and a `meta` lookup),
/// which property indexes are active for it (a `format!` and a `meta` prefix
/// scan, paid even when there are none, which is the common case), and does this
/// endpoint exist (a lookup in a tree the size of the graph). Measured against
/// the storage layer, those and the id allocation were most of an edge insert:
/// the four LMDB writes an edge performs total about 1.1 µs against a measured
/// 4.9 µs per edge.
///
/// Every entry is safe for exactly one transaction and no longer. There is one
/// writer at a time, so nothing else can change a registry or an index
/// definition underneath this; what this transaction changes itself, it records
/// here too. The endpoint memo is the one that can go stale from inside, since a
/// node deleted later in the same transaction must stop counting as present, so
/// a delete clears it.
#[derive(Default)]
pub(super) struct WriteBatchCache {
    /// Relationship type name to id, including types created by this
    /// transaction.
    types: AHashMap<String, TypeId>,
    /// Active edge property indexes per type, as `get_active_edge_indexes`
    /// returns them.
    edge_indexes: AHashMap<TypeId, Vec<(PropKeyId, u8)>>,
    /// Node ids this transaction has already proved exist. It holds one id per
    /// distinct endpoint the transaction touches and is released only with the
    /// transaction, so a million-node bulk load carries roughly 18 MB of it.
    known_nodes: AHashSet<NodeId>,
}

impl WriteBatchCache {
    fn knows_node(&self, id: NodeId) -> bool {
        self.known_nodes.contains(&id)
    }

    fn remember_node(&mut self, id: NodeId) {
        self.known_nodes.insert(id);
    }

    fn type_id(&self, name: &str) -> Option<TypeId> {
        self.types.get(name).copied()
    }

    fn remember_type(&mut self, name: &str, id: TypeId) {
        self.types.insert(name.to_string(), id);
    }

    /// Returns the active edge indexes for `type_id`, computing them with `f` on
    /// the first ask.
    pub(super) fn edge_indexes_or_insert<E>(
        &mut self,
        type_id: TypeId,
        f: impl FnOnce() -> Result<Vec<(PropKeyId, u8)>, E>,
    ) -> Result<&[(PropKeyId, u8)], E> {
        if !self.edge_indexes.contains_key(&type_id) {
            let computed = f()?;
            self.edge_indexes.insert(type_id, computed);
        }
        Ok(&self.edge_indexes[&type_id])
    }

    /// Forgets the endpoint memo. Called by any node deletion, since a node this
    /// transaction removes must stop satisfying a later edge's existence check.
    /// It drops every entry rather than the deleted id alone, so a batch that
    /// interleaves deletions re-proves each endpoint against storage.
    pub(super) fn invalidate_nodes(&mut self) {
        self.known_nodes.clear();
    }
}

thread_local! {
    /// Identity of the LMDB environment whose `Graph::update` closure this
    /// thread is currently inside (0 when none). LMDB permits only one active
    /// writer transaction per environment; a stray call to an auto-committing
    /// `Graph` mutation method on the SAME environment (which opens its own
    /// writer transaction) while this is set would block forever on the
    /// writer lock `Graph::update` already holds. Keyed by environment so
    /// mutating a different, independent `Graph` inside the closure (a safe
    /// pattern, e.g. copying between databases) does not trip the assert.
    /// Checked at the top of every auto-committing mutation method and of
    /// `Graph::update` itself, so a missed conversion to the `WriteTxn`-based
    /// method (or a nested `update` on the same graph) becomes an immediate,
    /// precisely located debug-build panic instead of a silent hang.
    static IN_WRITE_TXN: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

struct WriteTxnGuard {
    previous: usize,
}

impl WriteTxnGuard {
    fn enter(env_id: usize) -> Self {
        let previous = IN_WRITE_TXN.with(|f| f.replace(env_id));
        WriteTxnGuard { previous }
    }
}

impl Drop for WriteTxnGuard {
    fn drop(&mut self) {
        IN_WRITE_TXN.with(|f| f.set(self.previous));
    }
}

impl Graph {
    /// A stable per-environment identity for the deadlock tripwire.
    fn write_txn_env_id(&self) -> usize {
        Arc::as_ptr(&self.storage) as usize
    }

    fn debug_assert_not_in_write_txn(&self) {
        debug_assert!(
            IN_WRITE_TXN.with(|f| f.get()) != self.write_txn_env_id(),
            "an auto-committing Graph method (or a nested Graph::update) was called while a \
             WriteTxn from Graph::update was already open on this graph on this thread; call \
             the WriteTxn method instead to avoid a same-thread deadlock on LMDB's \
             single-writer lock"
        );
    }
}

impl Graph {
    /// Open (creating if absent) the database at `path`.
    ///
    /// `map_size_gb` is the size of the LMDB memory map, and it is an upper bound
    /// on how large the database may grow for the lifetime of this handle, not an
    /// allocation: LMDB reserves the address range and commits pages as they are
    /// written, so a large value costs virtual address space rather than disk or
    /// RAM. There is no resize path. Once the data exceeds the bound, every write
    /// fails with the underlying `MDB_MAP_FULL` through [`Error::Storage`] until
    /// the database is reopened with a larger value, which is safe to do and keeps
    /// the existing data. Size it for the eventual database, not the current one.
    ///
    /// Opening builds none of the derived structures; see the comment inside.
    pub fn open(path: &Path, map_size_gb: usize) -> Result<Self, Error> {
        let storage = Storage::open(path, map_size_gb)?;
        // Older versions persisted the CSR snapshot next to the LMDB files but
        // never read it back; remove the stale artifact if one is present.
        let _ = std::fs::remove_file(path.join("csr_snapshot.bin"));
        let storage = Arc::new(storage);
        // Opening builds nothing. The CSR snapshot is built by the freshness gate
        // (`ensure_snapshot_fresh`) when a consumer that needs it first runs, and
        // every such consumer already calls that gate. Building it here instead cost
        // a full edge scan on every open, which is time a workload of point lookups,
        // property reads, or point adjacency never uses: those paths read LMDB
        // directly. On a large database that eager work dominated the whole session's
        // latency, and it was repaid on every reopen.
        let csr_cache = Arc::new(CsrCache::new_unbuilt());
        Ok(Self {
            storage,
            _write_lock: Arc::new(ReentrantMutex::new(())),
            csr_cache,
            prop_columns: Arc::new(crate::columns::ColumnsCache::default()),
            edge_columns: Arc::new(crate::columns::ColumnsCache::default()),
            edge_fanout: Arc::new(parking_lot::Mutex::new(None)),
            schema_probes: Arc::new(parking_lot::Mutex::new((0, AHashMap::new()))),
            group_codes_by_id: Arc::new(parking_lot::Mutex::new(
                crate::columns::IdGroupCodesCache::default(),
            )),
            label_scans: Arc::new(parking_lot::Mutex::new(index::LabelScanCache::default())),
            n_threads: Arc::new(std::sync::atomic::AtomicI32::new(0)),
            extensions: Arc::new(parking_lot::Mutex::new(AHashMap::new())),
            #[cfg(test)]
            test_hooks: Arc::new(TestHooks::default()),
        })
    }

    /// Set the thread count for the parallel read passes, overriding the
    /// `ISSUNDB_NUM_THREADS` environment variable. Set to 0 to restore the default
    /// behavior, which resolves through `threads::resolve`: `ISSUNDB_NUM_THREADS`,
    /// then `OMP_NUM_THREADS`, then the machine's parallelism.
    ///
    /// Every parallel consumer (the counting kernels and the analytics passes that
    /// split over nodes or sources) shares that resolution, so this one knob has one
    /// meaning. There is no pool to configure: each pass resolves the budget when it
    /// starts and spawns scoped threads for its own duration, so a call here takes
    /// effect on the next pass and never fails.
    pub fn set_thread_count(&self, n: i32) -> Result<(), Error> {
        self.n_threads
            .store(n, std::sync::atomic::Ordering::Release);
        Ok(())
    }

    /// Read one property of a node as the `serde_json::Value` that decoding the
    /// stored record would give. Returns `None` for a nonexistent node and
    /// `Some(Value::Null)` for a missing property. Either way the result
    /// reflects committed state.
    ///
    /// Served through the in-memory property columns once they exist, refreshing
    /// them against pending writes first; while they are absent the read goes
    /// straight to storage instead of building them (see
    /// [`crate::columns::ColumnsCache::should_serve_directly`]).
    pub fn node_prop_json(
        &self,
        id: NodeId,
        prop: &str,
    ) -> Result<Option<serde_json::Value>, Error> {
        // One property of one node is an LMDB point read. Serving it by building
        // every column costs a full node scan, which is the wrong trade for a
        // point query; the read only goes through the columns once they exist,
        // or once enough direct reads have amortized building them.
        if self.prop_columns.should_serve_directly(1) {
            let Some(obj) = self.direct_node_props(id)? else {
                return Ok(None);
            };
            return Ok(Some(
                obj.get(prop).cloned().unwrap_or(serde_json::Value::Null),
            ));
        }
        self.prop_columns.with_fresh(&self.storage, |cols| {
            cols.id_to_dense.get(&id).map(|&d| {
                cols.cols
                    .get(prop)
                    .and_then(|c| c.get_json_opt(d as usize))
                    .unwrap_or(serde_json::Value::Null)
            })
        })
    }

    /// Gathers `props` for each id in `ids` through the in-memory property columns,
    /// the bulk form of [`Graph::node_prop_json`], row-major (`out[i][j]` is
    /// `props[j]` on `ids[i]`). One columns refresh covers the whole gather,
    /// and each id resolves to its dense index once. A missing property reads
    /// as `Value::Null`; a nonexistent node is [`Error::NodeNotFound`].
    pub fn node_props_json_table(
        &self,
        ids: &[NodeId],
        props: &[&str],
    ) -> Result<Vec<Vec<serde_json::Value>>, Error> {
        if self.prop_columns.should_serve_directly(ids.len()) {
            // One transaction for the whole gather, so the request is a single
            // point in time and pays one begin/end pair rather than one per id.
            return self
                .direct_node_props_many(ids)?
                .into_iter()
                .zip(ids)
                .map(|(obj, &id)| {
                    let obj = obj.ok_or(Error::NodeNotFound(id))?;
                    Ok(props
                        .iter()
                        .map(|p| obj.get(*p).cloned().unwrap_or(serde_json::Value::Null))
                        .collect())
                })
                .collect();
        }
        self.prop_columns
            .with_fresh(&self.storage, |cols| cols.props_table(ids, props))?
    }

    /// Gathers one property as a flat column, the single-property form of
    /// [`Graph::node_props_json_table`]. `out[i]` is the value of `prop` on `ids[i]`, so a
    /// bulk single-property gather does not pay one row vector allocation per
    /// id. A missing property reads as `Value::Null`; a nonexistent node is
    /// [`Error::NodeNotFound`].
    pub fn node_prop_json_column(
        &self,
        ids: &[NodeId],
        prop: &str,
    ) -> Result<Vec<serde_json::Value>, Error> {
        if self.prop_columns.should_serve_directly(ids.len()) {
            return self
                .direct_node_props_many(ids)?
                .into_iter()
                .zip(ids)
                .map(|(obj, &id)| {
                    let obj = obj.ok_or(Error::NodeNotFound(id))?;
                    Ok(obj.get(prop).cloned().unwrap_or(serde_json::Value::Null))
                })
                .collect();
        }
        self.prop_columns
            .with_fresh(&self.storage, |cols| cols.prop_column(ids, prop))?
    }

    /// One node's user properties decoded from storage, the way the column
    /// build decodes them, so a gather served directly and the same gather
    /// served through the columns cannot disagree. `None` if the node is gone.
    fn direct_node_props(&self, id: NodeId) -> Result<Option<serde_json::Value>, Error> {
        <crate::columns::NodeSource as crate::columns::ColumnSource>::fetch_one(&self.storage, id)
    }

    /// [`Graph::direct_node_props`] for many ids under one transaction, in input
    /// order. `None` for a node that is gone.
    fn direct_node_props_many(
        &self,
        ids: &[NodeId],
    ) -> Result<Vec<Option<serde_json::Value>>, Error> {
        <crate::columns::NodeSource as crate::columns::ColumnSource>::fetch_many(&self.storage, ids)
    }

    /// Whether each of `ids` carries a non-null value for `prop`, in input order.
    ///
    /// A node that is not there reads as absent rather than raising, because the
    /// callers' ids come from the CSR snapshot, which can lag a deletion; treating
    /// the gap as a null value is what the row pipeline would effectively produce
    /// for a row a stale snapshot should no longer have offered.
    ///
    /// Honors the same small-request path the property gathers do, so resolving
    /// presence for a handful of nodes costs a handful of point reads instead of
    /// one full scan to build every column. That is the whole point of it existing
    /// separately: the counting kernels need presence for the neighbors they
    /// actually visit, not a dense mask over the entire graph.
    pub(super) fn nodes_prop_present(
        &self,
        ids: &[NodeId],
        prop: &str,
    ) -> Result<Vec<bool>, Error> {
        if self.prop_columns.should_serve_directly(ids.len()) {
            return Ok(self
                .direct_node_props_many(ids)?
                .into_iter()
                .map(|obj| obj.is_some_and(|o| o.get(prop).is_some_and(|v| !v.is_null())))
                .collect());
        }
        self.prop_columns.with_fresh(&self.storage, |cols| {
            ids.iter()
                .map(|id| match (cols.id_to_dense.get(id), cols.cols.get(prop)) {
                    (Some(&d), Some(col)) => col.is_present(d as usize),
                    // Either the columns never saw this entity or no such property
                    // exists anywhere; both read as null.
                    _ => false,
                })
                .collect()
        })
    }

    /// Evaluate `prop <op> rhs` for each of `ids` directly against the typed
    /// in-memory property column, one keep flag per id in input order, without
    /// materializing a `Value` per row. The semantics are exactly the outcome a
    /// Cypher comparison filter keeps a row on; see
    /// [`crate::columns::PropColumns::cmp_mask`] for the three rules. A
    /// nonexistent node is [`Error::NodeNotFound`].
    ///
    /// `Ok(None)` declines, and the caller falls back to gathering and
    /// comparing boxed values: a small request on a cold graph must not build
    /// every column (the same size test the property gathers apply), and a
    /// mixed-kind `Json` fallback column has no typed storage to compare
    /// against.
    pub fn nodes_prop_cmp_mask(
        &self,
        ids: &[NodeId],
        prop: &str,
        op: crate::columns::PropCmp,
        rhs: &serde_json::Value,
    ) -> Result<Option<Vec<bool>>, Error> {
        if self.prop_columns.should_serve_directly(ids.len()) {
            return Ok(None);
        }
        self.prop_columns
            .with_fresh(&self.storage, |cols| cols.cmp_mask(ids, prop, op, rhs))?
    }

    /// Build the in-memory property columns now, if they are not built already.
    ///
    /// Every reader either serves a small request without them or, for the advisory
    /// statistics, declines rather than pay for them, so nothing builds them as a
    /// side effect of a small workload. That is deliberate: the build is one full
    /// entity scan, and it used to dominate cold-start latency. This is the
    /// deliberate way to ask for it, for a caller that wants the optimizer's
    /// selectivity estimates and zone-map pruning available on a cold graph, or that
    /// would rather pay the scan once up front than have a later bulk read pay it.
    ///
    /// It replaces an accident: `node_prop_group_codes` used to build
    /// unconditionally, so "call it and discard the result" was the idiom for
    /// warming the columns. Grouping now follows the same size test as the other
    /// readers, and warming them is this call.
    /// It is also the columns cache file's save site (the counterpart of
    /// `rebuild_csr` for the CSR cache file): materializing persists the built
    /// set next to the LMDB files, so a later process loads it instead of
    /// scanning, and a repeat at an unchanged generation rewrites nothing. No
    /// lazy build saves, so a read-only workload never writes a file as a side
    /// effect of a query.
    pub fn materialize_property_columns(&self) -> Result<(), Error> {
        #[cfg(feature = "lmdb")]
        {
            // Captured before the build, and under the write lock. Every
            // mutation holds that lock from before its commit until after it
            // records its touched ids. So at the capture, every commit the
            // stamp counts has already recorded its delta, and the drain
            // inside `with_fresh` absorbs all of them. That is the invariant
            // the saved file needs: stamp <= absorbed content. A write landing
            // after the capture leaves the file stale, which is safe. Reading
            // the generation without the lock is not: it could see a commit
            // whose touched ids were not yet recorded, stamp the file one
            // generation ahead of its content, and a later process would load
            // stale values as fresh.
            let persisted_gen = {
                let _guard = self._write_lock.lock();
                let rtxn = self.storage.env.read_txn()?;
                crate::storage::ids::commit_gen(&self.storage, &rtxn)?
            };
            let _quiet = crate::columns::MaterializingColumns::install();
            self.prop_columns.with_fresh(&self.storage, |cols| {
                let _ = crate::cache_file::save_columns(&self.storage, cols, persisted_gen);
            })
        }
        #[cfg(not(feature = "lmdb"))]
        {
            let _quiet = crate::columns::MaterializingColumns::install();
            self.prop_columns.with_fresh(&self.storage, |_| ())
        }
    }

    /// Group `ids` by the exact value of `prop` through the in-memory
    /// property columns: one dense group code per id, plus one representative
    /// value per code (the first occurrence). Null and missing property
    /// values share one code represented by `Value::Null`; a nonexistent node
    /// is [`Error::NodeNotFound`]. Codes are assigned under value identity,
    /// which for the typed columns needs no per-row value materialization.
    pub fn node_prop_group_codes(
        &self,
        ids: &[NodeId],
        prop: &str,
    ) -> Result<(Vec<u32>, Vec<serde_json::Value>), Error> {
        // A small request is grouped over an ephemeral column set built from just
        // those nodes, rather than by building every column from a full scan.
        // Grouping is a bulk read only when the id set is bulk; a grouped count
        // whose groups are a handful of nodes was paying one full node scan with a
        // decode per node, which is the cold-start cost the small-gather path exists
        // to avoid, and the caller had no way to opt out.
        //
        // The ephemeral set goes through the same `from_items` and `group_codes` as
        // the shared one, so this is the same grouping code over a narrower
        // population, not a second implementation of it. A node that is gone is left
        // out, which makes `group_codes` report `NodeNotFound` for it exactly as the
        // shared columns would.
        if self.prop_columns.should_serve_directly(ids.len()) {
            let fetched = self.direct_node_props_many(ids)?;
            let items: Vec<(NodeId, serde_json::Value)> = ids
                .iter()
                .zip(fetched)
                .filter_map(|(&id, obj)| obj.map(|o| (id, o)))
                .collect();
            // Only the grouped property is columnarized; the rest would be built
            // and dropped.
            let cols = crate::columns::PropColumns::<crate::columns::NodeSource>::from_items_for(
                items,
                Some(prop),
            );
            return cols.group_codes(ids, prop);
        }
        self.prop_columns
            .with_fresh(&self.storage, |cols| cols.group_codes(ids, prop))?
    }

    /// The id-indexed form of [`Graph::node_prop_group_codes`], one shared
    /// array over every node, `codes[node_id]` the node's group code under
    /// exact value identity and [`crate::columns::ID_GROUP_ABSENT`] where no
    /// such node exists, plus one representative value per code. Cached per
    /// write generation and shared, so a grouped bulk aggregation pays the
    /// value interning once per generation and one array read per row per
    /// query afterward. Building it costs one pass over every node (and the
    /// full column build when the columns are absent), so a caller with a
    /// small row set wants [`Graph::node_prop_group_codes`] instead.
    pub fn node_prop_group_codes_by_id(
        &self,
        prop: &str,
    ) -> Result<std::sync::Arc<crate::columns::IdGroupCodes>, Error> {
        let mut cache = self.group_codes_by_id.lock();
        let generation = self.csr_cache.current_gen();
        if cache.generation != generation {
            cache.by_prop.clear();
            cache.generation = generation;
        }
        if let Some(hit) = cache.by_prop.get(prop) {
            return Ok(hit.clone());
        }
        let built = self.prop_columns.with_fresh(&self.storage, |cols| {
            let (dense_codes, reps) = cols.group_codes(&cols.dense_to_id, prop)?;
            let span = cols
                .dense_to_id
                .iter()
                .copied()
                .max()
                .map_or(0, |m| m as usize + 1);
            let mut codes = vec![crate::columns::ID_GROUP_ABSENT; span];
            for (dense, &id) in cols.dense_to_id.iter().enumerate() {
                codes[id as usize] = dense_codes[dense];
            }
            Ok::<_, Error>(crate::columns::IdGroupCodes {
                codes,
                reps: std::sync::Arc::new(reps),
            })
        })??;
        let arc = std::sync::Arc::new(built);
        cache.by_prop.insert(prop.to_string(), arc.clone());
        Ok(arc)
    }

    /// Build the in-memory edge property columns now, if they are not built
    /// already: the edge counterpart of [`Graph::materialize_property_columns`],
    /// with the same contract. Nothing builds the edge columns as a side effect
    /// of a small workload, so this is the deliberate warm-up, and it is the
    /// edge columns cache file's save site; a repeat at an unchanged generation
    /// rewrites nothing.
    pub fn materialize_edge_property_columns(&self) -> Result<(), Error> {
        #[cfg(feature = "lmdb")]
        {
            // Captured under the write lock and before the build, for the
            // stamp <= absorbed content invariant explained in
            // [`Graph::materialize_property_columns`].
            let persisted_gen = {
                let _guard = self._write_lock.lock();
                let rtxn = self.storage.env.read_txn()?;
                crate::storage::ids::commit_gen(&self.storage, &rtxn)?
            };
            let _quiet = crate::columns::MaterializingColumns::install();
            self.edge_columns.with_fresh(&self.storage, |cols| {
                let _ = crate::cache_file::save_columns(&self.storage, cols, persisted_gen);
            })
        }
        #[cfg(not(feature = "lmdb"))]
        {
            let _quiet = crate::columns::MaterializingColumns::install();
            self.edge_columns.with_fresh(&self.storage, |_| ())
        }
    }

    // ------------------------------------------------------------------
    // Edge property columns
    //
    // The edge counterparts of the node column readers above, backed by an
    // independent columnar cache over the `edges` sub-database. They let the
    // query layer gather edge (relationship) properties in bulk through a
    // dense-index read instead of an LMDB point lookup plus a msgpack decode
    // per access. Semantics mirror the node methods exactly: a missing
    // property reads as `Value::Null`; a nonexistent edge is
    // [`Error::EdgeNotFound`].
    // ------------------------------------------------------------------

    /// Read one property of an edge through the in-memory edge property
    /// columns. Returns `None` for a nonexistent edge and `Some(Value::Null)`
    /// for a missing property.
    pub fn edge_prop_json(
        &self,
        id: EdgeId,
        prop: &str,
    ) -> Result<Option<serde_json::Value>, Error> {
        self.edge_columns.with_fresh(&self.storage, |cols| {
            cols.id_to_dense.get(&id).map(|&d| {
                cols.cols
                    .get(prop)
                    .and_then(|c| c.get_json_opt(d as usize))
                    .unwrap_or(serde_json::Value::Null)
            })
        })
    }

    /// Bulk row-major gather of `props` for each edge id in `ids`.
    pub fn edge_props_json_table(
        &self,
        ids: &[EdgeId],
        props: &[&str],
    ) -> Result<Vec<Vec<serde_json::Value>>, Error> {
        self.edge_columns
            .with_fresh(&self.storage, |cols| cols.props_table(ids, props))?
    }

    /// Gathers one property column for edges, where `out[i]` is `prop` on `ids[i]`.
    pub fn edge_prop_json_column(
        &self,
        ids: &[EdgeId],
        prop: &str,
    ) -> Result<Vec<serde_json::Value>, Error> {
        self.edge_columns
            .with_fresh(&self.storage, |cols| cols.prop_column(ids, prop))?
    }

    /// Group `ids` by the exact value of edge property `prop`: one dense group
    /// code per id plus one representative value per code.
    pub fn edge_prop_group_codes(
        &self,
        ids: &[EdgeId],
        prop: &str,
    ) -> Result<(Vec<u32>, Vec<serde_json::Value>), Error> {
        self.edge_columns
            .with_fresh(&self.storage, |cols| cols.group_codes(ids, prop))?
    }

    /// The minimum and maximum non-null value of one node property, from the
    /// lazily computed statistics over the in-memory property columns.
    /// `None` when the property has no typed column or no non-null values, and
    /// also when the columns are not built yet: this reader never builds them,
    /// because it is advisory (see [`Graph::estimate_equality_selectivity`]).
    pub fn node_prop_min_max(
        &self,
        prop: &str,
    ) -> Result<Option<(serde_json::Value, serde_json::Value)>, Error> {
        Ok(self
            .prop_columns
            .with_existing_mut(&self.storage, |cols| {
                cols.prop_stats(prop)
                    .map(|s| (s.min.clone(), s.max.clone()))
            })?
            .flatten())
    }

    /// Estimated fraction of non-null values of `prop` inside the given
    /// bounds (either bound optional), from the property's equi-depth
    /// histogram. `None` when no statistics exist for the property or the
    /// columns are not built yet; this reader never builds them.
    pub fn estimate_range_selectivity(
        &self,
        prop: &str,
        lower: Option<&serde_json::Value>,
        upper: Option<&serde_json::Value>,
    ) -> Result<Option<f64>, Error> {
        Ok(self
            .prop_columns
            .with_existing_mut(&self.storage, |cols| {
                cols.prop_stats(prop)
                    .map(|s| s.histogram.estimate_range_selectivity(lower, upper))
            })?
            .flatten())
    }

    /// Estimated fraction of non-null values of `prop` equal to `val`: exact
    /// for the property's most common values, histogram-estimated otherwise.
    ///
    /// `None` when no statistics exist for the property, and also when the
    /// property columns have not been built yet: the estimate only weights plan
    /// choices, so answering is never worth one full node scan on a query that
    /// would not otherwise materialize the columns.
    pub fn estimate_equality_selectivity(
        &self,
        prop: &str,
        val: &serde_json::Value,
    ) -> Result<Option<f64>, Error> {
        Ok(self
            .prop_columns
            .with_existing_mut(&self.storage, |cols| {
                cols.prop_stats(prop).map(|s| s.equality_selectivity(val))
            })?
            .flatten())
    }

    /// Store an extension value (as `Arc`) keyed by its concrete type.
    /// Replaces any existing value of the same type.
    pub fn set_extension<T: Any + Send + Sync>(&self, val: Arc<T>) {
        self.extensions
            .lock()
            .insert(StdTypeId::of::<T>(), Box::new(val));
    }

    /// Retrieve an `Arc` to a previously stored extension value, or `None` if absent.
    pub fn get_extension<T: Any + Send + Sync>(&self) -> Option<Arc<T>> {
        self.extensions
            .lock()
            .get(&StdTypeId::of::<T>())
            .and_then(|b| b.downcast_ref::<Arc<T>>())
            .cloned()
    }

    /// Return the extension of type `T`, initializing it with `init` if absent.
    ///
    /// `init` runs without the extensions lock held, so it may call back into
    /// the graph (for example, to read from storage) without risking a lock
    /// ordering problem. If two threads initialize concurrently, both may run
    /// `init`, but only the first stored value is kept and every caller observes
    /// that same `Arc`. `init` is fallible; on error nothing is stored and the
    /// error is propagated.
    pub fn get_or_init_extension_with<T, E, F>(&self, init: F) -> Result<Arc<T>, E>
    where
        T: Any + Send + Sync,
        F: FnOnce() -> Result<Arc<T>, E>,
    {
        if let Some(existing) = self.get_extension::<T>() {
            return Ok(existing);
        }
        let value = init()?;
        let mut ext = self.extensions.lock();
        // Another thread may have initialized while we built ours; prefer the
        // already-stored value so all callers share one instance.
        if let Some(existing) = ext
            .get(&StdTypeId::of::<T>())
            .and_then(|b| b.downcast_ref::<Arc<T>>())
        {
            return Ok(existing.clone());
        }
        ext.insert(StdTypeId::of::<T>(), Box::new(value.clone()));
        Ok(value)
    }

    /// Execute a read-only transaction inside a closure.
    pub fn view<F, T>(&self, f: F) -> Result<T, Error>
    where
        F: FnOnce(&ReadTxn) -> Result<T, Error>,
    {
        let rtxn = self.storage.env.read_txn()?;
        let txn = ReadTxn { graph: self, rtxn };
        f(&txn)
    }

    /// Execute a read-write transaction inside a closure.
    pub fn update<F, T>(&self, f: F) -> Result<T, Error>
    where
        F: FnOnce(&mut WriteTxn) -> Result<T, Error>,
    {
        self.debug_assert_not_in_write_txn();
        let _guard = self._write_lock.lock();
        let wtxn = self.storage.env.write_txn()?;
        let mut txn = WriteTxn {
            graph: self,
            wtxn,
            mutations_count: 0,
            delta: crate::csr::GraphDelta::default(),
            cache: WriteBatchCache::default(),
        };
        let _txn_guard = WriteTxnGuard::enter(self.write_txn_env_id());
        match f(&mut txn) {
            Ok(val) => {
                let WriteTxn {
                    wtxn,
                    mutations_count,
                    delta,
                    graph: _,
                    cache: _,
                } = txn;
                // Publish before any other bookkeeping, so the window in which
                // the caches claim to be current while LMDB already holds this
                // write is one atomic increment wide rather than the width of
                // the batch. See `CsrCache::advance_write_gen`.
                self.commit_and_publish(wtxn, mutations_count)?;
                #[cfg(test)]
                TestHooks::fire(&self.test_hooks.after_commit_before_column_bookkeeping);
                // Column bookkeeping next. The CSR snapshot needs nothing here: the
                // generation bump above is what tells a reader its snapshot lags, and
                // the refresh rebuilds from storage rather than from a delta.
                //
                // The columns are still a window: for as long as this bookkeeping
                // takes, a reader can see committed data through a column set that
                // has not absorbed it. Closing that properly is the read-isolation
                // question, not an ordering one. Until then the generation bump,
                // which happens first and is a single atomic, is what the snapshot
                // gate reads.
                if delta.force_full {
                    self.prop_columns.record_force_full();
                } else {
                    self.prop_columns.record_touched_many(&delta.added_nodes);
                    self.prop_columns.record_touched_many(&delta.updated_nodes);
                }
                // Edge columns: an edge removal (or a node deletion that may
                // cascade to edges) reshuffles the dense edge mapping, so fall
                // back to a full rebuild; otherwise patch the added and
                // updated edges in.
                if delta.force_full || delta.removed_edge {
                    self.edge_columns.record_force_full();
                } else {
                    self.edge_columns.record_touched_many(&delta.added_edge_ids);
                    self.edge_columns.record_touched_many(&delta.updated_edges);
                }
                if mutations_count > 0 {
                    self.maybe_spawn_rebuild_n(mutations_count);
                }
                Ok(val)
            }
            Err(err) => {
                txn.wtxn.abort();
                Err(err)
            }
        }
    }

    /// Commit `wtxn` and publish the write to the caches' freshness counters as
    /// one step, where `count` is the number of mutations the transaction made.
    ///
    /// Every mutation that changes adjacency, an edge weight, or a node record
    /// commits through here rather than calling `wtxn.commit()` directly. The
    /// publish is what makes every freshness gate notice the write, so a method
    /// that committed without it would leave the caches permanently claiming to be
    /// current rather than briefly.
    ///
    /// This is convention plus a test, not an enforced invariant: `wtxn.commit()`
    /// is still called directly by the index, vector, and FTS writers, none of
    /// which touch adjacency or a cached property, so nothing structurally prevents
    /// a new mutation method from committing without publishing.
    /// `publish_tests::every_committing_mutation_publishes_the_write` enumerates
    /// today's methods by hand, so add a new one to it. Ordering inside here is
    /// deliberate: see [`crate::csr::CsrCache::advance_write_gen`].
    pub(super) fn commit_and_publish(
        &self,
        mut wtxn: crate::storage::RwTxn<'_>,
        count: usize,
    ) -> Result<(), Error> {
        // The persisted generation advances inside the transaction, so it is
        // atomic with the mutations it describes; it is what lets a later
        // process decide whether an on-disk derived structure (the CSR
        // cache file) still reflects storage, which the in-memory counter below
        // cannot, since that one restarts with the process.
        if count > 0 {
            crate::storage::ids::bump_commit_gen(&self.storage, &mut wtxn)?;
        }
        wtxn.commit()?;
        self.csr_cache.advance_write_gen(count as u64);
        Ok(())
    }

    /// Hold the write lock for the duration of `f`, executing `f` without
    /// starting an LMDB transaction. Use this to make a multi-step read-then-write
    /// sequence (such as MERGE) atomic with respect to other writers.
    pub fn with_write_lock<F, R>(&self, f: F) -> R
    where
        F: FnOnce() -> R,
    {
        let _guard = self._write_lock.lock();
        f()
    }

    /// Synchronously rebuild the CSR snapshot from LMDB. Useful after bulk
    /// loads or when tests need a consistent read view before the threshold
    /// has been crossed.
    ///
    /// It deliberately does not *ask* for per-edge weights, though it keeps loading
    /// them once something else has. This is the call every bulk load makes (`COPY
    /// ... FROM` and `IMPORT DATABASE` both end with it), and because the request is
    /// sticky, asking here would pin every process that ever loads data to the extra
    /// `edges` scan for the rest of its life, whether or not anything asks a
    /// weighted question. The one consumer that needs them
    /// (`shortest_path_dijkstra`) asks through its own gate on first use.
    #[instrument(skip(self))]
    pub fn rebuild_csr(&self) -> Result<(), Error> {
        // Serialize against every other maintenance path (a foreground refresh and
        // the background rebuild) so no two run concurrently.
        let _maint = self.csr_cache.maintenance.lock();
        // Capture the generation before reading LMDB so writes that land during the
        // build leave the snapshot conservatively stale.
        let built_gen = self.csr_cache.current_gen();
        // The persisted generation, captured before the build for the same
        // conservative reason: a write landing mid-build moves the persisted
        // counter past the value stamped into the cache file, so the file reads as
        // stale rather than claiming a freshness it does not have.
        #[cfg(feature = "lmdb")]
        let persisted_gen = {
            let rtxn = self.storage.env.read_txn()?;
            crate::storage::ids::commit_gen(&self.storage, &rtxn)?
        };
        // Always the full scan, never the cache-file load: this method is the
        // file's save site, so serving the file here would write back whatever
        // it already claimed and a wrong file could never be repaired.
        let snap = self.build_snapshot_from_storage()?;
        // This is the one save site, chosen because every bulk load ends here:
        // the freshness gate's per-write refreshes must not pay a file write per
        // rebuild. A failed save is ignored; the cache file is a cache, and the
        // stale or absent file it leaves behind is refused on load.
        #[cfg(feature = "lmdb")]
        let _ = crate::cache_file::save_csr(
            self.storage.env.path(),
            &snap,
            self.storage.db_id,
            persisted_gen,
        );
        self.csr_cache.install_full(snap, built_gen);
        Ok(())
    }

    /// Create a hot backup of this database to `destination`.
    ///
    /// `destination` is a **file path** for the backup snapshot (e.g.
    /// `/backups/mydb_2026-05-27.mdb`). The file is a complete, portable
    /// LMDB snapshot. Concurrent reads and writes are not blocked.
    ///
    /// To restore, create an empty directory, copy the snapshot file to
    /// `<dir>/data.mdb`, then call `Graph::open(<dir>, map_size_gb)`.
    pub fn backup(&self, destination: &Path) -> Result<(), Error> {
        self.storage.copy_to_file(destination, false)
    }

    /// Same as `backup` but compacts the database during the copy.
    ///
    /// The resulting file is smaller than a raw backup but the operation
    /// takes longer because it rewrites every live page.
    pub fn backup_compact(&self, destination: &Path) -> Result<(), Error> {
        self.storage.copy_to_file(destination, true)
    }

    /// Restore a backup snapshot created by `backup` or `backup_compact` into
    /// a new database directory.
    ///
    /// Creates `dst_dir` if it does not exist, then copies `snapshot_file` into
    /// `dst_dir/data.mdb`. After this call succeeds the caller can open the
    /// restored database with `Graph::open(dst_dir, map_size_gb)`.
    /// Delegates to the storage backend, which is what makes the pair symmetric: a
    /// backend that cannot produce a snapshot (`backup`) must not claim to consume
    /// one. Leaving the copy here meant the in-memory backend reported a successful
    /// restore having restored nothing, while its `backup` correctly refused.
    pub fn restore(snapshot_file: &Path, dst_dir: &Path) -> Result<(), Error> {
        Storage::restore_from_file(snapshot_file, dst_dir)
    }
}

#[cfg(test)]
mod extension_tests {
    use std::sync::Arc;

    use tempfile::TempDir;

    use super::Graph;

    fn open_tmp() -> (TempDir, Graph) {
        let dir = TempDir::new().unwrap();
        let g = Graph::open(dir.path(), 1).unwrap();
        (dir, g)
    }

    /// Extensions are keyed by concrete type: a stored value round-trips, an
    /// absent type returns `None`, and a second `set_extension` replaces the
    /// previous value of the same type.
    #[test]
    fn extension_roundtrip_by_type() {
        let (_dir, g) = open_tmp();
        assert!(g.get_extension::<String>().is_none());

        g.set_extension(Arc::new(String::from("cache")));
        let got = g.get_extension::<String>().expect("extension must exist");
        assert_eq!(*got, "cache");
        assert!(g.get_extension::<u64>().is_none(), "distinct type slot");

        g.set_extension(Arc::new(String::from("replaced")));
        assert_eq!(*g.get_extension::<String>().unwrap(), "replaced");
    }

    /// `get_or_init_extension_with` runs `init` only when the slot is empty;
    /// later callers observe the first stored value.
    #[test]
    fn get_or_init_extension_initializes_once() {
        let (_dir, g) = open_tmp();

        let v1 = g
            .get_or_init_extension_with::<u64, std::convert::Infallible, _>(|| Ok(Arc::new(7)))
            .unwrap();
        assert_eq!(*v1, 7);

        let v2 = g
            .get_or_init_extension_with::<u64, std::convert::Infallible, _>(|| Ok(Arc::new(9)))
            .unwrap();
        assert_eq!(*v2, 7, "second init must not replace the stored value");
    }

    /// An `init` failure stores nothing, so a later successful `init` runs.
    #[test]
    fn get_or_init_extension_propagates_init_error() {
        let (_dir, g) = open_tmp();

        let err = g
            .get_or_init_extension_with::<u64, &str, _>(|| Err("init failed"))
            .unwrap_err();
        assert_eq!(err, "init failed");
        assert!(g.get_extension::<u64>().is_none());

        let v = g
            .get_or_init_extension_with::<u64, &str, _>(|| Ok(Arc::new(7)))
            .unwrap();
        assert_eq!(*v, 7);
    }
}

#[cfg(test)]
mod encode_tests {
    use serde_json::json;

    use super::{MAX_INDEXED_STRING_LEN, decode_property_value, encode_property_value};

    /// A string up to the indexable bound encodes and round-trips; one byte over
    /// the bound is declined so it never overflows the LMDB key size.
    #[test]
    fn over_long_strings_are_not_indexed() {
        let at_limit = json!("a".repeat(MAX_INDEXED_STRING_LEN));
        let encoded = encode_property_value(&at_limit).expect("at-limit string indexes");
        assert_eq!(decode_property_value(&encoded), Some(at_limit));

        let too_long = json!("a".repeat(MAX_INDEXED_STRING_LEN + 1));
        assert_eq!(
            encode_property_value(&too_long),
            None,
            "a string over the bound must not be indexed",
        );
    }

    /// Distinct integers beyond 2^53 must encode to distinct keys. Encoding
    /// purely through `f64` (the previous behavior) collapsed them, causing
    /// index collisions and wrong `nodes_by_property` matches.
    #[test]
    fn large_integers_do_not_collide() {
        let a = encode_property_value(&json!(9_007_199_254_740_992_i64)).unwrap(); // 2^53
        let b = encode_property_value(&json!(9_007_199_254_740_993_i64)).unwrap(); // 2^53 + 1
        assert_ne!(a, b, "distinct large integers must encode distinctly");
    }

    /// An integer and the float of the same real value must encode identically
    /// so they keep comparing equal in the index (Cypher treats `30 = 30.0`).
    #[test]
    fn integer_and_equal_float_unify() {
        assert_eq!(
            encode_property_value(&json!(30)).unwrap(),
            encode_property_value(&json!(30.0)).unwrap(),
        );
        assert_eq!(
            encode_property_value(&json!(0)).unwrap(),
            encode_property_value(&json!(0.0)).unwrap(),
        );
    }

    /// Every numeric encoding must be the same length: property lookups match by
    /// key prefix, so a value whose encoding prefixes another's would alias.
    #[test]
    fn numeric_encoding_is_fixed_length() {
        for v in [
            json!(1),
            json!(-1),
            json!(0),
            json!(i64::MAX),
            json!(i64::MIN),
            json!(3.5),
            json!(-2.5e10),
        ] {
            assert_eq!(encode_property_value(&v).unwrap().len(), 17, "value {v}");
        }
    }

    /// Byte-lexicographic order of encodings must match numeric order, including
    /// across the 2^53 boundary where the disambiguator orders the tie.
    #[test]
    fn numeric_ordering_preserved() {
        let ascending: Vec<i64> = vec![
            i64::MIN,
            -1_000,
            -1,
            0,
            1,
            1_000,
            1 << 53,
            (1 << 53) + 1,
            i64::MAX,
        ];
        let encoded: Vec<Vec<u8>> = ascending
            .iter()
            .map(|v| encode_property_value(&json!(v)).unwrap())
            .collect();
        let mut sorted = encoded.clone();
        sorted.sort();
        assert_eq!(encoded, sorted, "encodings must sort in numeric order");
    }

    /// Large integers must decode back to the exact integer, not a rounded float.
    #[test]
    fn decode_round_trips_large_integer() {
        for v in [
            json!(0),
            json!(-1),
            json!(9_007_199_254_740_993_i64),
            json!(i64::MAX),
        ] {
            let enc = encode_property_value(&v).unwrap();
            assert_eq!(decode_property_value(&enc), Some(v.clone()), "value {v}");
        }
    }
}

// Persistence-dependent: these close a database and reopen the same path, or copy it
// to a file. The in-memory backend starts empty on every `open` by design (see
// `storage::memory`), so their premise does not hold there and the gate states that
// rather than letting them fail as though the backend were broken.
#[cfg(feature = "lmdb")]
#[cfg(test)]
mod restore_tests {
    use serde_json::json;
    use tempfile::TempDir;

    use super::Graph;

    /// Restoring over an existing database must fail rather than truncate it.
    ///
    /// The copy is `fs::copy`, which overwrites, so this used to destroy the
    /// destination and report success. Every front end reaches this function, so the
    /// refusal belongs here rather than in one of them.
    #[test]
    fn restore_refuses_an_existing_database() {
        let src = TempDir::new().unwrap();
        let snap_dir = TempDir::new().unwrap();
        let dst = TempDir::new().unwrap();
        let snap = snap_dir.path().join("a.mdb");

        {
            let a = Graph::open(src.path(), 1).unwrap();
            a.add_node("FromA", &json!({ "n": 1 })).unwrap();
            a.backup(&snap).unwrap();
        }
        {
            let b = Graph::open(dst.path(), 1).unwrap();
            for i in 0..5 {
                b.add_node("FromB", &json!({ "n": i })).unwrap();
            }
        }

        let err = Graph::restore(&snap, dst.path()).unwrap_err();
        assert!(
            err.to_string().contains("already contains a database"),
            "{err}"
        );

        // The destination is untouched.
        let b = Graph::open(dst.path(), 1).unwrap();
        assert_eq!(b.nodes_by_label("FromB").unwrap().len(), 5);
        assert!(b.nodes_by_label("FromA").unwrap().is_empty());

        // A fresh directory still works, including one that does not exist yet.
        let fresh = TempDir::new().unwrap();
        let nested = fresh.path().join("new");
        Graph::restore(&snap, &nested).unwrap();
        let restored = Graph::open(&nested, 1).unwrap();
        assert_eq!(restored.nodes_by_label("FromA").unwrap().len(), 1);
    }

    /// Restoring into a directory with leftover cache files removes them: they
    /// describe whatever database used to live there, and the database identity
    /// they carry means they could never serve the restored one anyway.
    #[test]
    fn restore_removes_leftover_cache_files() {
        let old = TempDir::new().unwrap();
        let snap_dir = TempDir::new().unwrap();
        let snap = snap_dir.path().join("b.mdb");

        {
            let a = Graph::open(old.path(), 1).unwrap();
            let n0 = a.add_node("N", &json!({ "x": 1 })).unwrap();
            let n1 = a.add_node("N", &json!({ "x": 2 })).unwrap();
            a.add_edge(n0, n1, "R", &json!({})).unwrap();
            a.rebuild_csr().unwrap();
            a.materialize_property_columns().unwrap();
        }
        {
            let b_dir = TempDir::new().unwrap();
            let b = Graph::open(b_dir.path(), 1).unwrap();
            b.add_node("FromB", &json!({})).unwrap();
            b.backup(&snap).unwrap();
        }
        // The old database goes away, its cache files stay behind.
        std::fs::remove_file(old.path().join("data.mdb")).unwrap();
        let _ = std::fs::remove_file(old.path().join("lock.mdb"));
        assert!(old.path().join("csr.cache").exists());

        Graph::restore(&snap, old.path()).unwrap();
        let leftover: Vec<_> = std::fs::read_dir(old.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .map(|e| e.path())
            .filter(|p| p.extension().is_some_and(|ext| ext == "cache"))
            .collect();
        assert!(leftover.is_empty(), "leftover cache files: {leftover:?}");

        let restored = Graph::open(old.path(), 1).unwrap();
        assert_eq!(restored.nodes_by_label("FromB").unwrap().len(), 1);
    }
}

// Reopening the same directory is the observable half of the race, so the test
// needs the persistent backend.
#[cfg(test)]
#[cfg(feature = "lmdb")]
mod stamp_race_tests {
    use std::sync::mpsc;

    use serde_json::json;
    use tempfile::TempDir;

    use super::Graph;

    /// The columns cache file must never be stamped ahead of its content.
    ///
    /// The interleaving under test: a writer commits, and a concurrent
    /// materialize reads the bumped persisted generation before the writer
    /// records its touched ids into the columns' pending buffer. Capturing the
    /// generation without the write lock let the materialize refresh against
    /// an empty pending buffer, save pre-write column values, and stamp them
    /// with the post-write generation; a reopened graph then loaded the file
    /// as fresh and served the pre-write value. The hook parks the writer in
    /// exactly that window. Under the fixed code the materialize blocks on the
    /// write lock instead, so it absorbs the write before saving.
    #[test]
    fn a_concurrent_materialize_does_not_stamp_the_cache_file_ahead_of_its_content() {
        let dir = TempDir::new().unwrap();
        let node;
        {
            let g = Graph::open(dir.path(), 1).unwrap();
            node = g.add_node("Person", &json!({ "v": 1 })).unwrap();
            // Build and persist the columns first, so the racing materialize
            // refreshes through the pending buffer rather than a full scan,
            // which would read the committed value and hide the race.
            g.materialize_property_columns().unwrap();

            let (reached_tx, reached_rx) = mpsc::channel::<()>();
            let (release_tx, release_rx) = mpsc::channel::<()>();
            g.test_hooks
                .after_commit_before_column_bookkeeping
                .lock()
                .replace(Box::new(move || {
                    reached_tx.send(()).unwrap();
                    release_rx.recv().unwrap();
                }));

            let writer = {
                let g = g.clone();
                std::thread::spawn(move || {
                    g.update(|txn| txn.update_node(node, &json!({ "v": 2 })))
                        .unwrap();
                })
            };
            // The writer has committed and is parked before its column
            // bookkeeping, still holding the write lock.
            reached_rx.recv().unwrap();
            let materializer = {
                let g = g.clone();
                std::thread::spawn(move || g.materialize_property_columns().unwrap())
            };
            // Ordering help only, not correctness: give the materializer a
            // moment to reach the generation capture before the writer is
            // released. Under the fixed code it blocks there on the write
            // lock; under the racy ordering it completes its save here.
            std::thread::sleep(std::time::Duration::from_millis(100));
            release_tx.send(()).unwrap();
            writer.join().unwrap();
            materializer.join().unwrap();
        }

        let g = Graph::open(dir.path(), 1).unwrap();
        // A full build serves from the cache file when its stamp matches the
        // persisted generation, which is exactly the load a stamp ahead of
        // its content poisons.
        g.materialize_property_columns().unwrap();
        assert_eq!(
            g.node_prop_json(node, "v").unwrap(),
            Some(json!(2)),
            "the reopened graph must serve the committed value through the loaded columns"
        );
    }
}

#[cfg(test)]
mod publish_tests {
    use serde_json::json;
    use tempfile::TempDir;

    use super::Graph;

    /// Every committing mutation must publish its write to the freshness
    /// counters, which is what [`Graph::commit_and_publish`] exists to make
    /// unforgettable. A method that committed without publishing would leave
    /// every gate reporting the caches as current, so a typed expansion or a
    /// graph algorithm would read pre-write state indefinitely rather than for
    /// the length of one atomic increment.
    #[test]
    fn every_committing_mutation_publishes_the_write() {
        let dir = TempDir::new().unwrap();
        let g = Graph::open(dir.path(), 1).unwrap();
        let a = g.add_node("P", &json!({ "n": 1 })).unwrap();
        let b = g.add_node("P", &json!({ "n": 2 })).unwrap();
        let edge = g.add_edge(a, b, "T", &json!({ "weight": 1.0 })).unwrap();
        // Targets for the cases that consume what they touch, created up front so
        // the mutation under test is the only write inside its own window.
        let victim_node = g.add_node("P", &json!({})).unwrap();
        let victim_edge = g.add_edge(a, b, "T", &json!({})).unwrap();
        let label_target = g.add_node("P", &json!({})).unwrap();

        macro_rules! assert_publishes {
            ($name:literal, $body:block) => {{
                g.rebuild_csr().unwrap();
                assert!(
                    !g.csr_cache.snapshot_is_stale(),
                    concat!($name, ": a fresh rebuild must report current")
                );
                $body
                assert!(
                    g.csr_cache.snapshot_is_stale(),
                    concat!($name, " committed without publishing the write generation")
                );
            }};
        }

        assert_publishes!("add_node", {
            g.add_node("P", &json!({})).unwrap();
        });
        assert_publishes!("add_node_multi", {
            g.add_node_multi(&["P", "Q"], &json!({})).unwrap();
        });
        assert_publishes!("add_edge", {
            g.add_edge(a, b, "T", &json!({})).unwrap();
        });
        assert_publishes!("update_node", {
            g.update_node(a, &json!({ "n": 9 })).unwrap();
        });
        assert_publishes!("update_edge", {
            g.update_edge(edge, &json!({ "weight": 2.0 })).unwrap();
        });
        assert_publishes!("add_label", {
            g.add_label(label_target, "R").unwrap();
        });
        assert_publishes!("remove_label", {
            g.remove_label(label_target, "R").unwrap();
        });
        assert_publishes!("delete_edge", {
            g.delete_edge(victim_edge).unwrap();
        });
        assert_publishes!("delete_node", {
            g.delete_node(victim_node).unwrap();
        });
        assert_publishes!("update", {
            g.update(|txn| {
                txn.add_node("P", &json!({}))?;
                Ok(())
            })
            .unwrap();
        });
    }

    /// A `Graph::update` closure that mutates nothing must not advance the
    /// generation, so a read-only use of the write transaction does not force
    /// every cache to rebuild.
    #[test]
    fn a_mutation_free_update_publishes_nothing() {
        let dir = TempDir::new().unwrap();
        let g = Graph::open(dir.path(), 1).unwrap();
        g.add_node("P", &json!({})).unwrap();
        g.rebuild_csr().unwrap();

        g.update(|txn| txn.get_node(1).map(|_| ())).unwrap();

        assert!(
            !g.csr_cache.snapshot_is_stale(),
            "a read-only update must leave the caches current"
        );
    }
}

// Persistence-dependent: these close a database and reopen the same path, or copy it
// to a file. The in-memory backend starts empty on every `open` by design (see
// `storage::memory`), so their premise does not hold there and the gate states that
// rather than letting them fail as though the backend were broken.
#[cfg(feature = "lmdb")]
#[cfg(test)]
mod lazy_open_tests {
    use serde_json::json;
    use tempfile::TempDir;

    use super::Graph;
    use crate::schema::NodeId;

    /// Populate a graph, force the CSR snapshot to build, then close it. Returns
    /// the directory so the caller can reopen the same path.
    fn seeded_dir() -> (TempDir, Vec<NodeId>) {
        let dir = TempDir::new().unwrap();
        let ids = {
            let g = Graph::open(dir.path(), 1).unwrap();
            // 80 nodes in a ring plus a chord, so a typed expansion over more
            // than `STALE_POINT_EXPAND_MAX` (64) sources takes the snapshot
            // path rather than the per-source LMDB path.
            let ids: Vec<_> = (0..80)
                .map(|i| g.add_node("Person", &json!({ "n": i })).unwrap())
                .collect();
            for i in 0..ids.len() {
                g.add_edge(ids[i], ids[(i + 1) % ids.len()], "FOLLOWS", &json!({}))
                    .unwrap();
            }
            g.add_edge(ids[0], ids[40], "LIKES", &json!({})).unwrap();
            // Touch an algorithm so this handle definitely built the snapshot.
            g.bfs(ids[0], 2).unwrap();
            assert!(
                !g.csr_cache.snapshot_is_stale(),
                "seed handle must build the snapshot"
            );
            ids
        };
        (dir, ids)
    }

    /// Opening an existing database does no CSR scan. That is the freshness gate's
    /// job, so a workload that only reads properties or point adjacency never pays
    /// for it.
    #[test]
    fn open_defers_the_csr_build() {
        let (dir, _ids) = seeded_dir();
        let g = Graph::open(dir.path(), 1).unwrap();

        assert_eq!(
            g.csr_cache.snapshot.load().dense_to_id.len(),
            0,
            "open must not build the CSR snapshot"
        );
        assert!(
            g.csr_cache.snapshot_is_stale(),
            "the unbuilt snapshot must report stale so a consumer rebuilds it"
        );
    }

    /// A freshly opened handle serves every consumer class correctly, each
    /// building what it needs through its own gate. This is the guard on the
    /// generation bookkeeping: if the unbuilt snapshot reported itself fresh,
    /// the typed-expansion path would read an empty CSR and silently return no
    /// rows instead of rebuilding.
    #[test]
    fn reopened_graph_serves_every_consumer_class() {
        let (dir, ids) = seeded_dir();

        // Each consumer gets its own handle, scoped so the LMDB environment is
        // closed before the next open, and so every gate is exercised from the
        // unbuilt state rather than riding on an earlier consumer's build.
        let reopen = || Graph::open(dir.path(), 1).unwrap();

        // Typed expansion over more sources than the stale-point-read cutoff,
        // so this goes through `ensure_snapshot_fresh`.
        {
            let g = reopen();
            let wide = g.expand_bulk(&ids, Some("FOLLOWS"), false).unwrap();
            assert_eq!(wide.len(), 80, "every ring edge must expand");
        }
        // Typed expansion under the cutoff, which reads LMDB point adjacency
        // directly and needs no snapshot at all.
        {
            let g = reopen();
            let narrow = g.expand_bulk(&ids[..4], Some("FOLLOWS"), false).unwrap();
            assert_eq!(narrow.len(), 4);
        }
        // Matrix-view consumer. Traversal is untyped, so one hop from `ids[0]`
        // reaches both the ring successor and the `LIKES` chord target.
        {
            let g = reopen();
            assert_eq!(
                g.bfs(ids[0], 1).unwrap().len(),
                3,
                "start plus both one-hop neighbors"
            );
        }
        // CSR-array consumer.
        {
            let g = reopen();
            assert_eq!(g.dfs(ids[0], 1).unwrap().len(), 3);
        }
        // Weighted matrix consumer.
        {
            let g = reopen();
            assert_eq!(g.page_rank(5, 0.85).unwrap().len(), 80);
        }
        {
            let g = reopen();
            let spec = crate::PathCountSpec {
                rel_types: vec![Some("FOLLOWS")],
                labels: vec![Some("Person"), Some("Person")],
                vertex_allow: Vec::new(),
            };
            assert_eq!(g.count_linear_paths(&spec).unwrap(), 80);
        }
        // Point adjacency, which never consults the snapshot.
        {
            let g = reopen();
            assert_eq!(g.out_neighbors(ids[0]).unwrap().len(), 2);
        }
    }

    /// The first gated consumer builds the snapshot, so the deferral is a delay
    /// rather than a permanent absence.
    #[test]
    fn first_algorithm_builds_what_open_skipped() {
        let (dir, ids) = seeded_dir();
        let g = Graph::open(dir.path(), 1).unwrap();
        assert!(g.csr_cache.snapshot_is_stale());

        assert_eq!(g.bfs(ids[0], 1).unwrap().len(), 3);

        assert!(
            !g.csr_cache.snapshot_is_stale(),
            "the snapshot gate must build on first use"
        );
    }

    /// Reopening an empty database is also lazy, and every consumer reports
    /// empty rather than erroring on the absent snapshot.
    #[test]
    fn empty_database_opens_lazily_and_reads_empty() {
        let dir = TempDir::new().unwrap();
        {
            Graph::open(dir.path(), 1).unwrap();
        }
        let g = Graph::open(dir.path(), 1).unwrap();
        assert!(g.all_nodes().unwrap().is_empty());
        assert!(g.connected_components().unwrap().is_empty());
        assert!(g.page_rank(3, 0.85).unwrap().is_empty());
    }
}