From 8b4cdfd6505a8b0fef95385522616d5392d28754 Mon Sep 17 00:00:00 2001 From: naps62-yolo Date: Sun, 31 May 2026 19:30:36 +0000 Subject: [PATCH] fix(maestro): tolerate unknown WS frame types instead of dropping the connection (NAP-269) `dispatch_text` (registry) and `handle_inbound` (daemon) treated any frame that didn't deserialize into a known `DaemonOutbound`/`DaemonInbound` variant as fatal: the registry returned `false` (breaking the handler loop and deregistering the daemon, wiping its catalog), and the daemon propagated the serde error out of the read loop. A newer peer emitting a not-yet-merged variant would therefore get stuck in a tight disconnect/reconnect loop. - Add a `#[serde(other)] Unknown` catch-all to both enums so unknown `type` tags decode instead of erroring. - Log-and-skip the `Unknown` variant on both sides. - Also log-and-skip genuinely malformed frames rather than tearing down the connection, so one bad frame can't wipe a daemon's catalog. Tested: unknown/malformed frames keep the connection alive (registry handler), and unknown `type` decodes to `Unknown` (protocol round-trip). Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/maestro-daemon/src/registry_client.rs | 15 +++++++- crates/maestro-protocol/src/lib.rs | 29 ++++++++++++-- crates/maestro-registry/src/handler.rs | 40 +++++++++++++++++++- 3 files changed, 77 insertions(+), 7 deletions(-) diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index 96a181f..838ea56 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/crates/maestro-daemon/src/registry_client.rs @@ -253,7 +253,15 @@ impl RegistryClient { where S: SinkExt + Unpin, { - let parsed: DaemonInbound = serde_json::from_str(text)?; + let parsed: DaemonInbound = match serde_json::from_str(text) { + Ok(p) => p, + Err(e) => { + // A single malformed frame must not drop the registry link — + // log and skip it (NAP-269). + tracing::warn!(error = %e, "malformed registry frame — skipping"); + return Ok(()); + } + }; match parsed { DaemonInbound::Ack => {} DaemonInbound::Error(e) => { @@ -295,6 +303,11 @@ impl RegistryClient { // NAP-246: peer-filtered catalog delta from the registry. self.catalog.apply(event).await; } + DaemonInbound::Unknown => { + // A newer registry emitted a frame type this build doesn't know. + // Skip it rather than dropping the link (NAP-269). + tracing::debug!("unknown registry frame — skipping (forward-compat, NAP-269)"); + } } Ok(()) } diff --git a/crates/maestro-protocol/src/lib.rs b/crates/maestro-protocol/src/lib.rs index 5e80032..870fb54 100644 --- a/crates/maestro-protocol/src/lib.rs +++ b/crates/maestro-protocol/src/lib.rs @@ -144,6 +144,11 @@ pub enum DaemonOutbound { reply_to: ReplyTo, accepted: bool, }, + /// Forward-compat catch-all: a frame whose `type` this build doesn't + /// recognize (e.g. a newer daemon emitting a not-yet-merged variant). + /// Readers log and skip it instead of tearing the connection down (NAP-269). + #[serde(other)] + Unknown, } /// Messages the registry sends back to a daemon over the same WS. @@ -182,6 +187,11 @@ pub enum DaemonInbound { CatalogEvent { event: RegistryEvent, }, + /// Forward-compat catch-all: a frame whose `type` this build doesn't + /// recognize (e.g. a newer registry emitting a not-yet-merged variant). + /// Readers log and skip it instead of tearing the connection down (NAP-269). + #[serde(other)] + Unknown, } // --------------------------------------------------------------------------- @@ -589,10 +599,21 @@ mod tests { } #[test] - fn unknown_message_type_is_rejected() { - let bad = serde_json::json!({ "type": "Bogus" }); - let err = serde_json::from_value::(bad).unwrap_err(); - assert!(err.to_string().contains("Bogus"), "{}", err); + fn unknown_message_type_deserializes_to_unknown() { + // A newer peer can introduce a frame `type` this build has never heard + // of. It must decode to the forward-compat `Unknown` variant rather + // than erroring, so the reader skips it instead of dropping the WS + // (NAP-269). + let bad = serde_json::json!({ "type": "Bogus", "extra": 1 }); + assert_eq!( + serde_json::from_value::(bad).unwrap(), + DaemonOutbound::Unknown + ); + let bad_inbound = serde_json::json!({ "type": "AlsoBogus" }); + assert_eq!( + serde_json::from_value::(bad_inbound).unwrap(), + DaemonInbound::Unknown + ); } #[test] diff --git a/crates/maestro-registry/src/handler.rs b/crates/maestro-registry/src/handler.rs index 6277cf8..5c8fe00 100644 --- a/crates/maestro-registry/src/handler.rs +++ b/crates/maestro-registry/src/handler.rs @@ -167,8 +167,10 @@ async fn dispatch_text( let parsed: DaemonOutbound = match serde_json::from_str(text) { Ok(p) => p, Err(e) => { - tracing::warn!(%daemon_id, error = %e, "malformed message"); - return false; + // A single malformed frame must not tear down the WS (and wipe this + // daemon's catalog) — log and skip it (NAP-269). + tracing::warn!(%daemon_id, error = %e, "malformed message — skipping"); + return true; } }; match parsed { @@ -217,6 +219,11 @@ async fn dispatch_text( // don't crash the connection; the registry doesn't act on it yet. tracing::debug!(%daemon_id, ?reply_to, "RelayAck received but T3 not implemented (NAP-238 pending)"); } + DaemonOutbound::Unknown => { + // A newer daemon emitted a frame type this build doesn't know. Skip + // it rather than dropping the connection (NAP-269). + tracing::debug!(%daemon_id, "unknown frame type — skipping (forward-compat, NAP-269)"); + } } true } @@ -233,3 +240,32 @@ async fn send_error( let text = serde_json::to_string(&err).expect("serialize error"); sink.send(Message::Text(text)).await } + +#[cfg(test)] +mod tests { + use super::*; + + // `dispatch_text` returning `false` is what breaks the handler loop and + // triggers a deregister (catalog wipe). A frame the registry can't act on + // must NOT do that, or a newer daemon emitting an unknown variant gets + // stuck in a tight reconnect loop (NAP-269). + + #[tokio::test] + async fn unknown_frame_keeps_connection_alive() { + let registry = Registry::new(); + let frame = serde_json::json!({ "type": "FutureVariant", "x": 1 }).to_string(); + assert!( + dispatch_text(®istry, "laptop", "tok", &frame).await, + "unknown frame type must be skipped, not drop the WS" + ); + } + + #[tokio::test] + async fn malformed_frame_keeps_connection_alive() { + let registry = Registry::new(); + assert!( + dispatch_text(®istry, "laptop", "tok", "{ not valid json").await, + "malformed frame must be skipped, not drop the WS" + ); + } +}