From cd337202ee8e3f54ccb2682ed9d8a4ce7ec03e89 Mon Sep 17 00:00:00 2001 From: Andreas Brenner Date: Thu, 14 May 2026 14:55:34 +0200 Subject: [PATCH] ANW-12 tantivy index: schema, upsert, rebuild on startup, scanner->writer wire --- Cargo.lock | 1 + Cargo.toml | 2 +- src/app.rs | 128 +++++++++++++++++---- src/index.rs | 306 +++++++++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 1 + src/vault.rs | 15 ++- 6 files changed, 424 insertions(+), 29 deletions(-) create mode 100644 src/index.rs diff --git a/Cargo.lock b/Cargo.lock index 9f9af3f..1931280 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -332,6 +332,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0" dependencies = [ "num-traits", + "serde", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index f780fc9..d719d6d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,7 @@ path = "src/main.rs" anyhow = "1" axum = "0.8" blake3 = "1" -chrono = { version = "0.4", default-features = false, features = ["std"] } +chrono = { version = "0.4", default-features = false, features = ["std", "serde"] } clap = { version = "4", features = ["derive", "env"] } hydra = "0.1" notify = "8" diff --git a/src/app.rs b/src/app.rs index 8d84117..77275a9 100644 --- a/src/app.rs +++ b/src/app.rs @@ -27,12 +27,15 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use hydra::{ - Application, ApplicationConfig, ChildSpec, ExitReason, From as HydraFrom, GenServer, + Application, ApplicationConfig, ChildSpec, Dest, ExitReason, From as HydraFrom, GenServer, GenServerOptions, Pid, SupervisionStrategy, Supervisor, SupervisorOptions, }; use serde::{Deserialize, Serialize}; -use crate::vault; +use crate::index::NoteIndex; +use crate::vault::{self, Note}; + +const INDEX_WRITER_NAME: &str = "index_writer"; /// Snapshot of per-process restart counters. Each role increments its own /// counter on every `init` *after the first*, so the count represents @@ -117,7 +120,17 @@ impl Application for Anwesen { } async fn start(&self) -> Result { + // Startup order matters: `vault_scanner` casts a `Rebuild` to the + // `index_writer` name at the end of its init. Hydra start_link is + // synchronous, so we put `index_writer` first to guarantee it is + // registered and in its message loop before `vault_scanner` runs. + // The `one_for_one` strategy makes the order irrelevant for restart + // semantics, only for the cold-start handshake. let children = [ + IndexWriter { + counters: self.counters.clone(), + } + .child_spec(), VaultScanner { vault: self.vault.clone(), counters: self.counters.clone(), @@ -127,10 +140,6 @@ impl Application for Anwesen { counters: self.counters.clone(), } .child_spec(), - IndexWriter { - counters: self.counters.clone(), - } - .child_spec(), HttpServer { bind: self.bind, counters: self.counters.clone(), @@ -154,6 +163,17 @@ pub enum VaultScannerMessage { RescanNow, } +fn call_not_supported(role: &str) -> Result, ExitReason> { + // Each role's GenServer trait shares one `Message` type for both call and + // cast (Hydra's API shape). To keep a stray `Foo::call(...)` from hanging + // forever waiting on a reply, every `handle_call` returns this error. + // Sie flagged the silent-Ok(None) pattern in ANW-17 review; pinning it + // here in ANW-12 before ANW-16 wires the watcher-to-scanner call. + Err(ExitReason::from(format!( + "{role} does not handle synchronous calls; use cast" + ))) +} + #[derive(Clone)] pub struct VaultScanner { vault: PathBuf, @@ -188,6 +208,13 @@ impl VaultScanner { "vault_scanner: skipped file" ); } + // Push the fresh record set to the index writer. The cast is + // address-by-name so we don't have to thread the index_writer Pid + // through child specs. + IndexWriterState::cast( + Dest::from(INDEX_WRITER_NAME), + IndexWriterMessage::Rebuild(result.notes), + ); } } @@ -216,8 +243,7 @@ impl GenServer for VaultScanner { _message: Self::Message, _from: HydraFrom, ) -> Result, ExitReason> { - // No synchronous protocol on vault_scanner; ignore. - Ok(None) + call_not_supported("vault_scanner") } } @@ -265,17 +291,26 @@ impl GenServer for FilesystemWatcher { _message: Self::Message, _from: HydraFrom, ) -> Result, ExitReason> { - Ok(None) + call_not_supported("filesystem_watcher") } } -// -- index_writer (stub for ANW-12) ----------------------------------------- +// -- index_writer ----------------------------------------------------------- +/// Messages handled by the [`IndexWriter`] `GenServer`. All variants are casts +/// (one-way fire-and-forget); calls return an error from `handle_call`. #[derive(Debug, Clone, Serialize, Deserialize)] pub enum IndexWriterMessage { - /// Placeholder; replaced by index upsert/delete in - /// [ANW-12](https://crvrs.youtrack.cloud/issue/ANW-12). - Noop, + /// Discard the current index and reindex the given notes. Sent by + /// `vault_scanner` at startup and after `rescan_now`. + Rebuild(Vec), + /// Insert-or-replace one note. Sent by `filesystem_watcher` + /// ([ANW-16](https://crvrs.youtrack.cloud/issue/ANW-16)) on + /// Create / Move-In / Modify / Close-Write events. + Upsert(Box), + /// Drop one note from the index. Sent by `filesystem_watcher` on + /// Delete / Move-Out events. + Delete(String), } #[derive(Clone)] @@ -286,25 +321,70 @@ pub struct IndexWriter { impl IndexWriter { fn child_spec(self) -> ChildSpec { let counters = self.counters.clone(); - ChildSpec::new("index_writer").start(move || { - IndexWriter { + ChildSpec::new(INDEX_WRITER_NAME).start(move || { + IndexWriterState { counters: counters.clone(), + index: None, } - .start_link(GenServerOptions::new().name("index_writer")) + .start_link(GenServerOptions::new().name(INDEX_WRITER_NAME)) }) } } -impl GenServer for IndexWriter { +/// Runtime state for the [`IndexWriter`] process. Held in a separate struct +/// so the `Clone`-friendly child-spec form (which doesn't carry the live +/// `NoteIndex`) stays simple. +struct IndexWriterState { + counters: Arc, + /// Created lazily in `init` so a Tantivy construction failure surfaces + /// as an `ExitReason` and triggers a supervisor restart, rather than + /// poisoning the child spec. + index: Option, +} + +impl GenServer for IndexWriterState { type Message = IndexWriterMessage; async fn init(&mut self) -> Result<(), ExitReason> { let restart = self.counters.index_writer.record_init(); - tracing::info!(restart, "index_writer: init (stub; ANW-12)"); + let index = NoteIndex::new() + .map_err(|e| ExitReason::from(format!("index_writer: NoteIndex::new failed: {e}")))?; + self.index = Some(index); + tracing::info!(restart, "index_writer: init"); Ok(()) } - async fn handle_cast(&mut self, _message: Self::Message) -> Result<(), ExitReason> { + async fn handle_cast(&mut self, message: Self::Message) -> Result<(), ExitReason> { + let Some(index) = self.index.as_mut() else { + return Err(ExitReason::from( + "index_writer: handle_cast invoked before init", + )); + }; + match message { + IndexWriterMessage::Rebuild(notes) => { + let count = notes.len(); + if let Err(e) = index.rebuild(¬es) { + tracing::error!(error = %e, "index_writer: rebuild failed"); + return Err(ExitReason::from(format!("rebuild: {e}"))); + } + tracing::info!(notes = count, "index_writer: rebuilt"); + } + IndexWriterMessage::Upsert(note) => { + let path = note.path.clone(); + if let Err(e) = index.upsert(¬e) { + tracing::error!(%path, error = %e, "index_writer: upsert failed"); + return Err(ExitReason::from(format!("upsert {path}: {e}"))); + } + tracing::debug!(%path, "index_writer: upserted"); + } + IndexWriterMessage::Delete(path) => { + if let Err(e) = index.delete(&path) { + tracing::error!(%path, error = %e, "index_writer: delete failed"); + return Err(ExitReason::from(format!("delete {path}: {e}"))); + } + tracing::debug!(%path, "index_writer: deleted"); + } + } Ok(()) } @@ -313,7 +393,7 @@ impl GenServer for IndexWriter { _message: Self::Message, _from: HydraFrom, ) -> Result, ExitReason> { - Ok(None) + call_not_supported("index_writer") } } @@ -351,7 +431,11 @@ impl GenServer for HttpServer { async fn init(&mut self) -> Result<(), ExitReason> { let restart = self.counters.http_server.record_init(); - tracing::info!(restart, bind = %self.bind, "http_server: init (stub; ANW-13)"); + tracing::warn!( + restart, + bind = %self.bind, + "http_server: stub -- no port bound yet (real binding lands in ANW-13)" + ); Ok(()) } @@ -364,7 +448,7 @@ impl GenServer for HttpServer { _message: Self::Message, _from: HydraFrom, ) -> Result, ExitReason> { - Ok(None) + call_not_supported("http_server") } } diff --git a/src/index.rs b/src/index.rs new file mode 100644 index 0000000..2f376c5 --- /dev/null +++ b/src/index.rs @@ -0,0 +1,306 @@ +//! In-memory Tantivy index over note frontmatter. +//! +//! Implements [[ADR-002 Tantivy as Frontmatter Index]] + the ANW-12 schema: +//! +//! - one `json` field (`frontmatter`) with the `raw` tokenizer pinned for v1 +//! (exact-match semantics on every key, matches the keyword-style operator +//! surface in the User Manual); +//! - one `string` field (`path`) with the `raw` tokenizer (exact lookup + +//! anchored prefix queries). +//! +//! The index lives in a [`RamDirectory`](tantivy::directory::RamDirectory) -- +//! rebuilt at startup from the scanner output and maintained incrementally +//! by [`upsert`](NoteIndex::upsert) / [`delete`](NoteIndex::delete) calls +//! driven from the filesystem watcher in [ANW-16]. + +use anyhow::{Context, Result}; +use serde_json::{Map as JsonMap, Value as JsonValue, json}; +use tantivy::schema::{Field, IndexRecordOption, Schema, TextFieldIndexing, TextOptions}; +use tantivy::{Index, IndexWriter, TantivyDocument, Term}; + +use crate::vault::{Note, Value}; + +/// 50 MB heap for the writer. Tantivy's documented minimum is 15 MB; this +/// sits comfortably above that for low-thousands-of-notes vaults without +/// bloating idle memory. +const WRITER_HEAP_BYTES: usize = 50 * 1024 * 1024; + +/// Strongly-typed handle to the two schema fields. Carried alongside the +/// [`Index`] so call sites don't restring field names. +#[derive(Debug, Clone, Copy)] +pub struct Fields { + pub path: Field, + pub frontmatter: Field, +} + +/// The in-memory note index. Owns the [`Index`] and a long-lived +/// [`IndexWriter`]; one `NoteIndex` per daemon. +pub struct NoteIndex { + fields: Fields, + index: Index, + writer: IndexWriter, +} + +impl NoteIndex { + /// Construct an empty in-memory index with the pinned v1 schema. + /// + /// # Errors + /// Returns the underlying Tantivy error if writer allocation fails (the + /// only failure mode reachable for a fresh `RamDirectory`). + pub fn new() -> Result { + let mut schema_builder = Schema::builder(); + + let raw_indexing = TextFieldIndexing::default() + .set_tokenizer("raw") + .set_index_option(IndexRecordOption::Basic); + + let path_options = TextOptions::default() + .set_indexing_options(raw_indexing.clone()) + .set_stored(); + + let path = schema_builder.add_text_field("path", path_options); + + // JSON field: open-schema frontmatter. `raw` tokenizer pins exact-match + // semantics on every key, per the open question closed in + // [[ADR-002 Tantivy as Frontmatter Index]]. + let json_options = tantivy::schema::JsonObjectOptions::default() + .set_indexing_options(raw_indexing) + .set_stored(); + let frontmatter = schema_builder.add_json_field("frontmatter", json_options); + + let schema = schema_builder.build(); + let index = Index::create_in_ram(schema); + let writer: IndexWriter = index + .writer(WRITER_HEAP_BYTES) + .context("create tantivy writer")?; + + Ok(Self { + fields: Fields { path, frontmatter }, + index, + writer, + }) + } + + /// Discard every document and reindex the given notes. Used at startup + /// and during overflow recovery per [[ADR-003 Filesystem Change Tracking]]. + /// + /// # Errors + /// Returns the underlying Tantivy error if any add/commit fails. + pub fn rebuild(&mut self, notes: &[Note]) -> Result<()> { + self.writer + .delete_all_documents() + .context("delete_all_documents")?; + for note in notes { + self.add_document(note)?; + } + self.writer.commit().context("commit rebuild")?; + Ok(()) + } + + /// Insert-or-replace one note by path. Subsequent reads through a fresh + /// reader will see the new content after this call returns. + /// + /// # Errors + /// Returns the underlying Tantivy error if the add or commit fails. + pub fn upsert(&mut self, note: &Note) -> Result<()> { + self.writer.delete_term(self.path_term(¬e.path)); + self.add_document(note)?; + self.writer.commit().context("commit upsert")?; + Ok(()) + } + + /// Drop one note by path. No-op (still commits) if the path was never + /// indexed; callers can call freely on delete-events. + /// + /// # Errors + /// Returns the underlying Tantivy error if the commit fails. + pub fn delete(&mut self, path: &str) -> Result<()> { + self.writer.delete_term(self.path_term(path)); + self.writer.commit().context("commit delete")?; + Ok(()) + } + + /// Number of indexed documents from a fresh reader. Read-only; intended + /// for tests and the `/health` endpoint ([ANW-8]). + /// + /// # Errors + /// Returns the underlying Tantivy error if opening the reader fails. + pub fn document_count(&self) -> Result { + let reader = self.index.reader().context("open reader")?; + let searcher = reader.searcher(); + Ok(searcher.num_docs()) + } + + /// Expose the [`Index`] so other layers (the future query handler in + /// [ANW-15]) can build their own readers and parsers without re-deriving + /// the schema. + #[must_use] + pub fn index(&self) -> &Index { + &self.index + } + + #[must_use] + pub fn fields(&self) -> Fields { + self.fields + } + + fn path_term(&self, path: &str) -> Term { + Term::from_field_text(self.fields.path, path) + } + + fn add_document(&mut self, note: &Note) -> Result<()> { + // Build the doc as JSON keyed by schema field name and let Tantivy + // parse it -- avoids restating the OwnedValue tree by hand. + let doc_json = json!({ + "path": ¬e.path, + "frontmatter": frontmatter_to_json(¬e.frontmatter), + }); + let doc = TantivyDocument::parse_json(&self.index.schema(), &doc_json.to_string()) + .context("parse_json")?; + self.writer.add_document(doc).context("add_document")?; + Ok(()) + } +} + +/// Convert a typed [`Value`] tree to `serde_json::Value`. Typed dates and +/// datetimes are emitted as their ISO-8601 / RFC 3339 string forms so they +/// sort correctly under range queries with the `raw` tokenizer. +fn value_to_json(v: &Value) -> JsonValue { + match v { + Value::Null => JsonValue::Null, + Value::Bool(b) => JsonValue::Bool(*b), + Value::Int(i) => json!(i), + Value::Float(f) => json!(f), + Value::String(s) => JsonValue::String(s.clone()), + Value::Date(d) => JsonValue::String(d.format("%Y-%m-%d").to_string()), + Value::DateTime(dt) => JsonValue::String(dt.to_rfc3339()), + Value::Sequence(seq) => JsonValue::Array(seq.iter().map(value_to_json).collect()), + Value::Mapping(m) => { + let mut map = JsonMap::new(); + for (k, v) in m { + map.insert(k.clone(), value_to_json(v)); + } + JsonValue::Object(map) + } + } +} + +fn frontmatter_to_json(fm: &crate::vault::Frontmatter) -> JsonValue { + let mut map = JsonMap::new(); + for (k, v) in fm { + map.insert(k.clone(), value_to_json(v)); + } + JsonValue::Object(map) +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{TimeZone, Utc}; + use std::collections::BTreeMap; + use tantivy::collector::TopDocs; + use tantivy::query::TermQuery; + + fn sample_note(path: &str, tag: &str) -> Note { + let mut fm: BTreeMap = BTreeMap::new(); + fm.insert( + "tags".into(), + Value::Sequence(vec![Value::String(tag.into())]), + ); + fm.insert("title".into(), Value::String(format!("note {path}"))); + Note { + path: path.into(), + frontmatter: fm, + body: String::new(), + raw_bytes: Vec::new(), + last_modified: Utc.with_ymd_and_hms(2026, 5, 14, 12, 0, 0).unwrap(), + etag: "\"deadbeef\"".into(), + size: 0, + } + } + + fn count_path(idx: &NoteIndex, path: &str) -> tantivy::Result { + let reader = idx.index().reader()?; + let searcher = reader.searcher(); + let term = Term::from_field_text(idx.fields().path, path); + let query = TermQuery::new(term, IndexRecordOption::Basic); + let hits = searcher.search(&query, &TopDocs::with_limit(10))?; + Ok(hits.len()) + } + + #[test] + fn rebuild_populates_index() { + let mut idx = NoteIndex::new().unwrap(); + idx.rebuild(&[sample_note("a.md", "x"), sample_note("b.md", "y")]) + .unwrap(); + assert_eq!(idx.document_count().unwrap(), 2); + assert_eq!(count_path(&idx, "a.md").unwrap(), 1); + assert_eq!(count_path(&idx, "b.md").unwrap(), 1); + } + + #[test] + fn rebuild_replaces_previous_contents() { + let mut idx = NoteIndex::new().unwrap(); + idx.rebuild(&[sample_note("old.md", "x")]).unwrap(); + idx.rebuild(&[sample_note("new.md", "y")]).unwrap(); + assert_eq!(idx.document_count().unwrap(), 1); + assert_eq!(count_path(&idx, "old.md").unwrap(), 0); + assert_eq!(count_path(&idx, "new.md").unwrap(), 1); + } + + #[test] + fn upsert_replaces_by_path() { + let mut idx = NoteIndex::new().unwrap(); + idx.upsert(&sample_note("a.md", "v1")).unwrap(); + // Mutate then upsert under the same path; document count must stay 1. + idx.upsert(&sample_note("a.md", "v2")).unwrap(); + assert_eq!(idx.document_count().unwrap(), 1); + assert_eq!(count_path(&idx, "a.md").unwrap(), 1); + } + + #[test] + fn delete_removes_path() { + let mut idx = NoteIndex::new().unwrap(); + idx.upsert(&sample_note("a.md", "x")).unwrap(); + idx.upsert(&sample_note("b.md", "y")).unwrap(); + idx.delete("a.md").unwrap(); + assert_eq!(idx.document_count().unwrap(), 1); + assert_eq!(count_path(&idx, "a.md").unwrap(), 0); + assert_eq!(count_path(&idx, "b.md").unwrap(), 1); + } + + #[test] + fn delete_unknown_path_is_noop() { + let mut idx = NoteIndex::new().unwrap(); + idx.upsert(&sample_note("a.md", "x")).unwrap(); + idx.delete("never-indexed.md").unwrap(); + assert_eq!(idx.document_count().unwrap(), 1); + } + + #[test] + fn value_to_json_coerces_dates_to_iso_strings() { + let d = chrono::NaiveDate::from_ymd_opt(2026, 5, 14).unwrap(); + assert_eq!( + value_to_json(&Value::Date(d)), + JsonValue::String("2026-05-14".into()) + ); + + let dt = chrono::DateTime::parse_from_rfc3339("2026-05-14T10:14:22Z").unwrap(); + assert_eq!( + value_to_json(&Value::DateTime(dt)), + JsonValue::String("2026-05-14T10:14:22+00:00".into()) + ); + } + + #[test] + fn value_to_json_handles_nested_structures() { + let mut inner: BTreeMap = BTreeMap::new(); + inner.insert("name".into(), Value::String("brn".into())); + let v = Value::Mapping(inner); + let j = value_to_json(&v); + let JsonValue::Object(obj) = j else { + panic!("expected object"); + }; + assert_eq!(obj.get("name"), Some(&JsonValue::String("brn".into()))); + } +} diff --git a/src/lib.rs b/src/lib.rs index 043bb29..99b04d1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,4 +5,5 @@ //! supervisor tree, and the HTTP surface as those issues land. pub mod app; +pub mod index; pub mod vault; diff --git a/src/vault.rs b/src/vault.rs index c09197d..d702edd 100644 --- a/src/vault.rs +++ b/src/vault.rs @@ -17,23 +17,26 @@ use std::collections::BTreeMap; use std::io; use std::path::{Path, PathBuf}; -use std::time::SystemTime; -use chrono::{DateTime, FixedOffset, NaiveDate}; +use chrono::{DateTime, FixedOffset, NaiveDate, Utc}; +use serde::{Deserialize, Serialize}; use thiserror::Error; use walkdir::WalkDir; /// One scanned note. `raw_bytes` is retained so `Accept: text/markdown` can /// return the exact bytes that produced `etag` without re-reading from disk /// between watcher events ([ANW-13](https://crvrs.youtrack.cloud/issue/ANW-13)). -#[derive(Debug, Clone)] +/// +/// Derives `Serialize` / `Deserialize` so the record can flow over Hydra +/// process messages (see [[ADR-004 Hydra as Process Runtime]]). +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct Note { /// Path relative to the vault root, forward-slash separated. pub path: String, pub frontmatter: Frontmatter, pub body: String, pub raw_bytes: Vec, - pub last_modified: SystemTime, + pub last_modified: DateTime, pub etag: String, pub size: u64, } @@ -42,7 +45,7 @@ pub type Frontmatter = BTreeMap; /// Frontmatter value with the type-coercion contract from /// [[ADR-005 Frontmatter Contract Type Coercion and Cross-Note Shapes]] applied. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub enum Value { Null, Bool(bool), @@ -135,7 +138,7 @@ fn read_one(vault_root: &Path, abs_path: &Path) -> Result { let raw_bytes = std::fs::read(abs_path)?; let metadata = std::fs::metadata(abs_path)?; let size = metadata.len(); - let last_modified = metadata.modified()?; + let last_modified: DateTime = metadata.modified()?.into(); let etag = format!("\"{}\"", blake3::hash(&raw_bytes).to_hex()); let text = std::str::from_utf8(&raw_bytes).map_err(|_| ScanIssueKind::NonUtf8Body)?;