jcp/SPEC.md

@@ -1265,6 +1265,14 @@ action: the grant's holder is the enforcement point that ran it, as a steward is
 in Mesh, rather than the agent the grant governs reporting its own consumption.
 The endpoint behind the enforcement point is invoked only on Allow.
 
+**What the endpoint holds survives it.** An endpoint that can restart is bound
+by §4.11: it MUST record each request it admits before deciding it, and each
+revocation it honours before applying it, and on restart MUST replay them
+through the gate, each request at the time it was decided, before deciding
+anything new. That re-establishes the spend, the revocations, and the memory of
+requests still inside the window. A request it cannot record is denied
+`unrecorded`.
+
 ## 12. Non-goals
 
 JCP does not define a model or a tool registry. It does

jcp/jcp-rs/crates/jcp-schema/src/lib.rs

@@ -609,6 +609,10 @@ pub enum DenyReason {
     /// A wire request the enforcement point has already decided, presented
     /// again (SPEC §11). It is decided once, and debited once.
     Replayed,
+    /// The enforcement point could not record the request before deciding it
+    /// (SPEC §4.11), so it does not decide it: a decision it could not record
+    /// would be lost, with its debit, on a restart.
+    Unrecorded,
 }
 
 impl DenyReason {
@@ -632,6 +636,7 @@ impl DenyReason {
             DenyReason::TaintBudgetExceeded => "taint-budget-exceeded",
             DenyReason::StaleRequest => "stale-request",
             DenyReason::Replayed => "replayed",
+            DenyReason::Unrecorded => "unrecorded",
         }
     }
 }

jcp/jcp-rs/crates/jcp-wire/src/journal.rs

@@ -0,0 +1,108 @@
+//! The endpoint's journal (SPEC §4.11): every request it decided and every
+//! revocation it honoured, recorded before it acts, and replayed through the
+//! gate when it restarts, so a restart gives no budget back, unrevokes
+//! nothing, and forgets no request it may still be shown again.
+
+use std::fs::{File, OpenOptions};
+use std::io::{self, Read, Write};
+use std::path::Path;
+
+use jcp_schema::{canonical_cbor, from_canonical_cbor, Grant, SignedRevocation};
+use serde::{Deserialize, Serialize};
+
+use crate::WireRequest;
+
+/// One record in an endpoint's journal.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub enum JournalRecord {
+    /// A request the endpoint admitted, and the time it decided it at.
+    /// Written before the decision is made, so a decision the endpoint was
+    /// interrupted in is still on record and is debited on replay.
+    Decided { at: u64, request: WireRequest },
+    /// A revocation the endpoint honoured, with the grant it revokes.
+    Revoked { revocation: SignedRevocation, grant: Grant },
+}
+
+/// Where an endpoint records what it decided and revoked.
+pub trait Journal {
+    /// Record one record durably. An error refuses what it was for: a request
+    /// that could not be recorded is denied `unrecorded`, and a revocation
+    /// that could not be recorded is not applied.
+    fn append(&mut self, record: &JournalRecord) -> io::Result<()>;
+}
+
+/// A journal in memory, and the default. It is enough for an endpoint that
+/// never restarts; one that can restart MUST use a durable journal such as
+/// [`FileJournal`] (SPEC §4.11).
+#[derive(Clone, Debug, Default)]
+pub struct MemoryJournal {
+    records: Vec<JournalRecord>,
+}
+
+impl MemoryJournal {
+    /// The records so far, in order.
+    pub fn records(&self) -> &[JournalRecord] {
+        &self.records
+    }
+}
+
+impl Journal for MemoryJournal {
+    fn append(&mut self, record: &JournalRecord) -> io::Result<()> {
+        self.records.push(record.clone());
+        Ok(())
+    }
+}
+
+/// A journal in an append-only file: records framed as on the wire, a 4-byte
+/// big-endian length then canonical det-CBOR, each synced to disk before
+/// [`append`](Journal::append) returns.
+pub struct FileJournal {
+    file: File,
+}
+
+impl FileJournal {
+    /// Open the journal at `path`, creating it if absent. Returns it with the
+    /// records it already holds, for [`Endpoint::with_journal`](crate::Endpoint::with_journal)
+    /// to replay.
+    ///
+    /// A final record cut short is an append the endpoint was interrupted in.
+    /// It is dropped and cut from the file, which is safe for both kinds: a
+    /// request never recorded was never decided, and a revocation never
+    /// recorded was never applied. A complete record that does not decode is an
+    /// error, because a journal that cannot be read cannot re-establish state.
+    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)?;
+        let mut bytes = Vec::new();
+        file.read_to_end(&mut bytes)?;
+        let mut records = Vec::new();
+        let mut at = 0usize;
+        while bytes.len() - at >= 4 {
+            let len = u32::from_be_bytes([bytes[at], bytes[at + 1], bytes[at + 2], bytes[at + 3]]) as usize;
+            if bytes.len() - at - 4 < len {
+                break;
+            }
+            let body = &bytes[at + 4..at + 4 + len];
+            let record = from_canonical_cbor(body).map_err(|e| {
+                io::Error::new(io::ErrorKind::InvalidData, format!("journal record {}: {e}", records.len()))
+            })?;
+            records.push(record);
+            at += 4 + len;
+        }
+        if at < bytes.len() {
+            file.set_len(at as u64)?;
+            file.sync_data()?;
+        }
+        Ok((FileJournal { file }, records))
+    }
+}
+
+impl Journal for FileJournal {
+    fn append(&mut self, record: &JournalRecord) -> io::Result<()> {
+        let body = canonical_cbor(record).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?;
+        let mut framed = Vec::with_capacity(4 + body.len());
+        framed.extend_from_slice(&(body.len() as u32).to_be_bytes());
+        framed.extend_from_slice(&body);
+        self.file.write_all(&framed)?;
+        self.file.sync_data()
+    }
+}

jcp/jcp-rs/crates/jcp-wire/src/lib.rs

@@ -25,6 +25,15 @@
 //! requester is the grant's holder and the party that metered the action
 //! (SPEC §11): an endpoint takes no figure from a party it does not hold to
 //! account for it.
+//!
+//! An endpoint that can restart keeps a [`Journal`] (SPEC §4.11): each admitted
+//! request is recorded before it is decided, each revocation before it is
+//! applied, and [`Endpoint::with_journal`] replays them through the gate at the
+//! times they were decided, so a restart gives no budget back, unrevokes
+//! nothing, and forgets no request that may still be shown again.
+
+mod journal;
+pub use journal::{FileJournal, Journal, JournalRecord, MemoryJournal};
 
 use std::collections::HashMap;
 use std::io::{Read, Write};
@@ -53,6 +62,21 @@ pub enum WireError {
     Closed,
 }
 
+/// Errors from an endpoint's journal: an append it could not make, a
+/// revocation it refused, or a journal that does not replay.
+#[derive(Debug, thiserror::Error)]
+pub enum EndpointError {
+    #[error("journal: {0}")]
+    Journal(#[from] std::io::Error),
+    #[error("revocation: {0}")]
+    Revocation(#[from] jcp_grant::GrantError),
+    /// A replayed record the gate would not have produced: a request whose
+    /// caller signature fails, or that was decided twice, or a revocation that
+    /// does not verify.
+    #[error("journal record {record} is inconsistent: {reason}")]
+    Inconsistent { record: usize, reason: String },
+}
+
 /// One action over the wire: decide `frame` against `chain` (root first, leaf
 /// last) for `caller` at metered `cost`. A single un-delegated grant is a
 /// one-link chain.
@@ -201,6 +225,8 @@ pub struct Endpoint {
     skew_secs: u64,
     /// The requests decided within the window, by content id, with their stamp.
     decided: HashMap<ContentId, u64>,
+    /// Where each admitted request and each revocation is recorded first.
+    journal: Box<dyn Journal + Send>,
 }
 
 impl Endpoint {
@@ -214,9 +240,54 @@ impl Endpoint {
             clock: system_clock,
             skew_secs: DEFAULT_SKEW_SECS,
             decided: HashMap::new(),
+            journal: Box::new(MemoryJournal::default()),
         }
     }
 
+    /// Record in `journal`, first replaying `recorded`, the records it already
+    /// holds, through the gate (SPEC §4.11). Each request is decided again at
+    /// the time it was decided, which re-establishes the spend and the memory
+    /// of requests still inside the window; each revocation is applied again.
+    /// Configure the runtime (trust roots, taint floors) as it was before
+    /// calling this, since the replay decides under it.
+    pub fn with_journal(mut self, journal: impl Journal + Send + 'static, recorded: &[JournalRecord]) -> Result<Self, EndpointError> {
+        let inconsistent = |record: usize, reason: &str| EndpointError::Inconsistent { record, reason: reason.into() };
+        for (i, record) in recorded.iter().enumerate() {
+            match record {
+                JournalRecord::Decided { at, request } => {
+                    if !request.verify_caller() {
+                        return Err(inconsistent(i, "the caller's signature does not verify"));
+                    }
+                    let id = ContentId::of(request).map_err(|e| inconsistent(i, &e.to_string()))?;
+                    if self.decided.insert(id, request.now).is_some() {
+                        return Err(inconsistent(i, "the request was decided twice"));
+                    }
+                    self.runtime.decide_chained(&request.frame, &request.chain, request.caller, request.cost, *at);
+                }
+                JournalRecord::Revoked { revocation, grant } => {
+                    self.runtime
+                        .apply_revocation(revocation, grant)
+                        .map_err(|e| inconsistent(i, &e.to_string()))?;
+                }
+            }
+        }
+        let now = (self.clock)();
+        let skew = self.skew_secs;
+        self.decided.retain(|_, stamp| stamp.saturating_add(skew) >= now);
+        self.journal = Box::new(journal);
+        Ok(self)
+    }
+
+    /// Honour a revocation: verify it, record it, then apply it, so it
+    /// survives a restart. Applying one to `runtime` directly works until the
+    /// endpoint restarts.
+    pub fn revoke(&mut self, revocation: &jcp_schema::SignedRevocation, grant: &jcp_schema::Grant) -> Result<(), EndpointError> {
+        jcp_grant::verify_revocation(revocation, grant)?;
+        self.journal.append(&JournalRecord::Revoked { revocation: revocation.clone(), grant: grant.clone() })?;
+        self.runtime.apply_revocation(revocation, grant)?;
+        Ok(())
+    }
+
     /// Decide at `clock` (unix seconds) instead of the system clock.
     pub fn with_clock(mut self, clock: fn() -> u64) -> Self {
         self.clock = clock;
@@ -251,7 +322,15 @@ impl Endpoint {
     /// other decision — the refused impersonation is evidence too.
     pub fn handle(&mut self, req: &WireRequest) -> WireResponse {
         let now = (self.clock)();
-        let refused = if req.verify_caller() { self.admit(req, now).err() } else { Some(jcp_schema::DenyReason::BadSignature) };
+        let mut refused = if req.verify_caller() { self.admit(req, now).err() } else { Some(jcp_schema::DenyReason::BadSignature) };
+        // Recorded before it is decided: a decision that is lost on a restart
+        // would take its debit with it (SPEC §4.11).
+        if refused.is_none() && self.journal.append(&JournalRecord::Decided { at: now, request: req.clone() }).is_err() {
+            if let Ok(id) = ContentId::of(req) {
+                self.decided.remove(&id);
+            }
+            refused = Some(jcp_schema::DenyReason::Unrecorded);
+        }
         let decision = match refused {
             None => self.runtime.decide_chained(&req.frame, &req.chain, req.caller, req.cost, now),
             Some(reason) => Decision {
@@ -460,6 +539,118 @@ mod tests {
         assert!(ep.decided.is_empty(), "nothing outside the window is remembered");
     }
 
+    /// A journal the test can read after the endpoint is gone.
+    #[derive(Clone, Default)]
+    struct Shared(std::sync::Arc<std::sync::Mutex<Vec<JournalRecord>>>);
+    impl Journal for Shared {
+        fn append(&mut self, record: &JournalRecord) -> std::io::Result<()> {
+            self.0.lock().unwrap().push(record.clone());
+            Ok(())
+        }
+    }
+
+    /// A journal that cannot be written.
+    struct Broken;
+    impl Journal for Broken {
+        fn append(&mut self, _: &JournalRecord) -> std::io::Result<()> {
+            Err(std::io::Error::other("disk full"))
+        }
+    }
+
+    fn endpoint(clock: fn() -> u64) -> Endpoint {
+        Endpoint::new(Runtime::new(), sk(7), "did:example:steward").with_clock(clock)
+    }
+
+    /// A restart is not a fresh budget: the endpoint replays what it decided,
+    /// so the leaf has what it had left.
+    #[test]
+    fn a_restart_keeps_the_spend() {
+        let (chain, _, r) = fixture();
+        let journal = Shared::default();
+        let mut before = endpoint(at_10).with_journal(journal.clone(), &[]).unwrap();
+        assert!(before.handle(&stamped(&chain, r, 40_000, 10)).decision.is_allow());
+        drop(before);
+        let recorded = journal.0.lock().unwrap().clone();
+        let mut after = endpoint(at_10).with_journal(MemoryJournal::default(), &recorded).unwrap();
+        assert_eq!(after.runtime.spent(&chain[1].grant.id()), 40_000, "the restart gave nothing back");
+        let d = after.handle(&stamped(&chain, r, 20_000, 9));
+        assert_eq!(d.decision.verdict, Verdict::Deny(DenyReason::BudgetExceeded));
+    }
+
+    /// A revocation honoured through the endpoint is recorded, and still
+    /// stands after a restart.
+    #[test]
+    fn a_revocation_survives_a_restart() {
+        let (chain, _, r) = fixture();
+        let journal = Shared::default();
+        let mut before = endpoint(at_10).with_journal(journal.clone(), &[]).unwrap();
+        let of_root = revoke(&sk(1), &chain[0].grant, [9; 16]).unwrap();
+        assert!(before.revoke(&of_root, &chain[1].grant).is_err(), "a revocation of another grant");
+        before.revoke(&of_root, &chain[0].grant).unwrap();
+        drop(before);
+        let recorded = journal.0.lock().unwrap().clone();
+        assert_eq!(recorded.len(), 1, "a refused revocation is not recorded");
+        let mut after = endpoint(at_10).with_journal(MemoryJournal::default(), &recorded).unwrap();
+        assert_eq!(after.handle(&stamped(&chain, r, 1, 10)).decision.verdict, Verdict::Deny(DenyReason::Revoked));
+    }
+
+    /// The endpoint remembers what it decided across a restart, so a request
+    /// still inside the window is refused, not decided a second time.
+    #[test]
+    fn a_request_decided_before_a_restart_is_still_refused() {
+        let (chain, _, r) = fixture();
+        let journal = Shared::default();
+        let req = stamped(&chain, r, 10_000, 10);
+        let mut before = endpoint(at_10).with_journal(journal.clone(), &[]).unwrap();
+        assert!(before.handle(&req).decision.is_allow());
+        drop(before);
+        let recorded = journal.0.lock().unwrap().clone();
+        let mut after = endpoint(at_10).with_journal(MemoryJournal::default(), &recorded).unwrap();
+        assert_eq!(after.handle(&req).decision.verdict, Verdict::Deny(DenyReason::Replayed));
+        assert_eq!(after.runtime.spent(&chain[1].grant.id()), 10_000);
+
+        let twice = [recorded[0].clone(), recorded[0].clone()];
+        assert!(matches!(endpoint(at_10).with_journal(MemoryJournal::default(), &twice), Err(EndpointError::Inconsistent { record: 1, .. })));
+    }
+
+    /// A request that cannot be recorded is not decided, and debits nothing.
+    #[test]
+    fn an_unrecordable_request_is_not_decided() {
+        let (chain, _, r) = fixture();
+        let mut ep = endpoint(at_10).with_journal(Broken, &[]).unwrap();
+        let req = stamped(&chain, r, 10_000, 10);
+        assert_eq!(ep.handle(&req).decision.verdict, Verdict::Deny(DenyReason::Unrecorded));
+        assert_eq!(ep.runtime.spent(&chain[1].grant.id()), 0);
+        assert!(ep.decided.is_empty(), "an undecided request is not remembered as decided");
+    }
+
+    /// The file journal round-trips, and a record a crash cut short is dropped
+    /// and cut from the file rather than failing the restart.
+    #[test]
+    fn the_file_journal_drops_a_torn_record() {
+        let dir = std::env::temp_dir().join(format!("jcp-wire-journal-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let path = dir.join("journal.cbor");
+        let _ = std::fs::remove_file(&path);
+        let (chain, _, r) = fixture();
+        {
+            let (journal, recorded) = FileJournal::open(&path).unwrap();
+            assert!(recorded.is_empty());
+            let mut ep = endpoint(at_10).with_journal(journal, &recorded).unwrap();
+            assert!(ep.handle(&stamped(&chain, r, 30_000, 10)).decision.is_allow());
+        }
+        let whole = std::fs::metadata(&path).unwrap().len();
+        let mut f = std::fs::OpenOptions::new().append(true).open(&path).unwrap();
+        f.write_all(&[0, 0, 1, 0, 7, 7]).unwrap();
+        drop(f);
+        let (journal, recorded) = FileJournal::open(&path).unwrap();
+        assert_eq!(recorded.len(), 1);
+        assert_eq!(std::fs::metadata(&path).unwrap().len(), whole, "the torn record was cut");
+        let after = endpoint(at_10).with_journal(journal, &recorded).unwrap();
+        assert_eq!(after.runtime.spent(&chain[1].grant.id()), 30_000);
+        std::fs::remove_dir_all(&dir).unwrap();
+    }
+
     // the fixture's caller is sk(3) — sign as them
     fn request(chain: &[SignedGrant], _caller: Did, r: ContentId, trust: Trust, cost: u64) -> WireRequest {
         let frame = Frame {