diff --git a/.changeset/nip50-search-index.md b/.changeset/nip50-search-index.md new file mode 100644 index 00000000..c79feaff --- /dev/null +++ b/.changeset/nip50-search-index.md @@ -0,0 +1,7 @@ +--- +--- + +feat(storage): NIP-50 search over the LMDB store + +No published package changes: the change is Rust-only, in `nostream-storage`, and +touches no TypeScript package the release set publishes. diff --git a/crates/nostream-storage/src/lmdb/events.rs b/crates/nostream-storage/src/lmdb/events.rs index b262a842..413bc943 100644 --- a/crates/nostream-storage/src/lmdb/events.rs +++ b/crates/nostream-storage/src/lmdb/events.rs @@ -22,6 +22,7 @@ use tokio::task::JoinHandle; use super::encoding::{self, EventHeader}; use super::rows; +use super::search; use super::{ DEFAULT_FILTER_LIMIT, Databases, Gates, MAX_PAGE_KEYS, RETENTION_BATCH_SIZE, STREAM_BATCH_ITEMS, STREAM_BATCH_KEYS, @@ -244,6 +245,7 @@ fn write_event( .events_expiring .put(txn, &encoding::time_key(expires_at, &id), &[])?; } + search::index_event(txn, databases, record)?; Ok(()) } @@ -326,6 +328,7 @@ fn hard_delete( .tags .delete(txn, &encoding::tag_key(name, value, created_at, &id)?)?; } + search::remove_event(txn, databases, record)?; if let Some(deduplication) = &record.event_deduplication && encoding::replaceable_kind(kind) { @@ -802,7 +805,12 @@ pub(crate) fn page( } } } - sort_page(&mut union, filters[0].limit.is_some()); + // A one-filter search read arrives in the ranker's order, and `sort_page` would replace it + // with the created-at order; every other read sorts as it did. + let searched = filters.len() == 1 && search::query_terms(&filters[0], gates)?.is_some(); + if !searched { + sort_page(&mut union, filters[0].limit.is_some()); + } Ok(union) } @@ -844,8 +852,17 @@ pub(crate) fn count( fresh } None => { - let fresh = count_matches(txn, databases, filter, &counted, gates, now)?; - counted.predicates.push(filter); + let terms = search::query_terms(filter, gates)?; + let fresh = count_matches( + txn, + databases, + filter, + &counted, + gates, + now, + terms.as_deref(), + )?; + counted.predicates.push(CountedFilter { filter, terms }); fresh } }; @@ -859,7 +876,14 @@ struct Counted<'a> { /// The ids of every earlier filter that had a limit: its whole page. pages: Vec>, /// Every earlier filter without a limit, whose page is every row its predicate matches. - predicates: Vec<&'a Filter>, + predicates: Vec>, +} + +/// An earlier filter of a count, with the terms its search resolved to — resolved once when the +/// filter was recorded, so deciding a later row does not re-tokenise the query. +struct CountedFilter<'a> { + filter: &'a Filter, + terms: Option>, } /// Whether the count already holds `id`, so counting it again would count the row twice. @@ -874,14 +898,19 @@ fn already_counted( if counted.pages.iter().any(|page| page.contains(id)) { return Ok(true); } - for filter in &counted.predicates { - let matcher = Matcher { - filter, + // + // shortcut: each earlier predicate's resolved terms are cloned per row, because the matcher + // owns the list it carries; upgrade when a count over a large store is dominated by that + // allocation, which needs the matcher to borrow a list the read owns. + for earlier in &counted.predicates { + let matcher = Matcher::with_terms( + earlier.filter, gates, now, - live: true, - proven: None, - }; + true, + None, + earlier.terms.clone(), + ); if matcher.matches(txn, databases, id)?.is_some() { return Ok(true); } @@ -906,12 +935,23 @@ fn count_matches( counted: &Counted<'_>, gates: &Gates, now: i64, + terms: Option<&[String]>, ) -> Result { - count_matches_bounded(txn, databases, filter, counted, gates, now, MAX_PAGE_KEYS) + count_matches_bounded( + txn, + databases, + filter, + counted, + gates, + now, + terms, + MAX_PAGE_KEYS, + ) } /// [`count_matches`], with the number of ids the walk may hold given explicitly, so a test can /// compare the bounded walk against an unbounded one over one corpus. +#[allow(clippy::too_many_arguments)] fn count_matches_bounded( txn: &RoTxn<'_>, databases: &Databases, @@ -919,19 +959,33 @@ fn count_matches_bounded( counted: &Counted<'_>, gates: &Gates, now: i64, + terms: Option<&[String]>, cap: usize, ) -> Result { - if !admits(filter, gates)? { + if !admits(filter)? { return Ok(0); } let plan = plan(filter, databases); - let matcher = Matcher { + let matcher = Matcher::with_terms( filter, gates, now, - live: true, - proven: plan.tag.map(|(name, _)| name), - }; + true, + plan.tag.map(|(name, _)| name), + terms.map(<[String]>::to_vec), + ); + // A filter whose terms resolve is counted from the term index — the same walk the page + // takes, with no page to stop at — so the count's predicate is the read's predicate. + if let Some(terms) = terms { + let mut total = 0; + search::walk(txn, databases, &matcher, terms, |id, _, _| { + if !already_counted(txn, databases, counted, gates, now, &id)? { + total += 1; + } + Ok(()) + })?; + return Ok(total); + } // A single exact criterion names one key per row, so only a second criterion or a `#g` // prefix can walk one row twice, and only then is the id set paid for. // @@ -1054,21 +1108,13 @@ fn count_limit(limit: u64) -> usize { usize::try_from(limit.min(MAX_PAGE_KEYS as u64)).unwrap_or(MAX_PAGE_KEYS) } -/// The refusals and short-circuits every read applies before it seeks an index. +/// The short-circuits every read applies before it seeks an index. /// /// `true` means the filter can match rows; `false` means a criterion family is present but -/// empty, which is the capture's `1 = 0`. -fn admits(filter: &Filter, gates: &Gates) -> Result { - if gates.nip50 - && filter - .search - .as_deref() - .is_some_and(|search| !search.trim().is_empty()) - { - return Err(StorageError::NotImplemented { - module: "storage::lmdb::nip50_search", - }); - } +/// empty, which is the capture's `1 = 0`. A `search` is not one of them: whether it is a search +/// at all, and what it asks for, is [`search::query_terms`]'s answer, and the terms it resolves +/// are the search walk's (or, when they resolve to nothing, a page with no rows). +fn admits(filter: &Filter) -> Result { Ok(!(filter.ids.as_ref().is_some_and(|ids| ids.is_empty()) || filter .authors @@ -1098,20 +1144,22 @@ fn branch( live: bool, limit: usize, ) -> Result, StorageError> { - if !admits(filter, gates)? { + if !admits(filter)? { return Ok(Vec::new()); } let plan = plan(filter, databases); // A `tags` range is built from the family's own criteria, so every key in it satisfies that - // family: the matcher skips it and no body is decoded for it. - let matcher = Matcher { - filter, - gates, - now, - live, - proven: plan.tag.map(|(name, _)| name), - }; + // family: the matcher skips it and no body is decoded for it. A search proves no family — + // the term index selects a row by its content — and the matcher drops the proof itself when + // it carries terms. + let matcher = Matcher::new(filter, gates, now, live, plan.tag.map(|(name, _)| name))?; + // The router every read method shares: a filter whose search resolves is answered from the + // term index, and `plan` is not consulted for it. A count's limited path arrives here too + // (`count`), so the rows it counts are the ones this page returns. + if let Some(terms) = matcher.terms() { + return search::page(txn, databases, &matcher, terms, limit); + } let descending = filter.limit.is_some(); let mut page = Page::new(descending); for range in &plan.ranges { @@ -1513,27 +1561,93 @@ fn candidate( Ok(Some((rows::event_header(value)?.created_at, id))) } +/// What one row answers a filter: the header every caller reads, and the rank a search's page +/// orders by — `Some` exactly when the matcher carries search terms and the row's body holds +/// them all. +pub(crate) struct Verdict { + pub(crate) header: EventHeader, + pub(crate) rank: Option, +} + /// One filter's criteria, evaluated against a stored event. -struct Matcher<'a> { +pub(crate) struct Matcher<'a> { filter: &'a Filter, gates: &'a Gates, now: i64, live: bool, - /// The tag family the index range that selected the row has already proved, if any. + /// The tag family the index range that selected the row has already proved, if any. A + /// searched filter's keys prove none — the term index selects a row by its content — so a + /// matcher carrying terms never sees one, whatever the caller's plan says. proven: Option<&'a str>, + /// The filter's resolved search terms, `None` when it is not a search. + terms: Option>, } -impl Matcher<'_> { - fn matches( +impl<'a> Matcher<'a> { + /// The matcher every read path builds: resolves the filter's search terms once, so the + /// language refusal reaches every read method rather than only the one that walks the page. + pub(crate) fn new( + filter: &'a Filter, + gates: &'a Gates, + now: i64, + live: bool, + proven: Option<&'a str>, + ) -> Result { + Ok(Self::with_terms( + filter, + gates, + now, + live, + proven, + search::query_terms(filter, gates)?, + )) + } + + /// A matcher for a read that has already resolved this filter's terms — a count deciding a + /// row against an earlier filter, or a reconciliation batch — so they are not re-resolved. + /// + /// `proven` is the family the caller's plan proved, and it is dropped when the filter is a + /// search: a searched filter is answered from `events_by_term`, whose keys say nothing about + /// any tag family, so every tag criterion has to be tested from the body. Deciding that here + /// is what keeps the page, the count and the reconciliation stream from each restating it. + pub(crate) fn with_terms( + filter: &'a Filter, + gates: &'a Gates, + now: i64, + live: bool, + proven: Option<&'a str>, + terms: Option>, + ) -> Self { + Self { + filter, + gates, + now, + live, + proven: if terms.is_some() { None } else { proven }, + terms, + } + } + + /// The search terms this matcher carries, `None` when the filter is not a search. + pub(crate) fn terms(&self) -> Option<&[String]> { + self.terms.as_deref() + } + + pub(crate) fn matches( &self, txn: &RoTxn<'_>, databases: &Databases, id: &EventId, - ) -> Result, StorageError> { + ) -> Result, StorageError> { let Some(value) = databases.events.get(txn, id.as_bytes())? else { return Ok(None); }; let header = rows::event_header(value)?; + // A search whose query held no token selects nothing: it refuses every row rather than + // falling through to "no terms, no restriction". + if self.terms.as_deref().is_some_and(<[String]>::is_empty) { + return Ok(None); + } if self.live && (header.deleted @@ -1577,12 +1691,25 @@ impl Matcher<'_> { if self.gates.nip56 && self.reported(txn, databases, id, &header.pubkey)? { return Ok(None); } - if self.carries_other_tag_criteria() - && !tag_criteria_match(&self.filter.tags, &rows::decode_event(value)?, self.proven) - { - return Ok(None); + // The tag families and the search verdict share one decode: a searched row that also + // carries tag criteria pays for its body once, and a row that needs neither reads only + // the header. + let mut rank = None; + if self.carries_other_tag_criteria() || self.terms.is_some() { + let record = rows::decode_event(value)?; + if self.carries_other_tag_criteria() + && !tag_criteria_match(&self.filter.tags, &record, self.proven) + { + return Ok(None); + } + if let Some(terms) = &self.terms { + match search::verdict(&record.event.content, terms) { + Some(found) => rank = Some(found), + None => return Ok(None), + } + } } - Ok(Some(header)) + Ok(Some(Verdict { header, rank })) } /// Whether the filter carries a tag family the row's key has not already proved. A row the @@ -1816,6 +1943,8 @@ pub(crate) struct ReconciliationStream { filter: Filter, gates: Gates, now: i64, + /// The filter's resolved search terms, kept so no batch re-resolves or re-refuses. + terms: Option>, range_count: usize, position: Position, pending: Option, @@ -1824,9 +1953,10 @@ pub(crate) struct ReconciliationStream { } impl ReconciliationStream { - /// Builds the stream for one filter, or refuses it as every other read does: a non-empty - /// `search` with the NIP-50 gate on is [`StorageError::NotImplemented`], and a criterion - /// family that is present but empty selects nothing (the capture's `1 = 0`). + /// Builds the stream for one filter, or refuses it as every other read does: a `search` + /// under a language the index cannot reproduce is refused by this call rather than yielded + /// as a stream item, and a criterion family that is present but empty selects nothing (the + /// capture's `1 = 0`). pub(crate) fn new( env: Env, databases: Databases, @@ -1834,7 +1964,12 @@ impl ReconciliationStream { gates: Gates, now: i64, ) -> Result { - let admitted = admits(filter, &gates)?; + let admitted = admits(filter)?; + // Resolved once, before the stream exists, so the refusal is this call's error and + // every batch reuses the list rather than re-resolving it. + let terms = Matcher::new(filter, &gates, now, true, None)? + .terms() + .map(<[String]>::to_vec); let range_count = if admitted { stream_plan(filter, &databases).ranges.len() } else { @@ -1846,6 +1981,7 @@ impl ReconciliationStream { filter: filter.clone(), gates, now, + terms, range_count, position: Position::new(), pending: None, @@ -1895,6 +2031,7 @@ impl Stream for ReconciliationStream { let filter = this.filter.clone(); let gates = this.gates; let now = this.now; + let terms = this.terms.clone(); let mut position = std::mem::replace(&mut this.position, Position::new()); let exhausted = position.index >= this.range_count; if exhausted { @@ -1909,6 +2046,7 @@ impl Stream for ReconciliationStream { &filter, &gates, now, + terms, &mut position, )?; Ok((items, position)) @@ -1955,22 +2093,25 @@ fn stream_plan<'a>(filter: &'a Filter, databases: &Databases) -> Plan<'a> { /// The key cap is what keeps one poll short however few of the keys it walks match: every /// transaction a stream opens is closed by the poll that opened it, so no read pins a page while /// the writer works. +#[allow(clippy::too_many_arguments)] fn scan_batch( txn: &RoTxn<'_>, databases: &Databases, filter: &Filter, gates: &Gates, now: i64, + terms: Option>, position: &mut Position, ) -> Result, StorageError> { let plan = stream_plan(filter, databases); - let matcher = Matcher { + let matcher = Matcher::with_terms( filter, gates, now, - live: true, - proven: plan.tag.map(|(name, _)| name), - }; + true, + plan.tag.map(|(name, _)| name), + terms, + ); if plan.path == AccessPath::Ids { return scan_ids(txn, databases, &matcher, &plan, position); } @@ -2641,6 +2782,7 @@ mod tests { &filter, &Gates::default(), 0, + None, &mut position, ) .expect("the batch"); @@ -2685,6 +2827,7 @@ mod tests { &filter, &Gates::default(), 0, + None, &mut position, ) .expect("the batch"); @@ -2706,6 +2849,7 @@ mod tests { &filter, &Gates::default(), 0, + None, &mut position, ) .expect("the batch"), @@ -2889,6 +3033,7 @@ mod tests { &filter, &Gates::default(), 0, + None, &mut position, ) .expect("the batch"); @@ -3294,6 +3439,7 @@ mod tests { &counted, &Gates::default(), 0, + None, MAX_PAGE_KEYS, ) .expect("the bounded count"); @@ -3306,6 +3452,7 @@ mod tests { &counted, &Gates::default(), 0, + None, usize::MAX, ) .expect("the unbounded count"); @@ -3388,19 +3535,14 @@ mod tests { databases: &Databases, filter: &Filter, ) -> Vec<(i64, EventId)> { - let matcher = Matcher { - filter, - gates: &Gates::default(), - now: 0, - live: false, - proven: None, - }; + let gates = Gates::default(); + let matcher = Matcher::with_terms(filter, &gates, 0, false, None, None); let mut page = Vec::new(); for entry in databases.events.iter(txn).expect("the store") { let (key, _) = entry.expect("a row"); let id = encoding::event_id_from(key).expect("an event id"); - if let Some(header) = matcher.matches(txn, databases, &id).expect("a match") { - page.push((header.created_at, id)); + if let Some(verdict) = matcher.matches(txn, databases, &id).expect("a match") { + page.push((verdict.header.created_at, id)); } } sort_page(&mut page, filter.limit.is_some()); @@ -3415,20 +3557,15 @@ mod tests { databases: &Databases, filter: &Filter, ) -> Vec { - let matcher = Matcher { - filter, - gates: &Gates::default(), - now: 0, - live: true, - proven: None, - }; + let gates = Gates::default(); + let matcher = Matcher::with_terms(filter, &gates, 0, true, None, None); let mut items = Vec::new(); for entry in databases.events.iter(txn).expect("the store") { let (key, _) = entry.expect("a row"); let id = encoding::event_id_from(key).expect("an event id"); - if let Some(header) = matcher.matches(txn, databases, &id).expect("a match") { + if let Some(verdict) = matcher.matches(txn, databases, &id).expect("a match") { items.push(ReconciliationItem { - created_at: header.created_at, + created_at: verdict.header.created_at, event_id: id, }); } @@ -4012,6 +4149,7 @@ mod tests { &filter, &Gates::default(), 0, + None, &mut position, ) .expect("the batch"); @@ -4036,6 +4174,7 @@ mod tests { &filter, &Gates::default(), 0, + None, &mut position, ) .expect("the batch"), diff --git a/crates/nostream-storage/src/lmdb/mod.rs b/crates/nostream-storage/src/lmdb/mod.rs index 373863c3..a418ef52 100644 --- a/crates/nostream-storage/src/lmdb/mod.rs +++ b/crates/nostream-storage/src/lmdb/mod.rs @@ -1,4 +1,4 @@ -//! The LMDB backend: one environment, the 29 named databases, and the nine repository traits. +//! The LMDB backend: one environment, the 30 named databases, and the nine repository traits. //! Mirrors `src/repositories/` through the seam the Postgres backend also implements, and //! `docs/rewrite/decisions/0011-lmdb-backend.md` records why each choice below was made. //! @@ -14,10 +14,14 @@ mod metadata; mod migration; mod notifications; mod rows; +mod search; mod users; -use std::path::PathBuf; -use std::sync::Arc; +pub use search::{SEARCH_INDEX_DB, SEARCH_INDEX_MARKER, SEARCH_INDEX_VERSION}; + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, LazyLock, Mutex}; use chrono::Utc; use heed::types::Bytes; @@ -33,8 +37,8 @@ use crate::repositories::{ EventRetentionOptions, InviteCode, InviteCodeRepository, Invoice, InvoiceRepository, InvoiceStatus, NewDeliveryLogEntry, NewReport, Nip05Verification, Nip05VerificationRepository, NotificationDeliveryLogRepository, NotificationOutboxRepository, OutboxMessage, OutboxPayload, - QueryStream, ReconciliationItem, Report, ReportRepository, StorageFuture, StreamedEvent, User, - UserPatch, UserRepository, + QueryStream, ReconciliationItem, Report, ReportRepository, SearchRule, StorageFuture, + StreamedEvent, User, UserPatch, UserRepository, }; /// The fixed map size every environment is opened with: `Env::resize` is `unsafe` and this @@ -44,9 +48,9 @@ use crate::repositories::{ /// approach it, which needs `Env::resize` (a second `unsafe` call) or a refusal at open. pub const LMDB_MAP_SIZE: usize = 64 << 30; -/// The number of named databases the environment reserves, comfortably above the 29 it opens. +/// The number of named databases the environment reserves, comfortably above the 30 it opens. /// -/// shortcut: 29 databases open under a reserve of 64; upgrade when a tenth repository or a new +/// shortcut: 30 databases open under a reserve of 64; upgrade when a tenth repository or a new /// query path crosses that reserve, which moves this value and every store's reopen path. pub const LMDB_MAX_DBS: u32 = 64; @@ -79,15 +83,31 @@ pub(crate) const STREAM_BATCH_ITEMS: usize = 256; /// [`STREAM_BATCH_ITEMS`], so the transaction never spans a whole range. pub(crate) const STREAM_BATCH_KEYS: usize = 256; -/// The settings gates the capture reads per query. The LMDB backend records only whether -/// NIP-50 is on: it has no search implementation (ADR 0011, "Oracle: two arms, one executor" — -/// entry 21, out of scope, not a permitted failure). -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +/// The settings gates the capture reads per query. +/// +/// The NIP-50 pair is what a search has to be answered under: whether the feature is on, how +/// many characters of the query survive `nip50.maxQueryLength`, and whether `nip50.language` is +/// the one text-search configuration this store's index reproduces. The language travels as that +/// flag rather than by name so the struct stays `Copy`, and the refusal's message is static. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) struct Gates { pub(crate) nip50: bool, + pub(crate) max_query_length: u64, + pub(crate) language_simple: bool, pub(crate) nip56: bool, } +impl Default for Gates { + fn default() -> Self { + Self { + nip50: false, + max_query_length: search::DEFAULT_MAX_SEARCH_QUERY_LENGTH, + language_simple: true, + nip56: false, + } + } +} + /// The named databases of one environment. #[derive(Debug, Clone, Copy)] pub(crate) struct Databases { @@ -119,6 +139,7 @@ pub(crate) struct Databases { pub(crate) notification_delivery_log: Database, pub(crate) delivery_log_by_created_at: Database, pub(crate) delivery_log_by_outbox: Database, + pub(crate) events_by_term: Database, pub(crate) meta: Database, } @@ -163,6 +184,7 @@ impl Databases { delivery_log_by_created_at: env .create_database(txn, Some("delivery_log_by_created_at"))?, delivery_log_by_outbox: env.create_database(txn, Some("delivery_log_by_outbox"))?, + events_by_term: env.create_database(txn, Some(search::SEARCH_INDEX_DB))?, meta: env.create_database(txn, Some("meta"))?, }) } @@ -229,6 +251,28 @@ pub struct MapUsage { pub used_bytes: u64, } +/// One lock per store path, held by [`LmdbBackend::open`] across the marker read, its comparison +/// and the whole rebuild, so two opens of one store cannot both decide the index is stale. +/// +/// Per path rather than one process-wide lock, because a single lock would hold one store's cold +/// start against every other store's open and buy nothing. An entry is created per store path and +/// left in place after use: the map is bounded by the number of distinct store paths a process +/// opens, which is the number of stores it owns, so it is not a leak. +/// +/// `LazyLock` rather than a `const` initialiser because `HashMap::new` is not a `const fn` on this +/// toolchain; it is the standard library's only way to hold a map in a `static`. +static INDEX_REBUILDS: LazyLock>>>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +/// The lock that serialises index rebuilds at `path`, keyed by the canonical path heed itself +/// keys its own environment cache by, so two spellings of one directory share one lock. +fn rebuild_lock(path: &Path) -> Arc> { + let mut locks = INDEX_REBUILDS + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + Arc::clone(locks.entry(path.to_path_buf()).or_default()) +} + /// An embedded LMDB event store with its metadata tail, built from the root a caller supplies. #[derive(Debug, Clone)] pub struct LmdbBackend { @@ -243,6 +287,10 @@ impl LmdbBackend { /// The usage it reports is what the map's approach to [`LMDB_MAP_SIZE`] is read from: the /// store never grows, so an operator sees it coming here rather than at the first failed /// write. + /// + /// Two opens of one store serialise on the per-path rebuild lock across the marker read and + /// any rebuild, so one store's index is rebuilt once however many callers open it at the + /// same time; the comparison below is only sound while nothing else moves the marker. pub async fn open(root: impl Into) -> Result { let root = root.into(); // One home for the shape, shared with the writable opener and the CLI. @@ -250,9 +298,28 @@ impl LmdbBackend { let opened = tokio::task::spawn_blocking(move || { let environment = LmdbEnv::open(&path, LMDB_MAP_SIZE)?; let handle = environment.env().clone(); + #[cfg(test)] + rendezvous::await_partner(handle.path()); + // A `std::sync::Mutex` taken here and dropped before the usage read: the whole of it + // runs on this blocking thread, so nothing holds it across an await, and no other + // code takes it, so it orders against nothing else. + let rebuilding = rebuild_lock(handle.path()); + let rebuilding = rebuilding + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); let mut transaction = handle.write_txn()?; let databases = Databases::create(&handle, &mut transaction)?; + let marker = databases + .meta + .get(&transaction, search::SEARCH_INDEX_MARKER)? + .map(<[u8]>::to_vec); transaction.commit()?; + // An index that was never built, or one a previous tokeniser wrote, is replaced + // before the store is handed out, so no search can read a half-built index. + if marker.as_deref() != Some(&[search::SEARCH_INDEX_VERSION][..]) { + search::rebuild(&handle, &databases)?; + } + drop(rebuilding); let used_bytes = handle.non_free_pages_size()?; Ok::<_, StorageError>((environment, databases, used_bytes)) }) @@ -356,6 +423,16 @@ impl LmdbBackend { .nip50 .as_ref() .is_some_and(|nip50| nip50.enabled == Some(true)), + max_query_length: settings + .nip50 + .as_ref() + .and_then(|nip50| nip50.max_query_length) + .unwrap_or(search::DEFAULT_MAX_SEARCH_QUERY_LENGTH), + language_simple: settings + .nip50 + .as_ref() + .and_then(|nip50| nip50.language.as_deref()) + .is_none_or(|language| language == search::SEARCH_LANGUAGE), nip56: settings .nip56 .as_ref() @@ -467,6 +544,19 @@ impl EventRepository for LmdbBackend { }) } + /// LMDB answers a searched filter from `events_by_term` under the `simple` tokeniser + /// (`search.rs`), which is Postgres' `ToTsvector { language: "simple" }` rule; with the gate + /// off no terms resolve, no predicate is applied, and the relay's own matcher is what decides. + fn search_rule(&self) -> SearchRule { + if self.gates().nip50 { + SearchRule::ToTsvector { + language: search::SEARCH_LANGUAGE.to_owned(), + } + } else { + SearchRule::SubstringTerms + } + } + /// LMDB's tag index holds no empty value: `encoding::tag_entries` writes a key only when the /// value is non-empty (`encoding.rs:387`), so the criterion `""` — which `Filter::matches` /// accepts — selects nothing. @@ -1055,6 +1145,60 @@ impl NotificationOutboxRepository for LmdbBackend { } } +/// A rendezvous the test that opens one store twice passes through between the environment and +/// the rebuild lock, so both callers are past `Env::open` before either reads the marker. Nothing +/// waits unless a test armed the store's path. +#[cfg(test)] +pub(crate) mod rendezvous { + use std::collections::HashMap; + use std::path::{Path, PathBuf}; + use std::sync::{Arc, Barrier, LazyLock, Mutex}; + + struct Armed { + barrier: Arc, + /// The callers still to arrive; the path is disarmed when the last one does. + waiting: usize, + } + + static ARMED: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + + /// Arms `path` so the next `arrivals` opens of it wait for each other before the rebuild lock. + pub(crate) fn arm(path: &Path, arrivals: usize) { + let mut armed = ARMED + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + armed.insert( + path.to_path_buf(), + Armed { + barrier: Arc::new(Barrier::new(arrivals)), + waiting: arrivals, + }, + ); + } + + /// Waits for the other armed opens of `path`. An unarmed path returns at once. + pub(crate) fn await_partner(path: &Path) { + let (barrier, last) = { + let mut armed = ARMED + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let Some(entry) = armed.get_mut(path) else { + return; + }; + entry.waiting -= 1; + (Arc::clone(&entry.barrier), entry.waiting == 0) + }; + if last { + ARMED + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .remove(path); + } + barrier.wait(); + } +} + #[cfg(test)] mod tests { #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] @@ -1164,7 +1308,7 @@ mod tests { .expect("the store opens"); let usage = backend.map_usage().await.expect("the usage"); assert_eq!(usage.map_size, LMDB_MAP_SIZE); - assert!(usage.used_bytes > 0, "the 29 databases occupy pages"); + assert!(usage.used_bytes > 0, "the 30 databases occupy pages"); assert!(usage.used_bytes <= LMDB_MAP_SIZE as u64); } @@ -1181,7 +1325,7 @@ mod tests { #[tokio::test] async fn a_named_database_past_the_maximum_is_refused() { - // The environment reserves LMDB_MAX_DBS handles; the 29 the backend opens leave the + // The environment reserves LMDB_MAX_DBS handles; the 30 the backend opens leave the // rest free, and the 65th creation has to fail rather than silently succeed. let directory = tempfile::tempdir().expect("tempdir"); let backend = LmdbBackend::open(directory.path()) diff --git a/crates/nostream-storage/src/lmdb/search.rs b/crates/nostream-storage/src/lmdb/search.rs new file mode 100644 index 00000000..6097a84d --- /dev/null +++ b/crates/nostream-storage/src/lmdb/search.rs @@ -0,0 +1,786 @@ +//! NIP-50 over the embedded store: the term index, the tokeniser both directions share, the +//! proximity rank, and the rebuild that gives a store written before this module existed its +//! index on its next open. +//! +//! The index is `events_by_term`, keyed `term ‖ event_id` with an empty value, so one term's +//! postings are one byte range in id order. It is maintained on every write path, whether or not +//! the NIP-50 gate is on, because the gate is a read gate: a store written with the feature off +//! answers a search the moment it is turned on. +//! +//! The index narrows candidates; it never decides. A query term finds the postings its key +//! holds, and the row's own body answers whether the term is *the* token or merely shares the +//! key — two tokens that share their first [`SEARCH_TERM_MAX_LEN`] bytes are one key and two +//! verdicts. `docs/rewrite/decisions/0012-nip50-search-index.md` records the decisions (D1–D9): +//! the index's shape, the tokeniser, the candidate-then-body predicate, the ranker and its +//! divergence from `ts_rank`, the marker, the rebuild on open, the gate-off rules, the language +//! guard, and the per-store lock that serialises two opens' rebuilds. ADR 0011 keeps the store +//! these live in. + +use std::cmp::Ordering; +use std::collections::{BinaryHeap, HashMap, HashSet}; +use std::ops::Bound; + +use heed::{Env, RoTxn, RwTxn}; +use nostream_core::{EventId, Filter}; + +use super::encoding; +use super::events::Matcher; +use super::rows; +use super::{Databases, Gates}; +use crate::error::StorageError; +use crate::repositories::EventRecord; + +/// The name of the term index database: `term ‖ event_id`, value empty, one posting per key. +pub const SEARCH_INDEX_DB: &str = "events_by_term"; + +/// The version of the index's contents, written into `meta` under [`SEARCH_INDEX_MARKER`] after +/// a rebuild. A store whose marker holds another value is rebuilt on the next open. +pub const SEARCH_INDEX_VERSION: u8 = 1; + +/// The `meta` key that records [`SEARCH_INDEX_VERSION`]. +pub const SEARCH_INDEX_MARKER: &[u8] = b"search-index-version"; + +/// The one text-search configuration this index reproduces: Postgres' `simple`, which has no +/// stemmer and no stop list, so a token is a token and `the` is an ordinary word. +pub(crate) const SEARCH_LANGUAGE: &str = "simple"; + +/// How many bytes of a term a key carries. A longer token is truncated on a `char` boundary, so +/// the index can only ever hold more candidates than the body accepts, never fewer. +pub(crate) const SEARCH_TERM_MAX_LEN: usize = 64; + +/// How many rows one rebuild write transaction indexes. +pub(crate) const SEARCH_REINDEX_BATCH: usize = 1000; + +/// The default `nip50.maxQueryLength` (`src/repositories/event-repository.ts:50-52`). +pub(crate) const DEFAULT_MAX_SEARCH_QUERY_LENGTH: u64 = 256; + +/// The message a language the index cannot reproduce is refused with. +const LANGUAGE_REFUSAL: &str = + "nip50.language must be the simple text-search configuration the lmdb store indexes with"; + +/// One term index key, as the write path composes it and the walk reads it. +/// +/// The term is cut to [`SEARCH_TERM_MAX_LEN`] bytes on a `char` boundary, so two tokens sharing +/// their first 64 bytes are one key; the id tail keeps a term's postings contiguous and in id +/// order, which is what makes the range seekable and its order stable. +fn term_key(term: &str, id: &EventId) -> Vec { + let bytes = term_bytes(term); + let mut key = Vec::with_capacity(bytes.len() + 32); + key.extend_from_slice(bytes); + key.extend_from_slice(id.as_bytes()); + key +} + +/// The bytes one term occupies in a key, cut to [`SEARCH_TERM_MAX_LEN`] on a `char` boundary. +fn term_bytes(term: &str) -> &[u8] { + let mut end = 0; + for (offset, character) in term.char_indices() { + let next = offset + character.len_utf8(); + if next > SEARCH_TERM_MAX_LEN { + break; + } + end = next; + } + &term.as_bytes()[..end] +} + +/// The bounds of one term's postings: the term's bytes, then the same bytes followed by the +/// largest possible id. Truncation can only merge two keys, so a query term's range covers every +/// posting a token sharing its prefix wrote. +fn term_range(term: &str) -> (Vec, Vec) { + let bytes = term_bytes(term); + let mut high = Vec::with_capacity(bytes.len() + 32); + high.extend_from_slice(bytes); + high.extend([0xff; 32]); + (bytes.to_vec(), high) +} + +/// The 32-byte id tail of an index key. +fn posting_id(key: &[u8]) -> Result { + let Some(id) = key.len().checked_sub(32).and_then(|start| key.get(start..)) else { + return Err(encoding::invalid( + SEARCH_INDEX_DB, + format!("a key of {} bytes", key.len()), + )); + }; + encoding::event_id_from(id) +} + +/// The tokens one text is searched by: the text lowercased by Unicode case mapping, then split +/// into maximal runs of [`char::is_alphanumeric`] characters, duplicates dropped. +/// +/// One statement of the rule, called by the write path that indexes a body and the read path +/// that asks whether a row carries a query's terms, so the index cannot disagree with the +/// predicate about a token. There is no stemming and no stop list, which is `simple`. +pub(crate) fn terms(text: &str) -> Vec { + let mut seen = HashSet::new(); + let mut terms = Vec::new(); + for token in tokens(text) { + if seen.insert(token.clone()) { + terms.push(token); + } + } + terms +} + +/// The maximal alphanumeric runs of `text`'s lowercase, in the order they occur. +/// +/// The lowercase comes first, by Unicode case mapping, because a mapping can turn one character +/// into several: `İ` lowers to `i` and a combining dot, which is not alphanumeric, so `İx` is two +/// tokens here and one if the runs were taken from the original text. +fn tokens(text: &str) -> Vec { + let mut tokens = Vec::new(); + let mut run = String::new(); + for character in text.to_lowercase().chars() { + if character.is_alphanumeric() { + run.push(character); + } else if !run.is_empty() { + tokens.push(std::mem::take(&mut run)); + } + } + if !run.is_empty() { + tokens.push(run); + } + tokens +} + +/// The terms a filter's search resolves to, or why it cannot be answered. +/// +/// `None` is "this filter is not a search": the gate is off, `search` is absent, or it trims to +/// nothing — the read then behaves as it does without the field, which is what Postgres does. An +/// empty list is a search that selects nothing: every token of a punctuation-only query was +/// dropped, which is the empty tsquery `plainto_tsquery` builds for it. +pub(crate) fn query_terms( + filter: &Filter, + gates: &Gates, +) -> Result>, StorageError> { + if !gates.nip50 { + return Ok(None); + } + let Some(search) = filter.search.as_deref() else { + return Ok(None); + }; + let query = search.trim(); + if query.is_empty() { + return Ok(None); + } + if !gates.language_simple { + return Err(StorageError::InvalidArgument { + message: LANGUAGE_REFUSAL, + }); + } + let limit = usize::try_from(gates.max_query_length).unwrap_or(usize::MAX); + let query: String = query.chars().take(limit).collect(); + Ok(Some(terms(&query))) +} + +/// The rank a body earns for a query's terms, or `None` when the body lacks one of them. +/// +/// The rank is the smallest window of consecutive tokens that contains every distinct term, so +/// two adjacent terms rank 2 however far apart their characters are, and a term repeated in the +/// body does not shorten a window the others stretch. Tokens are compared whole: the index may +/// have proposed the row on a shared 64-byte prefix, and only the full tokens decide. +pub(crate) fn verdict(content: &str, terms: &[String]) -> Option { + if terms.is_empty() { + return None; + } + let wanted: HashMap<&str, usize> = terms + .iter() + .enumerate() + .map(|(index, term)| (term.as_str(), index)) + .collect(); + let mut hits: Vec<(usize, usize)> = Vec::new(); + for (position, token) in tokens(content).into_iter().enumerate() { + if let Some(index) = wanted.get(token.as_str()) { + hits.push((position, *index)); + } + } + + let mut counts = vec![0usize; terms.len()]; + let mut covered = 0; + let mut left = 0; + let mut best: Option = None; + for right in 0..hits.len() { + let (position, term) = hits[right]; + counts[term] += 1; + if counts[term] == 1 { + covered += 1; + } + while covered == terms.len() { + let (first, first_term) = hits[left]; + let span = position - first + 1; + best = Some(best.map_or(span, |kept: usize| kept.min(span))); + counts[first_term] -= 1; + if counts[first_term] == 0 { + covered -= 1; + } + left += 1; + } + } + best.map(|span| u32::try_from(span).unwrap_or(u32::MAX)) +} + +/// Writes one row's postings: one key per distinct token of its content. +/// +/// Returns how many keys it wrote, which only the rebuild reports. +pub(crate) fn index_event( + txn: &mut RwTxn<'_>, + databases: &Databases, + record: &EventRecord, +) -> Result { + let id = record.event.id; + let mut written = 0; + for term in terms(&record.event.content) { + databases + .events_by_term + .put(txn, &term_key(&term, &id), &[])?; + written += 1; + } + Ok(written) +} + +/// Removes one row's postings: the same keys [`index_event`] wrote for it. +pub(crate) fn remove_event( + txn: &mut RwTxn<'_>, + databases: &Databases, + record: &EventRecord, +) -> Result<(), StorageError> { + let id = record.event.id; + for term in terms(&record.event.content) { + databases + .events_by_term + .delete(txn, &term_key(&term, &id))?; + } + Ok(()) +} + +/// One row the search page keeps, ordered so a max-heap roots at the worst kept row: the largest +/// rank, then the largest id. Popping that root while the heap holds more than the limit leaves +/// the rank-best rows, and [`BinaryHeap::into_sorted_vec`] drains them in the order the page is +/// returned in. +#[derive(Clone, Copy)] +struct SearchKept { + rank: u32, + id: EventId, + created_at: i64, +} + +impl Ord for SearchKept { + fn cmp(&self, other: &Self) -> Ordering { + self.rank + .cmp(&other.rank) + .then_with(|| self.id.cmp(&other.id)) + } +} + +impl PartialOrd for SearchKept { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl PartialEq for SearchKept { + fn eq(&self, other: &Self) -> bool { + self.cmp(other) == Ordering::Equal + } +} + +impl Eq for SearchKept {} + +/// The page a searched filter returns: the rank-best `limit` rows, in `(rank, event_id)` order. +/// +/// The page keeps its own heap because the index's key order is id order and rank order is not — +/// this is the one read whose order the page cannot take from the key it walked. +pub(crate) fn page( + txn: &RoTxn<'_>, + databases: &Databases, + matcher: &Matcher<'_>, + terms: &[String], + limit: usize, +) -> Result, StorageError> { + if limit == 0 { + return Ok(Vec::new()); + } + let mut kept = BinaryHeap::new(); + walk(txn, databases, matcher, terms, |id, created_at, rank| { + kept.push(SearchKept { + rank, + id, + created_at, + }); + while kept.len() > limit { + kept.pop(); + } + Ok(()) + })?; + Ok(kept + .into_sorted_vec() + .into_iter() + .map(|kept| (kept.created_at, kept.id)) + .collect()) +} + +/// Walks the postings of the first query term and hands every row the matcher accepts, with its +/// rank, to `keep`. +/// +/// The first term's range is the candidate list because a term's postings are one contiguous +/// range in id order. The range narrows by prefix: a key is `term ‖ event_id`, so the bounds +/// [`term_range`] builds admit every posting whose term begins with the first term's bytes — +/// `cats` sorts inside `cat`'s range. The walk then enforces term identity, comparing each +/// posting's own term bytes to the first term's, before the other terms are probed by exact key +/// and before any body is decoded. The verdict is the row's own (`Matcher::matches`): the walk +/// narrows by prefix and enforces identity, and the body decides. +/// +/// shortcut: the walk is bounded by the first query term's postings, not by the rarest term — +/// there are no statistics to choose one; upgrade when the benchmark rig can measure a corpus +/// and a stored document frequency, or an intersection of two ranges, is worth the second index. +/// The breadth #1112 measures remains: a prefix range still visits every longer token's postings +/// to reject them. +pub(crate) fn walk( + txn: &RoTxn<'_>, + databases: &Databases, + matcher: &Matcher<'_>, + terms: &[String], + mut keep: F, +) -> Result<(), StorageError> +where + F: FnMut(EventId, i64, u32) -> Result<(), StorageError>, +{ + let Some((first, others)) = terms.split_first() else { + return Ok(()); + }; + let first_bytes = term_bytes(first); + let (low, high) = term_range(first); + let bounds = encoding::bounds(&low, &high); + 'postings: for entry in databases.events_by_term.range(txn, &bounds)? { + let (key, _) = entry?; + let id = posting_id(key)?; + // The range is a prefix range, so a posting for a longer token that starts with the + // first term's bytes sorts inside it; only a posting whose own term is that term is a + // candidate. A key carries its term before the 32-byte id tail. + if key.get(..key.len() - 32) != Some(first_bytes) { + continue; + } + for other in others { + if databases + .events_by_term + .get(txn, &term_key(other, &id))? + .is_none() + { + continue 'postings; + } + } + let Some(verdict) = matcher.matches(txn, databases, &id)? else { + continue; + }; + let Some(rank) = verdict.rank else { + continue; + }; + keep(id, verdict.header.created_at, rank)?; + } + Ok(()) +} + +/// Rebuilds the index from the rows the store holds: clear, then index in +/// [`SEARCH_REINDEX_BATCH`]-row write transactions, then the marker. +/// +/// The clear is what makes the rebuild a replacement rather than an extension — an add-only pass +/// would keep every key a previous tokeniser wrote, and the body verdict would mask the stale +/// postings forever. The marker is written after the last batch, so a rebuild interrupted part +/// way leaves no valid marker and the next open runs it again. +pub(crate) fn rebuild(env: &Env, databases: &Databases) -> Result<(), StorageError> { + let started = std::time::Instant::now(); + let path = env.path(); + tracing::info!( + path = %path.display(), + "the lmdb search index is being rebuilt" + ); + + let mut after: Option> = None; + let mut indexed = 0; + let mut keys_written = 0; + let mut cleared = false; + loop { + let keys = batch_keys(env, databases, after.as_deref())?; + if keys.is_empty() { + break; + } + after = keys.last().cloned(); + let mut transaction = env.write_txn()?; + if !cleared { + databases.events_by_term.clear(&mut transaction)?; + cleared = true; + } + let mut written = 0; + for key in &keys { + let id = posting_id(key)?; + let Some(value) = databases.events.get(&transaction, id.as_bytes())? else { + continue; + }; + let record = rows::decode_event(value)?; + written += index_event(&mut transaction, databases, &record)?; + } + transaction.commit()?; + #[cfg(test)] + probes::batch(path, indexed + keys.len()); + #[cfg(test)] + probes::put_keys(path, written); + indexed += keys.len(); + keys_written += written; + if keys.len() < SEARCH_REINDEX_BATCH { + break; + } + } + + let mut transaction = env.write_txn()?; + if !cleared { + databases.events_by_term.clear(&mut transaction)?; + } + databases.meta.put( + &mut transaction, + SEARCH_INDEX_MARKER, + &[SEARCH_INDEX_VERSION], + )?; + transaction.commit()?; + #[cfg(test)] + probes::marker(path, indexed); + + tracing::info!( + path = %path.display(), + rows = indexed, + keys = keys_written, + elapsed_ms = started.elapsed().as_millis(), + "the lmdb search index is rebuilt" + ); + Ok(()) +} + +/// The next [`SEARCH_REINDEX_BATCH`] `created_at ‖ event_id` keys of the store, after `after`. +fn batch_keys( + env: &Env, + databases: &Databases, + after: Option<&[u8]>, +) -> Result>, StorageError> { + let transaction = env.read_txn()?; + let start = match after { + Some(after) => Bound::Excluded(after), + None => Bound::Unbounded, + }; + let mut keys = Vec::with_capacity(SEARCH_REINDEX_BATCH); + for entry in databases + .events_by_created_at + .range(&transaction, &(start, Bound::Unbounded))? + { + let (key, _) = entry?; + keys.push(key.to_vec()); + if keys.len() == SEARCH_REINDEX_BATCH { + break; + } + } + Ok(keys) +} + +/// The ordered record of a rebuild's commits, for the test that reads the order back. Process +/// wide and keyed by the store's canonical path, because the rebuild runs on the blocking pool's +/// thread and a sibling test rebuilds another store. +#[cfg(test)] +pub(crate) mod probes { + use std::collections::HashMap; + use std::path::{Path, PathBuf}; + use std::sync::{LazyLock, Mutex}; + + /// One commit of a rebuild: a batch's, or the marker's. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub(crate) enum Step { + Batch, + Marker, + } + + #[derive(Debug, Default)] + struct Records { + events: Vec<(Step, usize)>, + keys: usize, + } + + static RECORDS: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + + fn with(path: &Path, work: impl FnOnce(&mut Records) -> R) -> R { + let mut records = RECORDS + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + work(records.entry(path.to_path_buf()).or_default()) + } + + /// Drops what the store's path has recorded, so the next rebuild is read alone. + pub(crate) fn reset(path: &Path) { + with(path, |records| *records = Records::default()); + } + + /// The `(step, rows indexed so far)` list, in the order the commits happened. + pub(crate) fn events(path: &Path) -> Vec<(Step, usize)> { + with(path, |records| records.events.clone()) + } + + /// How many index keys the rebuilds at this path have written. + pub(crate) fn keys(path: &Path) -> usize { + with(path, |records| records.keys) + } + + pub(super) fn batch(path: &Path, rows: usize) { + with(path, |records| records.events.push((Step::Batch, rows))); + } + + pub(super) fn marker(path: &Path, rows: usize) { + with(path, |records| records.events.push((Step::Marker, rows))); + } + + pub(super) fn put_keys(path: &Path, written: usize) { + with(path, |records| records.keys += written); + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + + use super::*; + + use crate::LmdbBackend; + use crate::lmdb::rendezvous; + use crate::repositories::EventRepository; + + /// A store on a temporary directory, with the NIP-50 gate on. + async fn backend() -> (tempfile::TempDir, LmdbBackend) { + let directory = tempfile::tempdir().expect("tempdir"); + let backend = LmdbBackend::open(directory.path()) + .await + .expect("the store opens"); + (directory, backend) + } + + #[test] + fn the_tokenizer_reduces_content_to_lowercase_alphanumeric_runs() { + assert_eq!(terms("Bitcoin Lightning"), vec!["bitcoin", "lightning"]); + assert_eq!( + terms("#Bitcoin & Lightning Network"), + vec!["bitcoin", "lightning", "network"] + ); + assert_eq!(terms("#LightningNetwork"), vec!["lightningnetwork"]); + assert_eq!(terms("⚡"), Vec::::new()); + assert_eq!(terms("bitcoin bitcoin"), vec!["bitcoin"]); + assert_eq!(terms("study"), vec!["study"]); + assert_eq!(terms("studies"), vec!["studies"]); + assert_ne!(terms("study"), terms("studies")); + // R3's order is lowercase, then split: `İ` (U+0130) lowers to `i` plus a combining dot, + // which is not alphanumeric, so the run ends between it and the `x`. Splitting first and + // lowercasing each run yields the one token `i̇x`. + assert_eq!(terms("\u{130}x"), vec!["i", "x"]); + } + + #[test] + fn the_rank_is_the_smallest_cover_span_then_the_event_id() { + let query = terms("bitcoin lightning"); + assert_eq!(verdict("bitcoin lightning", &query), Some(2)); + assert_eq!(verdict("lightning bitcoin", &query), Some(2)); + assert_eq!(verdict("bitcoin x y lightning", &query), Some(4)); + assert_eq!(verdict("bitcoininveniice bitcoinisland", &query), None); + assert_eq!(verdict("lightning", &query), None); + + // Two tokens that share their first 64 bytes are one index key and two verdicts. + let first = format!("{}x", "c".repeat(64)); + let second = format!("{}y", "c".repeat(64)); + assert_eq!(terms(&first), vec![first.clone()]); + assert_eq!(verdict(&second, std::slice::from_ref(&first)), None); + assert_eq!(verdict(&first, std::slice::from_ref(&first)), Some(1)); + } + + #[tokio::test] + async fn two_concurrent_opens_rebuild_the_index_once() { + let (directory, backend) = backend().await; + // 1 500 rows, so one rebuild is two batches (1 000 then 500) and a second rebuild would + // be two more commits and a second marker: the counts tell one rebuild from two. + let records: Vec = (0..1_500) + .map(|index| { + let mut event = nostream_core::Event { + id: EventId::from_byte_array([0; 32]), + pubkey: nostream_core::Pubkey::from_byte_array([0; 32]), + created_at: 1_700_000_000 + index as i64, + kind: nostream_core::Kind(1), + tags: Vec::new(), + content: format!("token{index}"), + sig: nostream_core::Signature::from_byte_array([0; 64]), + }; + nostream_core::crypto::sign_event(&[3; 32], &mut event).expect("the event signs"); + EventRecord::from_event(&event) + }) + .collect(); + backend + .create_many(&records) + .await + .expect("the events store"); + + let path = backend.env().path().to_path_buf(); + probes::reset(&path); + // The C16 way: clear the index and delete the marker, so the next open rebuilds. + { + let env = backend.env(); + let read = env.read_txn().expect("a read transaction"); + let index = env + .open_database::( + &read, + Some(SEARCH_INDEX_DB), + ) + .expect("the index opens") + .expect("the index exists"); + let meta = env + .open_database::(&read, Some("meta")) + .expect("meta opens") + .expect("meta exists"); + drop(read); + let mut write = env.write_txn().expect("a write transaction"); + index.clear(&mut write).expect("the index clears"); + meta.delete(&mut write, SEARCH_INDEX_MARKER) + .expect("the marker clears"); + write.commit().expect("the commit"); + } + drop(backend); + + // Both opens are past `Env::open` before either reads the marker, so the second is not + // merely late — it is inside the window an unserialised read-modify-write loses. + rendezvous::arm(&path, 2); + let (first, second) = tokio::join!( + LmdbBackend::open(directory.path()), + LmdbBackend::open(directory.path()) + ); + let first = first.expect("the first store opens"); + let second = second.expect("the second store opens"); + + let steps = probes::events(&path); + let batches = steps + .iter() + .filter(|(step, _)| *step == probes::Step::Batch) + .count(); + let markers = steps + .iter() + .filter(|(step, _)| *step == probes::Step::Marker) + .count(); + assert_eq!(batches, 2, "one rebuild writes two batches: {steps:?}"); + assert_eq!(markers, 1, "and one marker: {steps:?}"); + assert_eq!( + steps.last(), + Some(&(probes::Step::Marker, 1_500)), + "the marker carries the whole store, written after the batches" + ); + drop((first, second)); + } + + #[tokio::test] + async fn a_rebuild_indexes_in_batches_and_marks_last() { + let (directory, backend) = backend().await; + let records: Vec = (0..SEARCH_REINDEX_BATCH + 1) + .map(|index| { + let mut event = nostream_core::Event { + id: EventId::from_byte_array([0; 32]), + pubkey: nostream_core::Pubkey::from_byte_array([0; 32]), + created_at: 1_700_000_000 + index as i64, + kind: nostream_core::Kind(1), + tags: Vec::new(), + content: format!("token{index}"), + sig: nostream_core::Signature::from_byte_array([0; 64]), + }; + nostream_core::crypto::sign_event(&[3; 32], &mut event).expect("the event signs"); + EventRecord::from_event(&event) + }) + .collect(); + backend + .create_many(&records) + .await + .expect("the events store"); + + let path = backend.env().path().to_path_buf(); + probes::reset(&path); + // The C16 way: clear the index and delete the marker, so the next open rebuilds. + { + let env = backend.env(); + let read = env.read_txn().expect("a read transaction"); + let index = env + .open_database::( + &read, + Some(SEARCH_INDEX_DB), + ) + .expect("the index opens") + .expect("the index exists"); + let meta = env + .open_database::(&read, Some("meta")) + .expect("meta opens") + .expect("meta exists"); + drop(read); + let mut write = env.write_txn().expect("a write transaction"); + index.clear(&mut write).expect("the index clears"); + meta.delete(&mut write, SEARCH_INDEX_MARKER) + .expect("the marker clears"); + write.commit().expect("the commit"); + } + drop(backend); + + let reopened = LmdbBackend::open(directory.path()) + .await + .expect("the store reopens"); + assert_eq!( + probes::events(&path), + vec![ + (probes::Step::Batch, 1000), + (probes::Step::Batch, 1001), + (probes::Step::Marker, 1001) + ], + "two batch commits, then the marker carrying the whole store's row count" + ); + assert!( + probes::keys(&path) >= 1001, + "the rebuild wrote a key per token: {}", + probes::keys(&path) + ); + { + let env = reopened.env(); + let read = env.read_txn().expect("a read transaction"); + let meta = env + .open_database::(&read, Some("meta")) + .expect("meta opens") + .expect("meta exists"); + assert_eq!( + meta.get(&read, SEARCH_INDEX_MARKER) + .expect("the marker reads"), + Some(&[SEARCH_INDEX_VERSION][..]) + ); + } + drop(reopened); + + probes::reset(&path); + let reopened = LmdbBackend::open(directory.path()) + .await + .expect("the store reopens again"); + assert_eq!( + probes::events(&path), + Vec::new(), + "a valid marker skips the rebuild" + ); + assert_eq!(probes::keys(&path), 0, "and writes no key"); + { + let env = reopened.env(); + let read = env.read_txn().expect("a read transaction"); + let index = env + .open_database::( + &read, + Some(SEARCH_INDEX_DB), + ) + .expect("the index opens") + .expect("the index exists"); + assert_eq!( + index.len(&read).expect("the index's length"), + SEARCH_REINDEX_BATCH as u64 + 1, + "the index still holds its keys" + ); + } + } +} diff --git a/crates/nostream-storage/tests/parity_lmdb.rs b/crates/nostream-storage/tests/parity_lmdb.rs index 804e0859..5a962695 100644 --- a/crates/nostream-storage/tests/parity_lmdb.rs +++ b/crates/nostream-storage/tests/parity_lmdb.rs @@ -19,34 +19,43 @@ use chrono::{DateTime, Duration, Utc}; use serde_json::{Value, json}; use common::{Repos, fixture, settings_from_fixture}; +use heed::Database; +use heed::types::Bytes; +use nostream_config::Settings; use nostream_core::{Event, EventId, Filter, Kind, Pubkey, Signature}; use nostream_storage::LmdbBackend; use nostream_storage::StorageError; +use nostream_storage::lmdb::{SEARCH_INDEX_DB, SEARCH_INDEX_MARKER, SEARCH_INDEX_VERSION}; use nostream_storage::repositories::{ CreateInviteCodeOptions, DeliveryLogFilters, DvmJobStatus, EventRecord, EventRepository, EventRetentionOptions, FieldUpdate, InviteCodeRepository, Invoice, InvoiceRepository, InvoiceStatus, InvoiceUnit, NewDeliveryLogEntry, NewReport, Nip05Verification, NotificationDeliveryLogRepository, NotificationDeliveryStatus, NotificationOutboxRepository, - NotificationTargetType, ReconciliationItem, ReportType, UserRepository, collect_records, + NotificationTargetType, ReconciliationItem, ReportType, SearchRule, UserRepository, + collect_records, }; mod common; -/// The entries this arm cannot replay, fixed by index: 21 is `search` (NIP-50 is out of scope), -/// 48 and 50 are `user.tos_accepted_at` (raw SQL, no seam form), and 112 and 115 are the two -/// `transaction` groups the capture rolled back — a seam with no rollback cannot reproduce them. +/// The entries this arm cannot replay, fixed by index: 48 and 50 are `user.tos_accepted_at` (raw +/// SQL, no seam form), and 112 and 115 are the two `transaction` groups the capture rolled back — +/// a seam with no rollback cannot reproduce them. /// The two groups it committed (111, 117) are replayed: their nested ops in order leave exactly /// the state the commit left, which is why entries after 111 observe its rows. -// shortcut: five fixture entries are excluded by index and the count is a constant; upgrade +/// +/// Entry 21 — the one filter carrying `search` — is replayed now: the LMDB store answers NIP-50 +/// from its own term index (`lmdb/search.rs`), which reproduces the capture's token predicate and +/// its recorded order on this corpus, so the exclusion has nothing left to stand on. +// shortcut: four fixture entries are excluded by index and the count is a constant; upgrade // when the capture gains an entry, which fails the count assertion and forces this list to be // re-derived rather than quietly skipped. -const EXCLUDED: [usize; 5] = [21, 48, 50, 112, 115]; +const EXCLUDED: [usize; 4] = [48, 50, 112, 115]; /// The entries that are replayed; `EXCLUDED.len() + REPLAYED` is the fixture's entry count. -const REPLAYED: usize = 113; +const REPLAYED: usize = 114; /// The exclusion list as the marker line prints it. -const EXCLUDED_MARKER: &str = "21,48,50,112,115"; +const EXCLUDED_MARKER: &str = "48,50,112,115"; /// Replays a transaction group the capture committed. The group's comparison is the aggregate /// `{failed: false, results: [...]}`, and running its nested ops in order through the seam @@ -94,6 +103,120 @@ fn event(secret: u8, kind: u64, created_at: i64) -> EventRecord { EventRecord::from_event(&event) } +/// A signed event of the given kind and creation time, carrying `content`. +fn event_with_content(secret: u8, kind: u64, created_at: i64, content: &str) -> EventRecord { + event_with_tags(secret, kind, created_at, content, Vec::new()) +} + +/// [`event_with_content`], carrying `tags`. +fn event_with_tags( + secret: u8, + kind: u64, + created_at: i64, + content: &str, + tags: Vec>, +) -> EventRecord { + let mut event = Event { + id: EventId::from_byte_array([0; 32]), + pubkey: Pubkey::from_byte_array([0; 32]), + created_at, + kind: Kind(kind), + tags, + content: content.to_owned(), + sig: Signature::from_byte_array([0; 64]), + }; + nostream_core::crypto::sign_event(&[secret; 32], &mut event).expect("the event signs"); + EventRecord::from_event(&event) +} + +/// The filter a search test sends. +fn search(query: &str) -> Filter { + Filter { + search: Some(query.to_owned()), + ..Filter::default() + } +} + +/// The ids of the records a read returned, in the order it returned them. +fn ids_of(records: &[EventRecord]) -> Vec { + records.iter().map(|record| record.event.id).collect() +} + +/// One store on a temporary directory, with `settings` as its gates. +async fn open_with(settings: Settings) -> (tempfile::TempDir, LmdbBackend) { + let directory = tempfile::tempdir().expect("tempdir"); + let backend = LmdbBackend::open(directory.path()) + .await + .expect("the store opens") + .with_settings(Arc::new(settings)); + (directory, backend) +} + +/// The fixture's settings with one override replaced. +fn settings_with(path: &str, value: Value) -> Settings { + let mut fixture = fixture(); + fixture["settings"]["overrides"] + .as_object_mut() + .expect("the fixture's overrides are an object") + .insert(path.to_owned(), value); + settings_from_fixture(&fixture) +} + +/// The records one filter's page returns, in the page's order. +async fn rows_of(backend: &LmdbBackend, filter: &Filter) -> Vec { + collect_records( + backend + .find_by_filters(std::slice::from_ref(filter)) + .await + .expect("the page"), + ) + .await + .expect("the stream") +} + +/// The term index and the `meta` database, opened by name: `Databases` is crate-private. +fn named_databases( + env: &heed::Env, + rtxn: &heed::RoTxn<'_>, +) -> (Database, Database) { + let index = env + .open_database::(rtxn, Some(SEARCH_INDEX_DB)) + .expect("the index opens") + .expect("the index database exists"); + let meta = env + .open_database::(rtxn, Some("meta")) + .expect("meta opens") + .expect("the meta database exists"); + (index, meta) +} + +/// How many `events_by_term` keys begin with `term`. +fn index_keys(backend: &LmdbBackend, term: &str) -> usize { + let env = backend.env(); + let rtxn = env.read_txn().expect("a read transaction"); + let (index, _) = named_databases(env, &rtxn); + index + .prefix_iter(&rtxn, term.as_bytes()) + .expect("the index reads") + .count() +} + +/// The ids a page returned, ascending, so a caller can compare a set before it compares an +/// order. +fn sorted_ids(records: &[EventRecord]) -> Vec { + let mut ids = ids_of(records); + ids.sort(); + ids +} + +/// The set `page` returned, and that it is in ascending `event_id` order. +fn assert_set_in_id_order(page: &[EventRecord], expected: &[&EventRecord], what: &str) { + let mut wanted: Vec = expected.iter().map(|record| record.event.id).collect(); + wanted.sort(); + assert_eq!(sorted_ids(page), wanted, "{what}: the set"); + assert_eq!(ids_of(page), sorted_ids(page), "{what}: ascending event_id"); +} + fn one_entry(outbox_id: Option) -> NewDeliveryLogEntry { NewDeliveryLogEntry { outbox_id, @@ -138,7 +261,6 @@ async fn the_typescript_script_replays_with_an_empty_diff() { "the fixture's entry count is the one the exclusion list was written against" ); for (index, name) in [ - (21, "events.find_by_filters"), (48, "user.tos_accepted_at"), (50, "user.tos_accepted_at"), (112, "transaction"), @@ -151,15 +273,14 @@ async fn the_typescript_script_replays_with_an_empty_diff() { ); } // Each entry is excluded exactly when the seam has no form for it: a transaction group - // the capture rolled back, the two raw-SQL probes, and the one filter carrying `search`. + // the capture rolled back, or one of the two raw-SQL probes. A filter carrying `search` is + // no longer one of them — the store answers it from its term index. for (index, op) in script.iter().enumerate() { let name = op["op"].as_str().expect("an op name"); let excluded_kind = match name { "transaction" => op["result"]["compare"]["failed"] != Value::Bool(false), "user.tos_accepted_at" => true, - _ => op["request"]["filters"] - .as_array() - .is_some_and(|filters| filters.iter().any(|filter| filter.get("search").is_some())), + _ => false, }; assert_eq!( EXCLUDED.contains(&index), @@ -1353,29 +1474,871 @@ async fn creating_a_job_that_exists_is_refused() { ); } -/// A non-empty `search` with the NIP-50 gate on is refused by every query method, the -/// reconciliation stream included: this backend cannot search, and returning a broader filter's -/// rows would answer a question it did not ask. +// --------------------------------------------------------------------------------------- +// NIP-50: the term index, the tokeniser and the rank, over a real store. +// --------------------------------------------------------------------------------------- + +/// The four rows the capture recorded for `{"search":"bitcoin lightning"}`, in its order — the +/// Postgres `ts_rank` order, which this store's proximity rank reproduces on this corpus. +const RECORDED_SEARCH_IDS: [&str; 4] = [ + "5454369c2445e12b001b51e5e48cf6722072ecdb187453c028cac8a12040633a", + "c004000000000000000000000000000000000000000000000000000000000000", + "cb41289d08a5784a73b7bac2bf97620f5c927487d0c5a029ba81f840fe21fa36", + "9ae1fef4857bb0174e8fb1221f8da1da6705ebef42e143de7a8f8fe965ba9ee3", +]; + +/// A search over the fixture's corpus returns the capture's four rows, in the capture's order, +/// and not the fifth row the relay's substring matcher would select (`bitcoininveniice`). #[tokio::test] -async fn reconciliation_refuses_a_search_the_gate_enables() { +async fn a_search_returns_the_recorded_rows_in_rank_order() { + let fixture = fixture(); let (_directory, backend) = open_store().await; - let filter = Filter { - search: Some("nostr".to_owned()), - ..Filter::default() + let corpus: Vec = fixture["corpus"] + .as_array() + .expect("corpus") + .iter() + .map(common::event_record) + .collect(); + backend + .create_many(&corpus) + .await + .expect("the corpus inserts"); + + let page = rows_of(&backend, &search("bitcoin lightning")).await; + let ids: Vec = page + .iter() + .map(|record| record.event.id.to_string()) + .collect(); + assert_eq!( + ids, RECORDED_SEARCH_IDS, + "R3, R5, R6: the recorded rows in rank order" + ); +} + +#[tokio::test] +async fn a_search_that_matches_nothing_yields_no_rows() { + let (_directory, backend) = open_store().await; + backend + .create_many(&[event_with_content(90, 1, 1_700_000_000, "bitcoin")]) + .await + .expect("the event stores"); + + let page = rows_of(&backend, &search("bitcoin zzznotpresent")).await; + assert!(page.is_empty(), "a term no body carries selects nothing"); +} + +#[tokio::test] +async fn a_search_applies_every_criterion_it_is_combined_with() { + let (_directory, backend) = open_store().await; + let e1 = event_with_tags( + 1, + 1, + 1_700_000_100, + "commonterm", + vec![vec!["t".to_owned(), "alpha".to_owned()]], + ); + let e2 = event_with_content(2, 0, 1_700_000_200, "commonterm"); + let e3 = event_with_content(1, 1, 1_700_000_300, "commonterm"); + let e4 = event_with_content(3, 1, 1_700_000_400, "commonterm"); + let e5 = event_with_content(1, 1, 1_700_000_500, "commonterm"); + backend + .create_many(&[e1.clone(), e2.clone(), e3.clone(), e4.clone(), e5.clone()]) + .await + .expect("the events store"); + nostream_storage::repositories::ReportRepository::create( + &backend, + &NewReport { + event_id: EventId::from_byte_array([44; 32]), + reporter_pubkey: Pubkey::from_byte_array([45; 32]), + reported_pubkey: Some(e4.event.pubkey), + reported_event_id: Some(e4.event.id), + report_type: ReportType::Spam, + weight: 1.0, + actionable: true, + }, + ) + .await + .expect("the report stores"); + + let cases = [ + ( + Filter { + kinds: Some(vec![Kind(1)]), + ..search("commonterm") + }, + vec![&e1, &e3, &e5], + "kinds", + ), + ( + search("commonterm"), + vec![&e1, &e2, &e3, &e5], + "search alone", + ), + ( + Filter { + ids: Some(vec![e2.event.id.to_string()]), + ..search("commonterm") + }, + vec![&e2], + "ids", + ), + ( + Filter { + authors: Some(vec![e2.event.pubkey.to_string()]), + ..search("commonterm") + }, + vec![&e2], + "authors", + ), + ( + Filter { + since: Some(1_700_000_200), + ..search("commonterm") + }, + vec![&e2, &e3, &e5], + "since", + ), + ( + Filter { + until: Some(1_700_000_200), + ..search("commonterm") + }, + vec![&e1, &e2], + "until", + ), + ( + Filter { + tags: BTreeMap::from([("#t".to_owned(), vec!["alpha".to_owned()])]), + ..search("commonterm") + }, + vec![&e1], + "#t", + ), + ]; + for (filter, expected, what) in cases { + let page = rows_of(&backend, &filter).await; + assert_set_in_id_order(&page, &expected, what); + } +} + +#[tokio::test] +async fn a_search_orders_by_proximity_then_event_id() { + let (_directory, backend) = open_store().await; + let adjacent_one = event_with_content(10, 1, 1_700_000_000, "bitcoin lightning"); + let adjacent_two = event_with_content(11, 1, 1_700_000_000, "lightning bitcoin"); + let adjacent_three = event_with_content(12, 1, 1_700_000_000, "lightning bitcoin"); + let spread = event_with_content(13, 1, 1_700_000_000, "bitcoin x y lightning"); + backend + .create_many(&[ + spread.clone(), + adjacent_three.clone(), + adjacent_one.clone(), + adjacent_two.clone(), + ]) + .await + .expect("the events store"); + + let mut tied = [adjacent_one, adjacent_two, adjacent_three]; + tied.sort_by_key(|record| record.event.id); + for query in ["bitcoin lightning", "Bitcoin LIGHTNING"] { + let page = rows_of(&backend, &search(query)).await; + let expected = [&tied[0], &tied[1], &tied[2], &spread]; + assert_eq!( + ids_of(&page), + expected + .iter() + .map(|record| record.event.id) + .collect::>(), + "{query}: three span-2 rows in id order, then the span-4 row" + ); + } +} + +#[tokio::test] +async fn a_search_finds_a_term_at_the_end_of_a_large_content() { + let (_directory, backend) = open_store().await; + let long = format!("{} needle", "a".repeat(200_000)); + let long_run = event_with_content(20, 1, 1_700_000_000, &long); + let hundred = event_with_content(21, 1, 1_700_000_000, &"b".repeat(100)); + let first = format!("{}x", "c".repeat(64)); + let second = format!("{}y", "c".repeat(64)); + let shared_one = event_with_content(22, 1, 1_700_000_000, &first); + let shared_two = event_with_content(23, 1, 1_700_000_000, &second); + backend + .create_many(&[ + long_run.clone(), + hundred.clone(), + shared_one.clone(), + shared_two.clone(), + ]) + .await + .expect("the events store"); + + for (query, expected, what) in [ + ( + "needle".to_owned(), + &long_run, + "a token after a 200 000-character run is found", + ), + ( + "b".repeat(100), + &hundred, + "a term longer than the key cap is found", + ), + ( + first.clone(), + &shared_one, + "the first of two tokens sharing 64 bytes", + ), + ( + second.clone(), + &shared_two, + "the second of two tokens sharing 64 bytes", + ), + ] { + let page = rows_of(&backend, &search(&query)).await; + assert_eq!(ids_of(&page), vec![expected.event.id], "{what}"); + } +} + +#[tokio::test] +async fn the_term_index_follows_create_replace_and_retention() { + let (_directory, backend) = open_store().await; + + let alpha = event_with_content(30, 1, 1_700_000_000, "uniquealpha"); + EventRepository::create(&backend, &alpha) + .await + .expect("the event stores"); + assert_eq!( + ids_of(&rows_of(&backend, &search("uniquealpha")).await), + vec![alpha.event.id] + ); + assert_eq!(index_keys(&backend, "uniquealpha"), 1, "one posting"); + + let beta = event_with_content(31, 0, 1_700_000_000, "uniquebeta"); + assert_eq!( + EventRepository::upsert(&backend, &beta) + .await + .expect("the upsert"), + 1 + ); + let gamma = event_with_content(31, 0, 1_700_000_100, "uniquegamma"); + assert_eq!( + EventRepository::upsert(&backend, &gamma) + .await + .expect("the replacement"), + 1 + ); + assert!(rows_of(&backend, &search("uniquebeta")).await.is_empty()); + assert_eq!( + index_keys(&backend, "uniquebeta"), + 0, + "the replaced row's postings left" + ); + assert_eq!( + ids_of(&rows_of(&backend, &search("uniquegamma")).await), + vec![gamma.event.id] + ); + assert_eq!(index_keys(&backend, "uniquegamma"), 1); + + let delta = event_with_content(32, 1, 1_700_000_000, "uniquedelta") + .with_expires_at(Utc::now().timestamp() - 1); + backend + .create_many(std::slice::from_ref(&delta)) + .await + .expect("the event stores"); + let counts = backend + .delete_expired_and_retained(&EventRetentionOptions { + max_days: Some(1), + kind_whitelist: None, + pubkey_whitelist: None, + }) + .await + .expect("the retention pass"); + assert_eq!(counts.expired, 1); + assert!(rows_of(&backend, &search("uniquedelta")).await.is_empty()); + assert_eq!( + index_keys(&backend, "uniquedelta"), + 0, + "the retained row's postings left" + ); + + let token = event_with_content(33, 1, 1_700_000_000, "uniquetoken"); + EventRepository::create(&backend, &token) + .await + .expect("the event stores"); + assert_eq!( + backend + .delete_by_pubkey_and_ids(&token.event.pubkey, &[token.event.id]) + .await + .expect("the soft delete"), + 1 + ); + assert_eq!( + ids_of(&rows_of(&backend, &search("uniquetoken")).await), + vec![token.event.id], + "a soft-deleted row is still found, as `find_by_filters` reads it" + ); + assert_eq!( + index_keys(&backend, "uniquetoken"), + 1, + "the soft delete rewrites the value, not the terms" + ); +} + +#[tokio::test] +async fn a_search_page_honours_the_limit() { + let (_directory, backend) = open_store().await; + let records: Vec = (0..5) + .map(|step| event_with_content(40 + step, 1, 1_700_000_000 + i64::from(step), "commonterm")) + .collect(); + backend + .create_many(&records) + .await + .expect("the events store"); + let mut by_id = sorted_ids(&records); + + let page = rows_of(&backend, &search("commonterm")).await; + assert_eq!(ids_of(&page), by_id, "no limit: every row, in id order"); + + let empty = rows_of( + &backend, + &Filter { + limit: Some(0), + ..search("commonterm") + }, + ) + .await; + assert!(empty.is_empty(), "limit 0 yields no rows"); + + by_id.truncate(2); + let two = rows_of( + &backend, + &Filter { + limit: Some(2), + ..search("commonterm") + }, + ) + .await; + assert_eq!(ids_of(&two), by_id, "limit 2: the two best by (rank, id)"); +} + +/// A key is the whole term a row wrote, so `cats` sorts inside `cat`'s index range but is not a +/// `cat` posting: the walk compares the posting's own term before it accepts the row. +#[tokio::test] +async fn a_search_counts_a_row_once_though_a_longer_token_shares_its_prefix() { + let (_directory, backend) = open_store().await; + let both = event_with_content(80, 1, 1_700_000_000, "cat cats"); + backend + .create_many(std::slice::from_ref(&both)) + .await + .expect("the event stores"); + + let page = rows_of(&backend, &search("cat")).await; + assert_eq!( + ids_of(&page), + vec![both.event.id], + "one row for the id, not one per posting" + ); + assert_eq!( + backend + .count_by_filters(&[search("cat")]) + .await + .expect("the count"), + 1, + "and the id is counted once" + ); +} + +/// The prefix range's other half: a longer token's posting must not be accepted as the shorter +/// term, and the longer token is still found by its own term. +#[tokio::test] +async fn a_search_does_not_match_a_longer_token_that_shares_its_prefix() { + let (_directory, backend) = open_store().await; + let longer = event_with_content(81, 1, 1_700_000_000, "cats"); + backend + .create_many(std::slice::from_ref(&longer)) + .await + .expect("the event stores"); + + assert!( + rows_of(&backend, &search("cat")).await.is_empty(), + "`cats` has `cat`'s index prefix and not its token" + ); + assert_eq!( + ids_of(&rows_of(&backend, &search("cats")).await), + vec![longer.event.id], + "the longer token is found by its own term" + ); +} + +/// A duplicate posting takes a slot in the limited heap and evicts a distinct row before the +/// search page's union removes the duplicate, so a `limit: 2` page must still hold both rows. +/// The duplicated row has to be the id-smaller one for the eviction to happen — at equal rank +/// the heap drops the larger id — so the construction asserts that. +#[tokio::test] +async fn a_limited_search_page_holds_a_distinct_row_per_slot() { + let (_directory, backend) = open_store().await; + let both = event_with_content(83, 1, 1_700_000_000, "cat cats"); + let short = event_with_content(82, 1, 1_700_000_100, "cat"); + backend + .create_many(&[both.clone(), short.clone()]) + .await + .expect("the events store"); + assert!( + both.event.id < short.event.id, + "the duplicated row's id sorts first, so its second posting is the heap's worst entry" + ); + + let page = rows_of( + &backend, + &Filter { + limit: Some(2), + ..search("cat") + }, + ) + .await; + let mut expected = vec![both.event.id, short.event.id]; + expected.sort(); + assert_eq!(ids_of(&page), expected, "two slots, two distinct rows"); +} + +#[tokio::test] +async fn a_count_applies_the_search_predicate() { + let (directory, backend) = open_with(settings_from_fixture(&fixture())).await; + let live = event_with_content(50, 1, 1_700_000_000, "matchterm"); + let deleted = event_with_content(51, 1, 1_700_000_100, "matchterm more"); + let other = event_with_content(52, 1, 1_700_000_200, "otherthing"); + backend + .create_many(&[live, deleted.clone(), other]) + .await + .expect("the events store"); + backend + .delete_by_pubkey_and_ids(&deleted.event.pubkey, &[deleted.event.id]) + .await + .expect("the soft delete"); + + assert_eq!( + backend + .count_by_filters(&[search("matchterm")]) + .await + .expect("the count"), + 1, + "the live match is counted, the soft-deleted one is not" + ); + drop(backend); + + let reopened = LmdbBackend::open(directory.path()) + .await + .expect("the store reopens") + .with_settings(Arc::new(settings_with("nip50.enabled", json!(false)))); + assert_eq!( + reopened + .count_by_filters(&[search("matchterm")]) + .await + .expect("the gate-off count"), + 2, + "with the gate off the field is ignored, as Postgres ignores it" + ); + assert_eq!( + reopened + .count_by_filters(&[Filter::default()]) + .await + .expect("the unfiltered count"), + 2, + "and the count is the count of the other criteria's rows" + ); + + // A searched filter is counted from the term index, whose keys prove no tag family, so its + // tag criteria are the body's: the tagged row alone is counted, not every row the term + // index proposes. + let (_tagged_directory, tagged) = open_store().await; + let tagged_row = event_with_tags( + 53, + 1, + 1_700_000_300, + "matchterm", + vec![vec!["t".to_owned(), "alpha".to_owned()]], + ); + let untagged_row = event_with_content(54, 1, 1_700_000_400, "matchterm"); + tagged + .create_many(&[tagged_row, untagged_row]) + .await + .expect("the tagged corpus stores"); + let tagged_filter = Filter { + tags: BTreeMap::from([("#t".to_owned(), vec!["alpha".to_owned()])]), + ..search("matchterm") }; + assert_eq!( + tagged + .count_by_filters(&[search("matchterm")]) + .await + .expect("the search's count"), + 2, + "both rows carry the term" + ); + assert_eq!( + tagged + .count_by_filters(&[tagged_filter]) + .await + .expect("the counted filter"), + 1, + "the tagged row alone: the untagged row carrying the term is not counted" + ); +} - let error = backend - .find_reconciliation_items(&filter) +#[tokio::test] +async fn a_limited_count_counts_the_rank_page() { + let (_directory, backend) = open_store().await; + let spread: Vec = (1..=4) + .map(|step| { + event_with_content( + 60 + step, + 1, + 1_700_000_000 + i64::from(step), + "alpha x y z beta", + ) + }) + .collect(); + let m5 = event_with_content(70, 1, 1_700_000_500, "alpha beta"); + let m6 = event_with_content(71, 1, 1_700_000_600, "alpha beta"); + let mut all = spread.clone(); + all.push(m5.clone()); + all.push(m6.clone()); + backend.create_many(&all).await.expect("the events store"); + + let limited = Filter { + limit: Some(2), + ..search("alpha beta") + }; + let page = rows_of(&backend, &limited).await; + let mut best = vec![m5.event.id, m6.event.id]; + best.sort(); + assert_eq!(ids_of(&page), best, "the rank-best two, not the newest two"); + + let mut first_filter = [spread[0].event.id, spread[1].event.id]; + first_filter.sort(); + let total = backend + .count_by_filters(&[ + Filter { + ids: Some(first_filter.iter().map(ToString::to_string).collect()), + ..Filter::default() + }, + limited, + ]) + .await + .expect("the count"); + assert_eq!( + total, 4, + "the limited branch contributes its rank page's two rows" + ); +} + +/// A search is applied by the reconciliation stream too, and its rows arrive oldest first: the +/// stream keeps its created-at order and its batches, and the matcher refuses the rest. +#[tokio::test] +async fn reconciliation_applies_the_search_predicate() { + let (_directory, backend) = open_store().await; + let earlier = event_with_content(70, 1, 1_700_000_000, "reachableterm"); + let between = event_with_content(71, 1, 1_700_000_050, "unrelated"); + let later = event_with_content(72, 1, 1_700_000_100, "reachableterm again"); + backend + .create_many(&[later, earlier.clone(), between]) .await - .err() - .expect("the refusal"); + .expect("the events store"); + + let items = common::collect( + backend + .find_reconciliation_items(&search("reachableterm")) + .await + .expect("the stream"), + ) + .await + .expect("the items"); + + assert_eq!( + items.len(), + 2, + "only the two rows whose bodies carry the term" + ); + assert_eq!( + items.iter().map(|item| item.created_at).collect::>(), + vec![1_700_000_000, 1_700_000_100], + "oldest first, as the stream promises" + ); + assert_eq!(items[0].event_id, earlier.event.id); +} + +#[tokio::test] +async fn a_blank_search_is_ignored_and_a_tokenless_query_selects_nothing() { + let (_directory, backend) = open_store().await; + let the_cat = event_with_content(80, 1, 1_700_000_000, "the cat"); + let a_cat = event_with_content(81, 1, 1_700_000_100, "a cat"); + let bitcoin = event_with_content(82, 1, 1_700_000_200, "bitcoin"); + backend + .create_many(&[the_cat.clone(), a_cat, bitcoin.clone()]) + .await + .expect("the events store"); + + for blank in ["", " "] { + assert_eq!( + rows_of(&backend, &search(blank)).await.len(), + 3, + "{blank:?} is not a search, so every row of the other criteria arrives" + ); + } assert!( - matches!( - error, - StorageError::NotImplemented { - module: "storage::lmdb::nip50_search" - } - ), - "a search the gate enables is NotImplemented, got {error:?}" + rows_of(&backend, &search("!!!")).await.is_empty(), + "a query whose every character is punctuation has no token, so it selects nothing" + ); + assert_eq!( + ids_of(&rows_of(&backend, &search("the")).await), + vec![the_cat.event.id], + "`simple` has no stop list, so `the` is an ordinary token" + ); + + let padded = format!("{} bitcoin", ".".repeat(250)); + assert!( + rows_of(&backend, &search(&padded)).await.is_empty(), + "the 256-character cut leaves `bitco`, which is not the token `bitcoin`" + ); + assert_eq!( + ids_of(&rows_of(&backend, &search("bitcoin")).await), + vec![bitcoin.event.id], + "and the unprefixed query still finds it" + ); +} + +#[tokio::test] +async fn a_search_never_matches_a_tag_value() { + let (_directory, backend) = open_store().await; + let record = event_with_tags( + 85, + 1, + 1_700_000_000, + "hello world", + vec![vec!["t".to_owned(), "bitcoin".to_owned()]], + ); + backend + .create_many(std::slice::from_ref(&record)) + .await + .expect("the event stores"); + + assert!( + rows_of(&backend, &search("bitcoin")).await.is_empty(), + "the index is built from `content`, so a tag value is never a posting" + ); + assert_eq!( + ids_of(&rows_of(&backend, &search("hello")).await), + vec![record.event.id] + ); +} + +#[tokio::test] +async fn a_search_under_a_language_the_index_cannot_reproduce_is_refused() { + let (directory, backend) = open_with(settings_with("nip50.language", json!("english"))).await; + let record = event_with_content(90, 1, 1_700_000_000, "study"); + backend + .create_many(std::slice::from_ref(&record)) + .await + .expect("the event stores"); + + let filter = search("study"); + let page = backend.find_by_filters(std::slice::from_ref(&filter)).await; + let count = backend + .count_by_filters(std::slice::from_ref(&filter)) + .await; + let reconciliation = backend.find_reconciliation_items(&filter).await; + + for error in [ + page.err().unwrap_or_else(|| panic!("the page is refused")), + count + .err() + .unwrap_or_else(|| panic!("the count is refused")), + reconciliation + .err() + .expect("the stream is refused, not its items"), + ] { + assert!( + matches!(error, StorageError::InvalidArgument { .. }), + "a language the index cannot reproduce is InvalidArgument, got {error:?}" + ); + assert!( + error.to_string().contains("simple"), + "the message names the configuration: {error}" + ); + } + drop(backend); + + let reopened = LmdbBackend::open(directory.path()) + .await + .expect("the store reopens") + .with_settings(Arc::new(settings_from_fixture(&fixture()))); + assert!( + rows_of(&reopened, &search("studies")).await.is_empty(), + "`simple` has no stemmer, so `studies` does not find `study`" + ); + assert_eq!( + ids_of(&rows_of(&reopened, &search("study")).await), + vec![record.event.id] + ); +} + +#[tokio::test] +async fn the_search_rule_names_the_simple_configuration_when_the_gate_is_on() { + let (_directory, enabled) = open_with(settings_from_fixture(&fixture())).await; + assert_eq!( + enabled.search_rule(), + SearchRule::ToTsvector { + language: "simple".to_owned() + }, + "the declared rule is the rule the term index applies" + ); + + let directory = tempfile::tempdir().expect("tempdir"); + let bare = LmdbBackend::open(directory.path()) + .await + .expect("the store opens"); + assert_eq!( + bare.search_rule(), + SearchRule::SubstringTerms, + "with the gate off no predicate is applied, and the relay's matcher is what decides" + ); +} + +#[tokio::test] +async fn a_store_whose_index_was_maintained_with_the_gate_off_answers_the_search_when_it_is_on() { + let (directory, backend) = open_with(settings_with("nip50.enabled", json!(false))).await; + let record = event_with_content(95, 1, 1_700_000_000, "latentterm"); + backend + .create_many(std::slice::from_ref(&record)) + .await + .expect("the event stores"); + drop(backend); + + let reopened = LmdbBackend::open(directory.path()) + .await + .expect("the store reopens") + .with_settings(Arc::new(settings_from_fixture(&fixture()))); + assert_eq!( + ids_of(&rows_of(&reopened, &search("latentterm")).await), + vec![record.event.id], + "the write path indexes whether or not the read gate is on" + ); +} + +/// Three events whose contents are one distinctive token each, the store the two rebuild tests +/// start from, with the index forgotten the way a store written before it existed has none. +async fn store_without_an_index() -> (tempfile::TempDir, Vec) { + let (directory, backend) = open_with(settings_from_fixture(&fixture())).await; + let records: Vec = ["rebuiltalpha", "rebuiltbeta", "rebuiltgamma"] + .iter() + .enumerate() + .map(|(step, content)| event_with_content(100 + step as u8, 1, 1_700_000_000, content)) + .collect(); + backend + .create_many(&records) + .await + .expect("the events store"); + { + let env = backend.env(); + let rtxn = env.read_txn().expect("a read transaction"); + let (index, meta) = named_databases(env, &rtxn); + drop(rtxn); + let mut wtxn = env.write_txn().expect("a write transaction"); + index.clear(&mut wtxn).expect("the index clears"); + meta.delete(&mut wtxn, SEARCH_INDEX_MARKER) + .expect("the marker clears"); + wtxn.commit().expect("the commit"); + } + drop(backend); + (directory, records) +} + +#[tokio::test] +async fn a_store_written_without_the_index_is_rebuilt_on_open() { + let (directory, records) = store_without_an_index().await; + let reopened = LmdbBackend::open(directory.path()) + .await + .expect("the store reopens") + .with_settings(Arc::new(settings_from_fixture(&fixture()))); + + { + let env = reopened.env(); + let rtxn = env.read_txn().expect("a read transaction"); + let (index, meta) = named_databases(env, &rtxn); + assert_eq!( + index.len(&rtxn).expect("the index's length"), + 3, + "the index is populated before the first search, not lazily by it" + ); + assert_eq!( + meta.get(&rtxn, SEARCH_INDEX_MARKER) + .expect("the marker reads"), + Some(&[SEARCH_INDEX_VERSION][..]), + "and the marker records the rebuild" + ); + } + assert_eq!( + ids_of(&rows_of(&reopened, &search("rebuiltalpha")).await), + vec![records[0].event.id] + ); +} + +#[tokio::test] +async fn an_index_from_an_older_tokenizer_is_rebuilt_on_open() { + let (directory, records) = store_without_an_index().await; + { + let env = { + let backend = LmdbBackend::open(directory.path()) + .await + .expect("the store opens"); + backend.env().clone() + }; + let rtxn = env.read_txn().expect("a read transaction"); + let (index, meta) = named_databases(&env, &rtxn); + drop(rtxn); + let mut wtxn = env.write_txn().expect("a write transaction"); + meta.put(&mut wtxn, SEARCH_INDEX_MARKER, &[SEARCH_INDEX_VERSION + 1]) + .expect("the stale marker"); + let mut ghost = b"ghostterm".to_vec(); + ghost.extend([0u8; 32]); + index + .put(&mut wtxn, &ghost, &[]) + .expect("the stale posting"); + wtxn.commit().expect("the commit"); + } + + let reopened = LmdbBackend::open(directory.path()) + .await + .expect("the store reopens") + .with_settings(Arc::new(settings_from_fixture(&fixture()))); + { + let env = reopened.env(); + let rtxn = env.read_txn().expect("a read transaction"); + let (index, meta) = named_databases(env, &rtxn); + assert_eq!( + meta.get(&rtxn, SEARCH_INDEX_MARKER) + .expect("the marker reads"), + Some(&[SEARCH_INDEX_VERSION][..]), + "the rebuild repaired the marker before the first search" + ); + assert_eq!( + index.len(&rtxn).expect("the index's length"), + 3, + "the index was replaced, not extended: an add-only rebuild would read 4" + ); + assert_eq!( + index + .prefix_iter(&rtxn, b"ghostterm") + .expect("the index reads") + .count(), + 0, + "the posting no current row carries is gone" + ); + } + assert_eq!( + ids_of(&rows_of(&reopened, &search("rebuiltbeta")).await), + vec![records[1].event.id] ); } diff --git a/docs/rewrite/decisions/0011-lmdb-backend.md b/docs/rewrite/decisions/0011-lmdb-backend.md index c8f34cf4..4f53d114 100644 --- a/docs/rewrite/decisions/0011-lmdb-backend.md +++ b/docs/rewrite/decisions/0011-lmdb-backend.md @@ -49,7 +49,7 @@ author — inside the same read transaction that already holds the event row, in second store read per candidate. An embedded SQL store would have needed a second dependency and could still tear against the LMDB file. -The price is recorded with the decision: 29 named databases maintained by hand, `DISTINCT` +The price is recorded with the decision: 30 named databases maintained by hand, `DISTINCT` and `OFFSET` become cursor walks (`BTreeSet` for `find_actionable_targets`, a skip count for `find_pending_invoices`), and the columns' `CHECK` and foreign-key constraints move into Rust (§ Constraints move into Rust, below). @@ -173,7 +173,7 @@ transaction outlives a poll, and a dropped stream pins no pages. `Env::resize` is `unsafe` and this crate has one audited `unsafe` call, so the map is preallocated at `LMDB_MAP_SIZE = 64 GiB` and never grown (D6). The same single -`#[allow(unsafe_code)]` open sets `LMDB_MAX_DBS = 64` (the 29 named databases, comfortably +`#[allow(unsafe_code)]` open sets `LMDB_MAX_DBS = 64` (the 30 named databases, comfortably inside) and `LMDB_MAX_READERS = 512`. Because the map never grows, its approach has to be visible before the first failed write: @@ -215,6 +215,7 @@ The index is chosen from the filter, and which ranges can stop early follows fro | `authors` + `kinds` | `events_by_author`, one range per whole-pubkey criterion and kind, and one whole-tail range per prefix criterion | when every criterion is a whole pubkey, so a range is one `pubkey ‖ kind ‖ created_at` run; a prefix range spans every kind | | `authors` alone | `events_by_created_at` | yes — see below | | a `#` + letter criterion | `tags`, one range per criterion | for an exact value; a `#g` prefix spans values, so its range runs whole | +| `search` with the NIP-50 gate on | `events_by_term`, one range per posting of the first distinct query term (ADR 0012, D1) | never: a term's postings run in id order and the page is ordered by rank, so the page's own heap bounds it | | anything else | `events_by_created_at` | yes | Every hex criterion's bounds are widened to the field they name — 32 bytes for an id or a @@ -261,6 +262,21 @@ Two consequences of the same bound: criterion names one key per row. The tags index has no order that visits each row once, which is why the cap trades a slot past it for a body decode. +## NIP-50 search is a database of this store; its contract is ADR 0012 + +The store answers `Filter::search` from a thirtieth named database, `events_by_term`, keyed +`term ‖ event_id` with an empty value, maintained by `write_event` and `hard_delete` and rebuilt on +open when the `meta` marker `search-index-version` is stale. That is all this record needs to keep +about it, because the database count, the `LMDB_MAX_DBS` headroom and the `search` row of the index +table above are facts of the store's shape. + +Everything a reader would otherwise look for here — why an index rather than a scan-time filter, +the tokeniser and what it loses against Postgres' `simple`, the cover-span rank and its divergence +from `ts_rank`, the marker and what stale means, the batched rebuild and the open cost it adds, and +what would overturn any of it — is **ADR 0012 (`0012-nip50-search-index.md`, D1–D9)**. It is a +separate number because those are decisions about the store's on-disk format and its open cost, not +about the shape of a query. + ## Constraints move into Rust The store has no server to enforce the columns' `CHECK` and foreign-key constraints, so the @@ -324,14 +340,18 @@ Postgres can reach: raw SQL on `users.tos_accepted_at` and the serializable `tra groups. `tests/parity_lmdb.rs` opens a store on a `tempfile::TempDir`, inserts the fixture's -483-event corpus and replays **113 of 118** entries. The five it does not replay: +483-event corpus and replays **114 of 118** entries. The four it does not replay: |index|op|why the seam cannot run it| |---|---|---| -|21|`events.find_by_filters` with `search`|NIP-50 is out of scope; the backend answers `NotImplemented`| |48, 50|`user.tos_accepted_at`|raw SQL against `backend.master()`, no seam form| |112, 115|`transaction`|the capture **rolled these back**; reproducing a rollback needs a transaction type this job deliberately does not add| +Entry 21 — the one filter carrying `search` — **is** replayed, from the term index (ADR 0012, +D1–D4): its four rows and their order are the capture's, and the marker line reads +`LMDB_PARITY_COMPLETE replayed=114 excluded=48,50,112,115`. It was excluded while the backend +refused the query; the refusal is gone with job 028. + The other two `transaction` groups, 111 and 117, **committed** in the capture and are replayed: running their nested ops in order through the seam leaves exactly the state the commit left, and their comparison — the aggregate `{failed: false, results: [...]}` — is what @@ -341,13 +361,13 @@ cannot pass. What the transaction decision leaves unmeasured is rollback: the ar committed groups' effects, not that two calls can fail together. The target asserts every exclusion against the fixture — an entry is excluded exactly when -its op is one of the five above — so the list cannot grow without the diff showing it. +its op is one of the four above — so the list cannot grow without the diff showing it. ## Consequences - One environment, one write transaction, one error taxonomy; no second dependency, and no tear between two files. -- 29 named databases and their key encodings are maintained by hand; `DISTINCT`/`OFFSET` +- 30 named databases and their key encodings are maintained by hand; `DISTINCT`/`OFFSET` became cursor walks; two `CHECK`/foreign-key constraints became Rust code. - Postgres remains the reference: every query path is written against `sql.rs`'s predicates and orderings, and the shared executor is what proves it. diff --git a/docs/rewrite/decisions/0012-nip50-search-index.md b/docs/rewrite/decisions/0012-nip50-search-index.md new file mode 100644 index 00000000..2a47ad4c --- /dev/null +++ b/docs/rewrite/decisions/0012-nip50-search-index.md @@ -0,0 +1,350 @@ +# ADR 0012: NIP-50 search over the embedded store — a term index, a tokeniser and a rank (job 028) + +- **Status:** accepted (job 028, Phase 2) +- **Deciders:** the rewrite team, on the operator's decision of 2026-11-11 and the evidence in job 028's recon and spec +- **Date:** 2026-11-11 + +## Context + +ADR 0011 records the LMDB event store and leaves one read unanswered there: a filter carrying +`Filter::search` reached `admits` and came back `NotImplemented { module: +"storage::lmdb::nip50_search" }` on all three read paths. The capture answers the field on Postgres +with `to_tsvector @@ plainto_tsquery` and orders by `ts_rank`; its in-memory matcher applies a +lowercase substring rule instead, and the repository records that the two rules contain neither the +other (`repositories.rs`, `SearchRule`). The fixture's entry 21 is the Postgres answer over the +crate's own 483-event corpus — four rows, in `ts_rank` order — and the parity arm replays it now. + +This record is its own number rather than a section of 0011 because the change is to the store's +format and its open cost, not to the shape of a query: a named database is added, a key shape is +fixed, a marker joins `meta`, and `LmdbBackend::open` gains a step that a store written before the +index pays once. Those are the parts a later reader judges by the on-disk contract, and the next one +is that reader's to reopen. + +## D1 — A thirtieth named database, `events_by_term`, keyed `term ‖ event_id` + +`Databases` gains one field and `Databases::create` one line +(`crates/nostream-storage/src/lmdb/mod.rs:141`, `:186`), so the environment holds 30 named databases +under `LMDB_MAX_DBS = 64` (`:54`). The key is one token of the event's `content` — truncated to 64 +bytes on a `char` boundary (D2) — followed by the row's 32-byte id, and the value is empty, as every +other index database's is (`crates/nostream-storage/src/lmdb/search.rs:64-94`). A term's postings +are therefore one contiguous byte range in id order, seekable as `term .. term ‖ 0xff×32`, and the +order the range is walked in is stable. + +The index is maintained by `write_event` and `hard_delete` (`lmdb/events.rs:248`, `:298`), the only +two functions that add or remove an `events` row, so `create`, `upsert`/`upsert_many` (which +hard-deletes the superseded row and writes the new one in one transaction) and +`delete_expired_and_retained` all reach it; a soft delete rewrites the value and leaves the terms, +because the content did not change. It is maintained **whether or not the NIP-50 gate is on**: the +gate is a read gate, so a store written with the feature off answers a search the moment it is +turned on, with no rebuild. + +**Alternatives rejected.** + +- **A scan-time token filter, no index and no rank.** This is the real alternative and it was built: + design A, commit `6b50424` on branch `factory/028a-nip50-scan`, ~96 added production lines across + `lmdb/events.rs`, `lmdb/mod.rs` and `repositories.rs` (the `SearchRule::TokenTerms` variant). It + answered the right *set* — the token rule selects exactly entry 21's four rows — and paid for it + in three ways. It had no rank, so its order was the store's page order (`created_at`), and the + fixture's recorded order could only be reproduced by re-sorting the capture's rows **inside the + test**; a limited search and a limited count therefore disagreed with Postgres on the branch's + set. Every candidate the plan offered cost a body decode, so a term carried by a handful of rows + walked `events_by_created_at` to the page bound, and a searched count walked its ranges whole. And + it could not grow into the rank without keeping that cost, because the scan has no posting to + narrow the candidate set with. +- **The core matcher's substring rule** (`Filter::matches`) was rejected on measurement: over the + same corpus it selects a fifth row, `f8ef9793…`, whose content carries `bitcoininveniice`, + `bitcoinisland` and `#LightningNetwork`. A backend that applied it could not pass the fixture. + +**Consequence.** Every event write now pays one `put` per distinct token of its content, and the +index holds one key per (term, row) — the 20 000-row measurement in D6 wrote 12 keys per row for a +12-token body. That is the write cost the read path buys back. + +**What would overturn it:** a benchmark showing the write amplification unacceptable on this +branch's own corpus, or a need for the index to carry tag keys as well (which changes the key shape +and makes the key, not the body, authoritative). Either is a store-format change with its own +record. + +## D2 — One hand-rolled tokeniser, `search::terms` + +Unicode lowercase of the whole text, **then** maximal runs of `char::is_alphanumeric()`, duplicates +dropped (`lmdb/search.rs:109-145`). The order is part of the rule, not an implementation detail: a +case mapping can turn one character into several, and `İ` (U+0130) lowers to `i` plus a combining +dot, which is not alphanumeric — so `İx` is the two tokens `i` and `x` here, and one token if the +runs were taken from the original text and each run lowered afterwards. It is what the write path +indexes, what the read path resolves queries with and what the verdict compares, so the index and +the predicate cannot disagree about a token. +**What that changes for real content.** A body containing a dotted `İ` (Turkish `İstanbul`, say) +now indexes as the two keys `i` and `stanbul` where the split-then-lower order produced the single +key `i`+U+0307+`stanbul` — one term became two, so a query for `stanbul` starts matching a body it +did not match before, and the store that indexed the old way answers differently until it rebuilds +(D5). No existing criterion observes this: the fixture's corpus has no such character, and every +test of the rule would pass under either order were it not pinned explicitly. + +Against Postgres' `simple` configuration it reproduces the no-stemmer, no-stop-list part — `the` is +an ordinary token and `study` does not match `studies` — and it loses the parser's classes (URLs, +e-mails, hosts) and any CJK segmentation. **On the fixture's 483-event corpus the two select the +same four rows, and no claim beyond that is made.** + +No dependency is added: no crate reproduces Postgres' parser, and this environment cannot verify one +against Postgres. A token is cut to `SEARCH_TERM_MAX_LEN = 64` bytes on a `char` boundary for the +key, which can only merge two keys, never split one: a shortened key *proposes* a candidate and the +body decides (D3). + +**Alternatives rejected:** an analyser crate such as `tantivy` — a second dependency and a rule this +environment cannot check against the configuration it claims to reproduce; and Postgres' parser +classes reproduced by hand — a parser is not a tokeniser, and hand-writing one buys a rule no test +here can compare. + +**What would overturn it:** a corpus or a deployment that needs stemming, a second language +configuration, or CJK segmentation. Then the tokeniser becomes a real analyser, the index's contents +change with it, and `SEARCH_INDEX_VERSION` (D5) is what makes every existing store rebuild. + +## D3 — The index narrows candidates and the body decides + +A row matches a search when every distinct query term is one of the tokens of its `content` — never +of a tag (`lmdb/search.rs:177-221`). The walk narrows in three stages: the **prefix range** of the +first distinct query term (which necessarily also visits longer terms sharing its bytes), then +**term identity** — a posting whose term slice is not the walked term's bytes is skipped before +anything else — then each other term's exact key for the same id, and finally `Matcher::matches`, +whose one body decode answers both the tag families and the search verdict (`lmdb/search.rs:323-382`). +Without the identity stage a row whose content holds both `cat` and `cats` reaches the keeper twice +for a query of `cat`: a count over-reports and a limited page can spend two slots on one row. A stale or truncated posting therefore cannot produce a false positive, +and completeness is the write path's (D1). A query whose tokens are empty — a punctuation-only +query, which `plainto_tsquery` turns into an empty tsquery — selects nothing and takes no walk. + +**Alternatives rejected:** making the index authoritative (a stored verdict per key) would remove +the body decode and would make every stale posting a wrong answer — the failure mode the truncation +in D2 exists to keep harmless. + +**What would overturn it:** an index exact enough to decide alone, which means no truncation, no +tokeniser change without a rebuild, and a key that carries the tag criteria too. That is D1's +trigger, not a separate decision. + +## D4 — The rank is the smallest cover span, and a tie breaks by `event_id` + +For a body that holds every term, the rank is the smallest window of consecutive tokens covering all +of them; the page is ordered by rank ascending, then `event_id` ascending, and keeps its own heap +because that order is not the index's key order (`lmdb/search.rs:177-221`, `:257-321`). + +**The fixture recorded `ts_rank`'s order and this differs.** It reproduces that order on this corpus +— spans 2, 2, 2, 4 for the four rows — which is **not** a claim of `ts_rank` parity: `ts_rank` also +weights term frequency and document length, the four recorded rows tie on frequency, and no Postgres +is reachable here to compare values. Two consequences follow, and both are deliberate: + +- a `limit` no longer truncates by `created_at` for a searched read: **on a limited search LMDB + ranks by (span, id) where Postgres keeps `created_at DESC`**; +- a limited count counts that rank page, where Postgres' limited branch counts its created-at + branch. The two agree on such a branch's size and can disagree on its set, which only a count over + more than one filter can see. `branch` is the one router, so the rows a limited count counts are + the rows the read returns. + +**Alternatives rejected.** A **frequency rank** reproduces neither: the four recorded rows tie on +term frequency, and a frequency order sorts them `5454369c…`, `9ae1fef4…`, `c004…`, `cb41289d…`, +which is not the recorded sequence. A **`ts_rank` clone** is rejected because its weights and its +document-length normalisation are Postgres' to define and this environment cannot measure them; a +rank that cannot be checked is a guess in a decision record. Ordering by `created_at` (design A) is +rejected because the fixture's order is not time order — that was the measurement that separated the +two designs. + +**What would overturn it:** a Postgres-compatible ranker landing — one that computes `ts_rank`'s +values and is verified against a live Postgres. The index does not have to carry term positions for +it (the body is decoded for the verdict), but the page's order and the limited count would both +become `ts_rank`'s, and the divergence above goes with a contract change, not a tweak. + +## D5 — `meta` carries `search-index-version`, and "stale" means absent or different + +`meta` holds `SEARCH_INDEX_MARKER = b"search-index-version"` whose one-byte value is +`SEARCH_INDEX_VERSION = 1` (`lmdb/search.rs:33-41`). `LmdbBackend::open` reads it in the transaction +that creates the databases (`lmdb/mod.rs:257-263`); when it matches, nothing else happens. + +**Stale** is the marker being absent — a store written by a binary that had no index — or holding +another byte — an index a previous tokeniser built. Both are replaced, not extended: `events_by_term` +is cleared in the rebuild's first write transaction, before that transaction writes a key. An +add-only rebuild would keep every key an older tokeniser wrote, so the index would be the union of +two rules forever, and no answer would show it, because the body verdict masks a stale posting. The +clear is what makes "rebuilt" mean **replaced**. + +The marker is written **last**, after the final batch commits. A rebuild killed after the clear +leaves keys missing and no valid marker, so the next open clears and rebuilds again; a marker written +first would let a half-built index answer a search. The probe tests read the marker after `open` and +before any search, which is what pins the ordering rather than the intent. + +**Alternatives rejected:** a per-row format byte inside the index value — the value is empty, and a +byte per posting would be paid on every key to answer a question one key answers once. A silent +add-only backfill is rejected above. + +**What would overturn it:** an incremental rebuild that knows which rows an older tokeniser indexed +differently. The clear would become a delta, and the marker would have to record more than one byte +— a new version, not an edit to this one. + +## D6 — The rebuild runs on open, in batches of 1000, and its cost is paid once + +`LmdbBackend::open` rebuilds when D5 says stale: it walks `events_by_created_at`, indexes +`SEARCH_REINDEX_BATCH = 1000` rows per write transaction (`lmdb/search.rs:52`, `:391-456`), and logs +one line before and one after, with the row count, the key count and the elapsed milliseconds, so a +slow first open is visible rather than mysterious. A fresh store writes the marker and indexes +nothing. + +**The open cost, measured.** A rebuild on this host indexes 20 000 rows of a 12-token prose body — +240 000 postings — in **1.37 s in the release profile** (14 600 rows/s) and 2.21 s in a debug build +(9 000 rows/s): the order of a second per 20 000 rows, about 7 s per 100 000 and about 70 s per +million on this machine. It is the open cost a store written before the index existed adds, **once**: +the marker makes the second open a read of one key, and a store written by this binary maintains the +index as it writes (D1), so it never rebuilds. + +**It blocks inside `open`, not in the background.** The rebuild runs on the caller's `open` (on a +blocking-pool thread that `open` awaits), so the first open after an upgrade returns late by that +duration rather than starting a daemon that answers searches eventually: a store of 20 000 events +opens about 1.4 s late, 100 000 about 7 s, a million about 70 s on this host. For any store where +that delay is unacceptable on a restart, the answer is an incremental or offline rebuild (below) and +not a background build that answers from a half-built index — the number is here so that decision is +taken in advance rather than discovered during one. + +**Alternatives rejected:** a background build after `open` returns — the store would answer searches +from a half-built index, which is the state D5's ordering exists to prevent. A per-row lazy build on +first search — the same defect, with the first search paying for it invisibly. + +**What would overturn it:** a production corpus where minutes on the first open is unacceptable to +operators. The answer then is either a background build with an explicit "search unavailable until +built" state, or a migration path (`#864`'s `pg-to-lmdb` slice) that builds the index as it inserts. + +## D7 — Gate off, blank, and tokenless queries follow Postgres + +With the NIP-50 gate off, or a `search` that trims to empty, every read behaves exactly as it does +without the field: this is `is_search`'s own rule (`postgres/sql.rs`), and it matters because the +COUNT handler forwards `search` to a backend with the gate off. A query whose every character is +punctuation is a search that selects nothing, not a blank one. + +The core matcher disagrees on the whitespace-only case (`Filter::matches` matches nothing for +`" "`); the spec follows Postgres, because the two backends' agreement is the property under test, +and this paragraph is where a reader meets that as a choice rather than a defect. + +**What would overturn it:** Postgres changing its rule, or a decision that the two backends may +answer a blank search differently. Neither is reachable from this tree. + +## D8 — Only `simple` is answered, and the declaration says so + +The LMDB gates carry the NIP-50 query length and whether `nip50.language` is the `simple` +configuration the index reproduces. When a search is answered under any other language the read +returns `InvalidArgument` naming `simple`, from `find_by_filters`, `count_by_filters` and +`find_reconciliation_items` alike; answering with simple-token semantics under `english` would be a +silent divergence from a stemmed Postgres predicate, which is the defect class this job removes. +`LmdbBackend::search_rule` declares `SearchRule::ToTsvector { language: "simple" }` when the gate is +on and `SubstringTerms` when it is off (`lmdb/mod.rs:453-461`), matching Postgres' shape, so the +relay withholds a `finish` hint over a searched read instead of claiming a completeness the +enumeration cannot establish. + +**Alternatives rejected:** a `SearchRule::TokenTerms` variant, which design A added to publish its +own rule. This store reproduces a Postgres rule instead of declaring a private one, and the variant +is not on this branch. + +**What would overturn it:** an index that reproduces a second configuration — stemmed `english` +first. The tokeniser becomes an analyser (D2), the marker's version moves (D5), and `search_rule` +has one more true answer. + +## D9 — The rebuild is serialised per store path, within a process + +Two `LmdbBackend::open` calls on one store, run at the same time, used to rebuild the index twice, +and the second rebuild is the dangerous one: the first writes the marker after its last batch, so a +caller that starts while that first rebuild is still committing batches clears an index the first is +filling and then writes a marker over a partly refilled one. The marker is the whole of D5's state, +so any open inside that window would skip the rebuild and serve an incomplete index with no error +anywhere. **Measured** at `a21b622`, with a store of 1 500 rows whose index and marker were cleared, +two concurrent opens recorded +`[(Batch,1000),(Batch,1000),(Batch,1500),(Batch,1500),(Marker,1500),(Marker,1500)]` — four batches +and two markers where one rebuild writes two and one. + +`open` therefore holds one `std::sync::Mutex` per store path across the marker read, its comparison +and the whole rebuild (`lmdb/mod.rs`, `INDEX_REBUILDS`), and releases it before the usage read. It is +per path rather than one process-wide lock, because a single lock would hold one store's cold start +against every other store's open and buy nothing; it is a `std::sync::Mutex` taken inside the +`spawn_blocking` closure, so nothing holds it across an `await`; and nothing else in the crate takes +it, so there is no ordering to get wrong. An entry is kept after use because the map is bounded by +the number of distinct store paths a process opens, which is the number of stores it owns. + +**Within a process is the whole of the contract**, and the protection one would assume is not there: +heed's process-wide `OPENED_ENV` cache keys an environment by canonical path and hands a second open +of the same directory the same `Env` (`heed-0.20.5/src/env.rs:40`, `:254-262`), so LMDB's +one-environment-per-path rule never sees the second caller and nothing below this decision +arbitrates the two. + +**A second process opening the same directory is outside this contract.** LMDB serialises write +transactions between processes but not this read-modify-write: two processes can each read an absent +marker and each rebuild. One process owns one environment, which is the deployment shape this store +has, and fencing a second one needs a protocol the store does not carry. + +**The rejected sentinel.** The design that would cover a second process is an `IN_PROGRESS` value +written to `search-index-version` by the rebuild's first write transaction — the one that clears the +index — and compared-and-set at the end, so a reader that finds `IN_PROGRESS` waits or refuses +instead of rebuilding. It is rejected because it needs a crash-recovery rule the store does not +have: a process killed mid-rebuild leaves `IN_PROGRESS` behind, and telling a dead writer from a +live one takes a lease with a timestamp and a boot identity, which is a second decision with its own +failure modes. With no second writer to justify it, the lock is the whole answer. + +**What would overturn it:** more than one process opening one store — a sidecar, a CLI against a +live store, or a relay started twice on one volume. The sentinel above, or a directory lock the +operating system drops when its holder dies, then stops being optional. + +**A second handle in the same process.** The lock closes the case the static stage found — two `open` +calls rebuilding at once — but it is taken by `open`, not by writers. A second `LmdbBackend` over the +same path (same heed `Env`) that *writes* while the first is inside its rebuild can therefore have its +postings swept by the rebuild's clear after the rebuild's scan snapshot was taken, and the marker then +describes an index missing them. Nothing in production opens a store twice today, and the window needs +a write to land inside a rebuild, so this is recorded rather than fixed here: **the contract is one +handle per store per process, and a rebuild must not overlap writes.** Closing it properly means the +rebuild taking the write path's own lock or indexing from a snapshot that later writers cannot +invalidate, and it needs its own failing test. + +## Consequences + +- One more named database, its key shape and its marker are maintained by hand; the environment + stays inside `LMDB_MAX_DBS = 64` with room. +- A searched read is a range over one term's postings, and its page is bounded by the same + `MAX_PAGE_KEYS` as every other page; a row is decoded only when every query term's posting is + present. +- A searched read's order is `(span, event_id)`, and a limited searched count counts that page. + Postgres orders a limited search by `created_at DESC`; the two agree on the set of the unlimited + form and can differ on a limited branch's set. +- Every event write pays a `put` per distinct token; a store written before the index pays one + open-time rebuild of about a second per 20 000 rows (measured above), once. +- The capture's `ts_rank` values are still recorded only as an order in the fixture. No arm here + proves the two rank *numbers* agree, and none claims it. +- Two opens of one store serialise on a lock per store path, so a store's index is rebuilt once + however many callers open it at the same time; a second **process** on one store is unsupported + and would need D9's rejected sentinel. + +## How this record was reviewed, and what was lost + +The spec this record implements was attacked twice by an adversarial pass before any code was +written. **The second pass's report was destroyed**, and this note exists so the loss is visible +where the obligations live rather than only in a conversation: a later job was written against the +same file name in the same folder and overwrote it, and the first pass survives only because it +had been renamed. + +**Pass one** (18 criteria red at base, 4 green, 0 undecidable) found nine holes: a criterion whose +literal wording broke a correct build, `C7` checking answers instead of the index, `C16` never +asserting population *before* the first search, `C17` unable to distinguish a marker written first, +`C4` pinning two of seven filter combinations, unpinned test commands that pass when they match no +test, two database names, long-token truncation, and an unresolved `Matcher` seam. + +**Pass two** (20 red, 2 green, 0 undecidable) found the class that matters most here — implementations +that are *nearly* right and fail nothing: a rebuild that adds keys without clearing (this record's +D6 needs the clear-before-write, and the spec's stale-key criterion is what forced it); a router +placed in `page` only, which a limited count bypasses; a language refusal implemented on the find +path and not the count path; plus the rank heap's **polarity** (a `Reverse` min-heap evicts the +best row when it caps, so D4's order needed the opposite type), `C5`'s arithmetic (three span-2 +rows, not two), and `heed`'s real `open_database` arity, which the criteria had written wrong. + +Every finding from both passes is folded into the spec the build was written against; what is lost +is the report, not the requirements. The lesson is recorded here rather than in a process note +because it is a record-keeping obligation of this sequence: **an adversarial report is an artifact +of the decision, so it is written to a path that names its subject and its pass**, not to a shared +`spec-redteam.md` that the next design will overwrite. + +## What would overturn this record + +Any one of: a verified Postgres-compatible ranker (D4), an index that must carry tag keys or decide +alone (D1, D3), a tokeniser that has to stem or segment another script (D2), an incremental rebuild +(D5, D6), an operator corpus where the open-time rebuild is unacceptable (D6), or a second process +opening one store (D9). Each is a store format, open-cost or concurrency-contract change, and each +belongs to a new record in this sequence. diff --git a/docs/rewrite/nip-parity.md b/docs/rewrite/nip-parity.md index 3418d469..570a6ea6 100644 --- a/docs/rewrite/nip-parity.md +++ b/docs/rewrite/nip-parity.md @@ -1001,7 +1001,7 @@ no relay-side test reaches them. | 43 | `core/nip43.rs`; `relay/strategies/{join,leave,invite}_request.rs` | `session invite-mint` | | 44 | `core/nip44.rs` | `divergent: nip44_parity::produces_the_canonical_payload_from_the_nip44_spec_test_vector` *(nostream-core)* | | 45 | `relay/handlers/count.rs`; `core/messages.rs` | `session count-disabled`, `measured-session` | -| 50 | `core/filter.rs`; `storage/postgres/sql.rs` | `vectors-only: nip_vectors::nip50_a_search_filter_is_served_only_when_enabled` | +| 50 | `core/filter.rs`; `storage/postgres/sql.rs`; `storage/lmdb/search.rs` | `vectors-only: nip_vectors::nip50_a_search_filter_is_served_only_when_enabled`; the embedded store's own order by `parity_lmdb::a_search_returns_the_recorded_rows_in_rank_order` | | 56 | `core/nip56.rs`; `relay/strategies/report.rs`; `relay/hidden_content.rs` | `session hidden-report` | | 59 | `relay/strategies/gift_wrap.rs`; `core/event.rs` | `session event-rejections`, step 3 | | 62 | `relay/strategies/vanish.rs` | `vectors-only: strategies::vanish_soft_deletes_the_other_events_and_stores_the_request` | @@ -1015,9 +1015,13 @@ no relay-side test reaches them. | 98 | `core/nip98.rs` (stub) | `no-wire-surface: header verification needs the admin HTTP surface (D1, #865)` | Two limits of the map, stated so a later reader does not take it for more than it is. -**NIP-50's ranking** is not in it: `ts_rank` lives in the SQL backend and needs Postgres, -so ranking stays with the `#[ignore]`d `parity` target while the vector above pins the -matcher's accept-and-drop behaviour. **The census is a different oracle**: it maps 658 +**NIP-50's ranking** is split twice over. The SQL backend's `ts_rank` needs Postgres, so its +*values* stay with the `#[ignore]`d `parity` target, while the vector above pins the matcher's +accept-and-drop behaviour. The embedded store does not use `ts_rank`: it ranks by the smallest +cover span, then `event_id` (`storage/lmdb/search.rs`, ADR 0012 D4), and the `parity_lmdb` +arm pins the order it produces on the captured corpus — through entry 21, which is replayed +now, and through `a_search_returns_the_recorded_rows_in_rank_order`. What neither arm +establishes is that the two rank *numbers* agree; the fixture records an order, not values. **The census is a different oracle**: it maps 658 TypeScript unit cases to Rust tests and is untouched here, because a NIP with a vector and no TypeScript case has nothing to map, and hand-adding one fails `ts_case_parity` by design.