1use std::collections::HashMap;
22
23use nodedb_types::{Surrogate, VectorQuantization};
24use serde::{Deserialize, Serialize};
25
26use crate::collection::payload_index::PayloadIndexSetSnapshot;
27use crate::collection::segment::{BuildingSegment, DEFAULT_SEAL_THRESHOLD, SealedSegment};
28use crate::collection::tier::StorageTier;
29use crate::error::VectorError;
30use crate::flat::FlatIndex;
31use crate::hnsw::{HnswIndex, HnswParams};
32use crate::ivf::IvfPqIndex;
33use crate::quantize::pq::PqCodec;
34use crate::quantize::sq8::Sq8Codec;
35
36use super::lifecycle::VectorCollection;
37
38const SEGV_MAGIC: [u8; 4] = *b"SEGV";
41
42fn encrypt_checkpoint(
44 key: &nodedb_wal::crypto::WalEncryptionKey,
45 plaintext: &[u8],
46) -> Result<Vec<u8>, VectorError> {
47 nodedb_wal::crypto::encrypt_segment_envelope(key, &SEGV_MAGIC, plaintext).map_err(|e| {
48 VectorError::CheckpointEncryptionError {
49 detail: e.to_string(),
50 }
51 })
52}
53
54fn decrypt_checkpoint(
56 key: &nodedb_wal::crypto::WalEncryptionKey,
57 blob: &[u8],
58) -> Result<Vec<u8>, VectorError> {
59 nodedb_wal::crypto::decrypt_segment_envelope(key, &SEGV_MAGIC, blob).map_err(|e| {
60 VectorError::CheckpointEncryptionError {
61 detail: e.to_string(),
62 }
63 })
64}
65
66#[derive(Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack)]
67pub(crate) struct CollectionSnapshot {
68 pub dim: usize,
69 pub params_m: usize,
70 pub params_m0: usize,
71 pub params_ef_construction: usize,
72 pub params_metric: u8,
73 pub next_id: u32,
74 pub growing_base_id: u32,
75 pub growing_vectors: Vec<Vec<f32>>,
76 pub growing_deleted: Vec<bool>,
77 pub sealed_segments: Vec<SealedSnapshot>,
78 pub building_segments: Vec<BuildingSnapshot>,
79 #[serde(default)]
81 pub surrogate_map: Vec<(u32, u32)>,
82 #[serde(default)]
84 pub multi_doc_map: Vec<(u32, Vec<u32>)>,
85 #[serde(default)]
89 pub quantization_tag: u8,
90 #[serde(default)]
93 pub payload_index_bytes: Vec<u8>,
94 #[serde(default)]
100 pub checkpoint_wal_lsn: u64,
101 pub index_config: crate::index_config::IndexConfig,
104 pub ivf_bytes: Option<Vec<u8>>,
107}
108
109#[derive(Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack)]
110pub(crate) struct SealedSnapshot {
111 pub base_id: u32,
112 pub hnsw_bytes: Vec<u8>,
113 #[serde(default)]
114 pub pq_bytes: Option<Vec<u8>>,
115 #[serde(default)]
116 pub pq_codes: Option<Vec<u8>>,
117 #[serde(default)]
121 pub sq8_bytes: Option<Vec<u8>>,
122 #[serde(default)]
126 pub sq8_codes: Option<Vec<u8>>,
127}
128
129#[derive(Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack)]
130pub(crate) struct BuildingSnapshot {
131 pub base_id: u32,
132 pub vectors: Vec<Vec<f32>>,
133 #[serde(default)]
134 pub deleted: Vec<bool>,
135}
136
137impl VectorCollection {
138 pub fn checkpoint_to_bytes(
147 &self,
148 kek: Option<&nodedb_wal::crypto::WalEncryptionKey>,
149 ) -> Result<Vec<u8>, VectorError> {
150 let snapshot = CollectionSnapshot {
151 dim: self.dim,
152 params_m: self.params.m,
153 params_m0: self.params.m0,
154 params_ef_construction: self.params.ef_construction,
155 params_metric: self.params.metric as u8,
156 next_id: self.next_id,
157 growing_base_id: self.growing_base_id,
158 growing_vectors: (0..self.growing.len() as u32)
159 .filter_map(|i| self.growing.get_vector_raw(i).map(|v| v.to_vec()))
160 .collect(),
161 growing_deleted: (0..self.growing.len() as u32)
162 .map(|i| self.growing.is_deleted(i))
163 .collect(),
164 sealed_segments: self
165 .sealed
166 .iter()
167 .map(|s| {
168 let (pq_bytes, pq_codes) = match &s.pq {
169 Some((codec, codes)) => (
170 Some(codec.to_bytes().map_err(|e| {
171 VectorError::CheckpointSerializationError {
172 detail: format!("PQ codec encode: {e}"),
173 }
174 })?),
175 Some(codes.clone()),
176 ),
177 None => (None, None),
178 };
179 let (sq8_bytes, sq8_codes) = if pq_bytes.is_none() {
181 match &s.sq8 {
182 Some((codec, codes)) => (Some(codec.to_bytes()), Some(codes.clone())),
183 None => (None, None),
184 }
185 } else {
186 (None, None)
187 };
188 Ok(SealedSnapshot {
189 base_id: s.base_id,
190 hnsw_bytes: s.index.checkpoint_to_bytes()?,
191 pq_bytes,
192 pq_codes,
193 sq8_bytes,
194 sq8_codes,
195 })
196 })
197 .collect::<Result<Vec<_>, VectorError>>()?,
198 building_segments: self
199 .building
200 .iter()
201 .map(|b| BuildingSnapshot {
202 base_id: b.base_id,
203 vectors: (0..b.flat.len() as u32)
204 .filter_map(|i| b.flat.get_vector_raw(i).map(|v| v.to_vec()))
205 .collect(),
206 deleted: (0..b.flat.len() as u32)
207 .map(|i| b.flat.is_deleted(i))
208 .collect(),
209 })
210 .collect(),
211 surrogate_map: self
212 .surrogate_map
213 .iter()
214 .map(|(&k, s)| (k, s.as_u32()))
215 .collect(),
216 multi_doc_map: self
217 .multi_doc_map
218 .iter()
219 .map(|(k, v)| (k.as_u32(), v.clone()))
220 .collect(),
221 quantization_tag: quantization_to_tag(self.quantization),
222 payload_index_bytes: {
223 let snap = self.payload.to_snapshot();
224 match zerompk::to_msgpack_vec(&snap) {
225 Ok(bytes) => bytes,
226 Err(e) => {
227 return Err(VectorError::CheckpointSerializationError {
228 detail: e.to_string(),
229 });
230 }
231 }
232 },
233 checkpoint_wal_lsn: self.checkpoint_wal_lsn.max(self.applied_wal_lsn),
234 index_config: self.index_config.clone(),
235 ivf_bytes: self.ivf.as_ref().map(IvfPqIndex::to_bytes).transpose()?,
236 };
237 let msgpack = match zerompk::to_msgpack_vec(&snapshot) {
238 Ok(bytes) => bytes,
239 Err(e) => {
240 return Err(VectorError::CheckpointSerializationError {
241 detail: e.to_string(),
242 });
243 }
244 };
245
246 if let Some(key) = kek {
247 encrypt_checkpoint(key, &msgpack)
248 } else {
249 Ok(msgpack)
250 }
251 }
252
253 pub fn from_checkpoint(
268 bytes: &[u8],
269 kek: Option<&nodedb_wal::crypto::WalEncryptionKey>,
270 memory: nodedb_mem::ScopedMemory,
271 ) -> Result<Self, VectorError> {
272 let is_encrypted = bytes.len() >= 4 && bytes[0..4] == SEGV_MAGIC;
273
274 let msgpack: Vec<u8>;
275 let msgpack_ref: &[u8];
276
277 if is_encrypted {
278 if let Some(key) = kek {
279 msgpack = decrypt_checkpoint(key, bytes)?;
280 msgpack_ref = &msgpack;
281 } else {
282 return Err(VectorError::CheckpointEncryptedNoKey);
283 }
284 } else if kek.is_some() {
285 return Err(VectorError::CheckpointPlaintextKeyRequired);
286 } else {
287 msgpack_ref = bytes;
288 }
289
290 let snap: CollectionSnapshot = match zerompk::from_msgpack(msgpack_ref) {
291 Ok(s) => s,
292 Err(e) => {
293 return Err(VectorError::CheckpointDeserializationError {
294 detail: format!("collection msgpack decode: {e}"),
295 });
296 }
297 };
298 let metric = crate::hnsw::checkpoint::metric_from_code(snap.params_metric)?;
299 let params = HnswParams {
300 m: snap.params_m,
301 m0: snap.params_m0,
302 ef_construction: snap.params_ef_construction,
303 metric,
304 dtype: nodedb_types::vector_dtype::VectorStorageDtype::F32,
305 };
306
307 let mut growing = FlatIndex::new(snap.dim, metric);
308 for (i, v) in snap.growing_vectors.iter().enumerate() {
309 let deleted = snap.growing_deleted.get(i).copied().unwrap_or(false);
310 let inserted = if deleted {
311 growing.insert_tombstoned(v.clone())
312 } else {
313 growing.insert(v.clone())
314 };
315 inserted.map_err(|e| VectorError::CheckpointDeserializationError {
316 detail: format!("growing-segment replay insert: {e}"),
317 })?;
318 }
319
320 let mut sealed = Vec::with_capacity(snap.sealed_segments.len());
321 for ss in &snap.sealed_segments {
322 let index = HnswIndex::from_checkpoint(&ss.hnsw_bytes)?.ok_or_else(|| {
328 VectorError::CheckpointDeserializationError {
329 detail: "sealed segment hnsw bytes missing RKHNS magic".into(),
330 }
331 })?;
332 let pq = match (&ss.pq_bytes, &ss.pq_codes) {
333 (Some(bytes), Some(codes)) => Some((
334 PqCodec::from_bytes(bytes, memory.clone()).map_err(|e| {
335 VectorError::CheckpointDeserializationError {
336 detail: format!("PQ codec decode: {e}"),
337 }
338 })?,
339 codes.clone(),
340 )),
341 _ => None,
342 };
343 let sq8 = if pq.is_some() {
346 None
347 } else {
348 match (&ss.sq8_bytes, &ss.sq8_codes) {
349 (Some(codec_bytes), Some(codes)) => Some((
350 Sq8Codec::from_bytes(codec_bytes).map_err(|e| {
351 VectorError::CheckpointDeserializationError {
352 detail: format!("SQ8 codec decode: {e}"),
353 }
354 })?,
355 codes.clone(),
356 )),
357 _ => None,
358 }
359 };
360 sealed.push(SealedSegment {
361 index,
362 base_id: ss.base_id,
363 sq8,
364 pq,
365 tier: StorageTier::L0Ram,
366 mmap_vectors: None,
367 });
368 }
369
370 let mut next_segment_id = (sealed.len() + 1) as u32;
373 let mut building = Vec::with_capacity(snap.building_segments.len());
374 for bs in &snap.building_segments {
375 let mut flat = FlatIndex::new(snap.dim, metric);
376 for (i, v) in bs.vectors.iter().enumerate() {
377 let inserted = if bs.deleted.get(i).copied().unwrap_or(false) {
378 flat.insert_tombstoned(v.clone())
379 } else {
380 flat.insert(v.clone())
381 };
382 inserted.map_err(|e| VectorError::CheckpointDeserializationError {
383 detail: format!("building-segment replay insert: {e}"),
384 })?;
385 }
386 building.push(BuildingSegment {
387 flat,
388 base_id: bs.base_id,
389 segment_id: next_segment_id,
390 });
391 next_segment_id += 1;
392 }
393
394 let index_config = crate::index_config::IndexConfig {
395 hnsw: params.clone(),
396 ..snap.index_config
397 };
398 let ivf = snap
399 .ivf_bytes
400 .as_deref()
401 .map(|bytes| IvfPqIndex::from_bytes(bytes, memory.clone()))
402 .transpose()?;
403 Ok(Self {
404 growing,
405 growing_base_id: snap.growing_base_id,
406 sealed,
407 building,
408 params,
409 next_id: snap.next_id,
410 next_segment_id,
411 dim: snap.dim,
412 data_dir: None,
413 ram_budget_bytes: 0,
414 mmap_fallback_count: 0,
415 mmap_segment_count: 0,
416 surrogate_map: snap
417 .surrogate_map
418 .iter()
419 .map(|&(k, s)| (k, Surrogate::new(s)))
420 .collect(),
421 surrogate_to_local: snap
422 .surrogate_map
423 .iter()
424 .map(|&(k, s)| (Surrogate::new(s), k))
425 .collect(),
426 multi_doc_map: snap
427 .multi_doc_map
428 .into_iter()
429 .map(|(k, v)| (Surrogate::new(k), v))
430 .collect::<HashMap<_, _>>(),
431 seal_threshold: DEFAULT_SEAL_THRESHOLD,
432 index_config,
433 codec_dispatch: None,
434 ivf,
435 quantization: quantization_from_tag(snap.quantization_tag),
436 payload: if snap.payload_index_bytes.is_empty() {
437 super::payload_index::PayloadIndexSet::default()
438 } else {
439 zerompk::from_msgpack::<PayloadIndexSetSnapshot>(&snap.payload_index_bytes)
440 .map(super::payload_index::PayloadIndexSet::from_snapshot)
441 .unwrap_or_default()
442 },
443 arena_index: None,
444 checkpoint_wal_lsn: snap.checkpoint_wal_lsn,
445 applied_wal_lsn: snap.checkpoint_wal_lsn,
446 builds_completed: 0,
447 builds_failed: 0,
448 })
449 }
450}
451
452fn quantization_to_tag(q: VectorQuantization) -> u8 {
454 match q {
455 VectorQuantization::None => 0,
456 VectorQuantization::Sq8 => 1,
457 VectorQuantization::Pq => 2,
458 VectorQuantization::RaBitQ => 3,
459 VectorQuantization::Bbq => 4,
460 VectorQuantization::Binary => 5,
461 VectorQuantization::Ternary => 6,
462 VectorQuantization::Opq => 7,
463 _ => 0,
464 }
465}
466
467fn quantization_from_tag(tag: u8) -> VectorQuantization {
469 match tag {
470 0 => VectorQuantization::None,
471 1 => VectorQuantization::Sq8,
472 2 => VectorQuantization::Pq,
473 3 => VectorQuantization::RaBitQ,
474 4 => VectorQuantization::Bbq,
475 5 => VectorQuantization::Binary,
476 6 => VectorQuantization::Ternary,
477 7 => VectorQuantization::Opq,
478 _ => VectorQuantization::None,
479 }
480}
481
482#[cfg(test)]
483mod tests {
484 use crate::collection::lifecycle::VectorCollection;
485 use crate::distance::DistanceMetric;
486 use crate::hnsw::HnswParams;
487 use crate::test_support::test_memory;
488
489 #[test]
494 fn checkpoint_roundtrip_preserves_sq8() {
495 use crate::collection::lifecycle::VectorCollection;
496 use crate::hnsw::{HnswIndex, HnswParams};
497
498 let params = HnswParams {
499 metric: crate::distance::DistanceMetric::L2,
500 ..HnswParams::default()
501 };
502 let mut coll = VectorCollection::with_seal_threshold(8, params, 1024);
506 for i in 0..1024u32 {
507 let mut v = vec![0.0f32; 8];
508 for (d, slot) in v.iter_mut().enumerate() {
509 *slot = ((i as f32) * 0.01 + (d as f32) * 0.1).sin();
510 }
511 coll.insert(v).unwrap();
512 }
513 let req = coll.seal("sq8_test").expect("seal produced request");
514 let mut idx = HnswIndex::new(req.dim, req.params.clone());
515 for v in &req.vectors {
516 idx.insert(v.clone()).unwrap();
517 }
518 coll.complete_build(req.segment_id, idx, test_memory());
519
520 let sealed = coll.sealed_segments();
521 assert!(!sealed.is_empty(), "expected at least one sealed segment");
522 let orig_sq8 = sealed[0]
523 .sq8
524 .as_ref()
525 .expect("sq8 must be Some after complete_build with ≥1000 vectors");
526 let orig_dim = orig_sq8.0.dim();
527 let orig_bytes = orig_sq8.0.to_bytes();
529
530 let checkpoint = coll.checkpoint_to_bytes(None).unwrap();
531 let restored = VectorCollection::from_checkpoint(&checkpoint, None, test_memory()).unwrap();
532
533 let restored_sealed = restored.sealed_segments();
534 assert!(!restored_sealed.is_empty());
535 let restored_sq8 = restored_sealed[0]
536 .sq8
537 .as_ref()
538 .expect("sq8 must be Some after restoring checkpoint — never recomputed");
539
540 assert_eq!(restored_sq8.0.dim(), orig_dim, "dim mismatch after restore");
541 assert_eq!(
543 restored_sq8.0.to_bytes(),
544 orig_bytes,
545 "sq8 codec bytes differ — calibration data was recomputed rather than persisted"
546 );
547 }
548
549 #[test]
550 fn checkpoint_roundtrip() {
551 let mut coll = VectorCollection::new(
552 3,
553 HnswParams {
554 metric: DistanceMetric::L2,
555 ..HnswParams::default()
556 },
557 );
558 for i in 0..50u32 {
559 coll.insert(vec![i as f32, 0.0, 0.0]).unwrap();
560 }
561 let bytes = coll.checkpoint_to_bytes(None).unwrap();
562 let restored = VectorCollection::from_checkpoint(&bytes, None, test_memory()).unwrap();
563 assert_eq!(restored.len(), 50);
564 assert_eq!(restored.dim(), 3);
565
566 let results = restored.search(&[25.0, 0.0, 0.0], 1, 64).unwrap();
567 assert_eq!(results[0].id, 25);
568 }
569
570 #[test]
579 fn checkpoint_roundtrip_preserves_wal_lsn() {
580 let mut coll = VectorCollection::new(
581 3,
582 HnswParams {
583 metric: DistanceMetric::L2,
584 ..HnswParams::default()
585 },
586 );
587 coll.insert(vec![1.0, 0.0, 0.0]).unwrap();
588 coll.note_checkpoint_lsn(42);
589 assert_eq!(coll.applied_wal_lsn(), 42);
590 assert_eq!(
591 coll.checkpoint_wal_lsn(),
592 0,
593 "gate is not advanced by note; only by load/save"
594 );
595
596 let bytes = coll.checkpoint_to_bytes(None).unwrap();
597 let restored = VectorCollection::from_checkpoint(&bytes, None, test_memory()).unwrap();
598 assert_eq!(
599 restored.checkpoint_wal_lsn(),
600 42,
601 "save folds applied_wal_lsn into the persisted gate, so restore exposes it"
602 );
603 }
604
605 #[test]
610 fn note_checkpoint_lsn_is_monotonic_and_ignores_zero() {
611 let mut coll = VectorCollection::new(3, HnswParams::default());
612 coll.note_checkpoint_lsn(10);
613 assert_eq!(coll.applied_wal_lsn(), 10);
614 coll.note_checkpoint_lsn(5); assert_eq!(coll.applied_wal_lsn(), 10);
616 coll.note_checkpoint_lsn(0); assert_eq!(coll.applied_wal_lsn(), 10);
618 coll.note_checkpoint_lsn(20); assert_eq!(coll.applied_wal_lsn(), 20);
620 }
621
622 #[test]
626 fn checkpoint_roundtrip_preserves_payload_bitmap() {
627 use crate::collection::PayloadIndexKind;
628 use crate::collection::payload_index::FilterPredicate;
629 use nodedb_types::Value;
630 use std::collections::HashMap;
631
632 let mut coll = VectorCollection::new(
633 3,
634 HnswParams {
635 metric: DistanceMetric::L2,
636 ..HnswParams::default()
637 },
638 );
639 coll.payload
640 .add_index("category".to_string(), PayloadIndexKind::Equality);
641 for i in 0u32..10 {
642 let node_id = coll.insert(vec![i as f32, 0.0, 0.0]).unwrap();
643 let mut fields = HashMap::new();
644 let cat = if i % 2 == 0 { "A" } else { "B" };
645 fields.insert("category".to_string(), Value::String(cat.to_string()));
646 coll.payload.insert_row(node_id, &fields);
647 }
648
649 let bytes = coll.checkpoint_to_bytes(None).unwrap();
650 let restored = VectorCollection::from_checkpoint(&bytes, None, test_memory()).unwrap();
651
652 let pred = FilterPredicate::Eq {
653 field: "category".to_string(),
654 value: Value::String("A".to_string()),
655 };
656 let bm = restored
657 .payload
658 .pre_filter(&pred)
659 .expect("payload index 'category' must be present after restore");
660 assert_eq!(
661 bm.len(),
662 5,
663 "5 rows of category=A must survive checkpoint round-trip"
664 );
665 }
666}