mesh/SPEC.md

@@ -137,6 +137,20 @@ journal is not one the steward produced, and the steward MUST NOT serve from it.
 Replayed presentations count as spent, and the receipt chain continues from the
 last `Sealed` entry (§6.3).
 
+**One steward per journal.** A steward MUST take exclusive access to the
+journal when it opens it, and MUST fail to start rather than serve where
+another holds it — or where the platform cannot provide exclusivity at all.
+
+The chain above **detects** a second writer; it does not prevent one. Two
+stewards sharing a journal each replay it, each re-establish the same spend and
+each serve against it, so the grant is overspent *while both are running*.
+Their records then collide on `seq`, the chain no longer forms, and the rule
+above requires the steward not to serve from the journal at all — so the
+detection arrives after the overspend and costs the steward its ability to
+restart. An exclusive claim the operating system releases when the holding
+process dies is preferable to one the steward must clean up, because a crash
+then leaves nothing to recover.
+
 An authorisation with no debit is a serve that was in flight when the steward
 stopped, and whether the engine finished is not knowable from the journal. The
 steward MUST NOT retry it or report it as completed or failed. It stays debited,

mesh/mesh-rs/crates/mesh-core/src/journal.rs

@@ -1,7 +1,7 @@
 //! The steward's decision journal (SPEC §2.2): where a grant's spend at this
 //! steward, and the steward's receipt chain, survive the process (JCP §4.11).
 
-use std::fs::{File, OpenOptions};
+use std::fs::{File, OpenOptions, TryLockError};
 use std::io::{self, Read, Write};
 use std::path::Path;
 
@@ -118,6 +118,11 @@ impl FileJournal {
     /// records it already holds, for
     /// [`Mesh::with_journal`](crate::Mesh::with_journal) to replay.
     ///
+    /// Takes an **exclusive** lock first, so a second steward cannot hold the
+    /// same journal. Fails with [`io::ErrorKind::WouldBlock`] where the
+    /// journal is already held, and fails rather than proceeding where the
+    /// platform cannot provide a lock at all.
+    ///
     /// A final line with no newline is an append the steward was interrupted
     /// in. It is dropped and cut from the file, which is safe for every entry:
     /// an authorisation never recorded never started its serve, a debit never
@@ -126,6 +131,10 @@ impl FileJournal {
     /// because a journal that cannot be read cannot re-establish spend.
     pub fn open(path: impl AsRef<Path>) -> io::Result<(Self, Vec<JournalRecord>)> {
         let mut file = OpenOptions::new().read(true).append(true).create(true).open(path)?;
+        // Before the read, and before the truncation below: a would-be second
+        // steward must be refused *before* it can cut a torn trailing line out
+        // from under the holder's appends.
+        lock_exclusively(&file)?;
         let mut bytes = Vec::new();
         file.read_to_end(&mut bytes)?;
         let complete = bytes.iter().rposition(|&b| b == b'\n').map_or(0, |i| i + 1);
@@ -147,6 +156,48 @@ impl FileJournal {
     }
 }
 
+/// Take an exclusive lock on the journal file, so a second steward cannot hold
+/// it.
+///
+/// **The chain detects a second writer; it does not prevent one.** Two
+/// stewards on one journal each replay it, each derive the same spend and each
+/// serve against it, so the budget is overspent *in the world* while both run.
+/// Afterwards their records collide on `seq`, the chain no longer forms and
+/// [`Mesh::with_journal`](crate::Mesh::with_journal) refuses the journal
+/// entirely — so the detection costs the steward its ability to restart,
+/// having already failed to stop the overspend. [`FileJournal::open`] also
+/// truncates a torn trailing line, so a concurrent opener can cut the file out
+/// from under an appender.
+///
+/// The lock is the operating system's, held by the open file and released when
+/// it closes **or the process dies**, so a crash leaves nothing to recover. A
+/// lock-file sidecar would need a story for the crash case, and that story is
+/// where such schemes go wrong.
+fn lock_exclusively(file: &File) -> io::Result<()> {
+    lock_outcome(file.try_lock())
+}
+
+/// Turn a lock attempt into an outcome for [`FileJournal::open`].
+///
+/// Separate from the I/O so every arm is testable. The
+/// [`TryLockError::Error`] arm cannot be reached on a platform whose locks
+/// work, so on this one it is unreachable through `open` — and an unreachable
+/// arm is exactly where a fail-**open** would sit unnoticed. Falsification
+/// found it: swallowing that error broke no test until this function existed.
+fn lock_outcome(attempt: Result<(), TryLockError>) -> io::Result<()> {
+    match attempt {
+        Ok(()) => Ok(()),
+        Err(TryLockError::WouldBlock) => Err(io::Error::new(
+            io::ErrorKind::WouldBlock,
+            "journal already held by another steward: a second one would serve against the same spend",
+        )),
+        // The platform could not provide exclusivity. Failing to open is the
+        // safe direction, and the only honest one: the alternative is serving
+        // against a record another steward may also be writing.
+        Err(TryLockError::Error(e)) => Err(e),
+    }
+}
+
 impl Journal for FileJournal {
     fn append(&mut self, record: &JournalRecord) -> io::Result<()> {
         let mut line = serde_json::to_vec(record)?;
@@ -155,3 +206,112 @@ impl Journal for FileJournal {
         self.file.sync_data()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn tmp_path(tag: &str) -> std::path::PathBuf {
+        static N: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
+        let n = N.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        let p = std::env::temp_dir()
+            .join(format!("mesh-core-journal-{}-{tag}-{n}.log", std::process::id()));
+        let _ = std::fs::remove_file(&p);
+        p
+    }
+
+    /// A second steward must not be able to hold the journal.
+    ///
+    /// `std::fs::File::try_lock` is the operating system's lock, so this needs
+    /// no platform caveat and no `unsafe`: where a platform cannot provide one
+    /// at all, `open` fails rather than proceeding.
+    #[test]
+    fn the_single_writer_claim_matches_what_open_does() {
+        let path = tmp_path("second-open");
+        let (_held, records) = FileJournal::open(&path).expect("the first open");
+        assert!(records.is_empty());
+
+        let err = FileJournal::open(&path)
+            .err()
+            .expect("a second open of a held journal must be refused");
+        assert_eq!(err.kind(), io::ErrorKind::WouldBlock, "{err}");
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// Every arm of the lock policy, including the one `open` cannot reach on
+    /// a platform whose locks work.
+    ///
+    /// A platform that cannot lock **MUST NOT** yield an open journal: the
+    /// alternative is serving against a record another steward may also be
+    /// writing, which is the overspend this whole mechanism exists to stop.
+    #[test]
+    fn a_platform_that_cannot_lock_fails_closed() {
+        assert!(lock_outcome(Ok(())).is_ok(), "an acquired lock opens the journal");
+
+        let contended = lock_outcome(Err(TryLockError::WouldBlock))
+            .expect_err("a contended lock must not open the journal");
+        assert_eq!(contended.kind(), io::ErrorKind::WouldBlock);
+
+        let unsupported =
+            lock_outcome(Err(TryLockError::Error(io::Error::from(io::ErrorKind::Unsupported))))
+                .expect_err("an unsupported lock must not open the journal");
+        assert_eq!(
+            unsupported.kind(),
+            io::ErrorKind::Unsupported,
+            "the platform's own error is propagated, not replaced"
+        );
+    }
+
+    /// The lock is the kernel's, held by the descriptor, so dropping the
+    /// journal frees it — and a crash frees it the same way, which is why this
+    /// is an advisory lock rather than a lock file.
+    #[test]
+    fn dropping_the_journal_releases_the_lock() {
+        let path = tmp_path("release");
+        {
+            let (_held, _) = FileJournal::open(&path).expect("the first open");
+        }
+        let (_again, records) = FileJournal::open(&path).expect("reopen after drop");
+        assert!(records.is_empty());
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// **The lock is taken before the truncation.** `open` cuts a trailing line
+    /// with no newline, so an opener refused *afterwards* would already have
+    /// cut the holder's file. A refused open must leave it byte-for-byte.
+    ///
+    /// Raw bytes rather than real records: the property under test is the
+    /// ordering of the lock against the truncation, and a torn line is all
+    /// that takes.
+    #[test]
+    fn a_refused_open_does_not_truncate_the_holders_journal() {
+        let path = tmp_path("no-truncate");
+        let (held, _) = FileJournal::open(&path).expect("the first open");
+
+        // A torn line appended behind the holder's back: no trailing newline,
+        // which `open` cuts away.
+        {
+            use std::io::Write;
+            let mut raw = OpenOptions::new().append(true).open(&path).expect("raw append");
+            raw.write_all(b"{\"partial\"").expect("torn line");
+            raw.sync_data().expect("sync");
+        }
+        let before = std::fs::metadata(&path).expect("stat").len();
+        assert!(before > 0, "the torn line is on disk");
+
+        assert!(FileJournal::open(&path).is_err(), "the second open must be refused");
+        assert_eq!(
+            std::fs::metadata(&path).expect("stat").len(),
+            before,
+            "a refused open truncated the file it was refused from"
+        );
+        drop(held);
+
+        // Once the holder is gone the torn line is cut, as it always was — the
+        // truncation still happens, just never from a refused writer.
+        let (_now, records) = FileJournal::open(&path).expect("reopen");
+        assert!(records.is_empty(), "the torn line yields no record");
+        assert_eq!(std::fs::metadata(&path).expect("stat").len(), 0, "and is cut");
+        let _ = std::fs::remove_file(&path);
+    }
+}