Rust

<EXTREMELY-IMPORTANT> This skill is the routing shell over the Rust reference set, not the whole systems handbook.

Non-negotiable rules: 1. Identify the subsystem before coding. Load only the relevant references. 2. Safe Rust by default. unsafe only when justified — every unsafe block gets a // SAFETY: comment. 3. Zero-copy where possible. mmap'd segments, &str/&[u8] references into pages. Never copy data without reason. 4. Error types per crate. thiserror enums in library crates. Never anyhow in libraries — only in binaries/tests. 5. Async with tokio. All I/O is async. CPU-bound work on spawn_blocking. Never block the tokio runtime. 6. Workspace crate structure. One crate per subsystem. Depend downward — never circular. 7. Property tests for invariants. proptest for round-trips, type coercion, binary parsing. </EXTREMELY-IMPORTANT>

rust

Inputs

• $request: The crate, subsystem, bug, feature, or review target

Goal

Route Rust work through the right subsystem conventions so changes follow the workspace’s systems-programming patterns instead of generic Rust defaults.

Step 0: Identify the subsystem

Decide which part of the Rust surface the task touches:

• storage engine
• binary formats
• type system
• Arrow or DataFusion
• wire protocols
• search or vectors
• arena or graph code
• geo indexing
• async or concurrency
• testing
• unsafe or error design

Success criteria: The task is mapped to the right subsystem before references are loaded.

Step 1: Load only the relevant references

Use the routing table to pick the matching files. Do not bulk-load the full reference tree.

Multiple tasks? Read multiple files.

Success criteria: Only the subsystem-relevant Rust guidance is active.

Step 2: Implement with the core Rust guardrails

Keep these rules active:

• safe Rust by default; unsafe only when justified with // SAFETY:
• error types match crate boundaries (thiserror in libs, anyhow only in bins/tests)
• async code does not block the runtime; CPU work on spawn_blocking
• owned vs borrowed data choices are deliberate; prefer &[u8] over Vec<u8> in signatures
• explicit use imports — no glob imports except in test modules
• #[must_use] on functions returning Result or computed values
• #[inline] only on small functions in hot loops — never on public API
• Send + Sync bounds on trait objects crossing async boundaries
• #[derive(Debug)] on all public types; #[derive(Clone)] only when cheap
• feature flags for optional crate deps — #[cfg(feature = "...")]
• tests match the risk: unit, integration, property, snapshot, or domain-specific verification

Success criteria: The implementation fits the workspace’s systems-level quality bar.

Step 3: Verify the change

Use the narrowest relevant verification loop:

• cargo fmt
• cargo clippy -- -D warnings
• focused crate tests
• subsystem-specific tests such as proptest or snapshots when appropriate

Success criteria: The Rust surface is validated the way this workspace expects.

Guardrails

• Do not inline the whole Rust handbook in SKILL.md.
• Do not skip subsystem identification.
• Do not use anyhow in library crates unless the project specifically allows it there.
• Do not add disable-model-invocation; this is a normal domain skill.
• Do not leave unsafe invariants undocumented.

When To Load References

• references/stack.md

Use for workspace/toolchain context.

• then only the task-relevant subsystem files under references/

Output Contract

Report:

1. which Rust references were loaded 2. the subsystem pattern applied 3. the change made 4. the verification run

Arena Allocation & Graph Engine

What

Arena allocation + index-based graph data structures give you a graph engine in safe Rust. Nodes and edges live in contiguous Vecs, referenced by typed indices (NodeIndex, EdgeIndex) instead of pointers or references. The borrow checker is satisfied because indices are Copy integers — no lifetimes, no Rc<RefCell<>>, no unsafe.

Why Arenas for Graphs

Key Dependencies

[dependencies]
bumpalo = "3"    # Bump allocator for batch allocation (optional — Vec-based arenas are often sufficient)

[dev-dependencies]
proptest = "1"   # Property-based testing for graph invariants

---

Arena Allocation with bumpalo

Basic Usage

use bumpalo::Bump;

// Create an arena — all allocations are contiguous in memory
let arena = Bump::new();

// Allocate individual values
let x: &mut i32 = arena.alloc(42);
let name: &str = arena.alloc_str("Alice");

// Allocate a slice from an iterator
let ids: &mut [u64] = arena.alloc_slice_copy(&[1, 2, 3, 4, 5]);

// Allocate with a closure (useful when construction needs arena references)
let node: &mut GraphNode = arena.alloc_with(|| GraphNode {
    id: 1,
    label: arena.alloc_str("person"),
    neighbors: bumpalo::vec![in &arena; 2, 3, 5],
});

// Everything freed when `arena` is dropped — single deallocation

Typed Arena Wrapper

When all allocations are the same type, wrap Bump for type safety:

use bumpalo::Bump;
use std::marker::PhantomData;

pub struct TypedArena<T> {
    bump: Bump,
    count: usize,
    _marker: PhantomData<T>,
}

impl<T> TypedArena<T> {
    pub fn new() -> Self {
        Self {
            bump: Bump::new(),
            count: 0,
            _marker: PhantomData,
        }
    }

    /// Allocate a value, return its index.
    pub fn alloc(&mut self, value: T) -> usize {
        let _ = self.bump.alloc(value);
        let idx = self.count;
        self.count += 1;
        idx
    }

    /// Total bytes allocated (useful for memory budgeting).
    pub fn allocated_bytes(&self) -> usize {
        self.bump.allocated_bytes()
    }
}

When to Use bumpalo vs Plain Vec

For the graph storage engine below, we use `Vec`-based arenas. bumpalo is used for transient query-time allocations (pattern matching intermediate results, traversal buffers).

---

Graph Data Structures

Typed Indices (Safe Rust, Zero-Cost)

Instead of pointers, use newtype wrappers around integers. This is the core pattern that makes the entire graph engine safe:

/// Index into GraphStorage::nodes
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct NodeIndex(pub u32);

/// Index into GraphStorage::edges
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct EdgeIndex(pub u32);

/// Compact label identifier — maps to/from String via LabelRegistry
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct LabelId(pub u16);

impl NodeIndex {
    #[inline]
    pub fn as_usize(self) -> usize {
        self.0 as usize
    }
}

impl EdgeIndex {
    #[inline]
    pub fn as_usize(self) -> usize {
        self.0 as usize
    }
}

// Display for debugging: Node(42), Edge(7)
impl std::fmt::Display for NodeIndex {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Node({})", self.0)
    }
}

impl std::fmt::Display for EdgeIndex {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Edge({})", self.0)
    }
}

Node and Edge Data

/// Per-node data — kept small for cache efficiency.
/// Properties live in a separate document store (BMAP), referenced by offset.
#[derive(Debug, Clone)]
pub struct NodeData {
    /// Unique node identifier (external-facing, stable across compaction)
    pub id: u64,
    /// Label — Person, Product, etc. Resolved via LabelRegistry.
    pub label_id: LabelId,
    /// Byte offset into the document/property store (BMAP).
    /// Properties are stored externally to keep NodeData small and cache-hot.
    pub properties_offset: u64,
}

/// Per-edge data — directional (from → to).
#[derive(Debug, Clone)]
pub struct EdgeData {
    /// Unique edge identifier
    pub id: u64,
    /// Label — KNOWS, PURCHASED, etc.
    pub label_id: LabelId,
    /// Source node
    pub from: NodeIndex,
    /// Target node
    pub to: NodeIndex,
    /// Byte offset into the document/property store (BMAP).
    pub properties_offset: u64,
}

Label Registry

Bidirectional mapping between human-readable label strings and compact LabelId values:

use std::collections::HashMap;

/// Bidirectional String ↔ LabelId mapping.
/// Labels are interned — each unique string is stored once.
#[derive(Debug, Default)]
pub struct LabelRegistry {
    to_id: HashMap<String, LabelId>,
    to_name: Vec<String>, // indexed by LabelId.0
}

impl LabelRegistry {
    pub fn new() -> Self {
        Self::default()
    }

    /// Get or create a LabelId for the given name.
    pub fn get_or_insert(&mut self, name: &str) -> LabelId {
        if let Some(&id) = self.to_id.get(name) {
            return id;
        }
        let id = LabelId(self.to_name.len() as u16);
        self.to_name.push(name.to_owned());
        self.to_id.insert(name.to_owned(), id);
        id
    }

    /// Resolve a LabelId back to its string name.
    #[must_use]
    pub fn resolve(&self, id: LabelId) -> Option<&str> {
        self.to_name.get(id.0 as usize).map(|s| s.as_str())
    }

    /// Look up by name without inserting.
    #[must_use]
    pub fn lookup(&self, name: &str) -> Option<LabelId> {
        self.to_id.get(name).copied()
    }
}

GraphStorage — The Core

/// Arena-based graph storage with bidirectional adjacency lists.
///
/// All nodes and edges are stored in contiguous Vec buffers.
/// Adjacency is maintained as sorted Vec<EdgeIndex> per node,
/// for both outgoing and incoming directions.
#[derive(Debug)]
pub struct GraphStorage {
    /// All nodes, indexed by NodeIndex
    nodes: Vec<NodeData>,
    /// All edges, indexed by EdgeIndex
    edges: Vec<EdgeData>,
    /// node → outgoing edges (sorted by EdgeIndex for binary search)
    outgoing: Vec<Vec<EdgeIndex>>,
    /// node → incoming edges (sorted by EdgeIndex for binary search)
    incoming: Vec<Vec<EdgeIndex>>,
    /// External ID → NodeIndex lookup
    node_id_map: HashMap<u64, NodeIndex>,
    /// Label interning
    labels: LabelRegistry,
}

impl GraphStorage {
    pub fn new() -> Self {
        Self {
            nodes: Vec::new(),
            edges: Vec::new(),
            outgoing: Vec::new(),
            incoming: Vec::new(),
            node_id_map: HashMap::new(),
            labels: LabelRegistry::new(),
        }
    }

    /// Pre-allocate capacity for known graph size.
    pub fn with_capacity(node_capacity: usize, edge_capacity: usize) -> Self {
        Self {
            nodes: Vec::with_capacity(node_capacity),
            edges: Vec::with_capacity(edge_capacity),
            outgoing: Vec::with_capacity(node_capacity),
            incoming: Vec::with_capacity(node_capacity),
            node_id_map: HashMap::with_capacity(node_capacity),
            labels: LabelRegistry::new(),
        }
    }

    // --- Node operations ---

    /// Add a node. Returns its NodeIndex.
    pub fn add_node(&mut self, id: u64, label: &str, properties_offset: u64) -> NodeIndex {
        let label_id = self.labels.get_or_insert(label);
        let idx = NodeIndex(self.nodes.len() as u32);
        self.nodes.push(NodeData {
            id,
            label_id,
            properties_offset,
        });
        self.outgoing.push(Vec::new());
        self.incoming.push(Vec::new());
        self.node_id_map.insert(id, idx);
        idx
    }

    /// Look up a node by its external ID.
    #[must_use]
    pub fn node_by_id(&self, id: u64) -> Option<NodeIndex> {
        self.node_id_map.get(&id).copied()
    }

    /// Get node data by index. Panics if out of bounds (debug assertion).
    #[must_use]
    pub fn node(&self, idx: NodeIndex) -> &NodeData {
        debug_assert!(idx.as_usize() < self.nodes.len(), "NodeIndex out of bounds: {idx}");
        &self.nodes[idx.as_usize()]
    }

    /// Total node count.
    #[must_use]
    pub fn node_count(&self) -> usize {
        self.nodes.len()
    }

    /// Iterate all node indices.
    pub fn node_indices(&self) -> impl Iterator<Item = NodeIndex> {
        (0..self.nodes.len() as u32).map(NodeIndex)
    }

    /// Iterate all nodes with label filter.
    pub fn nodes_with_label(&self, label_id: LabelId) -> impl Iterator<Item = NodeIndex> + '_ {
        self.nodes
            .iter()
            .enumerate()
            .filter(move |(_, n)| n.label_id == label_id)
            .map(|(i, _)| NodeIndex(i as u32))
    }

    // --- Edge operations ---

    /// Add a directed edge. Returns its EdgeIndex.
    /// Maintains bidirectional adjacency (outgoing from `from`, incoming to `to`).
    pub fn add_edge(
        &mut self,
        id: u64,
        label: &str,
        from: NodeIndex,
        to: NodeIndex,
        properties_offset: u64,
    ) -> EdgeIndex {
        debug_assert!(from.as_usize() < self.nodes.len(), "from NodeIndex out of bounds: {from}");
        debug_assert!(to.as_usize() < self.nodes.len(), "to NodeIndex out of bounds: {to}");

        let label_id = self.labels.get_or_insert(label);
        let idx = EdgeIndex(self.edges.len() as u32);
        self.edges.push(EdgeData {
            id,
            label_id,
            from,
            to,
            properties_offset,
        });
        self.outgoing[from.as_usize()].push(idx);
        self.incoming[to.as_usize()].push(idx);
        idx
    }

    /// Get edge data by index.
    #[must_use]
    pub fn edge(&self, idx: EdgeIndex) -> &EdgeData {
        debug_assert!(idx.as_usize() < self.edges.len(), "EdgeIndex out of bounds: {idx}");
        &self.edges[idx.as_usize()]
    }

    /// Total edge count.
    #[must_use]
    pub fn edge_count(&self) -> usize {
        self.edges.len()
    }

    // --- Adjacency queries ---

    /// Outgoing edges from a node.
    #[must_use]
    pub fn outgoing_edges(&self, node: NodeIndex) -> &[EdgeIndex] {
        &self.outgoing[node.as_usize()]
    }

    /// Incoming edges to a node.
    #[must_use]
    pub fn incoming_edges(&self, node: NodeIndex) -> &[EdgeIndex] {
        &self.incoming[node.as_usize()]
    }

    /// Outgoing neighbors (target nodes of outgoing edges).
    pub fn outgoing_neighbors(&self, node: NodeIndex) -> impl Iterator<Item = NodeIndex> + '_ {
        self.outgoing[node.as_usize()]
            .iter()
            .map(|&eidx| self.edges[eidx.as_usize()].to)
    }

    /// Incoming neighbors (source nodes of incoming edges).
    pub fn incoming_neighbors(&self, node: NodeIndex) -> impl Iterator<Item = NodeIndex> + '_ {
        self.incoming[node.as_usize()]
            .iter()
            .map(|&eidx| self.edges[eidx.as_usize()].from)
    }

    /// All neighbors (both directions), deduplicated.
    pub fn all_neighbors(&self, node: NodeIndex) -> Vec<NodeIndex> {
        let mut neighbors: Vec<NodeIndex> = self
            .outgoing_neighbors(node)
            .chain(self.incoming_neighbors(node))
            .collect();
        neighbors.sort_unstable();
        neighbors.dedup();
        neighbors
    }

    /// Outgoing edges filtered by label.
    pub fn outgoing_edges_with_label(
        &self,
        node: NodeIndex,
        label_id: LabelId,
    ) -> impl Iterator<Item = EdgeIndex> + '_ {
        self.outgoing[node.as_usize()]
            .iter()
            .copied()
            .filter(move |&eidx| self.edges[eidx.as_usize()].label_id == label_id)
    }

    /// Access the label registry.
    #[must_use]
    pub fn labels(&self) -> &LabelRegistry {
        &self.labels
    }

    /// Mutable access to the label registry.
    pub fn labels_mut(&mut self) -> &mut LabelRegistry {
        &mut self.labels
    }

    // --- Deletion ---

    /// Remove an edge by marking it as deleted (tombstone).
    /// Does NOT compact — adjacency lists retain the EdgeIndex but it is skipped in iteration.
    /// Call `compact()` periodically to reclaim space.
    pub fn remove_edge(&mut self, idx: EdgeIndex) {
        let edge = &self.edges[idx.as_usize()];
        let from = edge.from;
        let to = edge.to;
        self.outgoing[from.as_usize()].retain(|&e| e != idx);
        self.incoming[to.as_usize()].retain(|&e| e != idx);
        // Tombstone: set from == to == NodeIndex(u32::MAX)
        let edge = &mut self.edges[idx.as_usize()];
        edge.from = NodeIndex(u32::MAX);
        edge.to = NodeIndex(u32::MAX);
    }

    /// Check if an edge is a tombstone (deleted).
    #[must_use]
    pub fn is_edge_deleted(&self, idx: EdgeIndex) -> bool {
        self.edges[idx.as_usize()].from == NodeIndex(u32::MAX)
    }
}

AS NODE / AS EDGE Table Annotations

When SQL tables are annotated as graph elements, the engine auto-creates adjacency:

/// Declares how a SQL table maps to the graph overlay.
#[derive(Debug, Clone)]
pub struct GraphAnnotation {
    pub table_name: String,
    pub annotation_type: GraphAnnotationType,
}

#[derive(Debug, Clone)]
pub enum GraphAnnotationType {
    /// `CREATE TABLE people (...) AS NODE`
    /// Each row becomes a graph node. Row PK = node external ID.
    Node,
    /// `CREATE TABLE knows (...) AS EDGE FROM people TO people`
    /// Each row becomes an edge. Auto-adds `from_id` and `to_id` columns.
    Edge {
        from_table: String,
        to_table: String,
    },
}

/// Tracks all graph-annotated tables and syncs them with GraphStorage.
#[derive(Debug)]
pub struct GraphCatalog {
    annotations: Vec<GraphAnnotation>,
    /// Maps table_name → LabelId for quick lookup during WAL apply.
    table_to_label: HashMap<String, LabelId>,
}

impl GraphCatalog {
    pub fn new() -> Self {
        Self {
            annotations: Vec::new(),
            table_to_label: HashMap::new(),
        }
    }

    /// Register a table as a graph node source.
    /// After this call, INSERT into this table also creates a graph node.
    pub fn register_node_table(
        &mut self,
        table_name: &str,
        graph: &mut GraphStorage,
    ) {
        let label_id = graph.labels_mut().get_or_insert(table_name);
        self.annotations.push(GraphAnnotation {
            table_name: table_name.to_owned(),
            annotation_type: GraphAnnotationType::Node,
        });
        self.table_to_label.insert(table_name.to_owned(), label_id);
    }

    /// Register a table as a graph edge source.
    /// The table must have `from_id` and `to_id` columns referencing node tables.
    pub fn register_edge_table(
        &mut self,
        table_name: &str,
        from_table: &str,
        to_table: &str,
        graph: &mut GraphStorage,
    ) {
        let label_id = graph.labels_mut().get_or_insert(table_name);
        self.annotations.push(GraphAnnotation {
            table_name: table_name.to_owned(),
            annotation_type: GraphAnnotationType::Edge {
                from_table: from_table.to_owned(),
                to_table: to_table.to_owned(),
            },
        });
        self.table_to_label.insert(table_name.to_owned(), label_id);
    }
}

---

Traversal Algorithms

All traversal functions take &GraphStorage (immutable borrow) — they never mutate the graph.

BFS (Breadth-First Search)

use std::collections::VecDeque;

/// Breadth-first traversal from `start`, up to `max_depth` hops.
/// Returns visited nodes with their depth from start.
/// Optionally filters by edge label.
#[must_use]
pub fn bfs(
    graph: &GraphStorage,
    start: NodeIndex,
    max_depth: usize,
    edge_label_filter: Option<LabelId>,
) -> Vec<(NodeIndex, usize)> {
    let node_count = graph.node_count();
    let mut visited = vec![false; node_count];
    let mut result = Vec::new();
    let mut queue = VecDeque::new();

    visited[start.as_usize()] = true;
    queue.push_back((start, 0usize));
    result.push((start, 0));

    while let Some((current, depth)) = queue.pop_front() {
        if depth >= max_depth {
            continue;
        }

        let edges = graph.outgoing_edges(current);
        for &edge_idx in edges {
            let edge = graph.edge(edge_idx);

            // Skip if label filter is set and doesn't match
            if let Some(filter_label) = edge_label_filter {
                if edge.label_id != filter_label {
                    continue;
                }
            }

            let neighbor = edge.to;
            if !visited[neighbor.as_usize()] {
                visited[neighbor.as_usize()] = true;
                let next_depth = depth + 1;
                result.push((neighbor, next_depth));
                queue.push_back((neighbor, next_depth));
            }
        }
    }

    result
}

DFS (Depth-First Search) — Iterative

Iterative with explicit stack to avoid stack overflow on deep graphs:

/// Iterative depth-first traversal. Returns nodes in DFS visit order.
/// Uses an explicit stack — safe for graphs with millions of nodes.
#[must_use]
pub fn dfs(
    graph: &GraphStorage,
    start: NodeIndex,
    max_depth: usize,
    edge_label_filter: Option<LabelId>,
) -> Vec<(NodeIndex, usize)> {
    let node_count = graph.node_count();
    let mut visited = vec![false; node_count];
    let mut result = Vec::new();
    let mut stack = Vec::new();

    stack.push((start, 0usize));

    while let Some((current, depth)) = stack.pop() {
        if visited[current.as_usize()] {
            continue;
        }
        visited[current.as_usize()] = true;
        result.push((current, depth));

        if depth >= max_depth {
            continue;
        }

        // Push neighbors in reverse order so that the first neighbor is visited first
        let edges = graph.outgoing_edges(current);
        for &edge_idx in edges.iter().rev() {
            let edge = graph.edge(edge_idx);

            if let Some(filter_label) = edge_label_filter {
                if edge.label_id != filter_label {
                    continue;
                }
            }

            let neighbor = edge.to;
            if !visited[neighbor.as_usize()] {
                stack.push((neighbor, depth + 1));
            }
        }
    }

    result
}

Shortest Path (Dijkstra)

use std::cmp::Ordering;
use std::collections::BinaryHeap;

/// State for Dijkstra's priority queue.
/// Implements Ord to give us a min-heap (BinaryHeap is max-heap by default).
#[derive(Debug)]
struct DijkstraState {
    cost: f64,
    node: NodeIndex,
}

impl PartialEq for DijkstraState {
    fn eq(&self, other: &Self) -> bool {
        self.cost.to_bits() == other.cost.to_bits() && self.node == other.node
    }
}
impl Eq for DijkstraState {}

impl PartialOrd for DijkstraState {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for DijkstraState {
    fn cmp(&self, other: &Self) -> Ordering {
        // Reverse ordering for min-heap behavior
        other
            .cost
            .partial_cmp(&self.cost)
            .unwrap_or(Ordering::Equal)
            .then_with(|| self.node.0.cmp(&other.node.0))
    }
}

/// Dijkstra's shortest path from `from` to `to`.
/// `weight_fn` extracts the edge weight (must be non-negative).
/// Returns (total_cost, path) or None if unreachable.
#[must_use]
pub fn shortest_path(
    graph: &GraphStorage,
    from: NodeIndex,
    to: NodeIndex,
    weight_fn: impl Fn(&EdgeData) -> f64,
) -> Option<(f64, Vec<NodeIndex>)> {
    let node_count = graph.node_count();
    let mut dist = vec![f64::INFINITY; node_count];
    let mut prev: Vec<Option<NodeIndex>> = vec![None; node_count];
    let mut heap = BinaryHeap::new();

    dist[from.as_usize()] = 0.0;
    heap.push(DijkstraState {
        cost: 0.0,
        node: from,
    });

    while let Some(DijkstraState { cost, node }) = heap.pop() {
        // Found the target — reconstruct path
        if node == to {
            let mut path = Vec::new();
            let mut current = Some(to);
            while let Some(n) = current {
                path.push(n);
                current = prev[n.as_usize()];
            }
            path.reverse();
            return Some((cost, path));
        }

        // Skip if we already found a shorter path to this node
        if cost > dist[node.as_usize()] {
            continue;
        }

        for &edge_idx in graph.outgoing_edges(node) {
            let edge = graph.edge(edge_idx);
            let weight = weight_fn(edge);
            debug_assert!(weight >= 0.0, "Dijkstra requires non-negative weights");

            let next_cost = cost + weight;
            let neighbor = edge.to;

            if next_cost < dist[neighbor.as_usize()] {
                dist[neighbor.as_usize()] = next_cost;
                prev[neighbor.as_usize()] = Some(node);
                heap.push(DijkstraState {
                    cost: next_cost,
                    node: neighbor,
                });
            }
        }
    }

    None // Target unreachable
}

PageRank

/// Iterative PageRank computation.
///
/// - `damping`: probability of following a link (typically 0.85)
/// - `tolerance`: convergence threshold (typically 1e-6)
/// - `max_iterations`: safety cap to prevent infinite loops
///
/// Returns a Vec of scores indexed by NodeIndex.
#[must_use]
pub fn pagerank(
    graph: &GraphStorage,
    damping: f64,
    tolerance: f64,
    max_iterations: usize,
) -> Vec<f64> {
    let n = graph.node_count();
    if n == 0 {
        return Vec::new();
    }

    let initial_score = 1.0 / n as f64;
    let mut scores = vec![initial_score; n];
    let mut new_scores = vec![0.0f64; n];

    // Precompute out-degree for each node
    let out_degree: Vec<usize> = (0..n)
        .map(|i| graph.outgoing_edges(NodeIndex(i as u32)).len())
        .collect();

    for _iteration in 0..max_iterations {
        // Reset new scores to the random-jump baseline
        new_scores.fill((1.0 - damping) / n as f64);

        // Distribute scores along edges
        for i in 0..n {
            let deg = out_degree[i];
            if deg == 0 {
                // Dangling node — distribute its score evenly to all nodes
                let share = damping * scores[i] / n as f64;
                for s in new_scores.iter_mut() {
                    *s += share;
                }
            } else {
                let share = damping * scores[i] / deg as f64;
                for &edge_idx in graph.outgoing_edges(NodeIndex(i as u32)) {
                    let target = graph.edge(edge_idx).to.as_usize();
                    new_scores[target] += share;
                }
            }
        }

        // Check convergence (L1 norm of difference)
        let diff: f64 = scores
            .iter()
            .zip(new_scores.iter())
            .map(|(old, new)| (old - new).abs())
            .sum();

        std::mem::swap(&mut scores, &mut new_scores);

        if diff < tolerance {
            break;
        }
    }

    scores
}

Connected Components (Union-Find)

/// Disjoint-set (Union-Find) with path compression and union by rank.
/// Used for connected component detection.
pub struct UnionFind {
    parent: Vec<u32>,
    rank: Vec<u8>,
}

impl UnionFind {
    pub fn new(size: usize) -> Self {
        Self {
            parent: (0..size as u32).collect(),
            rank: vec![0; size],
        }
    }

    /// Find the root of the set containing `x`, with path compression.
    pub fn find(&mut self, x: u32) -> u32 {
        if self.parent[x as usize] != x {
            self.parent[x as usize] = self.find(self.parent[x as usize]);
        }
        self.parent[x as usize]
    }

    /// Union the sets containing `x` and `y`. Returns true if they were separate.
    pub fn union(&mut self, x: u32, y: u32) -> bool {
        let rx = self.find(x);
        let ry = self.find(y);
        if rx == ry {
            return false;
        }
        // Union by rank — attach shorter tree under taller tree
        match self.rank[rx as usize].cmp(&self.rank[ry as usize]) {
            std::cmp::Ordering::Less => self.parent[rx as usize] = ry,
            std::cmp::Ordering::Greater => self.parent[ry as usize] = rx,
            std::cmp::Ordering::Equal => {
                self.parent[ry as usize] = rx;
                self.rank[rx as usize] += 1;
            }
        }
        true
    }
}

/// Find all connected components (treating edges as undirected).
/// Returns a map from component root → list of node indices.
#[must_use]
pub fn connected_components(graph: &GraphStorage) -> HashMap<u32, Vec<NodeIndex>> {
    let n = graph.node_count();
    let mut uf = UnionFind::new(n);

    // Union all edges (both directions — treating as undirected)
    for i in 0..n {
        let node = NodeIndex(i as u32);
        for &edge_idx in graph.outgoing_edges(node) {
            let edge = graph.edge(edge_idx);
            uf.union(node.0, edge.to.0);
        }
    }

    // Group nodes by their component root
    let mut components: HashMap<u32, Vec<NodeIndex>> = HashMap::new();
    for i in 0..n as u32 {
        let root = uf.find(i);
        components.entry(root).or_default().push(NodeIndex(i));
    }

    components
}

Variable-Length Path Traversal

Used by graph pattern matching — traverse 1..N hops following a label:

/// Traverse variable-length paths: from `start`, follow edges with `label_id`
/// between `min_hops` and `max_hops` times.
/// Returns all reachable (node, hop_count) pairs.
#[must_use]
pub fn variable_length_traverse(
    graph: &GraphStorage,
    start: NodeIndex,
    label_id: LabelId,
    min_hops: usize,
    max_hops: usize,
) -> Vec<(NodeIndex, usize)> {
    let node_count = graph.node_count();
    let mut results = Vec::new();
    // (current_node, depth, visited_set_as_bitmask_or_hashset)
    let mut stack: Vec<(NodeIndex, usize, Vec<bool>)> = Vec::new();

    let mut initial_visited = vec![false; node_count];
    initial_visited[start.as_usize()] = true;
    stack.push((start, 0, initial_visited));

    while let Some((current, depth, visited)) = stack.pop() {
        // Collect if within hop range
        if depth >= min_hops {
            results.push((current, depth));
        }

        if depth >= max_hops {
            continue;
        }

        // Follow matching edges
        for &edge_idx in graph.outgoing_edges(current) {
            let edge = graph.edge(edge_idx);
            if edge.label_id != label_id {
                continue;
            }
            let neighbor = edge.to;
            if !visited[neighbor.as_usize()] {
                let mut next_visited = visited.clone();
                next_visited[neighbor.as_usize()] = true;
                stack.push((neighbor, depth + 1, next_visited));
            }
        }
    }

    results
}

---

Cypher-Lite Pattern Matching

Pattern AST

/// Direction of an edge in a pattern.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EdgeDirection {
    /// `-[r:LABEL]->` — left to right
    Outgoing,
    /// `<-[r:LABEL]-` — right to left
    Incoming,
    /// `-[r:LABEL]-` — either direction
    Both,
}

/// A single element in a graph pattern.
#[derive(Debug, Clone)]
pub enum PatternElement {
    Node {
        variable: String,
        label: Option<String>,
    },
    Edge {
        variable: String,
        label: Option<String>,
        direction: EdgeDirection,
        /// Variable-length: (min_hops, max_hops). None = exactly 1 hop.
        hops: Option<(usize, usize)>,
    },
}

/// A complete pattern: alternating Node, Edge, Node, Edge, ..., Node.
#[derive(Debug, Clone)]
pub struct GraphPattern {
    pub elements: Vec<PatternElement>,
}

/// A single row of pattern match results — variable name → NodeIndex or EdgeIndex.
#[derive(Debug, Clone)]
pub struct MatchBinding {
    pub nodes: HashMap<String, NodeIndex>,
    pub edges: HashMap<String, EdgeIndex>,
}

Compilation to Traversal Plan

A pattern compiles into a sequence of traversal steps:

/// A step in the traversal plan.
#[derive(Debug)]
pub enum TraversalStep {
    /// Start from nodes matching a label (and optional WHERE predicate).
    ScanNodes {
        variable: String,
        label_id: Option<LabelId>,
    },
    /// Follow edges from the previous node in the pattern.
    FollowEdges {
        edge_variable: String,
        edge_label_id: Option<LabelId>,
        direction: EdgeDirection,
        target_variable: String,
        target_label_id: Option<LabelId>,
        hops: Option<(usize, usize)>,
    },
}

/// Compile a GraphPattern into a TraversalPlan.
pub fn compile_pattern(
    pattern: &GraphPattern,
    labels: &LabelRegistry,
) -> Vec<TraversalStep> {
    let mut steps = Vec::new();
    let mut i = 0;

    while i < pattern.elements.len() {
        match &pattern.elements[i] {
            PatternElement::Node { variable, label } if i == 0 => {
                // First node — this is a scan
                let label_id = label
                    .as_ref()
                    .and_then(|l| labels.lookup(l));
                steps.push(TraversalStep::ScanNodes {
                    variable: variable.clone(),
                    label_id,
                });
                i += 1;
            }
            PatternElement::Edge {
                variable: edge_var,
                label: edge_label,
                direction,
                hops,
            } => {
                // Edge must be followed by a target node
                let edge_label_id = edge_label
                    .as_ref()
                    .and_then(|l| labels.lookup(l));

                let (target_var, target_label_id) =
                    if let Some(PatternElement::Node { variable, label }) =
                        pattern.elements.get(i + 1)
                    {
                        let lid = label.as_ref().and_then(|l| labels.lookup(l));
                        (variable.clone(), lid)
                    } else {
                        panic!("Edge must be followed by a Node in pattern");
                    };

                steps.push(TraversalStep::FollowEdges {
                    edge_variable: edge_var.clone(),
                    edge_label_id,
                    direction: direction.clone(),
                    target_variable: target_var,
                    target_label_id,
                    hops: *hops,
                });
                i += 2; // skip edge + target node
            }
            _ => {
                i += 1;
            }
        }
    }

    steps
}

Pattern Execution Engine

/// Execute a compiled traversal plan against the graph.
/// Returns all bindings (variable → index mappings) that satisfy the pattern.
pub fn execute_pattern(
    graph: &GraphStorage,
    steps: &[TraversalStep],
) -> Vec<MatchBinding> {
    let mut bindings: Vec<MatchBinding> = Vec::new();

    for step in steps {
        match step {
            TraversalStep::ScanNodes { variable, label_id } => {
                // Initial scan — create a binding for each matching node
                let iter: Box<dyn Iterator<Item = NodeIndex>> = match label_id {
                    Some(lid) => Box::new(graph.nodes_with_label(*lid)),
                    None => Box::new(graph.node_indices()),
                };

                for node_idx in iter {
                    let mut binding = MatchBinding {
                        nodes: HashMap::new(),
                        edges: HashMap::new(),
                    };
                    binding.nodes.insert(variable.clone(), node_idx);
                    bindings.push(binding);
                }
            }

            TraversalStep::FollowEdges {
                edge_variable,
                edge_label_id,
                direction,
                target_variable,
                target_label_id,
                hops: None, // single hop
            } => {
                let mut new_bindings = Vec::new();

                for binding in &bindings {
                    // Find the source node from the previous step
                    // It's the last node variable added
                    let source_idx = binding
                        .nodes
                        .values()
                        .last()
                        .copied()
                        .expect("No source node in binding");

                    let edge_list = match direction {
                        EdgeDirection::Outgoing => graph.outgoing_edges(source_idx),
                        EdgeDirection::Incoming => graph.incoming_edges(source_idx),
                        EdgeDirection::Both => {
                            // For Both, we handle outgoing here, incoming below
                            graph.outgoing_edges(source_idx)
                        }
                    };

                    for &eidx in edge_list {
                        let edge = graph.edge(eidx);

                        // Label filter on edge
                        if let Some(filter_lid) = edge_label_id {
                            if edge.label_id != *filter_lid {
                                continue;
                            }
                        }

                        let target = match direction {
                            EdgeDirection::Outgoing | EdgeDirection::Both => edge.to,
                            EdgeDirection::Incoming => edge.from,
                        };

                        // Label filter on target node
                        if let Some(filter_lid) = target_label_id {
                            if graph.node(target).label_id != *filter_lid {
                                continue;
                            }
                        }

                        let mut new_binding = binding.clone();
                        new_binding.edges.insert(edge_variable.clone(), eidx);
                        new_binding.nodes.insert(target_variable.clone(), target);
                        new_bindings.push(new_binding);
                    }

                    // Handle incoming edges for Both direction
                    if *direction == EdgeDirection::Both {
                        for &eidx in graph.incoming_edges(source_idx) {
                            let edge = graph.edge(eidx);
                            if let Some(filter_lid) = edge_label_id {
                                if edge.label_id != *filter_lid {
                                    continue;
                                }
                            }
                            let target = edge.from;
                            if let Some(filter_lid) = target_label_id {
                                if graph.node(target).label_id != *filter_lid {
                                    continue;
                                }
                            }
                            let mut new_binding = binding.clone();
                            new_binding.edges.insert(edge_variable.clone(), eidx);
                            new_binding.nodes.insert(target_variable.clone(), target);
                            new_bindings.push(new_binding);
                        }
                    }
                }

                bindings = new_bindings;
            }

            TraversalStep::FollowEdges { hops: Some(_), .. } => {
                // Variable-length path matching — delegate to variable_length_traverse
                // and expand bindings accordingly.
                // Implementation left to the specific query engine.
                todo!("Variable-length path execution");
            }
        }
    }

    bindings
}

SQL Interop

The same data is queryable via both graph syntax and standard SQL JOINs:

-- Graph syntax (Cypher-lite)
MATCH (a:people)-[k:knows]->(b:people)
WHERE a.name = 'Alice'
RETURN b.name;

-- Equivalent SQL (auto-generated from AS NODE / AS EDGE declarations)
SELECT b.name
FROM people a
JOIN knows k ON k.from_id = a.id
JOIN people b ON k.to_id = b.id
WHERE a.name = 'Alice';

The graph engine does NOT replace SQL — it provides an alternative syntax that compiles to the same execution plan. When a table is declared AS EDGE FROM people TO people, the system:

1. Auto-adds from_id BIGINT NOT NULL and to_id BIGINT NOT NULL columns 2. Creates foreign key constraints to the referenced node tables 3. Builds the adjacency list index (GraphStorage.outgoing / GraphStorage.incoming) 4. Maintains adjacency on every INSERT/UPDATE/DELETE via WAL hooks

---

WAL Integration

Graph index updates are derived from WAL (Write-Ahead Log) entries. The graph index is a secondary index — the source of truth is the relational tables.

/// WAL entry types that affect the graph index.
#[derive(Debug)]
pub enum WalGraphOp {
    /// Row inserted into a node table
    InsertNode {
        table: String,
        row_id: u64,
        properties_offset: u64,
    },
    /// Row inserted into an edge table
    InsertEdge {
        table: String,
        edge_id: u64,
        from_id: u64,
        to_id: u64,
        properties_offset: u64,
    },
    /// Row deleted from a node table
    DeleteNode {
        table: String,
        row_id: u64,
    },
    /// Row deleted from an edge table
    DeleteEdge {
        table: String,
        edge_id: u64,
    },
}

/// Apply a WAL entry to the graph index.
/// Called synchronously during WAL apply — must be fast.
pub fn apply_wal_op(
    graph: &mut GraphStorage,
    catalog: &GraphCatalog,
    op: &WalGraphOp,
) -> Result<(), GraphError> {
    match op {
        WalGraphOp::InsertNode {
            table,
            row_id,
            properties_offset,
        } => {
            graph.add_node(*row_id, table, *properties_offset);
            Ok(())
        }

        WalGraphOp::InsertEdge {
            table,
            edge_id,
            from_id,
            to_id,
            properties_offset,
        } => {
            let from_idx = graph
                .node_by_id(*from_id)
                .ok_or(GraphError::NodeNotFound(*from_id))?;
            let to_idx = graph
                .node_by_id(*to_id)
                .ok_or(GraphError::NodeNotFound(*to_id))?;
            graph.add_edge(*edge_id, table, from_idx, to_idx, *properties_offset);
            Ok(())
        }

        WalGraphOp::DeleteNode { row_id, .. } => {
            // Remove all edges connected to this node first
            if let Some(node_idx) = graph.node_by_id(*row_id) {
                let outgoing: Vec<EdgeIndex> =
                    graph.outgoing_edges(node_idx).to_vec();
                let incoming: Vec<EdgeIndex> =
                    graph.incoming_edges(node_idx).to_vec();
                for eidx in outgoing.into_iter().chain(incoming) {
                    graph.remove_edge(eidx);
                }
                // Mark node as deleted (tombstone)
                // Full removal happens during compaction
            }
            Ok(())
        }

        WalGraphOp::DeleteEdge { edge_id, .. } => {
            // Linear scan for edge by ID — in production, maintain an edge_id_map
            // similar to node_id_map for O(1) lookup.
            // Omitted here for clarity; the pattern is identical to node_id_map.
            Ok(())
        }
    }
}

/// Error types for graph operations.
#[derive(Debug, thiserror::Error)]
pub enum GraphError {
    #[error("Node not found: {0}")]
    NodeNotFound(u64),

    #[error("Edge not found: {0}")]
    EdgeNotFound(u64),

    #[error("Label not found: {0}")]
    LabelNotFound(String),

    #[error("Invalid pattern: {0}")]
    InvalidPattern(String),
}

WAL Sync Guarantees

---

Testing

Property-Based Tests with proptest

#[cfg(test)]
mod tests {
    use super::*;
    use proptest::prelude::*;

    /// Strategy: generate a random graph, then verify structural invariants.
    fn arb_graph() -> impl Strategy<Value = GraphStorage> {
        (1usize..200, 0usize..500).prop_flat_map(|(num_nodes, num_edges)| {
            let node_labels = prop::collection::vec(
                prop::sample::select(vec!["person", "product", "order"]),
                num_nodes,
            );
            let edges = prop::collection::vec(
                (0..num_nodes as u32, 0..num_nodes as u32),
                num_edges.min(num_nodes * 3), // cap edges relative to nodes
            );
            (Just(num_nodes), node_labels, edges)
        })
        .prop_map(|(num_nodes, node_labels, edges)| {
            let mut graph = GraphStorage::with_capacity(num_nodes, edges.len());
            for (i, label) in node_labels.iter().enumerate() {
                graph.add_node(i as u64, label, 0);
            }
            for (i, (from, to)) in edges.iter().enumerate() {
                graph.add_edge(i as u64, "knows", NodeIndex(*from), NodeIndex(*to), 0);
            }
            graph
        })
    }

    proptest! {
        /// Every edge endpoint must reference a valid node.
        #[test]
        fn edge_endpoints_are_valid(graph in arb_graph()) {
            let n = graph.node_count();
            for i in 0..graph.edge_count() {
                let edge = graph.edge(EdgeIndex(i as u32));
                prop_assert!(edge.from.as_usize() < n, "from out of bounds");
                prop_assert!(edge.to.as_usize() < n, "to out of bounds");
            }
        }

        /// Bidirectional consistency: if edge E is in outgoing[A], then E is in incoming[B].
        #[test]
        fn bidirectional_adjacency_consistent(graph in arb_graph()) {
            for i in 0..graph.node_count() {
                let node = NodeIndex(i as u32);
                for &eidx in graph.outgoing_edges(node) {
                    let edge = graph.edge(eidx);
                    let target = edge.to;
                    prop_assert!(
                        graph.incoming_edges(target).contains(&eidx),
                        "Edge {} in outgoing[{}] but not in incoming[{}]",
                        eidx, node, target
                    );
                }
            }
        }

        /// BFS visits every reachable node exactly once.
        #[test]
        fn bfs_no_duplicates(graph in arb_graph()) {
            if graph.node_count() == 0 { return Ok(()); }
            let results = bfs(&graph, NodeIndex(0), usize::MAX, None);
            let mut seen = std::collections::HashSet::new();
            for (node, _depth) in &results {
                prop_assert!(seen.insert(*node), "BFS visited {} twice", node);
            }
        }

        /// Dijkstra path is valid: consecutive nodes in the path are connected by edges.
        #[test]
        fn dijkstra_path_is_valid(graph in arb_graph()) {
            if graph.node_count() < 2 { return Ok(()); }
            let from = NodeIndex(0);
            let to = NodeIndex((graph.node_count() - 1) as u32);
            if let Some((_cost, path)) = shortest_path(&graph, from, to, |_| 1.0) {
                prop_assert_eq!(path[0], from);
                prop_assert_eq!(*path.last().unwrap(), to);
                for window in path.windows(2) {
                    let has_edge = graph
                        .outgoing_edges(window[0])
                        .iter()
                        .any(|&eidx| graph.edge(eidx).to == window[1]);
                    prop_assert!(has_edge, "No edge from {} to {} in Dijkstra path", window[0], window[1]);
                }
            }
        }

        /// PageRank scores sum to ~1.0 (within floating point tolerance).
        #[test]
        fn pagerank_scores_sum_to_one(graph in arb_graph()) {
            if graph.node_count() == 0 { return Ok(()); }
            let scores = pagerank(&graph, 0.85, 1e-8, 100);
            let sum: f64 = scores.iter().sum();
            prop_assert!((sum - 1.0).abs() < 0.01, "PageRank sum = {}, expected ~1.0", sum);
        }

        /// Connected components: every node is in exactly one component.
        #[test]
        fn components_partition_all_nodes(graph in arb_graph()) {
            let components = connected_components(&graph);
            let total: usize = components.values().map(|v| v.len()).sum();
            prop_assert_eq!(total, graph.node_count());
        }
    }
}

Known-Graph Fixtures

#[cfg(test)]
mod fixture_tests {
    use super::*;

    /// Build a known triangle graph: A -> B -> C -> A
    fn triangle_graph() -> GraphStorage {
        let mut g = GraphStorage::new();
        let a = g.add_node(1, "person", 0);
        let b = g.add_node(2, "person", 0);
        let c = g.add_node(3, "person", 0);
        g.add_edge(1, "knows", a, b, 0);
        g.add_edge(2, "knows", b, c, 0);
        g.add_edge(3, "knows", c, a, 0);
        g
    }

    #[test]
    fn triangle_bfs_visits_all() {
        let g = triangle_graph();
        let results = bfs(&g, NodeIndex(0), 10, None);
        assert_eq!(results.len(), 3);
    }

    #[test]
    fn triangle_shortest_path() {
        let g = triangle_graph();
        let (cost, path) =
            shortest_path(&g, NodeIndex(0), NodeIndex(2), |_| 1.0).unwrap();
        assert_eq!(cost, 2.0);
        assert_eq!(path, vec![NodeIndex(0), NodeIndex(1), NodeIndex(2)]);
    }

    #[test]
    fn triangle_is_one_component() {
        let g = triangle_graph();
        let components = connected_components(&g);
        assert_eq!(components.len(), 1);
    }

    #[test]
    fn disconnected_components() {
        let mut g = GraphStorage::new();
        let a = g.add_node(1, "person", 0);
        let b = g.add_node(2, "person", 0);
        let c = g.add_node(3, "person", 0);
        let d = g.add_node(4, "person", 0);
        // Component 1: A <-> B
        g.add_edge(1, "knows", a, b, 0);
        g.add_edge(2, "knows", b, a, 0);
        // Component 2: C <-> D
        g.add_edge(3, "knows", c, d, 0);
        g.add_edge(4, "knows", d, c, 0);

        let components = connected_components(&g);
        assert_eq!(components.len(), 2);
    }

    #[test]
    fn pattern_match_two_hop() {
        let mut g = GraphStorage::new();
        let alice = g.add_node(1, "people", 0);
        let bob = g.add_node(2, "people", 0);
        let charlie = g.add_node(3, "people", 0);
        g.add_edge(1, "knows", alice, bob, 0);
        g.add_edge(2, "knows", bob, charlie, 0);

        // Pattern: (a:people)-[:knows]->(b:people)-[:knows]->(c:people)
        let pattern = GraphPattern {
            elements: vec![
                PatternElement::Node {
                    variable: "a".into(),
                    label: Some("people".into()),
                },
                PatternElement::Edge {
                    variable: "k1".into(),
                    label: Some("knows".into()),
                    direction: EdgeDirection::Outgoing,
                    hops: None,
                },
                PatternElement::Node {
                    variable: "b".into(),
                    label: Some("people".into()),
                },
                PatternElement::Edge {
                    variable: "k2".into(),
                    label: Some("knows".into()),
                    direction: EdgeDirection::Outgoing,
                    hops: None,
                },
                PatternElement::Node {
                    variable: "c".into(),
                    label: Some("people".into()),
                },
            ],
        };

        let steps = compile_pattern(&pattern, g.labels());
        let bindings = execute_pattern(&g, &steps);

        // Only one match: alice -> bob -> charlie
        assert_eq!(bindings.len(), 1);
        assert_eq!(bindings[0].nodes["a"], alice);
        assert_eq!(bindings[0].nodes["b"], bob);
        assert_eq!(bindings[0].nodes["c"], charlie);
    }
}

Performance Benchmarks

Target metrics for a production graph engine:

Use criterion for benchmarking:

use criterion::{criterion_group, criterion_main, Criterion, BenchmarkId};

fn bench_bfs(c: &mut Criterion) {
    let graph = build_benchmark_graph(1_000_000, 10_000_000);
    c.bench_function("bfs_depth_3_1M", |b| {
        b.iter(|| bfs(&graph, NodeIndex(0), 3, None))
    });
}

fn bench_dijkstra(c: &mut Criterion) {
    let graph = build_benchmark_graph(1_000_000, 10_000_000);
    c.bench_function("dijkstra_1M", |b| {
        b.iter(|| {
            shortest_path(&graph, NodeIndex(0), NodeIndex(999_999), |_| 1.0)
        })
    });
}

fn bench_pagerank(c: &mut Criterion) {
    let graph = build_benchmark_graph(1_000_000, 10_000_000);
    c.bench_function("pagerank_10iter_1M", |b| {
        b.iter(|| pagerank(&graph, 0.85, 1e-6, 10))
    });
}

criterion_group!(benches, bench_bfs, bench_dijkstra, bench_pagerank);
criterion_main!(benches);

---

Never

• Never use `Rc<RefCell<Node>>` for graph nodes — use index-based references (NodeIndex, EdgeIndex). Rc/RefCell is slower, non-Send, and creates garbage collection pressure.
• Never use recursive DFS on large graphs — Rust's default stack is 8 MB. A graph with 100K+ depth will stack overflow. Always use iterative traversal with an explicit stack or queue.
• Never store `&NodeData` references across mutations — adding a node or edge can reallocate the backing Vec, invalidating all references. Use indices.
• Never use `HashMap<NodeIndex, Vec<EdgeIndex>>` for adjacency — use Vec<Vec<EdgeIndex>> indexed directly by NodeIndex.0. HashMap adds 30-50% overhead for this access pattern.
• Never skip the bidirectional invariant — when adding an edge from A to B, ALWAYS update both outgoing[A] and incoming[B]. Forgetting one direction breaks incoming-edge traversals silently.
• Never use `f64` equality in Dijkstra — compare with > / <, not ==. Floating point accumulation makes exact equality unreliable. The DijkstraState Ord impl handles this correctly.
• Never allocate per-traversal when you can reuse buffers — for hot-path traversals, accept &mut Vec<bool> visited buffers from the caller instead of allocating inside the function.
• Never use `unsafe` for graph structure — the entire graph engine is safe Rust. unsafe is only justified in the arena allocator internals (bumpalo handles this) or SIMD acceleration of bulk operations.

Async & Concurrency — Tokio Runtime, io_uring, Crossbeam Lock-Free, Concurrent Data Structures

Every concurrent Rust component assumes these patterns. Do not deviate.

Tokio Runtime

Runtime Setup

use std::time::Duration;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // For most services: use the macro. Tokio picks sensible defaults.
    run_server().await
}

When you need explicit control over threads, queue depth, or thread naming:

fn main() -> anyhow::Result<()> {
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(num_cpus::get())
        .max_blocking_threads(64)         // ceiling for spawn_blocking pool
        .thread_name("engine-worker")
        .thread_stack_size(4 * 1024 * 1024)
        .enable_all()
        .build()?;

    runtime.block_on(async {
        run_server().await
    })
}

When to customize:

• worker_threads — match physical cores for CPU-heavy async tasks; default is fine for I/O-bound workloads
• max_blocking_threads — raise above 512 if the blocking pool saturates (compression, hashing, fsync)
• thread_name — essential for profiling and top -H readability in production
• thread_stack_size — raise only for deeply recursive call stacks (parser, query planner)

Task Types — When to Use Each

tokio::spawn — Concurrent I/O Tasks

Use for anything that awaits network, disk, or channel I/O. The future must be Send + 'static.

use tokio::net::TcpStream;

async fn handle_connections(listener: tokio::net::TcpListener) {
    loop {
        let (stream, peer) = listener.accept().await.unwrap();
        tokio::spawn(async move {
            if let Err(e) = process_connection(stream).await {
                tracing::error!(?peer, error = %e, "connection failed");
            }
        });
    }
}

async fn process_connection(stream: TcpStream) -> anyhow::Result<()> {
    // All I/O here is async — reads, writes, channel sends
    todo!()
}

tokio::task::spawn_blocking — CPU-Bound Work

Use for anything that takes >10us without an await point: compression, hashing, CRC computation, SIMD distance calculations, serialization of large payloads.

use bytes::Bytes;

/// Compress a WAL segment before writing to object storage.
async fn compress_segment(data: Bytes) -> anyhow::Result<Bytes> {
    tokio::task::spawn_blocking(move || {
        let mut encoder = zstd::Encoder::new(Vec::new(), 3)?;
        std::io::Write::write_all(&mut encoder, &data)?;
        let compressed = encoder.finish()?;
        Ok(Bytes::from(compressed))
    })
    .await?
}

/// Hash a block for content-addressed storage.
async fn hash_block(data: Bytes) -> [u8; 32] {
    tokio::task::spawn_blocking(move || {
        use ring::digest;
        let hash = digest::digest(&digest::SHA256, &data);
        let mut out = [0u8; 32];
        out.copy_from_slice(hash.as_ref());
        out
    })
    .await
    .expect("blocking task panicked")
}

Never do this on the tokio runtime directly — it blocks the worker thread and starves other tasks:

// WRONG: blocks the async runtime
async fn bad_compress(data: &[u8]) -> Vec<u8> {
    zstd::encode_all(data, 3).unwrap() // This takes milliseconds — blocks the worker
}

Dedicated OS Threads — Long-Running Background Work

Use std::thread::spawn for work that runs for the lifetime of the process and does not need async I/O: compaction loops, WAL flush threads, background merge operations.

use crossbeam_channel::{Receiver, bounded};
use std::thread;

struct CompactionHandle {
    _thread: thread::JoinHandle<()>,
}

fn start_compaction_thread(rx: Receiver<CompactionRequest>) -> CompactionHandle {
    let handle = thread::Builder::new()
        .name("compaction".into())
        .stack_size(8 * 1024 * 1024)
        .spawn(move || {
            while let Ok(request) = rx.recv() {
                // CPU-bound: merge sorted runs, rewrite segments
                // This runs on its own OS thread — never touches tokio
                if let Err(e) = compact(request) {
                    tracing::error!(error = %e, "compaction failed");
                }
            }
            tracing::info!("compaction thread exiting");
        })
        .expect("failed to spawn compaction thread");

    CompactionHandle { _thread: handle }
}

tokio::task::spawn_local — !Send Futures

Rare. Use only when a future holds a non-Send type (e.g., Rc, raw pointers to thread-local data). Requires a LocalSet:

use tokio::task::LocalSet;

async fn run_local_work() {
    let local = LocalSet::new();
    local.run_until(async {
        tokio::task::spawn_local(async {
            // Can hold Rc, Cell, and other !Send types here
        }).await.unwrap();
    }).await;
}

Graceful Shutdown

Complete pattern using CancellationToken with drain period:

use std::time::Duration;
use tokio::signal;
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;

pub struct ShutdownController {
    token: CancellationToken,
    drain_tx: watch::Sender<bool>,
    drain_rx: watch::Receiver<bool>,
}

impl ShutdownController {
    pub fn new() -> Self {
        let (drain_tx, drain_rx) = watch::channel(false);
        Self {
            token: CancellationToken::new(),
            drain_tx,
            drain_rx,
        }
    }

    pub fn token(&self) -> CancellationToken {
        self.token.clone()
    }

    pub fn drain_rx(&self) -> watch::Receiver<bool> {
        self.drain_rx.clone()
    }

    /// Run the shutdown sequence:
    /// 1. Receive signal → enter drain mode (stop accepting new work)
    /// 2. Wait `drain_period` for in-flight work to complete
    /// 3. Cancel all remaining tasks
    pub async fn wait_for_shutdown(self, drain_period: Duration) {
        // Wait for SIGINT or SIGTERM
        let ctrl_c = signal::ctrl_c();
        #[cfg(unix)]
        let mut sigterm = signal::unix::signal(signal::unix::SignalKind::terminate())
            .expect("failed to register SIGTERM handler");

        #[cfg(unix)]
        tokio::select! {
            _ = ctrl_c => tracing::info!("received SIGINT"),
            _ = sigterm.recv() => tracing::info!("received SIGTERM"),
        }

        #[cfg(not(unix))]
        ctrl_c.await.expect("failed to listen for Ctrl+C");

        // Phase 1: Drain — stop accepting new connections
        tracing::info!("entering drain phase ({drain_period:?})");
        let _ = self.drain_tx.send(true);

        // Phase 2: Wait for in-flight work
        tokio::time::sleep(drain_period).await;

        // Phase 3: Cancel all remaining tasks
        tracing::info!("cancelling remaining tasks");
        self.token.cancel();
    }
}

Service integration:

async fn run_server() -> anyhow::Result<()> {
    let shutdown = ShutdownController::new();
    let token = shutdown.token();
    let mut drain_rx = shutdown.drain_rx();

    let listener = tokio::net::TcpListener::bind("0.0.0.0:5432").await?;

    // Spawn the shutdown controller
    let shutdown_handle = tokio::spawn(shutdown.wait_for_shutdown(Duration::from_secs(30)));

    loop {
        tokio::select! {
            // Accept new connections until drain starts
            result = listener.accept() => {
                let (socket, peer) = result?;
                let task_token = token.clone();
                tokio::spawn(async move {
                    tokio::select! {
                        result = handle_connection(socket) => {
                            if let Err(e) = result {
                                tracing::error!(?peer, error = %e, "connection error");
                            }
                        }
                        _ = task_token.cancelled() => {
                            tracing::debug!(?peer, "connection cancelled");
                        }
                    }
                });
            }
            // Stop accepting when drain signal arrives
            _ = drain_rx.changed() => {
                tracing::info!("drain active — stopped accepting connections");
                break;
            }
        }
    }

    // Wait for shutdown to complete
    shutdown_handle.await?;
    tracing::info!("shutdown complete");
    Ok(())
}

Cancellation Safety

tokio::select! drops the losing branch's future. This is safe for some operations and dangerous for others.

Safe to cancel

NOT safe to cancel

Pattern: Make cancellation-unsafe code safe

Wrap the entire unsafe sequence in a single tokio::spawn — the spawned task runs to completion even if the parent future is cancelled:

use tokio::sync::oneshot;

/// Flush a WAL buffer to disk. Must not be cancelled mid-write.
async fn flush_wal(wal: &WalWriter, buffer: Vec<u8>) -> anyhow::Result<()> {
    let (tx, rx) = oneshot::channel();
    let wal = wal.clone();

    // Spawn an uncancellable task — runs to completion even if the caller drops us
    tokio::spawn(async move {
        let result = wal.write_and_fsync(buffer).await;
        let _ = tx.send(result);
    });

    rx.await?
}

Pattern: Checkpoint before yield points

/// Process a batch of mutations. Checkpoint after each item so cancellation
/// never loses more than one item of work.
async fn process_batch(items: Vec<Mutation>, state: &mut ProcessState) -> anyhow::Result<()> {
    for item in items {
        // Apply the mutation
        state.apply(&item)?;

        // Checkpoint — if we get cancelled after this yield, the mutation is saved
        state.checkpoint().await?;

        // Yield to allow cancellation between items, not during an item
        tokio::task::yield_now().await;
    }
    Ok(())
}

Async Traits

Native Async Fn in Traits (Rust 1.75+)

Since Rust 1.75, async functions work directly in traits:

pub trait StorageBackend: Send + Sync {
    async fn read(&self, path: &str, offset: u64, len: u64) -> Result<bytes::Bytes, StorageError>;
    async fn write(&self, path: &str, data: &[u8]) -> Result<(), StorageError>;
    async fn delete(&self, path: &str) -> Result<(), StorageError>;
    async fn exists(&self, path: &str) -> Result<bool, StorageError>;
}

Limitation: native async trait methods do not automatically produce Send futures. When you need the future to be Send (for tokio::spawn), you must add the bound explicitly:

pub trait StorageBackend: Send + Sync {
    fn read(&self, path: &str, offset: u64, len: u64)
        -> impl Future<Output = Result<bytes::Bytes, StorageError>> + Send;
}

async_trait Crate — When Still Needed

Use async_trait when: 1. You need dyn Trait (object safety) — native async traits are not object-safe 2. You target older MSRV (<1.75) 3. A third-party crate requires it (e.g., pgwire, datafusion)

use async_trait::async_trait;

#[async_trait]
pub trait QueryExecutor: Send + Sync {
    async fn execute(&self, plan: &ExecutionPlan) -> Result<RecordBatchStream, QueryError>;
}

// Can be used as dyn Trait:
async fn run_query(executor: &dyn QueryExecutor, plan: &ExecutionPlan) {
    let stream = executor.execute(plan).await.unwrap();
    // ...
}

Decision: Native vs async_trait

io_uring (Linux)

Why io_uring

Standard POSIX file I/O (read/write/pread/pwrite) requires one syscall per operation. io_uring amortizes syscall overhead by batching:

• Submission Queue (SQ): userspace pushes I/O requests without a syscall
• Completion Queue (CQ): kernel pushes results without a syscall
• Single `io_uring_enter` syscall submits and reaps an entire batch

For random read-heavy workloads (page lookups, index traversal), io_uring achieves 2-3x the IOPS of pread because it eliminates per-operation syscall overhead and enables the kernel to optimize scheduling across the batch.

Basic Usage with the io-uring Crate

#[cfg(target_os = "linux")]
use io_uring::{IoUring, opcode, types, squeue};
use std::os::unix::io::AsRawFd;

#[cfg(target_os = "linux")]
pub struct UringReader {
    ring: IoUring,
}

#[cfg(target_os = "linux")]
impl UringReader {
    pub fn new(queue_depth: u32) -> std::io::Result<Self> {
        let ring = IoUring::builder()
            .setup_sqpoll(1000) // kernel-side polling — reduces syscalls further
            .build(queue_depth)?;
        Ok(Self { ring })
    }

    /// Read a single block from the file at the given offset.
    pub fn read_block(
        &mut self,
        fd: &std::fs::File,
        buf: &mut [u8],
        offset: u64,
    ) -> std::io::Result<usize> {
        let read_entry = opcode::Read::new(
            types::Fd(fd.as_raw_fd()),
            buf.as_mut_ptr(),
            buf.len() as u32,
        )
        .offset(offset)
        .build()
        .user_data(0x01);

        // Submit
        unsafe {
            self.ring.submission().push(&read_entry)
                .map_err(|_| std::io::Error::new(
                    std::io::ErrorKind::Other,
                    "submission queue full",
                ))?;
        }
        self.ring.submit_and_wait(1)?;

        // Reap
        let cqe = self.ring.completion().next()
            .ok_or_else(|| std::io::Error::new(
                std::io::ErrorKind::Other,
                "no completion entry",
            ))?;

        let result = cqe.result();
        if result < 0 {
            Err(std::io::Error::from_raw_os_error(-result))
        } else {
            Ok(result as usize)
        }
    }

    /// Submit a batch of reads and wait for all completions.
    /// Returns results in submission order.
    pub fn read_batch(
        &mut self,
        fd: &std::fs::File,
        requests: &mut [(Vec<u8>, u64)], // (buffer, offset) pairs
    ) -> std::io::Result<Vec<usize>> {
        let raw_fd = fd.as_raw_fd();

        // Submit all reads
        for (i, (buf, offset)) in requests.iter_mut().enumerate() {
            let entry = opcode::Read::new(
                types::Fd(raw_fd),
                buf.as_mut_ptr(),
                buf.len() as u32,
            )
            .offset(*offset)
            .build()
            .user_data(i as u64);

            unsafe {
                self.ring.submission().push(&entry)
                    .map_err(|_| std::io::Error::new(
                        std::io::ErrorKind::Other,
                        "submission queue full",
                    ))?;
            }
        }

        // Submit and wait for all
        self.ring.submit_and_wait(requests.len())?;

        // Collect results, ordered by user_data
        let mut results = vec![0usize; requests.len()];
        for cqe in self.ring.completion() {
            let idx = cqe.user_data() as usize;
            let result = cqe.result();
            if result < 0 {
                return Err(std::io::Error::from_raw_os_error(-result));
            }
            results[idx] = result as usize;
        }

        Ok(results)
    }
}

Feature Gating for Platform Portability

io_uring is Linux-only. Gate it behind both #[cfg(target_os = "linux")] and a feature flag so the crate compiles on all platforms:

// In Cargo.toml:
// [features]
// io-uring = ["dep:io-uring"]
//
// [target.'cfg(target_os = "linux")'.dependencies]
// io-uring = { workspace = true, optional = true }

#[cfg(all(target_os = "linux", feature = "io-uring"))]
mod uring_backend;

#[cfg(not(all(target_os = "linux", feature = "io-uring")))]
mod fallback_backend;

// Unified trait — callers do not know which backend is active
pub trait BlockReader: Send + Sync {
    fn read_block(&self, offset: u64, len: u32) -> std::io::Result<Vec<u8>>;
    fn read_batch(&self, requests: &[(u64, u32)]) -> std::io::Result<Vec<Vec<u8>>>;
}

// Fallback uses standard pread — works on macOS, Windows, all Linuxes
#[cfg(not(all(target_os = "linux", feature = "io-uring")))]
mod fallback_backend {
    use std::os::unix::fs::FileExt;

    pub struct PreadReader {
        file: std::fs::File,
    }

    impl super::BlockReader for PreadReader {
        fn read_block(&self, offset: u64, len: u32) -> std::io::Result<Vec<u8>> {
            let mut buf = vec![0u8; len as usize];
            self.file.read_exact_at(&mut buf, offset)?;
            Ok(buf)
        }

        fn read_batch(&self, requests: &[(u64, u32)]) -> std::io::Result<Vec<Vec<u8>>> {
            requests.iter()
                .map(|&(offset, len)| self.read_block(offset, len))
                .collect()
        }
    }
}

io_uring with O_DIRECT

For bypassing the page cache (large sequential scans, compaction reads where data is used once):

#[cfg(target_os = "linux")]
use std::os::unix::fs::OpenOptionsExt;

#[cfg(target_os = "linux")]
fn open_direct(path: &std::path::Path) -> std::io::Result<std::fs::File> {
    std::fs::OpenOptions::new()
        .read(true)
        .custom_flags(libc::O_DIRECT)
        .open(path)
}

O_DIRECT constraints: buffer must be aligned to 512 bytes (or filesystem block size), read length must be a multiple of 512 bytes. Use aligned_alloc or manually align buffers.

Crossbeam Lock-Free Structures

SkipMap — Concurrent Sorted Map

SkipMap provides lock-free concurrent reads and writes with sorted key ordering. Ideal for in-memory sorted indices:

use crossbeam_skiplist::SkipMap;
use bytes::Bytes;

pub struct Memtable {
    data: SkipMap<Bytes, Bytes>,
    size: std::sync::atomic::AtomicUsize,
}

impl Memtable {
    pub fn new() -> Self {
        Self {
            data: SkipMap::new(),
            size: std::sync::atomic::AtomicUsize::new(0),
        }
    }

    pub fn insert(&self, key: Bytes, value: Bytes) {
        let entry_size = key.len() + value.len();
        self.data.insert(key, value);
        self.size.fetch_add(entry_size, std::sync::atomic::Ordering::Relaxed);
    }

    pub fn get(&self, key: &[u8]) -> Option<Bytes> {
        self.data.get(key).map(|entry| entry.value().clone())
    }

    /// Iterate over a key range — lock-free, consistent snapshot.
    pub fn range(&self, start: &[u8], end: &[u8]) -> Vec<(Bytes, Bytes)> {
        self.data
            .range(Bytes::copy_from_slice(start)..Bytes::copy_from_slice(end))
            .map(|entry| (entry.key().clone(), entry.value().clone()))
            .collect()
    }

    pub fn approximate_size(&self) -> usize {
        self.size.load(std::sync::atomic::Ordering::Relaxed)
    }
}

Epoch-Based Reclamation

Crossbeam uses epoch-based garbage collection to safely reclaim memory in lock-free structures. Understanding this is essential for debugging memory growth:

How it works: 1. Each thread pins the current global epoch before accessing shared data 2. While pinned, no memory from the current or adjacent epochs is reclaimed 3. When a thread removes data, the removed node is deferred — placed on a garbage list tagged with the current epoch 4. Memory is reclaimed only when all threads have advanced past the epoch in which the data was removed

use crossbeam_epoch::{self as epoch, Atomic, Owned, Shared};
use std::sync::atomic::Ordering;

pub struct LockFreeStack<T> {
    head: Atomic<Node<T>>,
}

struct Node<T> {
    data: T,
    next: Atomic<Node<T>>,
}

impl<T> LockFreeStack<T> {
    pub fn new() -> Self {
        Self { head: Atomic::null() }
    }

    pub fn push(&self, data: T) {
        let node = Owned::new(Node {
            data,
            next: Atomic::null(),
        });

        let guard = epoch::pin();
        let mut node = node;

        loop {
            let head = self.head.load(Ordering::Relaxed, &guard);
            node.next.store(head, Ordering::Relaxed);

            match self.head.compare_exchange(
                head,
                node,
                Ordering::Release,
                Ordering::Relaxed,
                &guard,
            ) {
                Ok(_) => break,
                Err(err) => node = err.new,
            }
        }
    }

    pub fn pop(&self) -> Option<T> {
        let guard = epoch::pin();
        loop {
            let head = self.head.load(Ordering::Acquire, &guard);
            let head_ref = unsafe { head.as_ref()? };

            let next = head_ref.next.load(Ordering::Relaxed, &guard);

            if self.head
                .compare_exchange(head, next, Ordering::Release, Ordering::Relaxed, &guard)
                .is_ok()
            {
                // SAFETY: We won the CAS, so we have exclusive ownership of this node.
                // Defer deallocation until all threads have advanced past this epoch.
                unsafe {
                    let data = std::ptr::read(&head_ref.data);
                    guard.defer_destroy(head);
                    return Some(data);
                }
            }
        }
    }
}

Key rule: always pin() before accessing shared atomic pointers. Keep the pin duration short — a long-held pin prevents garbage collection across all threads.

SegQueue — Lock-Free FIFO

use crossbeam_queue::SegQueue;

/// Lock-free queue for passing completed I/O results back to the event loop.
struct IoCompletionQueue {
    queue: SegQueue<IoResult>,
}

struct IoResult {
    request_id: u64,
    data: Result<Vec<u8>, std::io::Error>,
}

impl IoCompletionQueue {
    fn new() -> Self {
        Self { queue: SegQueue::new() }
    }

    fn push(&self, result: IoResult) {
        self.queue.push(result);
    }

    fn drain(&self) -> Vec<IoResult> {
        let mut results = Vec::new();
        while let Some(result) = self.queue.pop() {
            results.push(result);
        }
        results
    }
}

Crossbeam Channels

Three channel types for different coordination patterns:

use crossbeam_channel::{bounded, unbounded, select, Receiver, Sender};
use std::time::Duration;

// Bounded — backpressure. Writer blocks when channel is full.
// Use between fast producers and slow consumers (WAL writer → compaction).
let (wal_tx, wal_rx): (Sender<WalEntry>, Receiver<WalEntry>) = bounded(1024);

// Unbounded — fire-and-forget events. Never blocks the sender.
// Use for metrics, logging, notifications where dropping is worse than memory growth.
let (event_tx, event_rx): (Sender<Event>, Receiver<Event>) = unbounded();

// select! — multiplex multiple channels, similar to Go's select.
fn compaction_loop(
    work_rx: Receiver<CompactionRequest>,
    shutdown_rx: Receiver<()>,
) {
    loop {
        select! {
            recv(work_rx) -> msg => {
                match msg {
                    Ok(request) => compact(request),
                    Err(_) => break, // channel closed
                }
            }
            recv(shutdown_rx) -> _ => {
                tracing::info!("compaction shutting down");
                break;
            }
            default(Duration::from_secs(60)) => {
                // No work for 60s — run periodic maintenance
                run_gc();
            }
        }
    }
}

When to use crossbeam channels vs tokio channels:

Concurrent Patterns for Database Engines

Reader-Writer Lock with Tokio

use tokio::sync::RwLock;
use std::sync::Arc;

pub struct Schema {
    pub columns: Vec<ColumnDef>,
    pub version: u64,
}

pub struct TableState {
    /// Many concurrent readers (queries), rare writers (schema changes).
    schema: RwLock<Arc<Schema>>,
}

impl TableState {
    /// Fast path — clone the Arc, release the lock immediately.
    pub async fn schema(&self) -> Arc<Schema> {
        self.schema.read().await.clone()
    }

    /// Slow path — exclusive lock for schema changes.
    pub async fn alter_schema(&self, new_schema: Schema) {
        let mut guard = self.schema.write().await;
        *guard = Arc::new(new_schema);
        // Lock released here — all waiting readers proceed
    }
}

Optimization: For extremely hot read paths, avoid even the RwLock by using arc_swap:

use arc_swap::ArcSwap;
use std::sync::Arc;

pub struct HotTableState {
    /// Lock-free reads via ArcSwap — no contention on the read path at all.
    schema: ArcSwap<Schema>,
}

impl HotTableState {
    pub fn schema(&self) -> Arc<Schema> {
        self.schema.load_full()
    }

    pub fn update_schema(&self, new_schema: Schema) {
        self.schema.store(Arc::new(new_schema));
    }
}

Double-Buffered Memtable

The double-buffer pattern allows writes to continue on a fresh buffer while the old buffer is being flushed to disk:

use crossbeam_skiplist::SkipMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use parking_lot::Mutex;
use bytes::Bytes;

pub struct DoubleBufferedMemtable {
    /// Index of the currently active table (0 or 1).
    active: AtomicUsize,
    /// The two memtables. Writers always write to tables[active].
    tables: [Arc<SkipMap<Bytes, Bytes>>; 2],
    /// Protects the swap operation — only one flush at a time.
    flush_lock: Mutex<()>,
}

impl DoubleBufferedMemtable {
    pub fn new() -> Self {
        Self {
            active: AtomicUsize::new(0),
            tables: [Arc::new(SkipMap::new()), Arc::new(SkipMap::new())],
            flush_lock: Mutex::new(()),
        }
    }

    /// Insert into the active memtable. Lock-free, concurrent-safe.
    pub fn insert(&self, key: Bytes, value: Bytes) {
        let idx = self.active.load(Ordering::Acquire);
        self.tables[idx].insert(key, value);
    }

    /// Read from the active memtable first, then check the inactive one
    /// (it may still be draining during a flush).
    pub fn get(&self, key: &[u8]) -> Option<Bytes> {
        let idx = self.active.load(Ordering::Acquire);
        if let Some(entry) = self.tables[idx].get(key) {
            return Some(entry.value().clone());
        }
        // Check the inactive table — may still have data being flushed
        let other = 1 - idx;
        self.tables[other].get(key).map(|e| e.value().clone())
    }

    /// Swap the active buffer and return the old one for flushing.
    /// Only one thread calls this at a time (protected by flush_lock).
    pub fn swap_for_flush(&self) -> Arc<SkipMap<Bytes, Bytes>> {
        let _guard = self.flush_lock.lock();
        let old_idx = self.active.load(Ordering::Acquire);
        let new_idx = 1 - old_idx;

        // The new table should be empty (cleared after previous flush completed)
        assert!(self.tables[new_idx].is_empty(), "new buffer not empty — previous flush incomplete");

        // Swap — new writes go to the empty table
        self.active.store(new_idx, Ordering::Release);

        // Return the old table for flushing
        Arc::clone(&self.tables[old_idx])
    }

    /// Called after flush completes — clear the flushed table so it's ready for reuse.
    pub fn clear_flushed(&self, idx: usize) {
        self.tables[idx].clear();
    }
}

MVCC Snapshot Manager

Multi-Version Concurrency Control lets readers see a consistent snapshot while writers proceed without blocking:

use std::collections::BTreeSet;
use std::sync::atomic::{AtomicU64, Ordering};
use parking_lot::RwLock;

pub struct SnapshotManager {
    /// Monotonically increasing transaction ID.
    current_txn: AtomicU64,
    /// Set of all active (in-use) snapshot IDs.
    active_snapshots: RwLock<BTreeSet<u64>>,
}

#[derive(Debug, Clone)]
pub struct Snapshot {
    pub txn_id: u64,
    /// This snapshot can see all versions with txn_id < visible_before.
    pub visible_before: u64,
}

impl SnapshotManager {
    pub fn new() -> Self {
        Self {
            current_txn: AtomicU64::new(1),
            active_snapshots: RwLock::new(BTreeSet::new()),
        }
    }

    /// Allocate a new transaction ID.
    pub fn next_txn_id(&self) -> u64 {
        self.current_txn.fetch_add(1, Ordering::SeqCst)
    }

    /// Create a snapshot that sees everything committed before this moment.
    pub fn acquire_snapshot(&self) -> Snapshot {
        let txn_id = self.current_txn.load(Ordering::SeqCst);
        self.active_snapshots.write().insert(txn_id);
        Snapshot {
            txn_id,
            visible_before: txn_id,
        }
    }

    /// Release a snapshot — allows GC of versions it was holding alive.
    pub fn release_snapshot(&self, snapshot: &Snapshot) {
        self.active_snapshots.write().remove(&snapshot.txn_id);
    }

    /// The oldest active snapshot. Versions older than this can be garbage collected
    /// because no reader can see them.
    pub fn min_active_snapshot(&self) -> Option<u64> {
        self.active_snapshots.read().iter().next().copied()
    }

    /// Check if a version is visible to a given snapshot.
    pub fn is_visible(&self, version_txn: u64, snapshot: &Snapshot) -> bool {
        version_txn < snapshot.visible_before
    }
}

/// RAII guard that releases the snapshot when dropped.
pub struct SnapshotGuard<'a> {
    manager: &'a SnapshotManager,
    pub snapshot: Snapshot,
}

impl<'a> SnapshotGuard<'a> {
    pub fn new(manager: &'a SnapshotManager) -> Self {
        let snapshot = manager.acquire_snapshot();
        Self { manager, snapshot }
    }
}

impl Drop for SnapshotGuard<'_> {
    fn drop(&mut self) {
        self.manager.release_snapshot(&self.snapshot);
    }
}

Usage:

fn query_with_snapshot(manager: &SnapshotManager) {
    let guard = SnapshotGuard::new(manager);

    // All reads during this scope see a consistent snapshot
    // Even if other threads are writing new versions concurrently
    let snapshot = &guard.snapshot;

    // When guard is dropped, the snapshot is released
    // GC can now reclaim versions that only this snapshot was keeping alive
}

Group Commit

Batch multiple WAL entries into a single fsync to amortize disk flush cost:

use std::time::{Duration, Instant};
use tokio::sync::{Mutex, Notify, oneshot};
use std::collections::VecDeque;

pub struct WalEntry {
    pub data: Vec<u8>,
}

struct PendingEntry {
    entry: WalEntry,
    done: oneshot::Sender<Result<u64, WalError>>,
}

pub struct GroupCommitter {
    pending: Mutex<VecDeque<PendingEntry>>,
    notify: Notify,
    max_batch_size: usize,
    max_batch_delay: Duration,
}

#[derive(Debug, thiserror::Error)]
pub enum WalError {
    #[error("WAL write failed: {0}")]
    Io(#[from] std::io::Error),
}

impl GroupCommitter {
    pub fn new(max_batch_size: usize, max_batch_delay: Duration) -> Self {
        Self {
            pending: Mutex::new(VecDeque::new()),
            notify: Notify::new(),
            max_batch_size,
            max_batch_delay,
        }
    }

    /// Append an entry. Returns the LSN (log sequence number) once the group
    /// commit flushes it to disk. The caller blocks until the flush completes.
    pub async fn append(&self, entry: WalEntry) -> Result<u64, WalError> {
        let (tx, rx) = oneshot::channel();
        {
            let mut pending = self.pending.lock().await;
            pending.push_back(PendingEntry { entry, done: tx });
        }
        self.notify.notify_one();
        rx.await.expect("group committer dropped sender")
    }

    /// Background loop: collect entries, flush in batches.
    pub async fn run(&self, wal_writer: &mut WalWriter) {
        loop {
            // Wait for at least one entry
            self.notify.notified().await;

            // Collect up to max_batch_size entries, or until max_batch_delay expires
            let deadline = Instant::now() + self.max_batch_delay;
            let mut batch = Vec::new();

            loop {
                {
                    let mut pending = self.pending.lock().await;
                    while let Some(entry) = pending.pop_front() {
                        batch.push(entry);
                        if batch.len() >= self.max_batch_size {
                            break;
                        }
                    }
                }

                if batch.len() >= self.max_batch_size || Instant::now() >= deadline {
                    break;
                }

                // Wait briefly for more entries to accumulate
                tokio::select! {
                    _ = self.notify.notified() => continue,
                    _ = tokio::time::sleep_until(deadline.into()) => break,
                }
            }

            if batch.is_empty() {
                continue;
            }

            // Write all entries and fsync once
            let entries: Vec<WalEntry> = batch.iter().map(|p| {
                WalEntry { data: p.entry.data.clone() }
            }).collect();

            let result = wal_writer.write_batch_and_fsync(&entries).await;

            // Notify all waiters
            match result {
                Ok(lsn) => {
                    for (i, pending) in batch.into_iter().enumerate() {
                        let _ = pending.done.send(Ok(lsn + i as u64));
                    }
                }
                Err(e) => {
                    let err_msg = e.to_string();
                    for pending in batch {
                        let _ = pending.done.send(Err(WalError::Io(
                            std::io::Error::new(std::io::ErrorKind::Other, err_msg.clone()),
                        )));
                    }
                }
            }
        }
    }
}

Connection Pool Limiter

Use a semaphore to limit concurrent connections or outstanding I/O:

use tokio::sync::Semaphore;
use std::sync::Arc;

pub struct ConnectionPool {
    semaphore: Arc<Semaphore>,
    max_connections: usize,
}

impl ConnectionPool {
    pub fn new(max_connections: usize) -> Self {
        Self {
            semaphore: Arc::new(Semaphore::new(max_connections)),
            max_connections,
        }
    }

    pub async fn acquire(&self) -> Result<ConnectionGuard, PoolError> {
        let permit = self.semaphore.clone().acquire_owned().await
            .map_err(|_| PoolError::Closed)?;

        let conn = establish_connection().await?;
        Ok(ConnectionGuard { conn, _permit: permit })
    }

    pub fn available(&self) -> usize {
        self.semaphore.available_permits()
    }

    pub fn in_use(&self) -> usize {
        self.max_connections - self.semaphore.available_permits()
    }
}

pub struct ConnectionGuard {
    conn: Connection,
    _permit: tokio::sync::OwnedSemaphorePermit, // released on drop
}

Config Broadcast with watch

When configuration changes must propagate to all tasks:

use tokio::sync::watch;
use std::sync::Arc;

#[derive(Debug, Clone)]
pub struct RuntimeConfig {
    pub max_batch_size: usize,
    pub flush_interval_ms: u64,
    pub compression_level: i32,
}

pub struct ConfigManager {
    tx: watch::Sender<Arc<RuntimeConfig>>,
}

impl ConfigManager {
    pub fn new(initial: RuntimeConfig) -> (Self, watch::Receiver<Arc<RuntimeConfig>>) {
        let (tx, rx) = watch::channel(Arc::new(initial));
        (Self { tx }, rx)
    }

    pub fn update(&self, new_config: RuntimeConfig) {
        let _ = self.tx.send(Arc::new(new_config));
        // All receivers are immediately notified
    }

    pub fn subscribe(&self) -> watch::Receiver<Arc<RuntimeConfig>> {
        self.tx.subscribe()
    }
}

/// In a worker task:
async fn worker(mut config_rx: watch::Receiver<Arc<RuntimeConfig>>) {
    let mut config = config_rx.borrow_and_update().clone();

    loop {
        tokio::select! {
            _ = config_rx.changed() => {
                config = config_rx.borrow_and_update().clone();
                tracing::info!(?config, "config updated");
            }
            _ = do_work(&config) => {}
        }
    }
}

Fan-Out Events with broadcast

When multiple consumers need every event (replication subscribers, change data capture):

use tokio::sync::broadcast;

#[derive(Debug, Clone)]
pub enum ChangeEvent {
    Insert { table: String, key: Vec<u8> },
    Delete { table: String, key: Vec<u8> },
    SchemaChange { table: String },
}

pub struct EventBus {
    tx: broadcast::Sender<ChangeEvent>,
}

impl EventBus {
    pub fn new(capacity: usize) -> Self {
        let (tx, _) = broadcast::channel(capacity);
        Self { tx }
    }

    pub fn publish(&self, event: ChangeEvent) {
        // Returns Err if there are no active receivers — that's fine
        let _ = self.tx.send(event);
    }

    pub fn subscribe(&self) -> broadcast::Receiver<ChangeEvent> {
        self.tx.subscribe()
    }
}

/// Replication subscriber:
async fn replication_stream(mut rx: broadcast::Receiver<ChangeEvent>) {
    loop {
        match rx.recv().await {
            Ok(event) => replicate(event).await,
            Err(broadcast::error::RecvError::Lagged(n)) => {
                tracing::warn!(skipped = n, "replication subscriber lagged — events dropped");
                // Must handle this: trigger a full resync or accept data loss
            }
            Err(broadcast::error::RecvError::Closed) => break,
        }
    }
}

Synchronization Primitives Cheat Sheet

Ordering Quick Reference

Rule of thumb: start with SeqCst for correctness, downgrade to Acquire/Release after profiling shows contention, use Relaxed only for statistics where stale reads are acceptable.

Testing Concurrent Code

loom — Deterministic Concurrency Testing

loom explores all possible thread interleavings to find race conditions that stress tests miss:

#[cfg(test)]
mod tests {
    #[test]
    fn test_concurrent_insert() {
        loom::model(|| {
            use loom::sync::Arc;
            use loom::sync::atomic::{AtomicUsize, Ordering};

            let counter = Arc::new(AtomicUsize::new(0));
            let c1 = counter.clone();
            let c2 = counter.clone();

            let t1 = loom::thread::spawn(move || {
                c1.fetch_add(1, Ordering::SeqCst);
            });

            let t2 = loom::thread::spawn(move || {
                c2.fetch_add(1, Ordering::SeqCst);
            });

            t1.join().unwrap();
            t2.join().unwrap();

            assert_eq!(counter.load(Ordering::SeqCst), 2);
        });
    }
}

Setup: loom replaces std::sync and std::thread with its own versions. Use feature flags to swap implementations:

#[cfg(not(loom))]
use std::sync::atomic::AtomicU64;

#[cfg(loom)]
use loom::sync::atomic::AtomicU64;

tokio::test — Async Test Runtime

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_snapshot_isolation() {
        let manager = SnapshotManager::new();

        // Writer creates version 1
        let txn1 = manager.next_txn_id();

        // Reader takes snapshot — should see version 1
        let snap = manager.acquire_snapshot();

        // Writer creates version 2 (after snapshot)
        let txn2 = manager.next_txn_id();

        // Snapshot should see txn1 but not txn2
        assert!(manager.is_visible(txn1, &snap));
        assert!(!manager.is_visible(txn2, &snap));

        manager.release_snapshot(&snap);
    }

    /// Multi-threaded tokio test — spawns actual worker threads.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_concurrent_memtable_writes() {
        let memtable = Arc::new(Memtable::new());
        let mut handles = Vec::new();

        for i in 0..100 {
            let mt = memtable.clone();
            handles.push(tokio::spawn(async move {
                let key = Bytes::from(format!("key-{i:04}"));
                let value = Bytes::from(format!("value-{i}"));
                mt.insert(key, value);
            }));
        }

        for handle in handles {
            handle.await.unwrap();
        }

        assert_eq!(memtable.approximate_size(), /* expected */);
    }
}

Stress Tests — Verify Invariants Under Load

#[cfg(test)]
mod stress_tests {
    use super::*;
    use std::sync::Arc;
    use std::time::Duration;

    /// Hammer the double-buffered memtable with concurrent writes and flushes.
    /// Invariant: no writes are lost, reads always return either the current
    /// or the just-flushed version.
    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
    async fn stress_double_buffer() {
        let buffer = Arc::new(DoubleBufferedMemtable::new());
        let total_writes = Arc::new(AtomicUsize::new(0));

        // Spawn 10 writer tasks
        let mut writer_handles = Vec::new();
        for writer_id in 0..10 {
            let buf = buffer.clone();
            let writes = total_writes.clone();
            writer_handles.push(tokio::spawn(async move {
                for i in 0..1000 {
                    let key = Bytes::from(format!("w{writer_id}-{i:06}"));
                    let value = Bytes::from(vec![0xABu8; 64]);
                    buf.insert(key, value);
                    writes.fetch_add(1, Ordering::Relaxed);
                    // Small random delay to increase interleaving
                    if i % 100 == 0 {
                        tokio::task::yield_now().await;
                    }
                }
            }));
        }

        // Spawn a flusher that periodically swaps buffers
        let flush_buf = buffer.clone();
        let flush_handle = tokio::spawn(async move {
            for _ in 0..5 {
                tokio::time::sleep(Duration::from_millis(50)).await;
                let old = flush_buf.swap_for_flush();
                // Simulate flush delay
                tokio::time::sleep(Duration::from_millis(10)).await;
                // In real code: write old to disk, then clear
            }
        });

        for handle in writer_handles {
            handle.await.unwrap();
        }
        flush_handle.await.unwrap();

        assert_eq!(total_writes.load(Ordering::Relaxed), 10_000);
    }
}

Key Dependencies

[dependencies]
tokio = { version = "1.43", features = ["rt-multi-thread", "io-util", "net", "sync", "macros", "signal", "fs", "time"] }
tokio-util = { version = "0.7", features = ["rt"] }  # CancellationToken
crossbeam = "0.8"                      # Umbrella: channels, epoch, queue, skiplist
crossbeam-skiplist = "0.1"             # SkipMap — lock-free sorted map
crossbeam-channel = "0.5"             # Bounded/unbounded channels, select!
crossbeam-epoch = "0.9"               # Epoch-based memory reclamation
crossbeam-queue = "0.3"               # SegQueue, ArrayQueue
parking_lot = "0.12"                   # Fast Mutex/RwLock — no poisoning
dashmap = "6"                          # Concurrent hash map — sharded locks
arc-swap = "1"                         # Lock-free Arc pointer swap
num_cpus = "1"                         # Detect physical/logical core count
futures = "0.3"                        # Stream, FutureExt, StreamExt

[target.'cfg(target_os = "linux")'.dependencies]
io-uring = { version = "0.7", optional = true }

[dev-dependencies]
loom = "0.7"                           # Deterministic concurrency testing