Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 75 additions & 23 deletions binaries/cli/src/command/record.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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<NodeId, &Node>,
) -> 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::*;
Expand All @@ -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<String> {
let descriptor: serde_yaml::Value = serde_yaml::from_str(yaml).unwrap();
discover_descriptor_outputs(&descriptor)
Expand Down
12 changes: 10 additions & 2 deletions binaries/cli/src/command/topic/echo.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down Expand Up @@ -185,14 +189,18 @@ fn inspect(

match event.inner {
InterDaemonEvent::Output {
metadata,
mut metadata,
data,
node_id,
output_id,
..
} => {
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))
Expand Down
33 changes: 31 additions & 2 deletions binaries/replay-node/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -265,17 +269,42 @@ 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;
use dora_node_api::arrow_v59::array::Int32Array;
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
Expand Down
22 changes: 15 additions & 7 deletions libraries/message/src/metadata.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down
Loading