weavatrix_memory/snapshot/
file.rs

1use super::SnapshotStore;
2use crate::{Codec, Durability, MemoryError, ProjectionSnapshot, Result};
3use std::{
4    fs::{self, OpenOptions},
5    io::Write,
6    marker::PhantomData,
7    path::{Path, PathBuf},
8    sync::atomic::{AtomicU64, Ordering},
9};
10
11mod frame;
12
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub struct SnapshotOptions {
15    pub durability: Durability,
16    pub max_snapshot_bytes: usize,
17}
18
19impl Default for SnapshotOptions {
20    fn default() -> Self {
21        Self {
22            durability: Durability::SyncAll,
23            max_snapshot_bytes: 512 * 1024 * 1024,
24        }
25    }
26}
27
28pub struct FileSnapshotStore<P, C> {
29    directory: PathBuf,
30    prefix: String,
31    codec: C,
32    options: SnapshotOptions,
33    #[cfg(feature = "mmap")]
34    mapped_reads: bool,
35    marker: PhantomData<fn() -> P>,
36}
37
38impl<P, C> FileSnapshotStore<P, C>
39where
40    C: Codec<ProjectionSnapshot<P>>,
41{
42    /// Creates an immutable, generation-named snapshot store.
43    ///
44    /// # Errors
45    ///
46    /// Rejects an invalid prefix or inaccessible directory.
47    pub fn open(
48        directory: impl AsRef<Path>,
49        prefix: impl Into<String>,
50        codec: C,
51        options: SnapshotOptions,
52    ) -> Result<Self> {
53        let prefix = prefix.into();
54        if prefix.is_empty() || prefix.trim() != prefix || prefix.contains(['/', '\\']) {
55            return Err(MemoryError::InvalidValue {
56                field: "snapshot.prefix",
57                reason: "must be a simple non-empty file prefix",
58            });
59        }
60        if options.max_snapshot_bytes == 0 {
61            return Err(MemoryError::InvalidValue {
62                field: "max_snapshot_bytes",
63                reason: "must be greater than zero",
64            });
65        }
66        let directory = directory.as_ref().to_path_buf();
67        fs::create_dir_all(&directory).map_err(|error| io("create snapshot directory", error))?;
68        Ok(Self {
69            directory,
70            prefix,
71            codec,
72            options,
73            #[cfg(feature = "mmap")]
74            mapped_reads: false,
75            marker: PhantomData,
76        })
77    }
78
79    /// Enables guarded, read-only memory mapping for snapshot loads.
80    ///
81    /// Generation files created by this store are immutable. Other processes
82    /// must still honor advisory file locks and must never truncate a mapped
83    /// generation.
84    #[cfg(feature = "mmap")]
85    #[must_use]
86    pub fn with_memory_mapped_reads(mut self) -> Self {
87        self.mapped_reads = true;
88        self
89    }
90
91    fn final_path(&self, position: u64) -> PathBuf {
92        self.directory
93            .join(format!("{}-{position:020}.wmsnap", self.prefix))
94    }
95
96    fn latest_path(&self) -> Result<Option<(u64, PathBuf)>> {
97        let start = format!("{}-", self.prefix);
98        let mut latest = None;
99        for entry in
100            fs::read_dir(&self.directory).map_err(|error| io("read snapshot directory", error))?
101        {
102            let entry = entry.map_err(|error| io("read snapshot entry", error))?;
103            let name = entry.file_name();
104            let Some(name) = name.to_str() else {
105                continue;
106            };
107            let Some(raw) = name
108                .strip_prefix(&start)
109                .and_then(|value| value.strip_suffix(".wmsnap"))
110            else {
111                continue;
112            };
113            let Ok(position) = raw.parse::<u64>() else {
114                continue;
115            };
116            if latest
117                .as_ref()
118                .is_none_or(|(current, _)| position > *current)
119            {
120                latest = Some((position, entry.path()));
121            }
122        }
123        Ok(latest)
124    }
125
126    fn read_path(&self, path: &Path, expected_position: u64) -> Result<ProjectionSnapshot<P>> {
127        #[cfg(feature = "mmap")]
128        let payload = frame::read(path, self.options.max_snapshot_bytes, self.mapped_reads)?;
129        #[cfg(not(feature = "mmap"))]
130        let payload = frame::read(path, self.options.max_snapshot_bytes)?;
131        let snapshot = self.codec.decode(payload.as_ref())?;
132        if snapshot.cursor.global_position != Some(expected_position) {
133            return Err(corrupt("snapshot filename and cursor disagree"));
134        }
135        Ok(snapshot)
136    }
137
138    fn encode_frame(&self, snapshot: &ProjectionSnapshot<P>) -> Result<Vec<u8>> {
139        let bytes = self.codec.encode(snapshot)?;
140        if bytes.len() > self.options.max_snapshot_bytes {
141            return Err(MemoryError::InvalidValue {
142                field: "snapshot",
143                reason: "encoded snapshot exceeds max_snapshot_bytes",
144            });
145        }
146        frame::encode(&bytes)
147    }
148}
149
150impl<P, C> SnapshotStore<P> for FileSnapshotStore<P, C>
151where
152    C: Codec<ProjectionSnapshot<P>>,
153{
154    fn save(&mut self, snapshot: &ProjectionSnapshot<P>) -> Result<()> {
155        let position = snapshot
156            .cursor
157            .global_position
158            .ok_or(MemoryError::InvalidValue {
159                field: "snapshot.cursor",
160                reason: "cannot persist an empty replay cursor",
161            })?;
162        let final_path = self.final_path(position);
163        let frame = self.encode_frame(snapshot)?;
164        if final_path.exists() {
165            let existing =
166                fs::read(&final_path).map_err(|error| io("read existing snapshot", error))?;
167            if existing == frame {
168                return Ok(());
169            }
170            return Err(MemoryError::InvalidValue {
171                field: "snapshot",
172                reason: "different snapshot already exists at this position",
173            });
174        }
175        let temporary = temporary_path(&final_path);
176        let mut guard = TempGuard::new(temporary.clone());
177        let mut file = OpenOptions::new()
178            .write(true)
179            .create_new(true)
180            .open(&temporary)
181            .map_err(|error| io("create temporary snapshot", error))?;
182        file.write_all(&frame)
183            .map_err(|error| io("write snapshot", error))?;
184        match self.options.durability {
185            Durability::Flush => file.flush().map_err(|error| io("flush snapshot", error))?,
186            Durability::SyncData => file
187                .sync_data()
188                .map_err(|error| io("sync snapshot", error))?,
189            Durability::SyncAll => file
190                .sync_all()
191                .map_err(|error| io("sync snapshot data and metadata", error))?,
192        }
193        drop(file);
194        fs::rename(&temporary, &final_path).map_err(|error| io("commit snapshot", error))?;
195        guard.committed = true;
196        Ok(())
197    }
198
199    fn load_latest(&self) -> Result<Option<ProjectionSnapshot<P>>> {
200        self.latest_path()?
201            .map(|(position, path)| self.read_path(&path, position))
202            .transpose()
203    }
204}
205
206fn temporary_path(final_path: &Path) -> PathBuf {
207    static NEXT: AtomicU64 = AtomicU64::new(0);
208    let id = NEXT.fetch_add(1, Ordering::Relaxed);
209    let name = final_path
210        .file_name()
211        .and_then(|name| name.to_str())
212        .unwrap_or("snapshot");
213    final_path.with_file_name(format!(".{name}.tmp-{}-{id}", std::process::id()))
214}
215
216struct TempGuard {
217    path: PathBuf,
218    committed: bool,
219}
220
221impl TempGuard {
222    fn new(path: PathBuf) -> Self {
223        Self {
224            path,
225            committed: false,
226        }
227    }
228}
229
230impl Drop for TempGuard {
231    fn drop(&mut self) {
232        if !self.committed {
233            let _ = fs::remove_file(&self.path);
234        }
235    }
236}
237
238fn corrupt(reason: &str) -> MemoryError {
239    MemoryError::CorruptLog {
240        offset: 0,
241        reason: reason.to_owned(),
242    }
243}
244
245#[allow(clippy::needless_pass_by_value)]
246fn io(operation: &'static str, error: std::io::Error) -> MemoryError {
247    MemoryError::Io {
248        operation,
249        message: error.to_string(),
250    }
251}
weavatrix_memory/snapshot/file.rs

weavatrix_memory/snapshot/
file.rs
