Skip to main content

nodedb_vector/collection/
checkpoint.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Checkpoint serialization and deserialization for `VectorCollection`.
4//!
5//! ## On-disk framing
6//!
7//! **Plaintext** checkpoints are raw MessagePack bytes (existing format).
8//! The first 4 bytes are never `SEGV`, so detection is unambiguous.
9//!
10//! **Encrypted** checkpoints use the following layout:
11//!
12//! ```text
13//! [SEGV (4B)] [version_u16_le (2B)] [cipher_alg_u8 (1B)] [kid_u8 (1B)]
14//! [HKDF salt (16B)] [random nonce (12B)] [AES-256-GCM ciphertext + tag]
15//! ```
16//!
17//! The complete current preamble is authenticated as AAD. Each envelope uses
18//! its salt and nonce to derive a one-use data key from the KEK, eliminating
19//! restart-sensitive nonce state.
20
21use 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
38/// Magic bytes identifying an encrypted vector checkpoint. Shared with
39/// `nodedb-spatial`'s SEGV checkpoint format.
40const SEGV_MAGIC: [u8; 4] = *b"SEGV";
41
42/// Encrypt `plaintext` into the SEGV envelope from [`nodedb_wal::crypto`].
43fn 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
54/// Decrypt an encrypted checkpoint blob (starting at byte 0, which is `SEGV`).
55fn 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    /// `(global_vector_id, surrogate_u32)` pairs.
80    #[serde(default)]
81    pub surrogate_map: Vec<(u32, u32)>,
82    /// `(document_surrogate_u32, [global_vector_ids])` pairs.
83    #[serde(default)]
84    pub multi_doc_map: Vec<(u32, Vec<u32>)>,
85    /// Quantization mode for the collection-level codec-dispatch index.
86    /// Serialised as a u8 matching `VectorQuantization` discriminants.
87    /// 0 = None (default, backward-compatible).
88    #[serde(default)]
89    pub quantization_tag: u8,
90    /// Serialised `PayloadIndexSetSnapshot` (msgpack bytes).
91    /// Empty vec = no payload indexes (default, backward-compatible).
92    #[serde(default)]
93    pub payload_index_bytes: Vec<u8>,
94    /// Highest WAL LSN whose write is included in this checkpoint. Startup WAL
95    /// replay skips records at or below it for this collection so a
96    /// straddling-segment record already absorbed here is not re-applied.
97    /// `0` (default) on legacy checkpoints predating this field — those gate
98    /// nothing and replay everything, exactly as before.
99    #[serde(default)]
100    pub checkpoint_wal_lsn: u64,
101    /// Index type and PQ/IVF parameters. Its HNSW parameters are the
102    /// `params_*` fields above.
103    pub index_config: crate::index_config::IndexConfig,
104    /// Encoded trained IVF-PQ index. `None` for a non-IVF collection or one
105    /// still buffering toward its training threshold.
106    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    /// Serialized [`Sq8Codec`] bytes (magic + version + msgpack).
118    /// Present when SQ8 quantization is active and PQ is absent.
119    /// `None` means no SQ8 quantization for this segment.
120    #[serde(default)]
121    pub sq8_bytes: Option<Vec<u8>>,
122    /// Pre-quantized SQ8 codes for all vectors in this segment.
123    /// Layout: `[v0_d0, v0_d1, ..., v1_d0, ...]` (dim bytes per vector).
124    /// `None` when SQ8 is not configured.
125    #[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    /// Serialize all segments for checkpointing.
139    ///
140    /// When `kek` is `Some`, the MessagePack payload is wrapped in an
141    /// AES-256-GCM encrypted envelope with a `SEGV` preamble. When `None`,
142    /// raw MessagePack bytes are returned (existing plaintext format).
143    ///
144    /// Returns `Err` on serialization or encryption failure — callers must
145    /// propagate rather than treat a failed encode as an empty checkpoint.
146    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                    // Only serialize SQ8 when PQ is absent — a segment never carries both.
180                    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    /// Restore a collection from checkpoint bytes.
254    ///
255    /// `kek` controls the expected framing:
256    /// - `None` → the file must be plaintext MessagePack (starting with bytes
257    ///   that are NOT `SEGV`). If the file starts with `SEGV` and no key is
258    ///   provided, returns `Err(CheckpointEncryptedNoKey)`.
259    /// - `Some(key)` → encryption is **required**. If the file starts with
260    ///   `SEGV`, it is decrypted with `key`. If the file is plaintext, returns
261    ///   `Err(CheckpointPlaintextKeyRequired)` — refuse to silently load
262    ///   unencrypted data when the operator has enabled at-rest encryption.
263    ///
264    /// Returns `Err(CheckpointDeserializationError)` if the (decrypted)
265    /// MessagePack payload, an embedded sealed-segment HNSW checkpoint, or an
266    /// embedded PQ/SQ8 codec is structurally corrupt.
267    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            // Sealed segments always carry the `RKHNS\0` magic — we wrote
323            // them ourselves. Both a decode `Err` and an `Ok(None)` (magic
324            // missing) mean the checkpoint is structurally corrupt; either
325            // way the whole restore must fail rather than silently drop the
326            // segment's vectors.
327            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            // Restore SQ8 from persisted bytes — never recompute on load.
344            // A segment never carries both PQ and SQ8.
345            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        // A segment sealed but not yet built comes back as a building
371        // segment, searched by brute force; the owning core queues its build.
372        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
452/// Encode a `VectorQuantization` to a u8 tag for storage.
453fn 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
467/// Decode a u8 tag back to `VectorQuantization`.
468fn 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    /// SQ8 calibration data must survive a checkpoint round-trip without
490    /// being recomputed. Verifies that the O(N*dim) rebuild-on-restart bug
491    /// is eliminated: `sq8` on the restored sealed segment is `Some` and
492    /// its `inv_scales` match the original exactly.
493    #[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        // 1024 vectors of dim=8 — enough to pass the ≥1000 threshold in
503        // `build_sq8_for_index`, so sq8 will be Some after complete_build.
504        // Uses plain HNSW (default IndexType::Hnsw) so SQ8 is selected.
505        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        // Capture serialized form as ground truth.
528        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        // Byte-level equality guarantees calibration data is persisted, not recomputed.
542        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    /// The checkpoint watermark (`checkpoint_wal_lsn`) must survive a
571    /// checkpoint round-trip. `note_checkpoint_lsn` only feeds the running
572    /// `applied_wal_lsn` max — it does NOT move the replay gate. The gate
573    /// moves only at checkpoint save (which folds `applied_wal_lsn` into the
574    /// persisted watermark via `max`) and at checkpoint load (which exposes
575    /// that persisted watermark as the new gate). A legacy checkpoint
576    /// predating the field decodes to `0` (via `#[serde(default)]`), which
577    /// gates nothing.
578    #[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    /// `note_checkpoint_lsn` is monotonic and ignores `0` (an unassigned LSN
606    /// must never move the watermark, or a stale-low checkpoint would fail to
607    /// gate a straddling replay). This behavior lives on `applied_wal_lsn`,
608    /// the running max `note_checkpoint_lsn` advances.
609    #[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); // lower — ignored
615        assert_eq!(coll.applied_wal_lsn(), 10);
616        coll.note_checkpoint_lsn(0); // unassigned — ignored
617        assert_eq!(coll.applied_wal_lsn(), 10);
618        coll.note_checkpoint_lsn(20); // higher — advances
619        assert_eq!(coll.applied_wal_lsn(), 20);
620    }
621
622    /// Payload bitmap indexes registered on a vector-primary collection
623    /// must survive a checkpoint round-trip — otherwise `WHERE` filters
624    /// would silently return zero rows after a node restart.
625    #[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}