feat: add error event propagation to downstream nodes

Cargo.toml

@@ -71,7 +71,7 @@ dora-ros2-bridge = { version = "0.3.13", path = "libraries/extensions/ros2-bridg
 dora-ros2-bridge-msg-gen = { version = "0.3.13", path = "libraries/extensions/ros2-bridge/msg-gen" }
 dora-ros2-bridge-python = { path = "libraries/extensions/ros2-bridge/python" }
 # versioned independently from the other dora crates
-dora-message = { version = "0.6.0", path = "libraries/message" }
+dora-message = { version = "0.7.0-alpha", path = "libraries/message" }
 arrow = { version = "54.2.1" }
 arrow-schema = { version = "54.2.1" }
 arrow-data = { version = "54.2.1" }

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

@@ -1,5 +1,5 @@
 use dora_arrow_convert::ArrowData;
-use dora_core::config::{DataId, OperatorId};
+use dora_core::config::{DataId, NodeId, OperatorId};
 use dora_message::metadata::Metadata;
 
 /// Represents an incoming Dora event.
@@ -41,6 +41,22 @@ pub enum Event {
         /// assigned to the input in the YAML file.
         id: DataId,
     },
+    /// A node failed and exited with a non-zero exit code.
+    ///
+    /// The daemon automatically creates this event when a node exits with a non-zero exit code.
+    /// This allows downstream nodes to handle the error gracefully (e.g., use cached data,
+    /// switch to backup source, log and continue).
+    NodeFailed {
+        /// The IDs of the inputs that are affected by the node failure, as specified in the YAML file.
+        ///
+        /// A node failure can affect multiple inputs if the failed node produced multiple outputs
+        /// that are consumed by this node.
+        affected_input_ids: Vec<DataId>,
+        /// The error message describing the failure.
+        error: String,
+        /// The ID of the node that failed.
+        source_node_id: NodeId,
+    },
     /// Notification that the event stream is about to close.
     ///
     /// The [`StopCause`] field contains the reason for the event stream closure.

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

@@ -383,6 +383,24 @@ impl EventStream {
                         });
                         Some(event_json)
                     }
+                    NodeEvent::NodeFailed {
+                        affected_input_ids,
+                        error,
+                        source_node_id,
+                    } => {
+                        let time_offset = self
+                            .clock
+                            .new_timestamp()
+                            .get_diff_duration(&self.start_timestamp);
+                        let event_json = serde_json::json!({
+                            "type": "NodeFailed",
+                            "affected_input_ids": affected_input_ids.iter().map(|id| id.to_string()).collect::<Vec<_>>(),
+                            "error": error,
+                            "source_node_id": source_node_id.to_string(),
+                            "time_offset_secs": time_offset.as_secs_f64(),
+                        });
+                        Some(event_json)
+                    }
                     NodeEvent::AllInputsClosed => {
                         let time_offset = self
                             .clock
@@ -486,6 +504,15 @@ impl EventStream {
                 NodeEvent::Stop => Event::Stop(event::StopCause::Manual),
                 NodeEvent::Reload { operator_id } => Event::Reload { operator_id },
                 NodeEvent::InputClosed { id } => Event::InputClosed { id },
+                NodeEvent::NodeFailed {
+                    affected_input_ids,
+                    error,
+                    source_node_id,
+                } => Event::NodeFailed {
+                    affected_input_ids,
+                    error,
+                    source_node_id,
+                },
                 NodeEvent::Input { id, metadata, data } => {
                     let data = data_to_arrow_array(data, &metadata, ack_channel);
                     match data {

binaries/cli/src/command/topic/echo.rs

@@ -267,6 +267,10 @@ async fn log_to_terminal(
                 eprintln!("Output {node_id}/{output_id} closed");
                 break;
             }
+            InterDaemonEvent::NodeFailed { .. } => {
+                // NodeFailed events are not relevant for topic echo
+                continue;
+            }
         }
     }
 

binaries/cli/src/command/topic/hz.rs

@@ -337,6 +337,10 @@ async fn subscribe_output(
             InterDaemonEvent::OutputClosed { .. } => {
                 break;
             }
+            InterDaemonEvent::NodeFailed { .. } => {
+                // NodeFailed events are not relevant for topic hz
+                continue;
+            }
         }
     }
 

binaries/cli/src/command/topic/info.rs

@@ -200,6 +200,10 @@ fn info(
                         InterDaemonEvent::OutputClosed { .. } => {
                             break;
                         }
+                        InterDaemonEvent::NodeFailed { .. } => {
+                            // NodeFailed events are not relevant for topic info
+                            continue;
+                        }
                     }
                 }
                 Err(_) => break,

binaries/coordinator/src/lib.rs

@@ -1553,9 +1553,7 @@ async fn destroy_daemon(
 
     tcp_send(&mut daemon_connection.stream, &message)
         .await
-        .wrap_err(format!(
-            "failed to send destroy message to daemon `{daemon_id}`"
-        ))?;
+        .wrap_err("failed to send destroy message to daemon")?;
 
     // wait for reply
     let reply_raw = tcp_receive(&mut daemon_connection.stream)