nautilus_dydx/websocket/
handler.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Message handler for dYdX WebSocket streams.
17//!
18//! The handler owns the WebSocketClient exclusively and runs in a dedicated
19//! Tokio task within the lock-free I/O boundary. It deserializes raw messages
20//! into venue-specific types without converting to Nautilus domain objects.
21
22use std::{
23    collections::VecDeque,
24    fmt::Debug,
25    sync::{
26        Arc,
27        atomic::{AtomicBool, Ordering},
28    },
29};
30
31use ahash::AHashMap;
32use nautilus_network::{
33    RECONNECTED,
34    retry::{RetryManager, create_websocket_retry_manager},
35    websocket::{SubscriptionState, WebSocketClient},
36};
37use tokio_tungstenite::tungstenite::Message;
38use ustr::Ustr;
39
40use super::{
41    DydxWsError, DydxWsResult,
42    client::DYDX_RATE_LIMIT_KEY_SUBSCRIPTION,
43    enums::{DydxWsChannel, DydxWsMessage, DydxWsOutputMessage},
44    error::DydxWebSocketError,
45    messages::{
46        DydxCandle, DydxMarketsContents, DydxOrderbookContents, DydxOrderbookSnapshotContents,
47        DydxSubscription, DydxTradeContents, DydxWsBlockHeightMessage, DydxWsCandlesMessage,
48        DydxWsChannelBatchDataMsg, DydxWsChannelDataMsg, DydxWsConnectedMsg, DydxWsFeedMessage,
49        DydxWsGenericMsg, DydxWsMarketsMessage, DydxWsOrderbookMessage,
50        DydxWsParentSubaccountsMessage, DydxWsSubaccountsChannelContents,
51        DydxWsSubaccountsChannelData, DydxWsSubaccountsMessage, DydxWsSubaccountsSubscribed,
52        DydxWsSubscriptionMsg, DydxWsTradesMessage,
53    },
54};
55
56/// Commands sent to the feed handler.
57#[derive(Debug, Clone)]
58pub enum HandlerCommand {
59    /// Registers a subscription message for replay.
60    RegisterSubscription {
61        topic: String,
62        subscription: DydxSubscription,
63    },
64    /// Unregisters a subscription message.
65    UnregisterSubscription { topic: String },
66    /// Sends a text message via WebSocket.
67    SendText(String),
68    /// Disconnects the WebSocket client.
69    Disconnect,
70}
71
72/// Deserializes incoming WebSocket messages into venue-specific types.
73///
74/// The handler owns the WebSocketClient exclusively within the lock-free I/O boundary,
75/// eliminating RwLock contention on the hot path.
76pub struct FeedHandler {
77    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
78    out_tx: tokio::sync::mpsc::UnboundedSender<DydxWsOutputMessage>,
79    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
80    client: WebSocketClient,
81    signal: Arc<AtomicBool>,
82    retry_manager: RetryManager<DydxWsError>,
83    subscriptions: SubscriptionState,
84    subscription_messages: AHashMap<String, DydxSubscription>,
85    message_buffer: VecDeque<DydxWsOutputMessage>,
86    book_sequence: AHashMap<String, u64>,
87}
88
89impl Debug for FeedHandler {
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        f.debug_struct(stringify!(FeedHandler))
92            .field("subscriptions", &self.subscriptions.len())
93            .finish_non_exhaustive()
94    }
95}
96
97impl FeedHandler {
98    /// Creates a new [`FeedHandler`].
99    #[must_use]
100    pub fn new(
101        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
102        out_tx: tokio::sync::mpsc::UnboundedSender<DydxWsOutputMessage>,
103        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
104        client: WebSocketClient,
105        signal: Arc<AtomicBool>,
106        subscriptions: SubscriptionState,
107    ) -> Self {
108        Self {
109            cmd_rx,
110            out_tx,
111            raw_rx,
112            client,
113            signal,
114            retry_manager: create_websocket_retry_manager(),
115            subscriptions,
116            subscription_messages: AHashMap::new(),
117            message_buffer: VecDeque::new(),
118            book_sequence: AHashMap::new(),
119        }
120    }
121
122    async fn send_with_retry(
123        &self,
124        payload: String,
125        rate_limit_keys: Option<&[Ustr]>,
126    ) -> Result<(), DydxWsError> {
127        let keys_owned: Option<Vec<Ustr>> = rate_limit_keys.map(|k| k.to_vec());
128        self.retry_manager
129            .execute_with_retry(
130                "websocket_send",
131                || {
132                    let payload = payload.clone();
133                    let keys = keys_owned.clone();
134                    async move {
135                        self.client
136                            .send_text(payload, keys.as_deref())
137                            .await
138                            .map_err(|e| DydxWsError::ClientError(format!("Send failed: {e}")))
139                    }
140                },
141                should_retry_dydx_error,
142                create_dydx_timeout_error,
143            )
144            .await
145    }
146
147    /// Main processing loop for the handler.
148    ///
149    /// # Panics
150    ///
151    /// This method will not panic. The `expect` call on `iter.next()` is safe
152    /// because we explicitly check that `msgs` is not empty before calling it.
153    pub async fn run(&mut self) {
154        log::debug!("WebSocket handler started");
155
156        loop {
157            // First drain any buffered messages
158            if !self.message_buffer.is_empty() {
159                let msg = self.message_buffer.pop_front().unwrap();
160                if self.out_tx.send(msg).is_err() {
161                    log::debug!("Receiver dropped, stopping handler");
162                    break;
163                }
164                continue;
165            }
166
167            tokio::select! {
168                Some(cmd) = self.cmd_rx.recv() => {
169                    if self.handle_command(cmd).await {
170                        break;
171                    }
172                }
173
174                Some(msg) = self.raw_rx.recv() => {
175                    log::trace!("Handler received raw message");
176                    let msgs = self.process_raw_message(msg).await;
177                    if !msgs.is_empty() {
178                        let mut iter = msgs.into_iter();
179                        // We just checked that msgs is not empty
180                        let first = iter.next().expect("non-empty vec has first element");
181                        self.message_buffer.extend(iter);
182                        log::trace!("Handler sending message: {:?}", std::mem::discriminant(&first));
183                        if self.out_tx.send(first).is_err() {
184                            log::debug!("Receiver dropped, stopping handler");
185                            break;
186                        }
187                    }
188                }
189
190                else => {
191                    log::debug!("Handler shutting down: channels closed");
192                    break;
193                }
194            }
195
196            if self.signal.load(Ordering::Acquire) {
197                log::debug!("Handler received stop signal");
198                break;
199            }
200        }
201    }
202
203    async fn process_raw_message(&mut self, msg: Message) -> Vec<DydxWsOutputMessage> {
204        match msg {
205            Message::Text(txt) => {
206                if txt == RECONNECTED {
207                    self.clear_state();
208
209                    if let Err(e) = self.replay_subscriptions().await {
210                        log::error!("Failed to replay subscriptions after reconnect: {e}");
211                    }
212                    let topics = self.subscriptions.all_topics();
213                    return vec![DydxWsOutputMessage::Reconnected { topics }];
214                }
215
216                // Hot path: zero-copy parse for feed messages (orderbook/trades/candles)
217                match serde_json::from_str::<DydxWsFeedMessage>(&txt) {
218                    Ok(feed_msg) => {
219                        return self.handle_feed_message(feed_msg);
220                    }
221                    Err(e) => {
222                        if txt.contains("v4_subaccounts") {
223                            log::warn!(
224                                "[WS_DESER] Failed to parse v4_subaccounts as DydxWsFeedMessage: {e}\nRaw: {txt}"
225                            );
226                        }
227                    }
228                }
229
230                // Cold path: infrequent control messages (connected/subscribed/error)
231                match serde_json::from_str::<serde_json::Value>(&txt) {
232                    Ok(val) => match serde_json::from_value::<DydxWsGenericMsg>(val.clone()) {
233                        Ok(meta) => {
234                            let result = if meta.is_connected() {
235                                serde_json::from_value::<DydxWsConnectedMsg>(val)
236                                    .map(DydxWsMessage::Connected)
237                            } else if meta.is_subscribed() {
238                                log::debug!("Processing subscribed message via fallback path");
239
240                                if let Ok(sub_msg) =
241                                    serde_json::from_value::<DydxWsSubscriptionMsg>(val.clone())
242                                {
243                                    if sub_msg.channel == DydxWsChannel::Subaccounts {
244                                        log::debug!("Parsing subaccounts subscription (fallback)");
245                                        serde_json::from_value::<DydxWsSubaccountsSubscribed>(val)
246                                            .map(DydxWsMessage::SubaccountsSubscribed)
247                                            .or_else(|e| {
248                                                log::warn!(
249                                                    "Failed to parse subaccounts subscription: {e}"
250                                                );
251                                                Ok(DydxWsMessage::Subscribed(sub_msg))
252                                            })
253                                    } else {
254                                        Ok(DydxWsMessage::Subscribed(sub_msg))
255                                    }
256                                } else {
257                                    serde_json::from_value::<DydxWsSubscriptionMsg>(val)
258                                        .map(DydxWsMessage::Subscribed)
259                                }
260                            } else if meta.is_unsubscribed() {
261                                serde_json::from_value::<DydxWsSubscriptionMsg>(val)
262                                    .map(DydxWsMessage::Unsubscribed)
263                            } else if meta.is_error() {
264                                serde_json::from_value::<DydxWebSocketError>(val)
265                                    .map(DydxWsMessage::Error)
266                            } else if meta.is_unknown() {
267                                log::warn!("Received unknown WebSocket message type: {txt}",);
268                                Ok(DydxWsMessage::Raw(val))
269                            } else {
270                                Ok(DydxWsMessage::Raw(val))
271                            };
272
273                            match result {
274                                Ok(dydx_msg) => self.handle_dydx_message(dydx_msg).await,
275                                Err(e) => {
276                                    log::error!(
277                                        "Failed to parse WebSocket message: {e}. Message type: {:?}, Channel: {:?}. Raw: {txt}",
278                                        meta.msg_type,
279                                        meta.channel,
280                                    );
281                                    vec![]
282                                }
283                            }
284                        }
285                        Err(e) => {
286                            log::error!(
287                                "Failed to parse WebSocket message envelope (DydxWsGenericMsg): {e}\nRaw JSON:\n{txt}"
288                            );
289                            vec![]
290                        }
291                    },
292                    Err(e) => {
293                        let err = DydxWebSocketError::from_message(e.to_string());
294                        vec![DydxWsOutputMessage::Error(err)]
295                    }
296                }
297            }
298            Message::Pong(_data) => vec![],
299            Message::Ping(_data) => vec![],
300            Message::Binary(_bin) => vec![],
301            Message::Close(_frame) => {
302                log::debug!("WebSocket close frame received");
303                vec![]
304            }
305            Message::Frame(_) => vec![],
306        }
307    }
308
309    async fn handle_dydx_message(&mut self, msg: DydxWsMessage) -> Vec<DydxWsOutputMessage> {
310        match self.handle_message(msg).await {
311            Ok(msgs) => msgs,
312            Err(e) => {
313                log::error!("Error handling message: {e}");
314                vec![]
315            }
316        }
317    }
318
319    fn handle_feed_message(&mut self, feed_msg: DydxWsFeedMessage) -> Vec<DydxWsOutputMessage> {
320        log::trace!(
321            "Handling feed message: {:?}",
322            std::mem::discriminant(&feed_msg)
323        );
324
325        match feed_msg {
326            DydxWsFeedMessage::Subaccounts(msg) => self.handle_subaccounts(msg),
327            DydxWsFeedMessage::Orderbook(msg) => self.handle_orderbook(msg),
328            DydxWsFeedMessage::Trades(msg) => self.handle_trades(msg),
329            DydxWsFeedMessage::Markets(msg) => self.handle_markets_feed(msg),
330            DydxWsFeedMessage::Candles(msg) => self.handle_candles_feed(msg),
331            DydxWsFeedMessage::ParentSubaccounts(msg) => self.handle_parent_subaccounts(msg),
332            DydxWsFeedMessage::BlockHeight(msg) => self.handle_block_height_feed(msg),
333        }
334    }
335
336    fn handle_subaccounts(&self, msg: DydxWsSubaccountsMessage) -> Vec<DydxWsOutputMessage> {
337        match msg {
338            DydxWsSubaccountsMessage::Subscribed(data) => {
339                let topic =
340                    self.topic_from_msg(&DydxWsChannel::Subaccounts, &Some(data.id.clone()));
341                self.subscriptions.confirm_subscribe(&topic);
342                log::debug!("Forwarding subaccount subscription to execution client");
343                vec![DydxWsOutputMessage::SubaccountSubscribed(Box::new(data))]
344            }
345            DydxWsSubaccountsMessage::ChannelData(data) => {
346                let has_orders = data.contents.orders.as_ref().is_some_and(|o| !o.is_empty());
347                let has_fills = data.contents.fills.as_ref().is_some_and(|f| !f.is_empty());
348
349                if has_orders || has_fills {
350                    log::debug!(
351                        "Received {} order(s), {} fill(s) - forwarding to execution client",
352                        data.contents.orders.as_ref().map_or(0, |o| o.len()),
353                        data.contents.fills.as_ref().map_or(0, |f| f.len())
354                    );
355                    vec![DydxWsOutputMessage::SubaccountsChannelData(Box::new(data))]
356                } else {
357                    vec![]
358                }
359            }
360            DydxWsSubaccountsMessage::Unsubscribed(data) => {
361                let topic = self.topic_from_msg(&DydxWsChannel::Subaccounts, &data.id);
362                self.subscriptions.confirm_unsubscribe(&topic);
363                vec![]
364            }
365        }
366    }
367
368    fn handle_orderbook(&mut self, msg: DydxWsOrderbookMessage) -> Vec<DydxWsOutputMessage> {
369        match msg {
370            DydxWsOrderbookMessage::Subscribed(data) => {
371                let topic = self.topic_from_msg(&DydxWsChannel::Orderbook, &data.id);
372                self.subscriptions.confirm_subscribe(&topic);
373
374                if let Some(id) = &data.id {
375                    self.book_sequence.insert(id.clone(), data.message_id);
376                }
377
378                self.deserialize_orderbook_snapshot(&data)
379            }
380            DydxWsOrderbookMessage::ChannelData(data) => {
381                if let Some(id) = &data.id {
382                    if let Some(last_id) = self.book_sequence.get(id)
383                        && data.message_id <= *last_id
384                    {
385                        log::warn!(
386                            "Orderbook sequence regression for {id}: last {last_id}, received {}",
387                            data.message_id
388                        );
389                    }
390                    self.book_sequence.insert(id.clone(), data.message_id);
391                }
392                self.deserialize_orderbook_update(&data)
393            }
394            DydxWsOrderbookMessage::ChannelBatchData(data) => {
395                if let Some(id) = &data.id {
396                    if let Some(last_id) = self.book_sequence.get(id)
397                        && data.message_id <= *last_id
398                    {
399                        log::warn!(
400                            "Orderbook batch sequence regression for {id}: last {last_id}, received {}",
401                            data.message_id
402                        );
403                    }
404                    self.book_sequence.insert(id.clone(), data.message_id);
405                }
406                self.deserialize_orderbook_batch(&data)
407            }
408            DydxWsOrderbookMessage::Unsubscribed(data) => {
409                let topic = self.topic_from_msg(&DydxWsChannel::Orderbook, &data.id);
410                self.subscriptions.confirm_unsubscribe(&topic);
411
412                if let Some(id) = &data.id {
413                    self.book_sequence.remove(id);
414                }
415                vec![]
416            }
417        }
418    }
419
420    fn handle_trades(&self, msg: DydxWsTradesMessage) -> Vec<DydxWsOutputMessage> {
421        match msg {
422            DydxWsTradesMessage::Subscribed(data) => {
423                let topic = self.topic_from_msg(&DydxWsChannel::Trades, &data.id);
424                self.subscriptions.confirm_subscribe(&topic);
425                self.deserialize_trades(&data)
426            }
427            DydxWsTradesMessage::ChannelData(data) => self.deserialize_trades(&data),
428            DydxWsTradesMessage::Unsubscribed(data) => {
429                let topic = self.topic_from_msg(&DydxWsChannel::Trades, &data.id);
430                self.subscriptions.confirm_unsubscribe(&topic);
431                vec![]
432            }
433        }
434    }
435
436    fn handle_markets_feed(&self, msg: DydxWsMarketsMessage) -> Vec<DydxWsOutputMessage> {
437        match msg {
438            DydxWsMarketsMessage::Subscribed(data) => {
439                let topic = self.topic_from_msg(&DydxWsChannel::Markets, &data.id);
440                self.subscriptions.confirm_subscribe(&topic);
441                self.deserialize_markets(&data)
442            }
443            DydxWsMarketsMessage::ChannelData(data) => self.deserialize_markets(&data),
444            DydxWsMarketsMessage::Unsubscribed(data) => {
445                let topic = self.topic_from_msg(&DydxWsChannel::Markets, &data.id);
446                self.subscriptions.confirm_unsubscribe(&topic);
447                vec![]
448            }
449        }
450    }
451
452    fn handle_candles_feed(&self, msg: DydxWsCandlesMessage) -> Vec<DydxWsOutputMessage> {
453        match msg {
454            DydxWsCandlesMessage::Subscribed(data) => {
455                let topic = self.topic_from_msg(&DydxWsChannel::Candles, &data.id);
456                self.subscriptions.confirm_subscribe(&topic);
457                vec![]
458            }
459            DydxWsCandlesMessage::ChannelData(data) => self.deserialize_candles(&data),
460            DydxWsCandlesMessage::Unsubscribed(data) => {
461                let topic = self.topic_from_msg(&DydxWsChannel::Candles, &data.id);
462                self.subscriptions.confirm_unsubscribe(&topic);
463                vec![]
464            }
465        }
466    }
467
468    fn handle_parent_subaccounts(
469        &self,
470        msg: DydxWsParentSubaccountsMessage,
471    ) -> Vec<DydxWsOutputMessage> {
472        match msg {
473            DydxWsParentSubaccountsMessage::Subscribed(data) => {
474                let topic = self.topic_from_msg(&DydxWsChannel::ParentSubaccounts, &data.id);
475                self.subscriptions.confirm_subscribe(&topic);
476                self.deserialize_parent_subaccounts(&data)
477            }
478            DydxWsParentSubaccountsMessage::ChannelData(data) => {
479                self.deserialize_parent_subaccounts(&data)
480            }
481            DydxWsParentSubaccountsMessage::Unsubscribed(data) => {
482                let topic = self.topic_from_msg(&DydxWsChannel::ParentSubaccounts, &data.id);
483                self.subscriptions.confirm_unsubscribe(&topic);
484                vec![]
485            }
486        }
487    }
488
489    fn handle_block_height_feed(&self, msg: DydxWsBlockHeightMessage) -> Vec<DydxWsOutputMessage> {
490        match msg {
491            DydxWsBlockHeightMessage::Subscribed(data) => {
492                let topic =
493                    self.topic_from_msg(&DydxWsChannel::BlockHeight, &Some(data.id.clone()));
494                self.subscriptions.confirm_subscribe(&topic);
495
496                match data.contents.height.parse::<u64>() {
497                    Ok(height) => vec![DydxWsOutputMessage::BlockHeight {
498                        height,
499                        time: data.contents.time,
500                    }],
501                    Err(e) => {
502                        log::warn!("Failed to parse block height from subscription: {e}");
503                        vec![]
504                    }
505                }
506            }
507            DydxWsBlockHeightMessage::ChannelData(data) => {
508                match data.contents.block_height.parse::<u64>() {
509                    Ok(height) => vec![DydxWsOutputMessage::BlockHeight {
510                        height,
511                        time: data.contents.time,
512                    }],
513                    Err(e) => {
514                        log::warn!("Failed to parse block height from channel data: {e}");
515                        vec![]
516                    }
517                }
518            }
519            DydxWsBlockHeightMessage::Unsubscribed(data) => {
520                let topic = self.topic_from_msg(&DydxWsChannel::BlockHeight, &data.id);
521                self.subscriptions.confirm_unsubscribe(&topic);
522                vec![]
523            }
524        }
525    }
526
527    fn deserialize_trades(&self, data: &DydxWsChannelDataMsg) -> Vec<DydxWsOutputMessage> {
528        let Some(id) = data.id.clone() else {
529            log::error!("Missing id for trades channel");
530            return vec![];
531        };
532
533        match serde_json::from_value::<DydxTradeContents>(data.contents.clone()) {
534            Ok(contents) => vec![DydxWsOutputMessage::Trades { id, contents }],
535            Err(e) => {
536                log::error!("Failed to deserialize trade contents: {e}");
537                vec![]
538            }
539        }
540    }
541
542    fn deserialize_orderbook_snapshot(
543        &self,
544        data: &DydxWsChannelDataMsg,
545    ) -> Vec<DydxWsOutputMessage> {
546        let Some(id) = data.id.clone() else {
547            log::error!("Missing id for orderbook snapshot");
548            return vec![];
549        };
550
551        match serde_json::from_value::<DydxOrderbookSnapshotContents>(data.contents.clone()) {
552            Ok(contents) => vec![DydxWsOutputMessage::OrderbookSnapshot { id, contents }],
553            Err(e) => {
554                log::error!("Failed to deserialize orderbook snapshot: {e}");
555                vec![]
556            }
557        }
558    }
559
560    fn deserialize_orderbook_update(
561        &self,
562        data: &DydxWsChannelDataMsg,
563    ) -> Vec<DydxWsOutputMessage> {
564        let Some(id) = data.id.clone() else {
565            log::error!("Missing id for orderbook update");
566            return vec![];
567        };
568
569        match serde_json::from_value::<DydxOrderbookContents>(data.contents.clone()) {
570            Ok(contents) => vec![DydxWsOutputMessage::OrderbookUpdate { id, contents }],
571            Err(e) => {
572                log::error!("Failed to deserialize orderbook contents: {e}");
573                vec![]
574            }
575        }
576    }
577
578    fn deserialize_orderbook_batch(
579        &self,
580        data: &DydxWsChannelBatchDataMsg,
581    ) -> Vec<DydxWsOutputMessage> {
582        let Some(id) = data.id.clone() else {
583            log::error!("Missing id for orderbook batch");
584            return vec![];
585        };
586
587        match serde_json::from_value::<Vec<DydxOrderbookContents>>(data.contents.clone()) {
588            Ok(updates) => vec![DydxWsOutputMessage::OrderbookBatch { id, updates }],
589            Err(e) => {
590                log::error!("Failed to deserialize orderbook batch: {e}");
591                vec![]
592            }
593        }
594    }
595
596    fn deserialize_candles(&self, data: &DydxWsChannelDataMsg) -> Vec<DydxWsOutputMessage> {
597        let Some(id) = data.id.clone() else {
598            log::error!("Missing id for candles channel");
599            return vec![];
600        };
601
602        match serde_json::from_value::<DydxCandle>(data.contents.clone()) {
603            Ok(contents) => vec![DydxWsOutputMessage::Candles { id, contents }],
604            Err(e) => {
605                log::error!("Failed to deserialize candle contents: {e}");
606                vec![]
607            }
608        }
609    }
610
611    fn deserialize_markets(&self, data: &DydxWsChannelDataMsg) -> Vec<DydxWsOutputMessage> {
612        match serde_json::from_value::<DydxMarketsContents>(data.contents.clone()) {
613            Ok(contents) => vec![DydxWsOutputMessage::Markets(contents)],
614            Err(e) => {
615                log::error!("Failed to deserialize markets contents: {e}");
616                vec![]
617            }
618        }
619    }
620
621    fn deserialize_parent_subaccounts(
622        &self,
623        data: &DydxWsChannelDataMsg,
624    ) -> Vec<DydxWsOutputMessage> {
625        match serde_json::from_value::<DydxWsSubaccountsChannelContents>(data.contents.clone()) {
626            Ok(contents) => {
627                let has_orders = contents.orders.as_ref().is_some_and(|o| !o.is_empty());
628                let has_fills = contents.fills.as_ref().is_some_and(|f| !f.is_empty());
629
630                if has_orders || has_fills {
631                    let channel_data = DydxWsSubaccountsChannelData {
632                        connection_id: data.connection_id.clone(),
633                        message_id: data.message_id,
634                        id: data.id.clone().unwrap_or_default(),
635                        version: data.version.clone().unwrap_or_default(),
636                        contents,
637                    };
638                    vec![DydxWsOutputMessage::SubaccountsChannelData(Box::new(
639                        channel_data,
640                    ))]
641                } else {
642                    vec![]
643                }
644            }
645            Err(e) => {
646                log::error!("Failed to deserialize parent subaccounts contents: {e}");
647                vec![]
648            }
649        }
650    }
651
652    async fn handle_command(&mut self, command: HandlerCommand) -> bool {
653        match command {
654            HandlerCommand::RegisterSubscription {
655                topic,
656                subscription,
657            } => {
658                self.subscription_messages.insert(topic, subscription);
659            }
660            HandlerCommand::UnregisterSubscription { topic } => {
661                self.subscription_messages.remove(&topic);
662            }
663            HandlerCommand::SendText(text) => {
664                if let Err(e) = self
665                    .send_with_retry(text, Some(DYDX_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
666                    .await
667                {
668                    log::error!("Failed to send WebSocket text after retries: {e}");
669                }
670            }
671            HandlerCommand::Disconnect => {
672                log::debug!("Disconnect command received");
673                self.client.disconnect().await;
674                return true;
675            }
676        }
677        false
678    }
679
680    fn topic_from_msg(&self, channel: &DydxWsChannel, id: &Option<String>) -> String {
681        match id {
682            Some(id) => format!(
683                "{}{}{}",
684                channel.as_ref(),
685                self.subscriptions.delimiter(),
686                id
687            ),
688            None => channel.as_ref().to_string(),
689        }
690    }
691
692    fn clear_state(&mut self) {
693        let buffer_count = self.message_buffer.len();
694        let seq_count = self.book_sequence.len();
695        self.message_buffer.clear();
696        self.book_sequence.clear();
697        log::debug!(
698            "Cleared reconnect state: message_buffer={buffer_count}, book_sequence={seq_count}"
699        );
700    }
701
702    async fn replay_subscriptions(&self) -> DydxWsResult<()> {
703        let topics = self.subscriptions.all_topics();
704        for topic in topics {
705            let Some(subscription) = self.subscription_messages.get(&topic).cloned() else {
706                log::warn!("No preserved subscription message for topic: {topic}");
707                continue;
708            };
709
710            let payload = serde_json::to_string(&subscription)?;
711            self.subscriptions.mark_subscribe(&topic);
712
713            if let Err(e) = self
714                .send_with_retry(payload, Some(DYDX_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
715                .await
716            {
717                self.subscriptions.mark_failure(&topic);
718                return Err(e);
719            }
720        }
721
722        Ok(())
723    }
724
725    /// Handles control messages from the fallback parsing path.
726    ///
727    /// Channel data is handled directly via `handle_feed_message()`.
728    ///
729    /// # Errors
730    ///
731    /// Returns an error if the message cannot be processed.
732    pub async fn handle_message(
733        &mut self,
734        msg: DydxWsMessage,
735    ) -> DydxWsResult<Vec<DydxWsOutputMessage>> {
736        match msg {
737            DydxWsMessage::Connected(_) => {
738                log::debug!("dYdX WebSocket connected");
739                Ok(vec![])
740            }
741            DydxWsMessage::Subscribed(sub) => {
742                log::debug!("Subscribed to {} (id: {:?})", sub.channel, sub.id);
743                let topic = self.topic_from_msg(&sub.channel, &sub.id);
744                self.subscriptions.confirm_subscribe(&topic);
745                Ok(vec![])
746            }
747            DydxWsMessage::SubaccountsSubscribed(msg) => {
748                log::debug!("Subaccounts subscribed with initial state (fallback path)");
749                let topic = self.topic_from_msg(&DydxWsChannel::Subaccounts, &Some(msg.id.clone()));
750                self.subscriptions.confirm_subscribe(&topic);
751                Ok(vec![DydxWsOutputMessage::SubaccountSubscribed(Box::new(
752                    msg,
753                ))])
754            }
755            DydxWsMessage::Unsubscribed(unsub) => {
756                log::debug!("Unsubscribed from {} (id: {:?})", unsub.channel, unsub.id);
757                let topic = self.topic_from_msg(&unsub.channel, &unsub.id);
758                self.subscriptions.confirm_unsubscribe(&topic);
759                Ok(vec![])
760            }
761            DydxWsMessage::Error(err) => Ok(vec![DydxWsOutputMessage::Error(err)]),
762            DydxWsMessage::Reconnected => {
763                self.clear_state();
764
765                if let Err(e) = self.replay_subscriptions().await {
766                    log::error!("Failed to replay subscriptions after reconnect message: {e}");
767                }
768                let topics = self.subscriptions.all_topics();
769                Ok(vec![DydxWsOutputMessage::Reconnected { topics }])
770            }
771            DydxWsMessage::Pong => Ok(vec![]),
772            DydxWsMessage::Raw(_) => Ok(vec![]),
773        }
774    }
775}
776
777/// Determines if a dYdX WebSocket error should trigger a retry.
778fn should_retry_dydx_error(error: &DydxWsError) -> bool {
779    match error {
780        DydxWsError::Transport(_) => true,
781        DydxWsError::Send(_) => true,
782        DydxWsError::ClientError(msg) => {
783            let msg_lower = msg.to_lowercase();
784            msg_lower.contains("timeout")
785                || msg_lower.contains("timed out")
786                || msg_lower.contains("connection")
787                || msg_lower.contains("network")
788        }
789        DydxWsError::NotConnected
790        | DydxWsError::Json(_)
791        | DydxWsError::Parse(_)
792        | DydxWsError::Authentication(_)
793        | DydxWsError::Subscription(_)
794        | DydxWsError::Venue(_) => false,
795    }
796}
797
798/// Creates a timeout error for the retry manager.
799fn create_dydx_timeout_error(msg: String) -> DydxWsError {
800    DydxWsError::ClientError(msg)
801}
nautilus_dydx/websocket/handler.rs

nautilus_dydx/websocket/
handler.rs
