rns_net/
resource.rs

1//! Standard-library Resource streaming and disk-delivery types.
2
3use std::fs::{self, File, OpenOptions};
4#[cfg(test)]
5use std::io::Read;
6use std::io::{self, Write};
7use std::path::{Path, PathBuf};
8use std::sync::atomic::{AtomicU64, Ordering};
9
10use rns_core::constants::RESOURCE_AUTO_COMPRESS_MAX_SIZE;
11
12static NEXT_PERSIST_TEMP: AtomicU64 = AtomicU64::new(1);
13static NEXT_RECEIVE_TEMP: AtomicU64 = AtomicU64::new(1);
14
15/// A node-local identifier for an outgoing streaming Resource transfer.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
17pub struct ResourceTransferId(pub u64);
18
19/// Failure reported for the new streaming Resource API.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub enum ResourceTransferError {
22    Source(String),
23    Storage(String),
24    Protocol(String),
25    Cancelled,
26}
27
28/// How independent incoming Resources are delivered to the application.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub enum ResourceReceiveMode {
31    /// Assemble the complete logical Resource in memory.
32    Memory { max_bytes: u64 },
33    /// Write verified segments to a managed temporary file.
34    TemporaryFile {
35        directory: PathBuf,
36        max_bytes: Option<u64>,
37    },
38}
39
40impl Default for ResourceReceiveMode {
41    fn default() -> Self {
42        Self::Memory {
43            max_bytes: RESOURCE_AUTO_COMPRESS_MAX_SIZE as u64,
44        }
45    }
46}
47
48/// A received disk-backed Resource.
49///
50/// The temporary file is removed when this value is dropped unless ownership
51/// is transferred with [`ReceivedResourceFile::persist`].
52#[derive(Debug)]
53pub struct ReceivedResourceFile {
54    path: Option<PathBuf>,
55    pub original_hash: [u8; 32],
56    pub size: u64,
57    pub metadata: Option<Vec<u8>>,
58}
59
60impl ReceivedResourceFile {
61    pub(crate) fn new(
62        path: PathBuf,
63        original_hash: [u8; 32],
64        size: u64,
65        metadata: Option<Vec<u8>>,
66    ) -> Self {
67        Self {
68            path: Some(path),
69            original_hash,
70            size,
71            metadata,
72        }
73    }
74
75    /// Path of the managed temporary file.
76    pub fn path(&self) -> &Path {
77        self.path
78            .as_deref()
79            .expect("persisted resource has no path")
80    }
81
82    /// Persist this Resource at `destination`.
83    ///
84    /// The final name appears atomically. Cross-filesystem moves are staged in
85    /// the destination directory before the final rename.
86    pub fn persist(
87        mut self,
88        destination: impl AsRef<Path>,
89        overwrite: bool,
90    ) -> io::Result<PathBuf> {
91        let destination = destination.as_ref();
92        let source = self.path.as_ref().expect("persisted resource has no path");
93        match rename_final(source, destination, overwrite) {
94            Ok(()) => {
95                self.path = None;
96                sync_parent(destination)?;
97                return Ok(destination.to_path_buf());
98            }
99            Err(error) if error.raw_os_error() != Some(libc::EXDEV) => return Err(error),
100            Err(_) => {}
101        }
102
103        let parent = destination.parent().unwrap_or_else(|| Path::new("."));
104        let sequence = NEXT_PERSIST_TEMP.fetch_add(1, Ordering::Relaxed);
105        let stage = parent.join(format!(
106            ".rns-resource-{}-{}.tmp",
107            std::process::id(),
108            sequence
109        ));
110        let mut input = File::open(source)?;
111        let mut output = OpenOptions::new()
112            .write(true)
113            .create_new(true)
114            .open(&stage)?;
115        let copied = io::copy(&mut input, &mut output)?;
116        if copied != self.size {
117            let _ = fs::remove_file(&stage);
118            return Err(io::Error::new(
119                io::ErrorKind::UnexpectedEof,
120                "resource size changed",
121            ));
122        }
123        output.flush()?;
124        output.sync_all()?;
125        drop(output);
126        rename_final(&stage, destination, overwrite).inspect_err(|_| {
127            let _ = fs::remove_file(&stage);
128        })?;
129        sync_parent(destination)?;
130        fs::remove_file(source)?;
131        self.path = None;
132        Ok(destination.to_path_buf())
133    }
134}
135
136#[cfg(target_os = "linux")]
137fn rename_final(source: &Path, destination: &Path, overwrite: bool) -> io::Result<()> {
138    if overwrite {
139        return fs::rename(source, destination);
140    }
141    use std::ffi::CString;
142    use std::os::unix::ffi::OsStrExt;
143    let source = CString::new(source.as_os_str().as_bytes())
144        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "source path contains NUL"))?;
145    let destination = CString::new(destination.as_os_str().as_bytes()).map_err(|_| {
146        io::Error::new(io::ErrorKind::InvalidInput, "destination path contains NUL")
147    })?;
148    let result = unsafe {
149        libc::renameat2(
150            libc::AT_FDCWD,
151            source.as_ptr(),
152            libc::AT_FDCWD,
153            destination.as_ptr(),
154            libc::RENAME_NOREPLACE,
155        )
156    };
157    if result == 0 {
158        Ok(())
159    } else {
160        Err(io::Error::last_os_error())
161    }
162}
163
164#[cfg(not(target_os = "linux"))]
165fn rename_final(source: &Path, destination: &Path, overwrite: bool) -> io::Result<()> {
166    if !overwrite && destination.exists() {
167        return Err(io::Error::new(
168            io::ErrorKind::AlreadyExists,
169            "resource destination already exists",
170        ));
171    }
172    fs::rename(source, destination)
173}
174
175impl Drop for ReceivedResourceFile {
176    fn drop(&mut self) {
177        if let Some(path) = self.path.take() {
178            let _ = fs::remove_file(path);
179        }
180    }
181}
182
183fn sync_parent(path: &Path) -> io::Result<()> {
184    if let Some(parent) = path.parent() {
185        File::open(parent)?.sync_all()?;
186    }
187    Ok(())
188}
189
190pub(crate) fn create_receive_file(
191    directory: &Path,
192    original_hash: &[u8; 32],
193) -> io::Result<(File, PathBuf)> {
194    fs::create_dir_all(directory)?;
195    for _ in 0..100 {
196        let sequence = NEXT_RECEIVE_TEMP.fetch_add(1, Ordering::Relaxed);
197        let path = directory.join(format!(
198            "rns-resource-{:02x}{:02x}{:02x}{:02x}-{}-{}.part",
199            original_hash[0],
200            original_hash[1],
201            original_hash[2],
202            original_hash[3],
203            std::process::id(),
204            sequence
205        ));
206        match OpenOptions::new().write(true).create_new(true).open(&path) {
207            Ok(file) => return Ok((file, path)),
208            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
209            Err(error) => return Err(error),
210        }
211    }
212    Err(io::Error::new(
213        io::ErrorKind::AlreadyExists,
214        "could not allocate unique Resource temporary file",
215    ))
216}
217
218/// Read exactly `length` bytes and reject both early EOF and trailing data.
219#[cfg(test)]
220fn read_declared(mut reader: impl Read, length: u64) -> io::Result<Vec<u8>> {
221    let length = usize::try_from(length)
222        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "length does not fit usize"))?;
223    let mut data = vec![0; length];
224    reader.read_exact(&mut data)?;
225    let mut trailing = [0u8; 1];
226    if reader.read(&mut trailing)? != 0 {
227        return Err(io::Error::new(
228            io::ErrorKind::InvalidData,
229            "resource source exceeds declared length",
230        ));
231    }
232    Ok(data)
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238
239    #[test]
240    fn managed_file_is_deleted_on_drop_and_can_be_persisted() {
241        let root = tempfile::tempdir().unwrap();
242        let temporary = root.path().join("temporary");
243        fs::write(&temporary, b"resource").unwrap();
244        {
245            let artifact = ReceivedResourceFile::new(temporary.clone(), [7; 32], 8, None);
246            assert_eq!(artifact.path(), temporary);
247        }
248        assert!(!temporary.exists());
249
250        fs::write(&temporary, b"resource").unwrap();
251        let destination = root.path().join("saved");
252        let artifact = ReceivedResourceFile::new(temporary.clone(), [7; 32], 8, None);
253        artifact.persist(&destination, false).unwrap();
254        assert_eq!(fs::read(destination).unwrap(), b"resource");
255        assert!(!temporary.exists());
256    }
257
258    #[test]
259    fn declared_reader_rejects_short_and_long_sources() {
260        assert_eq!(read_declared(&b"abc"[..], 3).unwrap(), b"abc");
261        assert_eq!(
262            read_declared(&b"ab"[..], 3).unwrap_err().kind(),
263            io::ErrorKind::UnexpectedEof
264        );
265        assert_eq!(
266            read_declared(&b"abcd"[..], 3).unwrap_err().kind(),
267            io::ErrorKind::InvalidData
268        );
269    }
270}
rns_net/resource.rs

rns_net/
resource.rs
