Skip to main content

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}