wire/server/
committed.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4// Copyright (c) 2026 2youg1 and the sprawling contributors
5
6//! A committed record and its one wire frame (`crates/wire/spec/Server/Committed.lean` §8-47).
7//!
8//! The frame is spelled once, before the broadcast, so a socket's share
9//! of an event is two reference counts and a write however many tabs a
10//! person has open.
11
12use std::sync::Arc;
13
14use axum::extract::ws::Utf8Bytes;
15use kernel::{AxCode, AxError, EventRecord};
16
17/// What the Ledger's writer broadcasts: a record, and the
18/// `ServerFrame::Event` text that carries it. Cloning copies no bytes.
19#[derive(Debug, Clone)]
20pub struct Committed {
21    record: Arc<EventRecord>,
22    frame: Utf8Bytes,
23}
24
25impl Committed {
26    /// Spells the frame: `{"event":`, the record's JSON, `}` - the shape
27    /// the externally tagged `ServerFrame::Event` serialises to.
28    ///
29    /// # Errors
30    /// `WireMismatch` when the record does not serialise; that record
31    /// reaches no socket, and the gap in `seq` it leaves makes each live
32    /// session send `Lagged` at the next record (`crates/wire/spec/Reception.lean` §8-41).
33    pub fn new(record: EventRecord) -> Result<Self, AxError> {
34        let unframed = |detail: String| {
35            AxError::failure(AxCode::WireMismatch, "frame a committed record", detail)
36                .with_recovery("ask for the record's range; the ledger holds it")
37        };
38        let mut frame = b"{\"event\":".to_vec();
39        serde_json::to_writer(&mut frame, &record)
40            .map_err(|source| unframed(source.to_string()))?;
41        frame.push(b'}');
42        let text = String::from_utf8(frame).map_err(|source| unframed(source.to_string()))?;
43        Ok(Self {
44            record: Arc::new(record),
45            frame: Utf8Bytes::from(text),
46        })
47    }
48
49    /// The record the frame carries.
50    #[must_use]
51    pub fn record(&self) -> &EventRecord {
52        &self.record
53    }
54
55    /// The frame's text, shared rather than copied.
56    pub(crate) fn frame(&self) -> Utf8Bytes {
57        self.frame.clone()
58    }
59}
60
61#[cfg(test)]
62#[allow(
63    clippy::unwrap_used,
64    clippy::expect_used,
65    clippy::panic,
66    clippy::disallowed_methods,
67    clippy::arithmetic_side_effects
68)]
69mod tests {
70    use std::time::Instant;
71
72    use kernel::{B3Hash, EventRecord, Seq, TimeMs};
73    use tokio::sync::broadcast;
74
75    use super::Committed;
76    use crate::frames::ServerFrame;
77
78    fn record(seq: u64) -> EventRecord {
79        EventRecord::from_draft(
80            kernel::EventDraft {
81                run: kernel::RunId::from_bytes([4u8; 16]),
82                t: TimeMs::new(seq),
83                who: "resident".to_owned(),
84                addr: None,
85                kind: kernel::EventKind::RunStarted,
86                data: kernel::Payload::empty(),
87                ig: false,
88            },
89            Seq::new(seq),
90            B3Hash::digest(b"prev"),
91        )
92    }
93
94    #[test]
95    fn the_spliced_frame_is_the_serialised_event_frame() {
96        let committed = Committed::new(record(7)).unwrap();
97        let whole = serde_json::to_string(&ServerFrame::Event(Box::new(record(7)))).unwrap();
98        assert_eq!(committed.frame().as_str(), whole);
99    }
100
101    /// Nanoseconds of CPU one event costs the fan-out: the frame spelled
102    /// once, one send, then every subscriber takes the text it writes.
103    fn fanout_ns_per_event(sockets: usize) -> u128 {
104        const EVENTS: u64 = 4096;
105        let (events, _) = broadcast::channel::<Committed>(64);
106        let mut subscribers: Vec<_> = (0..sockets).map(|_| events.subscribe()).collect();
107        let prepared: Vec<EventRecord> = (1..=EVENTS).map(record).collect();
108        let mut written = 0usize;
109        let started = Instant::now();
110        for event in prepared {
111            events.send(Committed::new(event).unwrap()).unwrap();
112            for subscriber in &mut subscribers {
113                let text = subscriber.try_recv().unwrap().frame();
114                written = written.wrapping_add(text.len());
115            }
116        }
117        let spent = started.elapsed().as_nanos();
118        assert!(written > 0);
119        spent / u128::from(EVENTS)
120    }
121
122    #[test]
123    #[ignore = "an instrument: run with --release --run-ignored only"]
124    fn instrument_fanout_cpu_per_event() {
125        let readings: Vec<(usize, u128)> = [1, 4, 16]
126            .into_iter()
127            .map(|sockets| (sockets, fanout_ns_per_event(sockets)))
128            .collect();
129        for (sockets, ns) in &readings {
130            println!("fanout: {sockets} sockets, {ns} ns per event");
131        }
132        let at_one = readings.first().map(|(_, ns)| *ns).unwrap();
133        let at_sixteen = readings.last().map(|(_, ns)| *ns).unwrap();
134        assert!(
135            at_sixteen <= at_one.saturating_mul(2),
136            "16 sockets cost {at_sixteen} ns per event against {at_one} ns at one: the record is serialised per socket again"
137        );
138    }
139}
wire/server/committed.rs

wire/server/
committed.rs
