dataflow_rs/run/
message.rs

1//! The message and its audit trail, on v3's wire shape.
2
3use crate::base::{Change, Context, ErrorRecord, Result};
4use crate::expr::Path;
5use chrono::{DateTime, Utc};
6use datavalue::OwnedDataValue;
7use serde::{Deserialize, Deserializer, Serialize};
8use serde_json::Value;
9use std::sync::Arc;
10
11/// One task's audit record, written as v3 wrote one.
12#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
13#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
14#[non_exhaustive]
15pub struct AuditEntry {
16    /// The workflow the task belongs to.
17    pub workflow_id: Arc<str>,
18    /// The task.
19    pub task_id: Arc<str>,
20    /// When the entry was recorded.
21    #[cfg_attr(feature = "ts", ts(type = "string"))]
22    pub timestamp: DateTime<Utc>,
23    /// The writes the task made, when changes are captured.
24    pub changes: Vec<Change>,
25    /// The task's status code.
26    pub status: usize,
27    /// The loop counter of the sweep that produced the entry.
28    #[serde(default, skip_serializing_if = "Option::is_none")]
29    #[cfg_attr(feature = "ts", ts(optional))]
30    pub loop_counter: Option<i64>,
31    /// The fan-out element the entry records.
32    #[serde(default, skip_serializing_if = "Option::is_none")]
33    #[cfg_attr(feature = "ts", ts(optional))]
34    pub element_index: Option<usize>,
35}
36
37/// Which audit entries a run keeps on the message. `Full` by default.
38///
39/// An entry is one recorded call, skip or engine write. A fan-out's `into`
40/// pre-fill is not one: before the first element runs, the engine writes one
41/// `null` per element there, so an element that fails or never runs keeps
42/// its slot. A halted fan-out therefore leaves `into` as `[0, 1, null,
43/// null]` while the trail holds only the entries for `.0` and `.1`, and
44/// replaying the trail does not rebuild the context. A trace does keep the
45/// pre-fill, as the first change of the first folded element's step, when
46/// [`TraceOptions::changes`](crate::observe::TraceOptions::changes) is on.
47#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
48#[non_exhaustive]
49pub enum AuditMode {
50    /// Keep every entry.
51    #[default]
52    Full,
53    /// Keep the most recent `n` entries; `Last(0)` keeps none.
54    Last(usize),
55    /// Keep none.
56    Off,
57}
58
59// `Context` is `base`'s; its reads and writes over a `Path` are here, since
60// `expr` sits above `base`.
61impl Context {
62    /// The value at `path`, if any.
63    pub fn get(&self, path: &Path) -> Option<&OwnedDataValue> {
64        path.get(&self.root)
65    }
66
67    /// Writes `value` at `path` by [`Path::set`]'s rules: a root takes only
68    /// an object.
69    pub fn set(&mut self, path: &Path, value: OwnedDataValue) -> Result<()> {
70        path.set(&mut self.root, value)
71    }
72
73    /// Removes and returns the value at `path`. A root cannot be removed.
74    pub fn remove(&mut self, path: &Path) -> Result<Option<OwnedDataValue>> {
75        path.remove(&mut self.root)
76    }
77}
78
79/// Pure wire data: `{id, payload, context, audit_trail, errors}`, in that
80/// order.
81///
82/// Read from JSON as v3's is: the five fields, each required, any other key
83/// ignored. Its context reads by [`Context`]'s rules, so a fourth root is
84/// refused, and each root must be an object: a root takes only an object,
85/// and a `null` or scalar `metadata` would silently drop every write under
86/// it, `metadata.progress` included (3.x took any value).
87#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
88pub struct Message {
89    id: String,
90    /// Read back as v4 wrote it.
91    #[serde(deserialize_with = "payload")]
92    payload: Arc<OwnedDataValue>,
93    #[serde(deserialize_with = "object_roots")]
94    context: Context,
95    #[serde(rename = "audit_trail")]
96    audit: Vec<AuditEntry>,
97    errors: Vec<ErrorRecord>,
98}
99
100/// A message's context, each of its roots an object. A trace's context is
101/// read by [`Context`]'s own rules, which take a redacted root's `null`.
102fn object_roots<'de, D: Deserializer<'de>>(d: D) -> std::result::Result<Context, D::Error> {
103    let context = Context::deserialize(d)?;
104    for (root, value) in [
105        ("data", context.data()),
106        ("metadata", context.metadata()),
107        ("temp_data", context.temp_data()),
108    ] {
109        if !value.is_object() {
110            return Err(serde::de::Error::custom(format_args!(
111                "`{root}` is a root and takes only an object, not {}",
112                crate::expr::path::kind(value)
113            )));
114        }
115    }
116    Ok(context)
117}
118
119/// A message's payload, read back exactly as v4 wrote it.
120fn payload<'de, D: Deserializer<'de>>(d: D) -> std::result::Result<Arc<OwnedDataValue>, D::Error> {
121    crate::base::exact(d).map(Arc::new)
122}
123
124#[cfg(feature = "ts")]
125crate::base::ts::ts_decl!(
126    Message,
127    "Message",
128    [" A message as JSON: `{id, payload, context, audit_trail, errors}`, in that order."],
129    |cfg| {
130        use crate::base::ts::{Prop, object};
131        object(&[
132            Prop::new("id", "string", &[" The id: a UUID v7 unless set."]),
133            Prop::new(
134                "payload",
135                "JsonValue",
136                &[
137                    " The payload the message arrived with. It is not part of the evaluation",
138                    " context.",
139                ],
140            ),
141            Prop::new("context", Context::name(cfg), &[" The context."]),
142            Prop::new(
143                "audit_trail",
144                Vec::<AuditEntry>::name(cfg),
145                &[" The audit entries the run kept."],
146            ),
147            Prop::new(
148                "errors",
149                Vec::<ErrorRecord>::name(cfg),
150                &[" The errors recorded while processing."],
151            ),
152        ])
153    }
154);
155
156/// The parts a run borrows separately: only `run` holds a `&mut Message`.
157#[derive(Debug)]
158pub(crate) struct Parts<'a> {
159    pub(crate) id: &'a str,
160    pub(crate) payload: &'a OwnedDataValue,
161    pub(crate) context: &'a mut OwnedDataValue,
162    pub(crate) audit: &'a mut Vec<AuditEntry>,
163    pub(crate) errors: &'a mut Vec<ErrorRecord>,
164}
165
166thread_local! {
167    /// The v7 context a thread's message ids are made with. One per thread,
168    /// so making an id takes no lock: uuid's process-wide context (v3's
169    /// `Uuid::now_v7`) serialises every message's construction across a
170    /// host's worker threads behind one mutex.
171    static IDS: uuid::ContextV7 = const { uuid::ContextV7::new() };
172}
173
174impl Message {
175    /// A message with a fresh UUID v7 id, as v3. Ids made on one thread sort
176    /// in the order they were made; ids made in the same millisecond on two
177    /// threads are distinct but not ordered. A payload passed as an `Arc` is
178    /// shared, not copied: messages built from one `Arc` hold one payload.
179    pub fn new(payload: impl Into<Arc<OwnedDataValue>>) -> Self {
180        Self::from_parts(payload, OwnedDataValue::Null, OwnedDataValue::Null)
181    }
182
183    /// A message seeded with `data` and `metadata`. A seed that is not an
184    /// object is ignored, as v3. A payload passed as an `Arc` is shared, not
185    /// copied.
186    pub fn from_parts(
187        payload: impl Into<Arc<OwnedDataValue>>,
188        data: OwnedDataValue,
189        metadata: OwnedDataValue,
190    ) -> Self {
191        // Encoded into a buffer, not `to_string`, which formats through
192        // `Display`: the same lowercase hyphenated text.
193        let mut id = [0u8; uuid::fmt::Hyphenated::LENGTH];
194        let uuid = uuid::Uuid::new_v7(uuid::Timestamp::now(
195            uuid::timestamp::context::ThreadLocalContext::new(&IDS),
196        ));
197        let id = String::from(&*uuid.hyphenated().encode_lower(&mut id));
198        Self {
199            id,
200            payload: payload.into(),
201            context: Context::new(data, metadata),
202            audit: Vec::new(),
203            errors: Vec::new(),
204        }
205    }
206
207    /// A message whose payload is `payload`, converted.
208    pub fn from_json(payload: &Value) -> Self {
209        Self::new(OwnedDataValue::from(payload))
210    }
211
212    /// Replaces the id.
213    #[must_use]
214    pub fn with_id(mut self, id: impl Into<String>) -> Self {
215        self.id = id.into();
216        self
217    }
218
219    /// The id: a UUID v7 unless set.
220    pub fn id(&self) -> &str {
221        &self.id
222    }
223
224    /// The payload the message arrived with. It is not part of the evaluation
225    /// context.
226    pub fn payload(&self) -> &OwnedDataValue {
227        &self.payload
228    }
229
230    /// The context.
231    pub fn context(&self) -> &Context {
232        &self.context
233    }
234
235    /// The context, mutably.
236    pub fn context_mut(&mut self) -> &mut Context {
237        &mut self.context
238    }
239
240    /// The audit entries the run kept.
241    pub fn audit(&self) -> &[AuditEntry] {
242        &self.audit
243    }
244
245    /// The errors recorded while processing.
246    pub fn errors(&self) -> &[ErrorRecord] {
247        &self.errors
248    }
249
250    /// The message as a JSON value holding `id`, `payload`, `context`,
251    /// `audit_trail` and `errors`.
252    ///
253    /// Its keys are in wire order: the crate turns on serde_json's
254    /// `preserve_order`. The wire text is `serde_json::to_string(&message)`.
255    pub fn to_json(&self) -> Value {
256        serde_json::to_value(self).expect("a message always serializes")
257    }
258
259    pub(crate) fn parts(&mut self) -> Parts<'_> {
260        let Self {
261            id,
262            payload,
263            context,
264            audit,
265            errors,
266        } = self;
267        Parts {
268            id,
269            payload,
270            context: &mut context.root,
271            audit,
272            errors,
273        }
274    }
275}
276
277/// Whether `mode` keeps any entry: `Off` and `Last(0)` keep none, so the
278/// engine records no changes under them and the recorder builds no
279/// entry.
280pub(crate) fn keeps_entries(mode: AuditMode) -> bool {
281    !matches!(mode, AuditMode::Off | AuditMode::Last(0))
282}
283
284/// Appends `entry` under `mode`, trimming `Last(n)` in v3's batches: the Vec
285/// holds up to `2n` entries until [`finish_audit`].
286pub(crate) fn record_audit(audit: &mut Vec<AuditEntry>, mode: AuditMode, entry: AuditEntry) {
287    match mode {
288        AuditMode::Full => audit.push(entry),
289        AuditMode::Off | AuditMode::Last(0) => {}
290        AuditMode::Last(n) => {
291            audit.push(entry);
292            if audit.len() >= n.saturating_mul(2) {
293                let excess = audit.len() - n;
294                audit.drain(..excess);
295            }
296        }
297    }
298}
299
300/// Trims `Last(n)` to exactly `n`: the message no longer carries the mode.
301pub(crate) fn finish_audit(audit: &mut Vec<AuditEntry>, mode: AuditMode) {
302    match mode {
303        AuditMode::Full => {}
304        AuditMode::Off | AuditMode::Last(0) => audit.clear(),
305        AuditMode::Last(n) => {
306            let excess = audit.len().saturating_sub(n);
307            audit.drain(..excess);
308        }
309    }
310}
311
312#[cfg(test)]
313mod tests {
314    use super::*;
315    use crate::base::ErrorKind;
316    use serde_json::json;
317
318    fn text<T: Serialize>(v: &T) -> String {
319        serde_json::to_string(v).unwrap()
320    }
321
322    fn v4_msg(payload: &Value, data: &Value, metadata: &Value) -> Message {
323        Message::from_parts(
324            OwnedDataValue::from(payload),
325            OwnedDataValue::from(data),
326            OwnedDataValue::from(metadata),
327        )
328        .with_id("m")
329    }
330
331    /// [`v4_msg`] from JSON text, so the keys keep their order in every
332    /// build.
333    fn msg_from_text(payload: &str, data: &str, metadata: &str) -> Message {
334        let parsed = |text: &str| OwnedDataValue::from_json(text).expect("JSON");
335        Message::from_parts(parsed(payload), parsed(data), parsed(metadata)).with_id("m")
336    }
337
338    #[test]
339    fn the_wire_shape_keeps_3x_bytes() {
340        let message = msg_from_text(
341            r#"{"a": [1, {"b": null}], "z": "s", "k": 1.5}"#,
342            r#"{"order": {"items": [1, 2, {"x": true}], "total": 10}, "a": 1}"#,
343            r#"{"source": "api", "tags": ["x", "y"]}"#,
344        );
345        assert_eq!(
346            text(&message),
347            concat!(
348                r#"{"id":"m","payload":{"a":[1,{"b":null}],"z":"s","k":1.5},"#,
349                r#""context":{"data":{"order":{"items":[1,2,{"x":true}],"total":10},"a":1},"#,
350                r#""metadata":{"source":"api","tags":["x","y"]},"temp_data":{}},"#,
351                r#""audit_trail":[],"errors":[]}"#
352            )
353        );
354        assert_eq!(
355            text(&msg_from_text("null", "{}", "{}")),
356            r#"{"id":"m","payload":null,"context":{"data":{},"metadata":{},"temp_data":{}},"audit_trail":[],"errors":[]}"#
357        );
358    }
359
360    #[test]
361    fn a_non_object_seed_is_ignored() {
362        let (d, m, p) = (json!(7), json!("x"), json!(null));
363        let v4 = v4_msg(&p, &d, &m);
364        assert_eq!(
365            text(v4.context()),
366            r#"{"data":{},"metadata":{},"temp_data":{}}"#
367        );
368        assert_eq!(text(&v4), text(&msg_from_text("null", "{}", "{}")));
369    }
370
371    #[test]
372    fn audit_entries_and_changes_keep_3x_bytes() {
373        let ts = "2026-10-14T00:00:00Z".parse::<DateTime<Utc>>().unwrap();
374        for ((lc, ei, removed), change, entry) in [
375            (
376                (None, None, false),
377                r#"{"path":"data.a","old_value":{"o":1},"new_value":[1]}"#,
378                r#"{"workflow_id":"w","task_id":"t","timestamp":"2026-10-14T00:00:00Z","changes":[{"path":"data.a","old_value":{"o":1},"new_value":[1]}],"status":200}"#,
379            ),
380            (
381                (Some(3), Some(2), true),
382                r#"{"path":"data.a","old_value":{"o":1},"new_value":[1],"removed":true}"#,
383                r#"{"workflow_id":"w","task_id":"t","timestamp":"2026-10-14T00:00:00Z","changes":[{"path":"data.a","old_value":{"o":1},"new_value":[1],"removed":true}],"status":200,"loop_counter":3,"element_index":2}"#,
384            ),
385            (
386                (Some(-1), None, false),
387                r#"{"path":"data.a","old_value":{"o":1},"new_value":[1]}"#,
388                r#"{"workflow_id":"w","task_id":"t","timestamp":"2026-10-14T00:00:00Z","changes":[{"path":"data.a","old_value":{"o":1},"new_value":[1]}],"status":200,"loop_counter":-1}"#,
389            ),
390        ] {
391            let mut c4 = Change::write(
392                "data.a",
393                OwnedDataValue::from(&json!({"o": 1})),
394                OwnedDataValue::from(&json!([1])),
395            );
396            c4.removed = removed;
397            assert_eq!(text(&c4), change);
398            let a4 = AuditEntry {
399                workflow_id: Arc::from("w"),
400                task_id: Arc::from("t"),
401                timestamp: ts,
402                changes: vec![c4],
403                status: 200,
404                loop_counter: lc,
405                element_index: ei,
406            };
407            assert_eq!(text(&a4), entry);
408        }
409    }
410
411    #[test]
412    fn the_id_is_the_uuids_lowercase_hyphenated_text() {
413        let m = Message::new(OwnedDataValue::Null);
414        let id = uuid::Uuid::parse_str(m.id()).unwrap();
415        assert_eq!(m.id(), id.hyphenated().to_string());
416        assert_eq!(m.id(), m.id().to_lowercase());
417    }
418
419    #[test]
420    fn a_new_message_has_a_uuid_v7_id() {
421        let m = Message::new(OwnedDataValue::Null);
422        let id = uuid::Uuid::parse_str(m.id()).unwrap();
423        assert_eq!(id.get_version_num(), 7);
424        assert_ne!(m.id(), Message::new(OwnedDataValue::Null).id());
425        assert_eq!(
426            Message::from_json(&json!({"a": 1})).payload(),
427            &OwnedDataValue::from(&json!({"a": 1}))
428        );
429    }
430
431    #[test]
432    fn ids_made_on_one_thread_sort_in_the_order_they_were_made() {
433        let ids: Vec<String> = (0..10_000)
434            .map(|_| Message::new(OwnedDataValue::Null).id().to_owned())
435            .collect();
436        assert!(ids.windows(2).all(|w| w[0] < w[1]));
437    }
438
439    #[test]
440    fn ids_made_on_several_threads_are_distinct() {
441        let threads: Vec<_> = (0..4)
442            .map(|_| {
443                std::thread::spawn(|| {
444                    (0..10_000)
445                        .map(|_| Message::new(OwnedDataValue::Null).id().to_owned())
446                        .collect::<Vec<_>>()
447                })
448            })
449            .collect();
450        let ids: std::collections::HashSet<String> = threads
451            .into_iter()
452            .flat_map(|t| t.join().unwrap())
453            .collect();
454        assert_eq!(ids.len(), 40_000);
455    }
456
457    #[test]
458    fn messages_built_from_one_arc_share_its_payload() {
459        let payload = Arc::new(OwnedDataValue::from(&json!({"a": [1, 2, 3]})));
460        let a = Message::new(Arc::clone(&payload));
461        let b = Message::from_parts(
462            Arc::clone(&payload),
463            OwnedDataValue::Null,
464            OwnedDataValue::Null,
465        );
466        assert_eq!(Arc::strong_count(&payload), 3);
467        assert!(std::ptr::eq(a.payload(), b.payload()));
468        assert!(std::ptr::eq(a.payload(), &*payload));
469    }
470
471    fn entry(task: &str) -> AuditEntry {
472        AuditEntry {
473            workflow_id: Arc::from("w"),
474            task_id: Arc::from(task),
475            timestamp: Utc::now(),
476            changes: Vec::new(),
477            status: 200,
478            loop_counter: None,
479            element_index: None,
480        }
481    }
482
483    #[test]
484    fn record_audit_keeps_the_last_entries_the_mode_asks_for() {
485        for (mode, want) in [
486            (AuditMode::Full, 0..10),
487            (AuditMode::Off, 0..0),
488            (AuditMode::Last(0), 0..0),
489            (AuditMode::Last(1), 9..10),
490            (AuditMode::Last(3), 7..10),
491            (AuditMode::Last(10), 0..10),
492            (AuditMode::Last(25), 0..10),
493        ] {
494            let want: Vec<String> = want.map(|i| format!("t{i}")).collect();
495            let mut audit = Vec::new();
496            for i in 0..10 {
497                record_audit(&mut audit, mode, entry(&format!("t{i}")));
498                if let AuditMode::Last(n) = mode {
499                    assert!(audit.len() < (2 * n).max(1), "{mode:?} bounds the Vec");
500                }
501            }
502            finish_audit(&mut audit, mode);
503            let got: Vec<String> = audit.iter().map(|e| e.task_id.to_string()).collect();
504            assert_eq!(got, want, "{mode:?}");
505        }
506    }
507
508    #[test]
509    fn context_set_at_a_root_takes_only_an_object() {
510        let mut m = Message::new(OwnedDataValue::Null);
511        let root: Path = "data".parse().unwrap();
512        let err = m
513            .context_mut()
514            .set(&root, OwnedDataValue::from(&json!(1)))
515            .unwrap_err();
516        assert_eq!(err.kind(), &ErrorKind::Config);
517        m.context_mut()
518            .set(&root, OwnedDataValue::from(&json!({"a": 1})))
519            .unwrap();
520        let a: Path = "data.a".parse().unwrap();
521        assert_eq!(m.context().get(&a), Some(&OwnedDataValue::from(&json!(1))));
522        assert_eq!(m.context().data(), &OwnedDataValue::from(&json!({"a": 1})));
523        assert_eq!(m.context().as_value()["data"], *m.context().data());
524        assert!(m.context().metadata().is_object() && m.context().temp_data().is_object());
525    }
526
527    #[test]
528    fn to_json_holds_the_wire_content_but_not_its_order() {
529        let m = v4_msg(&json!({"b": 1, "a": 2}), &json!({"x": [1]}), &json!({}));
530        let wire = serde_json::to_string(&m).unwrap();
531        assert!(wire.starts_with(r#"{"id":"#), "{wire}");
532        assert_eq!(m.to_json(), serde_json::from_str::<Value>(&wire).unwrap());
533    }
534
535    #[test]
536    fn context_remove_refuses_a_root() {
537        let mut m = Message::from_parts(
538            OwnedDataValue::Null,
539            OwnedDataValue::from(&json!({"a": 1})),
540            OwnedDataValue::Null,
541        );
542        let root: Path = "metadata".parse().unwrap();
543        assert_eq!(
544            m.context_mut().remove(&root).unwrap_err().kind(),
545            &ErrorKind::Config
546        );
547        let a: Path = "data.a".parse().unwrap();
548        assert_eq!(
549            m.context_mut().remove(&a).unwrap(),
550            Some(OwnedDataValue::from(&json!(1)))
551        );
552        assert_eq!(m.context_mut().remove(&a).unwrap(), None);
553    }
554
555    #[test]
556    fn parts_borrow_the_fields_separately() {
557        let mut m = Message::new(OwnedDataValue::from(&json!(5)));
558        let p = m.parts();
559        p.audit.push(entry("t"));
560        p.errors.push(ErrorRecord::new("C", "m"));
561        *p.context = OwnedDataValue::from(&json!({"data": {}, "metadata": {}, "temp_data": {}}));
562        assert_eq!(p.payload, &OwnedDataValue::from(&json!(5)));
563        assert_eq!(p.id.len(), 36);
564        assert_eq!((m.audit().len(), m.errors().len()), (1, 1));
565    }
566}
dataflow_rs/run/message.rs

dataflow_rs/run/
message.rs
