1use alloc::format;
17use alloc::string::String;
18use alloc::vec::Vec;
19
20use plugmem_arena::{
21 Arena, ArenaCfg, BlobHeap, BlobHeapCfg, BlobId, ChunkPool, ChunkPoolCfg, Interner, ListHandle,
22 ShardMode, Slot, TermId, key,
23};
24
25use crate::config::Config;
26use crate::error::Error;
27use crate::id::{EdgeId, EntityId, FactId, NONE_U32};
28use crate::index::IdListIndex;
29use crate::index::bm25::Bm25Index;
30use crate::index::hnsw::HnswGraph;
31use crate::index::vecpool::VecPool;
32use crate::journal::{JournalScan, Op, scan};
33use crate::model::{
34 EdgeHistorySlot, EdgeSlot, EntityByName, EntityRecord, FactAux, FactRecord, TemporalSlot,
35 VALID_TO_OPEN, close_edge_history_payload, edge_history_key, edge_key, fact_flags,
36};
37use crate::storage::Storage;
38use crate::tokenizer::Tokenizer;
39
40use maintain::TOKENIZER_INDEX_VERSION;
41
42const SIMILAR_CANDIDATE_CAP: usize = 32;
44
45mod maintain;
46mod migrations;
47mod persist;
48mod recall;
49mod shards;
50
51pub use maintain::{MaintainReport, MaintenanceMode, MaintenanceOptions};
52pub use recall::{RecallQuery, RecallResult, RecallScratch, RecalledEdge, RecalledFact, source};
53pub use shards::ShardLayout;
54
55#[derive(Clone, Copy, Debug)]
57#[cfg_attr(feature = "serde", derive(serde::Serialize))]
58pub struct RememberInput<'a> {
59 pub now: u64,
61 pub text: &'a str,
63 pub entity: Option<&'a str>,
65 pub tags: &'a [&'a str],
67 pub links: &'a [(&'a str, &'a str)],
69 pub vector: Option<&'a [f32]>,
73 pub valid_from: Option<u64>,
75 pub metadata: Option<&'a [(&'a str, &'a str)]>,
80}
81
82impl<'a> RememberInput<'a> {
83 pub fn text(now: u64, text: &'a str) -> Self {
85 Self {
86 now,
87 text,
88 entity: None,
89 tags: &[],
90 links: &[],
91 vector: None,
92 valid_from: None,
93 metadata: None,
94 }
95 }
96}
97
98#[derive(Clone, Copy, Debug)]
100#[cfg_attr(feature = "serde", derive(serde::Serialize))]
101pub struct LinkInput<'a> {
102 pub now: u64,
104 pub src: &'a str,
106 pub rel: &'a str,
108 pub dst: &'a str,
110 pub provenance: Option<FactId>,
112}
113
114#[derive(Clone, Copy, Debug)]
116#[cfg_attr(feature = "serde", derive(serde::Serialize))]
117pub struct UnlinkInput<'a> {
118 pub now: u64,
120 pub src: &'a str,
122 pub rel: &'a str,
124 pub dst: &'a str,
126}
127
128#[derive(Clone, Debug, PartialEq)]
130#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
131pub struct RememberOutcome {
132 pub id: FactId,
134 pub entity: Option<EntityId>,
136 pub similar: Vec<Similar>,
141}
142
143#[derive(Clone, Copy, Debug, PartialEq)]
145#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
146pub struct Similar {
147 pub id: FactId,
149 pub score: f32,
152 pub reason: SimilarReason,
154}
155
156#[derive(Clone, Copy, Debug, PartialEq, Eq)]
158#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
159pub enum SimilarReason {
160 LexicalOverlap,
163 VectorCosine,
166}
167
168#[derive(Clone, Copy, Debug)]
170#[cfg_attr(feature = "serde", derive(serde::Serialize))]
171pub struct FactView<'a> {
172 pub record: FactRecord,
174 pub text: &'a str,
176}
177
178#[derive(Clone, Copy, Debug, PartialEq, Eq)]
183#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
184pub enum FactFault {
185 Text,
187 Vector,
190 Metadata,
193}
194
195#[derive(Clone, Copy, Debug, PartialEq, Eq)]
199#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
200#[non_exhaustive]
201pub struct Stats {
202 pub facts: usize,
205 pub entities: usize,
207 pub terms: usize,
209 pub edges: usize,
212 pub edge_versions: usize,
214 pub vectors: usize,
216 pub tombstones: usize,
218 pub hnsw_indexed: u32,
221 pub next_fact: u32,
224 pub next_entity: u32,
226 pub next_edge: u32,
228 pub db_uuid: u128,
231 pub pool_bytes: usize,
234 pub shards: ShardLayout,
240}
241
242#[derive(Clone, Debug, Default, PartialEq, Eq)]
244#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
245pub struct OpenReport {
246 pub replayed: usize,
248 pub skipped: usize,
250 pub truncated_tail: bool,
252}
253
254pub struct Memory<'a> {
261 cfg: Config,
262 facts: Arena<'a, FactRecord>,
264 fact_aux: Arena<'a, FactAux>,
265 entities: Arena<'a, EntityRecord>,
266 by_name: Arena<'a, EntityByName>,
267 edges_out: Arena<'a, EdgeSlot>,
268 edges_in: Arena<'a, EdgeSlot>,
269 edges_hist_out: Arena<'a, EdgeHistorySlot>,
270 edges_hist_in: Arena<'a, EdgeHistorySlot>,
271 temporal: Arena<'a, TemporalSlot>,
272 texts: BlobHeap<'a>,
274 metas: BlobHeap<'a>,
279 terms: Interner<'a>,
282 tag_lists: ChunkPool<'a>,
284 bm25: Bm25Index<'a>,
286 tags_idx: IdListIndex<'a>,
287 entity_facts: IdListIndex<'a>,
288 vecs: VecPool<'a>,
290 hnsw: HnswGraph<'a>,
294 next_fact: u32,
296 next_entity: u32,
297 next_edge: u32,
298 tombstones: usize,
300 bm25_tokenizer_version: u32,
301 tokenizer: Tokenizer,
303 tf_scratch: Vec<(u32, u8)>,
304 name_scratch: String,
305}
306
307impl<'a> Memory<'a> {
308 pub fn new(cfg: Config) -> Result<Self, Error> {
315 cfg.validate()?;
316 let uni =
317 |shards: usize| ArenaCfg::new(shards, ShardMode::Uniform).with_max_bytes(cfg.max_bytes);
318 let ord =
319 |shards: usize| ArenaCfg::new(shards, ShardMode::Ordered).with_max_bytes(cfg.max_bytes);
320 let blob = BlobHeapCfg::new()
321 .with_max_bytes(cfg.max_bytes)
322 .with_max_blob(cfg.max_blob);
323 Ok(Self {
324 facts: Arena::new(uni(cfg.shards_facts))?,
325 fact_aux: Arena::new(uni(cfg.shards_facts))?,
326 entities: Arena::new(uni(cfg.shards_entities))?,
327 by_name: Arena::new(ord(cfg.shards_entities))?,
328 edges_out: Arena::new(ord(cfg.shards_edges))?,
329 edges_in: Arena::new(ord(cfg.shards_edges))?,
330 edges_hist_out: Arena::new(ord(cfg.shards_edges))?,
331 edges_hist_in: Arena::new(ord(cfg.shards_edges))?,
332 temporal: Arena::new(ord(cfg.shards_temporal))?,
333 texts: BlobHeap::new(blob),
334 metas: BlobHeap::new(blob),
335 terms: Interner::new(blob),
336 tag_lists: ChunkPool::new(ChunkPoolCfg::new().with_max_bytes(cfg.max_bytes)),
337 bm25: Bm25Index::new(cfg.shards_postings, cfg.max_bytes)?,
338 tags_idx: IdListIndex::new(cfg.shards_postings, cfg.max_bytes)?,
339 entity_facts: IdListIndex::new(cfg.shards_entities, cfg.max_bytes)?,
340 vecs: VecPool::new(cfg.dim, cfg.max_bytes),
341 hnsw: HnswGraph::new(cfg.hnsw_m, cfg.hnsw_m0, cfg.max_bytes)?,
342 next_fact: 0,
343 next_entity: 0,
344 next_edge: 0,
345 tombstones: 0,
346 bm25_tokenizer_version: maintain::TOKENIZER_INDEX_VERSION,
347 tokenizer: Tokenizer::new(),
348 tf_scratch: Vec::new(),
349 name_scratch: String::new(),
350 cfg,
351 })
352 }
353
354 pub fn open<S: Storage>(store: &mut S, cfg: Config) -> Result<(Self, OpenReport), Error> {
358 let snapshot = store
359 .read_snapshot()
360 .map_err(|e| Error::Storage(format!("{e:?}")))?;
361 let journal = store
362 .read_journal()
363 .map_err(|e| Error::Storage(format!("{e:?}")))?;
364 Self::from_bytes(snapshot.as_deref(), &journal, cfg)
365 }
366
367 pub fn from_bytes(
369 snapshot: Option<&[u8]>,
370 journal: &[u8],
371 cfg: Config,
372 ) -> Result<(Self, OpenReport), Error> {
373 let mut mem = match snapshot {
374 Some(bytes) => Self::load_snapshot(bytes, cfg)?,
375 None => Self::new(cfg)?,
376 };
377 let report = mem.replay(journal)?;
378 Ok((mem, report))
379 }
380
381 pub fn from_bytes_borrowed(
392 snapshot: &'a [u8],
393 journal: &[u8],
394 cfg: Config,
395 ) -> Result<Self, Error> {
396 let mem = Self::load_snapshot_borrowed(snapshot, cfg)?;
397 let JournalScan { entries, .. } = scan(journal)?;
398 if !entries.is_empty() {
399 return Err(Error::Invalid(
400 "read-only open requires a checkpointed (empty) journal",
401 ));
402 }
403 Ok(mem)
404 }
405
406 pub fn from_bytes_overlay(
421 snapshot: &'a [u8],
422 journal: &[u8],
423 cfg: Config,
424 ) -> Result<(Self, OpenReport), Error> {
425 let mut mem = Self::load_snapshot_borrowed(snapshot, cfg)?;
426 let report = mem.replay(journal)?;
427 Ok((mem, report))
428 }
429
430 fn replay(&mut self, journal: &[u8]) -> Result<OpenReport, Error> {
435 let JournalScan {
436 entries,
437 truncated_tail,
438 } = scan(journal)?;
439 let mut report = OpenReport {
440 truncated_tail,
441 ..OpenReport::default()
442 };
443 for entry in entries {
444 let op = Op::decode(entry.op, entry.payload)?;
445 match op {
446 Op::Remember {
447 now,
448 valid_from,
449 entity,
450 text,
451 ref tags,
452 ref links,
453 ref vector,
454 ref metadata,
455 revises,
456 assigned,
457 } => {
458 if assigned.0 < self.next_fact {
459 report.skipped += 1;
460 continue;
461 }
462 if assigned.0 != self.next_fact {
463 return Err(Error::Corrupt("journal fact ids are not contiguous"));
464 }
465 if !vector.is_empty() && vector.len() != self.cfg.dim {
466 return Err(Error::Corrupt(
467 "journal vector dimension disagrees with dim",
468 ));
469 }
470 if let Some(target) = revises.some() {
471 self.check_revisable(target)
472 .map_err(|_| Error::Corrupt("journal revises an unrevisable fact"))?;
473 }
474 self.apply_remember(
475 &RememberInput {
476 now,
477 text,
478 entity,
479 tags: &tags.to_vec(),
480 links: &links.to_vec(),
481 vector: (!vector.is_empty()).then_some(vector.as_slice()),
482 valid_from: Some(valid_from),
483 metadata: (!metadata.is_empty()).then_some(metadata.as_slice()),
484 },
485 revises,
486 )?;
487 if let Some(target) = revises.some() {
488 self.close_target(target, valid_from);
489 }
490 report.replayed += 1;
491 }
492 Op::Forget { fact, .. } => {
493 match self.apply_forget(fact) {
496 Ok(_) => report.replayed += 1,
497 Err(Error::NotFound(_)) => {
498 return Err(Error::Corrupt("journal forgets an unknown fact"));
499 }
500 Err(e) => return Err(e),
501 }
502 }
503 Op::Link {
504 now,
505 src,
506 rel,
507 dst,
508 provenance,
509 } => {
510 self.apply_link(now, src, rel, dst, provenance)?;
511 report.replayed += 1;
512 }
513 Op::Unlink { now, src, rel, dst } => {
514 self.apply_unlink(now, src, rel, dst)?;
515 report.replayed += 1;
516 }
517 Op::Maintain {
518 mode,
519 max_hnsw_inserts,
520 ..
521 } => {
522 let options =
526 maintain::MaintenanceOptions::from_journal(mode, max_hnsw_inserts)?;
527 self.replay_maintain_with_options(options)?;
528 report.replayed += 1;
529 }
530 }
531 }
532 Ok(report)
533 }
534
535 pub fn remember<S: Storage>(
537 &mut self,
538 store: &mut S,
539 input: RememberInput<'_>,
540 ) -> Result<RememberOutcome, Error> {
541 self.validate_input(&input)?;
542 let mut outcome = self.apply_remember(&input, FactId::NONE)?;
543 self.find_similar(&mut outcome);
544 self.journal_remember(store, &input, FactId::NONE, outcome.id)?;
545 Ok(outcome)
546 }
547
548 pub fn remember_batch<S: Storage>(
556 &mut self,
557 store: &mut S,
558 inputs: &[RememberInput<'_>],
559 skip_similar: bool,
560 ) -> Result<Vec<RememberOutcome>, Error> {
561 let mut out = Vec::with_capacity(inputs.len());
562 for input in inputs {
563 self.validate_input(input)?;
564 let mut outcome = self.apply_remember(input, FactId::NONE)?;
565 if !skip_similar {
566 self.find_similar(&mut outcome);
567 }
568 self.journal_remember(store, input, FactId::NONE, outcome.id)?;
569 out.push(outcome);
570 }
571 Ok(out)
572 }
573
574 fn find_similar(&mut self, outcome: &mut RememberOutcome) {
596 let Some(entity) = outcome.entity else { return };
597 let new_terms: Vec<u32> = self.tf_scratch.iter().map(|&(t, _)| t).collect();
600 let new_vec = self
602 .fact(outcome.id)
603 .filter(|r| r.has_vector())
604 .map(|r| r.vector);
605 if new_terms.is_empty() && new_vec.is_none() {
606 return;
607 }
608 let mut ring = [FactId::NONE; SIMILAR_CANDIDATE_CAP];
610 let mut n = 0usize;
611 for (fact, _) in self.entity_facts.entries(entity.0) {
612 if fact != outcome.id {
613 ring[n % SIMILAR_CANDIDATE_CAP] = fact;
614 n += 1;
615 }
616 }
617 let summaries_trustworthy = self.bm25_tokenizer_version == TOKENIZER_INDEX_VERSION;
618 let lexical_only = new_vec.is_none();
625 let mut cand_terms: Vec<u32> = Vec::new();
626 for &fact in ring.iter().take(n.min(SIMILAR_CANDIDATE_CAP)) {
627 let may_overlap = !new_terms.is_empty()
628 && self.overlap_possible(fact, &new_terms, summaries_trustworthy);
629 if lexical_only && !may_overlap {
630 continue;
631 }
632 let Some(record) = self.fact(fact) else {
633 continue;
634 };
635 if record.is_tombstone() || record.is_closed() {
636 continue;
637 }
638
639 let mut lexical = None;
641 if may_overlap && let Ok(text) = core::str::from_utf8(self.texts.get(record.text)) {
644 cand_terms.clear();
645 let terms = &self.terms;
646 let cand = &mut cand_terms;
647 self.tokenizer.tokenize(text, &mut |token| {
648 if let Some(term) = terms.lookup(token)
649 && !cand.contains(&term.0)
650 {
651 cand.push(term.0);
652 }
653 });
654 if !cand_terms.is_empty() {
655 let both = cand_terms.iter().filter(|t| new_terms.contains(t)).count();
656 let union = cand_terms.len() + new_terms.len() - both;
657 let jaccard = both as f32 / union as f32;
658 if jaccard > self.cfg.similar_jaccard {
659 lexical = Some(jaccard);
660 }
661 }
662 }
663
664 let mut vector = None;
666 if let (Some(a), true) = (new_vec, record.has_vector()) {
667 let cos = self.vecs.cosine_slots(a, record.vector);
668 if cos > self.cfg.similar_cos {
669 vector = Some(cos);
670 }
671 }
672
673 let best = match (lexical, vector) {
675 (Some(l), Some(v)) if v > l => Some((v, SimilarReason::VectorCosine)),
676 (Some(l), _) => Some((l, SimilarReason::LexicalOverlap)),
677 (None, Some(v)) => Some((v, SimilarReason::VectorCosine)),
678 (None, None) => None,
679 };
680 if let Some((score, reason)) = best {
681 outcome.similar.push(Similar {
682 id: fact,
683 score,
684 reason,
685 });
686 }
687 }
688 outcome
689 .similar
690 .sort_unstable_by(|a, b| b.score.total_cmp(&a.score).then(a.id.cmp(&b.id)));
691 outcome.similar.truncate(8);
692 }
693
694 fn overlap_possible(&self, candidate: FactId, new_terms: &[u32], trust_summary: bool) -> bool {
705 if !trust_summary {
706 return true;
707 }
708 let Some(doc) = self.bm25.doc(candidate) else {
709 return true;
710 };
711 if !doc.has_signature() {
712 return true;
713 }
714 let bound = doc.overlap_bound(new_terms);
715 debug_assert!(
721 bound <= new_terms.len(),
722 "the overlap bound counts query terms, so it cannot exceed them"
723 );
724 let union = (new_terms.len() - bound) + usize::from(doc.distinct);
725 bound as f32 / union as f32 > self.cfg.similar_jaccard
726 }
727
728 pub fn revise<S: Storage>(
732 &mut self,
733 store: &mut S,
734 target: FactId,
735 input: RememberInput<'_>,
736 ) -> Result<RememberOutcome, Error> {
737 self.validate_input(&input)?;
738 self.check_revisable(target)?;
742 let outcome = self.apply_remember(&input, target)?;
743 let valid_from = input.valid_from.unwrap_or(input.now);
744 self.close_target(target, valid_from);
745 self.journal_remember(store, &input, target, outcome.id)?;
746 Ok(outcome)
747 }
748
749 pub fn forget<S: Storage>(
753 &mut self,
754 store: &mut S,
755 now: u64,
756 id: FactId,
757 ) -> Result<bool, Error> {
758 let fresh = self.apply_forget(id)?;
759 let mut entry = Vec::new();
760 Op::Forget { now, fact: id }.encode(&mut entry);
761 store
762 .append_journal(&entry)
763 .map_err(|e| Error::Storage(format!("{e:?}")))?;
764 Ok(fresh)
765 }
766
767 pub fn link<S: Storage>(&mut self, store: &mut S, input: LinkInput<'_>) -> Result<(), Error> {
770 self.apply_link(
771 input.now,
772 input.src,
773 input.rel,
774 input.dst,
775 FactId::from_opt(input.provenance),
776 )?;
777 let mut entry = Vec::new();
778 Op::Link {
779 now: input.now,
780 src: input.src,
781 rel: input.rel,
782 dst: input.dst,
783 provenance: FactId::from_opt(input.provenance),
784 }
785 .encode(&mut entry);
786 store
787 .append_journal(&entry)
788 .map_err(|e| Error::Storage(format!("{e:?}")))?;
789 Ok(())
790 }
791
792 pub fn unlink<S: Storage>(
795 &mut self,
796 store: &mut S,
797 input: UnlinkInput<'_>,
798 ) -> Result<bool, Error> {
799 let fresh = self.apply_unlink(input.now, input.src, input.rel, input.dst)?;
800 let mut entry = Vec::new();
801 Op::Unlink {
802 now: input.now,
803 src: input.src,
804 rel: input.rel,
805 dst: input.dst,
806 }
807 .encode(&mut entry);
808 store
809 .append_journal(&entry)
810 .map_err(|e| Error::Storage(format!("{e:?}")))?;
811 Ok(fresh)
812 }
813
814 pub fn get(&self, id: FactId) -> Option<FactView<'_>> {
817 let record = self.fact(id)?;
818 if record.is_tombstone() {
819 return None;
820 }
821 let text = core::str::from_utf8(self.texts.get(record.text)).ok()?;
825 Some(FactView { record, text })
826 }
827
828 pub fn tags_of(&self, id: FactId, out: &mut Vec<TermId>) {
831 let Some(record) = self.fact(id) else { return };
832 if record.is_tombstone() {
833 return;
834 }
835 let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
836 return;
837 };
838 for chunk in self.tag_lists.iter(&aux.tags) {
839 for raw in chunk.chunks_exact(4) {
840 out.push(TermId(u32::from_be_bytes(raw.try_into().unwrap())));
841 }
842 }
843 }
844
845 pub fn metadata_of<'s>(&'s self, id: FactId, out: &mut Vec<(&'s str, &'s str)>) -> bool {
856 out.clear();
857 let Some(record) = self.fact(id) else {
858 return false;
859 };
860 if record.is_tombstone() {
861 return false;
862 }
863 let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
864 return false;
865 };
866 if aux.meta.0 == NONE_U32 || aux.meta.0 >= self.metas.len() as u32 {
867 return false;
868 }
869 crate::metadata::decode(self.metas.get(aux.meta), out).is_ok() && !out.is_empty()
870 }
871
872 pub fn entity(&mut self, name: &str) -> Option<EntityId> {
874 let mut norm = core::mem::take(&mut self.name_scratch);
875 normalize_name(&mut self.tokenizer, name, &mut norm);
876 let found = if norm.is_empty() {
877 None
878 } else {
879 self.lookup_entity_by_norm(&norm)
881 };
882 self.name_scratch = norm;
883 found
884 }
885
886 pub fn term(&self, id: TermId) -> &str {
888 self.terms.resolve(id)
889 }
890
891 pub fn entity_name(&self, id: EntityId) -> Option<&str> {
896 let record = self.entities.get(&id.0.to_be_bytes())?;
897 core::str::from_utf8(self.texts.get(record.name)).ok()
900 }
901
902 pub fn edges_each(&self, mut visit: impl FnMut(&str, &str, &str, FactId) -> bool) {
920 for slot in self.edges_out.iter() {
921 let (Some(src), Some(dst)) = (self.entity_name(slot.a), self.entity_name(slot.b))
922 else {
923 continue;
924 };
925 if !visit(src, self.term(slot.rel), dst, slot.fact) {
926 return;
927 }
928 }
929 }
930
931 pub fn facts_len(&self) -> usize {
935 self.facts.len()
936 }
937
938 pub fn entities_len(&self) -> usize {
940 self.entities.len()
941 }
942
943 pub fn cfg(&self) -> &Config {
945 &self.cfg
946 }
947
948 pub fn stats(&self) -> Stats {
950 Stats {
951 facts: self.facts.len(),
952 entities: self.entities.len(),
953 terms: self.terms.len(),
954 edges: self.edges_out.len(),
955 edge_versions: self.edges_hist_out.len(),
956 vectors: self.vecs.len(),
957 tombstones: self.tombstones,
958 hnsw_indexed: self.hnsw.indexed(),
959 next_fact: self.next_fact,
960 next_entity: self.next_entity,
961 next_edge: self.next_edge,
962 db_uuid: self.cfg.db_uuid,
963 pool_bytes: self.facts.pool_bytes()
964 + self.fact_aux.pool_bytes()
965 + self.entities.pool_bytes()
966 + self.by_name.pool_bytes()
967 + self.edges_out.pool_bytes()
968 + self.edges_in.pool_bytes()
969 + self.edges_hist_out.pool_bytes()
970 + self.edges_hist_in.pool_bytes()
971 + self.temporal.pool_bytes()
972 + self.texts.pool_bytes()
973 + self.terms.pool_bytes()
974 + self.tag_lists.pool_bytes()
975 + self.bm25.pool_bytes()
976 + self.tags_idx.pool_bytes()
977 + self.entity_facts.pool_bytes()
978 + self.vecs.pool_bytes()
979 + self.hnsw.pool_bytes(),
980 shards: shards::ShardLayout::of_config(&self.cfg),
981 }
982 }
983
984 fn fact(&self, id: FactId) -> Option<FactRecord> {
987 self.facts.get(&id.0.to_be_bytes())
988 }
989
990 fn validate_input(&self, input: &RememberInput<'_>) -> Result<(), Error> {
992 if input.text.len() > self.cfg.max_text {
993 return Err(Error::TooLarge {
994 what: "text",
995 len: input.text.len(),
996 max: self.cfg.max_text,
997 });
998 }
999 if input.tags.len() > 32 {
1000 return Err(Error::TooLarge {
1001 what: "tags",
1002 len: input.tags.len(),
1003 max: 32,
1004 });
1005 }
1006 if input.links.len() > 16 {
1007 return Err(Error::TooLarge {
1008 what: "links",
1009 len: input.links.len(),
1010 max: 16,
1011 });
1012 }
1013 if input.tags.iter().any(|t| t.is_empty()) {
1014 return Err(Error::Invalid("empty tag"));
1015 }
1016 if !input.links.is_empty() && input.entity.is_none() {
1017 return Err(Error::Invalid("links require a subject entity"));
1018 }
1019 if let Some(v) = input.vector {
1020 if self.cfg.dim == 0 {
1021 return Err(Error::Invalid("vector given but dim is 0"));
1022 }
1023 if v.len() != self.cfg.dim {
1024 return Err(Error::DimMismatch {
1025 got: v.len(),
1026 want: self.cfg.dim,
1027 });
1028 }
1029 }
1030 Ok(())
1031 }
1032
1033 fn check_revisable(&self, target: FactId) -> Result<(), Error> {
1035 let record = self.fact(target).ok_or(Error::NotFound(target))?;
1036 if record.is_tombstone() {
1037 return Err(Error::NotFound(target));
1038 }
1039 if record.is_closed() {
1040 return Err(Error::AlreadyClosed(target));
1041 }
1042 Ok(())
1043 }
1044
1045 fn close_target(&mut self, target: FactId, valid_to: u64) {
1048 let record = self.fact(target).expect("checked revisable");
1049 let payload = self
1050 .facts
1051 .payload_mut(&target.0.to_be_bytes())
1052 .expect("record fetched above");
1053 let flags = record.flags | fact_flags::CLOSED;
1056 payload[4..6].copy_from_slice(&flags.to_be_bytes());
1057 payload[36..44].copy_from_slice(&valid_to.to_be_bytes());
1058 }
1059
1060 fn apply_remember(
1063 &mut self,
1064 input: &RememberInput<'_>,
1065 revises: FactId,
1066 ) -> Result<RememberOutcome, Error> {
1067 let id = FactId(self.next_fact);
1068 let entity = match input.entity {
1069 Some(name) => Some(self.resolve_or_create_entity(name, input.now)?),
1070 None => None,
1071 };
1072 let text_id = self.texts.push(input.text.as_bytes())?;
1073
1074 let mut tfs = core::mem::take(&mut self.tf_scratch);
1076 tfs.clear();
1077 let terms = &mut self.terms;
1078 let mut intern_err = None;
1079 self.tokenizer.tokenize(input.text, &mut |token| {
1080 if intern_err.is_some() {
1081 return;
1082 }
1083 match terms.intern(token) {
1084 Ok(term) => match tfs.iter_mut().find(|(t, _)| *t == term.0) {
1085 Some((_, tf)) => *tf = tf.saturating_add(1),
1086 None => tfs.push((term.0, 1)),
1087 },
1088 Err(e) => intern_err = Some(e),
1089 }
1090 });
1091 if let Some(e) = intern_err {
1092 self.tf_scratch = tfs;
1093 return Err(Error::Arena(e));
1094 }
1095 self.bm25.index_doc(id, &tfs)?;
1096 self.tf_scratch = tfs;
1097
1098 let meta = match input.metadata {
1102 Some(pairs) if !pairs.is_empty() => {
1103 self.metas.push(&crate::metadata::encode(pairs)?)?
1104 }
1105 _ => BlobId(NONE_U32),
1106 };
1107
1108 let mut aux = FactAux {
1111 id,
1112 tags: ListHandle::EMPTY,
1113 meta,
1114 };
1115 let mut seen_tags: [u32; 32] = [NONE_U32; 32];
1116 let mut seen_cnt = 0usize;
1117 for tag in input.tags {
1118 let term = self.terms.intern(tag)?;
1119 if seen_tags[..seen_cnt].contains(&term.0) {
1120 continue;
1121 }
1122 seen_tags[seen_cnt] = term.0;
1123 seen_cnt += 1;
1124 self.tag_lists.push(&mut aux.tags, &term.0.to_be_bytes())?;
1125 self.tags_idx.push(term.0, id, 0)?;
1126 }
1127 self.fact_aux.insert(&aux)?;
1128
1129 if let Some(src) = entity {
1131 for &(rel, dst_name) in input.links {
1132 let dst = self.resolve_or_create_entity(dst_name, input.now)?;
1133 let rel = self.terms.intern(rel)?;
1134 self.open_edge(input.now, src, rel, dst, id)?;
1135 }
1136 self.entity_facts.push(src.0, id, 0)?;
1137 }
1138
1139 let (vector, flags) = match input.vector {
1143 Some(v) => (self.vecs.push(id, v)?, fact_flags::HAS_VECTOR),
1144 None => (NONE_U32, 0),
1145 };
1146
1147 let recorded_at = input.now;
1148 let valid_from = input.valid_from.unwrap_or(input.now);
1149 self.facts.insert(&FactRecord {
1150 id,
1151 entity: EntityId::from_opt(entity),
1152 flags,
1153 kind: 0,
1154 text: text_id,
1155 vector,
1156 revises,
1157 recorded_at,
1158 valid_from,
1159 valid_to: VALID_TO_OPEN,
1160 })?;
1161 self.temporal.insert(&TemporalSlot {
1162 recorded_at,
1163 fact: id,
1164 })?;
1165 self.next_fact += 1;
1166 Ok(RememberOutcome {
1167 id,
1168 entity,
1169 similar: Vec::new(),
1170 })
1171 }
1172
1173 fn apply_forget(&mut self, id: FactId) -> Result<bool, Error> {
1174 let record = self.fact(id).ok_or(Error::NotFound(id))?;
1175 if record.is_tombstone() {
1176 return Ok(false);
1177 }
1178 let payload = self
1179 .facts
1180 .payload_mut(&id.0.to_be_bytes())
1181 .expect("record fetched above");
1182 let flags = record.flags | fact_flags::TOMBSTONE;
1183 payload[4..6].copy_from_slice(&flags.to_be_bytes());
1184 self.tombstones += 1;
1185 Ok(true)
1186 }
1187
1188 fn apply_link(
1189 &mut self,
1190 now: u64,
1191 src: &str,
1192 rel: &str,
1193 dst: &str,
1194 provenance: FactId,
1195 ) -> Result<(), Error> {
1196 let src = self.resolve_or_create_entity(src, now)?;
1197 let dst = self.resolve_or_create_entity(dst, now)?;
1198 let rel = self.terms.intern(rel)?;
1199 self.open_edge(now, src, rel, dst, provenance)
1200 }
1201
1202 fn apply_unlink(&mut self, now: u64, src: &str, rel: &str, dst: &str) -> Result<bool, Error> {
1203 let Some(src) = self.lookup_entity_name(src) else {
1204 return Ok(false);
1205 };
1206 let Some(dst) = self.lookup_entity_name(dst) else {
1207 return Ok(false);
1208 };
1209 let Some(rel) = self.terms.lookup(rel) else {
1210 return Ok(false);
1211 };
1212 self.close_current_edge(now, src, rel, dst)
1213 }
1214
1215 fn open_edge(
1216 &mut self,
1217 now: u64,
1218 src: EntityId,
1219 rel: TermId,
1220 dst: EntityId,
1221 fact: FactId,
1222 ) -> Result<(), Error> {
1223 if let Some(current) = self.current_edge(src, rel, dst) {
1224 if current.fact == fact {
1225 return Ok(());
1226 }
1227 self.close_current_edge(now, src, rel, dst)?;
1228 }
1229 let edge = EdgeId(self.next_edge);
1230 let history = EdgeHistorySlot {
1231 a: src,
1232 rel,
1233 b: dst,
1234 edge,
1235 fact,
1236 flags: 0,
1237 kind: 0,
1238 recorded_at: now,
1239 valid_from: now,
1240 valid_to: VALID_TO_OPEN,
1241 };
1242 self.insert_history_edge(history)?;
1243 self.insert_current_edge(src, rel, dst, fact, edge, now)?;
1244 self.next_edge += 1;
1245 Ok(())
1246 }
1247
1248 fn insert_current_edge(
1249 &mut self,
1250 src: EntityId,
1251 rel: TermId,
1252 dst: EntityId,
1253 fact: FactId,
1254 edge: EdgeId,
1255 valid_from: u64,
1256 ) -> Result<(), Error> {
1257 for (arena, a, b) in [
1258 (&mut self.edges_out, src, dst),
1259 (&mut self.edges_in, dst, src),
1260 ] {
1261 let slot = EdgeSlot {
1262 a,
1263 rel,
1264 b,
1265 fact,
1266 edge,
1267 valid_from,
1268 };
1269 if !arena.insert(&slot)? {
1270 let payload = arena
1271 .payload_mut(&edge_key(a, rel, b))
1272 .expect("insert reported a duplicate");
1273 let mut full = [0u8; EdgeSlot::SIZE];
1274 slot.write(&mut full);
1275 payload.copy_from_slice(&full[EdgeSlot::KEY_LEN..]);
1276 }
1277 }
1278 Ok(())
1279 }
1280
1281 fn insert_history_edge(&mut self, edge: EdgeHistorySlot) -> Result<(), Error> {
1282 self.edges_hist_out.insert(&edge)?;
1283 self.edges_hist_in.insert(&EdgeHistorySlot {
1284 a: edge.b,
1285 b: edge.a,
1286 ..edge
1287 })?;
1288 Ok(())
1289 }
1290
1291 fn close_current_edge(
1297 &mut self,
1298 now: u64,
1299 src: EntityId,
1300 rel: TermId,
1301 dst: EntityId,
1302 ) -> Result<bool, Error> {
1303 let Some(current) = self.current_edge(src, rel, dst) else {
1304 return Ok(false);
1305 };
1306 let close_at = now.max(current.valid_from);
1309 let out_key = edge_history_key(src, current.valid_from, current.edge);
1310 let in_key = edge_history_key(dst, current.valid_from, current.edge);
1311 close_edge_history_payload(
1312 self.edges_hist_out
1313 .payload_mut(&out_key)
1314 .ok_or(Error::Corrupt("missing outgoing edge history"))?,
1315 close_at,
1316 );
1317 close_edge_history_payload(
1318 self.edges_hist_in
1319 .payload_mut(&in_key)
1320 .ok_or(Error::Corrupt("missing incoming edge history"))?,
1321 close_at,
1322 );
1323 let out_removed = self.edges_out.remove(&edge_key(src, rel, dst));
1324 let in_removed = self.edges_in.remove(&edge_key(dst, rel, src));
1325 if out_removed != in_removed {
1326 return Err(Error::Corrupt("edge mirrors disagree"));
1327 }
1328 Ok(out_removed)
1329 }
1330
1331 fn current_edge(&self, src: EntityId, rel: TermId, dst: EntityId) -> Option<EdgeSlot> {
1332 self.edges_out
1333 .get_slot(&edge_key(src, rel, dst))
1334 .map(EdgeSlot::read)
1335 }
1336
1337 fn lookup_entity_by_norm(&self, norm: &str) -> Option<EntityId> {
1340 let term = self.terms.lookup(norm)?;
1341 let mut from = [0u8; 8];
1342 key::write_u32(&mut from, term.0);
1343 let mut to = [0u8; 8];
1344 key::write_u32(&mut to, term.0);
1345 to[4..].copy_from_slice(&u32::MAX.to_be_bytes());
1346 self.by_name.range(&from, &to).next().map(|e| e.id)
1347 }
1348
1349 fn lookup_entity_name(&mut self, name: &str) -> Option<EntityId> {
1350 let mut norm = core::mem::take(&mut self.name_scratch);
1351 normalize_name(&mut self.tokenizer, name, &mut norm);
1352 let result = (!norm.is_empty())
1353 .then(|| self.lookup_entity_by_norm(&norm))
1354 .flatten();
1355 self.name_scratch = norm;
1356 result
1357 }
1358
1359 fn resolve_or_create_entity(&mut self, name: &str, now: u64) -> Result<EntityId, Error> {
1360 let mut norm = core::mem::take(&mut self.name_scratch);
1361 normalize_name(&mut self.tokenizer, name, &mut norm);
1362 if norm.is_empty() {
1363 self.name_scratch = norm;
1364 return Err(Error::Invalid("entity name has no indexable characters"));
1365 }
1366 let result = (|| {
1367 if let Some(found) = self.lookup_entity_by_norm(&norm) {
1368 return Ok(found);
1369 }
1370 let term = self.terms.intern(&norm)?;
1371 let id = EntityId(self.next_entity);
1372 let name_id = self.texts.push(name.as_bytes())?;
1373 self.entities.insert(&EntityRecord {
1374 id,
1375 name: name_id,
1376 name_term: term,
1377 created_at: now,
1378 flags: 0,
1379 })?;
1380 self.by_name.insert(&EntityByName {
1381 name_term: term,
1382 id,
1383 })?;
1384 self.next_entity += 1;
1385 Ok(id)
1386 })();
1387 self.name_scratch = norm;
1388 result
1389 }
1390
1391 fn journal_remember<S: Storage>(
1392 &mut self,
1393 store: &mut S,
1394 input: &RememberInput<'_>,
1395 revises: FactId,
1396 assigned: FactId,
1397 ) -> Result<(), Error> {
1398 let mut entry = Vec::new();
1399 Op::Remember {
1400 now: input.now,
1401 valid_from: input.valid_from.unwrap_or(input.now),
1402 entity: input.entity,
1403 text: input.text,
1404 tags: input.tags.to_vec(),
1405 links: input.links.to_vec(),
1406 vector: input.vector.map(<[f32]>::to_vec).unwrap_or_default(),
1407 metadata: input.metadata.map(<[_]>::to_vec).unwrap_or_default(),
1408 revises,
1409 assigned,
1410 }
1411 .encode(&mut entry);
1412 store
1413 .append_journal(&entry)
1414 .map_err(|e| Error::Storage(format!("{e:?}")))
1415 }
1416}
1417
1418impl core::fmt::Debug for Memory<'_> {
1419 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1422 f.debug_struct("Memory")
1423 .field("facts", &self.facts.len())
1424 .field("entities", &self.entities.len())
1425 .field("terms", &self.terms.len())
1426 .finish()
1427 }
1428}
1429
1430fn normalize_name(tokenizer: &mut Tokenizer, name: &str, out: &mut String) {
1434 out.clear();
1435 tokenizer.tokenize(name, &mut |token| {
1436 if !out.is_empty() {
1437 out.push(' ');
1438 }
1439 out.push_str(token);
1440 });
1441}