ANW-12 tantivy index: schema, upsert, rebuild on startup, scanner->writer wire

Cargo.lock

@@ -332,6 +332,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0"
 dependencies = [
  "num-traits",
+ "serde",
 ]
 
 [[package]]

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"

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<Pid, ExitReason> {
+        // 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<M>(role: &str) -> Result<Option<M>, 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<Option<Self::Message>, 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<Option<Self::Message>, 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<Note>),
+    /// 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<Note>),
+    /// 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<RestartCounters>,
+    /// 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<NoteIndex>,
+}
+
+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(&notes) {
+                    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(&note) {
+                    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<Option<Self::Message>, 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<Option<Self::Message>, ExitReason> {
-        Ok(None)
+        call_not_supported("http_server")
     }
 }
 

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<Self> {
+        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(&note.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<u64> {
+        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": &note.path,
+            "frontmatter": frontmatter_to_json(&note.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<String, Value> = 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<usize> {
+        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<String, Value> = 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())));
+    }
+}

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;

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<u8>,
-    pub last_modified: SystemTime,
+    pub last_modified: DateTime<Utc>,
     pub etag: String,
     pub size: u64,
 }
@@ -42,7 +45,7 @@ pub type Frontmatter = BTreeMap<String, Value>;
 
 /// 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<Note, ScanIssueKind> {
     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<Utc> = metadata.modified()?.into();
     let etag = format!("\"{}\"", blake3::hash(&raw_bytes).to_hex());
 
     let text = std::str::from_utf8(&raw_bytes).map_err(|_| ScanIssueKind::NonUtf8Body)?;