Merge branch 'linear/nap-269-registry-ws-drops-on-any-unknown-daemonoutbound-variant'
ci / rust (push) Successful in 3m58s
ci / rust (push) Successful in 3m58s
This commit is contained in:
@@ -253,7 +253,15 @@ impl RegistryClient {
|
||||
where
|
||||
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 {
|
||||
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(())
|
||||
}
|
||||
|
||||
@@ -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::<DaemonOutbound>(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::<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]
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user