btpc_core/create/
atomic_output.rs

1use std::fs;
2use std::path::Path;
3
4use crate::{Error, Result};
5
6use super::{DurabilityPolicy, OverwritePolicy};
7
8/// Atomically writes bytes using BTPC's overwrite and durability policies.
9///
10/// # Errors
11///
12/// Returns an I/O error while creating, syncing, or publishing the destination.
13// Spec: CLI-WRITE-001
14pub fn write_atomic(
15    destination: &Path,
16    bytes: &[u8],
17    overwrite: OverwritePolicy,
18    durability: DurabilityPolicy,
19) -> Result<()> {
20    atomic_write_with_hook(
21        destination,
22        bytes,
23        overwrite,
24        durability,
25        &NoAtomicWriteHook,
26    )
27}
28
29#[derive(Clone, Copy)]
30enum AtomicWriteStage {
31    Write,
32    SyncFile,
33    Publish,
34    SyncDirectory,
35}
36
37trait AtomicWriteHook {
38    fn before(
39        &self,
40        _stage: AtomicWriteStage,
41        _temporary: &Path,
42        _destination: &Path,
43    ) -> std::io::Result<()> {
44        Ok(())
45    }
46}
47
48struct NoAtomicWriteHook;
49
50impl AtomicWriteHook for NoAtomicWriteHook {}
51
52fn atomic_write_with_hook(
53    destination: &Path,
54    bytes: &[u8],
55    overwrite: OverwritePolicy,
56    durability: DurabilityPolicy,
57    hook: &impl AtomicWriteHook,
58) -> Result<()> {
59    let parent = destination.parent().unwrap_or_else(|| Path::new("."));
60    fs::create_dir_all(parent).map_err(|source| Error::io(parent, source))?;
61    let file_name = destination
62        .file_name()
63        .and_then(|name| name.to_str())
64        .unwrap_or("torrent");
65    let mut temporary = tempfile::Builder::new()
66        .prefix(&format!(".{file_name}.btpc-tmp-"))
67        .tempfile_in(parent)
68        .map_err(|source| Error::io(parent, source))?;
69    let temporary_path = temporary.path().to_path_buf();
70    preserve_replaced_file_permissions(&temporary, destination, overwrite)?;
71    {
72        let file = temporary.as_file_mut();
73        hook.before(AtomicWriteStage::Write, &temporary_path, destination)
74            .map_err(|source| Error::io(&temporary_path, source))?;
75        std::io::Write::write_all(file, bytes)
76            .map_err(|source| Error::io(&temporary_path, source))?;
77        hook.before(AtomicWriteStage::SyncFile, &temporary_path, destination)
78            .map_err(|source| Error::io(&temporary_path, source))?;
79        file.sync_all()
80            .map_err(|source| Error::io(&temporary_path, source))?;
81    }
82    hook.before(AtomicWriteStage::Publish, &temporary_path, destination)
83        .map_err(|source| Error::io(destination, source))?;
84    publish(temporary, destination, overwrite)?;
85    if durability == DurabilityPolicy::FileAndDirectory {
86        hook.before(
87            AtomicWriteStage::SyncDirectory,
88            &temporary_path,
89            destination,
90        )
91        .map_err(|source| Error::io(parent, source))?;
92        sync_parent(parent)?;
93    }
94    Ok(())
95}
96
97fn preserve_replaced_file_permissions(
98    temporary: &tempfile::NamedTempFile,
99    destination: &Path,
100    overwrite: OverwritePolicy,
101) -> Result<()> {
102    if overwrite != OverwritePolicy::Replace {
103        return Ok(());
104    }
105    match fs::symlink_metadata(destination) {
106        Ok(metadata) if metadata.file_type().is_file() => temporary
107            .as_file()
108            .set_permissions(metadata.permissions())
109            .map_err(|source| Error::io(temporary.path(), source)),
110        Ok(_) => Ok(()),
111        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
112        Err(source) => Err(Error::io(destination, source)),
113    }
114}
115
116fn publish(
117    temporary: tempfile::NamedTempFile,
118    destination: &Path,
119    overwrite: OverwritePolicy,
120) -> Result<()> {
121    match overwrite {
122        OverwritePolicy::Deny => temporary
123            .persist_noclobber(destination)
124            .map(|_| ())
125            .map_err(|error| {
126                let source = if error.error.kind() == std::io::ErrorKind::AlreadyExists {
127                    std::io::Error::new(
128                        std::io::ErrorKind::AlreadyExists,
129                        "destination already exists",
130                    )
131                } else {
132                    error.error
133                };
134                Error::io(destination, source)
135            }),
136        OverwritePolicy::Replace => temporary
137            .persist(destination)
138            .map(|_| ())
139            .map_err(|error| Error::io(destination, error.error)),
140    }
141}
142
143#[cfg(unix)]
144fn sync_parent(parent: &Path) -> Result<()> {
145    fs::File::open(parent)
146        .and_then(|directory| directory.sync_all())
147        .map_err(|source| Error::io(parent, source))
148}
149
150#[cfg(not(unix))]
151fn sync_parent(_parent: &Path) -> Result<()> {
152    Ok(())
153}
154
155#[cfg(test)]
156mod atomic_write_tests {
157    use std::fs;
158    use std::io;
159    use std::path::{Path, PathBuf};
160
161    use super::{
162        AtomicWriteHook, AtomicWriteStage, DurabilityPolicy, OverwritePolicy,
163        atomic_write_with_hook,
164    };
165    use tempfile::TempDir;
166
167    struct FailAt(AtomicWriteStage);
168
169    impl AtomicWriteHook for FailAt {
170        fn before(
171            &self,
172            stage: AtomicWriteStage,
173            _temporary: &Path,
174            _destination: &Path,
175        ) -> io::Result<()> {
176            if std::mem::discriminant(&stage) == std::mem::discriminant(&self.0) {
177                return Err(io::Error::new(io::ErrorKind::PermissionDenied, "injected"));
178            }
179            Ok(())
180        }
181    }
182
183    struct CreateDestinationBeforePublish(Vec<u8>);
184
185    impl AtomicWriteHook for CreateDestinationBeforePublish {
186        fn before(
187            &self,
188            stage: AtomicWriteStage,
189            _temporary: &Path,
190            destination: &Path,
191        ) -> io::Result<()> {
192            if matches!(stage, AtomicWriteStage::Publish) {
193                fs::write(destination, &self.0)?;
194            }
195            Ok(())
196        }
197    }
198
199    #[test]
200    fn destination_created_immediately_before_publish_is_never_clobbered() {
201        let temp = TempDir::new().unwrap();
202        let destination = temp.path().join("output.torrent");
203        let sentinel = b"other process".to_vec();
204        let error = atomic_write_with_hook(
205            &destination,
206            b"new output",
207            OverwritePolicy::Deny,
208            DurabilityPolicy::File,
209            &CreateDestinationBeforePublish(sentinel.clone()),
210        )
211        .unwrap_err();
212        assert_eq!(error.category(), crate::ErrorCategory::Io);
213        assert_eq!(fs::read(&destination).unwrap(), sentinel);
214        assert_eq!(temporary_paths(temp.path()), Vec::<PathBuf>::new());
215    }
216
217    #[test]
218    fn write_sync_and_publish_failures_remove_temporary_files() {
219        for stage in [
220            AtomicWriteStage::Write,
221            AtomicWriteStage::SyncFile,
222            AtomicWriteStage::Publish,
223        ] {
224            let temp = TempDir::new().unwrap();
225            let destination = temp.path().join("output.torrent");
226            assert!(
227                atomic_write_with_hook(
228                    &destination,
229                    b"new output",
230                    OverwritePolicy::Deny,
231                    DurabilityPolicy::File,
232                    &FailAt(stage),
233                )
234                .is_err()
235            );
236            assert!(!destination.exists());
237            assert_eq!(temporary_paths(temp.path()), Vec::<PathBuf>::new());
238        }
239    }
240
241    #[test]
242    fn directory_sync_failure_reports_after_complete_publication() {
243        let temp = TempDir::new().unwrap();
244        let destination = temp.path().join("output.torrent");
245        assert!(
246            atomic_write_with_hook(
247                &destination,
248                b"complete output",
249                OverwritePolicy::Deny,
250                DurabilityPolicy::FileAndDirectory,
251                &FailAt(AtomicWriteStage::SyncDirectory),
252            )
253            .is_err()
254        );
255        assert_eq!(fs::read(&destination).unwrap(), b"complete output");
256        assert_eq!(temporary_paths(temp.path()), Vec::<PathBuf>::new());
257    }
258
259    fn temporary_paths(directory: &Path) -> Vec<PathBuf> {
260        fs::read_dir(directory)
261            .unwrap()
262            .filter_map(Result::ok)
263            .map(|entry| entry.path())
264            .filter(|path| path.to_string_lossy().contains(".btpc-tmp"))
265            .collect()
266    }
267}
btpc_core/create/atomic_output.rs

btpc_core/create/
atomic_output.rs
