gatehouse_runtime_protocol/lifecycle/journal/
storage.rs

1//! Descriptor-relative Unix persistence for the new lifecycle journal only.
2//! Cooperative same-user locking, not protection from a hostile same-UID writer.
3use super::*;
4use crate::lifecycle::canonical;
5use std::ffi::CString;
6use std::fs::{self, File, OpenOptions};
7use std::io::{Read, Write};
8use std::os::fd::{AsRawFd, FromRawFd};
9use std::os::unix::fs::{MetadataExt, OpenOptionsExt};
10use std::path::{Path, PathBuf};
11use std::sync::atomic::{AtomicU64, Ordering};
12
13fn io<T>(r: std::io::Result<T>) -> Result<T> {
14    r.map_err(|_| Invalid("lifecycle journal I/O failure; retain state"))
15}
16fn private(m: &fs::Metadata, directory: bool) -> Result<()> {
17    // SAFETY: geteuid has no pointer arguments or preconditions.
18    if m.uid() != unsafe { libc::geteuid() }
19        || m.mode() & 0o777 != if directory { 0o700 } else { 0o600 }
20        || if directory {
21            !m.is_dir()
22        } else {
23            !m.is_file() || m.nlink() != 1
24        }
25    {
26        return Err(Invalid("lifecycle journal must be private"));
27    }
28    Ok(())
29}
30
31/// Explicitly revoke the owner's lock before descriptor close. CLOEXEC alone
32/// does not prevent an incidental fork child from prolonging a flock lifetime.
33struct LockedDirectory {
34    file: File,
35    owner_pid: libc::pid_t,
36}
37impl std::ops::Deref for LockedDirectory {
38    type Target = File;
39    fn deref(&self) -> &File {
40        &self.file
41    }
42}
43impl Drop for LockedDirectory {
44    fn drop(&mut self) {
45        // SAFETY: live owned directory descriptor. This ends the parent's guard,
46        // including incidental copies inherited by children before exec.
47        if unsafe { libc::getpid() } != self.owner_pid {
48            return;
49        }
50        loop {
51            let result = unsafe { libc::flock(self.file.as_raw_fd(), libc::LOCK_UN) };
52            if result == 0
53                || std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted
54            {
55                break;
56            }
57        }
58    }
59}
60
61/// Holds a nonblocking exclusive resource-directory lock. Dispatch is unavailable
62/// after reopen until a new request has been durably prepared. Directory must be
63/// provisioned by Core; this store never discovers or migrates another lease.
64pub struct Journal {
65    root: PathBuf,
66    directory: LockedDirectory,
67    record: Record,
68    bytes: Vec<u8>,
69    poisoned: bool,
70    ready: bool,
71    fresh_status: bool,
72}
73
74/// Single-use, non-Clone, borrowed from a held journal. The future installed host
75/// must accept this token, not a free Request. No engine execution exists here.
76///
77/// ```compile_fail
78/// let forged = gatehouse_runtime_protocol::lifecycle::journal::Dispatch { journal: unimplemented!() };
79/// ```
80/// ```compile_fail
81/// fn duplicate(token: gatehouse_runtime_protocol::lifecycle::journal::Dispatch<'_>) {
82///     let _ = token.clone();
83/// }
84/// ```
85#[must_use]
86pub struct Dispatch<'a> {
87    journal: &'a mut Journal,
88}
89impl Dispatch<'_> {
90    pub fn request(&self) -> &Request {
91        &self
92            .journal
93            .record
94            .pending
95            .as_ref()
96            .expect("prepared token")
97            .request
98    }
99    /// Call only after host exit/transport checks. None means an uncertain result.
100    pub fn observe(self, bytes: Option<&[u8]>) -> Result<()> {
101        let next = match self.journal.record.observe(bytes) {
102            Ok(next) => next,
103            Err(e) => {
104                self.journal.poisoned = true;
105                return Err(e);
106            }
107        };
108        let fresh = matches!(
109            next.attempts
110                .last()
111                .and_then(|a| a.response.as_ref())
112                .map(|r| (&r.operation, &r.outcome)),
113            Some((Operation::Status { .. }, Outcome::Status { .. }))
114        );
115        self.journal.commit(next)?;
116        self.journal.fresh_status = fresh;
117        Ok(())
118    }
119}
120impl Journal {
121    fn directory(root: &Path) -> Result<LockedDirectory> {
122        if !root.is_absolute() || io(fs::canonicalize(root))? != root {
123            return Err(Invalid("journal path must be canonical"));
124        }
125        // Ancestors must not be writable by another principal; ownership may be
126        // root or current UID. This also rejects /tmp and symlink ancestors.
127        for parent in root.ancestors().skip(1) {
128            let m = io(fs::symlink_metadata(parent))?;
129            if !m.is_dir()
130                || m.mode() & 0o022 != 0
131                || (m.uid() != 0 && m.uid() != unsafe { libc::geteuid() })
132            {
133                return Err(Invalid("unprotected journal ancestor"));
134            }
135        }
136        let directory = io(OpenOptions::new()
137            .read(true)
138            .custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW | libc::O_CLOEXEC)
139            .open(root))?;
140        private(&io(directory.metadata())?, true)?;
141        // SAFETY: the file owns a live directory descriptor for the guard lifetime.
142        if unsafe { libc::flock(directory.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 {
143            return Err(Invalid("lifecycle resource is locked"));
144        }
145        Ok(LockedDirectory {
146            file: directory,
147            owner_pid: unsafe { libc::getpid() },
148        })
149    }
150    fn at(directory: &File, name: &str, flags: i32) -> Result<Option<File>> {
151        let name = CString::new(name).map_err(|_| Invalid("invalid journal member"))?;
152        // SAFETY: NUL-terminated fixed internal name, valid directory descriptor.
153        let fd = unsafe {
154            libc::openat(
155                directory.as_raw_fd(),
156                name.as_ptr(),
157                flags | libc::O_NOFOLLOW | libc::O_CLOEXEC | libc::O_NONBLOCK,
158                0o600,
159            )
160        };
161        if fd < 0 {
162            let err = std::io::Error::last_os_error();
163            if err.kind() == std::io::ErrorKind::NotFound && flags & libc::O_CREAT == 0 {
164                return Ok(None);
165            }
166            return Err(Invalid("cannot open lifecycle journal member"));
167        }
168        // SAFETY: successful openat returns an owned descriptor.
169        let f = unsafe { File::from_raw_fd(fd) };
170        private(&io(f.metadata())?, false)?;
171        Ok(Some(f))
172    }
173    fn read(directory: &File, name: &str) -> Result<Option<Vec<u8>>> {
174        let Some(f) = Self::at(directory, name, libc::O_RDONLY)? else {
175            return Ok(None);
176        };
177        let mut b = vec![];
178        io(f.take((crate::MAX_MESSAGE_BYTES + 1) as u64)
179            .read_to_end(&mut b))?;
180        if b.len() > crate::MAX_MESSAGE_BYTES {
181            return Err(Invalid("journal member too large"));
182        }
183        Ok(Some(b))
184    }
185    fn check(&self) -> Result<()> {
186        if self.poisoned {
187            return Err(Invalid("journal poisoned; reopen required"));
188        }
189        let now = io(fs::symlink_metadata(&self.root))?;
190        let held = io(self.directory.metadata())?;
191        private(&now, true)?;
192        if now.dev() != held.dev() || now.ino() != held.ino() {
193            return Err(Invalid("journal directory replaced"));
194        }
195        if Self::read(&self.directory, "lease.json")?.as_deref() != Some(self.bytes.as_slice()) {
196            return Err(Invalid("journal changed outside lock"));
197        }
198        if self.record.state.phase() != Phase::VerifiedAbsent
199            && Self::read(&self.directory, "active")?.as_deref()
200                != Some(self.record.marker().as_bytes())
201        {
202            return Err(Invalid("active lifecycle marker changed"));
203        }
204        Ok(())
205    }
206    fn encode(record: &Record) -> Result<Vec<u8>> {
207        record.validate()?;
208        canonical::encode(
209            &serde_json::to_value(record).map_err(|_| Invalid("journal encoding failed"))?,
210        )
211    }
212    fn rename(directory: &File, from: &str, to: &str) -> Result<()> {
213        let from = CString::new(from).unwrap();
214        let to = CString::new(to).unwrap();
215        // SAFETY: both internal names and descriptor are valid for this call.
216        if unsafe {
217            libc::renameat(
218                directory.as_raw_fd(),
219                from.as_ptr(),
220                directory.as_raw_fd(),
221                to.as_ptr(),
222            )
223        } != 0
224        {
225            return Err(Invalid("journal publish failed"));
226        }
227        Ok(())
228    }
229    fn atomic(directory: &File, name: &str, bytes: &[u8]) -> Result<()> {
230        Self::atomic_with(directory, name, bytes, |_| Ok(()))
231    }
232    fn atomic_with(
233        directory: &File,
234        name: &str,
235        bytes: &[u8],
236        mut stage: impl FnMut(u8) -> Result<()>,
237    ) -> Result<()> {
238        static NEXT: AtomicU64 = AtomicU64::new(0);
239        let temp = format!(
240            "next-{}-{}",
241            std::process::id(),
242            NEXT.fetch_add(1, Ordering::Relaxed)
243        );
244        let mut f = Self::at(
245            directory,
246            &temp,
247            libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL,
248        )?
249        .ok_or(Invalid("missing journal temp"))?;
250        io(f.write_all(bytes))?;
251        stage(0)?;
252        io(f.sync_all())?;
253        stage(1)?;
254        Self::rename(directory, &temp, name)?;
255        stage(2)?;
256        io(directory.sync_all())?;
257        Ok(())
258    }
259    pub fn create(root: &Path, create: &Request) -> Result<Self> {
260        let directory = Self::directory(root)?;
261        let record = Record::new(create)?;
262        let bytes = Self::encode(&record)?;
263        for name in [
264            "lease.json",
265            "active",
266            "tombstone.json",
267            "reservation.json",
268            "reservation-tombstone.json",
269        ] {
270            if Self::read(&directory, name)?.is_some() {
271                return Err(Invalid("lifecycle resource already recorded"));
272            }
273        }
274        // Active marker first: a crash before lease persistence still blocks reuse.
275        let mut marker = Self::at(
276            &directory,
277            "active",
278            libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL,
279        )?
280        .ok_or(Invalid("missing active marker"))?;
281        io(marker.write_all(create.binding.operation_key.as_bytes()))?;
282        io(marker.sync_all())?;
283        io(directory.sync_all())?;
284        Self::atomic(&directory, "lease.json", &bytes)?;
285        Ok(Self {
286            root: root.into(),
287            directory,
288            record,
289            bytes,
290            poisoned: false,
291            ready: true,
292            fresh_status: false,
293        })
294    }
295    pub fn open(root: &Path, create: &Request) -> Result<Self> {
296        create.validate()?;
297        let directory = Self::directory(root)?;
298        Self::open_held(root, create, directory)
299    }
300    fn open_held(root: &Path, create: &Request, directory: LockedDirectory) -> Result<Self> {
301        let bytes = Self::read(&directory, "lease.json")?
302            .ok_or(Invalid("missing journal; no automatic adoption"))?;
303        let value = canonical::parse_canonical(&bytes)?;
304        let record: Record = serde_json::from_value(value)
305            .map_err(|_| Invalid("invalid lifecycle journal schema"))?;
306        if Self::encode(&record)? != bytes || record.create != *create {
307            return Err(Invalid("journal binding mismatch"));
308        }
309        let marker = Self::read(&directory, "active")?;
310        let tombstone = Self::read(&directory, "tombstone.json")?;
311        if let Some(t) = &tombstone {
312            if record.state.phase() != Phase::VerifiedAbsent || t != &bytes {
313                return Err(Invalid("journal tombstone mismatch"));
314            }
315        }
316        if let Some(m) = &marker {
317            if m != record.marker().as_bytes() {
318                return Err(Invalid("journal marker mismatch"));
319            }
320        } else if record.state.phase() != Phase::VerifiedAbsent
321            || tombstone.as_ref() != Some(&bytes)
322        {
323            return Err(Invalid("missing active marker without durable absence"));
324        }
325        let mut store = Self {
326            root: root.into(),
327            directory,
328            record,
329            bytes,
330            poisoned: false,
331            ready: false,
332            fresh_status: false,
333        };
334        let recovered = store.record.recover()?;
335        if recovered != store.record {
336            store.commit(recovered)?;
337        }
338        Ok(store)
339    }
340    /// Close a dropped/incomplete dispatch as unknown without replaying it.
341    /// The mutable borrow prevents use while a Dispatch token is still held.
342    pub fn retain_unknown(&mut self) -> Result<()> {
343        self.check()?;
344        let next = self.record.observe(None)?;
345        self.commit(next)
346    }
347    pub fn phase(&self) -> Phase {
348        self.record.state.phase()
349    }
350    pub fn revision(&self) -> u64 {
351        self.record.revision
352    }
353    pub fn target(&self) -> Result<Target> {
354        self.record.target()
355    }
356    fn commit(&mut self, next: Record) -> Result<()> {
357        self.commit_with(next, |_| Ok(()))
358    }
359    fn commit_with(&mut self, next: Record, stage: impl FnMut(u8) -> Result<()>) -> Result<()> {
360        self.check()?;
361        self.poisoned = true;
362        self.ready = false;
363        self.fresh_status = false;
364        let bytes = Self::encode(&next)?;
365        Self::atomic_with(&self.directory, "lease.json", &bytes, stage)?;
366        self.record = next;
367        self.bytes = bytes;
368        self.poisoned = false;
369        Ok(())
370    }
371    pub fn begin(&mut self, request: &Request, purpose: Purpose) -> Result<Dispatch<'_>> {
372        self.check()?;
373        let next = self.record.begin(request, purpose, self.fresh_status)?;
374        self.commit(next)?;
375        self.ready = true;
376        self.dispatch()
377    }
378    /// Only the initial just-created record can use this without begin().
379    /// Reopened records never have ready=true; borrowing then dropping a token
380    /// cannot authorize a second dispatch.
381    pub fn dispatch(&mut self) -> Result<Dispatch<'_>> {
382        self.check()?;
383        if !self.ready || self.record.pending.is_none() {
384            return Err(Invalid("no fresh durable dispatch authorization"));
385        }
386        self.ready = false;
387        Ok(Dispatch { journal: self })
388    }
389    fn persist_tombstone(&self) -> Result<()> {
390        match Self::read(&self.directory, "tombstone.json")? {
391            Some(b) if b != self.bytes => return Err(Invalid("conflicting tombstone")),
392            Some(_) => {}
393            None => Self::atomic(&self.directory, "tombstone.json", &self.bytes)?,
394        }
395        Ok(())
396    }
397    /// Evidence is already committed in the verified-absent record. Write its
398    /// tombstone durably before unlinking active, then fsync the directory. This
399    /// method is idempotent after a crash; it never removes the lease/evidence.
400    pub fn finish_release(&mut self) -> Result<()> {
401        self.check()?;
402        if self.record.state.phase() != Phase::VerifiedAbsent {
403            return Err(Invalid("absence evidence required before release"));
404        }
405        self.poisoned = true;
406        self.persist_tombstone()?;
407        let name = CString::new("active").unwrap();
408        // SAFETY: fixed name, live directory descriptor; never a container path.
409        if unsafe { libc::unlinkat(self.directory.as_raw_fd(), name.as_ptr(), 0) } != 0
410            && std::io::Error::last_os_error().kind() != std::io::ErrorKind::NotFound
411        {
412            return Err(Invalid("cannot remove lifecycle marker"));
413        }
414        io(self.directory.sync_all())?;
415        self.poisoned = false;
416        Ok(())
417    }
418}
419
420#[cfg(test)]
421mod tests;
422
423mod reservation;
424pub use reservation::{Reservation, ResolutionDispatch};
425
426mod slot;
427pub use slot::{OperatorRecovery, ServiceSlot};
gatehouse_runtime_protocol/lifecycle/journal/storage.rs

gatehouse_runtime_protocol/lifecycle/journal/
storage.rs
