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    ///
33    /// Nodes can verify compressed, single-segment application Resources of
34    /// 16 KiB through the efficient segment limit on a lazy receive worker.
35    /// Authentication remains on the driver. Worker admission reserves owned
36    /// input and worst-case output/metadata capacity: four jobs and 256 MiB,
37    /// allowing one job at the default 64 MiB decoder limit. This reservation
38    /// is not preallocated memory or a whole-node RSS cap; native codec workspace
39    /// and unrelated transfers are additional. Small, split, request/response
40    /// and saturated work stays synchronous. Earlier same-link assemblies finish
41    /// first, so overload can still stall the driver. Shutdown joins active work.
42    Memory { max_bytes: u64 },
43    /// Write verified segments to a managed temporary file.
44    TemporaryFile {
45        directory: PathBuf,
46        max_bytes: Option<u64>,
47    },
48}
49
50impl Default for ResourceReceiveMode {
51    fn default() -> Self {
52        Self::Memory {
53            max_bytes: RESOURCE_AUTO_COMPRESS_MAX_SIZE as u64,
54        }
55    }
56}
57
58/// A received disk-backed Resource.
59///
60/// The temporary file is removed when this value is dropped unless ownership
61/// is transferred with [`ReceivedResourceFile::persist`].
62#[derive(Debug)]
63pub struct ReceivedResourceFile {
64    path: Option<PathBuf>,
65    pub original_hash: [u8; 32],
66    pub size: u64,
67    pub metadata: Option<Vec<u8>>,
68}
69
70impl ReceivedResourceFile {
71    pub(crate) fn new(
72        path: PathBuf,
73        original_hash: [u8; 32],
74        size: u64,
75        metadata: Option<Vec<u8>>,
76    ) -> Self {
77        Self {
78            path: Some(path),
79            original_hash,
80            size,
81            metadata,
82        }
83    }
84
85    /// Path of the managed temporary file.
86    pub fn path(&self) -> &Path {
87        self.path
88            .as_deref()
89            .expect("persisted resource has no path")
90    }
91
92    /// Persist this Resource at `destination`.
93    ///
94    /// The final name appears atomically. Cross-filesystem moves are staged in
95    /// the destination directory before the final rename.
96    pub fn persist(
97        mut self,
98        destination: impl AsRef<Path>,
99        overwrite: bool,
100    ) -> io::Result<PathBuf> {
101        let destination = destination.as_ref();
102        let source = self.path.as_ref().expect("persisted resource has no path");
103        match rename_final(source, destination, overwrite) {
104            Ok(()) => {
105                self.path = None;
106                sync_parent(destination)?;
107                return Ok(destination.to_path_buf());
108            }
109            Err(error) if error.raw_os_error() != Some(libc::EXDEV) => return Err(error),
110            Err(_) => {}
111        }
112
113        let parent = destination.parent().unwrap_or_else(|| Path::new("."));
114        let sequence = NEXT_PERSIST_TEMP.fetch_add(1, Ordering::Relaxed);
115        let stage = parent.join(format!(
116            ".rns-resource-{}-{}.tmp",
117            std::process::id(),
118            sequence
119        ));
120        let mut input = File::open(source)?;
121        let mut output = OpenOptions::new()
122            .write(true)
123            .create_new(true)
124            .open(&stage)?;
125        let copied = io::copy(&mut input, &mut output)?;
126        if copied != self.size {
127            let _ = fs::remove_file(&stage);
128            return Err(io::Error::new(
129                io::ErrorKind::UnexpectedEof,
130                "resource size changed",
131            ));
132        }
133        output.flush()?;
134        output.sync_all()?;
135        drop(output);
136        rename_final(&stage, destination, overwrite).inspect_err(|_| {
137            let _ = fs::remove_file(&stage);
138        })?;
139        sync_parent(destination)?;
140        fs::remove_file(source)?;
141        self.path = None;
142        Ok(destination.to_path_buf())
143    }
144}
145
146#[cfg(target_os = "linux")]
147fn rename_final(source: &Path, destination: &Path, overwrite: bool) -> io::Result<()> {
148    if overwrite {
149        return fs::rename(source, destination);
150    }
151    use std::ffi::CString;
152    use std::os::unix::ffi::OsStrExt;
153    let source = CString::new(source.as_os_str().as_bytes())
154        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "source path contains NUL"))?;
155    let destination = CString::new(destination.as_os_str().as_bytes()).map_err(|_| {
156        io::Error::new(io::ErrorKind::InvalidInput, "destination path contains NUL")
157    })?;
158    let result = unsafe {
159        libc::renameat2(
160            libc::AT_FDCWD,
161            source.as_ptr(),
162            libc::AT_FDCWD,
163            destination.as_ptr(),
164            libc::RENAME_NOREPLACE,
165        )
166    };
167    if result == 0 {
168        Ok(())
169    } else {
170        Err(io::Error::last_os_error())
171    }
172}
173
174#[cfg(not(target_os = "linux"))]
175fn rename_final(source: &Path, destination: &Path, overwrite: bool) -> io::Result<()> {
176    if !overwrite && destination.exists() {
177        return Err(io::Error::new(
178            io::ErrorKind::AlreadyExists,
179            "resource destination already exists",
180        ));
181    }
182    fs::rename(source, destination)
183}
184
185impl Drop for ReceivedResourceFile {
186    fn drop(&mut self) {
187        if let Some(path) = self.path.take() {
188            let _ = fs::remove_file(path);
189        }
190    }
191}
192
193fn sync_parent(path: &Path) -> io::Result<()> {
194    if let Some(parent) = path.parent() {
195        File::open(parent)?.sync_all()?;
196    }
197    Ok(())
198}
199
200pub(crate) fn create_receive_file(
201    directory: &Path,
202    original_hash: &[u8; 32],
203) -> io::Result<(File, PathBuf)> {
204    fs::create_dir_all(directory)?;
205    for _ in 0..100 {
206        let sequence = NEXT_RECEIVE_TEMP.fetch_add(1, Ordering::Relaxed);
207        let path = directory.join(format!(
208            "rns-resource-{:02x}{:02x}{:02x}{:02x}-{}-{}.part",
209            original_hash[0],
210            original_hash[1],
211            original_hash[2],
212            original_hash[3],
213            std::process::id(),
214            sequence
215        ));
216        match OpenOptions::new().write(true).create_new(true).open(&path) {
217            Ok(file) => return Ok((file, path)),
218            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
219            Err(error) => return Err(error),
220        }
221    }
222    Err(io::Error::new(
223        io::ErrorKind::AlreadyExists,
224        "could not allocate unique Resource temporary file",
225    ))
226}
227
228/// Read exactly `length` bytes and reject both early EOF and trailing data.
229#[cfg(test)]
230fn read_declared(mut reader: impl Read, length: u64) -> io::Result<Vec<u8>> {
231    let length = usize::try_from(length)
232        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "length does not fit usize"))?;
233    let mut data = vec![0; length];
234    reader.read_exact(&mut data)?;
235    let mut trailing = [0u8; 1];
236    if reader.read(&mut trailing)? != 0 {
237        return Err(io::Error::new(
238            io::ErrorKind::InvalidData,
239            "resource source exceeds declared length",
240        ));
241    }
242    Ok(data)
243}
244
245#[cfg(test)]
246mod tests {
247    use super::*;
248
249    #[test]
250    fn managed_file_is_deleted_on_drop_and_can_be_persisted() {
251        let root = tempfile::tempdir().unwrap();
252        let temporary = root.path().join("temporary");
253        fs::write(&temporary, b"resource").unwrap();
254        {
255            let artifact = ReceivedResourceFile::new(temporary.clone(), [7; 32], 8, None);
256            assert_eq!(artifact.path(), temporary);
257        }
258        assert!(!temporary.exists());
259
260        fs::write(&temporary, b"resource").unwrap();
261        let destination = root.path().join("saved");
262        let artifact = ReceivedResourceFile::new(temporary.clone(), [7; 32], 8, None);
263        artifact.persist(&destination, false).unwrap();
264        assert_eq!(fs::read(destination).unwrap(), b"resource");
265        assert!(!temporary.exists());
266    }
267
268    #[test]
269    fn declared_reader_rejects_short_and_long_sources() {
270        assert_eq!(read_declared(&b"abc"[..], 3).unwrap(), b"abc");
271        assert_eq!(
272            read_declared(&b"ab"[..], 3).unwrap_err().kind(),
273            io::ErrorKind::UnexpectedEof
274        );
275        assert_eq!(
276            read_declared(&b"abcd"[..], 3).unwrap_err().kind(),
277            io::ErrorKind::InvalidData
278        );
279    }
280}
rns_net/resource.rs

rns_net/
resource.rs
