backbeat_cli/
trace.rs

1// Copyright (c) 2026 Cameron Bytheway
2// SPDX-License-Identifier: MIT
3
4//! `backbeat` → Chrome / Perfetto trace JSON.
5//!
6//! Spans become **async** events (`ph:"b"`/`"e"`, paired by `id`), not duration events (`B`/`E`):
7//! backbeat spans carry explicit ids, can overlap, may be emitted across different shards (thread
8//! migration), and may be orphaned by ring eviction — all of which violate the per-thread LIFO
9//! stack contract that `B`/`E` assume. Async events pair by explicit `id`, tolerate overlap, and
10//! need no nesting. Plain (non-span) events become instants (`ph:"i"`).
11//!
12//! Pairing is by `(instance_id, span_id)` so merged multi-process dumps never collide. Orphans (the
13//! other half evicted from the ring) are surfaced rather than dropped: an enter with no exit gets a
14//! synthetic close at the trace's max timestamp; an exit with no enter becomes a zero-width instant.
15//!
16//! The output is read by `chrome://tracing`, [Perfetto](https://ui.perfetto.dev/), and any
17//! Trace-Event-Format consumer.
18
19use crate::model::{self, Loaded};
20use anyhow::{Context, Result};
21use backbeat::schema::{FieldType, Phase};
22use serde_json::{json, Value as Json};
23use std::{
24    collections::{HashMap, HashSet},
25    fs::File,
26    io::{BufWriter, Write},
27    path::Path,
28};
29
30/// Writes the loaded dumps to `output` as Chrome Trace Event Format JSON. Returns the event count.
31///
32/// The output is **streamed**: events are written one at a time straight to a buffered file, never
33/// collected into an in-memory array. A trace can have hundreds of millions of events, and a
34/// `serde_json::Value` per event (a heap map of boxed values) is far larger than its serialized
35/// form — materializing them all is what made large dumps blow up memory. We instead make one cheap
36/// pass to find orphaned spans (which needs only each record's `(span_id, phase)`, not its decoded
37/// fields), then a second pass that decodes and emits each event's JSON and immediately drops it.
38/// Peak extra memory is therefore one event, not all of them.
39pub fn to_trace(dumps: &[Loaded], output: &Path) -> Result<usize> {
40    // Deduplicate up front: overlapping dumps re-capture shared ring contents, and a duplicated
41    // span enter/exit would corrupt the orphan accounting as well as double the output. The set is
42    // references into the dump buffers (no record copy) — see `model::unique_records`.
43    let records = model::unique_records(dumps);
44
45    // Pass 1: the trace's max timestamp (for synthetic orphan closes) and which (instance, span)
46    // keys have an enter / an exit. This touches only span ids and phases — no field decoding, no
47    // per-event allocation — so it is cheap even for a huge dump.
48    let mut max_ts = 0u64;
49    let mut has_enter: HashSet<(u64, u64)> = HashSet::new();
50    let mut has_exit: HashSet<(u64, u64)> = HashSet::new();
51    for (d, r) in &records {
52        let s = &d.schemas[r.schema_idx];
53        max_ts = max_ts.max(r.ts_nanos);
54        if let Some(sid) = s
55            .span_id()
56            .and_then(|f| read_u64(&r.fields, f.offset as usize))
57        {
58            match s.phase {
59                Phase::Enter => {
60                    has_enter.insert((d.instance_id, sid));
61                }
62                Phase::Exit => {
63                    has_exit.insert((d.instance_id, sid));
64                }
65                _ => {}
66            }
67        }
68    }
69
70    // Pass 2: stream the document. We hand-write the envelope around the `traceEvents` array and
71    // serialize each event into the array as we go, separated by commas, so the full array never
72    // exists in memory at once.
73    let file = File::create(output).with_context(|| format!("creating {}", output.display()))?;
74    let mut w = BufWriter::new(file);
75    write!(w, "{{\"displayTimeUnit\":\"ns\",\"traceEvents\":[").context("writing trace JSON")?;
76
77    let mut count = 0usize;
78    for (d, r) in &records {
79        let s = &d.schemas[r.schema_idx];
80        let span_id = s
81            .span_id()
82            .and_then(|f| read_u64(&r.fields, f.offset as usize));
83        // Decode this one record's fields. The args map lives only for this iteration.
84        let args: serde_json::Map<String, Json> = s
85            .fields
86            .iter()
87            .map(|f| (f.name.to_string(), decode_json(f, &r.fields, &d.intern)))
88            .collect();
89        let f = Flat {
90            instance_id: d.instance_id,
91            shard_id: r.shard_id,
92            ts_nanos: r.ts_nanos,
93            phase: s.phase,
94            span_id,
95            name: &s.qualified_name,
96        };
97        emit_record(&mut w, &f, args, max_ts, &has_enter, &has_exit, &mut count)?;
98    }
99
100    write!(w, "]}}").context("writing trace JSON")?;
101    w.flush().context("flushing trace JSON")?;
102    Ok(count)
103}
104
105/// The render-time context for one record, independent of which dump it came from. Args are passed
106/// alongside rather than stored, so a `Flat` is cheap and short-lived (one per emitted record).
107struct Flat<'a> {
108    instance_id: u64,
109    shard_id: u32,
110    ts_nanos: u64,
111    phase: Phase,
112    span_id: Option<u64>,
113    name: &'a str,
114}
115
116/// Writes a single serialized event into the `traceEvents` array, prefixing a comma before all but
117/// the first (tracked via `count`). Used by the streaming emitter so events never accumulate.
118fn write_event(w: &mut impl Write, event: &Json, count: &mut usize) -> Result<()> {
119    if *count > 0 {
120        write!(w, ",").context("writing trace JSON")?;
121    }
122    serde_json::to_writer(&mut *w, event).context("writing trace JSON")?;
123    *count += 1;
124    Ok(())
125}
126
127/// Emits the one or two trace events a record produces (a span enter/exit, plus a synthetic close
128/// for an orphan; or a plain instant), streaming each straight to `w`.
129fn emit_record(
130    w: &mut impl Write,
131    f: &Flat,
132    args: serde_json::Map<String, Json>,
133    max_ts: u64,
134    has_enter: &HashSet<(u64, u64)>,
135    has_exit: &HashSet<(u64, u64)>,
136    count: &mut usize,
137) -> Result<()> {
138    let cat = f.name.split("::").next().unwrap_or("");
139    // pid = process (instance), tid = shard/core. Pairing is by `id`, so tid is free to show
140    // per-core activity (a span's begin and end may even land on different tids after migration).
141    let pid = f.instance_id;
142    let tid = f.shard_id;
143
144    match (f.phase, f.span_id) {
145        (Phase::Enter, Some(sid)) => {
146            // Orphaned enter (exit evicted): synthesize a close at the trace's end.
147            if !has_exit.contains(&(f.instance_id, sid)) {
148                let mut a = args.clone();
149                a.insert("backbeat_open".to_string(), json!(true));
150                let close = async_event_at("e", f, sid, cat, pid, tid, max_ts, a);
151                write_event(w, &async_event("b", f, sid, cat, pid, tid, args), count)?;
152                write_event(w, &close, count)?;
153            } else {
154                write_event(w, &async_event("b", f, sid, cat, pid, tid, args), count)?;
155            }
156        }
157        (Phase::Exit, Some(sid)) => {
158            // Orphaned exit (enter evicted): a zero-width instant marks where it closed.
159            if !has_enter.contains(&(f.instance_id, sid)) {
160                let mut a = args;
161                a.insert("backbeat_orphan_exit".to_string(), json!(true));
162                write_event(w, &instant_event(f, cat, pid, tid, a), count)?;
163            } else {
164                write_event(w, &async_event("e", f, sid, cat, pid, tid, args), count)?;
165            }
166        }
167        // A plain event: an instant. If it carries a parent_span_id it still renders as an
168        // instant (Chrome has no "point attached to a span"); the parent link is in args.
169        _ => write_event(w, &instant_event(f, cat, pid, tid, args), count)?,
170    }
171    Ok(())
172}
173
174/// An async-event (`b`/`e`) at the record's own timestamp.
175fn async_event(
176    ph: &str,
177    f: &Flat,
178    sid: u64,
179    cat: &str,
180    pid: u64,
181    tid: u32,
182    args: serde_json::Map<String, Json>,
183) -> Json {
184    async_event_at(ph, f, sid, cat, pid, tid, f.ts_nanos, args)
185}
186
187/// An async-event at an explicit timestamp (used for synthetic orphan closes).
188#[allow(clippy::too_many_arguments)]
189fn async_event_at(
190    ph: &str,
191    f: &Flat,
192    sid: u64,
193    cat: &str,
194    pid: u64,
195    tid: u32,
196    ts_nanos: u64,
197    args: serde_json::Map<String, Json>,
198) -> Json {
199    json!({
200        "ph": ph,
201        "name": f.name,
202        "cat": cat,
203        "id": format!("{sid:#018x}"),
204        "pid": pid,
205        "tid": tid,
206        "ts": ts_micros(ts_nanos),
207        "args": args,
208    })
209}
210
211/// An instant event (`ph:"i"`) at the record's timestamp.
212fn instant_event(
213    f: &Flat,
214    cat: &str,
215    pid: u64,
216    tid: u32,
217    args: serde_json::Map<String, Json>,
218) -> Json {
219    json!({
220        "ph": "i",
221        "name": f.name,
222        "cat": cat,
223        "pid": pid,
224        "tid": tid,
225        "ts": ts_micros(f.ts_nanos),
226        "s": "t", // thread-scoped instant
227        "args": args,
228    })
229}
230
231/// Trace Event Format `ts` is microseconds (fractional ok). We keep nanosecond precision via the
232/// fraction and set `displayTimeUnit: "ns"` on the document.
233fn ts_micros(ts_nanos: u64) -> f64 {
234    ts_nanos as f64 / 1000.0
235}
236
237/// Reads a little-endian `u64` at `offset`, or `None` if out of range.
238fn read_u64(bytes: &[u8], offset: usize) -> Option<u64> {
239    bytes
240        .get(offset..offset + 8)
241        .map(|s| u64::from_le_bytes(s.try_into().unwrap()))
242}
243
244/// Decodes a field to a JSON value using only its schema descriptor (mirrors the Parquet decoder).
245fn decode_json(
246    field: &backbeat::wire::OwnedField,
247    bytes: &[u8],
248    intern: &HashMap<u32, String>,
249) -> Json {
250    let start = field.offset as usize;
251    let Some(slice) = bytes.get(start..start + field.width as usize) else {
252        return Json::Null;
253    };
254    // `field.width` is read from the (possibly corrupt or foreign) dump and is not guaranteed to
255    // match the natural width of `field.ty`, so every fixed-width read goes through a fallible
256    // conversion: a mismatch renders as `null` rather than panicking the tool. This mirrors the
257    // Parquet decoder, which is already fallible throughout.
258    match field.ty {
259        FieldType::U8 => le::<1>(slice).map_or(Json::Null, |b| json!(b[0])),
260        FieldType::U16 => le::<2>(slice).map_or(Json::Null, |b| json!(u16::from_le_bytes(b))),
261        FieldType::U32 => le::<4>(slice).map_or(Json::Null, |b| json!(u32::from_le_bytes(b))),
262        FieldType::U64 => le::<8>(slice).map_or(Json::Null, |b| json!(u64::from_le_bytes(b))),
263        FieldType::I8 => le::<1>(slice).map_or(Json::Null, |b| json!(b[0] as i8)),
264        FieldType::I16 => le::<2>(slice).map_or(Json::Null, |b| json!(i16::from_le_bytes(b))),
265        FieldType::I32 => le::<4>(slice).map_or(Json::Null, |b| json!(i32::from_le_bytes(b))),
266        FieldType::I64 => le::<8>(slice).map_or(Json::Null, |b| json!(i64::from_le_bytes(b))),
267        FieldType::Bool => slice.first().map_or(Json::Null, |b| json!(*b != 0)),
268        FieldType::Bytes => json!(hex(slice)),
269        FieldType::Enum { repr } => {
270            let Some(raw_bytes) = slice.get(..repr as usize) else {
271                return Json::Null;
272            };
273            let mut buf = [0u8; 8];
274            buf[..repr as usize].copy_from_slice(raw_bytes);
275            let raw = u64::from_le_bytes(buf);
276            match field.enum_labels.iter().find(|l| l.value == raw) {
277                Some(l) => json!(l.label),
278                None => json!(raw),
279            }
280        }
281        FieldType::Interned { .. } => match le::<4>(slice) {
282            Some(b) => {
283                let id = u32::from_le_bytes(b);
284                match intern.get(&id) {
285                    Some(s) => json!(s),
286                    None => json!(format!("#{id}")),
287                }
288            }
289            None => Json::Null,
290        },
291        _ => json!(hex(slice)),
292    }
293}
294
295/// Reads exactly `N` little-endian bytes from the front of `slice`, or `None` if the field's
296/// declared width is too small to hold the type. Guards against a corrupt/foreign dump whose
297/// `field.width` disagrees with `field.ty`.
298fn le<const N: usize>(slice: &[u8]) -> Option<[u8; N]> {
299    slice.get(..N)?.try_into().ok()
300}
301
302fn hex(bytes: &[u8]) -> String {
303    let mut s = String::with_capacity(bytes.len() * 2);
304    for b in bytes {
305        s.push_str(&format!("{b:02x}"));
306    }
307    s
308}
backbeat_cli/trace.rs

backbeat_cli/
trace.rs
