zeppelin_embed/lifecycle/
reindex.rs

1//! Explicit same-tokenizer reindex using immutable segment replacement.
2use super::{PublishedSnapshot, Store, StoreError, StoreState};
3use crate::manifest::io::{MANIFEST_FILE, load_manifest};
4use crate::segment::SegmentId;
5use std::sync::Arc;
6
7impl Store {
8    /// Rebuilds every sealed text index from stored text under the current tokenizer.
9    /// Active text is already indexed; seal it first. Publication replaces every
10    /// sealed segment in one manifest commit, preserving all non-postings regions.
11    pub fn reindex_text(&self) -> Result<u64, StoreError> {
12        self.seal()?;
13        let _maintenance = self
14            .maintenance
15            .lock()
16            .map_err(|_| StoreError::Synchronization {
17                component: "maintenance",
18            })?;
19        let state = self
20            .state
21            .lock()
22            .map_err(|_| StoreError::Synchronization { component: "state" })?;
23        match *state {
24            StoreState::Open => {}
25            StoreState::Closing => return Err(StoreError::Closing),
26            StoreState::Closed => return Err(StoreError::Closed),
27        }
28        let writer = self
29            .writer_lock
30            .lock()
31            .map_err(|_| StoreError::Synchronization {
32                component: "writer lock",
33            })?;
34        if writer.is_none() {
35            return Err(StoreError::ReadOnly);
36        }
37        let wal = self
38            .wal_writer
39            .lock()
40            .map_err(|_| StoreError::Synchronization {
41                component: "WAL writer",
42            })?;
43        let mut publication = wal
44            .as_ref()
45            .ok_or(StoreError::ReadOnly)?
46            .manifest_publication()?;
47        let durable_end = wal.as_ref().ok_or(StoreError::ReadOnly)?.durable_end();
48        let mut active = self
49            .active
50            .lock()
51            .map_err(|_| StoreError::Synchronization {
52                component: "active segment",
53            })?;
54        let current = active.as_mut().ok_or(StoreError::Closed)?;
55        let snapshot = self
56            .snapshot
57            .read()
58            .map_err(|_| StoreError::Synchronization {
59                component: "published snapshot",
60            })?
61            .clone()
62            .ok_or(StoreError::Closed)?;
63        if snapshot.all_segments().is_empty() {
64            return Ok(current.generation);
65        }
66        let mut manifest = load_manifest(
67            self.vfs.as_ref(),
68            &self.directory.join(MANIFEST_FILE),
69            durable_end,
70        )
71        .map_err(StoreError::Manifest)?;
72        for epoch in &manifest.epochs {
73            super::validate_tokenizer_epoch(
74                Some(crate::epoch::EpochIdentity {
75                    embedding: epoch.id,
76                    tokenizer: epoch.tokenizer,
77                }),
78                &self.tokenizer,
79            )?;
80        }
81        let generation = manifest
82            .generation
83            .max(current.generation)
84            .checked_add(1)
85            .ok_or(StoreError::GenerationOverflow)?;
86        let mut replacements = Vec::new();
87        for segment in snapshot.all_segments() {
88            let mut bytes = *segment.meta().id.as_bytes();
89            let hash = xxhash_rust::xxh3::xxh3_64_with_seed(&bytes, generation ^ 0x5245494e444558);
90            bytes
91                .get_mut(..8)
92                .ok_or(StoreError::GenerationOverflow)?
93                .copy_from_slice(&generation.to_be_bytes());
94            bytes
95                .get_mut(8..)
96                .ok_or(StoreError::GenerationOverflow)?
97                .copy_from_slice(&hash.to_be_bytes());
98            replacements.push(
99                crate::segment::writer::reindex_segment(
100                    self.vfs.as_ref(),
101                    &self.directory,
102                    segment,
103                    SegmentId::from_bytes(bytes),
104                    &self.tokenizer,
105                    self.durability_policy,
106                )
107                .map_err(StoreError::Segment)?,
108            );
109        }
110        manifest.segments = replacements;
111        manifest.generation = generation;
112        #[cfg(feature = "graph-cypher")]
113        manifest
114            .record_generation_bump(durable_end)
115            .map_err(StoreError::Manifest)?;
116        publication
117            .commit_manifest(
118                self.vfs.as_ref(),
119                &self.directory,
120                &manifest,
121                self.durability_policy,
122            )
123            .map_err(StoreError::Manifest)?;
124        let remapped =
125            PublishedSnapshot::load_on_vfs(&self.directory, &self.accounting, self.vfs.as_ref())?;
126        *self
127            .snapshot
128            .write()
129            .map_err(|_| StoreError::Synchronization {
130                component: "published snapshot",
131            })? = Some(Arc::new(remapped));
132        current.generation = generation;
133        publication.complete();
134        // Keep old files for pinned readers; ordinary open reachability cleanup reclaims them.
135        Ok(generation)
136    }
137}
zeppelin_embed/lifecycle/reindex.rs

zeppelin_embed/lifecycle/
reindex.rs
