zeppelin_embed/lifecycle/
snapshot_view.rs

1//! In-place read-only views: shared data, independent handle lifetime.
2
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::sync::{Arc, Condvar, Mutex, RwLock};
5
6use super::{LexicalIndexCache, Store, StoreError, StoreLock, StoreState};
7
8pub(super) struct SnapshotPin {
9    count: Arc<AtomicU64>,
10    // A source closed before its views must not permit a new writer to reclaim
11    // their retired files through open-time orphan cleanup.
12    _writer: Arc<StoreLock>,
13    #[cfg(feature = "graph-cypher")]
14    _graph: Option<super::native_graph::NativeSnapshotPin>,
15}
16
17impl Drop for SnapshotPin {
18    fn drop(&mut self) {
19        self.count.fetch_sub(1, Ordering::AcqRel);
20    }
21}
22
23impl Store {
24    /// Opens an in-place read-only view of one complete generation.
25    ///
26    /// Requires a writable source. Active data and sealed mappings are shared;
27    /// no store files are created or copied. The view remains valid after the
28    /// source closes, and must itself be closed to release its retention pin.
29    /// Physical purge is busy while any view is open. Retired segment paths are
30    /// retained until ordinary orphan cleanup on the next writable open.
31    pub fn open_snapshot(&self) -> Result<Self, StoreError> {
32        let state = self
33            .state
34            .lock()
35            .map_err(|_| StoreError::Synchronization { component: "state" })?;
36        match *state {
37            StoreState::Open => {}
38            StoreState::Closing => return Err(StoreError::Closing),
39            StoreState::Closed => return Err(StoreError::Closed),
40        }
41        let writer = self
42            .writer_lock
43            .lock()
44            .map_err(|_| StoreError::Synchronization {
45                component: "writer lock",
46            })?;
47        let writer = Arc::clone(writer.as_ref().ok_or(StoreError::ReadOnly)?);
48        let active = self
49            .active
50            .lock()
51            .map_err(|_| StoreError::Synchronization {
52                component: "active segment",
53            })?;
54        let active = active.as_ref().ok_or(StoreError::Closed)?;
55        let published = self
56            .snapshot
57            .read()
58            .map_err(|_| StoreError::Synchronization {
59                component: "published snapshot",
60            })?;
61        let published = published.as_ref().ok_or(StoreError::Closed)?;
62        let snapshot = published.fork_read_view(&self.accounting)?;
63        #[cfg(feature = "graph-cypher")]
64        let graph_pin = if published.graph_enabled {
65            Some(self.native_graph.pin_snapshot(&[])?)
66        } else {
67            None
68        };
69        #[cfg(feature = "graph-cypher")]
70        let native_graph = match graph_pin.as_ref() {
71            Some(pin) => {
72                super::native_graph::NativeGraphPublication::fork_snapshot(&self.accounting, pin)?
73            }
74            None => super::native_graph::NativeGraphPublication::new(&self.accounting, false)?,
75        };
76        self.snapshot_pins
77            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| {
78                count.checked_add(1)
79            })
80            .map_err(|_| StoreError::GenerationOverflow)?;
81        let pin = SnapshotPin {
82            count: Arc::clone(&self.snapshot_pins),
83            _writer: writer,
84            #[cfg(feature = "graph-cypher")]
85            _graph: graph_pin,
86        };
87        Ok(Self {
88            private_preparation: None,
89            directory: self.directory.clone(),
90            open_migrations: super::OpenMigrations {
91                generation: active.generation,
92                ..super::OpenMigrations::default()
93            },
94            vfs: Arc::clone(&self.vfs),
95            clock: Arc::clone(&self.clock),
96            state: Mutex::new(StoreState::Open),
97            state_changed: Condvar::new(),
98            background: Mutex::new(None),
99            query_pool: Mutex::new(None),
100            lexical_worker: Mutex::new(None),
101            #[cfg(feature = "graph-cypher")]
102            native_graph,
103            #[cfg(feature = "graph-cypher")]
104            graph_enable_pending: std::sync::atomic::AtomicBool::new(false),
105            snapshot: RwLock::new(Some(Arc::new(snapshot))),
106            active: Mutex::new(Some(crate::ingest::ActiveState {
107                generation: active.generation,
108                segment: Arc::clone(&active.segment),
109            })),
110            wal_writer: Mutex::new(None),
111            writer_lock: Mutex::new(None),
112            #[cfg(feature = "graph-cypher")]
113            reader_store_lock: Mutex::new(None),
114            logical_writer_lock: Mutex::new(None),
115            reclamation_pin: Mutex::new(
116                self.reclamation_pin
117                    .lock()
118                    .map_err(|_| StoreError::Synchronization {
119                        component: "reclamation pin",
120                    })?
121                    .clone(),
122            ),
123            snapshot_pins: Arc::clone(&self.snapshot_pins),
124            snapshot_pin: Mutex::new(Some(pin)),
125            maintenance: Mutex::new(()),
126            health_state: Mutex::new(crate::diag::HealthState::default()),
127            durability_policy: self.durability_policy,
128            reader_drain_timeout: self.reader_drain_timeout,
129            accounting: Arc::clone(&self.accounting),
130            active_queries: AtomicU64::new(0),
131            #[cfg(any(test, feature = "test-seams"))]
132            text_materialization_work: super::materialize::TestMaterializationWork::default(),
133            epoch: self.epoch.clone(),
134            epoch_alias: crate::epoch::EpochAliasCell::new(self.epoch_alias.load()),
135            tokenizer: self.tokenizer.clone(),
136            schema: self.schema.clone(),
137            lexical_index_cache: LexicalIndexCache::new(),
138            #[cfg(any(test, feature = "test-seams"))]
139            ingest_retention_fault_controller: None,
140            #[cfg(any(test, feature = "test-seams"))]
141            hybrid_leg_fault: Mutex::new(None),
142            #[cfg(any(test, feature = "test-seams"))]
143            hybrid_execution_receipt: Mutex::new(None),
144            #[cfg(any(test, feature = "test-seams"))]
145            metadata_test_controller: None,
146            #[cfg(any(test, feature = "test-seams"))]
147            vector_fault_controller: None,
148            #[cfg(any(test, feature = "test-seams"))]
149            kernel_fault_controller: None,
150            #[cfg(any(test, feature = "test-seams"))]
151            vector_seal_scheme: None,
152            #[cfg(test)]
153            teardown_probe: Arc::new(super::close::TeardownProbe::new()),
154        })
155    }
156
157    pub(crate) fn has_snapshot_views(&self) -> bool {
158        self.snapshot_pins.load(Ordering::Acquire) != 0
159    }
160
161    pub(crate) fn require_no_snapshot_views(&self) -> Result<(), StoreError> {
162        if self.has_snapshot_views() {
163            return Err(StoreError::SnapshotViewsOpen);
164        }
165        Ok(())
166    }
167}
zeppelin_embed/lifecycle/snapshot_view.rs

zeppelin_embed/lifecycle/
snapshot_view.rs
