rust-duckdb-cache-read-through-reconcile

Answer

The bug: the DuckDB edge cache is only as fresh as the last successful write path. If edges.jsonl was appended externally, or a write crashed/ran on an older binary that skipped the cache, the cache silently drifts from the canonical log. Since all 14 read entry points (7 CLI + 7 MCP) resolve to a GraphDB handle opened by GraphDB::open, reconciling once at open time repairs every read path with one change.

1. The free function — reconcile_edges_from_jsonl

// src/cache.rs (or src/reconcile.rs)
use duckdb::{params, Connection, Result};
use serde_json::Value;
use std::path::Path;

/// Reconcile the DuckDB edge cache against the canonical edges.jsonl log.
///
/// Reads every line of `edges.jsonl` and `INSERT OR IGNORE`s each valid edge
/// into the cache. Edges lost when the write path was skipped (crash before
/// flush, older binary, external append) are repaired here at open time.
///
/// Guarantees:
///   - Idempotent: re-running is a no-op beyond the first pass.
///   - Malformed / truncated JSON lines are skipped (a crash mid-append is the
///     common case) — never aborts the whole reconcile.
///   - Missing edges.jsonl is not an error: returns Ok(0).
///   - Existing cache rows win: INSERT OR IGNORE never clobbers local state.
pub fn reconcile_edges_from_jsonl(conn: &Connection, path: &Path) -> Result<usize> {
    let raw = match std::fs::read(path) {
        Ok(b) => b,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(e) => return Err(io_err(e)),
    };
    // Strip a UTF-8 BOM if present; tolerate CRLF.
    let raw = raw.strip_prefix(b"\xef\xbb\xbf").unwrap_or(&raw);

    let mut stmt = conn.prepare(
        "INSERT OR IGNORE INTO edges (source, target, edge_type, weight, metadata)
         VALUES (?1, ?2, ?3, ?4, ?5)",
    )?;
    // One transaction: a big log reconciles in O(1) commits, and a failed run
    // rolls back atomically so the cache is never left half-reconciled.
    let tx = conn.unchecked_transaction()?;

    let mut inserted = 0usize;
    for line in raw.split(|&b| b == b'\n') {
        let line = std::str::from_utf8(line).unwrap_or("").trim();
        if line.is_empty() {
            continue; // trailing newline / blank lines
        }
        // Parse as a JSON object; any malformed or non-object line is skipped.
        let Ok(Value::Object(m)) = serde_json::from_str::<Value>(line) else {
            continue;
        };
        // Missing required fields => not a usable edge, skip.
        let (Some(source), Some(target)) = (m.get("source").and_then(Value::as_str),
                                           m.get("target").and_then(Value::as_str)) else {
            continue;
        };
        let edge_type = m.get("edge_type").and_then(Value::as_str).unwrap_or("default");
        let weight = m.get("weight").and_then(Value::as_f64).unwrap_or(1.0);
        let metadata = m.get("metadata").map(Value::to_string);
        let n = stmt.execute(params![source, target, edge_type, weight, metadata])?;
        inserted += n; // INSERT OR IGNORE -> 0 when the row already exists
    }

    tx.commit()?;
    Ok(inserted)
}

fn io_err(e: std::io::Error) -> duckdb::Error {
    duckdb::Error::InvalidPath(e.to_string().into())
}

The INSERT OR IGNORE depends on a uniqueness constraint on (source, target, edge_type) — the natural graph-edge key — declared in the schema:

CREATE TABLE IF NOT EXISTS edges (
    source    VARCHAR NOT NULL,
    target    VARCHAR NOT NULL,
    edge_type VARCHAR NOT NULL DEFAULT 'default',
    weight    DOUBLE  NOT NULL DEFAULT 1.0,
    metadata  JSON,
    PRIMARY KEY (source, target, edge_type)
);

2. Wiring into GraphDB::open — on-disk only

// src/graph_db.rs
use std::path::{Path, PathBuf};

pub struct GraphDB {
    conn: Connection,
    cache_path: Option<PathBuf>, // None for :memory: / in-memory graphs
    edges_jsonl: PathBuf,
}

impl GraphDB {
    pub fn open(path: &str) -> Result<Self> {
        let conn = Connection::open(path)?;
        let db = Self {
            conn,
            cache_path: (path != ":memory:").then(|| PathBuf::from(path)),
            edges_jsonl: edges_jsonl_for(path), // canonical sibling, e.g. "<db>.edges.jsonl"
        };
        db.init_schema()?; // CREATE TABLE IF NOT EXISTS edges (… PRIMARY KEY …)

        // Reconcile ONLY for on-disk caches. :memory: has no edges.jsonl to
        // drift from and is rebuilt from scratch per process — skip it.
        if db.cache_path.is_some() {
            match reconcile_edges_from_jsonl(&db.conn, &db.edges_jsonl) {
                Ok(n) if n > 0 => log::info!("reconciled {n} drifted edges from edges.jsonl"),
                Ok(_) => {}
                Err(e) => return Err(e), // hard-fail: never serve a known-drifted cache
            }
        }
        Ok(db)
    }
}

Because GraphDB::open is the single funnel for both CLI commands and the 7 MCP tools (all 14 read entry points take a &GraphDB / &Connection obtained from it), this one call site fixes all of them. No entry-point-specific plumbing, no per-read reconciliation overhead.

3. Why this design

• Open-time, not read-time: reconciliation is O(log size) once, instead of a file read + parse on every query. Reads stay hot.
• INSERT OR IGNORE, not DELETE + reload: preserves any cache rows that are ahead of the log (e.g., un-flushed metadata) and any rows the log never had — the cache only ever gains missing canonical rows.
• Hard-fail on reconcile error (other than missing file): a cache that cannot be reconciled is known-drifted, so serving it silently would re-introduce the exact bug being fixed.
• Missing file = Ok(0): fresh checkouts and log-less deployments open cleanly.