kunobi_daemon/
publication.rs

1//! Atomic publication of discovery records in an application-owned directory.
2
3use std::{
4    ffi::OsString,
5    fs::{File, OpenOptions},
6    io::{self, Read, Write},
7    path::{Path, PathBuf},
8    sync::atomic::{AtomicU64, Ordering},
9};
10
11static NEXT: AtomicU64 = AtomicU64::new(0);
12
13struct Temporary(PathBuf);
14impl Drop for Temporary {
15    fn drop(&mut self) {
16        let _ = std::fs::remove_file(&self.0);
17    }
18}
19
20/// Replace a discovery record atomically, after syncing its complete contents.
21///
22/// The parent directory must already exist and belong to the application.
23/// Temporary files stay on the same filesystem, are created exclusively, and
24/// are removed on failure. On Unix their permissions are set to 0600 before any
25/// contents are written. The caller remains responsible for directory security,
26/// record format, ownership and eventual discovery repair after a crash. The
27/// directory is not synced; this does not promise survival across power loss.
28pub fn publish_record(path: &Path, contents: &[u8]) -> io::Result<()> {
29    let parent = path
30        .parent()
31        .filter(|p| !p.as_os_str().is_empty())
32        .unwrap_or_else(|| Path::new("."));
33    let filename = path.file_name().ok_or_else(|| {
34        io::Error::new(
35            io::ErrorKind::InvalidInput,
36            "discovery path needs a filename",
37        )
38    })?;
39    let (temporary, mut file): (Temporary, File) = loop {
40        let mut name = OsString::from(".");
41        name.push(filename);
42        name.push(format!(
43            ".{}.{}.tmp",
44            std::process::id(),
45            NEXT.fetch_add(1, Ordering::Relaxed)
46        ));
47        let temporary = parent.join(name);
48        let mut options = OpenOptions::new();
49        options.create_new(true).write(true);
50        #[cfg(unix)]
51        {
52            use std::os::unix::fs::OpenOptionsExt;
53            options.mode(0o600);
54        }
55        match options.open(&temporary) {
56            Ok(file) => break (Temporary(temporary), file),
57            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
58            Err(e) => return Err(e),
59        }
60    };
61    #[cfg(unix)]
62    {
63        use std::os::unix::fs::PermissionsExt;
64        file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
65    }
66    file.write_all(contents)?;
67    file.sync_all()?;
68    drop(file);
69    std::fs::rename(&temporary.0, path)
70}
71
72/// A discovery slot whose publishers and conditional cleanup share a lock.
73/// Use the same slot path in every generation. Lock files remain persistent.
74#[derive(Clone, Debug)]
75pub struct RecordSlot(PathBuf);
76impl RecordSlot {
77    /// Address an existing application-owned directory. This performs no I/O.
78    pub fn new(path: impl Into<PathBuf>) -> Self {
79        Self(path.into())
80    }
81
82    fn lock(&self) -> io::Result<crate::ProcessLock> {
83        let mut lock = self.0.as_os_str().to_owned();
84        lock.push(".record.lock");
85        crate::readiness::wait_until(
86            std::time::Instant::now() + std::time::Duration::from_secs(2),
87            |_| crate::ProcessLock::try_acquire(Path::new(&lock)),
88        )?
89        .ok_or_else(|| io::ErrorKind::TimedOut.into())
90    }
91
92    /// Atomically publish while excluding conditional cleanup by older owners.
93    pub fn replace(&self, contents: &[u8]) -> io::Result<()> {
94        let _lock = self.lock()?;
95        publish_record(&self.0, contents)
96    }
97
98    /// Restore an owned record only if its contents differ. The caller must
99    /// still be authorized to advertise this endpoint; a lock does not elect it.
100    pub fn repair(&self, contents: &[u8]) -> io::Result<bool> {
101        let _lock = self.lock()?;
102        if self.matches(contents)? {
103            return Ok(false);
104        }
105        publish_record(&self.0, contents)?;
106        Ok(true)
107    }
108
109    /// Remove only the exact record still belonging to this owner. A successor's
110    /// publication cannot race between the comparison and unlink.
111    pub fn remove_if_matches(&self, contents: &[u8]) -> io::Result<bool> {
112        let _lock = self.lock()?;
113        if !self.matches(contents)? {
114            return Ok(false);
115        }
116        std::fs::remove_file(&self.0)?;
117        Ok(true)
118    }
119
120    fn matches(&self, contents: &[u8]) -> io::Result<bool> {
121        let file = match File::open(&self.0) {
122            Ok(file) => file,
123            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
124            Err(error) => return Err(error),
125        };
126        let mut current = Vec::new();
127        file.take(contents.len().saturating_add(1) as u64)
128            .read_to_end(&mut current)?;
129        Ok(current == contents)
130    }
131}
kunobi_daemon/publication.rs

kunobi_daemon/
publication.rs
