dora_message/
node_to_daemon.rs

1pub use crate::common::{DataMessage, LogLevel, LogMessage, SharedMemoryId, Timestamped};
2use crate::{
3    DataflowId, current_crate_version,
4    id::{DataId, NodeId},
5    metadata::Metadata,
6    versions_compatible,
7};
8
9#[allow(clippy::large_enum_variant)]
10#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
11#[non_exhaustive]
12pub enum DaemonRequest {
13    Register(NodeRegisterRequest),
14    Subscribe,
15    SendMessage {
16        output_id: DataId,
17        metadata: Metadata,
18        data: Option<DataMessage>,
19    },
20    OutputSent {
21        output_id: DataId,
22        metadata: Metadata,
23    },
24    CloseOutputs(Vec<DataId>),
25    /// Signals that the node is finished sending outputs.
26    OutputsDone,
27    NextEvent,
28    EventStreamDropped,
29    NodeConfig {
30        node_id: NodeId,
31    },
32    /// Store an opaque value in the daemon's dataflow-scoped extension table.
33    ///
34    /// dora attaches no meaning to `namespace`, `key` or `value`: this is the
35    /// out-of-band channel for transports that live outside the tree (see
36    /// `docs/extensions.md`). The daemon tracks who stored a key and who has
37    /// read it, so it can notify readers on removal and reclaim the entry when
38    /// the dataflow ends.
39    ExtensionStore {
40        namespace: String,
41        key: String,
42        value: Vec<u8>,
43    },
44    /// Read an opaque value back. `remove: true` drops it in the same round
45    /// trip, so a consume-once handoff needs one request rather than two.
46    ExtensionLoad {
47        namespace: String,
48        key: String,
49        remove: bool,
50    },
51    /// Drop an opaque value. Every node that stored or loaded the key is sent
52    /// [`NodeEvent::ExtensionDropped`] so it can release whatever the value
53    /// referred to.
54    ExtensionDrop {
55        namespace: String,
56        key: String,
57    },
58    /// An opaque request from a node to the extension registered under
59    /// `namespace` on its daemon. dora never interprets either field.
60    ///
61    /// Complements the extension table (`ExtensionStore`/`ExtensionLoad`/
62    /// `ExtensionDrop`), which brokers descriptor *lifetime*. This one
63    /// carries a call the extension's daemon half must service — the
64    /// tensor-pool uses it for cross-machine registration and pool writes.
65    ExtensionRequest {
66        namespace: String,
67        #[serde(with = "crate::bulk_bytes::vec")]
68        payload: Vec<u8>,
69    },
70}
71
72impl DaemonRequest {
73    /// Bulk bytes this request will contribute to its encoding, for
74    /// [`crate::encode_presized`].
75    ///
76    /// Matched exhaustively on purpose: a new payload-carrying variant that
77    /// forgets to report its size would silently fall back to growing the
78    /// buffer from empty, which is the cost `encode_presized` exists to avoid
79    /// and which no test would catch.
80    pub fn encode_size_hint(&self) -> usize {
81        match self {
82            DaemonRequest::SendMessage { data, .. } => data.as_ref().map_or(0, DataMessage::len),
83            DaemonRequest::Register(_)
84            | DaemonRequest::Subscribe
85            | DaemonRequest::OutputSent { .. }
86            | DaemonRequest::CloseOutputs(_)
87            | DaemonRequest::OutputsDone
88            | DaemonRequest::NextEvent
89            | DaemonRequest::EventStreamDropped
90            | DaemonRequest::NodeConfig { .. } => 0,
91            // The stored value dominates; the namespace and key are short.
92            DaemonRequest::ExtensionStore { value, .. } => value.len(),
93            DaemonRequest::ExtensionLoad { .. } | DaemonRequest::ExtensionDrop { .. } => 0,
94            DaemonRequest::ExtensionRequest { payload, .. } => payload.len(),
95        }
96    }
97
98    pub fn expects_tcp_binary_reply(&self) -> bool {
99        #[allow(clippy::match_like_matches_macro)]
100        match self {
101            DaemonRequest::SendMessage { .. }
102            | DaemonRequest::OutputSent { .. }
103            | DaemonRequest::NodeConfig { .. } => false,
104            DaemonRequest::Register(NodeRegisterRequest { .. })
105            | DaemonRequest::Subscribe
106            | DaemonRequest::CloseOutputs(_)
107            | DaemonRequest::OutputsDone
108            | DaemonRequest::NextEvent
109            | DaemonRequest::EventStreamDropped
110            | DaemonRequest::ExtensionRequest { .. }
111            | DaemonRequest::ExtensionStore { .. }
112            | DaemonRequest::ExtensionLoad { .. }
113            | DaemonRequest::ExtensionDrop { .. } => true,
114        }
115    }
116
117    pub fn expects_tcp_json_reply(&self) -> bool {
118        #[allow(clippy::match_like_matches_macro)]
119        match self {
120            DaemonRequest::NodeConfig { .. } => true,
121            DaemonRequest::Register(NodeRegisterRequest { .. })
122            | DaemonRequest::Subscribe
123            | DaemonRequest::CloseOutputs(_)
124            | DaemonRequest::OutputsDone
125            | DaemonRequest::NextEvent
126            | DaemonRequest::SendMessage { .. }
127            | DaemonRequest::OutputSent { .. }
128            | DaemonRequest::EventStreamDropped
129            | DaemonRequest::ExtensionRequest { .. }
130            | DaemonRequest::ExtensionStore { .. }
131            | DaemonRequest::ExtensionLoad { .. }
132            | DaemonRequest::ExtensionDrop { .. } => false,
133        }
134    }
135}
136
137#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
138pub struct NodeRegisterRequest {
139    pub dataflow_id: DataflowId,
140    pub node_id: NodeId,
141    dora_version: semver::Version,
142    /// The metadata wire-format version this node was built against
143    /// ([`Metadata::CURRENT_VERSION`]).
144    ///
145    /// The `dora_version` semver check above is too coarse to catch a
146    /// message-format break within a release series: two builds that both
147    /// report `1.0.0-rc` are "compatible" by semver even when the payload
148    /// layout differs (as it did across #2366). Carrying the format version
149    /// explicitly lets the daemon reject an incompatible peer at register with
150    /// a clear message, instead of the node desyncing mid-stream into a
151    /// cryptic bad-enum-discriminant deserialization error (#2742).
152    ///
153    /// This covers *layout* drift within one encoding. It cannot cover a change
154    /// of the encoding itself: the register frame is encoded the same way as
155    /// everything else, so a peer from before the bincode→postcard move fails
156    /// while decoding the frame that carries this field, never reaching the
157    /// check. Such changes are gated by the release notes instead.
158    metadata_version: u16,
159}
160
161impl NodeRegisterRequest {
162    pub fn new(dataflow_id: DataflowId, node_id: NodeId) -> Self {
163        Self {
164            dataflow_id,
165            node_id,
166            dora_version: semver::Version::parse(env!("CARGO_PKG_VERSION")).unwrap(),
167            metadata_version: Metadata::CURRENT_VERSION,
168        }
169    }
170
171    pub fn check_version(&self) -> Result<(), String> {
172        let crate_version = current_crate_version();
173        let specified_version = &self.dora_version;
174
175        if !versions_compatible(&crate_version, specified_version)? {
176            return Err(format!(
177                "version mismatch: message format v{} is not compatible \
178                with expected message format v{crate_version}",
179                self.dora_version
180            ));
181        }
182
183        // Even when the semver check passes, the payload wire format can still
184        // differ within a release series (#2366 dropped a `Metadata` field
185        // without changing the version). Reject that here so the failure is a
186        // legible register-time error rather than a mid-stream desync (#2742).
187        if self.metadata_version != Metadata::CURRENT_VERSION {
188            return Err(format!(
189                "message wire-format mismatch: node speaks metadata format v{} \
190                but this daemon speaks v{}. The node and daemon were built from \
191                dora revisions with incompatible message formats; rebuild both \
192                from the same revision.",
193                self.metadata_version,
194                Metadata::CURRENT_VERSION
195            ));
196        }
197
198        Ok(())
199    }
200}
201
202#[derive(Debug, serde::Deserialize, serde::Serialize)]
203pub enum DynamicNodeEvent {
204    NodeConfig { node_id: NodeId },
205}
206
207#[cfg(test)]
208mod register_version_tests {
209    use super::*;
210
211    fn request() -> NodeRegisterRequest {
212        NodeRegisterRequest::new(uuid::Uuid::nil(), NodeId::from("test-node".to_string()))
213    }
214
215    #[test]
216    fn new_stamps_current_metadata_version() {
217        assert_eq!(request().metadata_version, Metadata::CURRENT_VERSION);
218    }
219
220    #[test]
221    fn check_version_accepts_a_matching_request() {
222        // A request built by this same crate version passes both the semver
223        // and the wire-format gate.
224        request().check_version().unwrap();
225    }
226
227    #[test]
228    fn check_version_rejects_a_wire_format_mismatch() {
229        // A peer built from a dora revision with a different metadata wire
230        // format — same semver, incompatible bytes. This is the #2366 / #2742
231        // shape: it must be rejected at register with a legible message rather
232        // than desyncing mid-stream into a cryptic deserialization error.
233        let mut req = request();
234        req.metadata_version = Metadata::CURRENT_VERSION.wrapping_add(1);
235        let err = req
236            .check_version()
237            .expect_err("mismatched metadata wire version must be rejected");
238        assert!(
239            err.contains("wire-format") && err.contains(&Metadata::CURRENT_VERSION.to_string()),
240            "error should name the wire-format mismatch and the expected version, got: {err}"
241        );
242    }
243}
dora_message/node_to_daemon.rs

dora_message/
node_to_daemon.rs
