zeppelin_embed/lifecycle/
snapshot_copy.rs

1//! ZE-220: a consistent point-in-time copy of an open store.
2//!
3//! A snapshot pins one generation for a short time, then copies what that
4//! generation references without holding any store lock:
5//!
6//! - Pin. Under the WAL writer mutex, which every manifest commit and
7//!   snapshot publication also holds, it takes a reader lease on the
8//!   published segment set, reads the committed manifest bytes, and clones
9//!   the retained WAL records after the manifest's absorbed prefix. The cost
10//!   is O(manifest + unsealed records); the records share the writer's
11//!   encoded buffers.
12//! - Copy. Segments are copied from the lease's mappings, so a segment that
13//!   seal, maintenance or a sealed-row rewrite unlinks during the copy is
14//!   still read in full. The WAL is rebuilt as a header whose first sequence
15//!   follows the absorbed prefix, then the pinned records. Close cancels the
16//!   lease, and the copy stops at its next chunk.
17//! - Publish. Everything is written into a hidden staging directory beside
18//!   the target, each file and the directory are fully synced, the staging
19//!   directory is renamed to the target, and the parent is synced. A crash
20//!   before the rename leaves no target; the staging directory is never a
21//!   snapshot.
22//!
23//! The snapshot is an ordinary store directory. Opening it read-only or
24//! read-write restores exactly the pinned state.
25
26use std::path::{Path, PathBuf};
27use std::sync::atomic::{AtomicU64, Ordering};
28
29use crate::manifest::decode_manifest;
30use crate::manifest::io::MANIFEST_FILE;
31use crate::vfs::{SyncKind, Vfs};
32use crate::wal::header::encode_header;
33use crate::wal::{LogSeq, VisibleRecord, WalWriteError};
34
35use super::{PublishedSnapshot, SnapshotLease, Store, StoreError, StoreState};
36
37const WAL_FILE: &str = "wal.ze";
38/// Copy granularity. Close cancellation is checked between chunks.
39const COPY_CHUNK_BYTES: usize = 8 * 1024 * 1024;
40
41static STAGING_SEQUENCE: AtomicU64 = AtomicU64::new(0);
42
43/// Why a snapshot target was rejected.
44#[derive(Clone, Copy, Debug, Eq, PartialEq)]
45pub enum SnapshotTargetReason {
46    /// The path has no final name component, such as `/` or `a/..`.
47    NoFinalName,
48    /// The target's parent directory does not exist.
49    ParentMissing,
50    /// The target is the store directory or lies inside it.
51    InsideStore,
52    /// The target exists and is not a directory, such as a file or a link.
53    NotADirectory,
54    /// The target is a directory that already has entries.
55    NotEmpty,
56}
57
58impl std::fmt::Display for SnapshotTargetReason {
59    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
60        formatter.write_str(match self {
61            Self::NoFinalName => "has no final name",
62            Self::ParentMissing => "parent directory does not exist",
63            Self::InsideStore => "is inside the store",
64            Self::NotADirectory => "exists and is not a directory",
65            Self::NotEmpty => "is not empty",
66        })
67    }
68}
69
70/// A validated target: its canonical parent and final path.
71struct Target {
72    parent: PathBuf,
73    path: PathBuf,
74    /// An existing empty directory is replaced at publication.
75    existed: bool,
76}
77
78/// Everything one snapshot copies, captured at one generation.
79struct Pinned {
80    lease: SnapshotLease,
81    /// Committed manifest bytes; `None` for a store that never committed one.
82    manifest: Option<Vec<u8>>,
83    wal_first_seq: LogSeq,
84    wal_records: Vec<VisibleRecord>,
85}
86
87impl Store {
88    /// Writes a consistent snapshot of this store into `target` and returns
89    /// the generation it captured.
90    ///
91    /// `target` must not exist, or must be an empty directory; its parent
92    /// must exist; and it must not be the store directory or lie inside it.
93    /// A rejected target is [`StoreError::SnapshotTarget`] and nothing is
94    /// written. The handle must be writable ([`StoreError::ReadOnly`]
95    /// otherwise), because only the writer owns the WAL tail. While a
96    /// physical purge is pending the call is refused with
97    /// [`StoreError::SnapshotPurgePending`].
98    ///
99    /// Writers are blocked only while the generation is pinned, which is
100    /// O(manifest + unsealed records) and copies no document bytes. Writes,
101    /// seal and maintenance run during the copy and are absent from the
102    /// snapshot. Close cancels an in-flight snapshot
103    /// ([`StoreError::ReadCancelled`]); a failed snapshot removes its
104    /// staging directory and never creates `target`.
105    ///
106    /// The snapshot directory is an ordinary store: opening it read-only or
107    /// read-write restores exactly the state at the returned generation.
108    pub fn write_snapshot(&self, target: impl AsRef<Path>) -> Result<u64, StoreError> {
109        let target = resolve_target(&self.directory, target.as_ref())?;
110        let pinned = self.pin_snapshot()?;
111        let vfs = self.vfs.as_ref();
112        let staging = target.staging_path();
113        vfs.create_directory(&staging)
114            .map_err(|source| io(&staging, source))?;
115        if let Err(error) =
116            stage(vfs, &staging, &pinned).and_then(|()| target.publish(vfs, &staging))
117        {
118            // The original error is the one reported. After a completed
119            // rename the staging path is gone and this finds nothing.
120            remove_staging(vfs, &staging);
121            return Err(error);
122        }
123        Ok(pinned.lease.generation())
124    }
125
126    fn pin_snapshot(&self) -> Result<Pinned, StoreError> {
127        let state = self
128            .state
129            .lock()
130            .map_err(|_| StoreError::Synchronization { component: "state" })?;
131        match *state {
132            StoreState::Open => {}
133            StoreState::Closing => return Err(StoreError::Closing),
134            StoreState::Closed => return Err(StoreError::Closed),
135        }
136        // Every manifest commit and snapshot publication holds this mutex,
137        // so the manifest file, the published snapshot, the active
138        // generation and the WAL tail agree while it is held.
139        let wal = self
140            .wal_writer
141            .lock()
142            .map_err(|_| StoreError::Synchronization {
143                component: "WAL writer",
144            })?;
145        let writer = wal.as_ref().ok_or(StoreError::ReadOnly)?;
146        let active = self
147            .active
148            .lock()
149            .map_err(|_| StoreError::Synchronization {
150                component: "active segment",
151            })?;
152        let generation = active.as_ref().ok_or(StoreError::Closed)?.generation;
153        // `purge` returns with its intent on disk and the bytes it will
154        // remove still in the store; copying them would defeat the purge.
155        let intent = self
156            .directory
157            .join(crate::ingest::purge_support::PURGE_INTENT_FILE);
158        match self.vfs.open(&intent) {
159            Ok(_) => return Err(StoreError::SnapshotPurgePending),
160            Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
161            Err(source) => return Err(io(&intent, source)),
162        }
163        let published = self
164            .snapshot
165            .read()
166            .map_err(|_| StoreError::Synchronization {
167                component: "published snapshot",
168            })?
169            .as_ref()
170            .cloned()
171            .ok_or(StoreError::Closed)?;
172        let manifest = self.pinned_manifest(&published)?;
173        let absorbed_through = published.absorbed_through();
174        let wal_records = writer.unabsorbed_records(absorbed_through)?;
175        drop(active);
176        drop(wal);
177        drop(state);
178        Ok(Pinned {
179            lease: SnapshotLease::new_at(published, generation),
180            manifest,
181            wal_first_seq: LogSeq::new(absorbed_through.saturating_add(1)),
182            wal_records,
183        })
184    }
185
186    /// Reads the committed manifest and proves it describes `published`.
187    fn pinned_manifest(
188        &self,
189        published: &PublishedSnapshot,
190    ) -> Result<Option<Vec<u8>>, StoreError> {
191        let path = self.directory.join(MANIFEST_FILE);
192        let bytes = match self.vfs.read(&path) {
193            Ok(bytes) => bytes,
194            Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
195                if published.all_segments().is_empty() && published.absorbed_through() == 0 {
196                    return Ok(None);
197                }
198                return Err(StoreError::SnapshotPin {
199                    detail: "the published snapshot has no committed manifest",
200                });
201            }
202            Err(source) => return Err(io(&path, source)),
203        };
204        let manifest =
205            decode_manifest(&path.display().to_string(), &bytes).map_err(StoreError::Manifest)?;
206        let mut committed = manifest
207            .segments
208            .iter()
209            .map(|segment| segment.id)
210            .collect::<Vec<_>>();
211        let mut pinned = published
212            .all_segments()
213            .iter()
214            .map(|segment| segment.meta().id)
215            .collect::<Vec<_>>();
216        committed.sort_unstable();
217        pinned.sort_unstable();
218        if manifest.generation != published.generation()
219            || manifest.log_seq != published.absorbed_through()
220            || committed != pinned
221        {
222            return Err(StoreError::SnapshotPin {
223                detail: "the committed manifest does not describe the published snapshot",
224            });
225        }
226        Ok(Some(bytes))
227    }
228}
229
230fn io(path: &Path, source: std::io::Error) -> StoreError {
231    StoreError::Io {
232        path: path.to_path_buf(),
233        source,
234    }
235}
236
237fn resolve_target(store_directory: &Path, target: &Path) -> Result<Target, StoreError> {
238    let reject = |reason| StoreError::SnapshotTarget {
239        path: target.to_path_buf(),
240        reason,
241    };
242    let name = target
243        .file_name()
244        .ok_or_else(|| reject(SnapshotTargetReason::NoFinalName))?;
245    let parent = match target.parent() {
246        Some(parent) if !parent.as_os_str().is_empty() => parent,
247        _ => Path::new("."),
248    };
249    let parent = match std::fs::canonicalize(parent) {
250        Ok(parent) => parent,
251        Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
252            return Err(reject(SnapshotTargetReason::ParentMissing));
253        }
254        Err(source) => return Err(io(parent, source)),
255    };
256    let store =
257        std::fs::canonicalize(store_directory).map_err(|source| io(store_directory, source))?;
258    let path = parent.join(name);
259    if path.starts_with(&store) {
260        return Err(reject(SnapshotTargetReason::InsideStore));
261    }
262    let existed = match std::fs::symlink_metadata(&path) {
263        Ok(metadata) if metadata.is_dir() => {
264            let mut entries = std::fs::read_dir(&path).map_err(|source| io(&path, source))?;
265            if entries.next().is_some() {
266                return Err(reject(SnapshotTargetReason::NotEmpty));
267            }
268            true
269        }
270        Ok(_) => return Err(reject(SnapshotTargetReason::NotADirectory)),
271        Err(source) if source.kind() == std::io::ErrorKind::NotFound => false,
272        Err(source) => return Err(io(&path, source)),
273    };
274    Ok(Target {
275        parent,
276        path,
277        existed,
278    })
279}
280
281impl Target {
282    /// A hidden sibling that no other snapshot, process or crash leftover
283    /// shares; it is created exclusively.
284    fn staging_path(&self) -> PathBuf {
285        let name = self
286            .path
287            .file_name()
288            .map(|name| name.to_string_lossy().into_owned())
289            .unwrap_or_default();
290        let nanos = std::time::SystemTime::now()
291            .duration_since(std::time::UNIX_EPOCH)
292            .map_or(0, |elapsed| elapsed.as_nanos());
293        let sequence = STAGING_SEQUENCE.fetch_add(1, Ordering::Relaxed);
294        self.parent.join(format!(
295            ".{name}.snapshot-{}-{nanos}-{sequence}.tmp",
296            std::process::id()
297        ))
298    }
299
300    fn publish(&self, vfs: &dyn Vfs, staging: &Path) -> Result<(), StoreError> {
301        if self.existed {
302            vfs.remove_directory(&self.path)
303                .map_err(|source| io(&self.path, source))?;
304        }
305        vfs.rename(staging, &self.path)
306            .map_err(|source| io(&self.path, source))?;
307        vfs.sync(&self.parent, SyncKind::Full)
308            .map_err(|source| io(&self.parent, source))
309    }
310}
311
312fn stage(vfs: &dyn Vfs, staging: &Path, pinned: &Pinned) -> Result<(), StoreError> {
313    let lease = &pinned.lease;
314    for segment in lease.all_segments() {
315        let bytes = segment.file_bytes();
316        if u64::try_from(bytes.len()).ok() != Some(segment.meta().file_size) {
317            return Err(StoreError::SnapshotPin {
318                detail: "a pinned segment mapping differs from its manifest length",
319            });
320        }
321        write_file(
322            vfs,
323            &staging.join(segment.meta().id.file_name()),
324            lease,
325            bytes.chunks(COPY_CHUNK_BYTES).map(Ok),
326        )?;
327    }
328    write_wal(vfs, &staging.join(WAL_FILE), pinned)?;
329    if let Some(manifest) = &pinned.manifest {
330        write_file(
331            vfs,
332            &staging.join(MANIFEST_FILE),
333            lease,
334            std::iter::once(Ok(manifest.as_slice())),
335        )?;
336    }
337    vfs.sync(staging, SyncKind::Full)
338        .map_err(|source| io(staging, source))
339}
340
341/// The snapshot's WAL: a header naming the first unabsorbed sequence, then
342/// every pinned record exactly as the writer encoded it.
343fn write_wal(vfs: &dyn Vfs, path: &Path, pinned: &Pinned) -> Result<(), StoreError> {
344    let header = encode_header(pinned.wal_first_seq)
345        .map_err(|error| StoreError::WalWrite(WalWriteError::Header(error)))?;
346    let records = pinned.wal_records.iter().map(|record| {
347        record.encoded().map_err(|source| StoreError::WalRecord {
348            seq: record.seq,
349            source,
350        })
351    });
352    write_file(
353        vfs,
354        path,
355        &pinned.lease,
356        std::iter::once(Ok(header.as_slice())).chain(records),
357    )
358}
359
360fn write_file<'a>(
361    vfs: &dyn Vfs,
362    path: &Path,
363    lease: &SnapshotLease,
364    chunks: impl IntoIterator<Item = Result<&'a [u8], StoreError>>,
365) -> Result<(), StoreError> {
366    let mut file = vfs.open_append(path).map_err(|source| io(path, source))?;
367    for chunk in chunks {
368        lease.check_active()?;
369        file.append(chunk?).map_err(|source| io(path, source))?;
370    }
371    file.sync(SyncKind::Full).map_err(|source| io(path, source))
372}
373
374fn remove_staging(vfs: &dyn Vfs, staging: &Path) {
375    if let Ok(paths) = vfs.list(staging) {
376        for path in paths {
377            let _ = vfs.delete(&path);
378        }
379    }
380    let _ = vfs.remove_directory(staging);
381}
zeppelin_embed/lifecycle/snapshot_copy.rs

zeppelin_embed/lifecycle/
snapshot_copy.rs
