s3_wire/client/
download.rs

1//! Atomic file download helpers built on the streaming object primitive.
2
3use std::path::{Path, PathBuf};
4
5use crate::error::S3Error;
6use crate::operation::{GetObjectRequest, ObjectMetadata};
7
8use super::S3Client;
9
10/// Result of an atomic object download to a local path.
11#[derive(Clone, Debug, Eq, PartialEq)]
12pub struct DownloadToPathOutput {
13    /// Metadata returned with the object.
14    pub metadata: ObjectMetadata,
15    /// Number of object bytes written.
16    pub bytes_written: u64,
17    /// Final destination path.
18    pub path: PathBuf,
19}
20
21impl S3Client {
22    /// Downloads an object into a private temporary file and atomically persists
23    /// it at `destination` only after the response body is fully verified.
24    ///
25    /// The temporary file is created in the destination directory, so the final
26    /// rename cannot cross filesystems. An existing destination is atomically
27    /// replaced where the platform supports replacement renames. Failed,
28    /// cancelled, truncated, or checksum-invalid downloads leave the previous
29    /// destination untouched and remove the temporary file on drop.
30    ///
31    /// # Errors
32    ///
33    /// Returns an error when the object request, destination I/O, integrity
34    /// verification, or atomic persist operation fails.
35    pub async fn download_to_path(
36        &self,
37        request: GetObjectRequest,
38        destination: impl AsRef<Path>,
39    ) -> Result<DownloadToPathOutput, S3Error> {
40        let destination = destination.as_ref().to_path_buf();
41        let parent = destination
42            .parent()
43            .filter(|path| !path.as_os_str().is_empty())
44            .unwrap_or_else(|| Path::new("."))
45            .to_path_buf();
46
47        let temporary =
48            tokio::task::spawn_blocking(move || tempfile::NamedTempFile::new_in(parent))
49                .await
50                .map_err(|_| {
51                    S3Error::cancellation("temporary download file creation was cancelled")
52                })?
53                .map_err(S3Error::transport)?;
54        let std_file = temporary
55            .as_file()
56            .try_clone()
57            .map_err(S3Error::transport)?;
58        let mut file = tokio::fs::File::from_std(std_file);
59
60        let output = self.get_object(request).await?;
61        let bytes_written = output.body.write_to(&mut file).await?;
62        file.sync_all().await.map_err(S3Error::transport)?;
63        drop(file);
64
65        // Persist directly: rename is a short filesystem operation, and keeping
66        // it in this future means cancellation cannot detach a background rename
67        // that later replaces the destination after the caller observed a drop.
68        persist_download(temporary, &destination)?;
69
70        Ok(DownloadToPathOutput {
71            metadata: output.metadata,
72            bytes_written,
73            path: destination,
74        })
75    }
76}
77
78fn persist_download(temporary: tempfile::NamedTempFile, destination: &Path) -> Result<(), S3Error> {
79    temporary
80        .persist(destination)
81        .map(|_| ())
82        .map_err(|error| S3Error::transport(error.error))
83}
84
85#[cfg(test)]
86mod tests {
87    use std::io::Write as _;
88
89    use super::*;
90
91    #[test]
92    fn atomic_persist_replaces_an_existing_destination() {
93        let directory = tempfile::tempdir().unwrap();
94        let destination = directory.path().join("object");
95        std::fs::write(&destination, b"old").unwrap();
96        let mut temporary = tempfile::NamedTempFile::new_in(directory.path()).unwrap();
97        temporary.write_all(b"new").unwrap();
98
99        persist_download(temporary, &destination).unwrap();
100
101        assert_eq!(std::fs::read(destination).unwrap(), b"new");
102    }
103}
s3_wire/client/download.rs

s3_wire/client/
download.rs
