From e71b82be9cb21cbf7dfcb91eb041fbcc11237c72 Mon Sep 17 00:00:00 2001 From: finch Date: Wed, 19 Aug 2026 20:46:41 -0400 Subject: [PATCH 1/5] tracing: add the rumors-tracing adapter crate (rumors#35) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A new workspace crate bridging the observation hook into the tracing ecosystem: one INFO session span per observed session (kind, protocol, ordinal), a role-elected event when the election is decided, one DEBUG stream span per directed stream, and one DEBUG event per wire item carrying its exact length and an RFC 8949 diagnostic-notation-style rendering — structure unfolded, embedded-CBOR tags shown as <<...>>, the registered atom tags named, every dimension of the rendering bounded by constants rather than by the input. Message events are stamped from a session-scoped atomic counter, dogfooding the hook's documented interleaving-reconstruction pattern. The core crate's dependency surface is untouched: the adapter lives in crates/rumors-tracing and consumes only the public API. The README is derived via tools/readme, which gains the crate in its roster. --- Cargo.lock | 41 +++ Cargo.toml | 1 + crates/rumors-tracing/Cargo.toml | 15 ++ crates/rumors-tracing/README.md | 109 ++++++++ crates/rumors-tracing/src/lib.rs | 235 +++++++++++++++++ crates/rumors-tracing/src/render.rs | 218 ++++++++++++++++ crates/rumors-tracing/src/render/tests.rs | 156 ++++++++++++ crates/rumors-tracing/tests/adapter.rs | 292 ++++++++++++++++++++++ tools/readme | 5 + 9 files changed, 1072 insertions(+) create mode 100644 crates/rumors-tracing/Cargo.toml create mode 100644 crates/rumors-tracing/README.md create mode 100644 crates/rumors-tracing/src/lib.rs create mode 100644 crates/rumors-tracing/src/render.rs create mode 100644 crates/rumors-tracing/src/render/tests.rs create mode 100644 crates/rumors-tracing/tests/adapter.rs diff --git a/Cargo.lock b/Cargo.lock index e556ad00..00d268b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1979,6 +1979,16 @@ dependencies = [ "tokio-stream", ] +[[package]] +name = "rumors-tracing" +version = "0.1.0" +dependencies = [ + "ciborium", + "rumors", + "tokio", + "tracing", +] + [[package]] name = "rustc_version" version = "0.4.1" @@ -2469,6 +2479,37 @@ dependencies = [ "tokio", ] +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + [[package]] name = "typenum" version = "1.20.1" diff --git a/Cargo.toml b/Cargo.toml index 02aeebf9..1c29591f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ members = [ "crates/before", "crates/before-viz", + "crates/rumors-tracing", "crates/surface-scan", "crates/suanpan", ] diff --git a/crates/rumors-tracing/Cargo.toml b/crates/rumors-tracing/Cargo.toml new file mode 100644 index 00000000..8dd00a2b --- /dev/null +++ b/crates/rumors-tracing/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "rumors-tracing" +version = "0.1.0" +edition = "2024" +description = "A tracing adapter for rumors gossip sessions: spans per session and stream, one structured event per observed wire item." +license = "MPL-2.0" +readme = "README.md" + +[dependencies] +rumors = { path = "../.." } +tracing = "0.1" +ciborium = "0.2" + +[dev-dependencies] +tokio = { version = "1", features = ["rt", "macros"] } diff --git a/crates/rumors-tracing/README.md b/crates/rumors-tracing/README.md new file mode 100644 index 00000000..1d0d56e8 --- /dev/null +++ b/crates/rumors-tracing/README.md @@ -0,0 +1,109 @@ +# rumors-tracing + + + +A `tracing` adapter for `rumors` gossip sessions: spans per +session and per directed stream, one structured event per observed +wire item. + +`rumors` exposes its wire traffic through the bytes-level +observation hook in `rumors::observe`. This crate is that hook's +bridge into the `tracing` ecosystem: attach a `TracingObserver` +to a peer, install whatever `tracing` subscriber your application +already uses, and every session the peer enters becomes a span tree +with each protocol message a structured event inside it. Because +the wire is CBOR end to end, the adapter needs no knowledge of the +protocol's message vocabulary: it unfolds each item generically and +names only the registered rumors atom tags (`rumors::tags`) — +deep inspection of application payloads comes free, since they are +the application's own CBOR. + +## Quickstart + +```rust +use std::sync::Arc; + +use rumors::Peer; +use rumors_tracing::TracingObserver; + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<(), rumors::Error> { + // Install your subscriber first (tracing_subscriber::fmt(), + // a test collector, anything): the adapter emits, it never + // installs. + let alice = Peer::::seed() + .observe(Arc::new(TracingObserver::new())) + .into_rumors(); + alice.send("the meeting is at noon".to_string()); + + let (mut near, mut far) = rumors::link::memory(); + let (served, joined) = tokio::join!(alice.gossip(&mut far), async { + Peer::::bootstrap().join(&mut near).await + }); + served?; + joined?.expect("alice is established, not herself bootstrapping"); + Ok(()) +} +``` + +## What the adapter emits + +Everything is emitted under the target `rumors`, so one directive +(`rumors=debug`) scopes it in any filter. Field values that name +protocol vocabulary (`SessionKind`, +`Role`, …) are recorded in their debug form. + +- **One `session` span per observed session** (level `INFO`): + fields `kind` (`Gossip`, `Bootstrap`, `Retire`), `protocol`, and + `ordinal` (the peer's session counter, so concurrent sessions + stay distinguishable). +- **One `role elected` event** (level `INFO`, inside the session + span) when the session's role election is decided, with the + elected `role`. Sessions whose greetings carry equal versions + hold no election and emit no such event. +- **One `stream` span per directed stream** (level `DEBUG`, child + of the session span): fields `kind` (`control` or `data`) and + `direction`, plus `speaker` and `index` for data streams. +- **One `message` event per wire item** (level `DEBUG`, inside its + stream span): fields `ordinal` (see below), `len` (the item's + exact wire size in bytes), and `item` — the item rendered in RFC + 8949 diagnostic-notation style, structure unfolded, embedded-CBOR + tags (24, 63) shown as `<<…>>`, rumors atom tags named + (`version(h'…')`), byte strings as hex. The rendering is bounded: + long byte strings, deep nesting, and long renderings all elide + with explicit marks, so events stay cheap under megabyte supply + runs. + +## Ordering across streams + +A session's streams pump concurrently and the hook deliberately +imposes no cross-stream ordering, so subscriber-side timestamps +interleave events only as precisely as your subscriber's clock. +The `ordinal` field is the sharper tool: the adapter stamps every +`message` event from one session-scoped atomic counter, so sorting +a session's events by `ordinal` reconstructs the observed +interleaving exactly — the consumer-side pattern the hook's +documentation recommends, built in. + +## Cost and back-pressure + +Handlers run synchronously inside the session's stream tasks (see +the hook's back-pressure contract in `rumors::observe`): the +adapter therefore does bounded work per item — one parse plus one +capped rendering — and only when the `message` event is enabled by +your subscriber; a disabled target costs the enabled-check alone. +A subscriber that blocks inside `event` stalls the emitting stream, +exactly as any slow `StreamObserver` would: keep slow sinks +behind a channel. + +## When not to use it + +- **To watch the replica's *contents*** — what messages exist, + in what order — use `rumors`' own pull-based observers + (`rumors::Rumors::unordered_messages` and friends); the wire + view is the wrong altitude for application state. +- **To capture sessions for byte-exact replay or assertion**, + implement `rumors::observe::Observer` directly and keep the + bytes: this adapter renders for human eyes and elides by design. + + diff --git a/crates/rumors-tracing/src/lib.rs b/crates/rumors-tracing/src/lib.rs new file mode 100644 index 00000000..6b53e113 --- /dev/null +++ b/crates/rumors-tracing/src/lib.rs @@ -0,0 +1,235 @@ +//! A [`tracing`] adapter for [`rumors`] gossip sessions: spans per +//! session and per directed stream, one structured event per observed +//! wire item. +//! +//! `rumors` exposes its wire traffic through the bytes-level +//! observation hook in [`rumors::observe`]. This crate is that hook's +//! bridge into the `tracing` ecosystem: attach a [`TracingObserver`] +//! to a peer, install whatever `tracing` subscriber your application +//! already uses, and every session the peer enters becomes a span tree +//! with each protocol message a structured event inside it. Because +//! the wire is CBOR end to end, the adapter needs no knowledge of the +//! protocol's message vocabulary: it unfolds each item generically and +//! names only the registered rumors atom tags ([`rumors::tags`]) — +//! deep inspection of application payloads comes free, since they are +//! the application's own CBOR. +//! +//! # Quickstart +//! +//! ``` +//! use std::sync::Arc; +//! +//! use rumors::Peer; +//! use rumors_tracing::TracingObserver; +//! +//! #[tokio::main(flavor = "current_thread")] +//! async fn main() -> Result<(), rumors::Error> { +//! // Install your subscriber first (tracing_subscriber::fmt(), +//! // a test collector, anything): the adapter emits, it never +//! // installs. +//! let alice = Peer::::seed() +//! .observe(Arc::new(TracingObserver::new())) +//! .into_rumors(); +//! alice.send("the meeting is at noon".to_string()); +//! +//! let (mut near, mut far) = rumors::link::memory(); +//! let (served, joined) = tokio::join!(alice.gossip(&mut far), async { +//! Peer::::bootstrap().join(&mut near).await +//! }); +//! served?; +//! joined?.expect("alice is established, not herself bootstrapping"); +//! Ok(()) +//! } +//! ``` +//! +//! # What the adapter emits +//! +//! Everything is emitted under the target `rumors`, so one directive +//! (`rumors=debug`) scopes it in any filter. Field values that name +//! protocol vocabulary ([`SessionKind`](rumors::observe::SessionKind), +//! [`Role`], …) are recorded in their debug form. +//! +//! - **One `session` span per observed session** (level `INFO`): +//! fields `kind` (`Gossip`, `Bootstrap`, `Retire`), `protocol`, and +//! `ordinal` (the peer's session counter, so concurrent sessions +//! stay distinguishable). +//! - **One `role elected` event** (level `INFO`, inside the session +//! span) when the session's role election is decided, with the +//! elected `role`. Sessions whose greetings carry equal versions +//! hold no election and emit no such event. +//! - **One `stream` span per directed stream** (level `DEBUG`, child +//! of the session span): fields `kind` (`control` or `data`) and +//! `direction`, plus `speaker` and `index` for data streams. +//! - **One `message` event per wire item** (level `DEBUG`, inside its +//! stream span): fields `ordinal` (see below), `len` (the item's +//! exact wire size in bytes), and `item` — the item rendered in RFC +//! 8949 diagnostic-notation style, structure unfolded, embedded-CBOR +//! tags (24, 63) shown as `<<…>>`, rumors atom tags named +//! (`version(h'…')`), byte strings as hex. The rendering is bounded: +//! long byte strings, deep nesting, and long renderings all elide +//! with explicit marks, so events stay cheap under megabyte supply +//! runs. +//! +//! # Ordering across streams +//! +//! A session's streams pump concurrently and the hook deliberately +//! imposes no cross-stream ordering, so subscriber-side timestamps +//! interleave events only as precisely as your subscriber's clock. +//! The `ordinal` field is the sharper tool: the adapter stamps every +//! `message` event from one session-scoped atomic counter, so sorting +//! a session's events by `ordinal` reconstructs the observed +//! interleaving exactly — the consumer-side pattern the hook's +//! documentation recommends, built in. +//! +//! # Cost and back-pressure +//! +//! Handlers run synchronously inside the session's stream tasks (see +//! the hook's back-pressure contract in [`rumors::observe`]): the +//! adapter therefore does bounded work per item — one parse plus one +//! capped rendering — and only when the `message` event is enabled by +//! your subscriber; a disabled target costs the enabled-check alone. +//! A subscriber that blocks inside `event` stalls the emitting stream, +//! exactly as any slow [`StreamObserver`] would: keep slow sinks +//! behind a channel. +//! +//! # When not to use it +//! +//! - **To watch the replica's *contents*** — what messages exist, +//! in what order — use `rumors`' own pull-based observers +//! ([`rumors::Rumors::unordered_messages`] and friends); the wire +//! view is the wrong altitude for application state. +//! - **To capture sessions for byte-exact replay or assertion**, +//! implement [`rumors::observe::Observer`] directly and keep the +//! bytes: this adapter renders for human eyes and elides by design. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use rumors::observe::{ + Observer, Role, SessionInfo, SessionObserver, StreamId, StreamInfo, StreamObserver, +}; +use tracing::Span; + +mod render; + +/// The bridge from [`rumors::observe`] to [`tracing`]: attach one to a +/// peer and every session it enters is emitted as spans and events. +/// +/// Attach with [`rumors::Peer::observe`] (or +/// [`rumors::Bootstrap::observe`], to watch the joining session +/// itself): +/// +/// ``` +/// use std::sync::Arc; +/// +/// use rumors::Peer; +/// use rumors_tracing::TracingObserver; +/// +/// let peer = Peer::::seed().observe(Arc::new(TracingObserver::new())); +/// ``` +/// +/// The adapter is stateless between sessions; one instance serves +/// every session of a peer, concurrent sessions included. See the +/// crate docs for the emitted vocabulary. +#[derive(Debug, Default, Clone, Copy)] +pub struct TracingObserver { + // Purely a future-proofing seam: construction goes through + // `new`/`Default` so configuration (level choices, render caps) + // can arrive without breaking attachment sites. + _private: (), +} + +impl TracingObserver { + /// Creates an adapter; attach it with [`rumors::Peer::observe`]. + pub fn new() -> Self { + Self::default() + } +} + +impl Observer for TracingObserver { + fn session(&self, session: &SessionInfo) -> Option> { + let span = tracing::info_span!( + target: "rumors", + "session", + kind = ?session.kind, + protocol = ?session.protocol, + ordinal = session.ordinal, + ); + Some(Box::new(SessionAdapter { + span, + order: Arc::new(AtomicU64::new(0)), + })) + } +} + +/// One observed session: owns the session span and the ordinal +/// counter every stream of the session stamps its events from. +struct SessionAdapter { + span: Span, + order: Arc, +} + +impl SessionObserver for SessionAdapter { + fn elected(&self, role: Role) { + tracing::info!(target: "rumors", parent: &self.span, role = ?role, "role elected"); + } + + fn stream(&self, stream: &StreamInfo) -> Option> { + let span = match stream.id { + StreamId::Control => tracing::debug_span!( + target: "rumors", + parent: &self.span, + "stream", + kind = "control", + direction = ?stream.direction, + ), + StreamId::Data { speaker, index } => tracing::debug_span!( + target: "rumors", + parent: &self.span, + "stream", + kind = "data", + speaker = ?speaker, + index, + direction = ?stream.direction, + ), + // `StreamId` is non-exhaustive: a stream kind this adapter + // does not know is still worth a span, identified by its + // debug form. + other => tracing::debug_span!( + target: "rumors", + parent: &self.span, + "stream", + kind = ?other, + direction = ?stream.direction, + ), + }; + Some(Box::new(StreamAdapter { + span, + order: Arc::clone(&self.order), + })) + } +} + +/// One observed directed stream: owns the stream span and emits one +/// event per item. +struct StreamAdapter { + span: Span, + order: Arc, +} + +impl StreamObserver for StreamAdapter { + fn message(&mut self, bytes: &[u8]) { + // The ordinal must advance even when the event is disabled: + // enabling a subscriber mid-session would otherwise emit + // colliding ordinals, and the counter is the interleaving. + let ordinal = self.order.fetch_add(1, Ordering::Relaxed); + tracing::debug!( + target: "rumors", + parent: &self.span, + ordinal, + len = bytes.len(), + item = %render::item(bytes), + "message", + ); + } +} diff --git a/crates/rumors-tracing/src/render.rs b/crates/rumors-tracing/src/render.rs new file mode 100644 index 00000000..92e394ca --- /dev/null +++ b/crates/rumors-tracing/src/render.rs @@ -0,0 +1,218 @@ +//! Rendering one wire item into RFC 8949 diagnostic-notation-style text. +//! +//! The adapter's events carry the observed item as one text field, so +//! the rendering must stay legible and bounded no matter what crosses +//! the wire: structure is unfolded (arrays, maps, embedded-CBOR tags), +//! rumors' registered atom tags are named, and every dimension of the +//! output — byte-string hex, embedded-item unfolding, nesting depth, +//! and total length — is capped by a constant, never by the input. + +use std::fmt::Write; + +use ciborium::value::Value; +use rumors::tags::{CLOCK_TAG, PARTY_TAG, VERSION_TAG}; + +/// The embedded-CBOR tags whose byte strings are unfolded and shown as +/// their contents: tag 24 holds one encoded item, tag 63 an encoded +/// sequence. Diagnostic notation writes both as `<<…>>`. +const TAG_EMBEDDED_ITEM: u64 = 24; +const TAG_EMBEDDED_SEQUENCE: u64 = 63; + +/// How many bytes of a byte string are shown as hex before elision. +/// Version and party atoms fit well inside this; supply payloads and +/// digests elide to their prefix plus a length. +const SHOWN_BYTES: usize = 32; + +/// How many levels of embedded-CBOR byte strings are re-parsed and +/// unfolded. +/// +/// The wire nests two levels today (a supply run holds records); the +/// budget leaves headroom without letting adversarial nesting demand +/// unbounded re-parsing. +const UNFOLD_BUDGET: u8 = 4; + +/// How deep into one parsed item's structure the renderer descends. +/// Rendering recurses on the parsed value's shape, so this constant — +/// not the input — bounds the stack. +const DEPTH_BUDGET: u8 = 64; + +/// The rendered form's length cap in bytes. Once an item's rendering +/// crosses it, the remainder elides: events stay cheap even when a +/// megabyte supply run is observed. +const LENGTH_BUDGET: usize = 2048; + +/// Renders exactly one wire item as diagnostic-notation-style text. +/// +/// The hook's contract is one CBOR item per invocation; bytes that are +/// not that (undecodable, or carrying trailing garbage) render as an +/// explicit defect note plus a hex prefix rather than panicking — the +/// adapter observes, it never judges. +pub(crate) fn item(bytes: &[u8]) -> String { + let mut cursor = std::io::Cursor::new(bytes); + let mut out = String::new(); + match ciborium::de::from_reader::(&mut cursor) { + Ok(value) => { + render(&value, &mut out, UNFOLD_BUDGET, DEPTH_BUDGET); + let consumed = cursor.position() as usize; + if consumed != bytes.len() { + let _ = write!(out, " !trailing({} B)", bytes.len() - consumed); + } + } + Err(_) => { + out.push_str("!undecodable "); + hex(bytes, &mut out); + } + } + if out.len() > LENGTH_BUDGET { + // Truncation must land on a character boundary: the budget is + // in bytes, and the rendering carries multibyte characters + // (elision marks, escaped text) that may straddle it. + let mut cut = LENGTH_BUDGET; + while !out.is_char_boundary(cut) { + cut -= 1; + } + out.truncate(cut); + out.push('…'); + } + out +} + +/// Renders one parsed value, appending to `out`. +/// +/// `unfold` prices embedded-CBOR re-parses and `depth` prices descent +/// into the value's own structure; both only ever shrink, so the +/// recursion is bounded by the two constants above regardless of +/// input. Output growth is checked against the length budget at every +/// level so a wide value cannot buy unbounded work with small nesting. +fn render(value: &Value, out: &mut String, unfold: u8, depth: u8) { + if out.len() > LENGTH_BUDGET { + return; + } + let Some(deeper) = depth.checked_sub(1) else { + out.push('…'); + return; + }; + match value { + Value::Integer(n) => { + let _ = write!(out, "{}", i128::from(*n)); + } + Value::Float(f) => { + let _ = write!(out, "{f}"); + } + Value::Bool(b) => { + let _ = write!(out, "{b}"); + } + Value::Null => out.push_str("null"), + Value::Text(t) => { + let _ = write!(out, "\"{}\"", t.escape_debug()); + } + Value::Bytes(b) => hex(b, out), + Value::Array(items) => { + out.push('['); + for (i, item) in items.iter().enumerate() { + if i > 0 { + out.push_str(", "); + } + render(item, out, unfold, deeper); + if out.len() > LENGTH_BUDGET { + return; + } + } + out.push(']'); + } + Value::Map(entries) => { + out.push('{'); + for (i, (k, v)) in entries.iter().enumerate() { + if i > 0 { + out.push_str(", "); + } + render(k, out, unfold, deeper); + out.push_str(": "); + render(v, out, unfold, deeper); + if out.len() > LENGTH_BUDGET { + return; + } + } + out.push('}'); + } + Value::Tag(tag, inner) => render_tag(*tag, inner, out, unfold, deeper), + // `Value` is non-exhaustive; an item kind this renderer does not + // know is still an observed item, so show its debug form rather + // than dropping it. + other => { + let _ = write!(out, "{other:?}"); + } + } +} + +/// Renders one tagged value: rumors' atom tags by name, embedded-CBOR +/// tags unfolded as `<<…>>`, everything else as `tag(content)`. +fn render_tag(tag: u64, inner: &Value, out: &mut String, unfold: u8, depth: u8) { + let atom = match tag { + PARTY_TAG => Some("party"), + VERSION_TAG => Some("version"), + CLOCK_TAG => Some("clock"), + _ => None, + }; + if let Some(name) = atom { + let _ = write!(out, "{name}("); + render(inner, out, unfold, depth); + out.push(')'); + return; + } + if matches!(tag, TAG_EMBEDDED_ITEM | TAG_EMBEDDED_SEQUENCE) + && let Value::Bytes(encoded) = inner + && let Some(remaining) = unfold.checked_sub(1) + && let Some(rendered) = embedded(tag, encoded, remaining, depth) + { + let _ = write!(out, "{tag}(<<{rendered}>>)"); + return; + } + let _ = write!(out, "{tag}("); + render(inner, out, unfold, depth); + out.push(')'); +} + +/// Re-parses an embedded-CBOR byte string and renders its contents: +/// one item for tag 24, a whole sequence for tag 63. +/// +/// `None` when the bytes do not parse to exactly the promised shape — +/// the caller then falls back to the raw byte-string form, which is +/// always honest. +fn embedded(tag: u64, encoded: &[u8], unfold: u8, depth: u8) -> Option { + let mut cursor = std::io::Cursor::new(encoded); + let mut out = String::new(); + let mut first = true; + while (cursor.position() as usize) < encoded.len() { + if !first { + out.push_str(", "); + } + first = false; + let value = ciborium::de::from_reader::(&mut cursor).ok()?; + render(&value, &mut out, unfold, depth); + if tag == TAG_EMBEDDED_ITEM && (cursor.position() as usize) < encoded.len() { + return None; + } + if out.len() > LENGTH_BUDGET { + break; + } + } + Some(out) +} + +/// Appends a byte string's diagnostic form: full hex up to the shown +/// cap, then an elision with the true length. +fn hex(bytes: &[u8], out: &mut String) { + let shown = bytes.len().min(SHOWN_BYTES); + out.push_str("h'"); + for byte in &bytes[..shown] { + let _ = write!(out, "{byte:02x}"); + } + out.push('\''); + if shown < bytes.len() { + let _ = write!(out, "…({} B)", bytes.len()); + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/rumors-tracing/src/render/tests.rs b/crates/rumors-tracing/src/render/tests.rs new file mode 100644 index 00000000..8837b0af --- /dev/null +++ b/crates/rumors-tracing/src/render/tests.rs @@ -0,0 +1,156 @@ +use ciborium::value::Value; +use rumors::tags::{CLOCK_TAG, PARTY_TAG, VERSION_TAG}; + +use super::{LENGTH_BUDGET, SHOWN_BYTES, UNFOLD_BUDGET, item, render}; + +/// Encodes one value to CBOR bytes, the shape `item` consumes. +fn encoded(value: &Value) -> Vec { + let mut bytes = Vec::new(); + ciborium::ser::into_writer(value, &mut bytes).expect("test values encode"); + bytes +} + +/// Scalars render in diagnostic notation: integers as decimals, text +/// quoted, booleans and null literally. +#[test] +fn scalars_render_in_diagnostic_notation() { + assert_eq!(item(&encoded(&Value::Integer(170.into()))), "170"); + assert_eq!(item(&encoded(&Value::Text("rumors".into()))), "\"rumors\""); + assert_eq!(item(&encoded(&Value::Bool(true))), "true"); + assert_eq!(item(&encoded(&Value::Null)), "null"); +} + +/// Short byte strings render as full hex; long ones elide to the shown +/// prefix plus their true length, so no input can inflate the output. +#[test] +fn byte_strings_render_as_capped_hex() { + assert_eq!(item(&encoded(&Value::Bytes(vec![0xab, 0xcd]))), "h'abcd'"); + let long = vec![0x5a; SHOWN_BYTES + 9]; + let rendered = item(&encoded(&Value::Bytes(long.clone()))); + assert!(rendered.starts_with("h'")); + assert!(rendered.ends_with(&format!("…({} B)", long.len()))); +} + +/// Arrays and maps unfold with their elements rendered in place, the +/// shape a wire frame or greeting map arrives in. +#[test] +fn containers_unfold() { + let frame = Value::Array(vec![Value::Integer(162.into()), Value::Bytes(vec![0x01])]); + assert_eq!(item(&encoded(&frame)), "[162, h'01']"); + let map = Value::Map(vec![( + Value::Text("protocol".into()), + Value::Text("rumors".into()), + )]); + assert_eq!(item(&encoded(&map)), "{\"protocol\": \"rumors\"}"); +} + +/// The registered atom tags render by name, the thin naming layer over +/// the otherwise rumors-blind skeleton. +#[test] +fn atom_tags_render_by_name() { + for (tag, name) in [ + (PARTY_TAG, "party"), + (VERSION_TAG, "version"), + (CLOCK_TAG, "clock"), + ] { + let atom = Value::Tag(tag, Box::new(Value::Bytes(vec![0x42]))); + assert_eq!(item(&encoded(&atom)), format!("{name}(h'42')")); + } +} + +/// Embedded-CBOR byte strings unfold as `<<…>>`: tag 24 to its one +/// item, tag 63 to its whole sequence — the deep-inspection payoff for +/// supply runs and their records. +#[test] +fn embedded_cbor_tags_unfold() { + let one = Value::Tag( + 24, + Box::new(Value::Bytes(encoded(&Value::Integer(7.into())))), + ); + assert_eq!(item(&encoded(&one)), "24(<<7>>)"); + + let mut sequence = encoded(&Value::Integer(1.into())); + sequence.extend(encoded(&Value::Text("x".into()))); + let run = Value::Tag(63, Box::new(Value::Bytes(sequence))); + assert_eq!(item(&encoded(&run)), "63(<<1, \"x\">>)"); +} + +/// A tag-24 byte string that does not hold exactly one item (or does +/// not parse) falls back to the honest raw byte-string form instead of +/// guessing. +#[test] +fn malformed_embeddings_fall_back_to_bytes() { + let two_items = { + let mut bytes = encoded(&Value::Integer(1.into())); + bytes.extend(encoded(&Value::Integer(2.into()))); + bytes + }; + let tagged = Value::Tag(24, Box::new(Value::Bytes(two_items))); + assert_eq!(item(&encoded(&tagged)), "24(h'0102')"); + + let garbage = Value::Tag(63, Box::new(Value::Bytes(vec![0xff]))); + assert_eq!(item(&encoded(&garbage)), "63(h'ff')"); +} + +/// Unknown tags render by number with their content unfolded: the +/// renderer stays total over foreign vocabulary. +#[test] +fn unknown_tags_render_by_number() { + let foreign = Value::Tag(55799, Box::new(Value::Text("rumors".into()))); + assert_eq!(item(&encoded(&foreign)), "55799(\"rumors\")"); +} + +/// Undecodable bytes and trailing garbage render as explicit defect +/// notes: the renderer never panics on wire input, whatever arrives. +#[test] +fn defects_render_as_notes_not_panics() { + assert!(item(&[0xff]).starts_with("!undecodable ")); + let mut trailing = encoded(&Value::Integer(1.into())); + trailing.push(0x00); + assert_eq!(item(&trailing), "1 !trailing(1 B)"); +} + +/// Nesting past the depth budget elides instead of recursing: the +/// budget parameter, not the input's shape, bounds the renderer's +/// stack. +/// +/// Driven through `render` directly with a small budget so the +/// property is the renderer's own, independent of the decoder's +/// nesting limits. +#[test] +fn depth_is_bounded_by_the_budget() { + let mut value = Value::Integer(0.into()); + for _ in 0..8 { + value = Value::Array(vec![value]); + } + let mut shallow = String::new(); + render(&value, &mut shallow, UNFOLD_BUDGET, 4); + assert!(shallow.contains('…')); + let mut deep = String::new(); + render(&value, &mut deep, UNFOLD_BUDGET, 16); + assert_eq!(deep, "[[[[[[[[0]]]]]]]]"); +} + +/// A multibyte character straddling the length budget truncates to the +/// preceding character boundary instead of panicking. +#[test] +fn truncation_respects_character_boundaries() { + let text: String = "é".repeat(LENGTH_BUDGET); + let rendered = item(&encoded(&Value::Text(text))); + assert!(rendered.ends_with('…')); + assert!(rendered.len() <= LENGTH_BUDGET + '…'.len_utf8()); +} + +/// A rendering that crosses the length budget truncates with an +/// elision marker: events stay cheap under megabyte supply runs. +#[test] +fn length_is_bounded_by_the_budget() { + let wide = Value::Array( + (0..LENGTH_BUDGET) + .map(|i| Value::Integer((i as i64 % 10).into())) + .collect(), + ); + let rendered = item(&encoded(&wide)); + assert!(rendered.len() <= LENGTH_BUDGET + '…'.len_utf8()); + assert!(rendered.ends_with('…')); +} diff --git a/crates/rumors-tracing/tests/adapter.rs b/crates/rumors-tracing/tests/adapter.rs new file mode 100644 index 00000000..7b33eb95 --- /dev/null +++ b/crates/rumors-tracing/tests/adapter.rs @@ -0,0 +1,292 @@ +//! End-to-end proof that the adapter bridges real sessions faithfully. +//! +//! Driven through the public API only: a bootstrapped-then-gossiping +//! peer under a capturing subscriber yields the documented span tree, +//! and every item the hook delivers surfaces as exactly one `message` +//! event (held against a counting wrapper around the adapter itself). + +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use rumors::Peer; +use rumors::observe::{Observer, Role, SessionInfo, SessionObserver, StreamInfo, StreamObserver}; +use rumors_tracing::TracingObserver; +use tracing::span::{Attributes, Id, Record}; +use tracing::{Event, Metadata, Subscriber}; + +/// One captured span: its name, recorded fields, and explicit parent. +#[derive(Debug)] +struct SpanRecord { + name: String, + fields: BTreeMap, + parent: Option, +} + +/// One captured event: its recorded fields and explicit parent span. +#[derive(Debug)] +struct EventRecord { + fields: BTreeMap, + parent: Option, +} + +#[derive(Default)] +struct State { + spans: BTreeMap, + events: Vec, +} + +/// A minimal capturing subscriber: retains every span and event with +/// its fields, resolving parents from the explicit parent the adapter +/// always passes. +#[derive(Clone, Default)] +struct Capture(Arc>); + +/// Collects a span's or event's fields into string form via their +/// debug rendering. +struct Fields<'a>(&'a mut BTreeMap); + +impl tracing::field::Visit for Fields<'_> { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0 + .insert(field.name().to_string(), format!("{value:?}")); + } +} + +impl Subscriber for Capture { + fn enabled(&self, _: &Metadata<'_>) -> bool { + true + } + + fn new_span(&self, span: &Attributes<'_>) -> Id { + let mut state = self.0.lock().unwrap(); + let id = state.spans.len() as u64 + 1; + let mut fields = BTreeMap::new(); + span.record(&mut Fields(&mut fields)); + state.spans.insert( + id, + SpanRecord { + name: span.metadata().name().to_string(), + fields, + parent: span.parent().map(Id::into_u64), + }, + ); + Id::from_u64(id) + } + + fn record(&self, span: &Id, values: &Record<'_>) { + let mut state = self.0.lock().unwrap(); + let record = state.spans.get_mut(&span.into_u64()).expect("known span"); + values.record(&mut Fields(&mut record.fields)); + } + + fn record_follows_from(&self, _: &Id, _: &Id) {} + + fn event(&self, event: &Event<'_>) { + let mut fields = BTreeMap::new(); + event.record(&mut Fields(&mut fields)); + self.0.lock().unwrap().events.push(EventRecord { + fields, + parent: event.parent().map(Id::into_u64), + }); + } + + fn enter(&self, _: &Id) {} + fn exit(&self, _: &Id) {} +} + +/// Wraps the adapter and counts every item delivery, so the captured +/// events can be held against the hook's own invocation count. +struct Counting { + inner: TracingObserver, + messages: Arc, +} + +impl Observer for Counting { + fn session(&self, session: &SessionInfo) -> Option> { + let inner = self.inner.session(session)?; + Some(Box::new(CountingSession { + inner, + messages: Arc::clone(&self.messages), + })) + } +} + +struct CountingSession { + inner: Box, + messages: Arc, +} + +impl SessionObserver for CountingSession { + fn elected(&self, role: Role) { + self.inner.elected(role); + } + + fn stream(&self, stream: &StreamInfo) -> Option> { + let inner = self.inner.stream(stream)?; + Some(Box::new(CountingStream { + inner, + messages: Arc::clone(&self.messages), + })) + } +} + +struct CountingStream { + inner: Box, + messages: Arc, +} + +impl StreamObserver for CountingStream { + fn message(&mut self, bytes: &[u8]) { + self.messages.fetch_add(1, Ordering::Relaxed); + self.inner.message(bytes); + } +} + +/// The event's `message` text, as the debug visitor records it. +fn message_text(event: &EventRecord) -> Option<&str> { + event.fields.get("message").map(String::as_str) +} + +/// A bootstrapped-then-gossiping peer emits the documented span tree, +/// with exactly one `message` event per item the hook delivered. +/// +/// The tree: a `session` span per session with kind and ordinal, +/// `stream` spans beneath it, and a `role elected` event once the +/// election is decided. Every `message` event is stamped with a +/// session-dense ordinal and carries a rendering free of defect notes. +#[test] +fn adapter_bridges_real_sessions() { + let capture = Capture::default(); + let messages = Arc::new(AtomicU64::new(0)); + let counting = Arc::new(Counting { + inner: TracingObserver::new(), + messages: Arc::clone(&messages), + }); + + tracing::subscriber::with_default(capture.clone(), || { + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .expect("current-thread runtime"); + runtime.block_on(async { + let alice = Peer::::seed().into_rumors(); + alice.send("from the seed".to_string()); + + // Bob's own bootstrap session is observed through the + // builder; the attachment then follows the joined peer. + let (mut near, mut far) = rumors::link::memory(); + let (served, joined) = tokio::join!( + alice.gossip(&mut far), + Peer::::bootstrap() + .observe(counting.clone()) + .join(&mut near), + ); + served.expect("provider session"); + let bob = joined + .expect("bootstrap session") + .expect("alice is established, not herself bootstrapping") + .into_rumors(); + + // Diverge both replicas so the follow-up gossip elects a + // role and moves data-stream frames both ways. + bob.send("from bob".to_string()); + alice.send("from alice".to_string()); + let (mut near, mut far) = rumors::link::memory(); + let (a, b) = tokio::join!(alice.gossip(&mut far), bob.gossip(&mut near)); + a.expect("alice's gossip session"); + b.expect("bob's gossip session"); + }); + }); + + let hook_items = messages.load(Ordering::Relaxed); + assert!(hook_items > 0, "the sessions moved wire items"); + + let state = capture.0.lock().unwrap(); + + // The span tree: one session span per observed session, in entry + // order, with stream spans parented beneath. + let sessions: Vec<(&u64, &SpanRecord)> = state + .spans + .iter() + .filter(|(_, s)| s.name == "session") + .collect(); + assert_eq!(sessions.len(), 2, "bootstrap then gossip: {sessions:?}"); + assert!(sessions[0].1.fields["kind"].contains("Bootstrap")); + assert_eq!(sessions[0].1.fields["ordinal"], "0"); + assert!(sessions[1].1.fields["kind"].contains("Gossip")); + assert_eq!(sessions[1].1.fields["ordinal"], "1"); + for (_, session) in &sessions { + assert!(session.fields["protocol"].contains("V2")); + assert_eq!(session.parent, None); + } + + for (id, span) in state.spans.iter().filter(|(_, s)| s.name == "stream") { + let parent = span.parent.expect("stream spans sit under a session"); + assert_eq!( + state.spans[&parent].name, "session", + "stream span {id} parents a session" + ); + } + let gossip_id = *sessions[1].0; + let gossip_streams: Vec<&SpanRecord> = state + .spans + .values() + .filter(|s| s.name == "stream" && s.parent == Some(gossip_id)) + .collect(); + assert!( + gossip_streams + .iter() + .any(|s| s.fields["kind"] == "\"control\""), + "the gossip session's control stream is observed" + ); + assert!( + gossip_streams + .iter() + .any(|s| s.fields["kind"] == "\"data\""), + "divergent gossip opens observed data streams" + ); + + // The election is reported into the diverged gossip session. + assert!( + state + .events + .iter() + .any(|e| { message_text(e) == Some("role elected") && e.parent == Some(gossip_id) }), + "the diverged gossip session elects a role" + ); + + // Every hook delivery surfaced as exactly one message event, and + // every rendering is a clean single item. + let message_events: Vec<&EventRecord> = state + .events + .iter() + .filter(|e| message_text(e) == Some("message")) + .collect(); + assert_eq!(message_events.len() as u64, hook_items); + for event in &message_events { + let item = &event.fields["item"]; + assert!(!item.is_empty()); + assert!(!item.contains("!undecodable"), "clean item: {item}"); + assert!(!item.contains("!trailing"), "single item: {item}"); + } + + // Sorting one session's events by ordinal reconstructs the + // observed interleaving: per session, the stamps are exactly + // 0..n, each used once. + for (session_id, _) in &sessions { + let mut ordinals: Vec = message_events + .iter() + .filter(|e| { + let stream = e.parent.expect("message events sit in stream spans"); + state.spans[&stream].parent == Some(**session_id) + }) + .map(|e| e.fields["ordinal"].parse().expect("ordinal is a number")) + .collect(); + ordinals.sort_unstable(); + let expected: Vec = (0..ordinals.len() as u64).collect(); + assert_eq!( + ordinals, expected, + "session {session_id} ordinals are dense" + ); + } +} diff --git a/tools/readme b/tools/readme index 7e6fa110..09bdb37e 100755 --- a/tools/readme +++ b/tools/readme @@ -77,6 +77,11 @@ class Crate: CRATES = [ Crate(name="rumors", readme="README.md"), + Crate( + name="rumors-tracing", + readme="crates/rumors-tracing/README.md", + rdme_args=["-w", "rumors-tracing"], + ), Crate( name="suanpan", readme="crates/suanpan/README.md", From ce9378aa54a37039b0db2497c963f52179064c37 Mon Sep 17 00:00:00 2001 From: finch Date: Wed, 19 Aug 2026 22:09:55 -0400 Subject: [PATCH 2/5] tracing: the adapter numbers its own sessions The hook's session identity deliberately carries no ordinal (session numbering is consumer-side, like message interleaving), so the adapter counts the sessions it observes itself and stamps each session span from that internal counter; emitted vocabulary and field names are unchanged. Merges the ruling commits from the wire lane. Owner-ruled (rumors#35 follow-up rulings, 2026-08-19). --- crates/rumors-tracing/README.md | 6 ++++-- crates/rumors-tracing/src/lib.rs | 24 +++++++++++++----------- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/crates/rumors-tracing/README.md b/crates/rumors-tracing/README.md index 1d0d56e8..77d659b1 100644 --- a/crates/rumors-tracing/README.md +++ b/crates/rumors-tracing/README.md @@ -55,8 +55,10 @@ protocol vocabulary (`SessionKind`, - **One `session` span per observed session** (level `INFO`): fields `kind` (`Gossip`, `Bootstrap`, `Retire`), `protocol`, and - `ordinal` (the peer's session counter, so concurrent sessions - stay distinguishable). + `ordinal` — the adapter's own count of the sessions it has + observed, so concurrent sessions stay distinguishable. (The hook + deliberately carries no session number; numbering is the + observer's concern, and this adapter counts internally.) - **One `role elected` event** (level `INFO`, inside the session span) when the session's role election is decided, with the elected `role`. Sessions whose greetings carry equal versions diff --git a/crates/rumors-tracing/src/lib.rs b/crates/rumors-tracing/src/lib.rs index 6b53e113..879ade59 100644 --- a/crates/rumors-tracing/src/lib.rs +++ b/crates/rumors-tracing/src/lib.rs @@ -51,8 +51,10 @@ //! //! - **One `session` span per observed session** (level `INFO`): //! fields `kind` (`Gossip`, `Bootstrap`, `Retire`), `protocol`, and -//! `ordinal` (the peer's session counter, so concurrent sessions -//! stay distinguishable). +//! `ordinal` — the adapter's own count of the sessions it has +//! observed, so concurrent sessions stay distinguishable. (The hook +//! deliberately carries no session number; numbering is the +//! observer's concern, and this adapter counts internally.) //! - **One `role elected` event** (level `INFO`, inside the session //! span) when the session's role election is decided, with the //! elected `role`. Sessions whose greetings carry equal versions @@ -128,15 +130,15 @@ mod render; /// let peer = Peer::::seed().observe(Arc::new(TracingObserver::new())); /// ``` /// -/// The adapter is stateless between sessions; one instance serves -/// every session of a peer, concurrent sessions included. See the -/// crate docs for the emitted vocabulary. -#[derive(Debug, Default, Clone, Copy)] +/// One instance serves every session of a peer, concurrent sessions +/// included; its only state is the counter it numbers their spans +/// from. See the crate docs for the emitted vocabulary. +#[derive(Debug, Default)] pub struct TracingObserver { - // Purely a future-proofing seam: construction goes through - // `new`/`Default` so configuration (level choices, render caps) - // can arrive without breaking attachment sites. - _private: (), + /// Sessions this adapter has observed: the next span's `ordinal`. + /// The hook carries no session number, so the adapter counts for + /// itself — relaxed suffices, `session` being `&self`-concurrent. + sessions: AtomicU64, } impl TracingObserver { @@ -153,7 +155,7 @@ impl Observer for TracingObserver { "session", kind = ?session.kind, protocol = ?session.protocol, - ordinal = session.ordinal, + ordinal = self.sessions.fetch_add(1, Ordering::Relaxed), ); Some(Box::new(SessionAdapter { span, From 8dcdadeba15853e9715ab6a22cfb03edd3d74b9b Mon Sep 17 00:00:00 2001 From: finch Date: Thu, 20 Aug 2026 10:07:34 -0400 Subject: [PATCH 3/5] rumors-tracing: true up the cost doc and pin numbering and unfold boundaries Three review-driven changes, all local to the adapter crate: - The crate doc's cost sentence now states the full disabled-target cost: the enabled check plus one relaxed atomic increment (the message ordinal must advance even unobserved, or enabling a subscriber mid-session would emit colliding ordinals). README re-derived from the rustdoc. - A concurrent-sessions test: two counterparties gossip with clones of one observed peer's handle inside one tokio::join! on a current-thread runtime, so the sessions interleave at await points. It pins that the two session spans carry ordinals {0, 1} as a set and that each session's message ordinals stay dense from 0 -- session numbering under genuine concurrency, not only sequentially. - An unfold-budget boundary test: tag-24 embedded CBOR nested one level past UNFOLD_BUDGET renders the innermost embedded byte string as raw hex, pinning that the budget spans embedded-CBOR boundaries rather than resetting at each re-parse. Both new tests were held against their known-bad mechanisms before landing: a budget that resets per boundary and a session counter that stops advancing each fail the respective test. --- crates/rumors-tracing/README.md | 4 +- crates/rumors-tracing/src/lib.rs | 4 +- crates/rumors-tracing/src/render/tests.rs | 23 ++++ crates/rumors-tracing/tests/adapter.rs | 128 ++++++++++++++++++++++ 4 files changed, 157 insertions(+), 2 deletions(-) diff --git a/crates/rumors-tracing/README.md b/crates/rumors-tracing/README.md index 77d659b1..2047f039 100644 --- a/crates/rumors-tracing/README.md +++ b/crates/rumors-tracing/README.md @@ -93,7 +93,9 @@ Handlers run synchronously inside the session's stream tasks (see the hook's back-pressure contract in `rumors::observe`): the adapter therefore does bounded work per item — one parse plus one capped rendering — and only when the `message` event is enabled by -your subscriber; a disabled target costs the enabled-check alone. +your subscriber; a disabled target costs the enabled check plus one +relaxed atomic increment (the ordinal must advance even unobserved; +see `StreamAdapter::message`). A subscriber that blocks inside `event` stalls the emitting stream, exactly as any slow `StreamObserver` would: keep slow sinks behind a channel. diff --git a/crates/rumors-tracing/src/lib.rs b/crates/rumors-tracing/src/lib.rs index 879ade59..e6268dd7 100644 --- a/crates/rumors-tracing/src/lib.rs +++ b/crates/rumors-tracing/src/lib.rs @@ -89,7 +89,9 @@ //! the hook's back-pressure contract in [`rumors::observe`]): the //! adapter therefore does bounded work per item — one parse plus one //! capped rendering — and only when the `message` event is enabled by -//! your subscriber; a disabled target costs the enabled-check alone. +//! your subscriber; a disabled target costs the enabled check plus one +//! relaxed atomic increment (the ordinal must advance even unobserved; +//! see `StreamAdapter::message`). //! A subscriber that blocks inside `event` stalls the emitting stream, //! exactly as any slow [`StreamObserver`] would: keep slow sinks //! behind a channel. diff --git a/crates/rumors-tracing/src/render/tests.rs b/crates/rumors-tracing/src/render/tests.rs index 8837b0af..9f2f4007 100644 --- a/crates/rumors-tracing/src/render/tests.rs +++ b/crates/rumors-tracing/src/render/tests.rs @@ -92,6 +92,29 @@ fn malformed_embeddings_fall_back_to_bytes() { assert_eq!(item(&encoded(&garbage)), "63(h'ff')"); } +/// Tag-24 embedded CBOR nested one level past the unfold budget stops +/// unfolding at the boundary: the innermost embedded byte string +/// renders as raw hex (`h'…'`), not as its decoded contents. +/// +/// Every level the budget covers unfolds as `<<…>>`. This pins that +/// the budget spans embedded-CBOR boundaries — each re-parse draws +/// down the one shared budget rather than starting a fresh one. +#[test] +fn unfold_budget_spans_embedded_boundaries() { + // One more tag-24 level than the budget can unfold. + let mut value = Value::Integer(7.into()); + for _ in 0..=UNFOLD_BUDGET { + value = Value::Tag(24, Box::new(Value::Bytes(encoded(&value)))); + } + // The innermost tag's byte string (the encoding of 7) stays raw; + // every level above it unfolds. + let mut expected = "24(h'07')".to_string(); + for _ in 0..UNFOLD_BUDGET { + expected = format!("24(<<{expected}>>)"); + } + assert_eq!(item(&encoded(&value)), expected); +} + /// Unknown tags render by number with their content unfolded: the /// renderer stays total over foreign vocabulary. #[test] diff --git a/crates/rumors-tracing/tests/adapter.rs b/crates/rumors-tracing/tests/adapter.rs index 7b33eb95..fb6f90f8 100644 --- a/crates/rumors-tracing/tests/adapter.rs +++ b/crates/rumors-tracing/tests/adapter.rs @@ -290,3 +290,131 @@ fn adapter_bridges_real_sessions() { ); } } + +/// Concurrent sessions through clones of one observed handle number +/// cleanly: the two session spans carry the ordinals {0, 1} as a set, +/// and each session's message ordinals are dense from 0. +/// +/// The ordinals are held as a set because order between concurrent +/// sessions is unspecified. +/// Two counterparties gossip with the observed peer inside one +/// `tokio::join!` on a current-thread runtime, so the sessions +/// interleave at await points — the re-entrancy of the adapter's +/// `session` and the sharing of its counters is the property under +/// test, not thread parallelism. +#[test] +fn concurrent_sessions_number_cleanly() { + let capture = Capture::default(); + + tracing::subscriber::with_default(capture.clone(), || { + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .expect("current-thread runtime"); + runtime.block_on(async { + let alice = Peer::::seed().into_rumors(); + alice.send("from the seed".to_string()); + + // Bootstrap both counterparties unobserved, and attach the + // adapter to bob only after his join: the two concurrent + // gossip sessions are then the first sessions the adapter + // sees. + let (mut near, mut far) = rumors::link::memory(); + let (served, joined) = tokio::join!( + alice.gossip(&mut far), + Peer::::bootstrap().join(&mut near), + ); + served.expect("provider session"); + let bob = joined + .expect("bootstrap session") + .expect("alice is established, not herself bootstrapping") + .observe(Arc::new(TracingObserver::new())) + .into_rumors(); + + let (mut near, mut far) = rumors::link::memory(); + let (served, joined) = tokio::join!( + alice.gossip(&mut far), + Peer::::bootstrap().join(&mut near), + ); + served.expect("provider session"); + let carol = joined + .expect("bootstrap session") + .expect("alice is established, not herself bootstrapping") + .into_rumors(); + + // Diverge all three replicas so both sessions elect roles + // and move data-stream frames. + alice.send("from alice".to_string()); + bob.send("from bob".to_string()); + carol.send("from carol".to_string()); + + // Both of bob's sessions run inside one join, through the + // one shared observer, over separate links. + let bob_too = bob.clone(); + let (mut near_a, mut far_a) = rumors::link::memory(); + let (mut near_c, mut far_c) = rumors::link::memory(); + let (a, b1, b2, c) = tokio::join!( + alice.gossip(&mut far_a), + bob.gossip(&mut near_a), + bob_too.gossip(&mut near_c), + carol.gossip(&mut far_c), + ); + a.expect("alice's session"); + b1.expect("bob's session with alice"); + b2.expect("bob's session with carol"); + c.expect("carol's session"); + }); + }); + + let state = capture.0.lock().unwrap(); + + // Exactly the two concurrent gossip sessions were observed, and + // their ordinals are {0, 1} as a set: each session numbered once, + // no collision and no gap, whichever entered first. + let sessions: Vec<(&u64, &SpanRecord)> = state + .spans + .iter() + .filter(|(_, s)| s.name == "session") + .collect(); + assert_eq!(sessions.len(), 2, "two concurrent sessions: {sessions:?}"); + let mut ordinals: Vec<&str> = sessions + .iter() + .map(|(_, s)| s.fields["ordinal"].as_str()) + .collect(); + ordinals.sort_unstable(); + assert_eq!( + ordinals, + ["0", "1"], + "session ordinals are the set {{0, 1}}" + ); + for (_, session) in &sessions { + assert!(session.fields["kind"].contains("Gossip")); + } + + // Each session's message ordinals are dense from 0: the sessions + // share the adapter but not their message counters. + let message_events: Vec<&EventRecord> = state + .events + .iter() + .filter(|e| message_text(e) == Some("message")) + .collect(); + for (session_id, _) in &sessions { + let mut ordinals: Vec = message_events + .iter() + .filter(|e| { + let stream = e.parent.expect("message events sit in stream spans"); + state.spans[&stream].parent == Some(**session_id) + }) + .map(|e| e.fields["ordinal"].parse().expect("ordinal is a number")) + .collect(); + assert!( + !ordinals.is_empty(), + "session {session_id} moved wire items" + ); + ordinals.sort_unstable(); + let expected: Vec = (0..ordinals.len() as u64).collect(); + assert_eq!( + ordinals, expected, + "session {session_id} ordinals are dense" + ); + } +} From 5b92ee9068d53f27173637abea4797534c7a63f0 Mon Sep 17 00:00:00 2001 From: finch Date: Thu, 20 Aug 2026 20:12:26 -0400 Subject: [PATCH 4/5] The cost doc keeps its reason inline, without citing a private symbol Public crate rustdoc must not point at source-only items; the parenthetical already states why the ordinal advances unobserved. --- crates/rumors-tracing/README.md | 3 +-- crates/rumors-tracing/src/lib.rs | 3 +-- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/crates/rumors-tracing/README.md b/crates/rumors-tracing/README.md index 2047f039..71e5d3fc 100644 --- a/crates/rumors-tracing/README.md +++ b/crates/rumors-tracing/README.md @@ -94,8 +94,7 @@ the hook's back-pressure contract in `rumors::observe`): the adapter therefore does bounded work per item — one parse plus one capped rendering — and only when the `message` event is enabled by your subscriber; a disabled target costs the enabled check plus one -relaxed atomic increment (the ordinal must advance even unobserved; -see `StreamAdapter::message`). +relaxed atomic increment (the ordinal must advance even unobserved). A subscriber that blocks inside `event` stalls the emitting stream, exactly as any slow `StreamObserver` would: keep slow sinks behind a channel. diff --git a/crates/rumors-tracing/src/lib.rs b/crates/rumors-tracing/src/lib.rs index e6268dd7..2e446518 100644 --- a/crates/rumors-tracing/src/lib.rs +++ b/crates/rumors-tracing/src/lib.rs @@ -90,8 +90,7 @@ //! adapter therefore does bounded work per item — one parse plus one //! capped rendering — and only when the `message` event is enabled by //! your subscriber; a disabled target costs the enabled check plus one -//! relaxed atomic increment (the ordinal must advance even unobserved; -//! see `StreamAdapter::message`). +//! relaxed atomic increment (the ordinal must advance even unobserved). //! A subscriber that blocks inside `event` stalls the emitting stream, //! exactly as any slow [`StreamObserver`] would: keep slow sinks //! behind a channel. From ab58e185fb61a6d22971c7e40f77576b520e978b Mon Sep 17 00:00:00 2001 From: finch Date: Thu, 20 Aug 2026 20:16:36 -0400 Subject: [PATCH 5/5] The adapter tests hold send to its typed admission contract Rumors::send is fallible: admission runs the receiver's decode and can reject. The tests now assert the sends they depend on were admitted, instead of discarding the verdict they were built to observe. --- crates/rumors-tracing/tests/adapter.rs | 26 +++++++++++++++++++------- 1 file changed, 19 insertions(+), 7 deletions(-) diff --git a/crates/rumors-tracing/tests/adapter.rs b/crates/rumors-tracing/tests/adapter.rs index fb6f90f8..2d85e02c 100644 --- a/crates/rumors-tracing/tests/adapter.rs +++ b/crates/rumors-tracing/tests/adapter.rs @@ -170,7 +170,9 @@ fn adapter_bridges_real_sessions() { .expect("current-thread runtime"); runtime.block_on(async { let alice = Peer::::seed().into_rumors(); - alice.send("from the seed".to_string()); + alice + .send("from the seed".to_string()) + .expect("a flat string admits"); // Bob's own bootstrap session is observed through the // builder; the attachment then follows the joined peer. @@ -189,8 +191,11 @@ fn adapter_bridges_real_sessions() { // Diverge both replicas so the follow-up gossip elects a // role and moves data-stream frames both ways. - bob.send("from bob".to_string()); - alice.send("from alice".to_string()); + bob.send("from bob".to_string()) + .expect("a flat string admits"); + alice + .send("from alice".to_string()) + .expect("a flat string admits"); let (mut near, mut far) = rumors::link::memory(); let (a, b) = tokio::join!(alice.gossip(&mut far), bob.gossip(&mut near)); a.expect("alice's gossip session"); @@ -312,7 +317,9 @@ fn concurrent_sessions_number_cleanly() { .expect("current-thread runtime"); runtime.block_on(async { let alice = Peer::::seed().into_rumors(); - alice.send("from the seed".to_string()); + alice + .send("from the seed".to_string()) + .expect("a flat string admits"); // Bootstrap both counterparties unobserved, and attach the // adapter to bob only after his join: the two concurrent @@ -343,9 +350,14 @@ fn concurrent_sessions_number_cleanly() { // Diverge all three replicas so both sessions elect roles // and move data-stream frames. - alice.send("from alice".to_string()); - bob.send("from bob".to_string()); - carol.send("from carol".to_string()); + alice + .send("from alice".to_string()) + .expect("a flat string admits"); + bob.send("from bob".to_string()) + .expect("a flat string admits"); + carol + .send("from carol".to_string()) + .expect("a flat string admits"); // Both of bob's sessions run inside one join, through the // one shared observer, over separate links.