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, commit_manifest, 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 durable_end = wal.as_ref().ok_or(StoreError::ReadOnly)?.durable_end();
44        let mut active = self
45            .active
46            .lock()
47            .map_err(|_| StoreError::Synchronization {
48                component: "active segment",
49            })?;
50        let current = active.as_mut().ok_or(StoreError::Closed)?;
51        let snapshot = self
52            .snapshot
53            .read()
54            .map_err(|_| StoreError::Synchronization {
55                component: "published snapshot",
56            })?
57            .clone()
58            .ok_or(StoreError::Closed)?;
59        if snapshot.all_segments().is_empty() {
60            return Ok(current.generation);
61        }
62        let mut manifest = load_manifest(
63            self.vfs.as_ref(),
64            &self.directory.join(MANIFEST_FILE),
65            durable_end,
66        )
67        .map_err(StoreError::Manifest)?;
68        for epoch in &manifest.epochs {
69            super::validate_tokenizer_epoch(
70                Some(crate::epoch::EpochIdentity {
71                    embedding: epoch.id,
72                    tokenizer: epoch.tokenizer,
73                }),
74                &self.tokenizer,
75            )?;
76        }
77        let generation = manifest
78            .generation
79            .max(current.generation)
80            .checked_add(1)
81            .ok_or(StoreError::GenerationOverflow)?;
82        let mut replacements = Vec::new();
83        for segment in snapshot.all_segments() {
84            let mut bytes = *segment.meta().id.as_bytes();
85            let hash = xxhash_rust::xxh3::xxh3_64_with_seed(&bytes, generation ^ 0x5245494e444558);
86            bytes
87                .get_mut(..8)
88                .ok_or(StoreError::GenerationOverflow)?
89                .copy_from_slice(&generation.to_be_bytes());
90            bytes
91                .get_mut(8..)
92                .ok_or(StoreError::GenerationOverflow)?
93                .copy_from_slice(&hash.to_be_bytes());
94            replacements.push(
95                crate::segment::writer::reindex_segment(
96                    self.vfs.as_ref(),
97                    &self.directory,
98                    segment,
99                    SegmentId::from_bytes(bytes),
100                    &self.tokenizer,
101                    self.durability_policy,
102                )
103                .map_err(StoreError::Segment)?,
104            );
105        }
106        manifest.segments = replacements;
107        manifest.generation = generation;
108        commit_manifest(
109            self.vfs.as_ref(),
110            &self.directory,
111            &manifest,
112            self.durability_policy,
113        )
114        .map_err(StoreError::Manifest)?;
115        let remapped =
116            PublishedSnapshot::load_on_vfs(&self.directory, &self.accounting, self.vfs.as_ref())?;
117        *self
118            .snapshot
119            .write()
120            .map_err(|_| StoreError::Synchronization {
121                component: "published snapshot",
122            })? = Some(Arc::new(remapped));
123        current.generation = generation;
124        // Keep old files for pinned readers; ordinary open reachability cleanup reclaims them.
125        Ok(generation)
126    }
127}
zeppelin_embed/lifecycle/reindex.rs

zeppelin_embed/lifecycle/
reindex.rs
