coven_foundation/
atomic_file.rs

1//! Installing a local file's complete contents so a crash leaves either the
2//! old bytes or the new ones, never a partial write.
3//!
4//! This module is the crate's blessed implementation of the durable write, and
5//! the only place allowed to call the raw sync methods that `clippy.toml`
6//! disallows everywhere else. Keeping it to one place is what lets the
7//! platform split below — Unix flushes the parent directory, nothing else can —
8//! stay correct in a single spot instead of in every caller.
9#![allow(clippy::disallowed_methods)]
10
11use std::io::Write as _;
12use std::path::{Path, PathBuf};
13
14#[cfg(any(test, feature = "test-utils"))]
15use std::sync::atomic::{AtomicUsize, Ordering};
16
17#[derive(Clone, Debug)]
18pub(crate) enum FileSync {
19    Enabled,
20    Disabled,
21    #[cfg(any(test, feature = "test-utils"))]
22    ObservedDisabled(std::sync::Arc<AtomicUsize>),
23}
24
25impl FileSync {
26    fn requested(&self) {
27        #[cfg(any(test, feature = "test-utils"))]
28        if let Self::ObservedDisabled(requests) = self {
29            requests.fetch_add(1, Ordering::SeqCst);
30        }
31    }
32
33    pub(crate) async fn finish_async_write(
34        &self,
35        file: &mut tokio::fs::File,
36    ) -> std::io::Result<()> {
37        use tokio::io::AsyncWriteExt;
38
39        file.flush().await?;
40        self.sync_file(file).await
41    }
42
43    async fn sync_file(&self, file: &tokio::fs::File) -> std::io::Result<()> {
44        self.requested();
45        match self {
46            Self::Enabled => file.sync_all().await,
47            Self::Disabled => Ok(()),
48            #[cfg(any(test, feature = "test-utils"))]
49            Self::ObservedDisabled(_) => Ok(()),
50        }
51    }
52
53    pub(crate) fn sync_file_blocking(&self, file: &std::fs::File) -> std::io::Result<()> {
54        self.requested();
55        match self {
56            Self::Enabled => file.sync_all(),
57            Self::Disabled => Ok(()),
58            #[cfg(any(test, feature = "test-utils"))]
59            Self::ObservedDisabled(_) => Ok(()),
60        }
61    }
62
63    pub(crate) async fn sync_parent(&self, path: &Path) -> Result<(), FileError> {
64        let parent = parent_of(path)?;
65        self.requested();
66        match self {
67            Self::Enabled => flush_directory(parent)
68                .await
69                .map_err(|source| FileError::at("fsync parent directory", parent, source)),
70            Self::Disabled => Ok(()),
71            #[cfg(any(test, feature = "test-utils"))]
72            Self::ObservedDisabled(_) => Ok(()),
73        }
74    }
75
76    pub(crate) fn sync_parent_blocking(&self, path: &Path) -> Result<(), FileError> {
77        let parent = parent_of(path)?;
78        self.requested();
79        match self {
80            Self::Enabled => flush_directory_blocking(parent)
81                .map_err(|source| FileError::at("fsync parent directory", parent, source)),
82            Self::Disabled => Ok(()),
83            #[cfg(any(test, feature = "test-utils"))]
84            Self::ObservedDisabled(_) => Ok(()),
85        }
86    }
87}
88
89/// A local filesystem operation that failed without erasing its I/O cause.
90#[derive(Debug, thiserror::Error)]
91pub enum FileError {
92    #[error("{operation} {}: {source}", path.display())]
93    Path {
94        operation: &'static str,
95        path: PathBuf,
96        #[source]
97        source: std::io::Error,
98    },
99    #[error("{subject} is not a file: {}", path.display())]
100    NotFile {
101        subject: &'static str,
102        path: PathBuf,
103    },
104    #[error("{operation} {} -> {}: {source}", from.display(), to.display())]
105    BetweenPaths {
106        operation: &'static str,
107        from: PathBuf,
108        to: PathBuf,
109        #[source]
110        source: std::io::Error,
111    },
112    #[error("path has no parent directory: {}", path.display())]
113    NoParent { path: PathBuf },
114    #[error("atomic stage in {} cannot commit to {}", stage_parent.display(), destination.display())]
115    InvalidAtomicDestination {
116        stage_parent: PathBuf,
117        destination: PathBuf,
118    },
119    #[error("{subject} size overflow: {}", path.display())]
120    SizeOverflow {
121        subject: &'static str,
122        path: PathBuf,
123    },
124    #[error("local file range is too large: {len} bytes")]
125    RangeTooLarge { len: u64 },
126    #[error("file modification time for {} predates the Unix epoch: {source}", path.display())]
127    ModifiedBeforeUnixEpoch {
128        path: PathBuf,
129        #[source]
130        source: std::time::SystemTimeError,
131    },
132    #[error("atomic write {}: {source}", path.display())]
133    AtomicWrite {
134        path: PathBuf,
135        #[source]
136        source: WriteError<std::io::Error>,
137    },
138    #[error("{operation}; rollback failed: {rollback}")]
139    RollbackFailed {
140        operation: Box<FileError>,
141        rollback: Box<FileError>,
142    },
143}
144
145impl FileError {
146    pub fn at(operation: &'static str, path: impl Into<PathBuf>, source: std::io::Error) -> Self {
147        Self::Path {
148            operation,
149            path: path.into(),
150            source,
151        }
152    }
153
154    pub fn between(
155        operation: &'static str,
156        from: impl Into<PathBuf>,
157        to: impl Into<PathBuf>,
158        source: std::io::Error,
159    ) -> Self {
160        Self::BetweenPaths {
161            operation,
162            from: from.into(),
163            to: to.into(),
164            source,
165        }
166    }
167
168    pub fn rollback(operation: FileError, rollback: FileError) -> Self {
169        Self::RollbackFailed {
170            operation: Box::new(operation),
171            rollback: Box::new(rollback),
172        }
173    }
174}
175
176pub(crate) const TEMP_FILE_PREFIX: &str = ".tmp.";
177
178/// Whether a directory entry is an unpublished sibling left by this module's
179/// atomic writer. Callers that enumerate committed records exclude these;
180/// they are not part of the durable record set until renamed onto a target.
181pub fn is_atomic_staging_file_name(name: &std::ffi::OsStr) -> bool {
182    name.to_string_lossy().starts_with(TEMP_FILE_PREFIX)
183}
184
185/// A failed atomic write, tagged with whether the write had already committed.
186///
187/// The distinction is what a caller holding in-memory state needs: after
188/// [`WriteError::committed`] the bytes are installed and readers already see
189/// them, so the caller's own copy must move forward even though the call
190/// failed. Before commit the target file is untouched and the caller keeps
191/// what it had.
192#[derive(Debug)]
193pub enum WriteError<E> {
194    /// The target file is untouched; nothing was installed.
195    BeforeCommit(E),
196    /// The rename landed and readers see the new bytes; only the durability
197    /// work that follows it failed.
198    AfterCommit(E),
199}
200
201impl<E> WriteError<E> {
202    /// Whether the new bytes are already installed at the target path.
203    pub fn committed(&self) -> bool {
204        matches!(self, Self::AfterCommit(_))
205    }
206
207    pub fn into_inner(self) -> E {
208        match self {
209            Self::BeforeCommit(error) | Self::AfterCommit(error) => error,
210        }
211    }
212
213    /// Convert the payload, preserving the commit phase.
214    pub fn map<F>(self, convert: impl FnOnce(E) -> F) -> WriteError<F> {
215        match self {
216            Self::BeforeCommit(error) => WriteError::BeforeCommit(convert(error)),
217            Self::AfterCommit(error) => WriteError::AfterCommit(convert(error)),
218        }
219    }
220}
221
222impl<E: std::fmt::Display> std::fmt::Display for WriteError<E> {
223    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
224        match self {
225            Self::BeforeCommit(error) => write!(f, "{error}"),
226            Self::AfterCommit(error) => write!(f, "{error} (after the write committed)"),
227        }
228    }
229}
230
231impl<E: std::error::Error + 'static> std::error::Error for WriteError<E> {
232    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
233        match self {
234            Self::BeforeCommit(error) | Self::AfterCommit(error) => Some(error),
235        }
236    }
237}
238
239/// Install `bytes` as the complete contents of `path`.
240///
241/// The bytes go to a temporary sibling that is flushed to disk and then
242/// renamed onto `path`, so a concurrent reader sees either the old file or the
243/// whole new one. `path`'s parent directory must already exist.
244pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), WriteError<std::io::Error>> {
245    let parent = path
246        .parent()
247        .filter(|parent| !parent.as_os_str().is_empty())
248        .unwrap_or_else(|| Path::new("."));
249
250    let mut temp = tempfile::Builder::new()
251        .prefix(TEMP_FILE_PREFIX)
252        .tempfile_in(parent)
253        .map_err(WriteError::BeforeCommit)?;
254    temp.write_all(bytes).map_err(WriteError::BeforeCommit)?;
255    temp.as_file()
256        .sync_all()
257        .map_err(WriteError::BeforeCommit)?;
258    // Dropping the `NamedTempFile` on any failure above removes the sibling.
259    temp.persist(path)
260        .map_err(|error| WriteError::BeforeCommit(error.error))?;
261    flush_directory_blocking(parent).map_err(WriteError::AfterCommit)
262}
263
264/// One unpublished file whose destination is chosen after its bytes have been
265/// written. This is the streaming counterpart to [`write_atomic`]: callers can
266/// compute a content address while implementing [`std::io::Write`], then commit the
267/// completed file under that address with the owning durability policy's
268/// file-sync, rename, and directory-sync sequence.
269pub struct AtomicFileStage {
270    parent: PathBuf,
271    temp: tempfile::NamedTempFile,
272    file_sync: FileSync,
273}
274
275impl AtomicFileStage {
276    pub fn create_in(parent: &Path) -> Result<Self, std::io::Error> {
277        Self::create_in_with_file_sync(parent, FileSync::Enabled)
278    }
279
280    pub(crate) fn create_in_with_file_sync(
281        parent: &Path,
282        file_sync: FileSync,
283    ) -> Result<Self, std::io::Error> {
284        std::fs::create_dir_all(parent)?;
285        let temp = tempfile::Builder::new()
286            .prefix(TEMP_FILE_PREFIX)
287            .tempfile_in(parent)?;
288        Ok(Self {
289            parent: parent.to_path_buf(),
290            temp,
291            file_sync,
292        })
293    }
294
295    pub fn commit(self, destination: &Path) -> Result<(), WriteError<FileError>> {
296        if destination.parent() != Some(self.parent.as_path()) {
297            return Err(WriteError::BeforeCommit(
298                FileError::InvalidAtomicDestination {
299                    stage_parent: self.parent,
300                    destination: destination.to_path_buf(),
301                },
302            ));
303        }
304        let staged_path = self.temp.path().to_path_buf();
305        self.file_sync
306            .sync_file_blocking(self.temp.as_file())
307            .map_err(|source| {
308                WriteError::BeforeCommit(FileError::at("sync staged file", &staged_path, source))
309            })?;
310        self.temp.persist(destination).map_err(|error| {
311            WriteError::BeforeCommit(FileError::between(
312                "persist atomic stage",
313                &staged_path,
314                destination,
315                error.error,
316            ))
317        })?;
318        self.file_sync
319            .sync_parent_blocking(destination)
320            .map_err(WriteError::AfterCommit)
321    }
322}
323
324impl std::io::Write for AtomicFileStage {
325    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
326        self.temp.write(bytes)
327    }
328
329    fn flush(&mut self) -> std::io::Result<()> {
330        self.temp.flush()
331    }
332}
333
334/// One local file whose complete contents are installed with a durable rename.
335pub struct AtomicFile {
336    path: PathBuf,
337}
338
339impl AtomicFile {
340    pub fn new(path: PathBuf) -> Self {
341        Self { path }
342    }
343
344    pub fn read_optional(&self) -> Result<Option<Vec<u8>>, FileError> {
345        match std::fs::read(&self.path) {
346            Ok(bytes) => Ok(Some(bytes)),
347            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
348            Err(source) => Err(FileError::at("read file", &self.path, source)),
349        }
350    }
351
352    pub fn replace(&self, bytes: &[u8]) -> Result<(), FileError> {
353        let parent = parent_of(&self.path)?;
354        std::fs::create_dir_all(parent)
355            .map_err(|source| FileError::at("create parent directory", parent, source))?;
356        write_atomic(&self.path, bytes).map_err(|source| FileError::AtomicWrite {
357            path: self.path.clone(),
358            source,
359        })
360    }
361
362    pub fn remove(&self) -> Result<(), FileError> {
363        match std::fs::remove_file(&self.path) {
364            Ok(()) => Ok(()),
365            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
366            Err(source) => Err(FileError::at("remove file", &self.path, source)),
367        }
368    }
369
370    #[cfg(any(test, feature = "test-utils"))]
371    pub fn path(&self) -> &Path {
372        &self.path
373    }
374}
375
376/// The directory holding the entry that a rename onto `path` creates.
377fn parent_of(path: &Path) -> Result<&Path, FileError> {
378    path.parent().ok_or_else(|| FileError::NoParent {
379        path: path.to_path_buf(),
380    })
381}
382
383/// Flush the directory entry a rename onto `path` just wrote, so the installed
384/// file survives a crash. Every durable rename in the crate ends here.
385pub async fn sync_parent_dir(path: &Path) -> Result<(), FileError> {
386    let parent = parent_of(path)?;
387    flush_directory(parent)
388        .await
389        .map_err(|source| FileError::at("fsync parent directory", parent, source))
390}
391
392/// [`sync_parent_dir`] for callers that are not on the async runtime.
393pub fn sync_parent_dir_blocking(path: &Path) -> Result<(), FileError> {
394    let parent = parent_of(path)?;
395    flush_directory_blocking(parent)
396        .map_err(|source| FileError::at("fsync parent directory", parent, source))
397}
398
399#[cfg(unix)]
400async fn flush_directory(directory: &Path) -> std::io::Result<()> {
401    tokio::fs::File::open(directory).await?.sync_all().await
402}
403
404#[cfg(unix)]
405fn flush_directory_blocking(directory: &Path) -> std::io::Result<()> {
406    std::fs::File::open(directory)?.sync_all()
407}
408
409// Outside Unix the POSIX idiom has no counterpart: `FlushFileBuffers` on a
410// Windows directory handle needs write access that opening a directory cannot
411// grant, so the fsync fails with `ERROR_ACCESS_DENIED` (os error 5) and takes
412// every blob install and spool removal down with it. NTFS journals its own
413// metadata, so a rename's durability does not hang on a directory flush the way
414// it does on POSIX — the rename plus the file's own `sync_all` is the durability
415// the platform offers, which is why storage engines skip directory syncing here.
416#[cfg(not(unix))]
417async fn flush_directory(_directory: &Path) -> std::io::Result<()> {
418    Ok(())
419}
420
421#[cfg(not(unix))]
422fn flush_directory_blocking(_directory: &Path) -> std::io::Result<()> {
423    Ok(())
424}
425
426#[cfg(test)]
427mod tests {
428    use super::*;
429
430    #[test]
431    fn write_atomic_replaces_existing_contents() {
432        let directory = tempfile::tempdir().expect("temporary directory");
433        let path = directory.path().join("config.yaml");
434        std::fs::write(&path, b"old").expect("seed");
435
436        write_atomic(&path, b"new").expect("atomic write");
437
438        assert_eq!(std::fs::read(&path).expect("read"), b"new");
439    }
440
441    #[test]
442    fn write_atomic_leaves_no_temporary_sibling() {
443        let directory = tempfile::tempdir().expect("temporary directory");
444
445        write_atomic(&directory.path().join("config.yaml"), b"bytes").expect("atomic write");
446
447        let leftovers: Vec<_> = std::fs::read_dir(directory.path())
448            .expect("read dir")
449            .map(|entry| entry.expect("entry").file_name())
450            .filter(|name| name.to_string_lossy().starts_with(TEMP_FILE_PREFIX))
451            .collect();
452        assert!(leftovers.is_empty(), "{leftovers:?}");
453    }
454
455    #[test]
456    fn write_atomic_requires_an_existing_parent() {
457        let directory = tempfile::tempdir().expect("temporary directory");
458        let path = directory.path().join("missing").join("config.yaml");
459
460        let error = write_atomic(&path, b"bytes").expect_err("absent parent");
461
462        assert!(!error.committed());
463        assert_eq!(error.into_inner().kind(), std::io::ErrorKind::NotFound);
464        assert!(!path.exists());
465    }
466
467    #[test]
468    fn streaming_stage_commits_only_the_complete_file() {
469        let directory = tempfile::tempdir().expect("temporary directory");
470        let path = directory.path().join("payload");
471        let mut stage = AtomicFileStage::create_in(directory.path()).expect("create stage");
472
473        stage.write_all(b"first ").expect("write first part");
474        stage.write_all(b"second").expect("write second part");
475        assert!(!path.exists());
476        stage.commit(&path).expect("commit stage");
477
478        assert_eq!(std::fs::read(path).expect("read payload"), b"first second");
479    }
480
481    /// The durable-rename tail must succeed on every platform coven supports,
482    /// not only the ones with a POSIX directory fsync.
483    #[test]
484    fn write_atomic_commits_without_a_posix_directory_fsync() {
485        let directory = tempfile::tempdir().expect("temporary directory");
486        let path = directory.path().join("installed");
487
488        write_atomic(&path, b"bytes").expect("atomic write");
489        sync_parent_dir_blocking(&path).expect("sync parent directory");
490
491        assert_eq!(std::fs::read(&path).expect("read"), b"bytes");
492    }
493
494    #[tokio::test]
495    async fn sync_parent_dir_accepts_an_installed_file() {
496        let directory = tempfile::tempdir().expect("temporary directory");
497        let path = directory.path().join("installed");
498        tokio::fs::write(&path, b"bytes").await.expect("write");
499
500        sync_parent_dir(&path).await.expect("sync parent directory");
501    }
502
503    #[tokio::test]
504    async fn sync_parent_dir_rejects_a_path_with_no_parent() {
505        let error = sync_parent_dir(Path::new("/"))
506            .await
507            .expect_err("root has no parent");
508
509        assert!(matches!(
510            error,
511            FileError::NoParent { path } if path == Path::new("/")
512        ));
513    }
514
515    #[test]
516    fn atomic_file_round_trips_through_a_created_parent() {
517        let directory = tempfile::tempdir().expect("temporary directory");
518        let file = AtomicFile::new(directory.path().join("nested").join("config.yaml"));
519
520        assert_eq!(file.read_optional().expect("absent read"), None);
521        file.replace(b"first").expect("install");
522        file.replace(b"second").expect("replace");
523
524        assert_eq!(
525            file.read_optional().expect("read"),
526            Some(b"second".to_vec())
527        );
528    }
529}
coven_foundation/atomic_file.rs

coven_foundation/
atomic_file.rs
