1use alloc::vec::Vec;
19
20use plugmem_arena::{
21 Arena, ArenaCfg, BlobHeap, BlobHeapCfg, ChunkPool, ChunkPoolCfg, Interner, ShardMode, Slot,
22};
23
24use crate::config::Config;
25use crate::error::Error;
26use crate::id::{FactId, NONE_U32};
27use crate::index::IdListIndex;
28use crate::index::bm25::Bm25Index;
29use crate::index::hnsw::HnswGraph;
30use crate::index::postings::PostingStore;
31use crate::index::varint::decode_u32;
32use crate::index::vecpool::VecPool;
33use crate::memory::FactFault;
34use crate::memory::migrations::{self, STATE_LEN};
35use crate::memory::shards::ShardLayout;
36use crate::model::{
37 EdgeHistorySlot, EdgeSlot, EntityByName, EntityRecord, FactAux, FactRecord, TemporalSlot,
38 VALID_TO_OPEN, edge_history_key, edge_key,
39};
40use crate::snapshot::{Prefix, SectionMeta, Snapshot, SnapshotSink, build_prefix, pad_len};
41use xxhash_rust::xxh3::Xxh3;
42
43use super::Memory;
44
45mod kind {
48 pub const FACTS_META: u16 = 1;
49 pub const FACTS_POOL: u16 = 2;
50 pub const AUX_META: u16 = 3;
51 pub const AUX_POOL: u16 = 4;
52 pub const ENTITIES_META: u16 = 5;
53 pub const ENTITIES_POOL: u16 = 6;
54 pub const BY_NAME_META: u16 = 7;
55 pub const BY_NAME_POOL: u16 = 8;
56 pub const TEMPORAL_META: u16 = 13;
59 pub const TEMPORAL_POOL: u16 = 14;
60 pub const TEXTS_INDEX: u16 = 15;
61 pub const TEXTS_POOL: u16 = 16;
62 pub const TERMS_INDEX: u16 = 17;
63 pub const TERMS_POOL: u16 = 18;
64 pub const TERMS_TABLE: u16 = 19;
65 pub const TAG_LISTS_META: u16 = 20;
66 pub const TAG_LISTS_POOL: u16 = 21;
67 pub const BM25_HANDLES_META: u16 = 22;
68 pub const BM25_HANDLES_POOL: u16 = 23;
69 pub const BM25_CHUNKS_META: u16 = 24;
70 pub const BM25_CHUNKS_POOL: u16 = 25;
71 pub const TAGS_HANDLES_META: u16 = 28;
74 pub const TAGS_HANDLES_POOL: u16 = 29;
75 pub const TAGS_CHUNKS_META: u16 = 30;
76 pub const TAGS_CHUNKS_POOL: u16 = 31;
77 pub const ENTFACTS_HANDLES_META: u16 = 32;
78 pub const ENTFACTS_HANDLES_POOL: u16 = 33;
79 pub const ENTFACTS_CHUNKS_META: u16 = 34;
80 pub const ENTFACTS_CHUNKS_POOL: u16 = 35;
81 pub const ENGINE_STATE: u16 = 36;
82 pub const VEC_POOL: u16 = 37;
83 pub const HNSW_META: u16 = 38;
84 pub const HNSW_LEVEL0: u16 = 39;
85 pub const HNSW_UPPER_META: u16 = 40;
86 pub const HNSW_UPPER_POOL: u16 = 41;
87 pub const HNSW_LISTS_META: u16 = 42;
88 pub const HNSW_LISTS_POOL: u16 = 43;
89 pub const METAS_INDEX: u16 = 44;
90 pub const METAS_POOL: u16 = 45;
91 pub const EDGES_OUT_META: u16 = 50;
93 pub const EDGES_OUT_POOL: u16 = 51;
94 pub const EDGES_IN_META: u16 = 52;
95 pub const EDGES_IN_POOL: u16 = 53;
96 pub const EDGE_HIST_OUT_META: u16 = 54;
98 pub const EDGE_HIST_OUT_POOL: u16 = 55;
99 pub const EDGE_HIST_IN_META: u16 = 56;
100 pub const EDGE_HIST_IN_POOL: u16 = 57;
101 pub const BM25_DOCLEN_META: u16 = 58;
103 pub const BM25_DOCLEN_POOL: u16 = 59;
104}
105
106type SectionFn<'f> = dyn FnMut(u16, &[&[u8]]) -> Result<(), Error> + 'f;
110
111pub(crate) struct Sections<'r, 'a> {
125 pub(crate) facts: &'r Arena<'a, FactRecord>,
126 pub(crate) fact_aux: &'r Arena<'a, FactAux>,
127 pub(crate) entities: &'r Arena<'a, EntityRecord>,
128 pub(crate) by_name: &'r Arena<'a, EntityByName>,
129 pub(crate) temporal: &'r Arena<'a, TemporalSlot>,
130 pub(crate) texts: &'r BlobHeap<'a>,
131 pub(crate) metas: &'r BlobHeap<'a>,
132 pub(crate) tag_lists: &'r ChunkPool<'a>,
133 pub(crate) bm25: &'r Bm25Index<'a>,
134 pub(crate) tags_idx: &'r IdListIndex<'a>,
135 pub(crate) entity_facts: &'r IdListIndex<'a>,
136 pub(crate) vecs: &'r VecPool<'a>,
137 pub(crate) hnsw: &'r HnswGraph<'a>,
138 pub(crate) edges_out: &'r Arena<'a, EdgeSlot>,
139 pub(crate) edges_in: &'r Arena<'a, EdgeSlot>,
140 pub(crate) edges_hist_out: &'r Arena<'a, EdgeHistorySlot>,
141 pub(crate) edges_hist_in: &'r Arena<'a, EdgeHistorySlot>,
142 pub(crate) layout: ShardLayout,
149}
150
151fn arena_sections<T: Slot>(a: &Arena<'_, T>) -> (Vec<u8>, Vec<u8>) {
153 let (mut meta, mut pool) = (Vec::new(), Vec::new());
154 a.dump_meta(&mut meta);
155 a.dump_pool(&mut pool);
156 (meta, pool)
157}
158
159fn section<'a>(snap: &Snapshot<'a>, kind: u16) -> Result<&'a [u8], Error> {
161 snap.section(kind)
162 .ok_or(Error::Corrupt("snapshot is missing a required section"))
163}
164
165struct EdgeSections<'a> {
167 out_meta: &'a [u8],
168 out_pool: &'a [u8],
169 in_meta: &'a [u8],
170 in_pool: &'a [u8],
171 hist_out_meta: &'a [u8],
172 hist_out_pool: &'a [u8],
173 hist_in_meta: &'a [u8],
174 hist_in_pool: &'a [u8],
175}
176
177fn edge_sections<'a>(snap: &Snapshot<'a>) -> Result<Option<EdgeSections<'a>>, Error> {
182 const KINDS: [u16; 8] = [
183 kind::EDGES_OUT_META,
184 kind::EDGES_OUT_POOL,
185 kind::EDGES_IN_META,
186 kind::EDGES_IN_POOL,
187 kind::EDGE_HIST_OUT_META,
188 kind::EDGE_HIST_OUT_POOL,
189 kind::EDGE_HIST_IN_META,
190 kind::EDGE_HIST_IN_POOL,
191 ];
192 let found = KINDS.map(|k| snap.section(k));
193 if found.iter().all(Option::is_none) {
194 return Ok(None);
195 }
196 let [
197 out_meta,
198 out_pool,
199 in_meta,
200 in_pool,
201 hist_out_meta,
202 hist_out_pool,
203 hist_in_meta,
204 hist_in_pool,
205 ] = found.map(|s| s.ok_or(Error::Corrupt("snapshot has incomplete edge sections")));
206 Ok(Some(EdgeSections {
207 out_meta: out_meta?,
208 out_pool: out_pool?,
209 in_meta: in_meta?,
210 in_pool: in_pool?,
211 hist_out_meta: hist_out_meta?,
212 hist_out_pool: hist_out_pool?,
213 hist_in_meta: hist_in_meta?,
214 hist_in_pool: hist_in_pool?,
215 }))
216}
217
218impl<'a, const TF: bool> PostingStore<'a, TF> {
219 pub(crate) fn dump_sections(&self) -> [Vec<u8>; 4] {
221 let (hm, hp) = (self.handles_meta(), self.handles_pool());
222 let (cm, cp) = (self.chunks_meta(), self.chunks_pool());
223 [hm, hp, cm, cp]
224 }
225
226 pub(crate) fn load_sections(
233 shards: usize,
234 max_bytes: usize,
235 hm: &[u8],
236 hp: &[u8],
237 cm: &[u8],
238 cp: &[u8],
239 ) -> Result<Self, Error> {
240 let handles = Arena::<crate::index::postings::IdListSlot>::load(
241 ArenaCfg::new(shards, ShardMode::Uniform).with_max_bytes(max_bytes),
242 hm,
243 hp,
244 )?;
245 let pool = ChunkPool::load(ChunkPoolCfg::new().with_max_bytes(max_bytes), cm, cp)?;
246 Self::validate_lists(&handles, &pool)?;
247 Ok(Self::from_parts(handles, pool))
248 }
249
250 pub(crate) fn load_sections_borrowed(
255 shards: usize,
256 max_bytes: usize,
257 hm: &[u8],
258 hp: &'a [u8],
259 cm: &[u8],
260 cp: &'a [u8],
261 ) -> Result<Self, Error> {
262 let handles = Arena::<crate::index::postings::IdListSlot>::load_borrowed(
263 ArenaCfg::new(shards, ShardMode::Uniform).with_max_bytes(max_bytes),
264 hm,
265 hp,
266 )?;
267 let pool = ChunkPool::load_borrowed(ChunkPoolCfg::new().with_max_bytes(max_bytes), cm, cp)?;
268 Self::validate_lists(&handles, &pool)?;
269 Ok(Self::from_parts(handles, pool))
270 }
271
272 fn validate_lists(
277 handles: &Arena<'_, crate::index::postings::IdListSlot>,
278 pool: &ChunkPool<'_>,
279 ) -> Result<(), Error> {
280 let mut visited = alloc::vec![false; pool.chunks()];
281 for slot in handles.iter() {
282 pool.validate_chain(&slot.handle, &mut visited)?;
283 let mut count = 0u32;
284 let mut last = 0u32;
285 let mut first = true;
286 for chunk in pool.iter(&slot.handle) {
287 let mut cur = chunk;
288 while !cur.is_empty() {
289 let Some((delta, used)) = decode_u32(cur) else {
290 return Err(Error::Corrupt("posting entry is malformed"));
291 };
292 let mut entry_len = used;
293 if TF {
294 if cur.len() < used + 1 {
295 return Err(Error::Corrupt("posting entry is malformed"));
296 }
297 entry_len += 1;
298 }
299 cur = &cur[entry_len..];
300 let id = if first {
301 first = false;
302 delta
303 } else {
304 if delta == 0 {
305 return Err(Error::Corrupt("posting ids are not ascending"));
306 }
307 last.checked_add(delta)
308 .ok_or(Error::Corrupt("posting id overflows"))?
309 };
310 last = id;
311 count += 1;
312 }
313 }
314 if count != slot.count || (count > 0 && last != slot.last) {
315 return Err(Error::Corrupt("posting list disagrees with its handle"));
316 }
317 }
318 if pool.orphan_count(&visited) != 0 {
319 return Err(Error::Corrupt("posting pool has orphan chunks"));
320 }
321 Ok(())
322 }
323}
324
325impl<'a> Bm25Index<'a> {
326 fn dump_pairs(&self) -> [(u16, Vec<u8>); 6] {
328 let [hm, hp, cm, cp] = self.postings().dump_sections();
329 let (dm, dp) = arena_sections(self.doc_len_arena());
330 [
331 (kind::BM25_HANDLES_META, hm),
332 (kind::BM25_HANDLES_POOL, hp),
333 (kind::BM25_CHUNKS_META, cm),
334 (kind::BM25_CHUNKS_POOL, cp),
335 (kind::BM25_DOCLEN_META, dm),
336 (kind::BM25_DOCLEN_POOL, dp),
337 ]
338 }
339
340 fn load_from(snap: &Snapshot<'_>, cfg: &Config) -> Result<Self, Error> {
343 let postings = PostingStore::<true>::load_sections(
344 cfg.shards_postings,
345 cfg.max_bytes,
346 section(snap, kind::BM25_HANDLES_META)?,
347 section(snap, kind::BM25_HANDLES_POOL)?,
348 section(snap, kind::BM25_CHUNKS_META)?,
349 section(snap, kind::BM25_CHUNKS_POOL)?,
350 )?;
351 let (doc_len, migrated) = match migrations::legacy_doc_len(snap, cfg)? {
352 Some(upgraded) => (upgraded, true),
353 None => (
354 Arena::load(
355 migrations::doc_len_cfg(cfg),
356 section(snap, kind::BM25_DOCLEN_META)?,
357 section(snap, kind::BM25_DOCLEN_POOL)?,
358 )?,
359 false,
360 ),
361 };
362 Self::assemble(postings, doc_len, snap, migrated)
363 }
364
365 fn load_from_borrowed(snap: &Snapshot<'a>, cfg: &Config) -> Result<Self, Error> {
368 let postings = PostingStore::<true>::load_sections_borrowed(
369 cfg.shards_postings,
370 cfg.max_bytes,
371 section(snap, kind::BM25_HANDLES_META)?,
372 section(snap, kind::BM25_HANDLES_POOL)?,
373 section(snap, kind::BM25_CHUNKS_META)?,
374 section(snap, kind::BM25_CHUNKS_POOL)?,
375 )?;
376 let (doc_len, migrated) = match migrations::legacy_doc_len(snap, cfg)? {
379 Some(upgraded) => (upgraded, true),
380 None => (
381 Arena::load_borrowed(
382 migrations::doc_len_cfg(cfg),
383 section(snap, kind::BM25_DOCLEN_META)?,
384 section(snap, kind::BM25_DOCLEN_POOL)?,
385 )?,
386 false,
387 ),
388 };
389 Self::assemble(postings, doc_len, snap, migrated)
390 }
391
392 fn assemble(
396 postings: PostingStore<'a, true>,
397 doc_len: Arena<'a, crate::index::bm25::DocLenSlot>,
398 snap: &Snapshot<'_>,
399 migrated: bool,
400 ) -> Result<Self, Error> {
401 let state = section(snap, kind::ENGINE_STATE)?;
402 migrations::decode_engine_state(state)?;
405 let total_docs = u64::from_le_bytes(state[8..16].try_into().unwrap());
406 let total_len = u64::from_le_bytes(state[16..24].try_into().unwrap());
407 if total_docs != doc_len.len() as u64 {
408 return Err(Error::Corrupt("bm25 document total disagrees with doc_len"));
409 }
410 let mut index = Self::from_parts(postings, doc_len, total_docs, total_len);
411 if migrated {
412 index.mark_unsummarized();
413 }
414 Ok(index)
415 }
416}
417
418impl<'a> Memory<'a> {
419 pub(super) fn sections(&self) -> Sections<'_, 'a> {
422 Sections {
423 facts: &self.facts,
424 fact_aux: &self.fact_aux,
425 entities: &self.entities,
426 by_name: &self.by_name,
427 temporal: &self.temporal,
428 texts: &self.texts,
429 metas: &self.metas,
430 tag_lists: &self.tag_lists,
431 bm25: &self.bm25,
432 tags_idx: &self.tags_idx,
433 entity_facts: &self.entity_facts,
434 vecs: &self.vecs,
435 hnsw: &self.hnsw,
436 edges_out: &self.edges_out,
437 edges_in: &self.edges_in,
438 edges_hist_out: &self.edges_hist_out,
439 edges_hist_in: &self.edges_hist_in,
440 layout: ShardLayout::of_config(&self.cfg),
441 }
442 }
443
444 fn emit_sections_from(&self, s: &Sections<'_, '_>, f: &mut SectionFn<'_>) -> Result<(), Error> {
454 for (mk, pk, arena) in [
455 (kind::FACTS_META, kind::FACTS_POOL, arena_sections(s.facts)),
456 (kind::AUX_META, kind::AUX_POOL, arena_sections(s.fact_aux)),
457 (
458 kind::ENTITIES_META,
459 kind::ENTITIES_POOL,
460 arena_sections(s.entities),
461 ),
462 (
463 kind::BY_NAME_META,
464 kind::BY_NAME_POOL,
465 arena_sections(s.by_name),
466 ),
467 (
468 kind::EDGES_OUT_META,
469 kind::EDGES_OUT_POOL,
470 arena_sections(s.edges_out),
471 ),
472 (
473 kind::EDGES_IN_META,
474 kind::EDGES_IN_POOL,
475 arena_sections(s.edges_in),
476 ),
477 (
478 kind::EDGE_HIST_OUT_META,
479 kind::EDGE_HIST_OUT_POOL,
480 arena_sections(s.edges_hist_out),
481 ),
482 (
483 kind::EDGE_HIST_IN_META,
484 kind::EDGE_HIST_IN_POOL,
485 arena_sections(s.edges_hist_in),
486 ),
487 (
488 kind::TEMPORAL_META,
489 kind::TEMPORAL_POOL,
490 arena_sections(s.temporal),
491 ),
492 ] {
493 let (m, p) = arena;
494 f(mk, &[&m])?;
495 f(pk, &[&p])?;
496 }
497 let (mut i, mut p) = (Vec::new(), Vec::new());
498 s.texts.dump_index(&mut i);
499 s.texts.dump_pool(&mut p);
500 f(kind::TEXTS_INDEX, &[&i])?;
501 f(kind::TEXTS_POOL, &[&p])?;
502 let (mut i, mut p) = (Vec::new(), Vec::new());
503 s.metas.dump_index(&mut i);
504 s.metas.dump_pool(&mut p);
505 f(kind::METAS_INDEX, &[&i])?;
506 f(kind::METAS_POOL, &[&p])?;
507 let (mut i, mut p, mut t) = (Vec::new(), Vec::new(), Vec::new());
508 self.terms.dump_index(&mut i);
509 self.terms.dump_pool(&mut p);
510 self.terms.dump_table(&mut t);
511 f(kind::TERMS_INDEX, &[&i])?;
512 f(kind::TERMS_POOL, &[&p])?;
513 f(kind::TERMS_TABLE, &[&t])?;
514 let (mut m, mut p) = (Vec::new(), Vec::new());
515 s.tag_lists.dump_meta(&mut m);
516 s.tag_lists.dump_pool(&mut p);
517 f(kind::TAG_LISTS_META, &[&m])?;
518 f(kind::TAG_LISTS_POOL, &[&p])?;
519 for (k, bytes) in s.bm25.dump_pairs() {
520 f(k, &[&bytes])?;
521 }
522 let [hm, hp, cm, cp] = s.tags_idx.dump_sections();
523 f(kind::TAGS_HANDLES_META, &[&hm])?;
524 f(kind::TAGS_HANDLES_POOL, &[&hp])?;
525 f(kind::TAGS_CHUNKS_META, &[&cm])?;
526 f(kind::TAGS_CHUNKS_POOL, &[&cp])?;
527 let [hm, hp, cm, cp] = s.entity_facts.dump_sections();
528 f(kind::ENTFACTS_HANDLES_META, &[&hm])?;
529 f(kind::ENTFACTS_HANDLES_POOL, &[&hp])?;
530 f(kind::ENTFACTS_CHUNKS_META, &[&cm])?;
531 f(kind::ENTFACTS_CHUNKS_POOL, &[&cp])?;
532 let mut state = Vec::with_capacity(STATE_LEN);
533 state.extend_from_slice(&self.next_fact.to_le_bytes());
534 state.extend_from_slice(&self.next_entity.to_le_bytes());
535 state.extend_from_slice(&s.bm25.docs().to_le_bytes());
536 state.extend_from_slice(&s.bm25.total_len().to_le_bytes());
537 state.extend_from_slice(&self.bm25_tokenizer_version.to_le_bytes());
538 state.extend_from_slice(&0u32.to_le_bytes());
539 state.extend_from_slice(&self.next_edge.to_le_bytes());
540 state.extend_from_slice(&0u32.to_le_bytes());
541 f(kind::ENGINE_STATE, &[&state])?;
542 f(kind::VEC_POOL, &s.vecs.pieces())?;
545 f(kind::HNSW_META, &[&s.hnsw.dump_meta()])?;
548 f(kind::HNSW_LEVEL0, &[&s.hnsw.dump_level0()])?;
549 let [um, up, lm, lp] = s.hnsw.dump_upper();
550 f(kind::HNSW_UPPER_META, &[&um])?;
551 f(kind::HNSW_UPPER_POOL, &[&up])?;
552 f(kind::HNSW_LISTS_META, &[&lm])?;
553 f(kind::HNSW_LISTS_POOL, &[&lp])?;
554 Ok(())
555 }
556
557 pub fn write_snapshot_to(&self, created_at: u64, sink: impl SnapshotSink) -> Result<(), Error> {
569 self.write_snapshot_with(&self.sections(), created_at, sink)
570 }
571
572 pub(crate) fn write_snapshot_with(
579 &self,
580 s: &Sections<'_, '_>,
581 created_at: u64,
582 mut sink: impl SnapshotSink,
583 ) -> Result<(), Error> {
584 let mut cfg_bytes = Vec::new();
585 let mut cfg = self.cfg.clone();
586 s.layout.apply(&mut cfg);
587 cfg.encode(&mut cfg_bytes);
588 let flags = if self.cfg.dim > 0 {
589 crate::snapshot::FLAG_VECTORS
590 } else {
591 0
592 };
593
594 let mut metas: Vec<SectionMeta> = Vec::new();
596 self.emit_sections_from(s, &mut |kind, pieces| {
597 let mut h = Xxh3::new();
598 let mut len = 0u64;
599 for p in pieces {
600 h.update(p);
601 len += p.len() as u64;
602 }
603 metas.push(SectionMeta {
604 kind,
605 len,
606 hash: h.digest(),
607 });
608 Ok(())
609 })?;
610
611 let Prefix {
612 bytes: prefix,
613 offsets,
614 file_len: _,
615 } = build_prefix(
616 &cfg_bytes,
617 flags,
618 created_at,
619 env!("CARGO_PKG_VERSION"),
620 &metas,
621 );
622 sink.write(&prefix)?;
623 let mut file_hash = Xxh3::new();
624 file_hash.update(&prefix);
625
626 let zero = [0u8; 64]; let mut idx = 0usize;
629 self.emit_sections_from(s, &mut |_, pieces| {
630 for p in pieces {
631 sink.write(p)?;
632 file_hash.update(p);
633 }
634 let n = pad_len(offsets[idx], metas[idx].len);
635 sink.write(&zero[..n])?;
636 file_hash.update(&zero[..n]);
637 idx += 1;
638 Ok(())
639 })?;
640
641 sink.patch(
642 crate::snapshot::FILE_HASH_OFFSET,
643 &file_hash.digest().to_le_bytes(),
644 )
645 }
646
647 pub fn snapshot_bytes(&self, created_at: u64) -> Vec<u8> {
652 let mut out = Vec::new();
653 self.write_snapshot_to(created_at, &mut out)
654 .expect("writing a snapshot into a Vec is infallible");
655 out
656 }
657
658 pub fn snapshot<S: crate::storage::Storage>(
660 &mut self,
661 store: &mut S,
662 now: u64,
663 ) -> Result<(), Error> {
664 let bytes = self.snapshot_bytes(now);
665 store
666 .write_snapshot(&bytes)
667 .map_err(|e| Error::Storage(alloc::format!("{e:?}")))?;
668 store
669 .clear_journal()
670 .map_err(|e| Error::Storage(alloc::format!("{e:?}")))?;
671 Ok(())
672 }
673
674 pub(super) fn load_snapshot(bytes: &[u8], cfg: Config) -> Result<Self, Error> {
679 cfg.validate()?;
680 let snap = Snapshot::parse(bytes)?;
681 let cfg = Self::reconcile_config(&snap, cfg)?;
682 let mut mem = Self::new(cfg)?;
683 let cfg = &mem.cfg;
684 let uni =
685 |shards: usize| ArenaCfg::new(shards, ShardMode::Uniform).with_max_bytes(cfg.max_bytes);
686 let ord =
687 |shards: usize| ArenaCfg::new(shards, ShardMode::Ordered).with_max_bytes(cfg.max_bytes);
688 let blob = BlobHeapCfg::new()
689 .with_max_bytes(cfg.max_bytes)
690 .with_max_blob(cfg.max_blob);
691 mem.facts = Arena::load(
692 uni(cfg.shards_facts),
693 section(&snap, kind::FACTS_META)?,
694 section(&snap, kind::FACTS_POOL)?,
695 )?;
696 mem.fact_aux = Arena::load(
697 uni(cfg.shards_facts),
698 section(&snap, kind::AUX_META)?,
699 section(&snap, kind::AUX_POOL)?,
700 )?;
701 mem.entities = Arena::load(
702 uni(cfg.shards_entities),
703 section(&snap, kind::ENTITIES_META)?,
704 section(&snap, kind::ENTITIES_POOL)?,
705 )?;
706 mem.by_name = Arena::load(
707 ord(cfg.shards_entities),
708 section(&snap, kind::BY_NAME_META)?,
709 section(&snap, kind::BY_NAME_POOL)?,
710 )?;
711 if let Some(edges) = edge_sections(&snap)? {
714 mem.edges_out = Arena::load(ord(cfg.shards_edges), edges.out_meta, edges.out_pool)?;
715 mem.edges_in = Arena::load(ord(cfg.shards_edges), edges.in_meta, edges.in_pool)?;
716 mem.edges_hist_out = Arena::load(
717 ord(cfg.shards_edges),
718 edges.hist_out_meta,
719 edges.hist_out_pool,
720 )?;
721 mem.edges_hist_in = Arena::load(
722 ord(cfg.shards_edges),
723 edges.hist_in_meta,
724 edges.hist_in_pool,
725 )?;
726 }
727 mem.temporal = Arena::load(
728 ord(cfg.shards_temporal),
729 section(&snap, kind::TEMPORAL_META)?,
730 section(&snap, kind::TEMPORAL_POOL)?,
731 )?;
732 mem.texts = BlobHeap::load(
733 blob,
734 section(&snap, kind::TEXTS_INDEX)?,
735 section(&snap, kind::TEXTS_POOL)?,
736 )?;
737 mem.metas = BlobHeap::load(
738 blob,
739 section(&snap, kind::METAS_INDEX)?,
740 section(&snap, kind::METAS_POOL)?,
741 )?;
742 mem.terms = Interner::load(
743 blob,
744 section(&snap, kind::TERMS_INDEX)?,
745 section(&snap, kind::TERMS_POOL)?,
746 section(&snap, kind::TERMS_TABLE)?,
747 )?;
748 mem.tag_lists = ChunkPool::load(
749 ChunkPoolCfg::new().with_max_bytes(cfg.max_bytes),
750 section(&snap, kind::TAG_LISTS_META)?,
751 section(&snap, kind::TAG_LISTS_POOL)?,
752 )?;
753 mem.bm25 = Bm25Index::load_from(&snap, cfg)?;
754 mem.tags_idx = IdListIndex::load_sections(
755 cfg.shards_postings,
756 cfg.max_bytes,
757 section(&snap, kind::TAGS_HANDLES_META)?,
758 section(&snap, kind::TAGS_HANDLES_POOL)?,
759 section(&snap, kind::TAGS_CHUNKS_META)?,
760 section(&snap, kind::TAGS_CHUNKS_POOL)?,
761 )?;
762 mem.entity_facts = IdListIndex::load_sections(
763 cfg.shards_entities,
764 cfg.max_bytes,
765 section(&snap, kind::ENTFACTS_HANDLES_META)?,
766 section(&snap, kind::ENTFACTS_HANDLES_POOL)?,
767 section(&snap, kind::ENTFACTS_CHUNKS_META)?,
768 section(&snap, kind::ENTFACTS_CHUNKS_POOL)?,
769 )?;
770 mem.vecs = VecPool::from_parts(cfg.dim, cfg.max_bytes, section(&snap, kind::VEC_POOL)?)?;
771 mem.hnsw = crate::index::hnsw::HnswGraph::from_parts(
772 cfg.hnsw_m,
773 cfg.hnsw_m0,
774 cfg.max_bytes,
775 section(&snap, kind::HNSW_META)?,
776 section(&snap, kind::HNSW_LEVEL0)?,
777 section(&snap, kind::HNSW_UPPER_META)?,
778 section(&snap, kind::HNSW_UPPER_POOL)?,
779 section(&snap, kind::HNSW_LISTS_META)?,
780 section(&snap, kind::HNSW_LISTS_POOL)?,
781 )?;
782 Self::finish_load(mem, &snap)
783 }
784
785 pub(super) fn load_snapshot_borrowed(bytes: &'a [u8], cfg: Config) -> Result<Self, Error> {
793 cfg.validate()?;
794 let snap = Snapshot::parse(bytes)?;
795 let cfg = Self::reconcile_config(&snap, cfg)?;
796 let mut mem = Self::new(cfg)?;
797 let cfg = &mem.cfg;
798 let uni =
799 |shards: usize| ArenaCfg::new(shards, ShardMode::Uniform).with_max_bytes(cfg.max_bytes);
800 let ord =
801 |shards: usize| ArenaCfg::new(shards, ShardMode::Ordered).with_max_bytes(cfg.max_bytes);
802 let blob = BlobHeapCfg::new()
803 .with_max_bytes(cfg.max_bytes)
804 .with_max_blob(cfg.max_blob);
805 mem.facts = Arena::load_borrowed(
806 uni(cfg.shards_facts),
807 section(&snap, kind::FACTS_META)?,
808 section(&snap, kind::FACTS_POOL)?,
809 )?;
810 mem.fact_aux = Arena::load_borrowed(
811 uni(cfg.shards_facts),
812 section(&snap, kind::AUX_META)?,
813 section(&snap, kind::AUX_POOL)?,
814 )?;
815 mem.entities = Arena::load_borrowed(
816 uni(cfg.shards_entities),
817 section(&snap, kind::ENTITIES_META)?,
818 section(&snap, kind::ENTITIES_POOL)?,
819 )?;
820 mem.by_name = Arena::load_borrowed(
821 ord(cfg.shards_entities),
822 section(&snap, kind::BY_NAME_META)?,
823 section(&snap, kind::BY_NAME_POOL)?,
824 )?;
825 if let Some(edges) = edge_sections(&snap)? {
829 mem.edges_out =
830 Arena::load_borrowed(ord(cfg.shards_edges), edges.out_meta, edges.out_pool)?;
831 mem.edges_in =
832 Arena::load_borrowed(ord(cfg.shards_edges), edges.in_meta, edges.in_pool)?;
833 mem.edges_hist_out = Arena::load_borrowed(
834 ord(cfg.shards_edges),
835 edges.hist_out_meta,
836 edges.hist_out_pool,
837 )?;
838 mem.edges_hist_in = Arena::load_borrowed(
839 ord(cfg.shards_edges),
840 edges.hist_in_meta,
841 edges.hist_in_pool,
842 )?;
843 }
844 mem.temporal = Arena::load_borrowed(
845 ord(cfg.shards_temporal),
846 section(&snap, kind::TEMPORAL_META)?,
847 section(&snap, kind::TEMPORAL_POOL)?,
848 )?;
849 mem.texts = BlobHeap::load_borrowed(
850 blob,
851 section(&snap, kind::TEXTS_INDEX)?,
852 section(&snap, kind::TEXTS_POOL)?,
853 )?;
854 mem.metas = BlobHeap::load_borrowed(
855 blob,
856 section(&snap, kind::METAS_INDEX)?,
857 section(&snap, kind::METAS_POOL)?,
858 )?;
859 mem.terms = Interner::load_borrowed(
860 blob,
861 section(&snap, kind::TERMS_INDEX)?,
862 section(&snap, kind::TERMS_POOL)?,
863 section(&snap, kind::TERMS_TABLE)?,
864 )?;
865 mem.tag_lists = ChunkPool::load_borrowed(
866 ChunkPoolCfg::new().with_max_bytes(cfg.max_bytes),
867 section(&snap, kind::TAG_LISTS_META)?,
868 section(&snap, kind::TAG_LISTS_POOL)?,
869 )?;
870 mem.bm25 = Bm25Index::load_from_borrowed(&snap, cfg)?;
871 mem.tags_idx = IdListIndex::load_sections_borrowed(
872 cfg.shards_postings,
873 cfg.max_bytes,
874 section(&snap, kind::TAGS_HANDLES_META)?,
875 section(&snap, kind::TAGS_HANDLES_POOL)?,
876 section(&snap, kind::TAGS_CHUNKS_META)?,
877 section(&snap, kind::TAGS_CHUNKS_POOL)?,
878 )?;
879 mem.entity_facts = IdListIndex::load_sections_borrowed(
880 cfg.shards_entities,
881 cfg.max_bytes,
882 section(&snap, kind::ENTFACTS_HANDLES_META)?,
883 section(&snap, kind::ENTFACTS_HANDLES_POOL)?,
884 section(&snap, kind::ENTFACTS_CHUNKS_META)?,
885 section(&snap, kind::ENTFACTS_CHUNKS_POOL)?,
886 )?;
887 mem.vecs =
888 VecPool::from_parts_borrowed(cfg.dim, cfg.max_bytes, section(&snap, kind::VEC_POOL)?)?;
889 mem.hnsw = crate::index::hnsw::HnswGraph::from_parts_borrowed(
890 cfg.hnsw_m,
891 cfg.hnsw_m0,
892 cfg.max_bytes,
893 section(&snap, kind::HNSW_META)?,
894 section(&snap, kind::HNSW_LEVEL0)?,
895 section(&snap, kind::HNSW_UPPER_META)?,
896 section(&snap, kind::HNSW_UPPER_POOL)?,
897 section(&snap, kind::HNSW_LISTS_META)?,
898 section(&snap, kind::HNSW_LISTS_POOL)?,
899 )?;
900 Self::finish_load(mem, &snap)
901 }
902
903 fn reconcile_config(snap: &Snapshot<'_>, mut cfg: Config) -> Result<Config, Error> {
908 let stored = Config::decode(snap.config())?;
909 if stored.dim != cfg.dim {
910 return Err(Error::ConfigMismatch("stored dim differs"));
911 }
912 ShardLayout::of_config(&stored).apply(&mut cfg);
921 if stored.max_bytes != cfg.max_bytes
922 || stored.max_text != cfg.max_text
923 || stored.max_blob != cfg.max_blob
924 {
925 return Err(Error::ConfigMismatch("stored size limits differ"));
926 }
927 if stored.hnsw_m != cfg.hnsw_m || stored.hnsw_m0 != cfg.hnsw_m0 {
936 return Err(Error::ConfigMismatch("stored hnsw degrees differ"));
937 }
938 if cfg.db_uuid != 0 && stored.db_uuid != cfg.db_uuid {
942 return Err(Error::ConfigMismatch("stored db_uuid differs"));
943 }
944 cfg.db_uuid = stored.db_uuid;
945 Ok(cfg)
946 }
947
948 fn finish_load(mut mem: Self, snap: &Snapshot<'_>) -> Result<Self, Error> {
957 let state = migrations::decode_engine_state(section(snap, kind::ENGINE_STATE)?)?;
958 mem.next_fact = state.next_fact;
959 mem.next_entity = state.next_entity;
960 mem.bm25_tokenizer_version = state.bm25_tokenizer_version;
961 mem.next_edge = state.next_edge;
962 let cfg = mem.cfg.clone();
965 if mem.edges_hist_out.is_empty() && mem.edges_out.is_empty() {
966 mem.migrate_edges(snap, &cfg)?;
967 }
968 let derived_next_edge = mem
969 .edges_hist_out
970 .iter()
971 .map(|edge| edge.edge.0)
972 .max()
973 .map(|edge| edge.saturating_add(1))
974 .unwrap_or(0);
975 if mem.next_edge < derived_next_edge {
976 if !state.predates_edge_versions {
977 return Err(Error::Corrupt("engine edge id counter below record count"));
978 }
979 mem.next_edge = derived_next_edge;
980 }
981 if (mem.next_fact as usize) < mem.facts.len()
982 || (mem.next_entity as usize) < mem.entities.len()
983 {
984 return Err(Error::Corrupt("engine id counters below record counts"));
985 }
986 mem.tombstones = mem.facts.iter().filter(|fact| fact.is_tombstone()).count();
987 mem.validate_references()?;
988 Ok(mem)
989 }
990
991 fn validate_references(&self) -> Result<(), Error> {
1000 let texts = self.texts.len() as u32;
1001 let terms = self.terms.len() as u32;
1002 self.hnsw.validate(&self.vecs)?;
1006 for fact in self.facts.iter() {
1007 if fact.id.0 >= self.next_fact
1008 || fact.text.0 >= texts
1009 || (fact.entity.0 != NONE_U32 && fact.entity.0 >= self.next_entity)
1010 || (fact.revises.0 != NONE_U32 && fact.revises.0 >= self.next_fact)
1011 || fact.kind != 0
1012 {
1013 return Err(Error::Corrupt("fact record references out of range"));
1014 }
1015 if !fact.has_vector() && fact.vector != NONE_U32 {
1019 return Err(Error::Corrupt("fact without a vector flag carries a slot"));
1020 }
1021 }
1022 let metas = self.metas.len() as u32;
1023 let mut visited = alloc::vec![false; self.tag_lists.chunks()];
1024 for aux in self.fact_aux.iter() {
1025 if aux.id.0 >= self.next_fact || (aux.meta.0 != NONE_U32 && aux.meta.0 >= metas) {
1026 return Err(Error::Corrupt("aux record references out of range"));
1027 }
1028 self.tag_lists.validate_chain(&aux.tags, &mut visited)?;
1029 for chunk in self.tag_lists.iter(&aux.tags) {
1030 if !chunk.len().is_multiple_of(4) {
1031 return Err(Error::Corrupt("tag list is not a term-id sequence"));
1032 }
1033 for raw in chunk.chunks_exact(4) {
1034 if u32::from_be_bytes(raw.try_into().unwrap()) >= terms {
1035 return Err(Error::Corrupt("tag term out of range"));
1036 }
1037 }
1038 }
1039 }
1040 if self.tag_lists.orphan_count(&visited) != 0 {
1041 return Err(Error::Corrupt("tag pool has orphan chunks"));
1042 }
1043 for entity in self.entities.iter() {
1044 if entity.id.0 >= self.next_entity
1045 || entity.name.0 >= texts
1046 || entity.name_term.0 >= terms
1047 {
1048 return Err(Error::Corrupt("entity record references out of range"));
1049 }
1050 }
1051 for by_name in self.by_name.iter() {
1052 if by_name.name_term.0 >= terms || !self.entities.contains(&by_name.id.0.to_be_bytes())
1053 {
1054 return Err(Error::Corrupt("by-name record references out of range"));
1055 }
1056 }
1057 for arena in [&self.edges_out, &self.edges_in] {
1065 for edge in arena.iter() {
1066 if edge.a.0 >= self.next_entity
1067 || edge.b.0 >= self.next_entity
1068 || edge.rel.0 >= terms
1069 || edge.edge.0 >= self.next_edge
1070 || (edge.fact.0 != NONE_U32 && edge.fact.0 >= self.next_fact)
1071 {
1072 return Err(Error::Corrupt("edge record references out of range"));
1073 }
1074 }
1075 }
1076 if self.edges_out.len() != self.edges_in.len()
1077 || self.edges_hist_out.len() != self.edges_hist_in.len()
1078 {
1079 return Err(Error::Corrupt("edge mirrors disagree"));
1080 }
1081 for edge in self.edges_hist_out.iter() {
1082 if edge.a.0 >= self.next_entity
1083 || edge.b.0 >= self.next_entity
1084 || edge.edge.0 >= self.next_edge
1085 || edge.rel.0 >= terms
1086 || edge.kind != 0
1087 || edge.valid_from > edge.valid_to
1088 || (edge.fact.0 != NONE_U32 && edge.fact.0 >= self.next_fact)
1089 {
1090 return Err(Error::Corrupt("edge history references out of range"));
1091 }
1092 }
1093 for slot in self.temporal.iter() {
1094 if slot.fact.0 >= self.next_fact {
1095 return Err(Error::Corrupt("temporal record references out of range"));
1096 }
1097 }
1098 Ok(())
1099 }
1100
1101 pub fn verify(&self) -> Result<(), Error> {
1131 self.verify_graph()?;
1132 for (_, text) in self.texts.iter() {
1135 if core::str::from_utf8(text).is_err() {
1136 return Err(Error::Corrupt("stored text is not valid UTF-8"));
1137 }
1138 }
1139 let mut pairs = Vec::new();
1144 for aux in self.fact_aux.iter() {
1145 if aux.meta.0 != NONE_U32 {
1146 crate::metadata::decode(self.metas.get(aux.meta), &mut pairs)?;
1147 }
1148 }
1149 self.vecs.validate()?;
1153 let vslots = self.vecs.len() as u32;
1154 let mut with_vec = 0u32;
1155 for fact in self.facts.iter() {
1156 if fact.has_vector() {
1157 if fact.vector >= vslots || self.vecs.slot_fact(fact.vector as usize) != fact.id.0 {
1158 return Err(Error::Corrupt(
1159 "fact vector slot is out of range or mismatched",
1160 ));
1161 }
1162 with_vec += 1;
1163 }
1164 }
1165 if with_vec != vslots {
1166 return Err(Error::Corrupt("vector pool has orphan slots"));
1167 }
1168 Ok(())
1169 }
1170
1171 fn verify_graph(&self) -> Result<(), Error> {
1174 for arena in [&self.edges_out, &self.edges_in] {
1175 for edge in arena.iter() {
1176 if !self.entities.contains(&edge.a.0.to_be_bytes())
1177 || !self.entities.contains(&edge.b.0.to_be_bytes())
1178 {
1179 return Err(Error::Corrupt("edge names an entity that does not exist"));
1180 }
1181 }
1182 }
1183 for edge in self.edges_out.iter() {
1184 if !self.edges_in.contains(&edge_key(edge.b, edge.rel, edge.a)) {
1185 return Err(Error::Corrupt("edge mirrors disagree"));
1186 }
1187 let version = self
1190 .edges_hist_out
1191 .get(&edge_history_key(edge.a, edge.valid_from, edge.edge))
1192 .ok_or(Error::Corrupt("current edge has no history record"))?;
1193 if version.valid_to != VALID_TO_OPEN
1194 || version.rel != edge.rel
1195 || version.b != edge.b
1196 || version.fact != edge.fact
1197 {
1198 return Err(Error::Corrupt("current edge disagrees with its history"));
1199 }
1200 }
1201 for edge in self.edges_hist_out.iter() {
1202 if !self.entities.contains(&edge.a.0.to_be_bytes())
1203 || !self.entities.contains(&edge.b.0.to_be_bytes())
1204 {
1205 return Err(Error::Corrupt(
1206 "edge history names an entity that does not exist",
1207 ));
1208 }
1209 if !self
1210 .edges_hist_in
1211 .contains(&edge_history_key(edge.b, edge.valid_from, edge.edge))
1212 {
1213 return Err(Error::Corrupt("edge history mirrors disagree"));
1214 }
1215 if edge.valid_to == VALID_TO_OPEN
1219 && !self.edges_out.contains(&edge_key(edge.a, edge.rel, edge.b))
1220 {
1221 return Err(Error::Corrupt("open edge version is not a current edge"));
1222 }
1223 }
1224 Ok(())
1225 }
1226
1227 pub fn faulty_facts(&self) -> Vec<(FactId, FactFault)> {
1239 let vslots = self.vecs.len() as u32;
1240 let metas = self.metas.len() as u32;
1241 let mut pairs = Vec::new();
1242 let mut out = Vec::new();
1243 for i in self.fact_ids_ascending() {
1244 let id = FactId(i);
1245 let Some(record) = self.fact(id) else {
1246 continue; };
1248 if record.is_tombstone() {
1249 continue;
1250 }
1251 if core::str::from_utf8(self.texts.get(record.text)).is_err() {
1252 out.push((id, FactFault::Text));
1253 continue;
1254 }
1255 if record.has_vector()
1256 && (record.vector >= vslots || self.vecs.slot_fact(record.vector as usize) != id.0)
1257 {
1258 out.push((id, FactFault::Vector));
1259 continue;
1260 }
1261 if let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes())
1265 && aux.meta.0 != NONE_U32
1266 && (aux.meta.0 >= metas
1267 || crate::metadata::decode(self.metas.get(aux.meta), &mut pairs).is_err())
1268 {
1269 out.push((id, FactFault::Metadata));
1270 }
1271 }
1272 out
1273 }
1274}