plugmem_host/storage.rs
1//! `FileStorage`: the engine's `Storage` trait over a **versioned** on-disk
2//! layout — immutable snapshot generations named by a tiny manifest (
3//! ). This is what lets a reader map a stable snapshot while a writer
4//! keeps working: the writer never overwrites a live file, it publishes a new
5//! generation and repoints the manifest.
6//!
7//! Layout for `base = "agent.plugmem"`:
8//!
9//! | file | role |
10//! |---|---|
11//! | `agent.plugmem` | the **manifest** — a tiny record naming the current snapshot generation |
12//! | `agent.plugmem.snap.<N>` | **generation N** — an immutable full snapshot image; never rewritten |
13//! | `agent.plugmem.journal` | the append-only journal since the current generation |
14//! | `agent.plugmem.lock` | the advisory-lock file (writer-vs-writer) |
15//! | `agent.plugmem.snap.<N>.tmp`, `agent.plugmem.manifest.tmp` | staging for the atomic writes |
16//!
17//! A checkpoint streams the fresh image to `…snap.<N+1>.tmp`, fsyncs it,
18//! renames it to `…snap.<N+1>` (an immutable file, never overwritten), then
19//! atomically repoints the manifest (tmp + fsync + rename + directory fsync).
20//! The old generation is reclaimed once nothing maps it. A reader always
21//! observes a manifest pointing at a generation that already exists on disk.
22//! The lock is held from `open` until drop; the OS releases it even on
23//! abnormal termination.
24
25use std::fs::{File, OpenOptions};
26use std::io::{BufWriter, Seek, SeekFrom, Write as _};
27use std::path::{Path, PathBuf};
28
29use memmap2::Mmap;
30use plugmem_core::snapshot::SnapshotSink;
31use plugmem_core::{Error, Scratch, Storage};
32
33use crate::error::HostError;
34
35/// Maps a filesystem error into the engine's storage-error variant so it can
36/// cross the [`SnapshotSink`] boundary (which speaks [`plugmem_core::Error`]).
37fn sink_io(e: std::io::Error) -> Error {
38 Error::Storage(format!("{e}"))
39}
40
41/// When journal appends reach the disk.
42#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
43#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
44pub enum FsyncPolicy {
45 /// Fsync after every appended journal record — every acknowledged
46 /// mutation survives a power cut. The default: durability is worth
47 /// microseconds at this write volume.
48 #[default]
49 EachOp,
50 /// Fsync only at snapshot boundaries. Faster; an OS crash may lose
51 /// the journal tail written since the last snapshot.
52 OnSnapshot,
53}
54
55/// File-backed [`Storage`] holding an advisory lock on its database.
56#[derive(Debug)]
57pub struct FileStorage {
58 /// The manifest path (what the caller points at).
59 base: PathBuf,
60 journal_path: PathBuf,
61 /// Staging path for the atomic manifest publish.
62 manifest_tmp: PathBuf,
63 /// The generation the manifest currently names; `0` = no snapshot yet.
64 current_gen: u64,
65 /// Keeps the advisory lock alive; the handle itself is never read.
66 _lock: File,
67 /// The journal in append mode, kept open across appends.
68 journal: File,
69 fsync: FsyncPolicy,
70 /// While `true`, `append_journal` skips its per-record fsync so a bulk write
71 /// amortizes durability into one [`sync_journal`](Self::sync_journal) at the
72 /// end (see [`set_batch`](Self::set_batch)). Only meaningful under `EachOp`.
73 batch: bool,
74}
75
76impl FileStorage {
77 /// Opens (creating as needed) the database at `base` and takes the
78 /// **exclusive writer lock** — one writer at a time. Readers do *not* take
79 /// this lock (they pin a generation file via
80 /// [`pin_current_generation`] instead), so a writer and any number of
81 /// readers coexist across threads or processes (the versioned
82 /// MVCC layout).
83 ///
84 /// # Errors
85 ///
86 /// [`HostError::Locked`] when another **writer** owns the lock;
87 /// [`HostError::Io`] for filesystem failures.
88 pub fn open(base: impl Into<PathBuf>, fsync: FsyncPolicy) -> Result<Self, HostError> {
89 let base = base.into();
90 let lock_path = suffixed(&base, "lock");
91 let journal_path = suffixed(&base, "journal");
92 let manifest_tmp = suffixed(&base, "manifest.tmp");
93
94 let lock = OpenOptions::new()
95 .create(true)
96 .write(true)
97 .truncate(false)
98 .open(&lock_path)
99 .map_err(|e| HostError::io(&lock_path, e))?;
100 match lock.try_lock() {
101 Ok(()) => {}
102 Err(std::fs::TryLockError::WouldBlock) => {
103 return Err(HostError::Locked { path: base });
104 }
105 Err(std::fs::TryLockError::Error(e)) => {
106 return Err(HostError::io(&lock_path, e));
107 }
108 }
109
110 let current_gen = read_manifest(&base)?.unwrap_or(0);
111
112 // Crash recovery + GC: drop unpublished orphan generations and staging
113 // tmps, and reclaim any unpinned superseded generation. Safe with live
114 // readers — reclaim is pin-aware (a reader's shared lock keeps its
115 // generation). Only the writer sweeps.
116 sweep_generations(&base, current_gen)?;
117
118 let journal = OpenOptions::new()
119 .create(true)
120 .append(true)
121 .open(&journal_path)
122 .map_err(|e| HostError::io(&journal_path, e))?;
123
124 Ok(Self {
125 base,
126 journal_path,
127 manifest_tmp,
128 current_gen,
129 _lock: lock,
130 journal,
131 fsync,
132 batch: false,
133 })
134 }
135
136 /// Enters (`true`) or leaves (`false`) **batch mode**. While on,
137 /// [`append_journal`](plugmem_core::Storage::append_journal) writes each
138 /// record **without** its per-record fsync, so a bulk write (`remember_many`)
139 /// amortizes durability into a single [`sync_journal`](Self::sync_journal) at
140 /// the end. Only affects the `EachOp` policy — `OnSnapshot` never fsyncs per
141 /// record anyway. The caller must pair `set_batch(true)` with a final
142 /// `set_batch(false)` + `sync_journal`, even on error, so a later single
143 /// write is durable again.
144 pub(crate) fn set_batch(&mut self, on: bool) {
145 self.batch = on;
146 }
147
148 /// Fsyncs the journal now — the durability point for records appended in
149 /// batch mode. Idempotent (a plain `sync_data`), so calling it after a
150 /// partially-written batch makes exactly the records that reached the file
151 /// durable.
152 pub(crate) fn sync_journal(&mut self) -> Result<(), HostError> {
153 self.journal
154 .sync_data()
155 .map_err(|e| HostError::io(&self.journal_path, e))
156 }
157
158 /// The database base path.
159 pub fn path(&self) -> &Path {
160 &self.base
161 }
162
163 /// Current journal size in bytes (drives the snapshot policy).
164 pub fn journal_bytes(&self) -> u64 {
165 self.journal.metadata().map(|m| m.len()).unwrap_or(0)
166 }
167
168 /// The path of the snapshot file the manifest currently names, or `None`
169 /// for a fresh database with no published snapshot. Callers that map the
170 /// snapshot (`open_engine`, `ReadOnlyDatabase`, `Scrub`, `recover`) resolve
171 /// through this instead of mapping `base` (which is now the manifest).
172 pub(crate) fn current_snapshot_path(&self) -> Result<Option<PathBuf>, HostError> {
173 Ok(read_manifest(&self.base)?.map(|g| gen_path(&self.base, g)))
174 }
175
176 /// The next generation number this storage will publish.
177 fn next_gen(&self) -> u64 {
178 self.current_gen + 1
179 }
180
181 /// Streams a snapshot into the next generation's tmp file and fsyncs it,
182 /// **without** publishing — the durable-but-not-yet-visible half of a
183 /// checkpoint. `write` drives the engine's streaming snapshot
184 /// writer against a buffered file sink, so the image never all lives in RAM
185 /// at once. Split from [`FileStorage::commit_snapshot`] because the caller
186 /// must drop the mmap of the *old* generation **between** the two: staging
187 /// reads through that map, and the reclaim in `commit` deletes it. A staged
188 /// tmp is cleaned up on the next exclusive open.
189 pub(crate) fn stage_snapshot(
190 &mut self,
191 write: impl FnOnce(&mut FileSink) -> Result<(), HostError>,
192 ) -> Result<(), HostError> {
193 let tmp = gen_tmp_path(&self.base, self.next_gen());
194 let file = File::create(&tmp).map_err(|e| HostError::io(&tmp, e))?;
195 let mut sink = FileSink::new(file, tmp.clone());
196 write(&mut sink)?;
197 let file = sink.finish()?;
198 file.sync_all().map_err(|e| HostError::io(&tmp, e))?;
199 Ok(())
200 }
201
202 /// Publishes the staged generation: rename its tmp to the immutable
203 /// `snap.<N+1>`, repoint the manifest, then GC superseded generations
204 /// (pin-aware). Call only after [`FileStorage::stage_snapshot`] and after
205 /// dropping any mmap of the old generation.
206 pub(crate) fn commit_snapshot(&mut self) -> Result<(), HostError> {
207 let next = self.next_gen();
208 let tmp = gen_tmp_path(&self.base, next);
209 let genp = gen_path(&self.base, next);
210 std::fs::rename(&tmp, &genp).map_err(|e| HostError::io(&genp, e))?;
211 sync_dir(&self.base)?;
212 publish_manifest(&self.base, &self.manifest_tmp, next)?;
213 self.current_gen = next;
214 // Reclaim every unpinned superseded generation (a reader on an old one
215 // keeps it until it drops). Best-effort — leftovers go on the next pass.
216 let _ = sweep_generations(&self.base, self.current_gen);
217 Ok(())
218 }
219}
220
221/// A streaming [`SnapshotSink`] over a buffered file: sequential section
222/// writes are buffered, and the single `patch` (the header file-hash, once
223/// the running hash is known) flushes and seeks. Lets a snapshot stream to
224/// disk without a full-image buffer.
225pub(crate) struct FileSink {
226 buf: BufWriter<File>,
227 path: PathBuf,
228}
229
230impl FileSink {
231 fn new(file: File, path: PathBuf) -> Self {
232 Self {
233 buf: BufWriter::new(file),
234 path,
235 }
236 }
237
238 /// Flushes the buffer and returns the underlying file for fsync.
239 fn finish(self) -> Result<File, HostError> {
240 self.buf
241 .into_inner()
242 .map_err(|e| HostError::io(&self.path, e.into_error()))
243 }
244}
245
246impl SnapshotSink for &mut FileSink {
247 fn write(&mut self, bytes: &[u8]) -> Result<(), Error> {
248 self.buf.write_all(bytes).map_err(sink_io)
249 }
250
251 fn patch(&mut self, at: u64, bytes: &[u8]) -> Result<(), Error> {
252 // The one non-sequential write: flush buffered bytes, seek to the
253 // header field, patch it, then restore the position to the end.
254 self.buf.flush().map_err(sink_io)?;
255 let file = self.buf.get_mut();
256 file.seek(SeekFrom::Start(at)).map_err(sink_io)?;
257 file.write_all(bytes).map_err(sink_io)?;
258 file.seek(SeekFrom::End(0)).map_err(sink_io)?;
259 Ok(())
260 }
261}
262
263/// `"a.plugmem"` + `"lock"` → `"a.plugmem.lock"`.
264fn suffixed(base: &Path, ext: &str) -> PathBuf {
265 let mut s = base.as_os_str().to_os_string();
266 s.push(".");
267 s.push(ext);
268 PathBuf::from(s)
269}
270
271/// Manifest magic ("PMGL" — distinct from the snapshot's own `MAGIC`).
272const MANIFEST_MAGIC: u32 = 0x504D_474C;
273/// On-disk manifest version (the layout, not the snapshot format).
274const MANIFEST_VERSION: u16 = 1;
275/// Manifest length: magic(4) + version(2) + pad(2) + gen(8) + checksum(8).
276const MANIFEST_LEN: usize = 24;
277
278/// How many times a reader retries an open across the Windows manifest-swap
279/// window. Publishing the manifest is a `rename(tmp, base)`, which on Windows
280/// briefly leaves the old `base` *delete-pending*; a reader opening it in that
281/// window gets a transient `ERROR_ACCESS_DENIED` (5) / `ERROR_SHARING_VIOLATION`
282/// (32). POSIX renames are atomic with respect to a concurrent open, so this
283/// never happens there. `100 * 1ms` dwarfs the microsecond swap while adding a
284/// negligible delay to a genuine failure.
285#[cfg(windows)]
286const SHARE_RETRIES: u32 = 100;
287
288/// True for the transient Windows sharing errors a rename-replace throws at a
289/// concurrent open. Always `false` off Windows (codes 5/32 mean unrelated things
290/// on POSIX), so the retry paths below collapse to the original single shot.
291fn is_transient_share_error(e: &std::io::Error) -> bool {
292 #[cfg(windows)]
293 {
294 matches!(e.raw_os_error(), Some(5) | Some(32))
295 }
296 #[cfg(not(windows))]
297 {
298 let _ = e;
299 false
300 }
301}
302
303/// `std::fs::read`, retried across the transient Windows rename-swap window so a
304/// reader rides over a concurrent manifest publish instead of spuriously failing
305/// with "Access is denied". On non-Windows it is a plain `std::fs::read`.
306fn read_across_rename(path: &Path) -> std::io::Result<Vec<u8>> {
307 #[cfg(windows)]
308 {
309 let mut attempts = 0u32;
310 loop {
311 match std::fs::read(path) {
312 Err(e) if is_transient_share_error(&e) && attempts < SHARE_RETRIES => {
313 attempts += 1;
314 std::thread::sleep(std::time::Duration::from_millis(1));
315 }
316 other => return other,
317 }
318 }
319 }
320 #[cfg(not(windows))]
321 {
322 std::fs::read(path)
323 }
324}
325
326/// 64-bit FNV-1a — a dependency-free integrity check for the manifest. The
327/// manifest is written atomically (tmp + rename), so it can never be torn; this
328/// only catches external garbage / bit-rot in the tiny fixed record.
329fn fnv1a(bytes: &[u8]) -> u64 {
330 let mut h: u64 = 0xcbf2_9ce4_8422_2325;
331 for &b in bytes {
332 h ^= u64::from(b);
333 h = h.wrapping_mul(0x0000_0100_0000_01b3);
334 }
335 h
336}
337
338/// The snapshot file for generation `n`: `base` + `.snap.<n>`.
339fn gen_path(base: &Path, n: u64) -> PathBuf {
340 suffixed(base, &format!("snap.{n}"))
341}
342
343/// The staging path for generation `n`: `base` + `.snap.<n>.tmp`.
344fn gen_tmp_path(base: &Path, n: u64) -> PathBuf {
345 suffixed(base, &format!("snap.{n}.tmp"))
346}
347
348/// Reads and validates the manifest at `base`. `Ok(None)` when it is absent (a
349/// fresh database); `Err(Corrupt)` when it is present but malformed; `Err(Io)`
350/// on a real filesystem failure. The returned generation is always ≥ 1.
351pub(crate) fn read_manifest(base: &Path) -> Result<Option<u64>, HostError> {
352 let bytes = match read_across_rename(base) {
353 Ok(b) => b,
354 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
355 Err(e) => return Err(HostError::io(base, e)),
356 };
357 let ok = bytes.len() == MANIFEST_LEN
358 && u32::from_le_bytes(bytes[0..4].try_into().unwrap()) == MANIFEST_MAGIC
359 && u16::from_le_bytes(bytes[4..6].try_into().unwrap()) == MANIFEST_VERSION
360 && fnv1a(&bytes[0..16]) == u64::from_le_bytes(bytes[16..24].try_into().unwrap());
361 if !ok {
362 return Err(HostError::Engine(Error::Corrupt("manifest is corrupt")));
363 }
364 Ok(Some(u64::from_le_bytes(bytes[8..16].try_into().unwrap())))
365}
366
367/// Atomically publishes `gen` as the current generation: write a fresh manifest
368/// to `manifest_tmp`, fsync, rename over `base`, fsync the directory.
369fn publish_manifest(base: &Path, manifest_tmp: &Path, generation: u64) -> Result<(), HostError> {
370 let mut buf = [0u8; MANIFEST_LEN];
371 buf[0..4].copy_from_slice(&MANIFEST_MAGIC.to_le_bytes());
372 buf[4..6].copy_from_slice(&MANIFEST_VERSION.to_le_bytes());
373 buf[8..16].copy_from_slice(&generation.to_le_bytes());
374 let sum = fnv1a(&buf[0..16]);
375 buf[16..24].copy_from_slice(&sum.to_le_bytes());
376 let mut f = File::create(manifest_tmp).map_err(|e| HostError::io(manifest_tmp, e))?;
377 f.write_all(&buf)
378 .and_then(|()| f.sync_all())
379 .map_err(|e| HostError::io(manifest_tmp, e))?;
380 drop(f);
381 std::fs::rename(manifest_tmp, base).map_err(|e| HostError::io(base, e))?;
382 sync_dir(base)
383}
384
385/// Fsyncs the directory holding the database (unix only — the rename's
386/// durability point).
387fn sync_dir(base: &Path) -> Result<(), HostError> {
388 #[cfg(unix)]
389 {
390 let dir = base.parent().filter(|p| !p.as_os_str().is_empty());
391 let dir = dir.unwrap_or_else(|| Path::new("."));
392 File::open(dir)
393 .and_then(|d| d.sync_all())
394 .map_err(|e| HostError::io(dir, e))?;
395 }
396 #[cfg(not(unix))]
397 let _ = base;
398 Ok(())
399}
400
401/// Reclaims one superseded generation file, but only if nothing pins it. A
402/// reader holds a **shared** lock on the generation file for as long as it maps
403/// it (see [`pin_current_generation`]), so a successful **exclusive** try-lock
404/// proves no reader is using it, and the delete under that lock cannot race a
405/// new pin. Best-effort: a pinned (or, on Windows, an open-mapped) generation is
406/// simply left for a later pass. Never deletes a live reader's snapshot.
407fn try_reclaim_generation(genp: &Path) {
408 if let Ok(f) = File::open(genp)
409 && f.try_lock().is_ok()
410 {
411 let _ = std::fs::remove_file(genp);
412 }
413}
414
415/// Sweeps the generation files around `current`: the current one stays; a
416/// higher number is crash debris from a checkpoint that never published its
417/// manifest (never pinned — delete it); a lower number is a superseded
418/// generation reclaimed only if unpinned (a reader may still map it). Staging
419/// `.tmp`s and the manifest tmp go unconditionally. Called by the exclusive
420/// writer — on open (crash recovery) and after each checkpoint (GC).
421fn sweep_generations(base: &Path, current: u64) -> Result<(), HostError> {
422 let dir = base
423 .parent()
424 .filter(|p| !p.as_os_str().is_empty())
425 .unwrap_or_else(|| Path::new("."));
426 let name = base
427 .file_name()
428 .and_then(|n| n.to_str())
429 .unwrap_or_default();
430 let prefix = format!("{name}.snap.");
431 let entries = match std::fs::read_dir(dir) {
432 Ok(e) => e,
433 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
434 Err(e) => return Err(HostError::io(dir, e)),
435 };
436 for entry in entries {
437 let entry = entry.map_err(|e| HostError::io(dir, e))?;
438 let fname = entry.file_name();
439 let Some(fname) = fname.to_str() else {
440 continue;
441 };
442 let Some(rest) = fname.strip_prefix(&prefix) else {
443 continue;
444 };
445 if rest.ends_with(".tmp") {
446 let _ = std::fs::remove_file(entry.path()); // staging scrap
447 continue;
448 }
449 match rest.parse::<u64>() {
450 Ok(n) if n == current => {} // the live generation
451 Ok(n) if n > current => {
452 let _ = std::fs::remove_file(entry.path()); // unpublished orphan
453 }
454 Ok(_) => try_reclaim_generation(&entry.path()), // superseded — pin-aware
455 Err(_) => {}
456 }
457 }
458 let _ = std::fs::remove_file(suffixed(base, "manifest.tmp"));
459 Ok(())
460}
461
462/// Opens and **shared-locks** the current snapshot generation, pinning it
463/// against the writer's GC for as long as the returned [`File`] is held; returns
464/// it with the generation path and the generation number it pinned. `Ok(None)`
465/// when there is no published generation (a fresh database). Readers do not take
466/// the writer lock, so a reader and the writer coexist — this is what makes the
467/// cross-process MVCC work.
468///
469/// Retries the open→lock race with the collector: if the manifest names a
470/// generation that GC reclaims in the window between resolving and locking it,
471/// the open (or the post-lock existence recheck) fails and we retry against the
472/// fresh manifest. Once the shared lock is held and the file still exists, GC's
473/// exclusive try-lock must fail, so the pin is stable. The returned number is the
474/// generation actually pinned, which a caller can compare against a stale one to
475/// tell whether the writer has published a newer snapshot (see
476/// [`ReadOnlyDatabase::refresh`](crate::ReadOnlyDatabase::refresh)).
477pub(crate) fn pin_current_generation(
478 base: &Path,
479) -> Result<Option<(File, PathBuf, u64)>, HostError> {
480 loop {
481 let Some(generation) = read_manifest(base)? else {
482 return Ok(None);
483 };
484 let genp = gen_path(base, generation);
485 let file = match File::open(&genp) {
486 Ok(f) => f,
487 // GC reclaimed it between the manifest read and the open — retry.
488 Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
489 // Windows: a gen file being GC-reclaimed can surface as a transient
490 // delete-pending share error rather than NotFound — also a retry.
491 Err(e) if is_transient_share_error(&e) => {
492 std::thread::sleep(std::time::Duration::from_millis(1));
493 continue;
494 }
495 Err(e) => return Err(HostError::io(&genp, e)),
496 };
497 match file.try_lock_shared() {
498 Ok(()) => {}
499 // A transient exclusive holder (GC probing) — retry.
500 Err(std::fs::TryLockError::WouldBlock) => continue,
501 Err(std::fs::TryLockError::Error(e)) => return Err(HostError::io(&genp, e)),
502 }
503 // Confirm the generation survived the open→lock window. If GC reclaimed
504 // it just before we locked, the path is gone; drop the lock and retry
505 // for the fresh generation. If it exists, our shared lock now blocks
506 // GC's exclusive try-lock, so the pin holds.
507 if genp.exists() {
508 return Ok(Some((file, genp, generation)));
509 }
510 }
511}
512
513impl Storage for FileStorage {
514 type Error = HostError;
515
516 fn read_snapshot(&mut self) -> Result<Option<Vec<u8>>, HostError> {
517 match self.current_snapshot_path()? {
518 Some(p) => Ok(Some(std::fs::read(&p).map_err(|e| HostError::io(&p, e))?)),
519 None => Ok(None),
520 }
521 }
522
523 fn write_snapshot(&mut self, bytes: &[u8]) -> Result<(), HostError> {
524 // Publish a new immutable generation (the non-streaming path): stage
525 // its tmp, rename to snap.<N+1>, repoint the manifest, reclaim the old.
526 let next = self.next_gen();
527 let tmp = gen_tmp_path(&self.base, next);
528 let mut f = File::create(&tmp).map_err(|e| HostError::io(&tmp, e))?;
529 f.write_all(bytes)
530 .and_then(|()| f.sync_all())
531 .map_err(|e| HostError::io(&tmp, e))?;
532 drop(f);
533 let genp = gen_path(&self.base, next);
534 std::fs::rename(&tmp, &genp).map_err(|e| HostError::io(&genp, e))?;
535 sync_dir(&self.base)?;
536 publish_manifest(&self.base, &self.manifest_tmp, next)?;
537 self.current_gen = next;
538 let _ = sweep_generations(&self.base, self.current_gen);
539 Ok(())
540 }
541
542 fn read_journal(&mut self) -> Result<Vec<u8>, HostError> {
543 std::fs::read(&self.journal_path).map_err(|e| HostError::io(&self.journal_path, e))
544 }
545
546 fn append_journal(&mut self, entry: &[u8]) -> Result<(), HostError> {
547 self.journal
548 .write_all(entry)
549 .map_err(|e| HostError::io(&self.journal_path, e))?;
550 // In batch mode the fsync is deferred to one `sync_journal` at the end
551 // of the batch (durability amortized across the whole bulk write).
552 if self.fsync == FsyncPolicy::EachOp && !self.batch {
553 self.journal
554 .sync_data()
555 .map_err(|e| HostError::io(&self.journal_path, e))?;
556 }
557 Ok(())
558 }
559
560 fn clear_journal(&mut self) -> Result<(), HostError> {
561 // Truncate through a dedicated write handle, not `set_len` on the
562 // append handle: Rust opens append handles with FILE_WRITE_DATA
563 // masked off (append can only extend, never overwrite), so on
564 // Windows `SetEndOfFile` is denied with ERROR_ACCESS_DENIED. A
565 // `write + truncate` open empties the file portably; then the
566 // append handle is re-established so later appends target the
567 // fresh, empty journal.
568 let truncated = OpenOptions::new()
569 .create(true)
570 .write(true)
571 .truncate(true)
572 .open(&self.journal_path)
573 .map_err(|e| HostError::io(&self.journal_path, e))?;
574 truncated
575 .sync_data()
576 .map_err(|e| HostError::io(&self.journal_path, e))?;
577 drop(truncated);
578 self.journal = OpenOptions::new()
579 .create(true)
580 .append(true)
581 .open(&self.journal_path)
582 .map_err(|e| HostError::io(&self.journal_path, e))?;
583 Ok(())
584 }
585}
586
587/// A host [`Scratch`] over a temp file (milestone H): sequential
588/// appends go through a buffered writer; [`freeze`](Scratch::freeze) flushes
589/// and memory-maps the file, so the staged pool is read (randomly and
590/// sequentially) straight from the map instead of RAM. Dropping it unmaps and
591/// deletes the temp file.
592pub struct FileScratch {
593 path: PathBuf,
594 /// `Some` while writing, taken by the first `freeze`.
595 writer: Option<BufWriter<File>>,
596 /// `Some` after `freeze` — the read-back mapping the borrow points into.
597 map: Option<Mmap>,
598 len: u64,
599}
600
601impl FileScratch {
602 /// Creates (truncating) a staging file at `path`, ready for appends.
603 ///
604 /// # Errors
605 ///
606 /// [`HostError::Io`] if the file cannot be created.
607 pub fn create(path: impl Into<PathBuf>) -> Result<Self, HostError> {
608 let path = path.into();
609 let file = OpenOptions::new()
610 .write(true)
611 .create(true)
612 .truncate(true)
613 .open(&path)
614 .map_err(|e| HostError::io(&path, e))?;
615 Ok(Self {
616 path,
617 writer: Some(BufWriter::new(file)),
618 map: None,
619 len: 0,
620 })
621 }
622}
623
624impl Scratch for FileScratch {
625 type Error = HostError;
626
627 fn write(&mut self, bytes: &[u8]) -> Result<(), HostError> {
628 let Self {
629 writer, path, len, ..
630 } = self;
631 let w = writer.as_mut().ok_or(HostError::Engine(Error::Invalid(
632 "scratch write after freeze",
633 )))?;
634 w.write_all(bytes).map_err(|e| HostError::io(path, e))?;
635 *len += bytes.len() as u64;
636 Ok(())
637 }
638
639 fn len(&self) -> u64 {
640 self.len
641 }
642
643 fn freeze(&mut self) -> Result<&[u8], HostError> {
644 if self.map.is_none() {
645 // Flush and fsync the staged bytes, then map the file fresh.
646 let writer = self
647 .writer
648 .take()
649 .ok_or(HostError::Engine(Error::Invalid("scratch frozen twice")))?;
650 let file = writer
651 .into_inner()
652 .map_err(|e| HostError::io(&self.path, e.into_error()))?;
653 file.sync_all().map_err(|e| HostError::io(&self.path, e))?;
654 drop(file);
655 let file = File::open(&self.path).map_err(|e| HostError::io(&self.path, e))?;
656 // SAFETY: this is our private temp file — created by `create`,
657 // owned by this `FileScratch` for its whole life, deleted on drop —
658 // so no other process writes or truncates it under the map (the
659 // same argument as the read-only snapshot map).
660 let map = unsafe { Mmap::map(&file) }.map_err(|e| HostError::io(&self.path, e))?;
661 self.map = Some(map);
662 }
663 Ok(&self.map.as_ref().expect("just set")[..])
664 }
665}
666
667impl Drop for FileScratch {
668 fn drop(&mut self) {
669 // Unmap before delete: Windows refuses to remove a mapped file (the
670 // same constraint as renaming over one).
671 self.map = None;
672 self.writer = None;
673 let _ = std::fs::remove_file(&self.path);
674 }
675}