dora_node_api/event_stream/
event.rs

1use dora_arrow_convert::DoraArray;
2use dora_core::config::{DataId, NodeId, OperatorId};
3use dora_message::metadata::Metadata;
4
5/// Represents an incoming Dora event.
6///
7/// Events might be triggered by other nodes, by Dora itself, or by some external user input.
8///
9/// It's safe to ignore event types that are not relevant to the node.
10///
11/// This enum is marked as `non_exhaustive` because we might add additional
12/// variants in the future. Please ignore unknown event types instead of throwing an
13/// error to avoid breakage when updating Dora.
14#[derive(Debug)]
15#[non_exhaustive]
16#[allow(clippy::large_enum_variant)]
17pub enum Event {
18    /// An input was received from another node.
19    ///
20    /// This event corresponds to one of the `inputs` of the node as specified
21    /// in the dataflow YAML file.
22    Input {
23        /// The input ID, as specified in the YAML file.
24        ///
25        /// Note that this is not the output ID of the sender, but the ID
26        /// assigned to the input in the YAML file.
27        id: DataId,
28        /// Meta information about this input, e.g. the timestamp.
29        metadata: Metadata,
30        /// The actual data in the Apache Arrow data format.
31        data: DoraArray,
32    },
33    /// An input was closed by the sender.
34    ///
35    /// The sending node mapped to an input exited, so this input will receive
36    /// no more data.
37    InputClosed {
38        /// The ID of the input that was closed, as specified in the YAML file.
39        ///
40        /// Note that this is not the output ID of the sender, but the ID
41        /// assigned to the input in the YAML file.
42        id: DataId,
43    },
44    /// A previously closed input has recovered and will receive data again.
45    ///
46    /// This happens when an upstream node that timed out (via `input_timeout`)
47    /// starts producing data again. The circuit breaker automatically re-opens
48    /// the input.
49    InputRecovered {
50        /// The ID of the recovered input, as specified in the YAML file.
51        id: DataId,
52    },
53    /// An upstream node has restarted.
54    ///
55    /// Sent to downstream nodes when a node with a restart policy successfully
56    /// restarts after a failure. Nodes can use this to reset state, clear caches,
57    /// or log the recovery.
58    NodeRestarted {
59        /// The ID of the upstream node that restarted.
60        id: NodeId,
61    },
62    /// Notification that the event stream is about to close.
63    ///
64    /// The [`StopCause`] field contains the reason for the event stream closure.
65    ///
66    /// Nodes should exit once the event stream closes.
67    Stop(StopCause),
68    /// Instructs the node to reload itself or one of its operators.
69    ///
70    /// This event is currently only used for reloading Python operators that are
71    /// started by a `dora runtime` process. So this event should not be sent to normal
72    /// nodes yet.
73    Reload {
74        /// The ID of the operator that should be reloaded.
75        ///
76        /// `Some(id)` targets a single operator (the Python-operator reload
77        /// path always sets this). `None` requests reloading the whole runtime
78        /// node, which the runtime currently rejects with a warning rather than
79        /// acting on ("Reloading runtime nodes is not supported").
80        operator_id: Option<OperatorId>,
81    },
82    /// A runtime parameter has been updated via `dora param set`.
83    ///
84    /// Nodes can use this to dynamically adjust behavior (e.g., thresholds,
85    /// rates) without restarting.
86    ParamUpdate {
87        /// The parameter key that was set.
88        key: String,
89        /// The new JSON value.
90        value: serde_json::Value,
91    },
92    /// A runtime parameter has been deleted via `dora param delete`.
93    ///
94    /// Nodes can use this to remove local overrides and fall back to defaults.
95    ParamDeleted {
96        /// The parameter key that was deleted.
97        key: String,
98    },
99    /// An upstream node has failed.
100    ///
101    /// Sent to downstream nodes when an upstream node exits with a
102    /// non-zero exit code. Downstream nodes can use this to handle
103    /// the failure gracefully (e.g. switch to cached data, log, retry).
104    NodeFailed {
105        /// The IDs of the inputs affected by the failure.
106        affected_input_ids: Vec<DataId>,
107        /// Human-readable error message from the failed node.
108        error: String,
109        /// The ID of the node that failed.
110        source_node_id: NodeId,
111    },
112    /// Notifies the node about an unexpected error that happened inside Dora.
113    ///
114    /// It's a good idea to output or log this error for debugging.
115    Error(String),
116}
117
118/// The reason for closing the event stream.
119///
120/// This enum is marked as `non_exhaustive` because we might add additional
121/// variants in the future.
122#[derive(Debug, Clone)]
123#[non_exhaustive]
124pub enum StopCause {
125    /// The dataflow is stopped early after a `dora stop` command (or on `ctrl-c`).
126    ///
127    /// Nodes should exit as soon as possible if they receive a stop event of
128    /// this type. Dora will kill nodes that keep running for too long after
129    /// receiving such a stop event.
130    Manual,
131    /// The event stream is closed because all of the node's inputs were closed.
132    ///
133    /// This stop event type is only sent for nodes that have at least one input.
134    AllInputsClosed,
135}
dora_node_api/event_stream/event.rs

dora_node_api/event_stream/
event.rs
