kernel/event/
payload.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//! Event payloads, drafts, records, and refs.
7
8use serde::{Deserialize, Serialize};
9
10use crate::address::Address;
11use crate::consts_external::EVENT_LOG_V;
12use crate::error::{AxCode, AxError};
13use crate::locator::B3Hash;
14
15use super::identity::{RunId, Seq, TimeMs};
16use super::kind::EventKind;
17
18/// The deepest nesting a payload may carry, its own object counted as
19/// the first level: serde_json's recursion limit of 128 refuses the
20/// 128th nested container, so a line holds 127, and the [`EventRecord`]
21/// envelope spends one of them.
22const PAYLOAD_DEPTH_MAX: usize = 126;
23
24/// Ledger payload: a JSON object with every float refused, at construction
25/// and again on read (determinism rule 6). Keys serialize sorted
26/// (serde_json's default BTreeMap), which is part of the canonical bytes.
27#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
28#[serde(transparent)]
29#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
30pub struct Payload(serde_json::Map<String, serde_json::Value>);
31
32impl Payload {
33    /// Sole constructor.
34    pub fn new(map: serde_json::Map<String, serde_json::Value>) -> Result<Self, AxError> {
35        reject_floats_and_depth(&map)?;
36        Ok(Payload(map))
37    }
38
39    pub fn empty() -> Self {
40        Payload(serde_json::Map::new())
41    }
42
43    /// The way in: one of the record structs in [`crate::event::record`],
44    /// encoded as the object a ledger line carries. Every writer goes
45    /// through here, so the keys of a kind are spelled in exactly one
46    /// place: the struct.
47    ///
48    /// # Errors
49    /// Refuses a record that does not encode to a JSON object, and any
50    /// float inside one, for the reasons [`Payload::new`] gives.
51    pub fn of(record: &impl Serialize) -> Result<Self, AxError> {
52        let value = serde_json::to_value(record).map_err(|err| {
53            AxError::failure(
54                AxCode::InvalidArgs,
55                "encode a ledger payload",
56                err.to_string(),
57            )
58            .with_recovery("give the record string keys and integer numbers only")
59        })?;
60        let serde_json::Value::Object(map) = value else {
61            return Err(AxError::failure(
62                AxCode::InvalidArgs,
63                "encode a ledger payload",
64                "the record did not encode to a JSON object",
65            )
66            .with_recovery("a ledger payload is an object; record this fact as a struct"));
67        };
68        Payload::new(map)
69    }
70
71    /// The way out: the record struct this payload was written from.
72    ///
73    /// A field a record written by an older build may omit carries
74    /// `#[serde(default)]` on the struct, so the tolerance is declared
75    /// once per key rather than at each reader. Unknown keys pass, so a
76    /// newer writer never breaks an older reader.
77    ///
78    /// Deserializes from the held map by reference, so a read allocates
79    /// only what `T` owns, never a second copy of the payload.
80    ///
81    /// # Errors
82    /// A payload this build cannot read as `T`.
83    pub fn read<T: serde::de::DeserializeOwned>(&self) -> Result<T, AxError> {
84        T::deserialize(&self.0).map_err(|err| {
85            AxError::failure(
86                AxCode::WireMismatch,
87                "read a ledger payload",
88                err.to_string(),
89            )
90            .with_recovery("replay with the build that wrote this record")
91        })
92    }
93
94    pub fn as_map(&self) -> &serde_json::Map<String, serde_json::Value> {
95        &self.0
96    }
97
98    /// The map itself, for a reader that rewrites it in place and builds
99    /// a new payload from the result, so no second copy is made on the
100    /// way through.
101    pub fn into_map(self) -> serde_json::Map<String, serde_json::Value> {
102        self.0
103    }
104}
105
106impl<'de> Deserialize<'de> for Payload {
107    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
108        let map = serde_json::Map::deserialize(deserializer)?;
109        Payload::new(map).map_err(serde::de::Error::custom)
110    }
111}
112
113/// Refuses a float anywhere under `map`, and containers nested past
114/// [`PAYLOAD_DEPTH_MAX`], in one iterative walk over one stack: a hostile
115/// payload cannot spend the writer's stack, and a well-formed one costs a
116/// single allocation however wide or deep it is. A container is judged
117/// before anything under it, so a payload too deep for the reader is
118/// refused for its depth even when a float waits at the bottom.
119fn reject_floats_and_depth(
120    map: &serde_json::Map<String, serde_json::Value>,
121) -> Result<(), AxError> {
122    // The payload's own object is the first level, so what it holds
123    // starts one container down.
124    let mut pending: Vec<(&serde_json::Value, usize)> =
125        map.values().map(|value| (value, 1)).collect();
126    while let Some((value, above)) = pending.pop() {
127        match value {
128            serde_json::Value::Number(n) if !n.is_i64() && !n.is_u64() => {
129                return Err(
130                    AxError::failure(AxCode::InvalidArgs, "build payload", n.to_string())
131                        .with_recovery("ledger payloads never carry floats; scale to integers"),
132                );
133            }
134            serde_json::Value::Array(items) => {
135                let level = opened(above)?;
136                pending.extend(items.iter().map(|item| (item, level)));
137            }
138            serde_json::Value::Object(inner) => {
139                let level = opened(above)?;
140                pending.extend(inner.values().map(|item| (item, level)));
141            }
142            serde_json::Value::Null
143            | serde_json::Value::Bool(_)
144            | serde_json::Value::Number(_)
145            | serde_json::Value::String(_) => {}
146        }
147    }
148    Ok(())
149}
150
151/// The level a container sits at when `above` containers hold it, or the
152/// refusal when that passes [`PAYLOAD_DEPTH_MAX`].
153fn opened(above: usize) -> Result<usize, AxError> {
154    above
155        .checked_add(1)
156        .filter(|level| *level <= PAYLOAD_DEPTH_MAX)
157        .ok_or_else(|| {
158            AxError::failure(
159                AxCode::InvalidArgs,
160                "build payload",
161                format!("nesting deeper than {PAYLOAD_DEPTH_MAX} levels"),
162            )
163            .with_recovery(
164                "store the body in CAS and carry its locator in the payload; the ledger reader cannot parse this depth back",
165            )
166        })
167}
168
169/// What a recording party supplies; the Ledger implementation owns the
170/// rest (seq, prev, v).
171#[derive(Debug, Clone, PartialEq)]
172pub struct EventDraft {
173    pub run: RunId,
174    pub t: TimeMs,
175    pub who: String,
176    pub addr: Option<Address>,
177    pub kind: EventKind,
178    pub data: Payload,
179    /// "Ignorable": a future reader may skip this line without changing
180    /// any rebuild result. Writers set it only when that holds.
181    pub ig: bool,
182}
183
184fn ig_is_false(ig: &bool) -> bool {
185    !*ig
186}
187
188/// One serialized history line. Fields private: the only paths to a
189/// record are [`EventRecord::from_draft`] (append side) and
190/// [`EventRecord::parse_line`] (read side).
191#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
192#[serde(deny_unknown_fields)]
193#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
194pub struct EventRecord {
195    v: u32,
196    run: RunId,
197    seq: Seq,
198    prev: B3Hash,
199    t: TimeMs,
200    who: String,
201    #[serde(skip_serializing_if = "Option::is_none", default)]
202    addr: Option<Address>,
203    kind: EventKind,
204    data: Payload,
205    #[serde(skip_serializing_if = "ig_is_false", default)]
206    ig: bool,
207}
208
209impl EventRecord {
210    /// Append-side assembly: the caller (a Ledger implementation) owns seq
211    /// and prev; `v` is pinned to [`EVENT_LOG_V`].
212    pub fn from_draft(draft: EventDraft, seq: Seq, prev: B3Hash) -> Self {
213        EventRecord {
214            v: EVENT_LOG_V,
215            run: draft.run,
216            seq,
217            prev,
218            t: draft.t,
219            who: draft.who,
220            addr: draft.addr,
221            kind: draft.kind,
222            data: draft.data,
223            ig: draft.ig,
224        }
225    }
226
227    /// The canonical bytes (no line terminator). The one byte producer
228    /// library-wide; adapters append these bytes verbatim and hash exactly
229    /// them for the next line's prev.
230    pub fn canonical_line(&self) -> Result<Vec<u8>, AxError> {
231        serde_json::to_vec(self).map_err(|e| {
232            AxError::failure(
233                AxCode::InvalidArgs,
234                "serialize event",
235                format!("{:?} at seq {}", self.kind, self.seq.value()),
236            )
237            .with_recovery(format!(
238                "JSON refused this payload ({e}); give the event's data only string keys and finite numbers"
239            ))
240        })
241    }
242
243    /// Read-side entrance: full field revalidation, fail-closed. Unknown
244    /// kinds fail here; the `ig:true` skip belongs to replay, which probes
245    /// the envelope before committing to a typed parse.
246    pub fn parse_line(raw: &[u8]) -> Result<Self, AxError> {
247        serde_json::from_slice(raw).map_err(|e| {
248            AxError::failure(AxCode::InvalidArgs, "parse event line", e.to_string())
249                .with_recovery("the line is not a canonical EventRecord")
250        })
251    }
252
253    /// Minting a reference requires holding a whole record (15.3-1): the
254    /// append path holds what it just assembled, replay holds what it just
255    /// verified. There is no third path.
256    pub fn to_ref(&self) -> EventRef {
257        EventRef {
258            seq: self.seq,
259            kind: self.kind,
260        }
261    }
262
263    pub fn v(&self) -> u32 {
264        self.v
265    }
266
267    pub fn run(&self) -> RunId {
268        self.run
269    }
270
271    pub fn seq(&self) -> Seq {
272        self.seq
273    }
274
275    pub fn prev(&self) -> B3Hash {
276        self.prev
277    }
278
279    pub fn t(&self) -> TimeMs {
280        self.t
281    }
282
283    pub fn who(&self) -> &str {
284        &self.who
285    }
286
287    pub fn addr(&self) -> Option<&Address> {
288        self.addr.as_ref()
289    }
290
291    pub fn kind(&self) -> EventKind {
292        self.kind
293    }
294
295    pub fn data(&self) -> &Payload {
296        &self.data
297    }
298
299    pub fn ig(&self) -> bool {
300        self.ig
301    }
302}
303
304/// Unforgeable pointer at a ledger line: private fields, no public
305/// constructor, no serde (deserialization would be a third mint).
306#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
307pub struct EventRef {
308    seq: Seq,
309    kind: EventKind,
310}
311
312impl EventRef {
313    pub fn seq(&self) -> Seq {
314        self.seq
315    }
316
317    pub fn kind(&self) -> EventKind {
318        self.kind
319    }
320}
321
322#[cfg(test)]
323#[allow(
324    clippy::unwrap_used,
325    clippy::expect_used,
326    clippy::panic,
327    clippy::indexing_slicing,
328    reason = "test code"
329)]
330mod tests;
kernel/event/payload.rs

kernel/event/
payload.rs
