From d0aa8e0e20b1e7c12ec79f64c20a14c857270186 Mon Sep 17 00:00:00 2001 From: Henry Su Date: Mon, 26 Oct 2026 11:15:40 -0500 Subject: [PATCH] perf(core): write the FTS index once per ingest, in rowid order FTS5 flushes its pending terms at each statement that writes it, and ingest inserts one search row per statement. The AFTER INSERT trigger on search_document therefore wrote one index segment per row. Migration 0015 drops that trigger. persist_rows and append_rows now read the highest search row id after their deletes, and index every row above it with one INSERT ... SELECT ... ORDER BY id at the end. The DELETE and UPDATE triggers stay: each delete is one statement, so it already flushes once, and nothing updates search_document. --- .../lore-core/migrations/0015_batch_fts.sql | 14 + crates/lore-core/src/ingest.rs | 34 +- crates/lore-core/src/storage/migrations.rs | 5 + crates/lore-core/tests/fts_sync.rs | 303 ++++++++++++++++++ crates/lore-core/tests/incremental_ingest.rs | 7 +- crates/lore-core/tests/schema.rs | 14 +- 6 files changed, 367 insertions(+), 10 deletions(-) create mode 100644 crates/lore-core/migrations/0015_batch_fts.sql create mode 100644 crates/lore-core/tests/fts_sync.rs diff --git a/crates/lore-core/migrations/0015_batch_fts.sql b/crates/lore-core/migrations/0015_batch_fts.sql new file mode 100644 index 0000000..8cae1ef --- /dev/null +++ b/crates/lore-core/migrations/0015_batch_fts.sql @@ -0,0 +1,14 @@ +-- Migration 0015: ingest indexes its new search rows itself, once per +-- transaction. +-- +-- FTS5 flushes its pending terms to disk at each statement that writes it. +-- Ingest inserts search rows one statement at a time, so the AFTER INSERT +-- trigger wrote one index segment per row. Ingest now inserts a transaction's +-- new rows into search_fts with one statement, in rowid order +-- (`crate::ingest`). +-- +-- The DELETE and UPDATE triggers stay. Every delete removes its rows in one +-- statement, so the DELETE trigger already flushes once per delete, and +-- nothing updates search_document. The existing index is already in sync, so +-- this migration changes only how later rows reach it. +DROP TRIGGER search_document_ai; diff --git a/crates/lore-core/src/ingest.rs b/crates/lore-core/src/ingest.rs index 5ae3613..786a695 100644 --- a/crates/lore-core/src/ingest.rs +++ b/crates/lore-core/src/ingest.rs @@ -110,6 +110,7 @@ fn persist_rows( "DELETE FROM agent_session WHERE id = ?1", params![session_id], )?; + let high_water = search_high_water(tx)?; // Deleting the logical session above cascades its source links. Preserve // links from copies/continued artifacts so replaying one source does not @@ -164,9 +165,35 @@ fn persist_rows( patch_blobs, RowBase::default(), )?; + index_search_rows(tx, high_water)?; Ok(session_id) } +/// The highest search row id. Rows inserted after reading it get higher ids, +/// since a new INTEGER PRIMARY KEY is one past the largest. Read it after the +/// transaction's deletes, which can free the largest id for reuse. +fn search_high_water(tx: &Connection) -> Result { + Ok(tx.query_row( + "SELECT coalesce(max(id), 0) FROM search_document", + [], + |r| r.get(0), + )?) +} + +/// Index the search rows above `high_water` with one statement, in rowid +/// order. FTS5 flushes at each statement that writes it, and ingest inserts +/// one search row per statement, so migration 0015 dropped the insert trigger +/// that indexed each row as it landed. A rowid below the last one also forces +/// a flush, hence the order. +fn index_search_rows(tx: &Connection, high_water: i64) -> Result<()> { + tx.execute( + "INSERT INTO search_fts(rowid, redacted_text) + SELECT id, redacted_text FROM search_document WHERE id > ?1 ORDER BY id", + [high_water], + )?; + Ok(()) +} + /// A native title (custom/ai) is distinct content, so it is scanned and /// indexed like any cleartext. A synthetic fallback title merely echoes the /// first user message — already scanned and indexed — so re-scanning or @@ -552,7 +579,8 @@ fn append_rows( )?; // Only a native title is projected, so only a new one changes the index. - if from.native_title != end.native_title { + let title_changed = from.native_title != end.native_title; + if title_changed { tx.execute( "DELETE FROM search_document WHERE session_id = ?1 AND source_kind = 'session' AND source_id = ?1 @@ -564,6 +592,9 @@ fn append_rows( WHERE source_kind = 'session' AND source_id = ?1 AND field = 'title'", [session_id], )?; + } + let high_water = search_high_water(tx)?; + if title_changed { project_title(tx, session_id, agent_id, parsed)?; } @@ -579,6 +610,7 @@ fn append_rows( file_events: from.file_event_count, }, )?; + index_search_rows(tx, high_water)?; for result in &append.earlier_results { let (seq, ordinal) = result.result_ref; diff --git a/crates/lore-core/src/storage/migrations.rs b/crates/lore-core/src/storage/migrations.rs index ff0b52b..ff22748 100644 --- a/crates/lore-core/src/storage/migrations.rs +++ b/crates/lore-core/src/storage/migrations.rs @@ -88,6 +88,11 @@ const MIGRATIONS: &[Migration] = &[ name: "ingest_resume", sql: include_str!("../../migrations/0014_ingest_resume.sql"), }, + Migration { + version: 15, + name: "batch_fts", + sql: include_str!("../../migrations/0015_batch_fts.sql"), + }, ]; /// How many migrations exist. Tests assert against this rather than a literal diff --git a/crates/lore-core/tests/fts_sync.rs b/crates/lore-core/tests/fts_sync.rs new file mode 100644 index 0000000..16680d8 --- /dev/null +++ b/crates/lore-core/tests/fts_sync.rs @@ -0,0 +1,303 @@ +//! Ingest writes the FTS index itself, once per transaction (migration 0015). +//! After every way rows reach or leave `search_document`, a search finds +//! exactly the rows it holds, and FTS5 agrees its index matches that table. +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use lore_core::adapters::claude_code::ClaudeCodeAdapter; +use lore_core::forget::forget_session; +use lore_core::ingest::{ingest_file, ChangeClass, IngestOutcome}; +use lore_core::storage::blob::BlobStore; +use rusqlite::Connection; +use serde_json::json; + +/// The text of each row the index returns for `word`, in rowid order. `None` +/// is an index entry whose row is gone. +fn hits(conn: &Connection, word: &str) -> Vec> { + let mut stmt = conn + .prepare( + "SELECT search_document.redacted_text FROM search_fts + LEFT JOIN search_document ON search_document.id = search_fts.rowid + WHERE search_fts MATCH ?1 ORDER BY search_fts.rowid", + ) + .unwrap(); + let rows = stmt + .query_map([word], |r| r.get(0)) + .unwrap() + .collect::>>() + .unwrap(); + rows +} + +fn hit(text: &str) -> Vec> { + vec![Some(text.to_owned())] +} + +/// FTS5's integrity check. The `rank = 1` form also compares the index with +/// the external content table; the plain form only checks the index itself. +fn integrity(conn: &Connection) -> rusqlite::Result<()> { + conn.execute_batch( + "INSERT INTO search_fts(search_fts) VALUES('integrity-check'); + INSERT INTO search_fts(search_fts, rank) VALUES('integrity-check', 1);", + ) +} + +fn line(session: &str, event: serde_json::Value) -> String { + let mut v = json!({"sessionId": session, "cwd": "/repo/app", + "timestamp": "2026-08-10T10:00:00.000Z"}); + v.as_object_mut() + .unwrap() + .extend(event.as_object().unwrap().clone()); + format!("{v}\n") +} + +fn user(session: &str, uuid: &str, text: &str) -> String { + line( + session, + json!({"type": "user", "uuid": uuid, "parentUuid": null, + "message": {"role": "user", "content": text}}), + ) +} + +fn assistant(session: &str, uuid: &str, parent: &str, text: &str) -> String { + line( + session, + json!({"type": "assistant", "uuid": uuid, "parentUuid": parent, + "message": {"role": "assistant", "model": "claude-x", + "content": [{"type": "text", "text": text}]}}), + ) +} + +fn custom_title(session: &str, title: &str) -> String { + line( + session, + json!({"type": "custom-title", "customTitle": title}), + ) +} + +struct Archive { + _dir: tempfile::TempDir, + root: std::path::PathBuf, + blobs: BlobStore, + conn: Connection, +} + +impl Archive { + fn new() -> Self { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().to_path_buf(); + let blobs = BlobStore::open(root.join("blobs")).unwrap(); + let conn = lore_core::storage::open_in_memory().unwrap(); + conn.execute_batch( + "CREATE TEMP TABLE replaced (session_id TEXT); + CREATE TEMP TRIGGER log_replace AFTER DELETE ON agent_session + BEGIN INSERT INTO replaced VALUES (old.id); END;", + ) + .unwrap(); + Archive { + _dir: dir, + root, + blobs, + conn, + } + } + + /// Write `content` to `file` and ingest it. Returns the session id, the + /// change class, and whether ingest replaced an existing session. + fn ingest(&self, file: &str, content: &str) -> (String, ChangeClass, bool) { + let path = self.root.join(file); + std::fs::write(&path, content).unwrap(); + self.conn.execute_batch("DELETE FROM replaced").unwrap(); + let outcome = ingest_file(&self.conn, &ClaudeCodeAdapter::new(), &path, &self.blobs); + let replaced: i64 = self + .conn + .query_row("SELECT count(*) FROM replaced", [], |r| r.get(0)) + .unwrap(); + let (session_id, change) = match outcome.unwrap() { + IngestOutcome::Ingested { + session_id, change, .. + } => Some((session_id, change)), + IngestOutcome::Skipped => None, + } + .expect("a changed file must be ingested"); + (session_id, change, replaced > 0) + } +} + +#[test] +fn the_content_check_catches_a_row_missing_from_the_index() { + let archive = Archive::new(); + let (session, _, _) = archive.ingest("a.jsonl", &user("s-a", "u1", "Tune the quokka cache")); + integrity(&archive.conn).unwrap(); + archive + .conn + .execute( + "INSERT INTO search_document + (session_id, source_kind, source_id, field, redacted_text, created_at) + VALUES (?1, 'session', ?1, 'title', 'An unindexed wombat', 1)", + [&session], + ) + .unwrap(); + assert_eq!(hits(&archive.conn, "wombat"), Vec::>::new()); + assert!(integrity(&archive.conn).is_err()); +} + +#[test] +fn search_matches_the_archive_through_ingest_append_replace_and_forget() { + let archive = Archive::new(); + let first = user("s-a", "u1", "Tune the quokka cache"); + let second = assistant("s-a", "a1", "u1", "The wombat route is cold"); + let third = user("s-a", "u2", "Restart the numbat worker"); + + let (session, change, _) = archive.ingest("a.jsonl", &format!("{first}{second}")); + assert_eq!(change, ChangeClass::New); + assert_eq!(hits(&archive.conn, "quokka"), hit("Tune the quokka cache")); + assert_eq!( + hits(&archive.conn, "wombat"), + hit("The wombat route is cold") + ); + integrity(&archive.conn).unwrap(); + + let (appended, change, replaced) = + archive.ingest("a.jsonl", &format!("{first}{second}{third}")); + assert_eq!( + (appended.as_str(), change, replaced), + (session.as_str(), ChangeClass::Appended, false) + ); + assert_eq!(hits(&archive.conn, "quokka"), hit("Tune the quokka cache")); + assert_eq!( + hits(&archive.conn, "wombat"), + hit("The wombat route is cold") + ); + assert_eq!( + hits(&archive.conn, "numbat"), + hit("Restart the numbat worker") + ); + integrity(&archive.conn).unwrap(); + + let rewritten = user("s-a", "u1", "Tune the platypus cache"); + let (replaced_id, change, replaced) = + archive.ingest("a.jsonl", &format!("{rewritten}{second}{third}")); + assert_eq!( + (replaced_id.as_str(), change, replaced), + (session.as_str(), ChangeClass::Rewritten, true) + ); + assert_eq!(hits(&archive.conn, "quokka"), Vec::>::new()); + assert_eq!( + hits(&archive.conn, "platypus"), + hit("Tune the platypus cache") + ); + assert_eq!( + hits(&archive.conn, "wombat"), + hit("The wombat route is cold") + ); + assert_eq!( + hits(&archive.conn, "numbat"), + hit("Restart the numbat worker") + ); + integrity(&archive.conn).unwrap(); + + archive.ingest("b.jsonl", &user("s-b", "u1", "Plant the kiwi seeds")); + forget_session(&archive.conn, &archive.blobs, &session).unwrap(); + for word in ["platypus", "wombat", "numbat"] { + assert_eq!( + hits(&archive.conn, word), + Vec::>::new(), + "{word}" + ); + } + assert_eq!(hits(&archive.conn, "kiwi"), hit("Plant the kiwi seeds")); + integrity(&archive.conn).unwrap(); +} + +#[test] +fn a_source_rewritten_into_another_session_drops_the_old_sessions_rows() { + let archive = Archive::new(); + let (old, _, _) = archive.ingest("a.jsonl", &user("s-a", "u1", "Tune the quokka cache")); + let (new, change, _) = archive.ingest("a.jsonl", &user("s-z", "u1", "Tune the platypus cache")); + assert_ne!(new, old); + assert_eq!(change, ChangeClass::Rewritten); + assert_eq!(hits(&archive.conn, "quokka"), Vec::>::new()); + assert_eq!( + hits(&archive.conn, "platypus"), + hit("Tune the platypus cache") + ); + integrity(&archive.conn).unwrap(); +} + +/// The title is the archive's highest search row here, so deleting it frees +/// that id for the new title an append writes. +#[test] +fn an_appended_title_change_indexes_the_new_title() { + let archive = Archive::new(); + let thinking = line( + "s-t", + json!({"type": "assistant", "uuid": "a1", "parentUuid": null, + "message": {"role": "assistant", "model": "claude-x", + "content": [{"type": "thinking", "thinking": "hidden", "signature": "sig"}]}}), + ); + let head = format!("{}{thinking}", custom_title("s-t", "Quokka plans")); + archive.ingest("t.jsonl", &head); + assert_eq!(hits(&archive.conn, "quokka"), hit("Quokka plans")); + + let tail = format!( + "{}{}", + custom_title("s-t", "Numbat plans"), + user("s-t", "u1", "Feed the wombat") + ); + let (_, change, replaced) = archive.ingest("t.jsonl", &format!("{head}{tail}")); + assert_eq!((change, replaced), (ChangeClass::Appended, false)); + assert_eq!(hits(&archive.conn, "quokka"), Vec::>::new()); + assert_eq!(hits(&archive.conn, "numbat"), hit("Numbat plans")); + assert_eq!(hits(&archive.conn, "wombat"), hit("Feed the wombat")); + integrity(&archive.conn).unwrap(); +} + +/// A schema-14 archive is this schema plus the per-row insert trigger, with an +/// index already in sync. Migration 0015 must leave it in sync, and later +/// writes must keep it so. +#[test] +fn an_archive_indexed_by_the_insert_trigger_stays_in_sync_after_migration_0015() { + let archive = Archive::new(); + let quokka = user("s-a", "u1", "Tune the quokka cache"); + archive.ingest("a.jsonl", &quokka); + archive + .conn + .execute_batch( + "CREATE TRIGGER search_document_ai AFTER INSERT ON search_document BEGIN + INSERT INTO search_fts(rowid, redacted_text) VALUES (new.id, new.redacted_text); + END; + DELETE FROM schema_migrations WHERE version = 15; + INSERT INTO agent (id, display_name) VALUES ('old', 'Old'); + INSERT INTO agent_session (id, agent_id, dedupe_key) VALUES ('s-old', 'old', 'k'); + INSERT INTO search_document + (session_id, source_kind, source_id, field, redacted_text, created_at) + VALUES ('s-old', 'session', 's-old', 'title', 'Migrate the axolotl ledger', 1);", + ) + .unwrap(); + assert_eq!( + hits(&archive.conn, "axolotl"), + hit("Migrate the axolotl ledger") + ); + integrity(&archive.conn).unwrap(); + + lore_core::storage::migrations::run(&archive.conn).unwrap(); + assert_eq!(hits(&archive.conn, "quokka"), hit("Tune the quokka cache")); + assert_eq!( + hits(&archive.conn, "axolotl"), + hit("Migrate the axolotl ledger") + ); + integrity(&archive.conn).unwrap(); + + archive.ingest("b.jsonl", &user("s-b", "u1", "Plant the kiwi seeds")); + assert_eq!(hits(&archive.conn, "kiwi"), hit("Plant the kiwi seeds")); + let (_, change, _) = archive.ingest("a.jsonl", &user("s-a", "u1", "Tune the platypus cache")); + assert_eq!(change, ChangeClass::Rewritten); + assert_eq!(hits(&archive.conn, "quokka"), Vec::>::new()); + assert_eq!( + hits(&archive.conn, "platypus"), + hit("Tune the platypus cache") + ); + forget_session(&archive.conn, &archive.blobs, "s-old").unwrap(); + assert_eq!(hits(&archive.conn, "axolotl"), Vec::>::new()); + integrity(&archive.conn).unwrap(); +} diff --git a/crates/lore-core/tests/incremental_ingest.rs b/crates/lore-core/tests/incremental_ingest.rs index 4d8b649..b52aa0f 100644 --- a/crates/lore-core/tests/incremental_ingest.rs +++ b/crates/lore-core/tests/incremental_ingest.rs @@ -168,8 +168,11 @@ fn grow_and_compare( .unwrap(); kept.push(replaced == 0); } - live.execute_batch("INSERT INTO search_fts(search_fts) VALUES('integrity-check');") - .unwrap(); + // `rank = 1` also checks the index against search_document. + live.execute_batch( + "INSERT INTO search_fts(search_fts, rank) VALUES('integrity-check', 1);", + ) + .unwrap(); let fresh = lore_core::storage::open_in_memory().unwrap(); let fresh_id = session_id(ingest_file(&fresh, adapter, &path, &blobs).unwrap()); diff --git a/crates/lore-core/tests/schema.rs b/crates/lore-core/tests/schema.rs index 55059e2..83baaae 100644 --- a/crates/lore-core/tests/schema.rs +++ b/crates/lore-core/tests/schema.rs @@ -3,8 +3,6 @@ //! external-content FTS5 index tracks `search_document`. #![allow(clippy::unwrap_used, clippy::expect_used)] -use rusqlite::params; - fn db() -> rusqlite::Connection { lore_core::storage::open_in_memory().unwrap() } @@ -180,14 +178,16 @@ fn session_uniqueness_null_semantics() { } #[test] -fn fts_tracks_search_document() { +fn deleting_a_search_document_de_indexes_it() { let conn = db(); seed_session(&conn); - conn.execute( + // Ingest indexes new rows itself (migration 0015), as here. + conn.execute_batch( "INSERT INTO search_document (session_id, source_kind, source_id, field, redacted_text, created_at) - VALUES ('s1','message_part','p1','text','fix the stripe webhook signature',1)", - params![], + VALUES ('s1','message_part','p1','text','fix the stripe webhook signature',1); + INSERT INTO search_fts(rowid, redacted_text) + SELECT id, redacted_text FROM search_document;", ) .unwrap(); let hits: i64 = conn @@ -197,7 +197,7 @@ fn fts_tracks_search_document() { |r| r.get(0), ) .unwrap(); - assert_eq!(hits, 1, "inserting a search_document must index it in FTS5"); + assert_eq!(hits, 1); // Deleting the document must remove it from the FTS index too. conn.execute("DELETE FROM search_document WHERE source_id = 'p1'", [])