nautilus_lighter/websocket/
client.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//! Outer WebSocket client orchestrating connection lifecycle and subscriptions.
17
18use std::{
19    fmt::Debug,
20    sync::{
21        Arc,
22        atomic::{AtomicBool, AtomicU8, Ordering},
23    },
24    time::Duration,
25};
26
27use arc_swap::ArcSwap;
28use dashmap::DashMap;
29use nautilus_common::live::get_runtime;
30use nautilus_model::{
31    identifiers::{AccountId, InstrumentId},
32    instruments::InstrumentAny,
33};
34use nautilus_network::{
35    mode::ConnectionMode,
36    websocket::{
37        SubscriptionState, TransportBackend, WebSocketClient, WebSocketConfig,
38        channel_message_handler,
39    },
40};
41
42use crate::{
43    common::{
44        consts::{HEARTBEAT_INTERVAL, RECONNECT_BASE_BACKOFF, RECONNECT_MAX_BACKOFF},
45        enums::{LighterCandleResolution, LighterEnvironment},
46        rate_limit::ws_message_rate_limiter,
47        symbol::MarketRegistry,
48        urls::lighter_ws_url,
49    },
50    websocket::{
51        error::LighterWsError,
52        handler::{FeedHandler, HandlerCommand},
53        messages::{LighterMarketSelection, LighterWsChannel, NautilusWsMessage},
54    },
55};
56
57const RECONNECT_TIMEOUT_MS: u64 = 15_000;
58const RECONNECT_JITTER_MS: u64 = 200;
59const RECONNECT_BACKOFF_FACTOR: f64 = 2.0;
60const DISCONNECT_TIMEOUT: Duration = Duration::from_secs(2);
61
62/// Outer Lighter WebSocket client.
63///
64/// Orchestrates the connection lifecycle and subscription bookkeeping for the
65/// Lighter streaming API. The inner feed handler runs on a dedicated tokio
66/// task and exclusively owns the underlying [`WebSocketClient`]; this outer
67/// type communicates with it through a command channel and consumes events
68/// over an unbounded mpsc.
69///
70/// Authenticated channels store their auth token in `subscription_args` and
71/// replay it verbatim on reconnect. That stored token stays valid because the
72/// execution client rotates it on a 6h cadence (inside the venue's 7h TTL) by
73/// re-issuing `subscribe_account`, so a reconnect never replays a connect-time,
74/// potentially-expired token.
75pub struct LighterWebSocketClient {
76    url: String,
77    connection_mode: Arc<ArcSwap<AtomicU8>>,
78    signal: Arc<AtomicBool>,
79    cmd_tx: Arc<tokio::sync::RwLock<tokio::sync::mpsc::UnboundedSender<HandlerCommand>>>,
80    out_rx: Option<tokio::sync::mpsc::UnboundedReceiver<NautilusWsMessage>>,
81    subscriptions: SubscriptionState,
82    subscription_args: Arc<DashMap<String, (LighterWsChannel, Option<String>)>>,
83    instruments: Arc<DashMap<i16, InstrumentAny>>,
84    registry: Arc<MarketRegistry>,
85    task_handle: Option<tokio::task::JoinHandle<()>>,
86    transport_backend: TransportBackend,
87    proxy_url: Option<String>,
88}
89
90impl Debug for LighterWebSocketClient {
91    /// Custom `Debug` that redacts the auth token in `subscription_args`.
92    ///
93    /// Authenticated channel subscriptions store the venue bearer token
94    /// alongside the channel for reconnect replay; deriving `Debug` would
95    /// otherwise print the token verbatim.
96    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
97        let subscription_topics: Vec<String> = self
98            .subscription_args
99            .iter()
100            .map(|entry| {
101                let (channel, auth) = entry.value();
102                format!(
103                    "topic={} channel={:?} authed={}",
104                    entry.key(),
105                    channel,
106                    auth.is_some(),
107                )
108            })
109            .collect();
110
111        f.debug_struct(stringify!(LighterWebSocketClient))
112            .field("url", &self.url)
113            .field("is_active", &self.is_active())
114            .field("subscription_count", &self.subscriptions.len())
115            .field("subscription_args", &subscription_topics)
116            .field("instruments_len", &self.instruments.len())
117            .field("transport_backend", &self.transport_backend)
118            .field("proxy_url", &self.proxy_url)
119            .finish_non_exhaustive()
120    }
121}
122
123impl Clone for LighterWebSocketClient {
124    fn clone(&self) -> Self {
125        Self {
126            url: self.url.clone(),
127            connection_mode: Arc::clone(&self.connection_mode),
128            signal: Arc::clone(&self.signal),
129            cmd_tx: Arc::clone(&self.cmd_tx),
130            out_rx: None,
131            subscriptions: self.subscriptions.clone(),
132            subscription_args: Arc::clone(&self.subscription_args),
133            instruments: Arc::clone(&self.instruments),
134            registry: Arc::clone(&self.registry),
135            task_handle: None,
136            transport_backend: self.transport_backend,
137            proxy_url: self.proxy_url.clone(),
138        }
139    }
140}
141
142impl LighterWebSocketClient {
143    /// Creates a new client without connecting.
144    ///
145    /// `url` overrides the resolved environment URL when supplied.
146    #[must_use]
147    pub fn new(
148        url: Option<String>,
149        environment: LighterEnvironment,
150        registry: Arc<MarketRegistry>,
151        transport_backend: TransportBackend,
152        proxy_url: Option<String>,
153    ) -> Self {
154        let url = url.unwrap_or_else(|| lighter_ws_url(environment).to_string());
155        let connection_mode = Arc::new(ArcSwap::new(Arc::new(AtomicU8::new(
156            ConnectionMode::Closed as u8,
157        ))));
158
159        let (placeholder_tx, _) = tokio::sync::mpsc::unbounded_channel();
160
161        Self {
162            url,
163            connection_mode,
164            signal: Arc::new(AtomicBool::new(false)),
165            cmd_tx: Arc::new(tokio::sync::RwLock::new(placeholder_tx)),
166            out_rx: None,
167            subscriptions: SubscriptionState::new(':'),
168            subscription_args: Arc::new(DashMap::new()),
169            instruments: Arc::new(DashMap::new()),
170            registry,
171            task_handle: None,
172            transport_backend,
173            proxy_url,
174        }
175    }
176
177    /// Returns the resolved WebSocket URL.
178    #[must_use]
179    pub fn url(&self) -> &str {
180        &self.url
181    }
182
183    /// Returns `true` when the underlying connection is active.
184    #[must_use]
185    pub fn is_active(&self) -> bool {
186        self.connection_mode.load().load(Ordering::Relaxed) == ConnectionMode::Active as u8
187    }
188
189    /// Waits until the underlying connection reports active, or returns an
190    /// error after `timeout_secs`.
191    ///
192    /// Polls [`Self::is_active`] every 10ms. Mirrors the documented
193    /// `wait_until_active` contract for adapter WebSocket clients in
194    /// `docs/developer_guide/adapters.md`.
195    ///
196    /// # Errors
197    ///
198    /// Returns [`LighterWsError::Client`] if the connection does not reach
199    /// the active state within `timeout_secs`.
200    pub async fn wait_until_active(&self, timeout_secs: f64) -> Result<(), LighterWsError> {
201        let timeout = Duration::from_secs_f64(timeout_secs);
202
203        tokio::time::timeout(timeout, async {
204            while !self.is_active() {
205                tokio::time::sleep(Duration::from_millis(10)).await;
206            }
207        })
208        .await
209        .map_err(|_| {
210            LighterWsError::Client(format!(
211                "WebSocket connection timeout after {timeout_secs} seconds"
212            ))
213        })
214    }
215
216    /// Returns the count of confirmed subscriptions.
217    #[must_use]
218    pub fn subscription_count(&self) -> usize {
219        self.subscriptions.len()
220    }
221
222    /// Returns a clone of the shared instrument cache.
223    #[must_use]
224    pub fn instruments_cache(&self) -> Arc<DashMap<i16, InstrumentAny>> {
225        Arc::clone(&self.instruments)
226    }
227
228    /// Caches a batch of instruments along with their venue `market_index`,
229    /// replaying them to the handler if a connection is already established.
230    pub fn cache_instruments(&self, instruments: Vec<(i16, InstrumentAny)>) {
231        self.instruments.clear();
232        for (market_index, instrument) in &instruments {
233            self.instruments.insert(*market_index, instrument.clone());
234        }
235        log::debug!(
236            "Lighter instrument cache initialized with {} instruments",
237            instruments.len()
238        );
239
240        if let Ok(cmd_tx) = self.cmd_tx.try_read() {
241            let _ = cmd_tx.send(HandlerCommand::InitializeInstruments(instruments));
242        }
243    }
244
245    /// Caches a single instrument and pushes it to the handler if connected.
246    pub fn cache_instrument(&self, market_index: i16, instrument: InstrumentAny) {
247        self.instruments.insert(market_index, instrument.clone());
248
249        if let Ok(cmd_tx) = self.cmd_tx.try_read() {
250            let _ = cmd_tx.send(HandlerCommand::UpdateInstrument {
251                market_index,
252                instrument,
253            });
254        }
255    }
256
257    /// Establishes the WebSocket connection and spawns the feed-handler task.
258    ///
259    /// # Errors
260    ///
261    /// Returns an error if the underlying [`WebSocketClient::connect`] fails
262    /// or the handler cannot be initialized.
263    pub async fn connect(&mut self) -> anyhow::Result<()> {
264        if self.is_active() {
265            log::warn!("Lighter WebSocket already connected");
266            return Ok(());
267        }
268
269        let (message_handler, raw_rx) = channel_message_handler();
270        let cfg = WebSocketConfig {
271            url: self.url.clone(),
272            headers: vec![],
273            heartbeat: Some(HEARTBEAT_INTERVAL.as_secs()),
274            heartbeat_msg: None,
275            reconnect_timeout_ms: Some(RECONNECT_TIMEOUT_MS),
276            reconnect_delay_initial_ms: Some(RECONNECT_BASE_BACKOFF.as_millis() as u64),
277            reconnect_delay_max_ms: Some(RECONNECT_MAX_BACKOFF.as_millis() as u64),
278            reconnect_backoff_factor: Some(RECONNECT_BACKOFF_FACTOR),
279            reconnect_jitter_ms: Some(RECONNECT_JITTER_MS),
280            reconnect_max_attempts: None,
281            idle_timeout_ms: None,
282            backend: self.transport_backend,
283            proxy_url: self.proxy_url.clone(),
284        };
285        let client = WebSocketClient::connect_with_rate_limiter(
286            cfg,
287            Some(message_handler),
288            None,
289            None,
290            ws_message_rate_limiter(&self.url),
291        )
292        .await?;
293
294        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel::<HandlerCommand>();
295        let (out_tx, out_rx) = tokio::sync::mpsc::unbounded_channel::<NautilusWsMessage>();
296
297        // Capture the connection-mode atomic before moving `client` into the
298        // SetClient command below.
299        let connection_mode_atomic = client.connection_mode_atomic();
300
301        // Queue SetClient (and the instrument cache replay) onto the new
302        // command channel BEFORE publishing it to clones or marking the
303        // connection active. Otherwise a clone observing `is_active()` could
304        // race in and send a Subscribe before SetClient lands, and the
305        // handler would drop the subscription because `inner == None`.
306        if let Err(e) = cmd_tx.send(HandlerCommand::SetClient(client)) {
307            anyhow::bail!("Failed to send SetClient command: {e}");
308        }
309
310        let initial_instruments: Vec<(i16, InstrumentAny)> = self
311            .instruments
312            .iter()
313            .map(|entry| (*entry.key(), entry.value().clone()))
314            .collect();
315
316        if !initial_instruments.is_empty()
317            && let Err(e) = cmd_tx.send(HandlerCommand::InitializeInstruments(initial_instruments))
318        {
319            log::error!("Failed to send InitializeInstruments: {e}");
320        }
321
322        // Publish the new command channel and connection-mode atomic last.
323        // Any clone-driven subscribe call queued from this point lands
324        // behind SetClient and InitializeInstruments in cmd_rx.
325        *self.cmd_tx.write().await = cmd_tx.clone();
326        self.out_rx = Some(out_rx);
327        self.connection_mode.store(connection_mode_atomic);
328
329        log::debug!("Lighter WebSocket connected: {}", self.url);
330
331        let signal = Arc::clone(&self.signal);
332        let subscriptions = self.subscriptions.clone();
333        let subscription_args = Arc::clone(&self.subscription_args);
334        let cmd_tx_for_reconnect = cmd_tx.clone();
335
336        let task = get_runtime().spawn(async move {
337            let mut handler =
338                FeedHandler::new(Arc::clone(&signal), cmd_rx, raw_rx, out_tx, subscriptions);
339            handler.set_command_sender(cmd_tx_for_reconnect.clone());
340
341            let restore_subscriptions = || {
342                if subscription_args.is_empty() {
343                    log::debug!("No active Lighter subscriptions to restore after reconnect");
344                    return;
345                }
346                log::debug!(
347                    "Restoring {} Lighter subscriptions after reconnect",
348                    subscription_args.len(),
349                );
350
351                // Stored token stays inside the TTL: the execution client rotates it every 6h
352                for entry in subscription_args.iter() {
353                    let (channel, auth) = entry.value().clone();
354                    if let Err(e) =
355                        cmd_tx_for_reconnect.send(HandlerCommand::Subscribe { channel, auth })
356                    {
357                        log::error!("Failed to resend Lighter subscribe command: {e}");
358                    }
359                }
360            };
361
362            loop {
363                match handler.next().await {
364                    Some(NautilusWsMessage::Reconnected) => {
365                        log::debug!("Lighter WebSocket reconnected");
366                        restore_subscriptions();
367
368                        if handler.send(NautilusWsMessage::Reconnected).is_err() {
369                            if handler.is_stopped() {
370                                log::debug!("Failed to forward Reconnected (receiver dropped)");
371                            } else {
372                                log::error!("Failed to forward Reconnected (receiver dropped)");
373                            }
374                            break;
375                        }
376                    }
377                    Some(msg) => {
378                        if handler.send(msg).is_err() {
379                            if handler.is_stopped() {
380                                log::debug!("Failed to send Lighter message (receiver dropped)");
381                            } else {
382                                log::error!("Failed to send Lighter message (receiver dropped)");
383                            }
384                            break;
385                        }
386                    }
387                    None => {
388                        if handler.is_stopped() {
389                            log::debug!("Lighter handler stop signal observed, exiting loop");
390                            break;
391                        }
392                        log::warn!("Lighter WebSocket stream ended unexpectedly");
393                        break;
394                    }
395                }
396            }
397            log::debug!("Lighter handler task completed");
398        });
399        self.task_handle = Some(task);
400        Ok(())
401    }
402
403    /// Disconnects gracefully: signals shutdown, drains the handler, then
404    /// awaits the task handle with a timeout.
405    ///
406    /// # Errors
407    ///
408    /// This function currently completes best-effort shutdown and returns `Ok(())`.
409    pub async fn disconnect(&mut self) -> Result<(), LighterWsError> {
410        log::debug!("Disconnecting Lighter WebSocket");
411
412        if let Err(e) = self.cmd_tx.read().await.send(HandlerCommand::Disconnect) {
413            log::debug!("Failed to send Lighter disconnect command: {e}");
414        }
415        self.signal.store(true, Ordering::Release);
416
417        if let Some(handle) = self.task_handle.take() {
418            let abort_handle = handle.abort_handle();
419            tokio::select! {
420                result = handle => match result {
421                    Ok(()) => log::debug!("Lighter handler task completed"),
422                    Err(e) if e.is_cancelled() => log::debug!("Lighter handler task cancelled"),
423                    Err(e) => log::error!("Lighter handler task error: {e:?}"),
424                },
425                () = tokio::time::sleep(DISCONNECT_TIMEOUT) => {
426                    log::warn!("Timeout waiting for Lighter handler task, aborting");
427                    abort_handle.abort();
428                }
429            }
430        }
431
432        self.connection_mode
433            .store(Arc::new(AtomicU8::new(ConnectionMode::Closed as u8)));
434        Ok(())
435    }
436
437    /// Receives the next message from the handler, or `None` if the receiver
438    /// has been taken or the handler has shut down.
439    pub async fn next_event(&mut self) -> Option<NautilusWsMessage> {
440        if let Some(rx) = self.out_rx.as_mut() {
441            rx.recv().await
442        } else {
443            None
444        }
445    }
446
447    /// Takes the feed-handler task handle, leaving `None` behind.
448    ///
449    /// Used by callers that connect on a cloned client and want to await the
450    /// inner handler task on a different instance during disconnect.
451    #[must_use]
452    pub fn take_task_handle(&mut self) -> Option<tokio::task::JoinHandle<()>> {
453        self.task_handle.take()
454    }
455
456    /// Installs a feed-handler task handle previously obtained from
457    /// [`Self::take_task_handle`].
458    pub fn set_task_handle(&mut self, handle: tokio::task::JoinHandle<()>) {
459        self.task_handle = Some(handle);
460    }
461
462    /// Subscribe to L2 order-book updates for an instrument.
463    ///
464    /// # Errors
465    ///
466    /// Returns an error if the instrument is not registered or the command
467    /// cannot be queued.
468    pub async fn subscribe_book(&self, instrument_id: InstrumentId) -> Result<(), LighterWsError> {
469        let market_index = self.market_index_for(&instrument_id)?;
470        self.send_cmd(HandlerCommand::SetBookDeltasSub {
471            market_index,
472            subscribed: true,
473        })
474        .await?;
475
476        if let Err(e) = self.subscribe_order_book_stream(market_index).await {
477            let _ = self
478                .send_cmd(HandlerCommand::SetBookDeltasSub {
479                    market_index,
480                    subscribed: false,
481                })
482                .await;
483            return Err(e);
484        }
485
486        Ok(())
487    }
488
489    /// Unsubscribe from L2 order-book updates.
490    ///
491    /// # Errors
492    ///
493    /// Returns an error if the instrument is not registered or the command
494    /// cannot be queued.
495    pub async fn unsubscribe_book(
496        &self,
497        instrument_id: InstrumentId,
498    ) -> Result<(), LighterWsError> {
499        let market_index = self.market_index_for(&instrument_id)?;
500        self.send_cmd(HandlerCommand::SetBookDeltasSub {
501            market_index,
502            subscribed: false,
503        })
504        .await?;
505        self.unsubscribe_order_book_stream(market_index).await
506    }
507
508    /// Subscribe to depth-10 snapshots derived from the same `order_book`
509    /// stream as [`Self::subscribe_book`].
510    ///
511    /// # Errors
512    ///
513    /// Returns an error if the instrument is not registered or the command
514    /// cannot be queued.
515    pub async fn subscribe_book_depth10(
516        &self,
517        instrument_id: InstrumentId,
518    ) -> Result<(), LighterWsError> {
519        let market_index = self.market_index_for(&instrument_id)?;
520        self.send_cmd(HandlerCommand::SetDepth10Sub {
521            market_index,
522            subscribed: true,
523        })
524        .await?;
525
526        if let Err(e) = self.subscribe_order_book_stream(market_index).await {
527            let _ = self
528                .send_cmd(HandlerCommand::SetDepth10Sub {
529                    market_index,
530                    subscribed: false,
531                })
532                .await;
533            return Err(e);
534        }
535
536        Ok(())
537    }
538
539    /// Unsubscribe from depth-10 snapshots.
540    ///
541    /// Clears the depth-10 emission flag without tearing down the underlying
542    /// `order_book` stream so any active deltas subscriber keeps receiving
543    /// updates.
544    ///
545    /// # Errors
546    ///
547    /// Returns an error if the instrument is not registered or the command
548    /// cannot be queued.
549    pub async fn unsubscribe_book_depth10(
550        &self,
551        instrument_id: InstrumentId,
552    ) -> Result<(), LighterWsError> {
553        let market_index = self.market_index_for(&instrument_id)?;
554        self.send_cmd(HandlerCommand::SetDepth10Sub {
555            market_index,
556            subscribed: false,
557        })
558        .await?;
559        self.unsubscribe_order_book_stream(market_index).await
560    }
561
562    /// Subscribe to ticker (best bid/offer) updates.
563    ///
564    /// # Errors
565    ///
566    /// Returns an error if the instrument is not registered or the command
567    /// cannot be queued.
568    pub async fn subscribe_quotes(
569        &self,
570        instrument_id: InstrumentId,
571    ) -> Result<(), LighterWsError> {
572        let market_index = self.market_index_for(&instrument_id)?;
573        self.send_subscribe(LighterWsChannel::Ticker(market_index), None)
574            .await
575    }
576
577    /// Unsubscribe from ticker updates.
578    ///
579    /// # Errors
580    ///
581    /// Returns an error if the instrument is not registered or the command
582    /// cannot be queued.
583    pub async fn unsubscribe_quotes(
584        &self,
585        instrument_id: InstrumentId,
586    ) -> Result<(), LighterWsError> {
587        let market_index = self.market_index_for(&instrument_id)?;
588        self.send_unsubscribe(LighterWsChannel::Ticker(market_index))
589            .await
590    }
591
592    /// Subscribe to trade updates.
593    ///
594    /// # Errors
595    ///
596    /// Returns an error if the instrument is not registered or the command
597    /// cannot be queued.
598    pub async fn subscribe_trades(
599        &self,
600        instrument_id: InstrumentId,
601    ) -> Result<(), LighterWsError> {
602        let market_index = self.market_index_for(&instrument_id)?;
603        self.send_subscribe(LighterWsChannel::Trade(market_index), None)
604            .await
605    }
606
607    /// Unsubscribe from trade updates.
608    ///
609    /// # Errors
610    ///
611    /// Returns an error if the instrument is not registered or the command
612    /// cannot be queued.
613    pub async fn unsubscribe_trades(
614        &self,
615        instrument_id: InstrumentId,
616    ) -> Result<(), LighterWsError> {
617        let market_index = self.market_index_for(&instrument_id)?;
618        self.send_unsubscribe(LighterWsChannel::Trade(market_index))
619            .await
620    }
621
622    /// Subscribe to the `candle/{market_id}/{resolution}` stream for an
623    /// instrument and resolution.
624    ///
625    /// # Errors
626    ///
627    /// Returns an error if the instrument is not registered, the resolution
628    /// is not offered on the WebSocket stream, or the command cannot be
629    /// queued.
630    pub async fn subscribe_candles(
631        &self,
632        instrument_id: InstrumentId,
633        resolution: LighterCandleResolution,
634    ) -> Result<(), LighterWsError> {
635        if !resolution.is_ws_streamable() {
636            return Err(LighterWsError::Client(format!(
637                "resolution {resolution:?} is not offered on the Lighter candle WebSocket stream",
638            )));
639        }
640        let market_index = self.market_index_for(&instrument_id)?;
641        self.send_subscribe(
642            LighterWsChannel::Candle {
643                market_index,
644                resolution,
645            },
646            None,
647        )
648        .await
649    }
650
651    /// Unsubscribe from a candle stream.
652    ///
653    /// # Errors
654    ///
655    /// Returns an error if the instrument is not registered or the command
656    /// cannot be queued.
657    pub async fn unsubscribe_candles(
658        &self,
659        instrument_id: InstrumentId,
660        resolution: LighterCandleResolution,
661    ) -> Result<(), LighterWsError> {
662        let market_index = self.market_index_for(&instrument_id)?;
663        self.send_unsubscribe(LighterWsChannel::Candle {
664            market_index,
665            resolution,
666        })
667        .await
668    }
669
670    /// Subscribe to a market-stats stream covering all markets or a single
671    /// market index.
672    ///
673    /// # Errors
674    ///
675    /// Returns an error if the command cannot be queued.
676    pub async fn subscribe_market_stats(
677        &self,
678        selection: LighterMarketSelection,
679    ) -> Result<(), LighterWsError> {
680        self.send_subscribe(LighterWsChannel::MarketStats(selection), None)
681            .await
682    }
683
684    /// Unsubscribe from a market-stats stream.
685    ///
686    /// # Errors
687    ///
688    /// Returns an error if the command cannot be queued.
689    pub async fn unsubscribe_market_stats(
690        &self,
691        selection: LighterMarketSelection,
692    ) -> Result<(), LighterWsError> {
693        self.send_unsubscribe(LighterWsChannel::MarketStats(selection))
694            .await
695    }
696
697    /// Subscribe to a spot market-stats stream covering all spot markets or a
698    /// single spot market index.
699    ///
700    /// # Errors
701    ///
702    /// Returns an error if the command cannot be queued.
703    pub async fn subscribe_spot_market_stats(
704        &self,
705        selection: LighterMarketSelection,
706    ) -> Result<(), LighterWsError> {
707        self.send_subscribe(LighterWsChannel::SpotMarketStats(selection), None)
708            .await
709    }
710
711    /// Unsubscribe from a spot market-stats stream.
712    ///
713    /// # Errors
714    ///
715    /// Returns an error if the command cannot be queued.
716    pub async fn unsubscribe_spot_market_stats(
717        &self,
718        selection: LighterMarketSelection,
719    ) -> Result<(), LighterWsError> {
720        self.send_unsubscribe(LighterWsChannel::SpotMarketStats(selection))
721            .await
722    }
723
724    /// Subscribe to the chain-height stream.
725    ///
726    /// # Errors
727    ///
728    /// Returns an error if the command cannot be queued.
729    pub async fn subscribe_height(&self) -> Result<(), LighterWsError> {
730        self.send_subscribe(LighterWsChannel::Height, None).await
731    }
732
733    /// Unsubscribe from the chain-height stream.
734    ///
735    /// # Errors
736    ///
737    /// Returns an error if the command cannot be queued.
738    pub async fn unsubscribe_height(&self) -> Result<(), LighterWsError> {
739        self.send_unsubscribe(LighterWsChannel::Height).await
740    }
741
742    /// Provides the execution context the feed handler stamps onto reports
743    /// parsed from `account_*` frames.
744    ///
745    /// Without this context account frames fall back to
746    /// [`NautilusWsMessage::Raw`]; once it is set the handler emits typed
747    /// [`crate::websocket::messages::ExecutionReport`] and
748    /// [`crate::websocket::messages::NautilusWsMessage::AccountState`]
749    /// messages stamped with `account_id`. The `account_index` is used by
750    /// the fill parser to determine which side of each account-trade frame
751    /// the configured account took.
752    ///
753    /// # Errors
754    ///
755    /// Returns an error if the command cannot be queued.
756    pub async fn set_execution_context(
757        &self,
758        account_id: AccountId,
759        account_index: i64,
760    ) -> Result<(), LighterWsError> {
761        self.send_cmd(HandlerCommand::SetExecutionContext {
762            account_id,
763            account_index,
764        })
765        .await
766    }
767
768    /// Subscribe to a private account channel using a venue auth token.
769    ///
770    /// The auth token must be a valid Lighter L2 auth signature; see the
771    /// `signing` module for token construction. Re-issuing this with a fresh
772    /// token (as the execution client's auth-token rotation does) overwrites
773    /// the stored reconnect-replay token, keeping it within the venue TTL.
774    ///
775    /// # Errors
776    ///
777    /// Returns an error if the command cannot be queued.
778    pub async fn subscribe_account(
779        &self,
780        channel: LighterWsChannel,
781        auth_token: String,
782    ) -> Result<(), LighterWsError> {
783        self.send_subscribe(channel, Some(auth_token)).await
784    }
785
786    /// Unsubscribe from a private account channel.
787    ///
788    /// # Errors
789    ///
790    /// Returns an error if the command cannot be queued.
791    pub async fn unsubscribe_account(
792        &self,
793        channel: LighterWsChannel,
794    ) -> Result<(), LighterWsError> {
795        self.send_unsubscribe(channel).await
796    }
797
798    /// Dispatch a signed L2 transaction over the WebSocket.
799    ///
800    /// `tx_type` is the venue's [`crate::common::enums::LighterTxType`]
801    /// discriminant; `tx_info` is the JSON body produced by the matching
802    /// [`crate::signing::tx::TxInfoJson`] renderer. The venue confirms
803    /// acceptance via the `account_*` streams.
804    ///
805    /// # Errors
806    ///
807    /// Returns an error if the command cannot be queued.
808    pub async fn send_tx(
809        &self,
810        tx_type: u8,
811        tx_info: Box<serde_json::value::RawValue>,
812    ) -> Result<(), LighterWsError> {
813        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
814        self.send_cmd(HandlerCommand::SendTx {
815            tx_type,
816            tx_info,
817            response_tx,
818        })
819        .await?;
820
821        response_rx
822            .await
823            .map_err(|e| LighterWsError::Client(format!("handler dropped sendTx result: {e}")))?
824    }
825
826    async fn send_subscribe(
827        &self,
828        channel: LighterWsChannel,
829        auth: Option<String>,
830    ) -> Result<(), LighterWsError> {
831        let topic = channel.topic_key();
832        let previous = self
833            .subscription_args
834            .insert(topic.clone(), (channel.clone(), auth.clone()));
835        if let Err(e) = self
836            .send_cmd(HandlerCommand::Subscribe { channel, auth })
837            .await
838        {
839            if let Some(previous) = previous {
840                self.subscription_args.insert(topic, previous);
841            } else {
842                self.subscription_args.remove(&topic);
843            }
844            return Err(e);
845        }
846
847        Ok(())
848    }
849
850    async fn send_unsubscribe(&self, channel: LighterWsChannel) -> Result<(), LighterWsError> {
851        let topic = channel.topic_key();
852        self.send_cmd(HandlerCommand::Unsubscribe { channel })
853            .await?;
854        self.subscription_args.remove(&topic);
855        Ok(())
856    }
857
858    async fn subscribe_order_book_stream(&self, market_index: i16) -> Result<(), LighterWsError> {
859        let channel = LighterWsChannel::OrderBook(market_index);
860        let topic = channel.topic_key();
861
862        if !self.subscriptions.add_reference(topic.as_str()) {
863            return Ok(());
864        }
865
866        if let Err(e) = self.send_subscribe(channel, None).await {
867            self.subscriptions.remove_reference(topic.as_str());
868            return Err(e);
869        }
870
871        Ok(())
872    }
873
874    async fn unsubscribe_order_book_stream(&self, market_index: i16) -> Result<(), LighterWsError> {
875        let channel = LighterWsChannel::OrderBook(market_index);
876        let topic = channel.topic_key();
877
878        if !self.subscriptions.remove_reference(topic.as_str()) {
879            return Ok(());
880        }
881
882        if let Err(e) = self.send_unsubscribe(channel).await {
883            self.subscriptions.add_reference(topic.as_str());
884            return Err(e);
885        }
886
887        Ok(())
888    }
889
890    async fn send_cmd(&self, cmd: HandlerCommand) -> Result<(), LighterWsError> {
891        self.cmd_tx
892            .read()
893            .await
894            .send(cmd)
895            .map_err(|e| LighterWsError::Client(format!("handler unavailable: {e}")))
896    }
897
898    fn market_index_for(&self, instrument_id: &InstrumentId) -> Result<i16, LighterWsError> {
899        self.registry.market_index(instrument_id).ok_or_else(|| {
900            LighterWsError::Client(format!(
901                "no Lighter market_index registered for instrument: {instrument_id}"
902            ))
903        })
904    }
905}
906
907#[cfg(test)]
908mod tests {
909    use nautilus_core::UnixNanos;
910    use nautilus_model::{
911        identifiers::Symbol,
912        instruments::CryptoPerpetual,
913        types::{Currency, Price, Quantity},
914    };
915    use rstest::rstest;
916
917    use super::*;
918    use crate::common::{consts::LIGHTER_VENUE, enums::LighterProductType};
919
920    fn registry_with(
921        market_index: i16,
922        symbol: &str,
923        product: LighterProductType,
924    ) -> Arc<MarketRegistry> {
925        let registry = Arc::new(MarketRegistry::new());
926        registry.insert(market_index, symbol, product);
927        registry
928    }
929
930    #[rstest]
931    fn market_index_for_returns_registered_index() {
932        let registry = registry_with(7, "ETH", LighterProductType::Perp);
933        let client = LighterWebSocketClient::new(
934            Some("wss://example/test".to_string()),
935            LighterEnvironment::Testnet,
936            Arc::clone(&registry),
937            TransportBackend::default(),
938            None,
939        );
940        let id = registry.instrument_id(7).expect("registered");
941        assert_eq!(client.market_index_for(&id).unwrap(), 7);
942    }
943
944    #[rstest]
945    fn market_index_for_unregistered_returns_error() {
946        let registry = Arc::new(MarketRegistry::new());
947        let client = LighterWebSocketClient::new(
948            Some("wss://example/test".to_string()),
949            LighterEnvironment::Testnet,
950            registry,
951            TransportBackend::default(),
952            None,
953        );
954        let id = InstrumentId::new(Symbol::from_str_unchecked("UNKNOWN-PERP"), *LIGHTER_VENUE);
955        assert!(client.market_index_for(&id).is_err());
956    }
957
958    #[rstest]
959    fn cache_instrument_populates_lookup() {
960        let registry = registry_with(0, "ETH", LighterProductType::Perp);
961        let client = LighterWebSocketClient::new(
962            Some("wss://example/test".to_string()),
963            LighterEnvironment::Testnet,
964            Arc::clone(&registry),
965            TransportBackend::default(),
966            None,
967        );
968        let id = registry.instrument_id(0).expect("registered");
969        let instrument = stub_instrument(id);
970        client.cache_instrument(0, instrument);
971        assert!(client.instruments_cache().contains_key(&0));
972    }
973
974    fn stub_instrument(id: InstrumentId) -> InstrumentAny {
975        InstrumentAny::CryptoPerpetual(CryptoPerpetual::new(
976            id,
977            id.symbol,
978            Currency::from("ETH"),
979            Currency::from("USDC"),
980            Currency::from("USDC"),
981            false,
982            2,
983            4,
984            Price::from("0.01"),
985            Quantity::from("0.0001"),
986            None,
987            None,
988            None,
989            None,
990            None,
991            None,
992            None,
993            None,
994            None,
995            None,
996            None,
997            None,
998            None,
999            None,
1000            UnixNanos::default(),
1001            UnixNanos::default(),
1002        ))
1003    }
1004}
nautilus_lighter/websocket/client.rs

nautilus_lighter/websocket/
client.rs
