Struct Journal 

pub struct Journal { /* private fields */ }

The journal of one runtime. Mutex<BTreeMap>: a handful of events a second, and a deterministic order falls out for free.

One journal holds every stream this runtime knows of, not only its own: what arrives from a peer is stored beside what this runtime wrote, which is what lets it relay. It is the only thing in the mesh that writes, and everything a surface shows about cooperation is a fold of what it holds.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal, SessionInfo};

let a = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let b = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000002",
    "repo".into(), None).unwrap();
a.append_own(EventBody::SessionOpened { info: SessionInfo::named("s1", "cli") }).unwrap();

// what B receives is held beside what B writes, so B can relay A's stream onwards
b.ingest(&a.missing_for(&b.marks(), 1 << 20), &|_| Ok(()));
assert_eq!(b.tallies().streams, 2, "B's own stream, and the one it heard from A");
assert!(b.events().iter().any(|e| &e.stream == a.own_stream()));

Implementations

impl Journal

pub fn open( identity: Arc<NodeIdentity>, runtime: &str, repo: String, path: Option<PathBuf>, ) -> Result<Self, MeshError>

Open the journal of this runtime: the own stream is <node>-<runtime>-<instance> of the identity given, and path, when given, is the JSONL file events persist to and are reloaded from. Reloaded events keep their streams; none of them is fresh until a beat is heard again, so a restart never resurrects anybody’s ownership.

That last part is the reason persistence is safe here: reloading events would otherwise bring back claims whose holders are long gone, and a restarted runtime would refuse work on behalf of machines that are not running. Persistence is optional because a journal without a path is a complete journal — it simply starts empty and refills from its peers.

use std::sync::Arc;
use std::time::Duration;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal, StreamLiveness};

let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("journal.jsonl");

let before = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()),
    "0000000000000001", "repo".into(), Some(path.clone())).unwrap();
before.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();

// another runtime opens the same file: the events are back, and whoever wrote them
// is not alive here until a beat is heard again
let after = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()),
    "0000000000000002", "repo".into(), Some(path)).unwrap();
assert_eq!(after.events().len(), 1);
assert_eq!(
    after.liveness(before.own_stream(), Duration::from_secs(30)),
    StreamLiveness::Expired,
);

// a runtime slot that is not 16 hex is not a runtime
let ephemeral = Arc::new(NodeIdentity::ephemeral().unwrap());
assert!(Journal::open(ephemeral, "nope", "repo".into(), None).is_err());

This opening believes every origin the file holds, which is what a journal without a trust policy of its own can do. A runtime that has one opens with Journal::open_trusting, so that a key withdrawn from its allowlist is not believed again from its own disk.

pub fn open_trusting( identity: Arc<NodeIdentity>, runtime: &str, repo: String, path: Option<PathBuf>, accept: &dyn Fn(&MeshEvent) -> Result<(), Rejection>, ) -> Result<Self, MeshError>

Journal::open, with the trust decision the live path makes applied to every event reloaded from path.

A journal file is input, the same as a sync round: it holds whatever was trusted when it was written, and the policy may have changed since. A key removed from the allowlist and a runtime restarted would otherwise bring that key’s sessions, claims and handovers back from disk — events the live path would refuse the moment they arrived over a link. So every reloaded event passes the same checks as a received one and then accept, the same predicate the caller hands to Journal::ingest; an event signed by this node’s own key is accepted without asking, because the key is the machine’s own and a restarted runtime of it is the same machine. What is refused is not dropped silently: it is in Journal::reload_report, by reason.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal, Rejection};

let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("journal.jsonl");
let machine = Arc::new(NodeIdentity::ephemeral().unwrap());
let peer = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()),
    "0000000000000001", "repo".into(), None).unwrap();
let peers_event = peer.append_own(EventBody::SessionClosed { session: "p".into() })
    .unwrap();

// this machine stores a peer's event and one of its own
let before = Journal::open(Arc::clone(&machine), "0000000000000002", "repo".into(),
    Some(path.clone())).unwrap();
before.ingest(&[peers_event], &|_| Ok(()));
before.append_own(EventBody::SessionClosed { session: "own".into() }).unwrap();

// the peer is no longer trusted when another server of this machine — the same key,
// another stream, as a restart is — opens the file
let after = Journal::open_trusting(Arc::clone(&machine), "0000000000000003",
    "repo".into(), Some(path), &|_| Err(Rejection::Untrusted)).unwrap();
assert_eq!(after.events().len(), 1, "the machine's own event, and only it");
assert_eq!(after.reload_report().rejected[&Rejection::Untrusted], 1);

pub fn reload_report(&self) -> &IngestReport

What reloading the journal file did when this journal was opened: how many events came back, and how many were refused and why — a key no longer trusted is refused Rejection::Untrusted here rather than believed again from disk. Empty for a journal without a file, or one whose file did not exist yet.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::Journal;

let j = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
assert_eq!(j.reload_report().accepted + j.reload_report().rejected_total(), 0);

pub fn own_stream(&self) -> &StreamId

This runtime’s own stream: the one stream this journal may write to, and the prefix every claim, session and review key it creates carries. Nothing else in the journal is this runtime’s to number or to sign.

pub fn repository(&self) -> &str

The mesh repository id this journal belongs to.

pub fn append_own(&self, body: EventBody) -> Result<MeshEvent, MeshError>

Sign, store and persist an event of this runtime. The body’s bounds are checked first: nothing out of bounds is ever signed.

The sequence is dense from 1 and the Lamport stamp rises above every stamp this runtime has seen, so a peer can tell from the numbers alone whether it is missing something, and every runtime folds concurrent events in the same order. Writing is the only way an event enters the mesh; a relay never creates one.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal};

let j = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let first = j.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();
let second = j.append_own(EventBody::SessionClosed { session: "s2".into() }).unwrap();
assert_eq!((first.seq, second.seq), (1, 2), "dense from 1");
assert!(second.lamport > first.lamport);
assert_eq!(first.stream, *j.own_stream());

// out of bounds is refused before anything is signed or stored
let escaping = EventBody::ClaimAcquired { claim: "c1".into(), session: "s1".into(),
    scope: vec!["/etc".into()], intent: None, mode: Default::default(), issue: None };
assert!(j.append_own(escaping).is_err());
assert_eq!(j.events().len(), 2, "and nothing was written");

This answers the event and not what happened on disk; a caller that reports the write to anyone uses Journal::append_acknowledged, which says.

pub fn append_acknowledged( &self, body: EventBody, ) -> Result<Appended, MeshError>

Journal::append_own, answering whether the event reached this machine’s disk.

An event whose write failed is still held and replicated — a peer that receives it is the only copy left once this runtime restarts — and is answered Durability::NotDurable: the append is not acknowledged as durable, the failure is counted in JournalTallies::write_failures, and mesh doctor fails on it.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{Durability, EventBody, Journal};

let dir = tempfile::tempdir().unwrap();
let on_disk = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()),
    "0000000000000001", "repo".into(), Some(dir.path().join("journal.jsonl"))).unwrap();
let written = on_disk.append_acknowledged(EventBody::SessionClosed { session: "s1".into() })
    .unwrap();
assert_eq!(written.durability, Durability::Synced);

let in_memory = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()),
    "0000000000000002", "repo".into(), None).unwrap();
let held = in_memory.append_acknowledged(EventBody::SessionClosed { session: "s1".into() })
    .unwrap();
assert_eq!(held.durability, Durability::Memory, "no file, by configuration");

pub fn beat_own(&self)

Raise this runtime’s own beat: one heartbeat, signed, so that no relay can raise it.

The beat is how a runtime says “still here” without writing an event for it — it travels in marks, which every sync round carries anyway, so liveness costs nothing extra. The signature is what makes a relay safe: a runtime can pass on somebody else’s beat and cannot invent one, so a crashed runtime stops being alive for everybody at once and its claims end on their own.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::Journal;

let j = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
j.beat_own();
let first = j.marks()[j.own_stream()].clone();
j.beat_own();
let second = j.marks()[j.own_stream()].clone();

assert!(second.beat > first.beat, "the beat only rises");
assert_ne!(second.sig, first.sig, "each beat is its own signed statement");

pub fn marks(&self) -> Marks

Every stream’s mark, own included, ages measured now. This is the whole of what a runtime tells a peer about its state: a sequence per stream and a beat per stream, from which the peer computes what to send. Nothing is asked for and nothing is flooded, which is why an event reaches every runtime once per link and stops.

The ages are measured at the moment of the call rather than stored, because an age is only meaningful relative to when it was taken.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal};

let j = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
assert_eq!(j.marks()[j.own_stream()].seq, 0, "nothing written yet");

j.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();
j.beat_own();
let mark = j.marks()[j.own_stream()].clone();
assert_eq!(mark.seq, 1);
assert!(mark.age_ms.is_some(), "a beat has been heard, and how long ago");

pub fn marks_for(&self, peer: &Marks) -> Marks

The marks to tell one peer, whose own marks are peer: every stream this runtime holds, and a compacted stream’s tombstone only when the peer holds that stream.

A tombstone is in the marks to stop a peer re-sending a stream this runtime dropped, so it says something only to a peer that has the stream to send. Told to everyone, the up to four tombstones per stream slot weighed as much as the live marks, and together with a full event budget they took a sync message past the link bound (I2139). A peer whose marks are not known yet is told no tombstone; what it then sends of a dropped stream is absorbed by the tombstone, once, and its marks are known from that round on.

use std::sync::Arc;
use std::time::Duration;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal, Marks, HANDOVER_RETENTION};

let open = |n: &str| Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), n,
    "repo".into(), None).unwrap();
let (a, b) = (open("0000000000000001"), open("0000000000000002"));
a.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();
b.ingest(&a.missing_for(&b.marks(), 1 << 20), &|_| Ok(()));
let zero = Duration::ZERO;
assert_eq!(b.compact(zero, zero, HANDOVER_RETENTION, true), 1);

// a peer that holds the dropped stream is told how far it got, so it sends nothing
assert_eq!(b.marks_for(&a.marks())[a.own_stream()].seq, 1);
// a peer that does not is told nothing about it
assert!(!b.marks_for(&Marks::new()).contains_key(a.own_stream()));
// and the streams b holds are told to every peer
assert!(b.marks_for(&Marks::new()).contains_key(b.own_stream()));

pub fn merge_marks( &self, marks: &Marks, trusted: &dyn Fn(&str) -> bool, expiry: Duration, )

Merge a peer’s marks. Only a higher beat carrying its origin’s valid signature, from an origin trusted accepts, makes a stream fresher here — a relay forwards beats and cannot mint one — and the relayed age is clamped to the expiry, so a stale report can age a stream but never hold a dead one alive past one expiry after its last real beat. Sequences are not merged: only events move high-water marks. A stream is created from a mark only when the mark verifies, only within the stream quotas, and only when the beat it carries is fresh: a mark that can only report a stream dead — a relay still advertising a stream this runtime has compacted — gives the journal nothing to hold, and must not take a slot of its node’s quota.

A stream this runtime has compacted comes back the same way, and only on a beat above the one it was compacted at: a laptop that slept past the retention wakes as the same stream and beats again, while a relay replaying the old beat is remembering the stream, not hearing it. Heard again, the stream’s tombstone is lifted and its mark falls to what is held of it — nothing — so the peers send it again from its first event. The whole stream, not the part after the tombstone’s mark: what it said before it was compacted (a claim it still holds, a session still open) is live again with it, and a suffix would leave this runtime admitting claims that conflict with it.

use std::sync::Arc;
use std::time::Duration;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{Journal, StreamLiveness};

let a = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let b = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000002",
    "repo".into(), None).unwrap();
a.beat_own();

// B learns that A is beating, because A signed the beat
b.merge_marks(&a.marks(), &|_| true, Duration::from_secs(30));
assert_eq!(b.liveness(a.own_stream(), Duration::from_secs(30)), StreamLiveness::Live);

// an untrusted origin's beat tells B nothing, however well signed
let c = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000003",
    "repo".into(), None).unwrap();
c.beat_own();
b.merge_marks(&c.marks(), &|_| false, Duration::from_secs(30));
assert_eq!(b.liveness(c.own_stream(), Duration::from_secs(30)), StreamLiveness::Expired);

// a trusted beat heard a minute ago says only that C is dead: B holds nothing for it
let mut stale = c.marks();
stale.get_mut(c.own_stream()).unwrap().age_ms = Some(60_000);
b.merge_marks(&stale, &|_| true, Duration::from_secs(30));
assert_eq!(b.tallies().streams, 2, "B's own stream and A's");

pub fn missing_for(&self, peer: &Marks, budget: usize) -> Vec<MeshEvent>

The events a peer lacks by its marks, in stream-then-sequence order, at most budget serialized bytes (always at least one event when any is missing).

This is the half of replication that decides what travels, and it decides it from the peer’s own marks rather than from a request: a runtime sends what the peer says it lacks and nothing else, so an event crosses each link once and stops. The budget keeps one round bounded, and the streams are visited from a rotating start, so a stream a peer keeps refusing cannot spend the whole budget round after round and starve the streams behind it.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal};

let a = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let b = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000002",
    "repo".into(), None).unwrap();
a.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();
a.append_own(EventBody::SessionClosed { session: "s2".into() }).unwrap();

// B has nothing, so both events are missing for it
let missing = a.missing_for(&b.marks(), 1 << 20);
assert_eq!(missing.len(), 2);

// once B holds them, its marks say so and nothing is sent again
b.ingest(&missing, &|_| Ok(()));
assert!(a.missing_for(&b.marks(), 1 << 20).is_empty());

// a budget of nothing still moves one event: progress is never zero
assert_eq!(a.missing_for(&Default::default(), 0).len(), 1);

pub fn ingest( &self, events: &[MeshEvent], accept: &dyn Fn(&MeshEvent) -> Result<(), Rejection>, ) -> IngestReport

Verify and store received events. accept is the trust decision for an event’s origin, made above the journal; everything else — size, version, identity, signature, repository, bounds, order, duplicates — is decided here, once, for every transport and relay alike.

Trust is the caller’s because it is policy, and everything else is decided here because it must be decided the same way for every arrival: a datagram, a sync round and a file reloaded from disk are all input, and none of them is believed. Events are applied in stream order, so one that arrives ahead of its gap waits rather than being applied out of turn, and one that arrives twice is counted and changes nothing.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal, Rejection};

let a = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let b = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000002",
    "repo".into(), None).unwrap();
let first = a.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();
let second = a.append_own(EventBody::SessionClosed { session: "s2".into() }).unwrap();

// the second event alone arrives first: it waits for the gap ahead of it
let report = b.ingest(&[second.clone()], &|_| Ok(()));
assert_eq!((report.accepted, report.pending), (0, 1));

// the first arrives and both are applied, in the stream's order
let report = b.ingest(&[first], &|_| Ok(()));
assert_eq!(report.accepted, 2);
assert_eq!(b.events().len(), 2);

// an untrusted origin is refused, with the reason the caller gave
let c = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000003",
    "repo".into(), None).unwrap();
let theirs = c.append_own(EventBody::SessionClosed { session: "s9".into() }).unwrap();
let report = b.ingest(&[theirs], &|_| Err(Rejection::Untrusted));
assert_eq!(report.rejected[&Rejection::Untrusted], 1);

pub fn liveness(&self, stream: &StreamId, expiry: Duration) -> StreamLiveness

A stream’s liveness on this runtime’s clock, against expiry. The measurement is local and monotonic — how long since this process saw the beat rise — so two machines never compare wall clocks and a machine with a wrong clock cannot hold a dead runtime’s claims alive. A stream nobody has ever beaten for is expired, which is the right answer for a stream learned from a relay that has not heard from it.

use std::sync::Arc;
use std::time::Duration;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{Journal, StreamLiveness};

let a = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let b = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000002",
    "repo".into(), None).unwrap();
let expiry = Duration::from_secs(30);

assert_eq!(a.liveness(a.own_stream(), expiry), StreamLiveness::Own);
assert_eq!(a.liveness(b.own_stream(), expiry), StreamLiveness::Expired, "never heard of");

b.beat_own();
a.merge_marks(&b.marks(), &|_| true, expiry);
assert_eq!(a.liveness(b.own_stream(), expiry), StreamLiveness::Live);

pub fn stream_liveness( &self, expiry: Duration, ) -> BTreeMap<StreamId, (StreamLiveness, Option<Duration>)>

Every stream’s liveness and time since its beat rose, in stream order. The age travels with the verdict because “expired” alone is not diagnosable: a stream whose beat rose a minute ago is a runtime that stopped, and one whose beat was never heard is a runtime this one has only been told about.

use std::sync::Arc;
use std::time::Duration;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{Journal, StreamLiveness};

let j = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let all = j.stream_liveness(Duration::from_secs(30));
let (liveness, age) = all[j.own_stream()];
assert_eq!(liveness, StreamLiveness::Own);
assert_eq!(age, None, "no beat has been raised yet");

j.beat_own();
assert!(j.stream_liveness(Duration::from_secs(30))[j.own_stream()].1.is_some());

pub fn events(&self) -> Vec<MeshEvent>

Every held event, in stream-then-sequence order. Events waiting for a gap are not here: what this returns is what the journal is prepared to stand behind, and the fold takes it whole, which is why the order it comes out in does not matter to the result.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal};

let a = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let b = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000002",
    "repo".into(), None).unwrap();
let first = a.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();
let second = a.append_own(EventBody::SessionClosed { session: "s2".into() }).unwrap();

// an event still waiting for the gap ahead of it is not yet held
b.ingest(&[second], &|_| Ok(()));
assert!(b.events().is_empty());
b.ingest(&[first], &|_| Ok(()));
assert_eq!(b.events().len(), 2);
assert_eq!(b.events()[0].seq, 1, "stream order, whatever the arrival order was");

pub fn events_after(&self, after: u64, limit: usize) -> Vec<MeshEvent>

Held events whose Lamport stamp is above after, in Lamport order, at most limit.

The Lamport stamp is what a reader can resume from: it rises with causality rather than with anybody’s clock, so a caller that remembers the last stamp it saw reads each event once however the machines involved disagree about the time. This is what mesh events --after is.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal};

let j = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let first = j.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();
j.append_own(EventBody::SessionClosed { session: "s2".into() }).unwrap();

assert_eq!(j.events_after(0, 100).len(), 2, "from the beginning");
let rest = j.events_after(first.lamport, 100);
assert_eq!(rest.len(), 1, "resumed where the reader left off");
assert!(rest[0].lamport > first.lamport);
assert_eq!(j.events_after(0, 1).len(), 1, "one page");

pub fn compact( &self, expiry: Duration, retention: Duration, handovers: Duration, force: bool, ) -> u64

Drop every stream that has been expired for longer than retention and that nothing still needs, when the journal is over its bound or force is set; the file is rewritten to what remains. Only dead streams go, so no live claim can lose its release to compaction.

A node that holds NODE_COMPACTION_STREAMS streams is due on its own, whatever the journal’s size: its dead streams go by the same rule, the other nodes’ stay. One machine’s worktrees share its key and every server restart is a new stream, so without this a machine’s own runs would fill its quota and nothing would ever free it — the journal-wide bounds are out of one node’s reach — and every later run of a sibling would be refused here, its claims with it.

What makes dropping events safe is what is kept. Only a stream that has been silent well past its expiry goes, so no claim can lose the release that would have ended it. A dead stream stays while it holds a published handover nobody has taken, because a handover is continuity somebody may still be waiting to pick up — for at most handovers past the retention (HANDOVER_RETENTION in a running server), or its node’s quota would fill with the dead. And a dead stream stays while it holds the taking of a handover or the answer to a review that stays, so that a handover that is still offered never looks untaken again and an answered review never opens again. Who took what and who answered what are read by the fold (super::state::fold), the one reading every surface shows.

A stream whose beat was never heard here counts as silent from when this journal learned of it, so the streams a restart reloads are not taken for dead before their runtimes could be heard. What is dropped leaves a tombstone carrying the sequence it reached, so a peer that still holds the stream is not sent it all over again, and the beat it was last heard at, which a stream that returns must rise above (Journal::merge_marks).

use std::sync::Arc;
use std::time::Duration;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, HandoverBody, Journal, HANDOVER_RETENTION};

let a = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
let b = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000002",
    "repo".into(), None).unwrap();
a.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();
b.ingest(&a.missing_for(&b.marks(), 1 << 20), &|_| Ok(()));
assert_eq!(b.events().len(), 1);

// under its bounds, compaction does nothing unless it is asked
let zero = Duration::ZERO;
assert_eq!(b.compact(zero, zero, HANDOVER_RETENTION, false), 0);

// asked, the silent stream goes, and its mark stays so it is not re-sent
assert_eq!(b.compact(zero, zero, HANDOVER_RETENTION, true), 1);
assert!(b.events().is_empty());
assert_eq!(b.marks()[a.own_stream()].seq, 1, "the tombstone remembers how far it got");

// a handover nobody has taken keeps its stream, until the handover retention passes
let c = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000003",
    "repo".into(), None).unwrap();
let body = "# Objective\nship\n".to_string();
let handover = HandoverBody { id: HandoverBody::digest_of(&body), task: None, issue: None,
    milestone: None, branch: None, head: None, created_at: None, name: None, body };
c.append_own(EventBody::HandoverPublished { handover }).unwrap();
b.ingest(&c.missing_for(&b.marks(), 1 << 20), &|_| Ok(()));
assert_eq!(b.compact(zero, zero, HANDOVER_RETENTION, true), 0);
assert_eq!(b.compact(zero, zero, zero, true), 1);

pub fn tallies(&self) -> JournalTallies

The tallies now: a snapshot of the counters, taken under the lock so that the numbers in one answer are consistent with each other rather than read one at a time while the journal moves. This is what mesh status and the cooperation status show.

use std::sync::Arc;
use majordomus_cli::mesh::identity::NodeIdentity;
use majordomus_cli::mesh::journal::{EventBody, Journal};

let j = Journal::open(Arc::new(NodeIdentity::ephemeral().unwrap()), "0000000000000001",
    "repo".into(), None).unwrap();
j.append_own(EventBody::SessionClosed { session: "s1".into() }).unwrap();

let snapshot = j.tallies();
assert_eq!(snapshot.written, 1);
assert_eq!(snapshot.events, j.events().len());
assert_eq!(snapshot.received, 0, "nothing came from a peer");