backbeat_cli/
inspect.rs

1// Copyright (c) 2026 Cameron Bytheway
2// SPDX-License-Identifier: MIT
3
4//! `backbeat inspect`: print a dump's envelope, schema registry, and per-shard record counts.
5//!
6//! Everything here is driven by the dump's embedded registry — there is no compiled-in knowledge of
7//! any event type. A record's `event_id` is matched against the registry to attribute it to an
8//! event and validate its size; unknown ids and size mismatches are reported rather than guessed.
9
10use anyhow::Result;
11use backbeat::{
12    record::RecordView,
13    ring::walk,
14    schema::{FieldRole, Phase},
15    wire::{DumpReader, OwnedSchema},
16};
17use bytes::Bytes;
18use std::{collections::BTreeMap, io::Write};
19
20/// Reads `bytes` as a dump and writes a human-readable summary to `out`.
21pub fn inspect(bytes: impl Into<Bytes>, out: &mut impl Write) -> Result<()> {
22    let bytes = bytes.into();
23    let total_len = bytes.len();
24    let reader = DumpReader::new(bytes).map_err(|e| anyhow::anyhow!("{e}"))?;
25    let schemas = reader.schemas().map_err(|e| anyhow::anyhow!("{e}"))?;
26    let intern = reader.intern_table().map_err(|e| anyhow::anyhow!("{e}"))?;
27    let shards = reader.shards().map_err(|e| anyhow::anyhow!("{e}"))?;
28
29    // Index schemas by id for record attribution.
30    let by_id: BTreeMap<u64, &OwnedSchema> = schemas.iter().map(|s| (s.id.get(), s)).collect();
31
32    writeln!(out, "envelope")?;
33    writeln!(out, "  size:      {total_len} bytes")?;
34    writeln!(out, "  flags:     {:#06x}", reader.flags())?;
35    writeln!(out, "  sections:  {}", reader.section_count())?;
36
37    writeln!(out, "\nschema registry ({} events)", schemas.len())?;
38    for s in &schemas {
39        let phase = match s.phase {
40            Phase::Enter => "  [span enter]",
41            Phase::Exit => "  [span exit]",
42            _ => "",
43        };
44        writeln!(
45            out,
46            "  {:#018x}  {}  ({} bytes, {} fields){}{}",
47            s.id.get(),
48            s.qualified_name,
49            s.record_size,
50            s.fields.len(),
51            phase,
52            s.description
53                .as_deref()
54                .map(|d| format!("  — {d}"))
55                .unwrap_or_default()
56        )?;
57        for f in &s.fields {
58            // Role marker: * key, $ span id, ^ parent span id.
59            let marker = match f.role {
60                FieldRole::Key => "*",
61                FieldRole::SpanId => "$",
62                FieldRole::ParentSpanId => "^",
63                _ => " ",
64            };
65            writeln!(
66                out,
67                "      {}{}: {:?}{}",
68                marker,
69                f.name,
70                f.ty,
71                f.unit
72                    .as_deref()
73                    .map(|u| format!(" [{u}]"))
74                    .unwrap_or_default()
75            )?;
76        }
77    }
78    writeln!(out, "  (* = key, $ = span id, ^ = parent span id)")?;
79
80    writeln!(out, "\nintern table: {} entries", intern.len())?;
81
82    // Walk each shard, attributing records to events and counting them. The closure is walk's
83    // validator: it accepts (and counts) only records whose event_id is in the registry with a
84    // matching record_size, and rejects anything else so walk resynchronizes past torn data.
85    writeln!(out, "\nshards ({})", shards.len())?;
86    let mut total = 0usize;
87    let mut per_event: BTreeMap<String, usize> = BTreeMap::new();
88
89    for shard in &shards {
90        let mut count = 0usize;
91        walk(
92            &shard.region,
93            shard.head as usize,
94            shard.capacity as usize,
95            |payload| match RecordView::parse(&payload[..]) {
96                Some(rec) => match by_id.get(&rec.event_id.get()) {
97                    Some(schema) if rec.fields.len() == schema.record_size as usize => {
98                        *per_event.entry(schema.qualified_name.clone()).or_default() += 1;
99                        count += 1;
100                        true
101                    }
102                    _ => false,
103                },
104                None => false,
105            },
106        );
107        total += count;
108        writeln!(
109            out,
110            "  shard {:>3}: head {:>10}  capacity {:>10}  {} records",
111            shard.shard_id, shard.head, shard.capacity, count
112        )?;
113    }
114
115    writeln!(out, "\nrecords: {total} total")?;
116    for (name, n) in &per_event {
117        writeln!(out, "  {name}: {n}")?;
118    }
119
120    Ok(())
121}
backbeat_cli/inspect.rs

backbeat_cli/
inspect.rs
