diff --git a/apis/rust/node/src/daemon_connection/interactive.rs b/apis/rust/node/src/daemon_connection/interactive.rs index de859a03df..3810f0d57d 100644 --- a/apis/rust/node/src/daemon_connection/interactive.rs +++ b/apis/rust/node/src/daemon_connection/interactive.rs @@ -85,6 +85,9 @@ impl InteractiveEvents { DaemonRequest::NodeConfig { .. } => { eyre::bail!("unexpected NodeConfig in interactive mode") } + // `DaemonRequest` is `#[non_exhaustive]`: reject an unknown + // request explicitly rather than silently answering `Ok`. + other => eyre::bail!("unsupported daemon request: {other:?}"), }; Ok(reply) } diff --git a/apis/rust/node/src/daemon_connection/node_integration_testing.rs b/apis/rust/node/src/daemon_connection/node_integration_testing.rs index ae29033180..1d95b0f358 100644 --- a/apis/rust/node/src/daemon_connection/node_integration_testing.rs +++ b/apis/rust/node/src/daemon_connection/node_integration_testing.rs @@ -138,6 +138,9 @@ impl IntegrationTestingEvents { DaemonRequest::NodeConfig { .. } => { eyre::bail!("unexpected NodeConfig in interactive mode") } + // `DaemonRequest` is `#[non_exhaustive]`: reject an unknown + // request explicitly rather than silently answering `Ok`. + other => eyre::bail!("unsupported daemon request: {other:?}"), }; Ok(reply) } diff --git a/binaries/cli/src/command/record.rs b/binaries/cli/src/command/record.rs index d714f6529c..a932178e9c 100644 --- a/binaries/cli/src/command/record.rs +++ b/binaries/cli/src/command/record.rs @@ -468,6 +468,8 @@ fn run_record_proxy(args: Record) -> eyre::Result<()> { node_id, output_id, .. } => (node_id.to_string(), output_id.to_string()), InterDaemonEvent::OutputClosed { .. } => continue, + // `InterDaemonEvent` is `#[non_exhaustive]`: skip events this build predates. + _ => continue, }; let now_nanos = SystemTime::now() diff --git a/binaries/cli/src/command/topic/echo.rs b/binaries/cli/src/command/topic/echo.rs index 118c9ccef6..69f5f500e8 100644 --- a/binaries/cli/src/command/topic/echo.rs +++ b/binaries/cli/src/command/topic/echo.rs @@ -243,6 +243,8 @@ fn inspect( } => { eprintln!("Output {node_id}/{output_id} closed"); } + // `InterDaemonEvent` is `#[non_exhaustive]`: skip events this build predates. + _ => {} } } diff --git a/binaries/cli/src/command/topic/hz.rs b/binaries/cli/src/command/topic/hz.rs index 91aaf84693..f96d69525b 100644 --- a/binaries/cli/src/command/topic/hz.rs +++ b/binaries/cli/src/command/topic/hz.rs @@ -337,6 +337,8 @@ fn run_hz( } } InterDaemonEvent::OutputClosed { .. } => {} + // `InterDaemonEvent` is `#[non_exhaustive]`: skip events this build predates. + _ => {} } } }); diff --git a/binaries/cli/src/command/topic/info.rs b/binaries/cli/src/command/topic/info.rs index 4e59fbb633..3cf4532f71 100644 --- a/binaries/cli/src/command/topic/info.rs +++ b/binaries/cli/src/command/topic/info.rs @@ -195,6 +195,8 @@ fn info( stats_clone.record(data_size, data_type, Instant::now()); } InterDaemonEvent::OutputClosed { .. } => break, + // `InterDaemonEvent` is `#[non_exhaustive]`: skip events this build predates. + _ => continue, } } Ok(Err(_)) => continue, diff --git a/binaries/coordinator/src/ws_daemon.rs b/binaries/coordinator/src/ws_daemon.rs index 21d19caee3..a0a5fb4f6d 100644 --- a/binaries/coordinator/src/ws_daemon.rs +++ b/binaries/coordinator/src/ws_daemon.rs @@ -209,6 +209,15 @@ async fn handle_daemon_request( true } } + // `CoordinatorRequest` is `#[non_exhaustive]`: a newer daemon may send a + // variant this coordinator predates. Keep the connection open and drop + // the request rather than tearing down a daemon over an unknown message. + _ => { + tracing::warn!( + "ignoring unrecognized request from daemon (daemon is likely newer than this coordinator)" + ); + true + } } } @@ -297,6 +306,13 @@ fn translate_daemon_event( node_id, clean_stop, }), + // `DaemonEvent` is `#[non_exhaustive]`: a newer daemon may report an + // event this coordinator has no translation for. `None` is already the + // "nothing to forward" signal, so an unknown event is dropped quietly. + _ => { + tracing::debug!("ignoring unrecognized daemon event from `{daemon_id}`"); + None + } } } diff --git a/binaries/daemon/src/lib.rs b/binaries/daemon/src/lib.rs index c68a8ea381..1e4e7d8c48 100644 --- a/binaries/daemon/src/lib.rs +++ b/binaries/daemon/src/lib.rs @@ -3805,6 +3805,13 @@ impl Daemon { } Ok(()) } + // `InterDaemonEvent` is `#[non_exhaustive]`: a peer daemon running a + // newer dora may send an event this one predates. Warn and continue — + // tearing down the dataflow over an unknown peer message would be worse. + other => { + tracing::warn!("ignoring unrecognized inter-daemon event: {other:?}"); + Ok(()) + } } } diff --git a/binaries/daemon/src/node_communication/mod.rs b/binaries/daemon/src/node_communication/mod.rs index f323054582..1de613d932 100644 --- a/binaries/daemon/src/node_communication/mod.rs +++ b/binaries/daemon/src/node_communication/mod.rs @@ -404,6 +404,16 @@ impl Listener { ) .await?; } + // `DaemonRequest` is `#[non_exhaustive]`: a node built against a newer + // dora-node-api may send a request this daemon predates. Answer with an + // explicit error so the node fails loudly instead of hanging on a reply + // that never comes. + other => { + let reply = DaemonReply::Result(Err(format!( + "unsupported request from node (node is likely newer than this daemon): {other:?}" + ))); + self.send_reply(reply, connection).await?; + } } Ok(()) } diff --git a/binaries/replay-node/src/main.rs b/binaries/replay-node/src/main.rs index a62404e665..84f5c43a5b 100644 --- a/binaries/replay-node/src/main.rs +++ b/binaries/replay-node/src/main.rs @@ -137,6 +137,12 @@ fn main() -> eyre::Result<()> { InterDaemonEvent::OutputClosed { .. } => { // Skip close events during replay } + // `InterDaemonEvent` is `#[non_exhaustive]`: a recording made by a + // newer dora may carry events this replay node predates. Skip them + // rather than aborting an otherwise-replayable recording. + _ => { + skipped += 1; + } } } diff --git a/libraries/message/src/daemon_to_coordinator.rs b/libraries/message/src/daemon_to_coordinator.rs index 709c1fd8ea..74dc771393 100644 --- a/libraries/message/src/daemon_to_coordinator.rs +++ b/libraries/message/src/daemon_to_coordinator.rs @@ -16,6 +16,7 @@ pub struct DataflowStatusEntry { #[allow(clippy::large_enum_variant)] #[derive(Debug, serde::Serialize, serde::Deserialize)] +#[non_exhaustive] pub enum CoordinatorRequest { Register(DaemonRegisterRequest), Event { @@ -157,6 +158,7 @@ mod register_version_tests { } #[derive(Debug, serde::Serialize, serde::Deserialize)] +#[non_exhaustive] pub enum DaemonEvent { BuildResult { build_id: BuildId, diff --git a/libraries/message/src/daemon_to_daemon.rs b/libraries/message/src/daemon_to_daemon.rs index cb4b70e640..849fb93cf4 100644 --- a/libraries/message/src/daemon_to_daemon.rs +++ b/libraries/message/src/daemon_to_daemon.rs @@ -8,6 +8,7 @@ use crate::{ #[derive(Debug, serde::Deserialize, serde::Serialize)] #[allow(clippy::large_enum_variant)] +#[non_exhaustive] pub enum InterDaemonEvent { Output { dataflow_id: DataflowId, diff --git a/libraries/message/src/daemon_to_node.rs b/libraries/message/src/daemon_to_node.rs index e3641dcb8f..65f8203033 100644 --- a/libraries/message/src/daemon_to_node.rs +++ b/libraries/message/src/daemon_to_node.rs @@ -88,6 +88,7 @@ pub enum DaemonCommunication { #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[must_use] #[allow(clippy::large_enum_variant)] +#[non_exhaustive] pub enum DaemonReply { Result(Result<(), String>), NextEvents(Vec>), diff --git a/libraries/message/src/node_to_daemon.rs b/libraries/message/src/node_to_daemon.rs index 96004e15a2..9b1f72944d 100644 --- a/libraries/message/src/node_to_daemon.rs +++ b/libraries/message/src/node_to_daemon.rs @@ -8,6 +8,7 @@ use crate::{ #[allow(clippy::large_enum_variant)] #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +#[non_exhaustive] pub enum DaemonRequest { Register(NodeRegisterRequest), Subscribe,