canic_host/durable_io/
mod.rs

1//! Module: durable_io
2//!
3//! Responsibility: own atomic durable regular-file publication and narrow canonical-document
4//! mechanics for `canic-host`.
5//! Does not own: domain schemas or transitions, path selection, ephemeral protocol files, open
6//! command-result descriptors, backup persistence, or multi-file transactions.
7//! Boundary: reads reject links/special files; writes own sibling staging, publication, cleanup,
8//! and filesystem syncs behind replace and create-new modes; document helpers add only bounded
9//! encoding, reads and exact replacement reconciliation.
10
11#[cfg(test)]
12mod tests;
13
14use std::{fs, io, path::Path};
15
16#[derive(Debug)]
17pub(crate) enum RegularFileReadError {
18    NotRegular,
19    Io(io::Error),
20    #[cfg(not(unix))]
21    UnsupportedPlatform,
22}
23
24#[derive(Debug)]
25pub(crate) enum BoundedRegularFileReadError {
26    Read(RegularFileReadError),
27    TooLarge,
28}
29
30#[derive(Debug)]
31pub(crate) enum RegularFileLockError {
32    NotRegular,
33    Io(io::Error),
34    #[cfg(windows)]
35    UnsupportedPlatform,
36}
37
38/// Read one optional regular file without following a final symlink.
39pub(crate) fn read_optional_regular_bytes(
40    path: &Path,
41) -> Result<Option<Vec<u8>>, RegularFileReadError> {
42    #[cfg(unix)]
43    {
44        supported::read_optional_regular_bytes(path)
45    }
46
47    #[cfg(not(unix))]
48    {
49        let _ = path;
50        Err(RegularFileReadError::UnsupportedPlatform)
51    }
52}
53
54/// Read at most `maximum_bytes + 1` from one optional regular no-follow file.
55///
56/// Descriptor metadata rejects an already oversized file before allocating its
57/// contents. The extra byte in the bounded read detects growth after that
58/// metadata observation.
59pub(crate) fn read_optional_regular_bytes_bounded(
60    path: &Path,
61    maximum_bytes: usize,
62) -> Result<Option<Vec<u8>>, BoundedRegularFileReadError> {
63    #[cfg(unix)]
64    {
65        supported::read_optional_regular_bytes_bounded(path, maximum_bytes)
66    }
67
68    #[cfg(not(unix))]
69    {
70        let _ = (path, maximum_bytes);
71        Err(BoundedRegularFileReadError::Read(
72            RegularFileReadError::UnsupportedPlatform,
73        ))
74    }
75}
76
77#[derive(Clone, Copy, Debug, Eq, PartialEq)]
78enum FileCommitMode {
79    Replace,
80    CreateNew,
81    CreateNewWithParents,
82}
83
84/// Durably replace one file through atomic publication of complete bytes.
85///
86/// Missing parent directories are created and durably linked before the file
87/// is published. Serialization must complete before calling this helper.
88pub fn write_bytes(path: &Path, bytes: &[u8]) -> io::Result<()> {
89    commit_bytes(path, bytes, FileCommitMode::Replace)
90}
91
92/// Durably create one file without replacing an existing destination.
93///
94/// The parent directory must already exist. Serialization must complete before
95/// calling this helper.
96pub fn create_new_bytes(path: &Path, bytes: &[u8]) -> io::Result<()> {
97    commit_bytes(path, bytes, FileCommitMode::CreateNew)
98}
99
100/// Durably create one file and its missing parent hierarchy without replacing
101/// an existing destination.
102pub fn create_new_bytes_with_parents(path: &Path, bytes: &[u8]) -> io::Result<()> {
103    commit_bytes(path, bytes, FileCommitMode::CreateNewWithParents)
104}
105
106/// Read a bounded regular file without following a final symlink.
107pub fn read_regular_bytes(path: &Path, maximum_bytes: usize) -> io::Result<Vec<u8>> {
108    match read_optional_regular_bytes_bounded(path, maximum_bytes) {
109        Ok(Some(bytes)) => Ok(bytes),
110        Ok(None) => Err(io::Error::from(io::ErrorKind::NotFound)),
111        Err(BoundedRegularFileReadError::Read(RegularFileReadError::Io(error))) => Err(error),
112        Err(error) => Err(io::Error::other(format!(
113            "invalid bounded regular file: {error:?}"
114        ))),
115    }
116}
117
118/// Hold an exclusive kernel lock on a durable regular file until the returned handle drops.
119pub fn lock_file(path: &Path) -> io::Result<fs::File> {
120    lock_regular_file_with_parents(path)
121        .map_err(|error| io::Error::other(format!("cannot lock regular file: {error:?}")))
122}
123
124/// Open and exclusively lock one durable regular no-follow file.
125///
126/// The lock file and missing parent hierarchy are durably created first. The
127/// returned descriptor owns the kernel lock and is close-on-exec.
128pub(crate) fn lock_regular_file_with_parents(
129    path: &Path,
130) -> Result<fs::File, RegularFileLockError> {
131    open_regular_lock_file(path, |file| {
132        #[cfg(not(windows))]
133        rustix::fs::flock(file, rustix::fs::FlockOperation::LockExclusive)
134            .map_err(errno_to_lock_error)?;
135        Ok(())
136    })
137}
138
139/// Acquire the same exclusive lock, reporting contention while preserving its lifetime.
140pub(crate) fn lock_file_with_progress(
141    path: &Path,
142    mut waiting: impl FnMut(std::time::Duration),
143) -> io::Result<fs::File> {
144    let started = std::time::Instant::now();
145    open_regular_lock_file(path, |file| {
146        #[cfg(not(windows))]
147        {
148            let mut next_report = std::time::Duration::from_secs(1);
149            loop {
150                match rustix::fs::flock(file, rustix::fs::FlockOperation::NonBlockingLockExclusive)
151                {
152                    Ok(()) => break,
153                    Err(rustix::io::Errno::WOULDBLOCK) => {
154                        let elapsed = started.elapsed();
155                        if elapsed >= next_report {
156                            waiting(elapsed);
157                            next_report = elapsed + std::time::Duration::from_secs(5);
158                        }
159                        std::thread::sleep(std::time::Duration::from_millis(100));
160                    }
161                    Err(error) => return Err(errno_to_lock_error(error)),
162                }
163            }
164        }
165        Ok(())
166    })
167    .map_err(|error| io::Error::other(format!("cannot lock regular file: {error:?}")))
168}
169
170fn open_regular_lock_file(
171    path: &Path,
172    acquire: impl FnOnce(&fs::File) -> Result<(), RegularFileLockError>,
173) -> Result<fs::File, RegularFileLockError> {
174    match create_new_bytes_with_parents(path, &[]) {
175        Ok(()) => {}
176        Err(source) if source.kind() == io::ErrorKind::AlreadyExists => {}
177        Err(source) => return Err(RegularFileLockError::Io(source)),
178    }
179    let metadata = fs::symlink_metadata(path).map_err(RegularFileLockError::Io)?;
180    if !metadata.file_type().is_file() {
181        return Err(RegularFileLockError::NotRegular);
182    }
183
184    #[cfg(not(windows))]
185    {
186        use rustix::{
187            fd::OwnedFd,
188            fs::{FileType, Mode, OFlags, fstat, open},
189        };
190
191        let fd: OwnedFd = open(
192            path,
193            OFlags::RDWR | OFlags::NOFOLLOW | OFlags::NONBLOCK | OFlags::CLOEXEC,
194            Mode::empty(),
195        )
196        .map_err(errno_to_lock_error)?;
197        let metadata = fstat(&fd).map_err(errno_to_lock_error)?;
198        if FileType::from_raw_mode(metadata.st_mode) != FileType::RegularFile {
199            return Err(RegularFileLockError::NotRegular);
200        }
201        let file = fs::File::from(fd);
202        acquire(&file)?;
203        Ok(file)
204    }
205
206    #[cfg(windows)]
207    {
208        Err(RegularFileLockError::UnsupportedPlatform)
209    }
210}
211
212#[cfg(not(windows))]
213fn errno_to_lock_error(source: rustix::io::Errno) -> RegularFileLockError {
214    RegularFileLockError::Io(io::Error::from_raw_os_error(source.raw_os_error()))
215}
216
217fn commit_bytes(path: &Path, bytes: &[u8], mode: FileCommitMode) -> io::Result<()> {
218    #[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
219    {
220        supported::commit_with_hook(path, bytes, mode, |_, _| Ok(()))
221    }
222
223    #[cfg(not(any(target_os = "linux", target_os = "android", target_vendor = "apple")))]
224    {
225        let _ = (path, bytes, mode);
226        Err(io::Error::new(
227            io::ErrorKind::Unsupported,
228            format!(
229                "durable atomic file publication is unsupported on platform {}",
230                std::env::consts::OS
231            ),
232        ))
233    }
234}
235
236#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
237mod supported {
238    use super::{BoundedRegularFileReadError, FileCommitMode, RegularFileReadError};
239
240    use std::{
241        ffi::{OsStr, OsString},
242        fs,
243        io::{self, Read, Write},
244        path::{Path, PathBuf},
245        sync::atomic::{AtomicU64, Ordering},
246    };
247
248    use rustix::{
249        fd::{AsFd, OwnedFd},
250        fs::{self as unix_fs, AtFlags, Mode, OFlags, RenameFlags},
251    };
252
253    const TEMP_ATTEMPTS: usize = 64;
254    static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0);
255
256    pub(super) fn read_optional_regular_bytes(
257        path: &Path,
258    ) -> Result<Option<Vec<u8>>, RegularFileReadError> {
259        let Some((mut file, _)) = open_optional_regular_file(path)? else {
260            return Ok(None);
261        };
262        let mut bytes = Vec::new();
263        file.read_to_end(&mut bytes)
264            .map_err(RegularFileReadError::Io)?;
265        Ok(Some(bytes))
266    }
267
268    pub(super) fn read_optional_regular_bytes_bounded(
269        path: &Path,
270        maximum_bytes: usize,
271    ) -> Result<Option<Vec<u8>>, BoundedRegularFileReadError> {
272        read_optional_regular_bytes_bounded_with_hook(path, maximum_bytes, || Ok(()))
273    }
274
275    #[cfg(test)]
276    pub(super) fn read_optional_regular_bytes_bounded_with_hook(
277        path: &Path,
278        maximum_bytes: usize,
279        before_read: impl FnOnce() -> io::Result<()>,
280    ) -> Result<Option<Vec<u8>>, BoundedRegularFileReadError> {
281        read_optional_regular_bytes_bounded_inner(path, maximum_bytes, before_read)
282    }
283
284    #[cfg(not(test))]
285    fn read_optional_regular_bytes_bounded_with_hook(
286        path: &Path,
287        maximum_bytes: usize,
288        before_read: impl FnOnce() -> io::Result<()>,
289    ) -> Result<Option<Vec<u8>>, BoundedRegularFileReadError> {
290        read_optional_regular_bytes_bounded_inner(path, maximum_bytes, before_read)
291    }
292
293    fn read_optional_regular_bytes_bounded_inner(
294        path: &Path,
295        maximum_bytes: usize,
296        before_read: impl FnOnce() -> io::Result<()>,
297    ) -> Result<Option<Vec<u8>>, BoundedRegularFileReadError> {
298        let Some((mut file, observed_size)) =
299            open_optional_regular_file(path).map_err(BoundedRegularFileReadError::Read)?
300        else {
301            return Ok(None);
302        };
303        if observed_size > u64::try_from(maximum_bytes).unwrap_or(u64::MAX) {
304            return Err(BoundedRegularFileReadError::TooLarge);
305        }
306        before_read().map_err(|source| {
307            BoundedRegularFileReadError::Read(RegularFileReadError::Io(source))
308        })?;
309
310        let read_limit = maximum_bytes
311            .checked_add(1)
312            .ok_or(BoundedRegularFileReadError::TooLarge)?;
313        let mut bytes = Vec::with_capacity(
314            usize::try_from(observed_size)
315                .unwrap_or(maximum_bytes)
316                .min(maximum_bytes),
317        );
318        Read::by_ref(&mut file)
319            .take(u64::try_from(read_limit).unwrap_or(u64::MAX))
320            .read_to_end(&mut bytes)
321            .map_err(RegularFileReadError::Io)
322            .map_err(BoundedRegularFileReadError::Read)?;
323        if bytes.len() > maximum_bytes {
324            return Err(BoundedRegularFileReadError::TooLarge);
325        }
326        Ok(Some(bytes))
327    }
328
329    fn open_optional_regular_file(
330        path: &Path,
331    ) -> Result<Option<(fs::File, u64)>, RegularFileReadError> {
332        let metadata = match fs::symlink_metadata(path) {
333            Ok(metadata) => metadata,
334            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
335            Err(error) => return Err(RegularFileReadError::Io(error)),
336        };
337        if !metadata.file_type().is_file() {
338            return Err(RegularFileReadError::NotRegular);
339        }
340
341        let fd: OwnedFd = unix_fs::open(
342            path,
343            OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::NONBLOCK | OFlags::CLOEXEC,
344            Mode::empty(),
345        )
346        .map_err(errno_to_io)
347        .map_err(RegularFileReadError::Io)?;
348        let metadata = unix_fs::fstat(&fd)
349            .map_err(errno_to_io)
350            .map_err(RegularFileReadError::Io)?;
351        if unix_fs::FileType::from_raw_mode(metadata.st_mode) != unix_fs::FileType::RegularFile {
352            return Err(RegularFileReadError::NotRegular);
353        }
354
355        let size = u64::try_from(metadata.st_size).map_err(|_| {
356            RegularFileReadError::Io(io::Error::new(
357                io::ErrorKind::InvalidData,
358                "regular file has a negative size",
359            ))
360        })?;
361        Ok(Some((fs::File::from(fd), size)))
362    }
363
364    #[derive(Clone, Copy, Debug, Eq, PartialEq)]
365    pub(super) enum FileCommitStep {
366        ParentDirectoryCreate,
367        CreatedDirectorySync,
368        CreatedDirectoryParentSync,
369        TemporaryFileCreate,
370        TemporaryFileWrite,
371        TemporaryFileSync,
372        Publication,
373        FinalParentSync,
374    }
375
376    pub(super) fn commit_with_hook(
377        path: &Path,
378        bytes: &[u8],
379        mode: FileCommitMode,
380        mut before: impl FnMut(FileCommitStep, &Path) -> io::Result<()>,
381    ) -> io::Result<()> {
382        let (parent, file_name) = split_target(path)?;
383        if matches!(
384            mode,
385            FileCommitMode::Replace | FileCommitMode::CreateNewWithParents
386        ) {
387            create_parent_hierarchy(parent, &mut before)?;
388        }
389        let parent_fd = open_directory(parent)?;
390        let (temp_name, temp_path, mut temp_file) =
391            create_sibling_temp(&parent_fd, parent, file_name, &mut before)?;
392
393        let staged = (|| {
394            before(FileCommitStep::TemporaryFileWrite, &temp_path)?;
395            temp_file.write_all(bytes)?;
396            before(FileCommitStep::TemporaryFileSync, &temp_path)?;
397            temp_file.sync_all()
398        })();
399        drop(temp_file);
400        if let Err(error) = staged {
401            remove_temp(&parent_fd, &temp_name);
402            return Err(error);
403        }
404
405        if let Err(error) = before(FileCommitStep::Publication, path) {
406            remove_temp(&parent_fd, &temp_name);
407            return Err(error);
408        }
409        let published = match mode {
410            FileCommitMode::Replace => {
411                unix_fs::renameat(&parent_fd, &temp_name, &parent_fd, file_name)
412            }
413            FileCommitMode::CreateNew | FileCommitMode::CreateNewWithParents => {
414                publish_create_new(&parent_fd, &temp_name, file_name)
415            }
416        };
417        if let Err(error) = published {
418            remove_temp(&parent_fd, &temp_name);
419            return Err(errno_to_io(error));
420        }
421
422        before(FileCommitStep::FinalParentSync, parent)?;
423        unix_fs::fsync(&parent_fd).map_err(errno_to_io)
424    }
425
426    fn publish_create_new(
427        parent_fd: &OwnedFd,
428        temp_name: &OsStr,
429        file_name: &OsStr,
430    ) -> rustix::io::Result<()> {
431        let renamed = unix_fs::renameat_with(
432            parent_fd,
433            temp_name,
434            parent_fd,
435            file_name,
436            RenameFlags::NOREPLACE,
437        );
438        finish_create_new_publication(parent_fd, temp_name, file_name, renamed)
439    }
440
441    fn finish_create_new_publication(
442        parent_fd: &OwnedFd,
443        temp_name: &OsStr,
444        file_name: &OsStr,
445        renamed: rustix::io::Result<()>,
446    ) -> rustix::io::Result<()> {
447        match renamed {
448            Ok(()) => Ok(()),
449            Err(
450                rustix::io::Errno::INVAL | rustix::io::Errno::NOSYS | rustix::io::Errno::OPNOTSUPP,
451            ) => {
452                unix_fs::linkat(parent_fd, temp_name, parent_fd, file_name, AtFlags::empty())?;
453                remove_temp(parent_fd, temp_name);
454                Ok(())
455            }
456            Err(error) => Err(error),
457        }
458    }
459
460    #[cfg(test)]
461    pub(super) fn publish_create_new_after_error(
462        parent: &Path,
463        temp_name: &OsStr,
464        file_name: &OsStr,
465        error: rustix::io::Errno,
466    ) -> io::Result<()> {
467        let parent_fd = open_directory(parent)?;
468        finish_create_new_publication(&parent_fd, temp_name, file_name, Err(error))
469            .map_err(errno_to_io)
470    }
471
472    fn split_target(path: &Path) -> io::Result<(&Path, &OsStr)> {
473        let file_name = path.file_name().ok_or_else(|| {
474            io::Error::new(
475                io::ErrorKind::InvalidInput,
476                format!("durable write target has no file name: {}", path.display()),
477            )
478        })?;
479        let parent = path
480            .parent()
481            .filter(|parent| !parent.as_os_str().is_empty())
482            .unwrap_or_else(|| Path::new("."));
483        Ok((parent, file_name))
484    }
485
486    fn create_parent_hierarchy(
487        parent: &Path,
488        before: &mut impl FnMut(FileCommitStep, &Path) -> io::Result<()>,
489    ) -> io::Result<()> {
490        let mut missing = Vec::new();
491        let mut current = parent;
492        loop {
493            match fs::symlink_metadata(current) {
494                Ok(metadata) if metadata.is_dir() => break,
495                Ok(_) => {
496                    return Err(io::Error::new(
497                        io::ErrorKind::NotADirectory,
498                        format!("output parent is not a directory: {}", current.display()),
499                    ));
500                }
501                Err(error) if error.kind() == io::ErrorKind::NotFound => {
502                    missing.push(current.to_path_buf());
503                    current = current
504                        .parent()
505                        .filter(|ancestor| !ancestor.as_os_str().is_empty())
506                        .unwrap_or_else(|| Path::new("."));
507                }
508                Err(error) => return Err(error),
509            }
510        }
511
512        for directory in missing.into_iter().rev() {
513            before(FileCommitStep::ParentDirectoryCreate, &directory)?;
514            match fs::create_dir(&directory) {
515                Ok(()) => sync_created_directory(&directory, before)?,
516                Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
517                    if !fs::symlink_metadata(&directory)?.is_dir() {
518                        return Err(io::Error::new(
519                            io::ErrorKind::NotADirectory,
520                            format!("output parent is not a directory: {}", directory.display()),
521                        ));
522                    }
523                }
524                Err(error) => return Err(error),
525            }
526        }
527        Ok(())
528    }
529
530    fn sync_created_directory(
531        directory: &Path,
532        before: &mut impl FnMut(FileCommitStep, &Path) -> io::Result<()>,
533    ) -> io::Result<()> {
534        before(FileCommitStep::CreatedDirectorySync, directory)?;
535        let directory_fd = open_directory(directory)?;
536        unix_fs::fsync(&directory_fd).map_err(errno_to_io)?;
537
538        let owner = directory
539            .parent()
540            .filter(|parent| !parent.as_os_str().is_empty())
541            .unwrap_or_else(|| Path::new("."));
542        before(FileCommitStep::CreatedDirectoryParentSync, owner)?;
543        let owner_fd = open_directory(owner)?;
544        unix_fs::fsync(&owner_fd).map_err(errno_to_io)
545    }
546
547    fn create_sibling_temp(
548        parent_fd: &impl AsFd,
549        parent: &Path,
550        file_name: &OsStr,
551        before: &mut impl FnMut(FileCommitStep, &Path) -> io::Result<()>,
552    ) -> io::Result<(OsString, PathBuf, fs::File)> {
553        for _ in 0..TEMP_ATTEMPTS {
554            let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
555            let mut temp_name = OsString::from(".");
556            temp_name.push(file_name);
557            temp_name.push(format!(".canic-tmp-{}-{sequence}", std::process::id()));
558            let temp_path = parent.join(&temp_name);
559            before(FileCommitStep::TemporaryFileCreate, &temp_path)?;
560            match unix_fs::openat(
561                parent_fd,
562                &temp_name,
563                OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC,
564                Mode::from_raw_mode(0o666),
565            ) {
566                Ok(file) => return Ok((temp_name, temp_path, fs::File::from(file))),
567                Err(error) if error == rustix::io::Errno::EXIST => {}
568                Err(error) => return Err(errno_to_io(error)),
569            }
570        }
571
572        Err(io::Error::new(
573            io::ErrorKind::AlreadyExists,
574            format!(
575                "could not allocate a unique sibling temporary file for {}",
576                parent.join(file_name).display()
577            ),
578        ))
579    }
580
581    fn open_directory(path: &Path) -> io::Result<OwnedFd> {
582        unix_fs::open(
583            path,
584            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
585            Mode::empty(),
586        )
587        .map_err(errno_to_io)
588    }
589
590    fn remove_temp(parent_fd: &impl AsFd, temp_name: &OsStr) {
591        let _ = unix_fs::unlinkat(parent_fd, temp_name, AtFlags::empty());
592    }
593
594    fn errno_to_io(error: rustix::io::Errno) -> io::Error {
595        io::Error::from_raw_os_error(error.raw_os_error())
596    }
597}
canic_host/durable_io/mod.rs

canic_host/durable_io/
mod.rs
