async_fs_io/
atomic.rs

1//! Atomic, streaming file replacement.
2
3use std::path::{Path, PathBuf};
4
5use std::pin::Pin;
6use std::task::{Context, Poll};
7
8use tokio::io::{AsyncRead, ReadBuf};
9
10use crate::{AsyncFile, FsError, Operation};
11
12/// Stream `reader` into a temporary sibling file and atomically rename it over
13/// `target`.
14pub async fn atomic_write<R>(target: impl AsRef<Path>, mut reader: R) -> Result<(), FsError>
15where
16    R: AsyncRead + Unpin,
17{
18    let target = target.as_ref().to_owned();
19    let parent = target.parent().ok_or_else(|| {
20        FsError::InvalidRequest(format!("target has no parent: {}", target.display()))
21    })?;
22    tokio::fs::create_dir_all(parent)
23        .await
24        .map_err(|error| FsError::io(Operation::Directory, parent, error))?;
25
26    let temporary = temporary_sibling(&target);
27    let result = async {
28        let mut file = AsyncFile::create_new(&temporary).await?;
29        tokio::io::copy(&mut reader, &mut file)
30            .await
31            .map_err(|error| FsError::io(Operation::Write, &temporary, error))?;
32        file.flush().await?;
33        tokio::fs::rename(&temporary, &target)
34            .await
35            .map_err(|error| FsError::io(Operation::Write, &target, error))
36    }
37    .await;
38
39    match result {
40        Ok(()) => Ok(()),
41        Err(primary) => {
42            // Cleanup is part of the async failure path; it is not hidden in
43            // Drop and a cleanup failure must remain visible to the caller.
44            match tokio::fs::remove_file(&temporary).await {
45                Ok(()) => Err(primary),
46                Err(cleanup) => Err(FsError::Write {
47                    path: temporary.display().to_string(),
48                    detail: format!("{primary}; temporary-file cleanup failed: {cleanup}"),
49                }),
50            }
51        }
52    }
53}
54
55/// Atomically replace a file with UTF-8 text.
56pub async fn atomic_write_string(target: impl AsRef<Path>, value: &str) -> Result<(), FsError> {
57    atomic_write(target, SliceReader::new(value.as_bytes())).await
58}
59
60/// Atomically copy one existing file to another path.
61pub async fn atomic_copy(
62    source: impl AsRef<Path>,
63    target: impl AsRef<Path>,
64) -> Result<(), FsError> {
65    let source = source.as_ref();
66    let file = AsyncFile::open(source).await?;
67    atomic_write(target, file).await
68}
69
70fn temporary_sibling(target: &Path) -> PathBuf {
71    let mut value = target.as_os_str().to_owned();
72    value.push(format!(".tmp.{}", uuid::Uuid::new_v4()));
73    value.into()
74}
75
76struct SliceReader<'a> {
77    bytes: &'a [u8],
78    position: usize,
79}
80
81impl<'a> SliceReader<'a> {
82    fn new(bytes: &'a [u8]) -> Self {
83        Self { bytes, position: 0 }
84    }
85}
86
87impl AsyncRead for SliceReader<'_> {
88    fn poll_read(
89        mut self: Pin<&mut Self>,
90        _context: &mut Context<'_>,
91        buffer: &mut ReadBuf<'_>,
92    ) -> Poll<std::io::Result<()>> {
93        let remaining = &self.bytes[self.position..];
94        if remaining.is_empty() {
95            return Poll::Ready(Ok(()));
96        }
97        let amount = remaining.len().min(buffer.remaining());
98        buffer.put_slice(&remaining[..amount]);
99        self.position += amount;
100        Poll::Ready(Ok(()))
101    }
102}
async_fs_io/atomic.rs

async_fs_io/
atomic.rs
