fix(node): record zenoh-delivered inputs in the write_events_to log

apis/rust/node/src/daemon_connection/node_integration_testing.rs

@@ -245,44 +245,107 @@ pub fn convert_output_to_json(
     start_timestamp: Timestamp,
     skip_output_time_offsets: bool,
 ) -> eyre::Result<serde_json::Map<String, serde_json::Value>> {
+    let mut output = json_header(
+        output_id,
+        metadata,
+        start_timestamp,
+        skip_output_time_offsets,
+    );
+    if data.is_some() {
+        let data_array = data_to_arrow_array(data.clone().map(std::sync::Arc::unwrap_or_clone))
+            .context("failed to convert output to arrow array")?;
+        append_arrow_array_json(&mut output, data_array)?;
+    }
+    Ok(output)
+}
+
+/// Serialize an already-decoded Arrow array (e.g. an input received over the
+/// zenoh data plane) into the same JSON shape as [`convert_output_to_json`].
+///
+/// The daemon-path `Input` events reach the recorder as an encoded
+/// [`DataMessage`], but zenoh-delivered inputs arrive already decoded as an
+/// `ArrayData`, so this is the entry point for recording those.
+pub fn convert_arrow_input_to_json(
+    input_id: &dora_message::id::DataId,
+    metadata: &Metadata,
+    data: arrow::array::ArrayRef,
+    start_timestamp: Timestamp,
+    skip_output_time_offsets: bool,
+) -> eyre::Result<serde_json::Map<String, serde_json::Value>> {
+    let mut output = json_header(
+        input_id,
+        metadata,
+        start_timestamp,
+        skip_output_time_offsets,
+    );
+    append_arrow_array_json(&mut output, data)?;
+    Ok(output)
+}
+
+/// Build the `id` (+ optional `time_offset_secs`) prefix shared by every
+/// recorded input/output event.
+fn json_header(
+    id: &dora_message::id::DataId,
+    metadata: &Metadata,
+    start_timestamp: Timestamp,
+    skip_output_time_offsets: bool,
+) -> serde_json::Map<String, serde_json::Value> {
     let mut output = serde_json::Map::new();
-    output.insert("id".into(), output_id.to_string().into());
+    output.insert("id".into(), id.to_string().into());
     if !skip_output_time_offsets {
-        let time_offset = metadata.timestamp().get_diff_duration(&start_timestamp);
+        let input_ts = metadata.timestamp();
+        // A zenoh-delivered input can carry a remote HLC timestamp that
+        // precedes this node's `start_timestamp` (a remote clock that is
+        // behind, or a producer that started earlier). `get_diff_duration` is
+        // an unguarded NTP64 (`u64`) subtraction that would underflow — a debug
+        // panic (which unwinds past `add_event`'s `Err` guard and kills the
+        // event loop) or a release wraparound to a garbage offset. Clamp to
+        // zero when the input predates start. The daemon path only ever records
+        // locally-timestamped inputs (always >= start), so this is a no-op
+        // there.
+        let time_offset = if input_ts.get_time() >= start_timestamp.get_time() {
+            input_ts.get_diff_duration(&start_timestamp)
+        } else {
+            std::time::Duration::ZERO
+        };
         output.insert("time_offset_secs".into(), time_offset.as_secs_f64().into());
     }
-    if data.is_some() {
-        let data_array = data_to_arrow_array(data.clone().map(std::sync::Arc::unwrap_or_clone))
-            .context("failed to convert output to arrow array")?;
+    output
+}
 
-        let data_type_json = serde_json::to_value(data_array.data_type())
-            .context("failed to serialize data type as JSON")?;
+/// Encode `data_array` into the `data` / `data_type` fields of a recorded
+/// event's JSON object.
+fn append_arrow_array_json(
+    output: &mut serde_json::Map<String, serde_json::Value>,
+    data_array: arrow::array::ArrayRef,
+) -> eyre::Result<()> {
+    let data_type_json = serde_json::to_value(data_array.data_type())
+        .context("failed to serialize data type as JSON")?;
 
-        let batch = RecordBatch::try_from_iter([("inner", data_array)])
-            .context("failed to create RecordBatch")?;
+    let batch = RecordBatch::try_from_iter([("inner", data_array)])
+        .context("failed to create RecordBatch")?;
 
-        let mut writer = arrow_json::ArrayWriter::new(Vec::new());
-        writer
-            .write(&batch)
-            .context("failed to encode data as JSON")?;
-        writer
-            .finish()
-            .context("failed to finish writing JSON data")?;
-        let json_data_encoded = writer.into_inner();
+    let mut writer = arrow_json::ArrayWriter::new(Vec::new());
+    writer
+        .write(&batch)
+        .context("failed to encode data as JSON")?;
+    writer
+        .finish()
+        .context("failed to finish writing JSON data")?;
+    let json_data_encoded = writer.into_inner();
 
-        // Reparse the string using serde_json
-        let json_data: Vec<serde_json::Map<String, serde_json::Value>> =
-            serde_json::from_reader(json_data_encoded.as_slice())
-                .context("failed to parse JSON data again")?;
-        // remove `inner` field again
-        let json_data_flattened: Vec<_> = json_data
-            .into_iter()
-            .map(|mut m| m.remove("inner"))
-            .collect();
-        output.insert("data".into(), json_data_flattened.into());
-        output.insert("data_type".into(), data_type_json);
-    }
-    Ok(output)
+    // Reparse the string using serde_json
+    let json_data: Vec<serde_json::Map<String, serde_json::Value>> =
+        serde_json::from_reader(json_data_encoded.as_slice())
+            .context("failed to parse JSON data again")?;
+    // remove `inner` field again
+    let json_data_flattened: Vec<_> = json_data
+        .into_iter()
+        .map(|mut m| m.remove("inner"))
+        .collect();
+    output.insert("data".into(), json_data_flattened.into());
+    output.insert("data_type".into(), data_type_json);
+    Ok(())
 }
 
 fn read_input_data(data: InputData) -> eyre::Result<arrow::array::ArrayData> {

apis/rust/node/src/event_stream/mod.rs

@@ -23,7 +23,10 @@ use scheduler::{NON_INPUT_EVENT, Scheduler};
 use self::thread::{EventItem, EventStreamThreadHandle};
 use crate::{
     DaemonCommunicationWrapper, PatternError,
-    daemon_connection::{DaemonChannel, node_integration_testing::convert_output_to_json},
+    daemon_connection::{
+        DaemonChannel,
+        node_integration_testing::{convert_arrow_input_to_json, convert_output_to_json},
+    },
     event_stream::data_conversion::RawData,
     node::{ZENOH_TEARDOWN_TIMEOUT, teardown_with_timeout},
 };
@@ -1057,6 +1060,23 @@ impl EventStream {
                     }
                     _ => None,
                 },
+                // Zenoh-delivered inputs surface to the user as `Event::Input`
+                // exactly like the daemon-path `NodeEvent::Input` above, but
+                // bypass the daemon, so neither this recorder nor the daemon
+                // would otherwise capture them — silently dropping inputs from
+                // the `write_events_to` recording. Record them here too.
+                EventItem::ZenohInput { id, metadata, data } => {
+                    let array = arrow::array::make_array(data.clone());
+                    let mut event_json = convert_arrow_input_to_json(
+                        id,
+                        metadata,
+                        array,
+                        self.start_timestamp,
+                        false,
+                    )?;
+                    event_json.insert("type".into(), "Input".into());
+                    Some(event_json.into())
+                }
                 _ => None,
             };
             if let Some(event_json) = event_json {
@@ -2865,6 +2885,72 @@ mod tests {
         assert!(!user_metadata.parameters.contains_key(FRAMING));
     }
 
+    /// A zenoh-delivered input must be serializable into the same recording
+    /// JSON shape as a daemon-path input, so `write_events_to` recordings do
+    /// not silently drop inputs that take the direct zenoh data plane.
+    #[test]
+    fn zenoh_input_serializes_into_recording_json() {
+        use crate::daemon_connection::node_integration_testing::convert_arrow_input_to_json;
+
+        let hlc = dora_core::uhlc::HLC::default();
+        let start = hlc.new_timestamp();
+        let metadata = Metadata::new(hlc.new_timestamp());
+        let array: arrow::array::ArrayRef =
+            std::sync::Arc::new(arrow::array::Int32Array::from(vec![1, 2, 3]));
+
+        let json = convert_arrow_input_to_json(
+            &DataId::from("in".to_string()),
+            &metadata,
+            array,
+            start,
+            true,
+        )
+        .expect("zenoh input must serialize");
+
+        assert_eq!(json["id"], "in");
+        assert!(json.contains_key("data"), "recorded event must carry data");
+        assert!(
+            json.contains_key("data_type"),
+            "recorded event must carry data_type"
+        );
+        assert_eq!(
+            json["data"].as_array().map(|a| a.len()),
+            Some(3),
+            "all array elements must be recorded"
+        );
+    }
+
+    /// A zenoh input can carry a remote HLC timestamp that predates this node's
+    /// `start_timestamp`. The recording must clamp the offset to zero instead of
+    /// underflowing the NTP64 subtraction (a debug panic that would kill the
+    /// event loop, or a release wraparound to a garbage offset).
+    #[test]
+    fn zenoh_input_with_earlier_timestamp_does_not_underflow() {
+        use crate::daemon_connection::node_integration_testing::convert_arrow_input_to_json;
+
+        let hlc = dora_core::uhlc::HLC::default();
+        // `input_ts` is created first, so it is strictly before `start`.
+        let input_ts = hlc.new_timestamp();
+        let start = hlc.new_timestamp();
+        let metadata = Metadata::new(input_ts);
+        let array: arrow::array::ArrayRef =
+            std::sync::Arc::new(arrow::array::Int32Array::from(vec![1]));
+
+        // `skip_output_time_offsets = false` exercises the time-offset path.
+        let json = convert_arrow_input_to_json(
+            &DataId::from("in".to_string()),
+            &metadata,
+            array,
+            start,
+            false,
+        )
+        .expect("recording an earlier-timestamped input must not fail");
+        assert_eq!(
+            json["time_offset_secs"], 0.0,
+            "an input predating start must clamp to a zero offset"
+        );
+    }
+
     /// The schema-plane FatalError only fires after the grace window: the
     /// producer's periodic full-stream refresh heals an unprimed input in-band,
     /// so a node with a dead `@schema` plane must not be killed on the first