feat(ws): add fail-closed ordered market stream and shutdown

src/clob/ws/interest.rs

@@ -108,6 +108,11 @@ impl InterestTracker {
         self.interest.fetch_or(interest.bits(), Ordering::Release);
     }
 
+    /// Remove interest in specific message types.
+    pub fn remove(&self, interest: MessageInterest) {
+        self.interest.fetch_and(!interest.bits(), Ordering::Release);
+    }
+
     /// Get the current interest set.
     #[must_use]
     pub fn get(&self) -> MessageInterest {

src/clob/ws/types/response.rs

@@ -1,15 +1,12 @@
-use bon::Builder;
-use serde::Deserialize;
-use serde_json::Value;
-use serde_with::{DefaultOnNull, DisplayFromStr, NoneAsEmptyString, serde_as};
-#[cfg(feature = "tracing")]
-use tracing::warn;
-
 use crate::auth::ApiKey;
 use crate::clob::types::{OrderStatusType, Side, TraderSide};
 use crate::clob::ws::interest::MessageInterest;
 use crate::error::Kind;
 use crate::types::{B256, Decimal, U256};
+use bon::Builder;
+use serde::Deserialize;
+use serde_json::Value;
+use serde_with::{DefaultOnNull, DisplayFromStr, NoneAsEmptyString, serde_as};
 
 /// Top-level WebSocket message wrapper.
 ///
@@ -490,8 +487,8 @@ pub struct MidpointUpdate {
 /// extracted to check interest before final deserialization via `from_value()`.
 /// This avoids re-parsing the JSON text twice.
 ///
-/// For arrays, messages are processed one-by-one with tolerant parsing: unknown or invalid
-/// event types are skipped rather than causing the entire batch to fail.
+/// For arrays, uninterested event types are skipped, while a malformed interested event fails
+/// the whole batch so consumers cannot silently retain stale state.
 pub fn parse_if_interested(
     bytes: &[u8],
     interest: &MessageInterest,
@@ -515,28 +512,27 @@ pub fn parse_if_interested(
                 }
             }
         }
-        Value::Array(arr) => Ok(arr
-            .iter()
-            .filter_map(|elem| {
-                let obj = elem.as_object()?;
-                let event_type = obj.get("event_type").and_then(Value::as_str)?;
+        Value::Array(arr) => {
+            let mut messages = Vec::with_capacity(arr.len());
+            for elem in arr {
+                let Some(obj) = elem.as_object() else {
+                    continue;
+                };
+                let Some(event_type) = obj.get("event_type").and_then(Value::as_str) else {
+                    continue;
+                };
 
                 if !interest.is_interested_in_event(event_type) {
-                    return None;
+                    continue;
                 }
 
-                serde_json::from_value(elem.clone())
-                    .inspect_err(|err| {
-                        #[cfg(feature = "tracing")]
-                        warn!(
-                            event_type = %event_type,
-                            error = %err,
-                            "Skipping unknown/invalid WS event in batch"
-                        );
-                    })
-                    .ok()
-            })
-            .collect()),
+                // Do not silently discard malformed events in an interested batch.
+                // A typed consumer must be able to fail closed rather than retain a
+                // book after missing one of its deltas.
+                messages.push(serde_json::from_value(elem.clone())?);
+            }
+            Ok(messages)
+        }
         _ => Ok(vec![]),
     }
 }

src/rtds/subscription.rs

@@ -16,8 +16,8 @@ use super::types::request::{Subscription, SubscriptionRequest};
 use super::types::response::{RtdsMessage, parse_messages};
 use crate::Result;
 use crate::auth::Credentials;
-use crate::ws::ConnectionManager;
-use crate::ws::connection::ConnectionState;
+use crate::ws::connection::{ConnectionEvent, ConnectionState};
+use crate::ws::{ConnectionManager, WsError};
 
 #[non_exhaustive]
 #[derive(Clone)]
@@ -178,6 +178,7 @@ impl SubscriptionManager {
         subscription: Subscription,
     ) -> Result<impl Stream<Item = Result<RtdsMessage>>> {
         let topic_type = TopicType::new(subscription.topic.clone(), subscription.msg_type.clone());
+        let mut rx = self.connection.subscribe_events();
 
         // Store auth for re-subscription on reconnect.
         // We can recover from poisoned lock because Option<Credentials> has no inconsistent intermediate state.
@@ -230,15 +231,13 @@ impl SubscriptionManager {
             },
         );
 
-        // Create filtered stream with its own receiver
-        let mut rx = self.connection.subscribe();
         let target_topic = topic_type.topic;
         let target_type = topic_type.msg_type;
 
         Ok(try_stream! {
             loop {
                 match rx.recv().await {
-                    Ok(msg) => {
+                    Ok(ConnectionEvent::Message(msg)) => {
                         // Filter messages by topic and type
                         let matches_topic = msg.topic == target_topic;
                         let matches_type = target_type == "*" || msg.msg_type == target_type;
@@ -247,11 +246,11 @@ impl SubscriptionManager {
                             yield msg;
                         }
                     }
+                    Ok(ConnectionEvent::ParseError(error)) => {
+                        Err(WsError::InvalidMessage(error.to_string()))?;
+                    }
                     Err(RecvError::Lagged(n)) => {
-                        #[cfg(not(feature = "tracing"))]
-                        let _ = n;
-                        #[cfg(feature = "tracing")]
-                        tracing::warn!("RTDS subscription lagged, missed {n} messages — continuing");
+                        Err(WsError::Lagged(n))?;
                     }
                     Err(RecvError::Closed) => {
                         break;

src/rtds/types/response.rs

@@ -167,13 +167,22 @@ pub fn parse_messages(bytes: &[u8]) -> crate::Result<Vec<RtdsMessage>> {
         return Ok(Vec::new());
     }
 
-    // Try parsing as array first, fall back to single object
-    if trimmed.first() == Some(&b'[') {
-        Ok(serde_json::from_slice(trimmed)?)
+    let values = if trimmed.first() == Some(&b'[') {
+        serde_json::from_slice::<Vec<serde_json::Value>>(trimmed)?
     } else {
-        let msg: RtdsMessage = serde_json::from_slice(trimmed)?;
-        Ok(vec![msg])
-    }
+        vec![serde_json::from_slice::<serde_json::Value>(trimmed)?]
+    };
+    values
+        .into_iter()
+        .filter_map(|value| {
+            let is_control = value.as_object().is_some_and(|object| {
+                !object.contains_key("topic")
+                    && !object.contains_key("type")
+                    && !object.contains_key("payload")
+            });
+            (!is_control).then(|| serde_json::from_value(value).map_err(Into::into))
+        })
+        .collect()
 }
 
 #[cfg(test)]
@@ -289,6 +298,18 @@ mod tests {
         assert_eq!(msgs[0].topic, "crypto_prices");
     }
 
+    #[test]
+    fn parse_control_ack_without_topic_is_ignored() {
+        let json = br#"{"status":"success","message":"subscribed","connection_id":"abc"}"#;
+        assert!(parse_messages(json).unwrap().is_empty());
+    }
+
+    #[test]
+    fn malformed_update_without_topic_still_fails() {
+        let json = br#"{"type":"update","payload":{"symbol":"btc/usd"}}"#;
+        assert!(parse_messages(json).is_err());
+    }
+
     #[test]
     fn parse_empty_input() {
         let msgs = parse_messages(b"").unwrap();