feat(message)!: mark the wire-protocol enums #[non_exhaustive]

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)
     }

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)
     }

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()

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.
+            _ => {}
         }
     }
 

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.
+                _ => {}
             }
         }
     });

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,

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
+        }
     }
 }
 

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(())
+            }
         }
     }
 

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(())
     }

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;
+                }
             }
         }
 

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,

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,

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<Timestamped<NodeEvent>>),

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,