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 facts_len(&self) -> usize {
906 self.facts.len()
907 }
908
909 pub fn entities_len(&self) -> usize {
911 self.entities.len()
912 }
913
914 pub fn cfg(&self) -> &Config {
916 &self.cfg
917 }
918
919 pub fn stats(&self) -> Stats {
921 Stats {
922 facts: self.facts.len(),
923 entities: self.entities.len(),
924 terms: self.terms.len(),
925 edges: self.edges_out.len(),
926 edge_versions: self.edges_hist_out.len(),
927 vectors: self.vecs.len(),
928 tombstones: self.tombstones,
929 hnsw_indexed: self.hnsw.indexed(),
930 next_fact: self.next_fact,
931 next_entity: self.next_entity,
932 next_edge: self.next_edge,
933 db_uuid: self.cfg.db_uuid,
934 pool_bytes: self.facts.pool_bytes()
935 + self.fact_aux.pool_bytes()
936 + self.entities.pool_bytes()
937 + self.by_name.pool_bytes()
938 + self.edges_out.pool_bytes()
939 + self.edges_in.pool_bytes()
940 + self.edges_hist_out.pool_bytes()
941 + self.edges_hist_in.pool_bytes()
942 + self.temporal.pool_bytes()
943 + self.texts.pool_bytes()
944 + self.terms.pool_bytes()
945 + self.tag_lists.pool_bytes()
946 + self.bm25.pool_bytes()
947 + self.tags_idx.pool_bytes()
948 + self.entity_facts.pool_bytes()
949 + self.vecs.pool_bytes()
950 + self.hnsw.pool_bytes(),
951 shards: shards::ShardLayout::of_config(&self.cfg),
952 }
953 }
954
955 fn fact(&self, id: FactId) -> Option<FactRecord> {
958 self.facts.get(&id.0.to_be_bytes())
959 }
960
961 fn validate_input(&self, input: &RememberInput<'_>) -> Result<(), Error> {
963 if input.text.len() > self.cfg.max_text {
964 return Err(Error::TooLarge {
965 what: "text",
966 len: input.text.len(),
967 max: self.cfg.max_text,
968 });
969 }
970 if input.tags.len() > 32 {
971 return Err(Error::TooLarge {
972 what: "tags",
973 len: input.tags.len(),
974 max: 32,
975 });
976 }
977 if input.links.len() > 16 {
978 return Err(Error::TooLarge {
979 what: "links",
980 len: input.links.len(),
981 max: 16,
982 });
983 }
984 if input.tags.iter().any(|t| t.is_empty()) {
985 return Err(Error::Invalid("empty tag"));
986 }
987 if !input.links.is_empty() && input.entity.is_none() {
988 return Err(Error::Invalid("links require a subject entity"));
989 }
990 if let Some(v) = input.vector {
991 if self.cfg.dim == 0 {
992 return Err(Error::Invalid("vector given but dim is 0"));
993 }
994 if v.len() != self.cfg.dim {
995 return Err(Error::DimMismatch {
996 got: v.len(),
997 want: self.cfg.dim,
998 });
999 }
1000 }
1001 Ok(())
1002 }
1003
1004 fn check_revisable(&self, target: FactId) -> Result<(), Error> {
1006 let record = self.fact(target).ok_or(Error::NotFound(target))?;
1007 if record.is_tombstone() {
1008 return Err(Error::NotFound(target));
1009 }
1010 if record.is_closed() {
1011 return Err(Error::AlreadyClosed(target));
1012 }
1013 Ok(())
1014 }
1015
1016 fn close_target(&mut self, target: FactId, valid_to: u64) {
1019 let record = self.fact(target).expect("checked revisable");
1020 let payload = self
1021 .facts
1022 .payload_mut(&target.0.to_be_bytes())
1023 .expect("record fetched above");
1024 let flags = record.flags | fact_flags::CLOSED;
1027 payload[4..6].copy_from_slice(&flags.to_be_bytes());
1028 payload[36..44].copy_from_slice(&valid_to.to_be_bytes());
1029 }
1030
1031 fn apply_remember(
1034 &mut self,
1035 input: &RememberInput<'_>,
1036 revises: FactId,
1037 ) -> Result<RememberOutcome, Error> {
1038 let id = FactId(self.next_fact);
1039 let entity = match input.entity {
1040 Some(name) => Some(self.resolve_or_create_entity(name, input.now)?),
1041 None => None,
1042 };
1043 let text_id = self.texts.push(input.text.as_bytes())?;
1044
1045 let mut tfs = core::mem::take(&mut self.tf_scratch);
1047 tfs.clear();
1048 let terms = &mut self.terms;
1049 let mut intern_err = None;
1050 self.tokenizer.tokenize(input.text, &mut |token| {
1051 if intern_err.is_some() {
1052 return;
1053 }
1054 match terms.intern(token) {
1055 Ok(term) => match tfs.iter_mut().find(|(t, _)| *t == term.0) {
1056 Some((_, tf)) => *tf = tf.saturating_add(1),
1057 None => tfs.push((term.0, 1)),
1058 },
1059 Err(e) => intern_err = Some(e),
1060 }
1061 });
1062 if let Some(e) = intern_err {
1063 self.tf_scratch = tfs;
1064 return Err(Error::Arena(e));
1065 }
1066 self.bm25.index_doc(id, &tfs)?;
1067 self.tf_scratch = tfs;
1068
1069 let meta = match input.metadata {
1073 Some(pairs) if !pairs.is_empty() => {
1074 self.metas.push(&crate::metadata::encode(pairs)?)?
1075 }
1076 _ => BlobId(NONE_U32),
1077 };
1078
1079 let mut aux = FactAux {
1082 id,
1083 tags: ListHandle::EMPTY,
1084 meta,
1085 };
1086 let mut seen_tags: [u32; 32] = [NONE_U32; 32];
1087 let mut seen_cnt = 0usize;
1088 for tag in input.tags {
1089 let term = self.terms.intern(tag)?;
1090 if seen_tags[..seen_cnt].contains(&term.0) {
1091 continue;
1092 }
1093 seen_tags[seen_cnt] = term.0;
1094 seen_cnt += 1;
1095 self.tag_lists.push(&mut aux.tags, &term.0.to_be_bytes())?;
1096 self.tags_idx.push(term.0, id, 0)?;
1097 }
1098 self.fact_aux.insert(&aux)?;
1099
1100 if let Some(src) = entity {
1102 for &(rel, dst_name) in input.links {
1103 let dst = self.resolve_or_create_entity(dst_name, input.now)?;
1104 let rel = self.terms.intern(rel)?;
1105 self.open_edge(input.now, src, rel, dst, id)?;
1106 }
1107 self.entity_facts.push(src.0, id, 0)?;
1108 }
1109
1110 let (vector, flags) = match input.vector {
1114 Some(v) => (self.vecs.push(id, v)?, fact_flags::HAS_VECTOR),
1115 None => (NONE_U32, 0),
1116 };
1117
1118 let recorded_at = input.now;
1119 let valid_from = input.valid_from.unwrap_or(input.now);
1120 self.facts.insert(&FactRecord {
1121 id,
1122 entity: EntityId::from_opt(entity),
1123 flags,
1124 kind: 0,
1125 text: text_id,
1126 vector,
1127 revises,
1128 recorded_at,
1129 valid_from,
1130 valid_to: VALID_TO_OPEN,
1131 })?;
1132 self.temporal.insert(&TemporalSlot {
1133 recorded_at,
1134 fact: id,
1135 })?;
1136 self.next_fact += 1;
1137 Ok(RememberOutcome {
1138 id,
1139 entity,
1140 similar: Vec::new(),
1141 })
1142 }
1143
1144 fn apply_forget(&mut self, id: FactId) -> Result<bool, Error> {
1145 let record = self.fact(id).ok_or(Error::NotFound(id))?;
1146 if record.is_tombstone() {
1147 return Ok(false);
1148 }
1149 let payload = self
1150 .facts
1151 .payload_mut(&id.0.to_be_bytes())
1152 .expect("record fetched above");
1153 let flags = record.flags | fact_flags::TOMBSTONE;
1154 payload[4..6].copy_from_slice(&flags.to_be_bytes());
1155 self.tombstones += 1;
1156 Ok(true)
1157 }
1158
1159 fn apply_link(
1160 &mut self,
1161 now: u64,
1162 src: &str,
1163 rel: &str,
1164 dst: &str,
1165 provenance: FactId,
1166 ) -> Result<(), Error> {
1167 let src = self.resolve_or_create_entity(src, now)?;
1168 let dst = self.resolve_or_create_entity(dst, now)?;
1169 let rel = self.terms.intern(rel)?;
1170 self.open_edge(now, src, rel, dst, provenance)
1171 }
1172
1173 fn apply_unlink(&mut self, now: u64, src: &str, rel: &str, dst: &str) -> Result<bool, Error> {
1174 let Some(src) = self.lookup_entity_name(src) else {
1175 return Ok(false);
1176 };
1177 let Some(dst) = self.lookup_entity_name(dst) else {
1178 return Ok(false);
1179 };
1180 let Some(rel) = self.terms.lookup(rel) else {
1181 return Ok(false);
1182 };
1183 self.close_current_edge(now, src, rel, dst)
1184 }
1185
1186 fn open_edge(
1187 &mut self,
1188 now: u64,
1189 src: EntityId,
1190 rel: TermId,
1191 dst: EntityId,
1192 fact: FactId,
1193 ) -> Result<(), Error> {
1194 if let Some(current) = self.current_edge(src, rel, dst) {
1195 if current.fact == fact {
1196 return Ok(());
1197 }
1198 self.close_current_edge(now, src, rel, dst)?;
1199 }
1200 let edge = EdgeId(self.next_edge);
1201 let history = EdgeHistorySlot {
1202 a: src,
1203 rel,
1204 b: dst,
1205 edge,
1206 fact,
1207 flags: 0,
1208 kind: 0,
1209 recorded_at: now,
1210 valid_from: now,
1211 valid_to: VALID_TO_OPEN,
1212 };
1213 self.insert_history_edge(history)?;
1214 self.insert_current_edge(src, rel, dst, fact, edge, now)?;
1215 self.next_edge += 1;
1216 Ok(())
1217 }
1218
1219 fn insert_current_edge(
1220 &mut self,
1221 src: EntityId,
1222 rel: TermId,
1223 dst: EntityId,
1224 fact: FactId,
1225 edge: EdgeId,
1226 valid_from: u64,
1227 ) -> Result<(), Error> {
1228 for (arena, a, b) in [
1229 (&mut self.edges_out, src, dst),
1230 (&mut self.edges_in, dst, src),
1231 ] {
1232 let slot = EdgeSlot {
1233 a,
1234 rel,
1235 b,
1236 fact,
1237 edge,
1238 valid_from,
1239 };
1240 if !arena.insert(&slot)? {
1241 let payload = arena
1242 .payload_mut(&edge_key(a, rel, b))
1243 .expect("insert reported a duplicate");
1244 let mut full = [0u8; EdgeSlot::SIZE];
1245 slot.write(&mut full);
1246 payload.copy_from_slice(&full[EdgeSlot::KEY_LEN..]);
1247 }
1248 }
1249 Ok(())
1250 }
1251
1252 fn insert_history_edge(&mut self, edge: EdgeHistorySlot) -> Result<(), Error> {
1253 self.edges_hist_out.insert(&edge)?;
1254 self.edges_hist_in.insert(&EdgeHistorySlot {
1255 a: edge.b,
1256 b: edge.a,
1257 ..edge
1258 })?;
1259 Ok(())
1260 }
1261
1262 fn close_current_edge(
1268 &mut self,
1269 now: u64,
1270 src: EntityId,
1271 rel: TermId,
1272 dst: EntityId,
1273 ) -> Result<bool, Error> {
1274 let Some(current) = self.current_edge(src, rel, dst) else {
1275 return Ok(false);
1276 };
1277 let close_at = now.max(current.valid_from);
1280 let out_key = edge_history_key(src, current.valid_from, current.edge);
1281 let in_key = edge_history_key(dst, current.valid_from, current.edge);
1282 close_edge_history_payload(
1283 self.edges_hist_out
1284 .payload_mut(&out_key)
1285 .ok_or(Error::Corrupt("missing outgoing edge history"))?,
1286 close_at,
1287 );
1288 close_edge_history_payload(
1289 self.edges_hist_in
1290 .payload_mut(&in_key)
1291 .ok_or(Error::Corrupt("missing incoming edge history"))?,
1292 close_at,
1293 );
1294 let out_removed = self.edges_out.remove(&edge_key(src, rel, dst));
1295 let in_removed = self.edges_in.remove(&edge_key(dst, rel, src));
1296 if out_removed != in_removed {
1297 return Err(Error::Corrupt("edge mirrors disagree"));
1298 }
1299 Ok(out_removed)
1300 }
1301
1302 fn current_edge(&self, src: EntityId, rel: TermId, dst: EntityId) -> Option<EdgeSlot> {
1303 self.edges_out
1304 .get_slot(&edge_key(src, rel, dst))
1305 .map(EdgeSlot::read)
1306 }
1307
1308 fn lookup_entity_by_norm(&self, norm: &str) -> Option<EntityId> {
1311 let term = self.terms.lookup(norm)?;
1312 let mut from = [0u8; 8];
1313 key::write_u32(&mut from, term.0);
1314 let mut to = [0u8; 8];
1315 key::write_u32(&mut to, term.0);
1316 to[4..].copy_from_slice(&u32::MAX.to_be_bytes());
1317 self.by_name.range(&from, &to).next().map(|e| e.id)
1318 }
1319
1320 fn lookup_entity_name(&mut self, name: &str) -> Option<EntityId> {
1321 let mut norm = core::mem::take(&mut self.name_scratch);
1322 normalize_name(&mut self.tokenizer, name, &mut norm);
1323 let result = (!norm.is_empty())
1324 .then(|| self.lookup_entity_by_norm(&norm))
1325 .flatten();
1326 self.name_scratch = norm;
1327 result
1328 }
1329
1330 fn resolve_or_create_entity(&mut self, name: &str, now: u64) -> Result<EntityId, Error> {
1331 let mut norm = core::mem::take(&mut self.name_scratch);
1332 normalize_name(&mut self.tokenizer, name, &mut norm);
1333 if norm.is_empty() {
1334 self.name_scratch = norm;
1335 return Err(Error::Invalid("entity name has no indexable characters"));
1336 }
1337 let result = (|| {
1338 if let Some(found) = self.lookup_entity_by_norm(&norm) {
1339 return Ok(found);
1340 }
1341 let term = self.terms.intern(&norm)?;
1342 let id = EntityId(self.next_entity);
1343 let name_id = self.texts.push(name.as_bytes())?;
1344 self.entities.insert(&EntityRecord {
1345 id,
1346 name: name_id,
1347 name_term: term,
1348 created_at: now,
1349 flags: 0,
1350 })?;
1351 self.by_name.insert(&EntityByName {
1352 name_term: term,
1353 id,
1354 })?;
1355 self.next_entity += 1;
1356 Ok(id)
1357 })();
1358 self.name_scratch = norm;
1359 result
1360 }
1361
1362 fn journal_remember<S: Storage>(
1363 &mut self,
1364 store: &mut S,
1365 input: &RememberInput<'_>,
1366 revises: FactId,
1367 assigned: FactId,
1368 ) -> Result<(), Error> {
1369 let mut entry = Vec::new();
1370 Op::Remember {
1371 now: input.now,
1372 valid_from: input.valid_from.unwrap_or(input.now),
1373 entity: input.entity,
1374 text: input.text,
1375 tags: input.tags.to_vec(),
1376 links: input.links.to_vec(),
1377 vector: input.vector.map(<[f32]>::to_vec).unwrap_or_default(),
1378 metadata: input.metadata.map(<[_]>::to_vec).unwrap_or_default(),
1379 revises,
1380 assigned,
1381 }
1382 .encode(&mut entry);
1383 store
1384 .append_journal(&entry)
1385 .map_err(|e| Error::Storage(format!("{e:?}")))
1386 }
1387}
1388
1389impl core::fmt::Debug for Memory<'_> {
1390 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1393 f.debug_struct("Memory")
1394 .field("facts", &self.facts.len())
1395 .field("entities", &self.entities.len())
1396 .field("terms", &self.terms.len())
1397 .finish()
1398 }
1399}
1400
1401fn normalize_name(tokenizer: &mut Tokenizer, name: &str, out: &mut String) {
1405 out.clear();
1406 tokenizer.tokenize(name, &mut |token| {
1407 if !out.is_empty() {
1408 out.push(' ');
1409 }
1410 out.push_str(token);
1411 });
1412}