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            .invocation(
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                |e| create_dydx_timeout_error(e.to_string()),
143            )
144            .execute()
145            .await
146    }
147
148    /// Main processing loop for the handler.
149    ///
150    /// Message extraction cannot panic because the message buffer and decoded message vector are
151    /// checked for emptiness first.
152    #[allow(
153        clippy::missing_panics_doc,
154        reason = "message buffers are checked for emptiness before extraction"
155    )]
156    pub async fn run(&mut self) {
157        log::debug!("WebSocket handler started");
158
159        loop {
160            // First drain any buffered messages
161            if !self.message_buffer.is_empty() {
162                let msg = self.message_buffer.pop_front().unwrap();
163                if self.out_tx.send(msg).is_err() {
164                    log::debug!("Receiver dropped, stopping handler");
165                    break;
166                }
167                continue;
168            }
169
170            tokio::select! {
171                Some(cmd) = self.cmd_rx.recv() => {
172                    if self.handle_command(cmd).await {
173                        break;
174                    }
175                }
176
177                Some(msg) = self.raw_rx.recv() => {
178                    log::trace!("Handler received raw message");
179                    let msgs = self.process_raw_message(msg).await;
180                    if !msgs.is_empty() {
181                        let mut iter = msgs.into_iter();
182                        // We just checked that msgs is not empty
183                        let first = iter.next().expect("non-empty vec has first element");
184                        self.message_buffer.extend(iter);
185                        log::trace!("Handler sending message: {:?}", std::mem::discriminant(&first));
186                        if self.out_tx.send(first).is_err() {
187                            log::debug!("Receiver dropped, stopping handler");
188                            break;
189                        }
190                    }
191                }
192
193                else => {
194                    log::debug!("Handler shutting down: channels closed");
195                    break;
196                }
197            }
198
199            if self.signal.load(Ordering::Acquire) {
200                log::debug!("Handler received stop signal");
201                break;
202            }
203        }
204    }
205
206    async fn process_raw_message(&mut self, msg: Message) -> Vec<DydxWsOutputMessage> {
207        match msg {
208            Message::Text(txt) => {
209                if txt == RECONNECTED {
210                    self.clear_state();
211                    self.subscriptions.reset_after_reconnect();
212
213                    if let Err(e) = self.replay_subscriptions().await {
214                        log::error!("Failed to replay subscriptions after reconnect: {e}");
215                    }
216                    let topics = self.subscriptions.all_topics();
217                    return vec![DydxWsOutputMessage::Reconnected { topics }];
218                }
219
220                // Hot path: zero-copy parse for feed messages (orderbook/trades/candles)
221                match serde_json::from_str::<DydxWsFeedMessage>(&txt) {
222                    Ok(feed_msg) => {
223                        return self.handle_feed_message(feed_msg);
224                    }
225                    Err(e) => {
226                        if txt.contains("v4_subaccounts") {
227                            log::warn!(
228                                "[WS_DESER] Failed to parse v4_subaccounts as DydxWsFeedMessage: {e}\nRaw: {txt}"
229                            );
230                        }
231                    }
232                }
233
234                // Cold path: infrequent control messages (connected/subscribed/error)
235                match serde_json::from_str::<serde_json::Value>(&txt) {
236                    Ok(val) => match serde_json::from_value::<DydxWsGenericMsg>(val.clone()) {
237                        Ok(meta) => {
238                            let result = if meta.is_connected() {
239                                serde_json::from_value::<DydxWsConnectedMsg>(val)
240                                    .map(DydxWsMessage::Connected)
241                            } else if meta.is_subscribed() {
242                                log::debug!("Processing subscribed message via fallback path");
243
244                                if let Ok(sub_msg) =
245                                    serde_json::from_value::<DydxWsSubscriptionMsg>(val.clone())
246                                {
247                                    if sub_msg.channel == DydxWsChannel::Subaccounts {
248                                        log::debug!("Parsing subaccounts subscription (fallback)");
249                                        serde_json::from_value::<DydxWsSubaccountsSubscribed>(val)
250                                            .map(DydxWsMessage::SubaccountsSubscribed)
251                                            .or_else(|e| {
252                                                log::warn!(
253                                                    "Failed to parse subaccounts subscription: {e}"
254                                                );
255                                                Ok(DydxWsMessage::Subscribed(sub_msg))
256                                            })
257                                    } else {
258                                        Ok(DydxWsMessage::Subscribed(sub_msg))
259                                    }
260                                } else {
261                                    serde_json::from_value::<DydxWsSubscriptionMsg>(val)
262                                        .map(DydxWsMessage::Subscribed)
263                                }
264                            } else if meta.is_unsubscribed() {
265                                serde_json::from_value::<DydxWsSubscriptionMsg>(val)
266                                    .map(DydxWsMessage::Unsubscribed)
267                            } else if meta.is_error() {
268                                serde_json::from_value::<DydxWebSocketError>(val)
269                                    .map(DydxWsMessage::Error)
270                            } else if meta.is_unknown() {
271                                log::warn!("Received unknown WebSocket message type: {txt}",);
272                                Ok(DydxWsMessage::Raw(val))
273                            } else {
274                                Ok(DydxWsMessage::Raw(val))
275                            };
276
277                            match result {
278                                Ok(dydx_msg) => self.handle_dydx_message(dydx_msg).await,
279                                Err(e) => {
280                                    log::error!(
281                                        "Failed to parse WebSocket message: {e}. Message type: {:?}, Channel: {:?}. Raw: {txt}",
282                                        meta.msg_type,
283                                        meta.channel,
284                                    );
285                                    vec![]
286                                }
287                            }
288                        }
289                        Err(e) => {
290                            log::error!(
291                                "Failed to parse WebSocket message envelope (DydxWsGenericMsg): {e}\nRaw JSON:\n{txt}"
292                            );
293                            vec![]
294                        }
295                    },
296                    Err(e) => {
297                        let err = DydxWebSocketError::from_message(e.to_string());
298                        vec![DydxWsOutputMessage::Error(err)]
299                    }
300                }
301            }
302            Message::Pong(_data) => vec![],
303            Message::Ping(_data) => vec![],
304            Message::Binary(_bin) => vec![],
305            Message::Close(_frame) => {
306                log::debug!("WebSocket close frame received");
307                vec![]
308            }
309            Message::Frame(_) => vec![],
310        }
311    }
312
313    async fn handle_dydx_message(&mut self, msg: DydxWsMessage) -> Vec<DydxWsOutputMessage> {
314        match self.handle_message(msg).await {
315            Ok(msgs) => msgs,
316            Err(e) => {
317                log::error!("Error handling message: {e}");
318                vec![]
319            }
320        }
321    }
322
323    fn handle_feed_message(&mut self, feed_msg: DydxWsFeedMessage) -> Vec<DydxWsOutputMessage> {
324        log::trace!(
325            "Handling feed message: {:?}",
326            std::mem::discriminant(&feed_msg)
327        );
328
329        match feed_msg {
330            DydxWsFeedMessage::Subaccounts(msg) => self.handle_subaccounts(msg),
331            DydxWsFeedMessage::Orderbook(msg) => self.handle_orderbook(msg),
332            DydxWsFeedMessage::Trades(msg) => self.handle_trades(msg),
333            DydxWsFeedMessage::Markets(msg) => self.handle_markets_feed(msg),
334            DydxWsFeedMessage::Candles(msg) => self.handle_candles_feed(msg),
335            DydxWsFeedMessage::ParentSubaccounts(msg) => self.handle_parent_subaccounts(msg),
336            DydxWsFeedMessage::BlockHeight(msg) => self.handle_block_height_feed(msg),
337        }
338    }
339
340    fn handle_subaccounts(&self, msg: DydxWsSubaccountsMessage) -> Vec<DydxWsOutputMessage> {
341        match msg {
342            DydxWsSubaccountsMessage::Subscribed(data) => {
343                let topic =
344                    self.topic_from_msg(&DydxWsChannel::Subaccounts, &Some(data.id.clone()));
345                self.subscriptions.confirm_subscribe(&topic);
346                log::debug!("Forwarding subaccount subscription to execution client");
347                vec![DydxWsOutputMessage::SubaccountSubscribed(Box::new(data))]
348            }
349            DydxWsSubaccountsMessage::ChannelData(data) => {
350                let has_orders = data.contents.orders.as_ref().is_some_and(|o| !o.is_empty());
351                let has_fills = data.contents.fills.as_ref().is_some_and(|f| !f.is_empty());
352
353                if has_orders || has_fills {
354                    log::debug!(
355                        "Received {} order(s), {} fill(s) - forwarding to execution client",
356                        data.contents.orders.as_ref().map_or(0, |o| o.len()),
357                        data.contents.fills.as_ref().map_or(0, |f| f.len())
358                    );
359                    vec![DydxWsOutputMessage::SubaccountsChannelData(Box::new(data))]
360                } else {
361                    vec![]
362                }
363            }
364            DydxWsSubaccountsMessage::Unsubscribed(data) => {
365                let topic = self.topic_from_msg(&DydxWsChannel::Subaccounts, &data.id);
366                self.subscriptions.confirm_unsubscribe(&topic);
367                vec![]
368            }
369        }
370    }
371
372    fn handle_orderbook(&mut self, msg: DydxWsOrderbookMessage) -> Vec<DydxWsOutputMessage> {
373        match msg {
374            DydxWsOrderbookMessage::Subscribed(data) => {
375                let topic = self.topic_from_msg(&DydxWsChannel::Orderbook, &data.id);
376                self.subscriptions.confirm_subscribe(&topic);
377
378                if let Some(id) = &data.id {
379                    self.book_sequence.insert(id.clone(), data.message_id);
380                }
381
382                self.deserialize_orderbook_snapshot(&data)
383            }
384            DydxWsOrderbookMessage::ChannelData(data) => {
385                if let Some(id) = &data.id {
386                    if let Some(last_id) = self.book_sequence.get(id)
387                        && data.message_id <= *last_id
388                    {
389                        log::warn!(
390                            "Orderbook sequence regression for {id}: last {last_id}, received {}",
391                            data.message_id
392                        );
393                    }
394                    self.book_sequence.insert(id.clone(), data.message_id);
395                }
396                self.deserialize_orderbook_update(&data)
397            }
398            DydxWsOrderbookMessage::ChannelBatchData(data) => {
399                if let Some(id) = &data.id {
400                    if let Some(last_id) = self.book_sequence.get(id)
401                        && data.message_id <= *last_id
402                    {
403                        log::warn!(
404                            "Orderbook batch sequence regression for {id}: last {last_id}, received {}",
405                            data.message_id
406                        );
407                    }
408                    self.book_sequence.insert(id.clone(), data.message_id);
409                }
410                self.deserialize_orderbook_batch(&data)
411            }
412            DydxWsOrderbookMessage::Unsubscribed(data) => {
413                let topic = self.topic_from_msg(&DydxWsChannel::Orderbook, &data.id);
414                self.subscriptions.confirm_unsubscribe(&topic);
415
416                if let Some(id) = &data.id {
417                    self.book_sequence.remove(id);
418                }
419                vec![]
420            }
421        }
422    }
423
424    fn handle_trades(&self, msg: DydxWsTradesMessage) -> Vec<DydxWsOutputMessage> {
425        match msg {
426            DydxWsTradesMessage::Subscribed(data) => {
427                let topic = self.topic_from_msg(&DydxWsChannel::Trades, &data.id);
428                self.subscriptions.confirm_subscribe(&topic);
429                self.deserialize_trades(&data)
430            }
431            DydxWsTradesMessage::ChannelData(data) => self.deserialize_trades(&data),
432            DydxWsTradesMessage::Unsubscribed(data) => {
433                let topic = self.topic_from_msg(&DydxWsChannel::Trades, &data.id);
434                self.subscriptions.confirm_unsubscribe(&topic);
435                vec![]
436            }
437        }
438    }
439
440    fn handle_markets_feed(&self, msg: DydxWsMarketsMessage) -> Vec<DydxWsOutputMessage> {
441        match msg {
442            DydxWsMarketsMessage::Subscribed(data) => {
443                let topic = self.topic_from_msg(&DydxWsChannel::Markets, &data.id);
444                self.subscriptions.confirm_subscribe(&topic);
445                self.deserialize_markets(&data)
446            }
447            DydxWsMarketsMessage::ChannelData(data) => self.deserialize_markets(&data),
448            DydxWsMarketsMessage::Unsubscribed(data) => {
449                let topic = self.topic_from_msg(&DydxWsChannel::Markets, &data.id);
450                self.subscriptions.confirm_unsubscribe(&topic);
451                vec![]
452            }
453        }
454    }
455
456    fn handle_candles_feed(&self, msg: DydxWsCandlesMessage) -> Vec<DydxWsOutputMessage> {
457        match msg {
458            DydxWsCandlesMessage::Subscribed(data) => {
459                let topic = self.topic_from_msg(&DydxWsChannel::Candles, &data.id);
460                self.subscriptions.confirm_subscribe(&topic);
461                vec![]
462            }
463            DydxWsCandlesMessage::ChannelData(data) => self.deserialize_candles(&data),
464            DydxWsCandlesMessage::Unsubscribed(data) => {
465                let topic = self.topic_from_msg(&DydxWsChannel::Candles, &data.id);
466                self.subscriptions.confirm_unsubscribe(&topic);
467                vec![]
468            }
469        }
470    }
471
472    fn handle_parent_subaccounts(
473        &self,
474        msg: DydxWsParentSubaccountsMessage,
475    ) -> Vec<DydxWsOutputMessage> {
476        match msg {
477            DydxWsParentSubaccountsMessage::Subscribed(data) => {
478                let topic = self.topic_from_msg(&DydxWsChannel::ParentSubaccounts, &data.id);
479                self.subscriptions.confirm_subscribe(&topic);
480                self.deserialize_parent_subaccounts(&data)
481            }
482            DydxWsParentSubaccountsMessage::ChannelData(data) => {
483                self.deserialize_parent_subaccounts(&data)
484            }
485            DydxWsParentSubaccountsMessage::Unsubscribed(data) => {
486                let topic = self.topic_from_msg(&DydxWsChannel::ParentSubaccounts, &data.id);
487                self.subscriptions.confirm_unsubscribe(&topic);
488                vec![]
489            }
490        }
491    }
492
493    fn handle_block_height_feed(&self, msg: DydxWsBlockHeightMessage) -> Vec<DydxWsOutputMessage> {
494        match msg {
495            DydxWsBlockHeightMessage::Subscribed(data) => {
496                let topic =
497                    self.topic_from_msg(&DydxWsChannel::BlockHeight, &Some(data.id.clone()));
498                self.subscriptions.confirm_subscribe(&topic);
499
500                match data.contents.height.parse::<u64>() {
501                    Ok(height) => vec![DydxWsOutputMessage::BlockHeight {
502                        height,
503                        time: data.contents.time,
504                    }],
505                    Err(e) => {
506                        log::warn!("Failed to parse block height from subscription: {e}");
507                        vec![]
508                    }
509                }
510            }
511            DydxWsBlockHeightMessage::ChannelData(data) => {
512                match data.contents.block_height.parse::<u64>() {
513                    Ok(height) => vec![DydxWsOutputMessage::BlockHeight {
514                        height,
515                        time: data.contents.time,
516                    }],
517                    Err(e) => {
518                        log::warn!("Failed to parse block height from channel data: {e}");
519                        vec![]
520                    }
521                }
522            }
523            DydxWsBlockHeightMessage::Unsubscribed(data) => {
524                let topic = self.topic_from_msg(&DydxWsChannel::BlockHeight, &data.id);
525                self.subscriptions.confirm_unsubscribe(&topic);
526                vec![]
527            }
528        }
529    }
530
531    fn deserialize_trades(&self, data: &DydxWsChannelDataMsg) -> Vec<DydxWsOutputMessage> {
532        let Some(id) = data.id.clone() else {
533            log::error!("Missing id for trades channel");
534            return vec![];
535        };
536
537        match serde_json::from_value::<DydxTradeContents>(data.contents.clone()) {
538            Ok(contents) => vec![DydxWsOutputMessage::Trades { id, contents }],
539            Err(e) => {
540                log::error!("Failed to deserialize trade contents: {e}");
541                vec![]
542            }
543        }
544    }
545
546    fn deserialize_orderbook_snapshot(
547        &self,
548        data: &DydxWsChannelDataMsg,
549    ) -> Vec<DydxWsOutputMessage> {
550        let Some(id) = data.id.clone() else {
551            log::error!("Missing id for orderbook snapshot");
552            return vec![];
553        };
554
555        match serde_json::from_value::<DydxOrderbookSnapshotContents>(data.contents.clone()) {
556            Ok(contents) => vec![DydxWsOutputMessage::OrderbookSnapshot { id, contents }],
557            Err(e) => {
558                log::error!("Failed to deserialize orderbook snapshot: {e}");
559                vec![]
560            }
561        }
562    }
563
564    fn deserialize_orderbook_update(
565        &self,
566        data: &DydxWsChannelDataMsg,
567    ) -> Vec<DydxWsOutputMessage> {
568        let Some(id) = data.id.clone() else {
569            log::error!("Missing id for orderbook update");
570            return vec![];
571        };
572
573        match serde_json::from_value::<DydxOrderbookContents>(data.contents.clone()) {
574            Ok(contents) => vec![DydxWsOutputMessage::OrderbookUpdate { id, contents }],
575            Err(e) => {
576                log::error!("Failed to deserialize orderbook contents: {e}");
577                vec![]
578            }
579        }
580    }
581
582    fn deserialize_orderbook_batch(
583        &self,
584        data: &DydxWsChannelBatchDataMsg,
585    ) -> Vec<DydxWsOutputMessage> {
586        let Some(id) = data.id.clone() else {
587            log::error!("Missing id for orderbook batch");
588            return vec![];
589        };
590
591        match serde_json::from_value::<Vec<DydxOrderbookContents>>(data.contents.clone()) {
592            Ok(updates) => vec![DydxWsOutputMessage::OrderbookBatch { id, updates }],
593            Err(e) => {
594                log::error!("Failed to deserialize orderbook batch: {e}");
595                vec![]
596            }
597        }
598    }
599
600    fn deserialize_candles(&self, data: &DydxWsChannelDataMsg) -> Vec<DydxWsOutputMessage> {
601        let Some(id) = data.id.clone() else {
602            log::error!("Missing id for candles channel");
603            return vec![];
604        };
605
606        match serde_json::from_value::<DydxCandle>(data.contents.clone()) {
607            Ok(contents) => vec![DydxWsOutputMessage::Candles { id, contents }],
608            Err(e) => {
609                log::error!("Failed to deserialize candle contents: {e}");
610                vec![]
611            }
612        }
613    }
614
615    fn deserialize_markets(&self, data: &DydxWsChannelDataMsg) -> Vec<DydxWsOutputMessage> {
616        match serde_json::from_value::<DydxMarketsContents>(data.contents.clone()) {
617            Ok(contents) => vec![DydxWsOutputMessage::Markets(contents)],
618            Err(e) => {
619                log::error!("Failed to deserialize markets contents: {e}");
620                vec![]
621            }
622        }
623    }
624
625    fn deserialize_parent_subaccounts(
626        &self,
627        data: &DydxWsChannelDataMsg,
628    ) -> Vec<DydxWsOutputMessage> {
629        match serde_json::from_value::<DydxWsSubaccountsChannelContents>(data.contents.clone()) {
630            Ok(contents) => {
631                let has_orders = contents.orders.as_ref().is_some_and(|o| !o.is_empty());
632                let has_fills = contents.fills.as_ref().is_some_and(|f| !f.is_empty());
633
634                if has_orders || has_fills {
635                    let channel_data = DydxWsSubaccountsChannelData {
636                        connection_id: data.connection_id.clone(),
637                        message_id: data.message_id,
638                        id: data.id.clone().unwrap_or_default(),
639                        version: data.version.clone().unwrap_or_default(),
640                        contents,
641                    };
642                    vec![DydxWsOutputMessage::SubaccountsChannelData(Box::new(
643                        channel_data,
644                    ))]
645                } else {
646                    vec![]
647                }
648            }
649            Err(e) => {
650                log::error!("Failed to deserialize parent subaccounts contents: {e}");
651                vec![]
652            }
653        }
654    }
655
656    async fn handle_command(&mut self, command: HandlerCommand) -> bool {
657        match command {
658            HandlerCommand::RegisterSubscription {
659                topic,
660                subscription,
661            } => {
662                self.subscription_messages.insert(topic, subscription);
663            }
664            HandlerCommand::UnregisterSubscription { topic } => {
665                self.subscription_messages.remove(&topic);
666            }
667            HandlerCommand::SendText(text) => {
668                if let Err(e) = self
669                    .send_with_retry(text, Some(DYDX_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
670                    .await
671                {
672                    log::error!("Failed to send WebSocket text after retries: {e}");
673                }
674            }
675            HandlerCommand::Disconnect => {
676                log::debug!("Disconnect command received");
677                self.client.disconnect().await;
678                return true;
679            }
680        }
681        false
682    }
683
684    fn topic_from_msg(&self, channel: &DydxWsChannel, id: &Option<String>) -> String {
685        if matches!(channel, DydxWsChannel::BlockHeight) {
686            return channel.as_ref().to_string();
687        }
688
689        match id {
690            Some(id) => format!(
691                "{}{}{}",
692                channel.as_ref(),
693                self.subscriptions.delimiter(),
694                id
695            ),
696            None => channel.as_ref().to_string(),
697        }
698    }
699
700    fn clear_state(&mut self) {
701        let buffer_count = self.message_buffer.len();
702        let seq_count = self.book_sequence.len();
703        self.message_buffer.clear();
704        self.book_sequence.clear();
705        log::debug!(
706            "Cleared reconnect state: message_buffer={buffer_count}, book_sequence={seq_count}"
707        );
708    }
709
710    async fn replay_subscriptions(&self) -> DydxWsResult<()> {
711        let topics = self.subscriptions.all_topics();
712        for topic in topics {
713            let Some(subscription) = self.subscription_messages.get(&topic).cloned() else {
714                log::warn!("No preserved subscription message for topic: {topic}");
715                continue;
716            };
717
718            let payload = serde_json::to_string(&subscription)?;
719            self.subscriptions.mark_subscribe(&topic);
720
721            if let Err(e) = self
722                .send_with_retry(payload, Some(DYDX_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
723                .await
724            {
725                self.subscriptions.mark_failure(&topic);
726                return Err(e);
727            }
728        }
729
730        Ok(())
731    }
732
733    /// Handles control messages from the fallback parsing path.
734    ///
735    /// Channel data is handled directly via `handle_feed_message()`.
736    ///
737    /// # Errors
738    ///
739    /// Returns an error if the message cannot be processed.
740    pub async fn handle_message(
741        &mut self,
742        msg: DydxWsMessage,
743    ) -> DydxWsResult<Vec<DydxWsOutputMessage>> {
744        match msg {
745            DydxWsMessage::Connected(_) => {
746                log::debug!("dYdX WebSocket connected");
747                Ok(vec![])
748            }
749            DydxWsMessage::Subscribed(sub) => {
750                log::debug!("Subscribed to {} (id: {:?})", sub.channel, sub.id);
751                let topic = self.topic_from_msg(&sub.channel, &sub.id);
752                self.subscriptions.confirm_subscribe(&topic);
753                Ok(vec![])
754            }
755            DydxWsMessage::SubaccountsSubscribed(msg) => {
756                log::debug!("Subaccounts subscribed with initial state (fallback path)");
757                let topic = self.topic_from_msg(&DydxWsChannel::Subaccounts, &Some(msg.id.clone()));
758                self.subscriptions.confirm_subscribe(&topic);
759                Ok(vec![DydxWsOutputMessage::SubaccountSubscribed(Box::new(
760                    msg,
761                ))])
762            }
763            DydxWsMessage::Unsubscribed(unsub) => {
764                log::debug!("Unsubscribed from {} (id: {:?})", unsub.channel, unsub.id);
765                let topic = self.topic_from_msg(&unsub.channel, &unsub.id);
766                self.subscriptions.confirm_unsubscribe(&topic);
767                Ok(vec![])
768            }
769            DydxWsMessage::Error(err) => Ok(vec![DydxWsOutputMessage::Error(err)]),
770            DydxWsMessage::Reconnected => {
771                self.clear_state();
772                self.subscriptions.reset_after_reconnect();
773
774                if let Err(e) = self.replay_subscriptions().await {
775                    log::error!("Failed to replay subscriptions after reconnect message: {e}");
776                }
777                let topics = self.subscriptions.all_topics();
778                Ok(vec![DydxWsOutputMessage::Reconnected { topics }])
779            }
780            DydxWsMessage::Pong => Ok(vec![]),
781            DydxWsMessage::Raw(_) => Ok(vec![]),
782        }
783    }
784}
785
786/// Determines if a dYdX WebSocket error should trigger a retry.
787fn should_retry_dydx_error(error: &DydxWsError) -> bool {
788    match error {
789        DydxWsError::Transport(_) => true,
790        DydxWsError::Send(_) => true,
791        DydxWsError::ClientError(msg) => {
792            let msg_lower = msg.to_lowercase();
793            msg_lower.contains("timeout")
794                || msg_lower.contains("timed out")
795                || msg_lower.contains("connection")
796                || msg_lower.contains("network")
797        }
798        DydxWsError::NotConnected
799        | DydxWsError::Json(_)
800        | DydxWsError::Parse(_)
801        | DydxWsError::Authentication(_)
802        | DydxWsError::Subscription(_)
803        | DydxWsError::Venue(_) => false,
804    }
805}
806
807/// Creates a timeout error for the retry manager.
808fn create_dydx_timeout_error(msg: String) -> DydxWsError {
809    DydxWsError::ClientError(msg)
810}
nautilus_dydx/websocket/handler.rs

nautilus_dydx/websocket/
handler.rs
