async_fs_io/
atomic.rs

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

async_fs_io/
atomic.rs
