fix(maestro): tolerate unknown WS frame types instead of dropping the connection (NAP-269)
ci / rust (push) Successful in 14m9s

`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) <noreply@anthropic.com>
This commit is contained in:
2026-05-31 19:30:36 +00:00
co-authored by Claude Opus 4.8
parent 3b60f67332
commit 8b4cdfd650
3 changed files with 77 additions and 7 deletions
+14 -1
View File
@@ -253,7 +253,15 @@ impl RegistryClient {
where where
S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin, S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + 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 { match parsed {
DaemonInbound::Ack => {} DaemonInbound::Ack => {}
DaemonInbound::Error(e) => { DaemonInbound::Error(e) => {
@@ -295,6 +303,11 @@ impl RegistryClient {
// NAP-246: peer-filtered catalog delta from the registry. // NAP-246: peer-filtered catalog delta from the registry.
self.catalog.apply(event).await; 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(()) Ok(())
} }
+25 -4
View File
@@ -144,6 +144,11 @@ pub enum DaemonOutbound {
reply_to: ReplyTo, reply_to: ReplyTo,
accepted: bool, 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. /// Messages the registry sends back to a daemon over the same WS.
@@ -182,6 +187,11 @@ pub enum DaemonInbound {
CatalogEvent { CatalogEvent {
event: RegistryEvent, 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] #[test]
fn unknown_message_type_is_rejected() { fn unknown_message_type_deserializes_to_unknown() {
let bad = serde_json::json!({ "type": "Bogus" }); // A newer peer can introduce a frame `type` this build has never heard
let err = serde_json::from_value::<DaemonOutbound>(bad).unwrap_err(); // of. It must decode to the forward-compat `Unknown` variant rather
assert!(err.to_string().contains("Bogus"), "{}", err); // 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::<DaemonOutbound>(bad).unwrap(),
DaemonOutbound::Unknown
);
let bad_inbound = serde_json::json!({ "type": "AlsoBogus" });
assert_eq!(
serde_json::from_value::<DaemonInbound>(bad_inbound).unwrap(),
DaemonInbound::Unknown
);
} }
#[test] #[test]
+38 -2
View File
@@ -167,8 +167,10 @@ async fn dispatch_text(
let parsed: DaemonOutbound = match serde_json::from_str(text) { let parsed: DaemonOutbound = match serde_json::from_str(text) {
Ok(p) => p, Ok(p) => p,
Err(e) => { Err(e) => {
tracing::warn!(%daemon_id, error = %e, "malformed message"); // A single malformed frame must not tear down the WS (and wipe this
return false; // daemon's catalog) — log and skip it (NAP-269).
tracing::warn!(%daemon_id, error = %e, "malformed message — skipping");
return true;
} }
}; };
match parsed { match parsed {
@@ -217,6 +219,11 @@ async fn dispatch_text(
// don't crash the connection; the registry doesn't act on it yet. // 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)"); 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 true
} }
@@ -233,3 +240,32 @@ async fn send_error(
let text = serde_json::to_string(&err).expect("serialize error"); let text = serde_json::to_string(&err).expect("serialize error");
sink.send(Message::Text(text)).await 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(&registry, "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(&registry, "laptop", "tok", "{ not valid json").await,
"malformed frame must be skipped, not drop the WS"
);
}
}