planarian_txn/
store.rs

1//! Durable storage under the journal (§12c): a preallocated log, an atomically
2//! replaced checkpoint, a fixed-size `RecoveryFailed` sentinel, and the per-
3//! transaction fence epochs that backends check before mutating.
4//!
5//! `MemStore` is the in-memory store for the fake harness. Cloning it gives
6//! another "process" a handle on the same storage; it can tear writes on
7//! demand. `FileStore` is the on-disk layout under `/var/lib/planarian/journal/`.
8
9use planarian_schema::TxnId;
10use std::collections::BTreeMap;
11use std::io;
12use std::sync::{Arc, Condvar, Mutex, MutexGuard};
13
14pub const DEFAULT_CAPACITY: u64 = 64 * 1024 * 1024;
15pub const SENTINEL_SIZE: usize = 4096;
16
17pub trait Store {
18    /// Held while a journal handle reads-then-appends; serializes all handles
19    /// (across processes for `FileStore`). Released on drop.
20    type Guard;
21    fn lock(&self) -> io::Result<Self::Guard>;
22    fn capacity(&self) -> u64;
23    /// Up to `len` bytes at `off`; shorter at the end of the preallocation.
24    fn read_at(&self, off: u64, len: usize) -> io::Result<Vec<u8>>;
25    fn write_at(&self, off: u64, bytes: &[u8]) -> io::Result<()>;
26    fn sync(&self) -> io::Result<()>;
27    fn read_checkpoint(&self) -> io::Result<Option<Vec<u8>>>;
28    /// Atomic: either the old checkpoint or the complete new one survives.
29    fn write_checkpoint(&self, bytes: &[u8]) -> io::Result<()>;
30    fn read_sentinel(&self) -> io::Result<Option<Vec<u8>>>;
31    fn write_sentinel(&self, bytes: &[u8]) -> io::Result<()>;
32    /// The epoch backends fence on. Written before the journal's Takeover record.
33    fn fence(&self, txn: TxnId) -> io::Result<Option<u64>>;
34    fn set_fence(&self, txn: TxnId, epoch: u64) -> io::Result<()>;
35    /// A new transaction's fence, created exclusively (§9 step 4, C5): fails
36    /// with `AlreadyExists` if anything is already there for this id.
37    fn create_fence(&self, txn: TxnId, epoch: u64) -> io::Result<()>;
38    /// Removes the fence of every transaction `keep` rejects; returns them.
39    /// A transaction without a fence is never admitted, so this only narrows.
40    fn forget_fences<F: Fn(TxnId) -> bool>(&self, keep: F) -> io::Result<Vec<TxnId>>;
41}
42
43// ---------- In-memory ----------
44
45#[derive(Debug, Default)]
46struct MemInner {
47    log: Vec<u8>,
48    checkpoint: Option<Vec<u8>>,
49    sentinel: Option<Vec<u8>>,
50    fences: BTreeMap<TxnId, u64>,
51    /// Fault injection: the next log write stores only this many bytes, then fails.
52    tear_next_write: Option<usize>,
53    /// Fault injection: the next log write whose bytes contain this stores
54    /// half of them, then fails; and whether that has happened.
55    tear_write_containing: Option<Vec<u8>>,
56    torn: bool,
57    /// Fault injection: the next checkpoint write fails without replacing the old one.
58    fail_next_checkpoint: bool,
59    /// Fault injection: log writes fail with ENOSPC.
60    enospc: bool,
61    /// P21 (§12 "fsync'd", §17a): log bytes written since the last `sync`,
62    /// as (offset, the bytes there before), oldest first. `sync` makes the
63    /// writes durable by forgetting them; `crash` undoes them. The
64    /// checkpoint, sentinel and fences are replaced atomically and durably
65    /// on write (`Store`'s contract), so they are not buffered.
66    unsynced: Vec<(usize, Vec<u8>)>,
67}
68
69#[derive(Clone, Debug)]
70pub struct MemStore {
71    inner: Arc<Mutex<MemInner>>,
72    lock: Arc<(Mutex<bool>, Condvar)>,
73    capacity: u64,
74}
75
76/// Owned guard for `MemStore::lock`.
77#[derive(Debug)]
78pub struct MemGuard(Arc<(Mutex<bool>, Condvar)>);
79
80impl Drop for MemGuard {
81    fn drop(&mut self) {
82        let (m, cv) = &*self.0;
83        *m.lock().unwrap_or_else(|e| e.into_inner()) = false;
84        cv.notify_one();
85    }
86}
87
88impl MemStore {
89    pub fn new(capacity: u64) -> MemStore {
90        MemStore {
91            inner: Arc::new(Mutex::new(MemInner { log: vec![0; capacity as usize], ..Default::default() })),
92            lock: Arc::new((Mutex::new(false), Condvar::new())),
93            capacity,
94        }
95    }
96
97    fn inner(&self) -> MutexGuard<'_, MemInner> {
98        self.inner.lock().unwrap_or_else(|e| e.into_inner())
99    }
100
101    /// A power loss: the log keeps exactly what the last `sync` covered; a
102    /// write after it is gone, as if never made.
103    pub fn crash(&self) {
104        let mut inner = self.inner();
105        while let Some((off, before)) = inner.unsynced.pop() {
106            inner.log[off..off + before.len()].copy_from_slice(&before);
107        }
108    }
109
110    pub fn tear_next_write(&self, keep: usize) {
111        self.inner().tear_next_write = Some(keep);
112    }
113
114    /// Tear the next log write that contains `needle` (e.g. `"Checking"`, to
115    /// tear the record of that state).
116    pub fn tear_write_containing(&self, needle: &str) {
117        self.inner().tear_write_containing = Some(needle.as_bytes().to_vec());
118    }
119
120    /// Whether an injected tear has happened.
121    pub fn torn(&self) -> bool {
122        self.inner().torn
123    }
124
125    pub fn fail_next_checkpoint(&self) {
126        self.inner().fail_next_checkpoint = true;
127    }
128
129    pub fn set_enospc(&self, on: bool) {
130        self.inner().enospc = on;
131    }
132}
133
134impl Store for MemStore {
135    type Guard = MemGuard;
136
137    fn lock(&self) -> io::Result<MemGuard> {
138        let (m, cv) = &*self.lock;
139        let mut held = m.lock().unwrap_or_else(|e| e.into_inner());
140        while *held {
141            held = cv.wait(held).unwrap_or_else(|e| e.into_inner());
142        }
143        *held = true;
144        Ok(MemGuard(self.lock.clone()))
145    }
146
147    fn capacity(&self) -> u64 {
148        self.capacity
149    }
150
151    fn read_at(&self, off: u64, len: usize) -> io::Result<Vec<u8>> {
152        let inner = self.inner();
153        let start = (off as usize).min(inner.log.len());
154        let end = start.saturating_add(len).min(inner.log.len());
155        Ok(inner.log[start..end].to_vec())
156    }
157
158    fn write_at(&self, off: u64, bytes: &[u8]) -> io::Result<()> {
159        let mut inner = self.inner();
160        if inner.enospc {
161            return Err(io::Error::from_raw_os_error(28));
162        }
163        let off = off as usize;
164        if off + bytes.len() > inner.log.len() {
165            return Err(io::Error::from_raw_os_error(28));
166        }
167        let matches = inner.tear_write_containing.as_ref().is_some_and(|n| bytes.windows(n.len()).any(|w| w == n));
168        if matches {
169            inner.tear_write_containing = None;
170            inner.tear_next_write = Some(bytes.len() / 2);
171        }
172        if let Some(keep) = inner.tear_next_write.take() {
173            inner.torn = true;
174            let keep = keep.min(bytes.len());
175            let before = inner.log[off..off + keep].to_vec();
176            inner.unsynced.push((off, before));
177            inner.log[off..off + keep].copy_from_slice(&bytes[..keep]);
178            return Err(io::Error::other("torn write (injected)"));
179        }
180        let before = inner.log[off..off + bytes.len()].to_vec();
181        inner.unsynced.push((off, before));
182        inner.log[off..off + bytes.len()].copy_from_slice(bytes);
183        Ok(())
184    }
185
186    fn sync(&self) -> io::Result<()> {
187        self.inner().unsynced.clear();
188        Ok(())
189    }
190
191    fn read_checkpoint(&self) -> io::Result<Option<Vec<u8>>> {
192        Ok(self.inner().checkpoint.clone())
193    }
194
195    fn write_checkpoint(&self, bytes: &[u8]) -> io::Result<()> {
196        let mut inner = self.inner();
197        if std::mem::take(&mut inner.fail_next_checkpoint) {
198            return Err(io::Error::other("interrupted checkpoint write (injected)"));
199        }
200        inner.checkpoint = Some(bytes.to_vec());
201        Ok(())
202    }
203
204    fn read_sentinel(&self) -> io::Result<Option<Vec<u8>>> {
205        Ok(self.inner().sentinel.clone())
206    }
207
208    fn write_sentinel(&self, bytes: &[u8]) -> io::Result<()> {
209        let mut s = vec![0; SENTINEL_SIZE];
210        let n = bytes.len().min(SENTINEL_SIZE);
211        s[..n].copy_from_slice(&bytes[..n]);
212        self.inner().sentinel = Some(s);
213        Ok(())
214    }
215
216    fn fence(&self, txn: TxnId) -> io::Result<Option<u64>> {
217        Ok(self.inner().fences.get(&txn).copied())
218    }
219
220    fn set_fence(&self, txn: TxnId, epoch: u64) -> io::Result<()> {
221        self.inner().fences.insert(txn, epoch);
222        Ok(())
223    }
224
225    fn create_fence(&self, txn: TxnId, epoch: u64) -> io::Result<()> {
226        let mut i = self.inner();
227        if i.fences.contains_key(&txn) {
228            return Err(io::Error::new(io::ErrorKind::AlreadyExists, format!("txn/{} exists", txn.0)));
229        }
230        i.fences.insert(txn, epoch);
231        Ok(())
232    }
233
234    fn forget_fences<F: Fn(TxnId) -> bool>(&self, keep: F) -> io::Result<Vec<TxnId>> {
235        let mut i = self.inner();
236        let gone: Vec<TxnId> = i.fences.keys().copied().filter(|t| !keep(*t)).collect();
237        for t in &gone {
238            i.fences.remove(t);
239        }
240        Ok(gone)
241    }
242}
243
244// ---------- On disk ----------
245
246/// `dir/journal` (preallocated), `dir/checkpoint`, `dir/sentinel` (4 KB,
247/// reserved at install), `dir/lock`, and fences at `dir/../txn/<txn>/lease`.
248///
249/// Filesystem-specific setup (`chattr +C` on btrfs, a dataset reservation on
250/// ZFS) is the installer's job (§12c). This type only `fallocate`s.
251#[derive(Debug)]
252pub struct FileStore {
253    dir: std::path::PathBuf,
254    txn_dir: std::path::PathBuf,
255    journal: std::fs::File,
256    capacity: u64,
257}
258
259impl FileStore {
260    pub fn open(dir: &std::path::Path, txn_dir: &std::path::Path, capacity: u64) -> io::Result<FileStore> {
261        use std::fs::{File, OpenOptions};
262        std::fs::create_dir_all(dir)?;
263        std::fs::create_dir_all(txn_dir)?;
264        let path = dir.join("journal");
265        let fresh = !path.exists();
266        let journal = OpenOptions::new().read(true).write(true).create(true).truncate(false).open(&path)?;
267        if fresh || journal.metadata()?.len() < capacity {
268            rustix::fs::fallocate(&journal, rustix::fs::FallocateFlags::empty(), 0, capacity)?;
269            journal.sync_all()?;
270            File::open(dir)?.sync_all()?;
271        }
272        let sentinel = dir.join("sentinel");
273        if !sentinel.exists() {
274            let f = File::create(&sentinel)?;
275            rustix::fs::fallocate(&f, rustix::fs::FallocateFlags::empty(), 0, SENTINEL_SIZE as u64)?;
276            f.sync_all()?;
277        }
278        Ok(FileStore { dir: dir.to_owned(), txn_dir: txn_dir.to_owned(), journal, capacity })
279    }
280
281    /// §12 step 4: a transaction's fence directory is durable before the
282    /// journal records the transaction as Prepared. `replace_atomically`
283    /// fsyncs the directory the lease is in; the entry naming that
284    /// directory lives in `txn_dir`, so `txn_dir` is fsynced once it is
285    /// created. Without it a power loss could keep Prepared and lose the
286    /// fence (fails safe, as an unfenced transaction, but not §12's intent).
287    fn sync_txn_dir(&self) -> io::Result<()> {
288        std::fs::File::open(&self.txn_dir)?.sync_all()
289    }
290
291    fn replace_atomically(&self, target: &std::path::Path, bytes: &[u8]) -> io::Result<()> {
292        use std::io::Write;
293        let parent = target.parent().expect("target has a parent");
294        let tmp =
295            parent.join(format!(".{}.tmp", target.file_name().expect("target has a file name").to_string_lossy()));
296        let mut f = std::fs::File::create(&tmp)?;
297        f.write_all(bytes)?;
298        f.sync_all()?;
299        std::fs::rename(&tmp, target)?;
300        std::fs::File::open(parent)?.sync_all()
301    }
302}
303
304impl Store for FileStore {
305    /// The open lock file; closing it releases the flock.
306    type Guard = std::fs::File;
307
308    fn lock(&self) -> io::Result<std::fs::File> {
309        let lock = std::fs::OpenOptions::new().create(true).truncate(false).write(true).open(self.dir.join("lock"))?;
310        rustix::fs::flock(&lock, rustix::fs::FlockOperation::LockExclusive)?;
311        Ok(lock)
312    }
313
314    fn capacity(&self) -> u64 {
315        self.capacity
316    }
317
318    fn read_at(&self, off: u64, len: usize) -> io::Result<Vec<u8>> {
319        use std::os::unix::fs::FileExt;
320        let end = off.saturating_add(len as u64).min(self.capacity);
321        let mut buf = vec![0; end.saturating_sub(off) as usize];
322        self.journal.read_exact_at(&mut buf, off)?;
323        Ok(buf)
324    }
325
326    fn write_at(&self, off: u64, bytes: &[u8]) -> io::Result<()> {
327        use std::os::unix::fs::FileExt;
328        if off + bytes.len() as u64 > self.capacity {
329            return Err(io::Error::from_raw_os_error(28));
330        }
331        self.journal.write_all_at(bytes, off)
332    }
333
334    fn sync(&self) -> io::Result<()> {
335        self.journal.sync_data()
336    }
337
338    fn read_checkpoint(&self) -> io::Result<Option<Vec<u8>>> {
339        match std::fs::read(self.dir.join("checkpoint")) {
340            Ok(b) => Ok(Some(b)),
341            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
342            Err(e) => Err(e),
343        }
344    }
345
346    fn write_checkpoint(&self, bytes: &[u8]) -> io::Result<()> {
347        self.replace_atomically(&self.dir.join("checkpoint"), bytes)
348    }
349
350    fn read_sentinel(&self) -> io::Result<Option<Vec<u8>>> {
351        let b = std::fs::read(self.dir.join("sentinel"))?;
352        Ok(if b.iter().all(|x| *x == 0) { None } else { Some(b) })
353    }
354
355    fn write_sentinel(&self, bytes: &[u8]) -> io::Result<()> {
356        use std::os::unix::fs::FileExt;
357        // In place: the file was preallocated at install, so this cannot need space.
358        let f = std::fs::OpenOptions::new().write(true).open(self.dir.join("sentinel"))?;
359        let n = bytes.len().min(SENTINEL_SIZE);
360        f.write_all_at(&bytes[..n], 0)?;
361        f.sync_data()
362    }
363
364    fn fence(&self, txn: TxnId) -> io::Result<Option<u64>> {
365        match std::fs::read_to_string(self.txn_dir.join(txn.0.to_string()).join("lease")) {
366            Ok(s) => Ok(s.trim().parse().ok()),
367            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
368            Err(e) => Err(e),
369        }
370    }
371
372    fn set_fence(&self, txn: TxnId, epoch: u64) -> io::Result<()> {
373        let d = self.txn_dir.join(txn.0.to_string());
374        if !d.is_dir() {
375            std::fs::create_dir_all(&d)?;
376            self.sync_txn_dir()?;
377        }
378        self.replace_atomically(&d.join("lease"), format!("{epoch}\n").as_bytes())
379    }
380
381    fn create_fence(&self, txn: TxnId, epoch: u64) -> io::Result<()> {
382        std::fs::create_dir_all(&self.txn_dir)?;
383        let d = self.txn_dir.join(txn.0.to_string());
384        std::fs::create_dir(&d)?;
385        self.sync_txn_dir()?;
386        self.replace_atomically(&d.join("lease"), format!("{epoch}\n").as_bytes())
387    }
388
389    fn forget_fences<F: Fn(TxnId) -> bool>(&self, keep: F) -> io::Result<Vec<TxnId>> {
390        let entries = match std::fs::read_dir(&self.txn_dir) {
391            Ok(e) => e,
392            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(vec![]),
393            Err(e) => return Err(e),
394        };
395        let mut gone = vec![];
396        for e in entries {
397            let e = e?;
398            // Only the directories this store names; anything else is left.
399            let Some(id) = e.file_name().to_str().and_then(|n| n.parse::<u64>().ok()).map(TxnId) else {
400                continue;
401            };
402            if !keep(id) {
403                std::fs::remove_dir_all(e.path())?;
404                gone.push(id);
405            }
406        }
407        if !gone.is_empty() {
408            std::fs::File::open(&self.txn_dir)?.sync_all()?;
409        }
410        gone.sort();
411        Ok(gone)
412    }
413}
planarian_txn/store.rs

planarian_txn/
store.rs
