rust-duckdb-cache-write-through

Answer

Root cause: parse_and_diff_sync appended new edges to the durable edges.jsonl log but never wrote them into the DuckDB graph cache, so every read path that hits DuckDB served stale data until a manual warm-up.

Fix (three moves, one SQL source of truth):

1. hilo-graph — extract ensure_schema() + insert_edges_into() as public free functions; GraphDB delegates.

// hilo-graph/src/lib.rs
use duckdb::{Connection, params};

/// Idempotently create the graph cache schema on *any* connection —
/// including raw connections opened without `GraphDB::open` (mount.rs case).
pub fn ensure_schema(conn: &Connection) -> duckdb::Result<()> {
    conn.execute_batch(
        "CREATE TABLE IF NOT EXISTS edges (
             id         TEXT PRIMARY KEY,
             source     TEXT NOT NULL,
             target     TEXT NOT NULL,
             label      TEXT NOT NULL,
             meta       TEXT,
             updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
         );
         CREATE INDEX IF NOT EXISTS idx_edges_source ON edges(source);
         CREATE INDEX IF NOT EXISTS idx_edges_target ON edges(target);
         CREATE INDEX IF NOT EXISTS idx_edges_label  ON edges(label);",
    )
}

/// Batch-insert edges into the graph cache.
/// `INSERT OR IGNORE` makes this idempotent: re-running the same
/// parse-and-diff (or a concurrent trigger firing twice) never duplicates rows.
pub fn insert_edges_into(conn: &Connection, edges: &[Edge]) -> duckdb::Result<()> {
    ensure_schema(conn)?; // raw conns are covered here — no separate bootstrap
    let mut stmt = conn.prepare(
        "INSERT OR IGNORE INTO edges (id, source, target, label, meta)
         VALUES (?1, ?2, ?3, ?4, ?5)",
    )?;
    for e in edges {
        stmt.execute(params![e.id, e.source, e.target, e.label, e.meta])?;
    }
    Ok(())
}

pub struct GraphDB { conn: Connection }

impl GraphDB {
    pub fn open(path: &str) -> duckdb::Result<Self> {
        let conn = Connection::open(path)?;
        ensure_schema(&conn)?; // delegate — GraphDB holds no private copy of the DDL
        Ok(Self { conn })
    }

    /// Delegate — write paths through GraphDB use the same function as the trigger.
    pub fn insert_edges(&self, edges: &[Edge]) -> duckdb::Result<()> {
        insert_edges_into(&self.conn, edges)
    }

    pub fn query_edges(&self) -> duckdb::Result<Vec<Edge>> {
        let mut stmt = self.conn.prepare("SELECT id, source, target, label, meta FROM edges ORDER BY id")?;
        let rows = stmt.query_map([], |r| Ok(Edge {
            id: r.get(0)?, source: r.get(1)?, target: r.get(2)?,
            label: r.get(3)?, meta: r.get(4)?,
        }))?;
        rows.collect()
    }
}

2. hilo-triggers — write through in parse_and_diff_sync immediately after append_edges.

// hilo-triggers/src/engine.rs
use hilo_graph::{self, Edge};

pub fn parse_and_diff_sync(conn: &duckdb::Connection, log_path: &Path) -> anyhow::Result<()> {
    let new_edges: Vec<Edge> = parse_and_diff()?;
    if new_edges.is_empty() {
        return Ok(()); // nothing appended, nothing to cache
    }

    // 1. durable append log (replay source of truth)
    append_edges(log_path, &new_edges)?;

    // 2. write-through to the DuckDB graph cache — previously missing.
    //    Read paths now observe new edges immediately; no manual warm required.
    hilo_graph::insert_edges_into(conn, &new_edges)?;

    Ok(())
}

3. mount.rs raw-connection case is covered for free — a connection opened with Connection::open/open_in_memory (no GraphDB::open) still gets its schema via the ensure_schema() call inside insert_edges_into():

// hilo-triggers/src/mount.rs
let conn = duckdb::Connection::open_in_memory()?; // raw conn — no GraphDB::open
hilo_graph::insert_edges_into(&conn, &new_edges)?; // ensure_schema runs internally

4. New regression test.

// hilo-triggers/tests/write_through.rs (or inline in engine.rs tests)
#[test]
fn test_parse_diff_writes_through_to_duckdb_cache() {
    // Raw in-memory conn, deliberately NOT created via GraphDB::open (mount.rs path).
    let conn = duckdb::Connection::open_in_memory().unwrap();
    let edges = vec![
        Edge { id: "e1".into(), source: "a".into(), target: "b".into(), label: "depends_on".into(), meta: None },
        Edge { id: "e2".into(), source: "b".into(), target: "c".into(), label: "depends_on".into(), meta: None },
    ];

    // What the trigger does after append_edges():
    hilo_graph::insert_edges_into(&conn, &edges).unwrap();

    // Read path sees the rows immediately — no manual warm.
    let ids: Vec<String> = conn
        .prepare("SELECT id FROM edges ORDER BY id").unwrap()
        .query_map([], |r| r.get(0)).unwrap()
        .collect::<duckdb::Result<_>>().unwrap();
    assert_eq!(ids, vec!["e1".to_string(), "e2".to_string()]);

    // Idempotency: re-running the trigger with the same edges must not duplicate.
    hilo_graph::insert_edges_into(&conn, &edges).unwrap();
    let count: i64 = conn.query_row("SELECT COUNT(*) FROM edges", [], |r| r.get(0)).unwrap();
    assert_eq!(count, 2);
}