pamoja_sync/
file.rs

1//! A crash-safe on-disk store-and-forward queue.
2
3use std::collections::VecDeque;
4use std::fs;
5use std::io::Write;
6use std::path::{Path, PathBuf};
7
8use pamoja_core::{Error, Result, Store};
9
10/// The filename extension for a stored record.
11const RECORD_EXTENSION: &str = "rec";
12
13/// The extension of a record still being written.
14const PARTIAL_EXTENSION: &str = "rec.tmp";
15
16/// A durable first-in first-out queue backed by one file per record.
17///
18/// Each [`append`](Store::append) writes a record to its own sequence-numbered
19/// file, flushes it to disk, and atomically renames it into place, so a power
20/// loss mid-write leaves the queue consistent: a partially written record is
21/// never visible, and [`open`](FileStore::open) clears what one left behind. On
22/// Linux and other Unix systems the directory is synced after the rename as well,
23/// so an append that returned is on disk. On Windows the directory entry is left
24/// to the file system, so an append that returned just before a power cut can be
25/// missing afterwards, though never half written. [`pop`](Store::pop) reads and
26/// deletes the oldest record.
27///
28/// Delivery is at-least-once: if the process stops between reading a record and
29/// deleting it, the next [`open`](FileStore::open) returns that record again, so
30/// consumers must tolerate the occasional redelivery.
31///
32/// # Examples
33///
34/// ```no_run
35/// use pamoja_core::Store;
36/// use pamoja_sync::FileStore;
37///
38/// # async fn run() -> pamoja_core::Result<()> {
39/// let mut store = FileStore::open("/var/lib/pamoja/outbox")?;
40/// store.append(b"reading").await?;
41/// if let Some(record) = store.pop().await? {
42///     // forward `record` over a transport, then it is gone from the queue
43///     let _ = record;
44/// }
45/// # Ok(())
46/// # }
47/// ```
48pub struct FileStore {
49    dir: PathBuf,
50    pending: VecDeque<u64>,
51    next: u64,
52    capacity: Option<usize>,
53}
54
55impl FileStore {
56    /// Opens a store rooted at `dir`, creating the directory if needed.
57    ///
58    /// Records left in the directory by a previous run are adopted in sequence
59    /// order, so the queue resumes where it left off, and a record a power cut
60    /// interrupted mid-write is deleted.
61    ///
62    /// # Arguments
63    ///
64    /// * `dir` - the directory that holds the queue's record files.
65    ///
66    /// # Returns
67    ///
68    /// A store with no limit on how many records it holds, ready to append and
69    /// drain records.
70    ///
71    /// # Errors
72    ///
73    /// Returns [`Error::Io`](pamoja_core::Error::Io) if the directory cannot be
74    /// created or scanned, or an interrupted record cannot be deleted.
75    pub fn open(dir: impl AsRef<Path>) -> Result<Self> {
76        let dir = dir.as_ref().to_path_buf();
77        fs::create_dir_all(&dir).map_err(io)?;
78
79        let mut sequences = Vec::new();
80        for entry in fs::read_dir(&dir).map_err(io)? {
81            let path = entry.map_err(io)?.path();
82            let name = path.file_name().and_then(|name| name.to_str());
83            if name.is_some_and(|name| name.ends_with(PARTIAL_EXTENSION)) {
84                fs::remove_file(&path).map_err(io)?;
85                continue;
86            }
87            if path.extension().and_then(|ext| ext.to_str()) != Some(RECORD_EXTENSION) {
88                continue;
89            }
90            if let Some(sequence) = path
91                .file_stem()
92                .and_then(|stem| stem.to_str())
93                .and_then(|stem| stem.parse::<u64>().ok())
94            {
95                sequences.push(sequence);
96            }
97        }
98        sequences.sort_unstable();
99        let next = sequences.last().map_or(0, |last| last + 1);
100
101        Ok(Self {
102            dir,
103            pending: sequences.into(),
104            next,
105            capacity: None,
106        })
107    }
108
109    /// Opens a store rooted at `dir` that holds at most `capacity` records.
110    ///
111    /// An append that would take the store past `capacity` is refused rather
112    /// than written, which on a device writing to an SD card is what keeps a long
113    /// outage from filling the card. A store reopened with more records than
114    /// `capacity` keeps them all and refuses appends until it is drained below it.
115    ///
116    /// # Arguments
117    ///
118    /// * `dir` - the directory that holds the queue's record files.
119    /// * `capacity` - the most records the store holds.
120    ///
121    /// # Returns
122    ///
123    /// A bounded store, ready to append and drain records.
124    ///
125    /// # Errors
126    ///
127    /// As for [`open`](FileStore::open).
128    pub fn open_with_capacity(dir: impl AsRef<Path>, capacity: usize) -> Result<Self> {
129        let mut store = Self::open(dir)?;
130        store.capacity = Some(capacity);
131        Ok(store)
132    }
133
134    /// Returns the path of the record file for a sequence number.
135    fn record_path(&self, sequence: u64) -> PathBuf {
136        self.dir.join(format!("{sequence:020}.{RECORD_EXTENSION}"))
137    }
138
139    /// Makes the directory's latest rename durable, where the platform allows it.
140    fn sync_dir(&self) -> Result<()> {
141        #[cfg(unix)]
142        fs::File::open(&self.dir)
143            .and_then(|dir| dir.sync_all())
144            .map_err(io)?;
145        Ok(())
146    }
147}
148
149impl Store for FileStore {
150    async fn append(&mut self, record: &[u8]) -> Result<()> {
151        if self
152            .capacity
153            .is_some_and(|capacity| self.pending.len() >= capacity)
154        {
155            return Err(Error::Io("store is at capacity".to_owned()));
156        }
157        let sequence = self.next;
158        let final_path = self.record_path(sequence);
159        let temp_path = final_path.with_extension(PARTIAL_EXTENSION);
160
161        let mut file = fs::File::create(&temp_path).map_err(io)?;
162        file.write_all(record).map_err(io)?;
163        file.sync_all().map_err(io)?;
164        drop(file);
165        fs::rename(&temp_path, &final_path).map_err(io)?;
166        self.sync_dir()?;
167
168        self.next += 1;
169        self.pending.push_back(sequence);
170        Ok(())
171    }
172
173    async fn peek(&self) -> Result<Option<Vec<u8>>> {
174        let Some(&sequence) = self.pending.front() else {
175            return Ok(None);
176        };
177        let record = fs::read(self.record_path(sequence)).map_err(io)?;
178        Ok(Some(record))
179    }
180
181    async fn pop(&mut self) -> Result<Option<Vec<u8>>> {
182        let Some(sequence) = self.pending.pop_front() else {
183            return Ok(None);
184        };
185        let path = self.record_path(sequence);
186        let record = fs::read(&path).map_err(io)?;
187        fs::remove_file(&path).map_err(io)?;
188        Ok(Some(record))
189    }
190
191    async fn len(&self) -> Result<usize> {
192        Ok(self.pending.len())
193    }
194}
195
196/// Maps a filesystem error onto the shared I/O error.
197fn io(error: std::io::Error) -> Error {
198    Error::Io(error.to_string())
199}
200
201#[cfg(test)]
202mod tests {
203    use super::*;
204
205    #[tokio::test]
206    async fn drains_in_first_in_first_out_order() {
207        let dir = tempfile::tempdir().expect("tempdir");
208        let mut store = FileStore::open(dir.path()).expect("open");
209
210        store.append(b"a").await.expect("append");
211        store.append(b"b").await.expect("append");
212        assert_eq!(store.len().await.expect("len"), 2);
213
214        assert_eq!(store.pop().await.expect("pop"), Some(b"a".to_vec()));
215        assert_eq!(store.pop().await.expect("pop"), Some(b"b".to_vec()));
216        assert_eq!(store.pop().await.expect("pop"), None);
217    }
218
219    #[tokio::test]
220    async fn records_survive_reopening() {
221        let dir = tempfile::tempdir().expect("tempdir");
222        {
223            let mut store = FileStore::open(dir.path()).expect("open");
224            store.append(b"durable").await.expect("append");
225            store.append(b"records").await.expect("append");
226        }
227
228        let mut reopened = FileStore::open(dir.path()).expect("reopen");
229        assert_eq!(reopened.len().await.expect("len"), 2);
230        assert_eq!(
231            reopened.pop().await.expect("pop"),
232            Some(b"durable".to_vec())
233        );
234        assert_eq!(
235            reopened.pop().await.expect("pop"),
236            Some(b"records".to_vec())
237        );
238    }
239
240    #[tokio::test]
241    async fn peek_reads_the_oldest_without_removing_it() {
242        let dir = tempfile::tempdir().expect("tempdir");
243        let mut store = FileStore::open(dir.path()).expect("open");
244        store.append(b"oldest").await.expect("append");
245        store.append(b"newer").await.expect("append");
246
247        assert_eq!(store.peek().await.expect("peek"), Some(b"oldest".to_vec()));
248        assert_eq!(store.len().await.expect("len"), 2);
249        assert_eq!(store.pop().await.expect("pop"), Some(b"oldest".to_vec()));
250    }
251
252    #[tokio::test]
253    async fn popping_removes_the_record_file() {
254        let dir = tempfile::tempdir().expect("tempdir");
255        let mut store = FileStore::open(dir.path()).expect("open");
256        store.append(b"once").await.expect("append");
257        let _ = store.pop().await.expect("pop");
258
259        let reopened = FileStore::open(dir.path()).expect("reopen");
260        assert!(reopened.is_empty().await.expect("is_empty"));
261    }
262
263    #[tokio::test]
264    async fn a_record_interrupted_mid_write_is_never_seen_and_is_cleared() {
265        let dir = tempfile::tempdir().expect("tempdir");
266        {
267            let mut store = FileStore::open(dir.path()).expect("open");
268            store.append(b"whole").await.expect("append");
269        }
270        let partial = dir.path().join(format!("{:020}.{PARTIAL_EXTENSION}", 1));
271        fs::write(&partial, b"hal").expect("a half-written record");
272
273        let mut reopened = FileStore::open(dir.path()).expect("reopen");
274        assert_eq!(reopened.len().await.expect("len"), 1);
275        assert!(!partial.exists(), "the interrupted write was cleared");
276        assert_eq!(reopened.pop().await.expect("pop"), Some(b"whole".to_vec()));
277        reopened.append(b"next").await.expect("append");
278        assert_eq!(reopened.pop().await.expect("pop"), Some(b"next".to_vec()));
279    }
280
281    #[tokio::test]
282    async fn a_bounded_store_refuses_the_append_that_would_overflow_it() {
283        let dir = tempfile::tempdir().expect("tempdir");
284        let mut store = FileStore::open_with_capacity(dir.path(), 2).expect("open");
285        store.append(b"1").await.expect("room");
286        store.append(b"2").await.expect("room");
287
288        match store.append(b"3").await {
289            Err(Error::Io(reason)) => assert_eq!(reason, "store is at capacity"),
290            other => panic!("a full store took a record: {other:?}"),
291        }
292        assert_eq!(store.len().await.expect("len"), 2);
293
294        let mut smaller = FileStore::open_with_capacity(dir.path(), 1).expect("reopen");
295        assert_eq!(
296            smaller.len().await.expect("len"),
297            2,
298            "a smaller bound keeps what is there"
299        );
300        assert!(smaller.append(b"3").await.is_err());
301        smaller.pop().await.expect("pop");
302        smaller.pop().await.expect("pop");
303        smaller.append(b"3").await.expect("room once drained");
304    }
305}
pamoja_sync/file.rs

pamoja_sync/
file.rs
