From c5c6d94b0d936a0b1fbaacb28c04b41c32ef00d6 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 8 Oct 2026 23:32:36 +0000 Subject: [PATCH] fix(cli,replay-node,message): keep the debug-only `_wire_size` stamp out of recordings and replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The daemon stamps `_wire_size` on every `dora topic` debug frame so `dora topic info` can charge the real on-wire size. `dora record --proxy` stored those frames with the stamp intact, and the replay node forwarded `metadata.parameters` verbatim into `send_output`, so every downstream node of a replayed proxy recording saw a stale `_wire_size` in its input metadata — a key documented as "never set on real node→node outputs". `dora topic echo` also printed it as if it were user metadata. Make `WIRE_SIZE` one of the internal keys `strip_internal_parameters` removes (the daemon stamps it after stripping, so `topic info` still sees it), and call that helper at the three consumers that store, print or re-send debug frames: - record --proxy: while rewriting the event (now in a `prepare_proxy_event` helper next to the existing wire→public id rewrite) - replay-node: before `send_output`, so recordings made before this fix replay clean too - topic echo: before rendering the metadata The node API already strips internal keys on receive, so a node can no longer see the stamp either. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01C38big8SF3jEa21cdm4q6B --- binaries/cli/src/command/record.rs | 98 ++++++++++++++++++++------ binaries/cli/src/command/topic/echo.rs | 12 +++- binaries/replay-node/src/main.rs | 33 ++++++++- libraries/message/src/metadata.rs | 22 ++++-- 4 files changed, 131 insertions(+), 34 deletions(-) diff --git a/binaries/cli/src/command/record.rs b/binaries/cli/src/command/record.rs index 0b3e5e06ab..ff24453493 100644 --- a/binaries/cli/src/command/record.rs +++ b/binaries/cli/src/command/record.rs @@ -5,12 +5,13 @@ use std::{ }; use clap::Args; -use dora_core::descriptor::Descriptor; +use dora_core::descriptor::{Descriptor, Node}; use dora_message::{ common::Timestamped, coordinator_to_cli::DataflowIdAndName, daemon_to_daemon::InterDaemonEvent, id::{DataId, NodeId}, + metadata::strip_internal_parameters, }; use dora_recording::{RecordEntry, RecordingHeader, RecordingWriter}; use eyre::{Context, bail}; @@ -664,28 +665,12 @@ fn run_record_proxy(args: Record) -> eyre::Result<()> { Err(_) => continue, }; - // The daemon reports the *wire* output id, which for a single - // `operator:` node is `op/image` where the descriptor (and - // `replay_node_outputs`) call it `image`. Rewrite it to the - // public id in both the entry index and the embedded event, so - // a `--proxy` recording is shaped exactly like one from the - // record node -- which stores the bare `source_output` in both - // places. Left as-is, replay declares `outputs: [image]` and - // then calls `send_output("op/image", ..)`; `validate_output` - // rejects it, `send_output` still returns `Ok(())`, and every - // message is dropped with a single warning (dora-rs/dora#2893). - let (node_id, output_id) = match &mut event.inner { - InterDaemonEvent::Output { - node_id, output_id, .. - } => { - if let Some(node) = nodes.get(node_id) { - *output_id = public_topic_output_id(node, output_id); - } - (node_id.to_string(), output_id.to_string()) - } - InterDaemonEvent::OutputClosed { .. } => continue, - // `InterDaemonEvent` is `#[non_exhaustive]`: skip events this build predates. - _ => continue, + // Rewrite the event in place so a `--proxy` recording is + // shaped exactly like one from the record node, which stores + // the bare `source_output` in both the entry and the event. + let Some((node_id, output_id)) = prepare_proxy_event(&mut event.inner, &nodes) + else { + continue; }; // Re-serialize rather than storing `payload`: the id inside the @@ -814,6 +799,42 @@ fn select_dataflow( }) } +/// Shape a debug-topic frame received by `dora record --proxy` like an event +/// recorded by the record node, returning its `(node_id, output_id)` entry key, +/// or `None` for an event that isn't an output. +/// +/// - The daemon reports the *wire* output id, which for a single `operator:` +/// node is `op/image` where the descriptor (and `replay_node_outputs`) call it +/// `image`. Rewrite it to the public id, so replay declares and sends the same +/// id. Left as-is, `send_output("op/image", ..)` is rejected by +/// `validate_output` and every message is dropped (dora-rs/dora#2893). +/// - Drop internal parameters, notably the daemon's `_wire_size` stamp. It +/// exists only for `dora topic info`'s bandwidth accounting on the debug +/// path; stored in the recording, replay would forward it into every +/// downstream node's input metadata. +fn prepare_proxy_event( + event: &mut InterDaemonEvent, + nodes: &HashMap, +) -> Option<(String, String)> { + match event { + InterDaemonEvent::Output { + node_id, + output_id, + metadata, + .. + } => { + if let Some(node) = nodes.get(node_id) { + *output_id = public_topic_output_id(node, output_id); + } + strip_internal_parameters(&mut metadata.parameters); + Some((node_id.to_string(), output_id.to_string())) + } + // `OutputClosed`, and (`InterDaemonEvent` is `#[non_exhaustive]`) + // events this build predates. + _ => None, + } +} + #[cfg(test)] mod tests { use super::*; @@ -837,6 +858,37 @@ mod tests { assert_eq!(parsed[0].1.to_string(), "op/out"); } + #[test] + fn proxy_event_drops_the_debug_wire_size_stamp() { + use dora_message::metadata::{Metadata, MetadataParameters, Parameter, WIRE_SIZE}; + + // The daemon stamps `_wire_size` on every debug frame for `dora topic + // info`. A proxy recording must not keep it: replay would forward it + // into every downstream node's input metadata. + let mut parameters = MetadataParameters::default(); + parameters.insert(WIRE_SIZE.to_string(), Parameter::Integer(42)); + parameters.insert("user_key".to_string(), Parameter::Integer(7)); + let mut event = InterDaemonEvent::Output { + dataflow_id: Uuid::new_v4(), + node_id: "camera".parse().unwrap(), + output_id: "image".parse().unwrap(), + metadata: Metadata::from_parameters( + dora_message::uhlc::HLC::default().new_timestamp(), + parameters, + ), + data: None, + }; + + let key = prepare_proxy_event(&mut event, &HashMap::new()); + + assert_eq!(key, Some(("camera".to_string(), "image".to_string()))); + let InterDaemonEvent::Output { metadata, .. } = event else { + unreachable!() + }; + assert!(!metadata.parameters.contains_key(WIRE_SIZE)); + assert!(metadata.parameters.contains_key("user_key")); + } + fn discovered_topics(yaml: &str) -> Vec { let descriptor: serde_yaml::Value = serde_yaml::from_str(yaml).unwrap(); discover_descriptor_outputs(&descriptor) diff --git a/binaries/cli/src/command/topic/echo.rs b/binaries/cli/src/command/topic/echo.rs index 0dddeaef3e..50ee1cb0f3 100644 --- a/binaries/cli/src/command/topic/echo.rs +++ b/binaries/cli/src/command/topic/echo.rs @@ -3,7 +3,11 @@ use std::{collections::HashMap, ptr::NonNull, sync::Arc, time::SystemTime}; use arrow::{buffer::OffsetBuffer, datatypes::Field}; use clap::Args; use colored::Colorize; -use dora_message::{common::Timestamped, daemon_to_daemon::InterDaemonEvent, metadata::Parameter}; +use dora_message::{ + common::Timestamped, + daemon_to_daemon::InterDaemonEvent, + metadata::{Parameter, strip_internal_parameters}, +}; use eyre::{Context, eyre}; use crate::{ @@ -185,7 +189,7 @@ fn inspect( match event.inner { InterDaemonEvent::Output { - metadata, + mut metadata, data, node_id, output_id, @@ -193,6 +197,10 @@ fn inspect( } => { use std::fmt::Write; + // The daemon stamps every debug frame with its on-wire size for + // `dora topic info`; it is not part of the node's metadata. + strip_internal_parameters(&mut metadata.parameters); + let display_output = nodes .get(&node_id) .map(|node| public_topic_output_id(node, &output_id)) diff --git a/binaries/replay-node/src/main.rs b/binaries/replay-node/src/main.rs index f8c4373875..d19dd892f2 100644 --- a/binaries/replay-node/src/main.rs +++ b/binaries/replay-node/src/main.rs @@ -3,7 +3,11 @@ use std::{ time::{Duration, Instant}, }; -use dora_message::{common::Timestamped, daemon_to_daemon::InterDaemonEvent}; +use dora_message::{ + common::Timestamped, + daemon_to_daemon::InterDaemonEvent, + metadata::{MetadataParameters, strip_internal_parameters}, +}; use dora_node_api::{ DoraArray, DoraNode, Event, EventStream, IntoArrow, TryRecvError, arrow_utils::decode_arrow_ipc, arrow_v59::array::NullArray, @@ -205,7 +209,7 @@ fn main() -> eyre::Result<()> { continue; } }; - node.send_output(output_id, metadata.parameters, array) + node.send_output(output_id, replay_parameters(metadata.parameters), array) .wrap_err("failed to send replay output")?; replayed += 1; } @@ -265,10 +269,21 @@ fn main() -> eyre::Result<()> { Ok(()) } +/// The metadata parameters to re-send for a recorded output. +/// +/// Recordings made with `dora record --proxy` before the proxy learned to +/// strip it carry the daemon's debug-only `_wire_size` stamp on every +/// message. Drop internal keys here, so they never reach downstream nodes. +fn replay_parameters(mut parameters: MetadataParameters) -> MetadataParameters { + strip_internal_parameters(&mut parameters); + parameters +} + #[cfg(test)] mod tests { use super::{ Replay, classify_event, decode_recorded_payload, pacing_gap, replay_emitted_nothing_usable, + replay_parameters, }; use dora_node_api::DoraArray; use dora_node_api::arrow_utils::encode_arrow_ipc; @@ -276,6 +291,20 @@ mod tests { use dora_node_api::{Event, StopCause}; use std::time::Duration; + #[test] + fn replay_drops_the_debug_wire_size_stamp() { + use dora_message::metadata::{MetadataParameters, Parameter, WIRE_SIZE}; + + let mut parameters = MetadataParameters::default(); + parameters.insert(WIRE_SIZE.to_string(), Parameter::Integer(42)); + parameters.insert("user_key".to_string(), Parameter::Integer(7)); + + let replayed = replay_parameters(parameters); + + assert!(!replayed.contains_key(WIRE_SIZE)); + assert!(replayed.contains_key("user_key")); + } + #[test] fn stop_event_ends_the_replay() { // Both stop causes end the replay. `Manual` is the one that matters in diff --git a/libraries/message/src/metadata.rs b/libraries/message/src/metadata.rs index b745ddca82..6a1d60d32f 100644 --- a/libraries/message/src/metadata.rs +++ b/libraries/message/src/metadata.rs @@ -348,7 +348,9 @@ pub const SCHEMA_HASH: &str = "_schema_hash"; /// accurate, the daemon stamps the original data-sample length here; the CLI /// measures this instead of the rebuilt stream length (dora-rs/dora#2584). /// -/// Debug/inspection path only — never set on real node→node outputs. +/// Debug/inspection path only — never set on real node→node outputs, and +/// removed by [`strip_internal_parameters`] so it can't leak into one through a +/// recorded or re-sent debug frame. pub const WIRE_SIZE: &str = "_wire_size"; /// Byte size to charge for a `dora topic` debug frame when accounting bandwidth. @@ -387,15 +389,21 @@ pub fn carries_pattern_correlation(params: &MetadataParameters) -> bool { || params.contains_key(GOAL_STATUS) } -/// Remove internal wire-protocol keys ([`SCHEMA_HASH`], [`FRAMING`]) from a -/// parameter map. Call this at every wire→user boundary: the keys are -/// meaningless after decode, and a stale [`SCHEMA_HASH`] forwarded from an -/// input's metadata into `send_output` parameters (a standard pattern, e.g. -/// replay) would ride onto outputs that don't overwrite it, making receivers -/// hash-mismatch and silently drop them (dora-rs/dora#2366 review). +/// Remove internal wire-protocol keys ([`SCHEMA_HASH`], [`FRAMING`], and the +/// debug-frame [`WIRE_SIZE`] stamp) from a parameter map. Call this at every +/// wire→user boundary: the keys are meaningless after decode, and a stale +/// [`SCHEMA_HASH`] forwarded from an input's metadata into `send_output` +/// parameters (a standard pattern, e.g. replay) would ride onto outputs that +/// don't overwrite it, making receivers hash-mismatch and silently drop them +/// (dora-rs/dora#2366 review). +/// +/// The daemon stamps [`WIRE_SIZE`] *after* stripping a debug frame, so `dora +/// topic info` still sees it; consumers that store, print or re-send debug +/// frames call this to drop it. pub fn strip_internal_parameters(params: &mut MetadataParameters) { params.remove(SCHEMA_HASH); params.remove(FRAMING); + params.remove(WIRE_SIZE); } /// FNV-1a-64 hash with a fixed seed (cross-process deterministic). Used to