nautilus_derive/
data.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//! Live market data client implementation for the Derive adapter.
17
18use std::{
19    borrow::Cow,
20    num::NonZeroUsize,
21    str::FromStr,
22    sync::{
23        Arc, Weak,
24        atomic::{AtomicBool, Ordering},
25    },
26    time::Duration,
27};
28
29use ahash::{AHashMap, AHashSet};
30use anyhow::Context;
31use async_trait::async_trait;
32use dashmap::DashMap;
33use nautilus_common::{
34    cache::{InstrumentLookupError, fifo::FifoCache, quote::QuoteCache},
35    clients::DataClient,
36    live::{
37        dst::time::{self, Instant},
38        runner::get_data_event_sender,
39        sender::EventSender,
40    },
41    messages::{
42        DataEvent,
43        data::{
44            BarsResponse, DataResponse, FundingRatesResponse, InstrumentResponse,
45            InstrumentsResponse, OptionChainReferencePriceResponse, QuotesResponse, RequestBars,
46            RequestFundingRates, RequestInstrument, RequestInstruments,
47            RequestOptionChainReferencePrice, RequestQuotes, RequestTrades, SubscribeBookDeltas,
48            SubscribeBookDepth, SubscribeFundingRates, SubscribeIndexPrices, SubscribeMarkPrices,
49            SubscribeOptionGreeks, SubscribeQuotes, SubscribeTrades, TradesResponse,
50            UnsubscribeBookDeltas, UnsubscribeBookDepth, UnsubscribeFundingRates,
51            UnsubscribeIndexPrices, UnsubscribeMarkPrices, UnsubscribeOptionGreeks,
52            UnsubscribeQuotes, UnsubscribeTrades,
53        },
54    },
55    providers::InstrumentProvider,
56};
57use nautilus_core::{
58    AtomicMap, AtomicSet, Params, UnixNanos,
59    datetime::{NANOSECONDS_IN_SECOND, datetime_to_unix_nanos},
60    time::{AtomicTime, get_atomic_clock_realtime},
61};
62use nautilus_live::{
63    SocketControl,
64    task::{TaskGroup, TaskGroupGuard},
65};
66use nautilus_model::{
67    data::{Bar, Data, QuoteTick},
68    enums::{AggregationSource, BookType, PriceType},
69    identifiers::{ClientId, InstrumentId, TradeId, Venue},
70    instruments::{Instrument, InstrumentAny},
71    types::Price,
72};
73use parking_lot::Mutex;
74use rust_decimal::Decimal;
75use tokio_util::sync::CancellationToken;
76
77use crate::{
78    book::{recovery::spawn_recovery, sync::BookSyncTracker},
79    common::{
80        consts::{
81            DERIVE_CANDLES_DEFAULT_LIMIT, DERIVE_CANDLES_MAX_PAGES, DERIVE_TRADES_PAGE_SIZE,
82            DERIVE_VENUE,
83        },
84        enums::{
85            DeriveInstrumentType, DeriveOrderbookDepth, DeriveOrderbookGroup, DeriveTickerInterval,
86        },
87        parse::{format_instrument_id, format_venue_symbol, parse_derive_instrument_any},
88    },
89    config::DeriveDataClientConfig,
90    http::{DeriveHttpClient, query::PaginationCursor},
91    providers::{
92        DeriveInstrumentProvider, fetch_instrument_definitions, parse_instrument_definitions,
93    },
94    websocket::{
95        DEFAULT_ORDERBOOK_DEPTH, DEFAULT_ORDERBOOK_GROUP, DEFAULT_TICKER_INTERVAL,
96        DeriveOrderbookMsg, DerivePublicWsData, DeriveTickerMsg, DeriveTradesMsg,
97        DeriveWebSocketClient, DeriveWebSocketSubscriptionHandle, DeriveWsError, DeriveWsMessage,
98        WsMessageContext, bar_spec_to_derive_period, dispatch::TRADE_DEDUP_CAPACITY,
99        orderbook_channel, parse_candle_record, parse_funding_rate,
100        parse_funding_rate_history_record, parse_index_price, parse_mark_price,
101        parse_option_greeks, parse_orderbook_deltas, parse_orderbook_depth, parse_public_ws_data,
102        parse_ticker_quote, parse_ticker_quote_from_rest, parse_trade_tick,
103        parse_trade_tick_from_rest, ticker_channel, trades_channel,
104    },
105};
106
107/// Derive live data client.
108#[derive(Debug)]
109pub struct DeriveDataClient {
110    client_id: ClientId,
111    config: DeriveDataClientConfig,
112    http_client: DeriveHttpClient,
113    provider: DeriveInstrumentProvider,
114    ws_client: DeriveWebSocketClient,
115    is_connected: Arc<AtomicBool>,
116    cancellation_token: CancellationToken,
117    session_tasks: TaskGroup,
118    pending_tasks: TaskGroup,
119    shutdown_errors: Vec<String>,
120    data_sender: EventSender<DataEvent>,
121    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
122    active_book_delta_channels: Arc<AtomicMap<InstrumentId, String>>,
123    active_book_depth_channels: Arc<AtomicMap<InstrumentId, String>>,
124    active_ticker_channels: Arc<AtomicMap<InstrumentId, String>>,
125    active_quote_subs: Arc<AtomicSet<InstrumentId>>,
126    active_trade_subs: Arc<AtomicSet<InstrumentId>>,
127    active_mark_subs: Arc<AtomicSet<InstrumentId>>,
128    active_index_subs: Arc<AtomicSet<InstrumentId>>,
129    active_funding_subs: Arc<AtomicSet<InstrumentId>>,
130    active_greeks_subs: Arc<AtomicSet<InstrumentId>>,
131    channel_subscriptions: Arc<ChannelSubscriptionRegistry>,
132    subscription_lock: Arc<Mutex<()>>,
133    book_sync: BookSyncTracker,
134    quote_cache: Arc<Mutex<QuoteCache>>,
135    trades_emitted: Arc<Mutex<FifoCache<TradeId, TRADE_DEDUP_CAPACITY>>>,
136    clock: &'static AtomicTime,
137}
138
139impl DeriveDataClient {
140    /// Creates a new [`DeriveDataClient`] instance.
141    ///
142    /// # Errors
143    ///
144    /// Returns an error if the HTTP client cannot be initialized.
145    pub fn new(client_id: ClientId, config: DeriveDataClientConfig) -> anyhow::Result<Self> {
146        let clock = get_atomic_clock_realtime();
147        let data_sender = get_data_event_sender();
148        let proxy_url = config
149            .proxy_url
150            .as_ref()
151            .map(|value| value.expose_secret().to_owned());
152
153        let http_client = DeriveHttpClient::new(
154            config.rest_url(),
155            Some(config.http_timeout_secs),
156            proxy_url.clone(),
157            None,
158        )?;
159        let provider = DeriveInstrumentProvider::with_expired(
160            http_client.clone(),
161            config.currencies.clone(),
162            config.include_expired,
163        );
164        let book_sync = BookSyncTracker::default();
165        let mut ws_client = DeriveWebSocketClient::new(
166            Some(config.ws_url()),
167            config.environment,
168            config.transport_backend,
169            proxy_url,
170        )
171        .with_book_sync(
172            book_sync.clone(),
173            Duration::from_secs(config.book_snapshot_timeout_secs),
174        )
175        .with_socket_control(SocketControl::new(
176            client_id,
177            Some(*DERIVE_VENUE),
178            "derive-data-streams",
179        ));
180
181        if let Some(secs) = config.ws_timeout_secs {
182            ws_client.set_request_timeout(Duration::from_secs(secs));
183        }
184
185        let session_tasks = TaskGroup::new();
186        let pending_tasks = TaskGroup::new();
187
188        Ok(Self {
189            client_id,
190            config,
191            http_client,
192            provider,
193            ws_client,
194            is_connected: Arc::new(AtomicBool::new(false)),
195            cancellation_token: CancellationToken::new(),
196            session_tasks,
197            pending_tasks,
198            shutdown_errors: Vec::new(),
199            data_sender,
200            instruments: Arc::new(AtomicMap::new()),
201            active_book_delta_channels: Arc::new(AtomicMap::new()),
202            active_book_depth_channels: Arc::new(AtomicMap::new()),
203            active_ticker_channels: Arc::new(AtomicMap::new()),
204            active_quote_subs: Arc::new(AtomicSet::new()),
205            active_trade_subs: Arc::new(AtomicSet::new()),
206            active_mark_subs: Arc::new(AtomicSet::new()),
207            active_index_subs: Arc::new(AtomicSet::new()),
208            active_funding_subs: Arc::new(AtomicSet::new()),
209            active_greeks_subs: Arc::new(AtomicSet::new()),
210            channel_subscriptions: Arc::new(ChannelSubscriptionRegistry::default()),
211            subscription_lock: Arc::new(Mutex::new(())),
212            book_sync,
213            quote_cache: Arc::new(Mutex::new(QuoteCache::new())),
214            trades_emitted: Arc::new(Mutex::new(FifoCache::new())),
215            clock,
216        })
217    }
218
219    /// Spawns a fire-and-forget task, tracks its handle in `pending_tasks`
220    /// for teardown abort, and logs failures with the task description.
221    fn spawn_task<F>(&self, description: impl Into<Cow<'static, str>>, fut: F)
222    where
223        F: std::future::Future<Output = anyhow::Result<()>> + Send + 'static,
224    {
225        let description = description.into();
226        let task_description = description.clone();
227
228        let future = async move {
229            if fut.await.is_err() {
230                log::warn!("{task_description} failed");
231            }
232        };
233
234        if let Err(e) = self.pending_tasks.spawn(future) {
235            log::warn!("Skipping Derive {description} after shutdown began: {e}");
236        }
237    }
238
239    fn abort_pending_tasks(&self) {
240        self.pending_tasks.begin_shutdown();
241    }
242
243    fn abort_session_tasks(&self) {
244        self.session_tasks.begin_shutdown();
245        self.ws_client.begin_shutdown();
246    }
247
248    /// Clears every local subscription map. Called from `disconnect` and
249    /// `reset` because the venue drops all subscriptions on socket close,
250    /// and aborted in-flight subscribe tasks can leak entries staged before
251    /// spawn (they never reach their on-error rollback branch).
252    fn clear_subscription_state(&self) {
253        let _guard = self.subscription_lock.lock();
254        self.channel_subscriptions.clear();
255        self.book_sync.clear();
256        self.active_book_delta_channels.store(AHashMap::new());
257        self.active_book_depth_channels.store(AHashMap::new());
258        self.active_ticker_channels.store(AHashMap::new());
259        self.active_quote_subs.store(AHashSet::new());
260        self.active_trade_subs.store(AHashSet::new());
261        self.active_mark_subs.store(AHashSet::new());
262        self.active_index_subs.store(AHashSet::new());
263        self.active_funding_subs.store(AHashSet::new());
264        self.active_greeks_subs.store(AHashSet::new());
265        self.quote_cache.lock().clear();
266        self.trades_emitted.lock().clear();
267    }
268
269    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
270        self.cancellation_token.cancel();
271        self.abort_session_tasks();
272        self.abort_pending_tasks();
273
274        if let Err(e) = self.ws_client.disconnect().await {
275            self.shutdown_errors
276                .push(format!("Derive WebSocket shutdown failed: {e}"));
277        }
278
279        let (session_result, pending_result) =
280            tokio::join!(self.join_session_tasks(), self.join_pending_tasks());
281        self.clear_subscription_state();
282        self.channel_subscriptions.clear_transitions();
283        self.is_connected.store(false, Ordering::Release);
284
285        if let Err(e) = session_result {
286            self.shutdown_errors.push(e.to_string());
287        }
288
289        if let Err(e) = pending_result {
290            self.shutdown_errors.push(e.to_string());
291        }
292
293        if !self.shutdown_errors.is_empty() {
294            anyhow::bail!(std::mem::take(&mut self.shutdown_errors).join("; "));
295        }
296
297        Ok(())
298    }
299
300    fn spawn_stream_task(
301        &self,
302        mut rx: tokio::sync::mpsc::UnboundedReceiver<DeriveWsMessage>,
303    ) -> anyhow::Result<()> {
304        let ctx = WsMessageContext {
305            clock: self.clock,
306            data_sender: self.data_sender.clone(),
307            instruments: Arc::clone(&self.instruments),
308            active_book_delta_channels: Arc::clone(&self.active_book_delta_channels),
309            active_book_depth_channels: Arc::clone(&self.active_book_depth_channels),
310            active_ticker_channels: Arc::clone(&self.active_ticker_channels),
311            active_quote_subs: Arc::clone(&self.active_quote_subs),
312            active_trade_subs: Arc::clone(&self.active_trade_subs),
313            active_mark_subs: Arc::clone(&self.active_mark_subs),
314            active_index_subs: Arc::clone(&self.active_index_subs),
315            active_funding_subs: Arc::clone(&self.active_funding_subs),
316            active_greeks_subs: Arc::clone(&self.active_greeks_subs),
317            subscription_lock: Arc::clone(&self.subscription_lock),
318            book_sync: self.book_sync.clone(),
319            quote_cache: Arc::clone(&self.quote_cache),
320            trades_emitted: Arc::clone(&self.trades_emitted),
321        };
322
323        let cancellation = self.cancellation_token.clone();
324        let is_connected = Arc::clone(&self.is_connected);
325
326        self.session_tasks.spawn(async move {
327            loop {
328                tokio::select! {
329                    biased;
330                    () = cancellation.cancelled() => {
331                        log::debug!("Derive WebSocket data stream task cancelled");
332                        break;
333                    }
334                    maybe_msg = rx.recv() => {
335                        match maybe_msg {
336                            Some(msg) => {
337                                if matches!(&msg, DeriveWsMessage::SessionRecoveryFailed(_)) {
338                                    is_connected.store(false, Ordering::Release);
339                                }
340                                Self::handle_ws_message(msg, &ctx);
341                            }
342                            None => {
343                                log::debug!("Derive WebSocket data stream ended");
344                                break;
345                            }
346                        }
347                    }
348                }
349            }
350
351            is_connected.store(false, Ordering::Release);
352
353            if !cancellation.is_cancelled() {
354                log::warn!("Derive data WebSocket stream ended unexpectedly");
355            }
356        })?;
357
358        Ok(())
359    }
360
361    fn spawn_book_monitor(&self) -> anyhow::Result<()> {
362        let tracker = self.book_sync.clone();
363        let ws = self.ws_handle();
364        let registry = Arc::clone(&self.channel_subscriptions);
365        let stale_timeout = Duration::from_secs(self.config.book_stale_threshold_secs);
366        let snapshot_timeout = Duration::from_secs(self.config.book_snapshot_timeout_secs);
367        let tasks = self.pending_tasks.spawner()?;
368        let cancellation = self.cancellation_token.clone();
369
370        self.session_tasks.spawn(async move {
371            let mut interval = time::interval(Duration::from_secs(1));
372
373            loop {
374                tokio::select! {
375                    biased;
376                    () = cancellation.cancelled() => break,
377                    _ = interval.tick() => {
378                        for (channel, signal, recovery) in tracker.recoveries(stale_timeout, Instant::now()) {
379                            signal.log();
380                            let transition = registry.transition(&channel);
381                            spawn_recovery(
382                                channel,
383                                signal.instrument_id,
384                                recovery,
385                                ws.clone(),
386                                transition,
387                                snapshot_timeout,
388                                &tasks,
389                            );
390                        }
391                    }
392                }
393            }
394        })?;
395
396        Ok(())
397    }
398
399    fn handle_ws_message(message: DeriveWsMessage, ctx: &WsMessageContext) {
400        match message {
401            DeriveWsMessage::Subscription(payload) => match parse_public_ws_data(&payload) {
402                Ok(data) => Self::handle_public_ws_data(data, ctx),
403                Err(e) => {
404                    // Include channel and a truncated payload snippet so the
405                    // wire shape that broke the parser is actionable from the
406                    // log alone.
407                    let snippet = truncated_payload_snippet(payload.data.get());
408                    log::warn!(
409                        "Failed to parse Derive public WS data on channel `{}`: {e}; payload: {snippet}",
410                        payload.channel,
411                    );
412                }
413            },
414            DeriveWsMessage::BookSubscription {
415                payload,
416                generation,
417            } => {
418                let _guard = ctx.subscription_lock.lock();
419
420                if generation.is_cancelled() {
421                    return;
422                }
423
424                match parse_public_ws_data(&payload) {
425                    Ok(DerivePublicWsData::Orderbook(msg)) => {
426                        Self::handle_public_book(&msg, ctx, Some(&generation));
427                    }
428                    Err(e) => {
429                        ctx.book_sync.reject_snapshot(
430                            payload.channel.as_str(),
431                            Arc::new(DeriveWsError::Transport(e.to_string())),
432                            Some(&generation),
433                        );
434                        log::warn!("Failed to parse Derive book snapshot: {e}");
435                    }
436                    Ok(_) => {}
437                }
438            }
439            DeriveWsMessage::Reconnected => {
440                let _guard = ctx.subscription_lock.lock();
441                ctx.quote_cache.lock().clear();
442                log::info!("Derive WebSocket reconnected");
443            }
444            DeriveWsMessage::SessionRecoveryFailed(_) => {
445                log::error!("Derive WebSocket session recovery failed");
446            }
447            DeriveWsMessage::Authenticated => log::debug!("Derive WebSocket authenticated"),
448        }
449    }
450
451    fn handle_public_ws_data(data: DerivePublicWsData, ctx: &WsMessageContext) {
452        // Lifecycle mutation takes this lock before QuoteCache, so a feed
453        // generation cannot change during cache mutation.
454        let _guard = ctx.subscription_lock.lock();
455        match data {
456            DerivePublicWsData::Orderbook(msg) => Self::handle_public_book(&msg, ctx, None),
457            DerivePublicWsData::Trades(msg) => Self::handle_public_trades(&msg, ctx),
458            DerivePublicWsData::Ticker(msg) => Self::handle_public_ticker(&msg, ctx),
459        }
460    }
461
462    fn handle_public_book(
463        msg: &DeriveOrderbookMsg,
464        ctx: &WsMessageContext,
465        generation: Option<&CancellationToken>,
466    ) {
467        let instrument_id = match msg.data.instrument_id() {
468            Ok(instrument_id) => instrument_id,
469            Err(e) => {
470                log::warn!("Invalid Derive feed instrument: {e:#}");
471                return;
472            }
473        };
474
475        let channel = msg.channel.as_str();
476        let deltas_active =
477            channel_is_active(&ctx.active_book_delta_channels, instrument_id, channel);
478        let depth_active =
479            channel_is_active(&ctx.active_book_depth_channels, instrument_id, channel);
480
481        if !deltas_active && !depth_active {
482            return;
483        }
484
485        let Some(instrument) = ctx.instruments.get_cloned(&instrument_id) else {
486            log::warn!("Orderbook message received for unknown instrument {instrument_id}");
487            return;
488        };
489
490        let ts_init = ctx.clock.get_time_ns();
491
492        let deltas = if deltas_active {
493            match parse_orderbook_deltas(
494                msg,
495                instrument.price_precision(),
496                instrument.size_precision(),
497                ts_init,
498            ) {
499                Ok(deltas) => Some(deltas),
500                Err(e) => {
501                    ctx.book_sync.reject_snapshot(
502                        channel,
503                        Arc::new(DeriveWsError::Transport(e.to_string())),
504                        generation,
505                    );
506
507                    log::warn!("Failed to parse Derive book deltas: {e}");
508                    return;
509                }
510            }
511        } else {
512            None
513        };
514
515        let depth = if depth_active {
516            match parse_orderbook_depth(
517                msg,
518                instrument.price_precision(),
519                instrument.size_precision(),
520                ts_init,
521            ) {
522                Ok(depth) => Some(depth),
523                Err(e) => {
524                    ctx.book_sync.reject_snapshot(
525                        channel,
526                        Arc::new(DeriveWsError::Transport(e.to_string())),
527                        generation,
528                    );
529
530                    log::warn!("Failed to parse Derive book depth: {e}");
531                    return;
532                }
533            }
534        } else {
535            None
536        };
537
538        if !ctx
539            .book_sync
540            .record_snapshot(channel, msg.data.timestamp, generation, Instant::now())
541        {
542            return;
543        }
544
545        if let Some(deltas) = deltas {
546            Self::send_data(ctx, Data::BookDeltas(Box::new(deltas)));
547        }
548
549        if let Some(depth) = depth {
550            Self::send_data(ctx, Data::BookDepth(Box::new(depth)));
551        }
552    }
553
554    fn handle_public_ticker(msg: &DeriveTickerMsg, ctx: &WsMessageContext) {
555        let instrument_id = match msg.data.instrument_id() {
556            Ok(instrument_id) => instrument_id,
557            Err(e) => {
558                log::warn!("Invalid Derive feed instrument: {e:#}");
559                return;
560            }
561        };
562
563        if !channel_is_active(
564            &ctx.active_ticker_channels,
565            instrument_id,
566            msg.channel.as_str(),
567        ) {
568            return;
569        }
570
571        let Some(instrument) = ctx.instruments.get_cloned(&instrument_id) else {
572            log::warn!("Ticker message received for unknown instrument {instrument_id}");
573            return;
574        };
575
576        let ts_init = ctx.clock.get_time_ns();
577        let price_precision = instrument.price_precision();
578
579        if ctx.active_quote_subs.contains(&instrument_id) {
580            let mut quote_cache = ctx.quote_cache.lock();
581
582            match process_ticker_quote(
583                msg,
584                price_precision,
585                instrument.size_precision(),
586                ts_init,
587                &mut quote_cache,
588            ) {
589                Ok(Some(quote)) => {
590                    Self::send_data(ctx, Data::Quote(quote));
591                }
592                Ok(None) => {}
593                Err(e) => log::warn!("Failed to parse Derive ticker quote: {e}"),
594            }
595        }
596
597        Self::handle_ticker_prices(msg, ctx, instrument_id, price_precision, ts_init);
598        Self::handle_ticker_derivatives(msg, ctx, instrument_id, ts_init);
599    }
600
601    fn handle_ticker_prices(
602        msg: &DeriveTickerMsg,
603        ctx: &WsMessageContext,
604        instrument_id: InstrumentId,
605        price_precision: u8,
606        ts_init: UnixNanos,
607    ) {
608        if ctx.active_mark_subs.contains(&instrument_id) {
609            match parse_mark_price(msg, price_precision, ts_init) {
610                Ok(Some(update)) => {
611                    Self::send_data(ctx, Data::MarkPrice(update));
612                }
613                Ok(None) => {}
614                Err(e) => log::warn!("Failed to parse Derive mark price: {e}"),
615            }
616        }
617
618        if ctx.active_index_subs.contains(&instrument_id) {
619            match parse_index_price(msg, price_precision, ts_init) {
620                Ok(Some(update)) => {
621                    Self::send_data(ctx, Data::IndexPrice(update));
622                }
623                Ok(None) => {}
624                Err(e) => log::warn!("Failed to parse Derive index price: {e}"),
625            }
626        }
627    }
628
629    fn handle_ticker_derivatives(
630        msg: &DeriveTickerMsg,
631        ctx: &WsMessageContext,
632        instrument_id: InstrumentId,
633        ts_init: UnixNanos,
634    ) {
635        if ctx.active_funding_subs.contains(&instrument_id) {
636            match parse_funding_rate(msg, ts_init) {
637                Ok(Some(update)) => {
638                    if let Err(e) = ctx.data_sender.send(DataEvent::FundingRate(update)) {
639                        log::error!("Failed to send Derive funding rate: {e}");
640                    }
641                }
642                Ok(None) => {}
643                Err(e) => log::warn!("Failed to parse Derive funding rate: {e}"),
644            }
645        }
646
647        if ctx.active_greeks_subs.contains(&instrument_id) {
648            match parse_option_greeks(msg, ts_init) {
649                Ok(Some(greeks)) => {
650                    if let Err(e) = ctx.data_sender.send(DataEvent::OptionGreeks(greeks)) {
651                        log::error!("Failed to send Derive option greeks: {e}");
652                    }
653                }
654                Ok(None) => {}
655                Err(e) => log::warn!("Failed to parse Derive option greeks: {e}"),
656            }
657        }
658    }
659
660    fn handle_public_trades(msg: &DeriveTradesMsg, ctx: &WsMessageContext) {
661        let ts_init = ctx.clock.get_time_ns();
662
663        for trade in &msg.trades {
664            let instrument_id = match format_instrument_id(trade.instrument_name) {
665                Ok(instrument_id) => instrument_id,
666                Err(e) => {
667                    log::warn!("Invalid Derive trade instrument: {e:#}");
668                    continue;
669                }
670            };
671
672            if !ctx.active_trade_subs.contains(&instrument_id) {
673                continue;
674            }
675
676            let Some(instrument) = ctx.instruments.get_cloned(&instrument_id) else {
677                log::warn!("Trade message received for unknown instrument {instrument_id}");
678                continue;
679            };
680
681            match parse_trade_tick(
682                trade,
683                instrument.price_precision(),
684                instrument.size_precision(),
685                ts_init,
686            ) {
687                Ok(tick) => {
688                    let mut emitted = ctx.trades_emitted.lock();
689                    if emitted.contains(&tick.trade_id) {
690                        continue;
691                    }
692
693                    if Self::send_data(ctx, Data::Trade(tick)) {
694                        emitted.add(tick.trade_id);
695                    }
696                }
697                Err(e) => log::warn!("Failed to parse Derive trade tick: {e}"),
698            }
699        }
700    }
701
702    fn send_data(ctx: &WsMessageContext, data: Data) -> bool {
703        match ctx.data_sender.send(DataEvent::Data(data)) {
704            Ok(()) => true,
705            Err(e) => {
706                log::error!("Failed to send Derive data event: {e}");
707                false
708            }
709        }
710    }
711
712    fn cache_provider_instruments(&self) {
713        let mut instruments = self
714            .provider
715            .store()
716            .get_all()
717            .values()
718            .cloned()
719            .collect::<Vec<_>>();
720
721        instruments.sort_by_key(|instrument| instrument.id());
722        for instrument in instruments {
723            self.cache_instrument(&instrument);
724            if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
725                log::warn!("Failed to send Derive instrument: {e}");
726            }
727        }
728    }
729
730    fn cache_instrument(&self, instrument: &InstrumentAny) {
731        cache_instrument(&self.instruments, instrument);
732    }
733
734    fn prepare_subscribe(&self, instrument_id: InstrumentId) -> anyhow::Result<bool> {
735        if self.instruments.contains_key(&instrument_id) {
736            return Ok(false);
737        }
738
739        if !self.config.auto_load_missing_instruments {
740            anyhow::bail!(
741                "Instrument {instrument_id} not found and `auto_load_missing_instruments` is disabled"
742            );
743        }
744
745        Ok(true)
746    }
747
748    async fn lazy_load_instrument(
749        http_client: DeriveHttpClient,
750        instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
751        instrument_id: InstrumentId,
752        include_expired: bool,
753    ) -> anyhow::Result<()> {
754        let currency = currency_from_instrument_id(&instrument_id)?;
755        let definitions = fetch_instrument_definitions(&http_client, currency, include_expired)
756            .await
757            .with_context(|| format!("failed to lazy-load Derive instruments for {currency}"))?;
758        let mut found = false;
759
760        for instrument in parse_instrument_definitions(definitions) {
761            if instrument.id() == instrument_id {
762                found = true;
763            }
764
765            cache_instrument(&instruments, &instrument);
766        }
767
768        if !found {
769            anyhow::bail!(InstrumentLookupError::not_found(instrument_id));
770        }
771
772        Ok(())
773    }
774
775    fn ws_handle(&self) -> DeriveWebSocketSubscriptionHandle {
776        self.ws_client.subscription_handle()
777    }
778
779    fn subscription_lifecycle(&self) -> SubscriptionLifecycle {
780        SubscriptionLifecycle {
781            registry: Arc::clone(&self.channel_subscriptions),
782            lock: Arc::clone(&self.subscription_lock),
783            snapshot_timeout: Duration::from_secs(self.config.book_snapshot_timeout_secs),
784            dispatch: SubscriptionDispatchState {
785                book_sync: self.book_sync.clone(),
786                active_book_delta_channels: Arc::clone(&self.active_book_delta_channels),
787                active_book_depth_channels: Arc::clone(&self.active_book_depth_channels),
788                active_ticker_channels: Arc::clone(&self.active_ticker_channels),
789                active_quote_subs: Arc::clone(&self.active_quote_subs),
790                active_trade_subs: Arc::clone(&self.active_trade_subs),
791                active_mark_subs: Arc::clone(&self.active_mark_subs),
792                active_index_subs: Arc::clone(&self.active_index_subs),
793                active_funding_subs: Arc::clone(&self.active_funding_subs),
794                active_greeks_subs: Arc::clone(&self.active_greeks_subs),
795                quote_cache: Arc::clone(&self.quote_cache),
796            },
797        }
798    }
799
800    fn subscribe_ticker_feed(
801        &self,
802        instrument_id: InstrumentId,
803        params: &Option<Params>,
804        feed: TickerFeed,
805        label: &'static str,
806    ) -> anyhow::Result<()> {
807        let owner = ChannelOwner::Ticker {
808            instrument_id,
809            feed,
810        };
811
812        let lifecycle = self.subscription_lifecycle();
813        if lifecycle.is_active(owner) {
814            return Ok(());
815        }
816
817        let channel = match self.active_ticker_channels.get_cloned(&instrument_id) {
818            Some(channel) => channel,
819            None => {
820                let instrument_name = format_venue_symbol(&instrument_id)?.to_string();
821                let interval = ticker_interval(params)?;
822                ticker_channel(&instrument_name, &interval)
823            }
824        };
825
826        let request = ChannelRequest::from_channel(&channel)?;
827        let needs_load = self.prepare_subscribe(instrument_id)?;
828
829        let Some(generation) = lifecycle.activate(owner, Some(&channel)) else {
830            return Ok(());
831        };
832
833        let ws = self.ws_handle();
834        let http_client = self.http_client.clone();
835        let include_expired = self.config.include_expired;
836        let instruments = Arc::clone(&self.instruments);
837
838        self.spawn_task("subscribe_ticker_feed", async move {
839            if needs_load
840                && let Err(e) = Self::lazy_load_instrument(
841                    http_client,
842                    instruments,
843                    instrument_id,
844                    include_expired,
845                )
846                .await
847            {
848                lifecycle.rollback(owner, generation);
849                log::error!("Lazy-load failed for {instrument_id} ({label}): {e}");
850                return Ok(());
851            }
852
853            run_channel_subscribe(lifecycle, owner, generation, request, ws).await
854        });
855
856        Ok(())
857    }
858
859    fn unsubscribe_channel_owner(&self, owner: ChannelOwner) -> anyhow::Result<()> {
860        let lifecycle = self.subscription_lifecycle();
861
862        let Some(removed) = lifecycle.remove(owner) else {
863            return Ok(());
864        };
865
866        if !removed.channel_empty {
867            return Ok(());
868        }
869
870        let Some(channel) = removed.channel else {
871            return Ok(());
872        };
873
874        let request = ChannelRequest::from_channel(&channel)?;
875        let ws = self.ws_handle();
876        self.spawn_task("unsubscribe_channel", async move {
877            run_channel_unsubscribe(lifecycle, request, ws).await
878        });
879
880        Ok(())
881    }
882}
883
884#[async_trait(?Send)]
885impl DataClient for DeriveDataClient {
886    fn client_id(&self) -> ClientId {
887        self.client_id
888    }
889
890    fn venue(&self) -> Option<Venue> {
891        Some(*DERIVE_VENUE)
892    }
893
894    fn start(&mut self) -> anyhow::Result<()> {
895        log::info!("Starting Derive data client: {}", self.client_id);
896        Ok(())
897    }
898
899    fn stop(&mut self) -> anyhow::Result<()> {
900        log::info!("Stopping Derive data client: {}", self.client_id);
901        self.cancellation_token.cancel();
902        self.abort_session_tasks();
903        self.abort_pending_tasks();
904        self.book_sync.clear();
905        self.is_connected.store(false, Ordering::Relaxed);
906        Ok(())
907    }
908
909    fn reset(&mut self) -> anyhow::Result<()> {
910        log::info!("Resetting Derive data client: {}", self.client_id);
911        self.cancellation_token.cancel();
912
913        self.abort_session_tasks();
914        self.abort_pending_tasks();
915        self.is_connected.store(false, Ordering::Relaxed);
916
917        // Leave the cancellation token cancelled; connect() refreshes it
918        // (and tears down the inner WS client) on the next lifecycle start.
919        self.instruments.store(AHashMap::new());
920        self.clear_subscription_state();
921        self.provider.store_mut().clear();
922        Ok(())
923    }
924
925    fn dispose(&mut self) -> anyhow::Result<()> {
926        log::debug!("Disposing Derive data client: {}", self.client_id);
927        self.stop()
928    }
929
930    fn is_connected(&self) -> bool {
931        self.is_connected.load(Ordering::SeqCst)
932    }
933
934    fn is_disconnected(&self) -> bool {
935        !self.is_connected()
936    }
937
938    async fn connect(&mut self) -> anyhow::Result<()> {
939        if self.is_connected()
940            && !self.cancellation_token.is_cancelled()
941            && self.session_tasks.is_open()
942            && self.pending_tasks.is_open()
943        {
944            return Ok(());
945        }
946
947        // Completes the async teardown deferred by sync reset()/stop().
948        if self.cancellation_token.is_cancelled()
949            || !self.session_tasks.is_open()
950            || !self.pending_tasks.is_open()
951        {
952            self.teardown_partial_connect().await?;
953            self.cancellation_token = CancellationToken::new();
954            self.session_tasks.start_generation().map_err(|e| {
955                anyhow::anyhow!("Failed to start Derive data session generation: {e}")
956            })?;
957
958            self.pending_tasks
959                .start_generation()
960                .map_err(|e| anyhow::anyhow!("Failed to start Derive data task generation: {e}"))?;
961        }
962
963        let cancellation_token = self.cancellation_token.clone();
964        let ws_shutdown = self.ws_client.shutdown_handle();
965
966        let setup_guard =
967            TaskGroupGuard::new(&[&self.session_tasks, &self.pending_tasks], move || {
968                cancellation_token.cancel();
969                ws_shutdown.begin_shutdown();
970            });
971
972        if !self.config.currencies.is_empty() {
973            self.provider
974                .load_all(None)
975                .await
976                .context("failed to load Derive instruments")?;
977            self.cache_provider_instruments();
978        }
979
980        self.ws_client
981            .connect()
982            .await
983            .context("failed to connect Derive WebSocket")?;
984        let session_result = self
985            .ws_client
986            .take_event_receiver()
987            .ok_or_else(|| anyhow::anyhow!("Derive WebSocket event receiver not initialized"))
988            .and_then(|rx| self.spawn_stream_task(rx))
989            .and_then(|()| self.spawn_book_monitor());
990
991        if let Err(e) = session_result {
992            if let Err(teardown_error) = self.teardown_partial_connect().await {
993                return Err(e.context(format!(
994                    "Derive data startup teardown failed: {teardown_error}"
995                )));
996            }
997
998            return Err(e);
999        }
1000
1001        self.is_connected.store(true, Ordering::Release);
1002        setup_guard.disarm();
1003        log::info!(
1004            "Connected Derive data client ({:?})",
1005            self.config.environment
1006        );
1007        Ok(())
1008    }
1009
1010    async fn disconnect(&mut self) -> anyhow::Result<()> {
1011        self.teardown_partial_connect().await?;
1012        log::info!("Disconnected Derive data client");
1013        Ok(())
1014    }
1015
1016    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
1017        if cmd.book_type != BookType::L2_MBP {
1018            anyhow::bail!("Derive only supports L2_MBP order book deltas");
1019        }
1020
1021        let instrument_id = cmd.instrument_id;
1022        let owner = ChannelOwner::BookDeltas(instrument_id);
1023        let lifecycle = self.subscription_lifecycle();
1024        if lifecycle.is_active(owner) {
1025            return Ok(());
1026        }
1027
1028        let instrument_name = format_venue_symbol(&instrument_id)?.to_string();
1029        let group = orderbook_group(&cmd.params)?;
1030        let depth = orderbook_depth(cmd.depth.map(|d| d.get()), &cmd.params)?;
1031        let channel = orderbook_channel(&instrument_name, &group, &depth);
1032        let request = ChannelRequest::from_channel(&channel)?;
1033        let needs_load = self.prepare_subscribe(instrument_id)?;
1034
1035        let Some(generation) = lifecycle.activate(owner, Some(&channel)) else {
1036            return Ok(());
1037        };
1038
1039        let ws = self.ws_handle();
1040        let http_client = self.http_client.clone();
1041        let include_expired = self.config.include_expired;
1042        let instruments = Arc::clone(&self.instruments);
1043
1044        self.spawn_task("subscribe_book_deltas", async move {
1045            if needs_load
1046                && let Err(e) = Self::lazy_load_instrument(
1047                    http_client,
1048                    instruments,
1049                    instrument_id,
1050                    include_expired,
1051                )
1052                .await
1053            {
1054                lifecycle.rollback(owner, generation);
1055                log::error!("Lazy-load failed for {instrument_id} (book deltas): {e}");
1056                return Ok(());
1057            }
1058
1059            run_channel_subscribe(lifecycle, owner, generation, request, ws).await
1060        });
1061
1062        Ok(())
1063    }
1064
1065    fn subscribe_book_depth(&mut self, cmd: SubscribeBookDepth) -> anyhow::Result<()> {
1066        if cmd.book_type != BookType::L2_MBP {
1067            anyhow::bail!("Derive only supports L2_MBP order book depth");
1068        }
1069
1070        let instrument_id = cmd.instrument_id;
1071        let owner = ChannelOwner::BookDepth(instrument_id);
1072        let lifecycle = self.subscription_lifecycle();
1073        if lifecycle.is_active(owner) {
1074            return Ok(());
1075        }
1076
1077        let instrument_name = format_venue_symbol(&instrument_id)?.to_string();
1078        let group = orderbook_group(&cmd.params)?;
1079        let depth = DeriveOrderbookDepth::D10.to_string();
1080        let channel = orderbook_channel(&instrument_name, &group, &depth);
1081        let request = ChannelRequest::from_channel(&channel)?;
1082        let needs_load = self.prepare_subscribe(instrument_id)?;
1083
1084        let Some(generation) = lifecycle.activate(owner, Some(&channel)) else {
1085            return Ok(());
1086        };
1087
1088        let ws = self.ws_handle();
1089        let http_client = self.http_client.clone();
1090        let include_expired = self.config.include_expired;
1091        let instruments = Arc::clone(&self.instruments);
1092
1093        self.spawn_task("subscribe_book_depth", async move {
1094            if needs_load
1095                && let Err(e) = Self::lazy_load_instrument(
1096                    http_client,
1097                    instruments,
1098                    instrument_id,
1099                    include_expired,
1100                )
1101                .await
1102            {
1103                lifecycle.rollback(owner, generation);
1104                log::error!("Lazy-load failed for {instrument_id} (book depth): {e}");
1105                return Ok(());
1106            }
1107
1108            run_channel_subscribe(lifecycle, owner, generation, request, ws).await
1109        });
1110
1111        Ok(())
1112    }
1113
1114    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
1115        self.subscribe_ticker_feed(cmd.instrument_id, &cmd.params, TickerFeed::Quote, "quotes")
1116    }
1117
1118    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
1119        let instrument_id = cmd.instrument_id;
1120        let owner = ChannelOwner::Trades(instrument_id);
1121        let lifecycle = self.subscription_lifecycle();
1122        if lifecycle.is_active(owner) {
1123            return Ok(());
1124        }
1125
1126        let needs_load = self.prepare_subscribe(instrument_id)?;
1127
1128        let Some(generation) = lifecycle.activate(owner, None) else {
1129            return Ok(());
1130        };
1131
1132        let ws = self.ws_handle();
1133        let http_client = self.http_client.clone();
1134        let include_expired = self.config.include_expired;
1135        let instruments = Arc::clone(&self.instruments);
1136
1137        self.spawn_task("subscribe_trades", async move {
1138            if needs_load
1139                && let Err(e) = Self::lazy_load_instrument(
1140                    http_client,
1141                    Arc::clone(&instruments),
1142                    instrument_id,
1143                    include_expired,
1144                )
1145                .await
1146            {
1147                lifecycle.rollback(owner, generation);
1148                log::error!("Lazy-load failed for {instrument_id} (trades): {e}");
1149                return Ok(());
1150            }
1151
1152            if !lifecycle.is_current(owner, generation) {
1153                return Ok(());
1154            }
1155
1156            let Some(instrument) = instruments.get_cloned(&instrument_id) else {
1157                lifecycle.rollback(owner, generation);
1158                log::error!("Instrument {instrument_id} not found for Derive trades");
1159                return Ok(());
1160            };
1161
1162            let channel = match trade_channel(&instrument) {
1163                Ok(channel) => channel,
1164                Err(e) => {
1165                    lifecycle.rollback(owner, generation);
1166                    log::error!("Failed to resolve Derive trades channel: {e}");
1167                    return Ok(());
1168                }
1169            };
1170
1171            let request = match ChannelRequest::from_channel(&channel) {
1172                Ok(request) => request,
1173                Err(e) => {
1174                    lifecycle.rollback(owner, generation);
1175                    log::error!("Invalid Derive trades channel `{channel}`: {e}");
1176                    return Ok(());
1177                }
1178            };
1179
1180            if !lifecycle.attach_channel(owner, generation, channel) {
1181                return Ok(());
1182            }
1183
1184            run_channel_subscribe(lifecycle, owner, generation, request, ws).await
1185        });
1186
1187        Ok(())
1188    }
1189
1190    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
1191        self.subscribe_ticker_feed(
1192            cmd.instrument_id,
1193            &cmd.params,
1194            TickerFeed::Mark,
1195            "mark prices",
1196        )
1197    }
1198
1199    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
1200        self.subscribe_ticker_feed(
1201            cmd.instrument_id,
1202            &cmd.params,
1203            TickerFeed::Index,
1204            "index prices",
1205        )
1206    }
1207
1208    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
1209        self.subscribe_ticker_feed(
1210            cmd.instrument_id,
1211            &cmd.params,
1212            TickerFeed::Funding,
1213            "funding rates",
1214        )
1215    }
1216
1217    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
1218        self.subscribe_ticker_feed(
1219            cmd.instrument_id,
1220            &cmd.params,
1221            TickerFeed::Greeks,
1222            "option greeks",
1223        )
1224    }
1225
1226    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
1227        self.unsubscribe_channel_owner(ChannelOwner::BookDeltas(cmd.instrument_id))
1228    }
1229
1230    fn unsubscribe_book_depth(&mut self, cmd: &UnsubscribeBookDepth) -> anyhow::Result<()> {
1231        self.unsubscribe_channel_owner(ChannelOwner::BookDepth(cmd.instrument_id))
1232    }
1233
1234    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
1235        self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1236            instrument_id: cmd.instrument_id,
1237            feed: TickerFeed::Quote,
1238        })
1239    }
1240
1241    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
1242        self.unsubscribe_channel_owner(ChannelOwner::Trades(cmd.instrument_id))
1243    }
1244
1245    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
1246        self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1247            instrument_id: cmd.instrument_id,
1248            feed: TickerFeed::Mark,
1249        })
1250    }
1251
1252    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
1253        self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1254            instrument_id: cmd.instrument_id,
1255            feed: TickerFeed::Index,
1256        })
1257    }
1258
1259    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
1260        self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1261            instrument_id: cmd.instrument_id,
1262            feed: TickerFeed::Funding,
1263        })
1264    }
1265
1266    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
1267        self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1268            instrument_id: cmd.instrument_id,
1269            feed: TickerFeed::Greeks,
1270        })
1271    }
1272
1273    fn request_quotes(&self, request: RequestQuotes) -> anyhow::Result<()> {
1274        // No historical quote endpoint; `public/get_tickers` is a current snapshot.
1275        let instrument_id = request.instrument_id;
1276        let instrument = self
1277            .instruments
1278            .get_cloned(&instrument_id)
1279            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1280        let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1281        let price_precision = instrument.price_precision();
1282        let size_precision = instrument.size_precision();
1283
1284        let http_client = self.http_client.clone();
1285        let sender = self.data_sender.clone();
1286        let clock = self.clock;
1287        let client_id = request.client_id.unwrap_or(self.client_id);
1288        let request_id = request.request_id;
1289        let params = request.params;
1290        let start_nanos = datetime_to_unix_nanos(request.start);
1291        let end_nanos = datetime_to_unix_nanos(request.end);
1292
1293        self.spawn_task(format!("request_quotes {instrument_id}"), async move {
1294            let ticker = match http_client.get_ticker(&venue_symbol).await {
1295                Ok(ticker) => ticker,
1296                Err(e) => {
1297                    log::error!("Failed to fetch Derive ticker for {instrument_id}: {e:?}");
1298                    return Ok(());
1299                }
1300            };
1301
1302            let ts_init = clock.get_time_ns();
1303
1304            let quotes = match parse_ticker_quote_from_rest(
1305                &ticker,
1306                price_precision,
1307                size_precision,
1308                ts_init,
1309            ) {
1310                Ok(quote) => {
1311                    // Drop the snapshot when bounded callers ask for a past window.
1312                    let within_start = start_nanos.is_none_or(|nanos| quote.ts_event >= nanos);
1313                    let within_end = end_nanos.is_none_or(|nanos| quote.ts_event <= nanos);
1314                    if within_start && within_end {
1315                        vec![quote]
1316                    } else {
1317                        Vec::new()
1318                    }
1319                }
1320                Err(e) => {
1321                    log::warn!("Failed to parse Derive ticker for {instrument_id}: {e}");
1322                    Vec::new()
1323                }
1324            };
1325
1326            let response = DataResponse::Quotes(QuotesResponse::new(
1327                request_id,
1328                client_id,
1329                instrument_id,
1330                quotes,
1331                start_nanos,
1332                end_nanos,
1333                clock.get_time_ns(),
1334                params,
1335            ));
1336
1337            if let Err(e) = sender.send(DataEvent::Response(response)) {
1338                log::error!("Failed to send Derive quotes response: {e}");
1339            }
1340
1341            Ok(())
1342        });
1343
1344        Ok(())
1345    }
1346
1347    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
1348        let instrument_id = request.instrument_id;
1349        let instrument = self
1350            .instruments
1351            .get_cloned(&instrument_id)
1352            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1353        let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1354        let price_precision = instrument.price_precision();
1355        let size_precision = instrument.size_precision();
1356
1357        let http_client = self.http_client.clone();
1358        let sender = self.data_sender.clone();
1359        let clock = self.clock;
1360        let client_id = request.client_id.unwrap_or(self.client_id);
1361        let request_id = request.request_id;
1362        let params = request.params;
1363        let start = request.start;
1364        let end = request.end;
1365        let limit = request.limit.map(NonZeroUsize::get);
1366        let start_nanos = datetime_to_unix_nanos(start);
1367        let end_nanos = datetime_to_unix_nanos(end);
1368        let from_timestamp = start.map(|dt| dt.as_millisecond());
1369
1370        let to_timestamp = Some(match end {
1371            Some(dt) => dt.as_millisecond(),
1372            None => i64::try_from(clock.get_time_ms())
1373                .context("Derive current time exceeds i64 milliseconds")?,
1374        });
1375
1376        self.spawn_task("request_trades", async move {
1377            // Hold page_size constant across requests: the venue paginates by
1378            // offset = (page - 1) * page_size, so shrinking page_size mid-walk
1379            // would skip and duplicate trades.
1380            let page_size = limit.map_or(DERIVE_TRADES_PAGE_SIZE, |cap| {
1381                cap.min(DERIVE_TRADES_PAGE_SIZE as usize) as u32
1382            });
1383
1384            let mut trades = Vec::new();
1385            let mut seen_trade_ids = AHashSet::new();
1386            let mut pages = PaginationCursor::new_public();
1387
1388            loop {
1389                let result = match http_client
1390                    .get_trade_history(
1391                        &venue_symbol,
1392                        from_timestamp,
1393                        to_timestamp,
1394                        pages.page(),
1395                        page_size,
1396                    )
1397                    .await
1398                {
1399                    Ok(result) => result,
1400                    Err(e) => {
1401                        log::error!("Failed to fetch Derive trades for {instrument_id}: {e:?}");
1402                        return Ok(());
1403                    }
1404                };
1405
1406                let row_ids: Vec<String> = result
1407                    .trades
1408                    .iter()
1409                    .map(|trade| {
1410                        format!(
1411                            "{}:{:?}:{:?}",
1412                            trade.trade_id, trade.liquidity_role, trade.subaccount_id
1413                        )
1414                    })
1415                    .collect();
1416
1417                let more =
1418                    match pages.advance(&result.pagination, row_ids.iter().map(String::as_str)) {
1419                        Ok(more) => more,
1420                        Err(e) => {
1421                            log::error!("Failed to collect Derive trades for {instrument_id}: {e}");
1422                            return Ok(());
1423                        }
1424                    };
1425
1426                let ts_init = clock.get_time_ns();
1427
1428                for trade in &result.trades {
1429                    match parse_trade_tick_from_rest(
1430                        trade,
1431                        price_precision,
1432                        size_precision,
1433                        ts_init,
1434                    ) {
1435                        Ok(tick) if seen_trade_ids.insert(tick.trade_id) => trades.push(tick),
1436                        Ok(_) => {}
1437                        Err(e) => log::warn!(
1438                            "Failed to parse Derive trade {} for {instrument_id}: {e}",
1439                            trade.trade_id,
1440                        ),
1441                    }
1442                }
1443
1444                if let Some(cap) = limit
1445                    && trades.len() >= cap
1446                {
1447                    break;
1448                }
1449
1450                if !more {
1451                    break;
1452                }
1453            }
1454
1455            trades.sort_by_key(|trade| trade.ts_event);
1456            if let Some(cap) = limit
1457                && trades.len() > cap
1458            {
1459                trades.drain(..trades.len() - cap);
1460            }
1461
1462            let response = DataResponse::Trades(TradesResponse::new(
1463                request_id,
1464                client_id,
1465                instrument_id,
1466                trades,
1467                start_nanos,
1468                end_nanos,
1469                clock.get_time_ns(),
1470                params,
1471            ));
1472
1473            if let Err(e) = sender.send(DataEvent::Response(response)) {
1474                log::error!("Failed to send Derive trades response: {e}");
1475            }
1476
1477            Ok(())
1478        });
1479
1480        Ok(())
1481    }
1482
1483    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
1484        let instrument_id = request.instrument_id;
1485        let instrument = self
1486            .instruments
1487            .get_cloned(&instrument_id)
1488            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1489        anyhow::ensure!(
1490            matches!(instrument, InstrumentAny::CryptoPerpetual(_)),
1491            "Funding rates are only available for Derive perpetual instruments (got {instrument_id})",
1492        );
1493        let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1494
1495        let http_client = self.http_client.clone();
1496        let sender = self.data_sender.clone();
1497        let clock = self.clock;
1498        let client_id = request.client_id.unwrap_or(self.client_id);
1499        let request_id = request.request_id;
1500        let params = request.params;
1501        let start = request.start;
1502        let end = request.end;
1503        let limit = request.limit.map(NonZeroUsize::get);
1504        let start_nanos = datetime_to_unix_nanos(start);
1505        let end_nanos = datetime_to_unix_nanos(end);
1506        let start_ms = start.map(|dt| dt.as_millisecond());
1507        let end_ms = end.map(|dt| dt.as_millisecond());
1508
1509        self.spawn_task("request_funding_rates", async move {
1510            let result = match http_client
1511                .get_funding_rate_history(&venue_symbol, start_ms, end_ms, None)
1512                .await
1513            {
1514                Ok(result) => result,
1515                Err(e) => {
1516                    log::error!(
1517                        "Failed to fetch Derive funding rate history for {instrument_id}: {e:?}",
1518                    );
1519                    return Ok(());
1520                }
1521            };
1522
1523            let ts_init = clock.get_time_ns();
1524            let mut updates = Vec::with_capacity(result.funding_rate_history.len());
1525
1526            for record in &result.funding_rate_history {
1527                match parse_funding_rate_history_record(record, instrument_id, None, ts_init) {
1528                    Ok(update) => updates.push(update),
1529                    Err(e) => log::warn!(
1530                        "Failed to parse Derive funding rate record for {instrument_id} at {}: {e}",
1531                        record.timestamp,
1532                    ),
1533                }
1534            }
1535
1536            updates.sort_by_key(|update| update.ts_event);
1537            if let Some(cap) = limit
1538                && updates.len() > cap
1539            {
1540                updates.drain(..updates.len() - cap);
1541            }
1542
1543            let response = DataResponse::FundingRates(FundingRatesResponse::new(
1544                request_id,
1545                client_id,
1546                instrument_id,
1547                updates,
1548                start_nanos,
1549                end_nanos,
1550                clock.get_time_ns(),
1551                params,
1552            ));
1553
1554            if let Err(e) = sender.send(DataEvent::Response(response)) {
1555                log::error!("Failed to send Derive funding rates response: {e}");
1556            }
1557
1558            Ok(())
1559        });
1560
1561        Ok(())
1562    }
1563
1564    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
1565        let bar_type = request.bar_type;
1566        anyhow::ensure!(
1567            bar_type.aggregation_source() == AggregationSource::External,
1568            "Derive only supports EXTERNAL aggregation source (got {bar_type})",
1569        );
1570        let spec = bar_type.spec();
1571        anyhow::ensure!(
1572            spec.price_type == PriceType::Last,
1573            "Derive candles are trade-based; only PriceType::Last is supported (got {bar_type})",
1574        );
1575
1576        let instrument_id = bar_type.instrument_id();
1577        let instrument = self
1578            .instruments
1579            .get_cloned(&instrument_id)
1580            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1581        let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1582        let price_precision = instrument.price_precision();
1583        let size_precision = instrument.size_precision();
1584
1585        let period = bar_spec_to_derive_period(spec.aggregation, spec.step.get() as u64)
1586            .with_context(|| format!("unsupported Derive bar spec for {bar_type}"))?;
1587
1588        let http_client = self.http_client.clone();
1589        let sender = self.data_sender.clone();
1590        let clock = self.clock;
1591        let client_id = request.client_id.unwrap_or(self.client_id);
1592        let request_id = request.request_id;
1593        let params = request.params;
1594        let start = request.start;
1595        let end = request.end;
1596        let limit = request.limit.map(NonZeroUsize::get);
1597        let start_nanos = datetime_to_unix_nanos(start);
1598        let end_nanos = datetime_to_unix_nanos(end);
1599
1600        // The venue requires both bounds in UNIX seconds. Default end to now
1601        // and start to one window of `limit` buckets (or 1000) before end.
1602        let request_time = clock.get_time_ns();
1603        let now_secs = (request_time.as_u64() / NANOSECONDS_IN_SECOND) as i64;
1604        let end_ts = end.map_or(now_secs, |dt| dt.as_second());
1605        let default_span = i64::from(period) * limit.unwrap_or(DERIVE_CANDLES_DEFAULT_LIMIT) as i64;
1606        let start_ts = start.map_or(end_ts - default_span, |dt| dt.as_second());
1607
1608        self.spawn_task("request_bars", async move {
1609            // Venue caps each call at 5000 candles; walk backwards by shrinking
1610            // `current_end` to one second before each page's earliest bucket.
1611            let mut seen_timestamps: AHashSet<i64> = AHashSet::new();
1612            let mut pages: Vec<Vec<Bar>> = Vec::new();
1613            let mut total_bars = 0usize;
1614            let mut current_end = end_ts;
1615            let mut page_count = 0;
1616            loop {
1617                page_count += 1;
1618
1619                let mut records = match http_client
1620                    .get_candles(&venue_symbol, start_ts, current_end, period)
1621                    .await
1622                {
1623                    Ok(records) => records,
1624                    Err(e) => {
1625                        log::error!("Failed to fetch Derive candles for {bar_type}: {e:?}");
1626                        return Ok(());
1627                    }
1628                };
1629
1630                if records.is_empty() {
1631                    break;
1632                }
1633
1634                records.sort_by_key(|r| r.timestamp_bucket);
1635
1636                let has_new = records
1637                    .iter()
1638                    .any(|r| !seen_timestamps.contains(&r.timestamp_bucket));
1639
1640                if !has_new {
1641                    break;
1642                }
1643
1644                let ts_init = clock.get_time_ns();
1645                let mut page_bars = Vec::with_capacity(records.len());
1646                let mut earliest_ts: Option<i64> = None;
1647
1648                for record in &records {
1649                    let bucket = record.timestamp_bucket;
1650                    if earliest_ts.is_none_or(|ts| bucket < ts) {
1651                        earliest_ts = Some(bucket);
1652                    }
1653
1654                    if seen_timestamps.contains(&bucket) {
1655                        continue;
1656                    }
1657
1658                    match parse_candle_record(
1659                        record,
1660                        bar_type,
1661                        price_precision,
1662                        size_precision,
1663                        ts_init,
1664                    ) {
1665                        Ok(bar) => {
1666                            seen_timestamps.insert(bucket);
1667
1668                            if bar.ts_event <= request_time {
1669                                page_bars.push(bar);
1670                            }
1671                        }
1672                        Err(e) => log::warn!(
1673                            "Failed to parse Derive candle for {bar_type} at {bucket}: {e}",
1674                        ),
1675                    }
1676                }
1677
1678                total_bars += page_bars.len();
1679                pages.push(page_bars);
1680
1681                if let Some(cap) = limit
1682                    && total_bars >= cap
1683                {
1684                    break;
1685                }
1686
1687                let Some(earliest) = earliest_ts else {
1688                    break;
1689                };
1690
1691                if earliest <= start_ts {
1692                    break;
1693                }
1694
1695                current_end = earliest - 1;
1696
1697                if page_count >= DERIVE_CANDLES_MAX_PAGES {
1698                    log::warn!(
1699                        "Derive bars pagination hit safety cap of {DERIVE_CANDLES_MAX_PAGES} pages for {bar_type}",
1700                    );
1701                    break;
1702                }
1703            }
1704
1705            let mut bars: Vec<Bar> = Vec::with_capacity(total_bars);
1706            for page in pages.into_iter().rev() {
1707                bars.extend(page);
1708            }
1709
1710            if let Some(cap) = limit
1711                && bars.len() > cap
1712            {
1713                let drop_count = bars.len() - cap;
1714                bars.drain(..drop_count);
1715            }
1716
1717            let response = DataResponse::Bars(BarsResponse::new(
1718                request_id,
1719                client_id,
1720                bar_type,
1721                bars,
1722                start_nanos,
1723                end_nanos,
1724                clock.get_time_ns(),
1725                params,
1726            ));
1727
1728            if let Err(e) = sender.send(DataEvent::Response(response)) {
1729                log::error!("Failed to send Derive bars response: {e}");
1730            }
1731
1732            Ok(())
1733        });
1734
1735        Ok(())
1736    }
1737
1738    fn request_option_chain_reference_price(
1739        &self,
1740        request: RequestOptionChainReferencePrice,
1741    ) -> anyhow::Result<()> {
1742        let series_id = request.series_id;
1743        let instrument_id = request.instrument_id;
1744        let instrument = self
1745            .instruments
1746            .get_cloned(&instrument_id)
1747            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1748        anyhow::ensure!(
1749            matches!(instrument, InstrumentAny::CryptoOption(_)),
1750            "Derive option-chain reference prices require an option instrument (got {instrument_id})",
1751        );
1752        let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1753
1754        let http_client = self.http_client.clone();
1755        let sender = self.data_sender.clone();
1756        let clock = self.clock;
1757        let client_id = request.client_id.unwrap_or(self.client_id);
1758        let request_id = request.request_id;
1759        let params = request.params;
1760
1761        self.spawn_task("request_option_chain_reference_price", async move {
1762            let price = match http_client.get_ticker(&venue_symbol).await {
1763                Ok(ticker) => match ticker.option_pricing.as_ref() {
1764                    Some(pricing) if pricing.forward_price > Decimal::ZERO => {
1765                        match Price::from_decimal(pricing.forward_price) {
1766                            Ok(price) => Some(price),
1767                            Err(e) => {
1768                                log::warn!(
1769                                    "Invalid Derive option-chain reference price for {instrument_id}: {e}"
1770                                );
1771                                None
1772                            }
1773                        }
1774                    }
1775                    None => {
1776                        log::warn!(
1777                            "Derive ticker for {instrument_id} has no option pricing reference"
1778                        );
1779                        None
1780                    }
1781                    Some(_) => None,
1782                },
1783                Err(e) => {
1784                    log::error!(
1785                        "Option-chain reference price request failed for {series_id}: {e:?}"
1786                    );
1787                    None
1788                }
1789            };
1790
1791            let response = DataResponse::OptionChainReferencePrice(
1792                OptionChainReferencePriceResponse::new(
1793                    request_id,
1794                    client_id,
1795                    series_id,
1796                    price,
1797                    clock.get_time_ns(),
1798                    params,
1799                ),
1800            );
1801
1802            if let Err(e) = sender.send(DataEvent::Response(response)) {
1803                log::error!("Failed to send option-chain reference price response: {e}");
1804            }
1805
1806            Ok(())
1807        });
1808
1809        Ok(())
1810    }
1811
1812    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
1813        let currencies = self.config.currencies.clone();
1814        if currencies.is_empty() {
1815            anyhow::bail!(
1816                "Derive request_instruments requires at least one configured currency \
1817                 (DeriveDataClientConfig::currencies)"
1818            );
1819        }
1820
1821        let http_client = self.http_client.clone();
1822        let include_expired = self.config.include_expired;
1823        let instruments_cache = Arc::clone(&self.instruments);
1824        let sender = self.data_sender.clone();
1825        let clock = self.clock;
1826        let venue = self.venue().unwrap_or(*DERIVE_VENUE);
1827        let client_id = request.client_id.unwrap_or(self.client_id);
1828        let request_id = request.request_id;
1829        let start_nanos = datetime_to_unix_nanos(request.start);
1830        let end_nanos = datetime_to_unix_nanos(request.end);
1831        let params = request.params;
1832
1833        self.spawn_task("request_instruments", async move {
1834            let mut all_instruments = Vec::new();
1835
1836            for currency in currencies {
1837                match fetch_instrument_definitions(&http_client, &currency, include_expired).await {
1838                    Ok(definitions) => {
1839                        for instrument in parse_instrument_definitions(definitions) {
1840                            cache_instrument(&instruments_cache, &instrument);
1841                            all_instruments.push(instrument);
1842                        }
1843                    }
1844                    Err(e) => {
1845                        log::error!("Failed to fetch Derive instruments for {currency}: {e:?}");
1846                    }
1847                }
1848            }
1849
1850            let response = DataResponse::Instruments(InstrumentsResponse::new(
1851                request_id,
1852                client_id,
1853                venue,
1854                all_instruments,
1855                start_nanos,
1856                end_nanos,
1857                clock.get_time_ns(),
1858                params,
1859            ));
1860
1861            if let Err(e) = sender.send(DataEvent::Response(response)) {
1862                log::error!("Failed to send Derive instruments response: {e}");
1863            }
1864
1865            Ok(())
1866        });
1867
1868        Ok(())
1869    }
1870
1871    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
1872        let instrument_id = request.instrument_id;
1873        let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1874
1875        let http_client = self.http_client.clone();
1876        let instruments_cache = Arc::clone(&self.instruments);
1877        let sender = self.data_sender.clone();
1878        let clock = self.clock;
1879        let client_id = request.client_id.unwrap_or(self.client_id);
1880        let request_id = request.request_id;
1881        let start_nanos = datetime_to_unix_nanos(request.start);
1882        let end_nanos = datetime_to_unix_nanos(request.end);
1883        let params = request.params;
1884
1885        self.spawn_task("request_instrument", async move {
1886            let definition = match http_client.get_instrument(&venue_symbol).await {
1887                Ok(definition) => definition,
1888                Err(e) => {
1889                    log::error!("Failed to fetch Derive instrument {instrument_id}: {e:?}");
1890                    return Ok(());
1891                }
1892            };
1893
1894            let ts_init = clock.get_time_ns();
1895
1896            let instrument = match parse_derive_instrument_any(&definition, ts_init) {
1897                Ok(Some(instrument)) => instrument,
1898                Ok(None) => {
1899                    log::warn!(
1900                        "Derive instrument {instrument_id} resolved to an unsupported type ({:?})",
1901                        definition.instrument_type,
1902                    );
1903                    return Ok(());
1904                }
1905                Err(e) => {
1906                    log::error!("Failed to parse Derive instrument {instrument_id}: {e}");
1907                    return Ok(());
1908                }
1909            };
1910
1911            cache_instrument(&instruments_cache, &instrument);
1912
1913            let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
1914                request_id,
1915                client_id,
1916                instrument.id(),
1917                instrument,
1918                start_nanos,
1919                end_nanos,
1920                clock.get_time_ns(),
1921                params,
1922            )));
1923
1924            if let Err(e) = sender.send(DataEvent::Response(response)) {
1925                log::error!("Failed to send Derive instrument response: {e}");
1926            }
1927
1928            Ok(())
1929        });
1930
1931        Ok(())
1932    }
1933}
1934
1935#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1936enum TickerFeed {
1937    Quote,
1938    Mark,
1939    Index,
1940    Funding,
1941    Greeks,
1942}
1943
1944#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1945enum ChannelOwner {
1946    BookDeltas(InstrumentId),
1947    BookDepth(InstrumentId),
1948    Ticker {
1949        instrument_id: InstrumentId,
1950        feed: TickerFeed,
1951    },
1952    Trades(InstrumentId),
1953}
1954
1955#[derive(Debug)]
1956struct OwnedChannel {
1957    generation: u64,
1958    channel: Option<String>,
1959}
1960
1961#[derive(Debug, Default)]
1962struct ChannelSubscriptionState {
1963    next_generation: u64,
1964    owners: AHashMap<ChannelOwner, OwnedChannel>,
1965    channels: AHashMap<String, AHashSet<ChannelOwner>>,
1966}
1967
1968#[derive(Debug, Default)]
1969struct ChannelSubscriptionRegistry {
1970    state: Mutex<ChannelSubscriptionState>,
1971    transitions: DashMap<String, Weak<tokio::sync::Mutex<()>>>,
1972}
1973
1974const TRANSITION_GC_THRESHOLD: usize = 256;
1975
1976#[derive(Debug)]
1977struct RemovedSubscription {
1978    channel: Option<String>,
1979    channel_empty: bool,
1980}
1981
1982#[derive(Debug, Clone)]
1983struct SubscriptionDispatchState {
1984    book_sync: BookSyncTracker,
1985    active_book_delta_channels: Arc<AtomicMap<InstrumentId, String>>,
1986    active_book_depth_channels: Arc<AtomicMap<InstrumentId, String>>,
1987    active_ticker_channels: Arc<AtomicMap<InstrumentId, String>>,
1988    active_quote_subs: Arc<AtomicSet<InstrumentId>>,
1989    active_trade_subs: Arc<AtomicSet<InstrumentId>>,
1990    active_mark_subs: Arc<AtomicSet<InstrumentId>>,
1991    active_index_subs: Arc<AtomicSet<InstrumentId>>,
1992    active_funding_subs: Arc<AtomicSet<InstrumentId>>,
1993    active_greeks_subs: Arc<AtomicSet<InstrumentId>>,
1994    quote_cache: Arc<Mutex<QuoteCache>>,
1995}
1996
1997#[derive(Debug, Clone)]
1998struct SubscriptionLifecycle {
1999    snapshot_timeout: Duration,
2000    registry: Arc<ChannelSubscriptionRegistry>,
2001    lock: Arc<Mutex<()>>,
2002    dispatch: SubscriptionDispatchState,
2003}
2004
2005#[derive(Debug)]
2006enum ChannelRequest {
2007    Orderbook {
2008        channel: String,
2009        instrument_name: String,
2010        group: String,
2011        depth: String,
2012    },
2013    Ticker {
2014        channel: String,
2015        instrument_name: String,
2016        interval: String,
2017    },
2018    Trades {
2019        channel: String,
2020        instrument_type: String,
2021        currency: String,
2022    },
2023}
2024
2025async fn run_channel_subscribe(
2026    lifecycle: SubscriptionLifecycle,
2027    owner: ChannelOwner,
2028    generation: u64,
2029    request: ChannelRequest,
2030    ws: DeriveWebSocketSubscriptionHandle,
2031) -> anyhow::Result<()> {
2032    let channel = request.channel().to_string();
2033    let transition = lifecycle.registry.transition(&channel);
2034    let _guard = transition.lock().await;
2035
2036    if !lifecycle.is_current_channel(owner, generation, &channel) {
2037        return Ok(());
2038    }
2039
2040    let book_pending = lifecycle
2041        .dispatch
2042        .book_sync
2043        .write(&channel)
2044        .is_some_and(|write| write.gate.lock().is_closed());
2045
2046    if ws.has_subscription(&channel) && !book_pending {
2047        return Ok(());
2048    }
2049
2050    let result = if matches!(&request, ChannelRequest::Orderbook { .. }) {
2051        ws.subscribe_book_gated(channel.clone()).await
2052    } else {
2053        request.subscribe(&ws).await
2054    };
2055
2056    if let Err(e) = result {
2057        lifecycle.rollback(owner, generation);
2058        log::error!(
2059            "Failed to subscribe to Derive channel `{channel}`: {}",
2060            e.redacted_message()
2061        );
2062
2063        let cleanup_error = match request.unsubscribe(&ws).await {
2064            Ok(()) => None,
2065            Err(cleanup_error) => {
2066                log::error!(
2067                    "Failed to clean up uncertain Derive channel `{channel}`: {}",
2068                    cleanup_error.redacted_message(),
2069                );
2070                Some(cleanup_error)
2071            }
2072        };
2073
2074        if retain_channel_for_reconnect(
2075            &ws,
2076            &channel,
2077            lifecycle.has_owners(&channel),
2078            cleanup_error.as_ref(),
2079        ) {
2080            log::error!(
2081                "Derive channel `{channel}` remains uncertain and will replay on reconnect",
2082            );
2083        }
2084
2085        return Ok(());
2086    }
2087
2088    lifecycle
2089        .dispatch
2090        .book_sync
2091        .subscription_written(&channel, lifecycle.snapshot_timeout);
2092    if !lifecycle.has_owners(&channel)
2093        && let Err(e) = request.unsubscribe(&ws).await
2094    {
2095        log::error!("Failed to clean up stale Derive channel `{channel}`: {e}");
2096        ws.forget_subscription(&channel);
2097    }
2098
2099    Ok(())
2100}
2101
2102async fn run_channel_unsubscribe(
2103    lifecycle: SubscriptionLifecycle,
2104    request: ChannelRequest,
2105    ws: DeriveWebSocketSubscriptionHandle,
2106) -> anyhow::Result<()> {
2107    let channel = request.channel().to_string();
2108    let transition = lifecycle.registry.transition(&channel);
2109    let _guard = transition.lock().await;
2110
2111    if lifecycle.has_owners(&channel) || !ws.has_subscription(&channel) {
2112        return Ok(());
2113    }
2114
2115    if let Err(e) = request.unsubscribe(&ws).await {
2116        log::error!("Failed to unsubscribe from Derive channel `{channel}`: {e}");
2117        ws.forget_subscription(&channel);
2118    }
2119
2120    Ok(())
2121}
2122
2123impl SubscriptionLifecycle {
2124    fn is_active(&self, owner: ChannelOwner) -> bool {
2125        let _guard = self.lock.lock();
2126        self.registry.state.lock().owners.contains_key(&owner)
2127    }
2128
2129    fn activate(&self, owner: ChannelOwner, channel: Option<&str>) -> Option<u64> {
2130        let _guard = self.lock.lock();
2131        let mut state = self.registry.state.lock();
2132        let generation = state.activate(owner, channel)?;
2133        self.dispatch.activate(owner, channel);
2134
2135        if let ChannelOwner::BookDeltas(instrument_id) | ChannelOwner::BookDepth(instrument_id) =
2136            owner
2137        {
2138            self.dispatch
2139                .book_sync
2140                .record_subscription(channel.expect("book channel present"), instrument_id);
2141        }
2142
2143        Some(generation)
2144    }
2145
2146    fn attach_channel(&self, owner: ChannelOwner, generation: u64, channel: String) -> bool {
2147        let _guard = self.lock.lock();
2148        self.registry
2149            .state
2150            .lock()
2151            .attach_channel(owner, generation, channel)
2152    }
2153
2154    fn is_current(&self, owner: ChannelOwner, generation: u64) -> bool {
2155        let _guard = self.lock.lock();
2156        self.registry.state.lock().is_current(owner, generation)
2157    }
2158
2159    fn is_current_channel(&self, owner: ChannelOwner, generation: u64, channel: &str) -> bool {
2160        let _guard = self.lock.lock();
2161        self.registry
2162            .state
2163            .lock()
2164            .is_current_channel(owner, generation, channel)
2165    }
2166
2167    fn rollback(&self, owner: ChannelOwner, generation: u64) -> bool {
2168        let _guard = self.lock.lock();
2169        let mut state = self.registry.state.lock();
2170
2171        let Some(removed) = state.remove_if_generation(owner, generation) else {
2172            return false;
2173        };
2174
2175        self.dispatch.deactivate(owner, removed.channel_empty);
2176        if removed.channel_empty
2177            && let Some(channel) = &removed.channel
2178        {
2179            self.dispatch.book_sync.remove(channel);
2180        }
2181
2182        true
2183    }
2184
2185    fn remove(&self, owner: ChannelOwner) -> Option<RemovedSubscription> {
2186        let _guard = self.lock.lock();
2187        let mut state = self.registry.state.lock();
2188        let removed = state.remove(owner)?;
2189        self.dispatch.deactivate(owner, removed.channel_empty);
2190        if removed.channel_empty
2191            && let Some(channel) = &removed.channel
2192        {
2193            self.dispatch.book_sync.remove(channel);
2194        }
2195
2196        Some(removed)
2197    }
2198
2199    fn has_owners(&self, channel: &str) -> bool {
2200        let _guard = self.lock.lock();
2201        self.registry.state.lock().has_owners(channel)
2202    }
2203}
2204
2205impl SubscriptionDispatchState {
2206    fn activate(&self, owner: ChannelOwner, channel: Option<&str>) {
2207        match owner {
2208            ChannelOwner::BookDeltas(instrument_id) => {
2209                self.active_book_delta_channels.insert(
2210                    instrument_id,
2211                    channel.expect("book channel present").to_string(),
2212                );
2213            }
2214            ChannelOwner::BookDepth(instrument_id) => {
2215                self.active_book_depth_channels.insert(
2216                    instrument_id,
2217                    channel.expect("book channel present").to_string(),
2218                );
2219            }
2220            ChannelOwner::Ticker {
2221                instrument_id,
2222                feed,
2223            } => {
2224                self.active_ticker_channels.insert(
2225                    instrument_id,
2226                    channel.expect("ticker channel present").to_string(),
2227                );
2228                self.ticker_subscriptions(feed).insert(instrument_id);
2229            }
2230            ChannelOwner::Trades(instrument_id) => {
2231                self.active_trade_subs.insert(instrument_id);
2232            }
2233        }
2234    }
2235
2236    fn deactivate(&self, owner: ChannelOwner, channel_empty: bool) {
2237        match owner {
2238            ChannelOwner::BookDeltas(instrument_id) => {
2239                self.active_book_delta_channels.remove(&instrument_id);
2240            }
2241            ChannelOwner::BookDepth(instrument_id) => {
2242                self.active_book_depth_channels.remove(&instrument_id);
2243            }
2244            ChannelOwner::Ticker {
2245                instrument_id,
2246                feed,
2247            } => {
2248                self.ticker_subscriptions(feed).remove(&instrument_id);
2249                if feed == TickerFeed::Quote {
2250                    self.quote_cache.lock().remove(&instrument_id);
2251                }
2252
2253                if channel_empty {
2254                    self.active_ticker_channels.remove(&instrument_id);
2255                }
2256            }
2257            ChannelOwner::Trades(instrument_id) => {
2258                self.active_trade_subs.remove(&instrument_id);
2259            }
2260        }
2261    }
2262
2263    fn ticker_subscriptions(&self, feed: TickerFeed) -> &AtomicSet<InstrumentId> {
2264        match feed {
2265            TickerFeed::Quote => &self.active_quote_subs,
2266            TickerFeed::Mark => &self.active_mark_subs,
2267            TickerFeed::Index => &self.active_index_subs,
2268            TickerFeed::Funding => &self.active_funding_subs,
2269            TickerFeed::Greeks => &self.active_greeks_subs,
2270        }
2271    }
2272}
2273
2274impl ChannelSubscriptionRegistry {
2275    fn transition(&self, channel: &str) -> Arc<tokio::sync::Mutex<()>> {
2276        if self.transitions.len() >= TRANSITION_GC_THRESHOLD {
2277            self.transitions
2278                .retain(|_, transition| transition.strong_count() > 0);
2279        }
2280
2281        let mut entry = self.transitions.entry(channel.to_string()).or_default();
2282        if let Some(transition) = entry.value().upgrade() {
2283            return transition;
2284        }
2285
2286        let transition = Arc::new(tokio::sync::Mutex::new(()));
2287        *entry.value_mut() = Arc::downgrade(&transition);
2288        transition
2289    }
2290
2291    fn clear(&self) {
2292        let mut state = self.state.lock();
2293        let next_generation = state.next_generation;
2294
2295        *state = ChannelSubscriptionState {
2296            next_generation,
2297            ..Default::default()
2298        };
2299    }
2300
2301    fn clear_transitions(&self) {
2302        self.transitions.clear();
2303    }
2304}
2305
2306impl ChannelSubscriptionState {
2307    fn activate(&mut self, owner: ChannelOwner, channel: Option<&str>) -> Option<u64> {
2308        if self.owners.contains_key(&owner) {
2309            return None;
2310        }
2311
2312        self.next_generation = self
2313            .next_generation
2314            .checked_add(1)
2315            .expect("subscription generation overflow");
2316        let generation = self.next_generation;
2317
2318        if let Some(channel) = channel {
2319            self.channels
2320                .entry(channel.to_string())
2321                .or_default()
2322                .insert(owner);
2323        }
2324
2325        self.owners.insert(
2326            owner,
2327            OwnedChannel {
2328                generation,
2329                channel: channel.map(ToOwned::to_owned),
2330            },
2331        );
2332
2333        Some(generation)
2334    }
2335
2336    fn attach_channel(&mut self, owner: ChannelOwner, generation: u64, channel: String) -> bool {
2337        let Some(owned) = self.owners.get(&owner) else {
2338            return false;
2339        };
2340
2341        if owned.generation != generation {
2342            return false;
2343        }
2344
2345        if let Some(active_channel) = &owned.channel {
2346            return active_channel == &channel;
2347        }
2348
2349        self.owners.get_mut(&owner).expect("owner present").channel = Some(channel.clone());
2350        self.channels.entry(channel).or_default().insert(owner);
2351        true
2352    }
2353
2354    fn is_current(&self, owner: ChannelOwner, generation: u64) -> bool {
2355        self.owners
2356            .get(&owner)
2357            .is_some_and(|owned| owned.generation == generation)
2358    }
2359
2360    fn is_current_channel(&self, owner: ChannelOwner, generation: u64, channel: &str) -> bool {
2361        self.owners.get(&owner).is_some_and(|owned| {
2362            owned.generation == generation && owned.channel.as_deref() == Some(channel)
2363        })
2364    }
2365
2366    fn remove_if_generation(
2367        &mut self,
2368        owner: ChannelOwner,
2369        generation: u64,
2370    ) -> Option<RemovedSubscription> {
2371        if !self.is_current(owner, generation) {
2372            return None;
2373        }
2374
2375        self.remove(owner)
2376    }
2377
2378    fn remove(&mut self, owner: ChannelOwner) -> Option<RemovedSubscription> {
2379        let owned = self.owners.remove(&owner)?;
2380
2381        let channel_empty = owned.channel.as_ref().is_some_and(|channel| {
2382            let Some(owners) = self.channels.get_mut(channel) else {
2383                return true;
2384            };
2385
2386            owners.remove(&owner);
2387            owners.is_empty()
2388        });
2389
2390        if channel_empty && let Some(channel) = &owned.channel {
2391            self.channels.remove(channel);
2392        }
2393
2394        Some(RemovedSubscription {
2395            channel: owned.channel,
2396            channel_empty,
2397        })
2398    }
2399
2400    fn has_owners(&self, channel: &str) -> bool {
2401        self.channels
2402            .get(channel)
2403            .is_some_and(|owners| !owners.is_empty())
2404    }
2405}
2406
2407impl ChannelRequest {
2408    fn from_channel(channel: &str) -> anyhow::Result<Self> {
2409        if channel.starts_with("orderbook.") {
2410            let (instrument_name, group, depth) = orderbook_channel_parts(channel)?;
2411            return Ok(Self::Orderbook {
2412                channel: channel.to_string(),
2413                instrument_name,
2414                group,
2415                depth,
2416            });
2417        }
2418
2419        if channel.starts_with("ticker_slim.") || channel.starts_with("ticker.") {
2420            let (instrument_name, interval) = ticker_channel_parts(channel)?;
2421            return Ok(Self::Ticker {
2422                channel: channel.to_string(),
2423                instrument_name,
2424                interval,
2425            });
2426        }
2427
2428        if let Some((instrument_type, currency)) = channel
2429            .strip_prefix("trades.")
2430            .and_then(|value| value.split_once('.'))
2431        {
2432            return Ok(Self::Trades {
2433                channel: channel.to_string(),
2434                instrument_type: instrument_type.to_string(),
2435                currency: currency.to_string(),
2436            });
2437        }
2438
2439        anyhow::bail!("invalid Derive subscription channel `{channel}`")
2440    }
2441
2442    fn channel(&self) -> &str {
2443        match self {
2444            Self::Orderbook { channel, .. }
2445            | Self::Ticker { channel, .. }
2446            | Self::Trades { channel, .. } => channel,
2447        }
2448    }
2449
2450    async fn subscribe(&self, ws: &DeriveWebSocketSubscriptionHandle) -> Result<(), DeriveWsError> {
2451        match self {
2452            Self::Orderbook {
2453                instrument_name,
2454                group,
2455                depth,
2456                ..
2457            } => {
2458                ws.subscribe_orderbook(instrument_name, group, depth)
2459                    .await?;
2460            }
2461            Self::Ticker {
2462                instrument_name,
2463                interval,
2464                ..
2465            } => ws.subscribe_ticker(instrument_name, interval).await?,
2466            Self::Trades {
2467                instrument_type,
2468                currency,
2469                ..
2470            } => ws.subscribe_trades(instrument_type, currency).await?,
2471        }
2472
2473        Ok(())
2474    }
2475
2476    async fn unsubscribe(
2477        &self,
2478        ws: &DeriveWebSocketSubscriptionHandle,
2479    ) -> Result<(), DeriveWsError> {
2480        match self {
2481            Self::Orderbook {
2482                instrument_name,
2483                group,
2484                depth,
2485                ..
2486            } => {
2487                ws.unsubscribe_orderbook(instrument_name, group, depth)
2488                    .await?;
2489            }
2490            Self::Ticker {
2491                instrument_name,
2492                interval,
2493                ..
2494            } => ws.unsubscribe_ticker(instrument_name, interval).await?,
2495            Self::Trades {
2496                instrument_type,
2497                currency,
2498                ..
2499            } => ws.unsubscribe_trades(instrument_type, currency).await?,
2500        }
2501
2502        Ok(())
2503    }
2504}
2505
2506fn retain_channel_for_reconnect(
2507    ws: &DeriveWebSocketSubscriptionHandle,
2508    channel: &str,
2509    has_surviving_owner: bool,
2510    cleanup_error: Option<&DeriveWsError>,
2511) -> bool {
2512    let replay = has_surviving_owner
2513        && cleanup_error.is_some_and(|e| {
2514            matches!(
2515                e,
2516                DeriveWsError::Transport(_)
2517                    | DeriveWsError::RequestCancelled { .. }
2518                    | DeriveWsError::Timeout { .. }
2519                    | DeriveWsError::NotConnected
2520            )
2521        });
2522
2523    if replay {
2524        ws.remember_subscription(channel);
2525    } else {
2526        ws.forget_subscription(channel);
2527    }
2528
2529    replay
2530}
2531
2532impl DeriveDataClient {
2533    async fn join_session_tasks(&self) -> anyhow::Result<()> {
2534        self.session_tasks.begin_shutdown();
2535        self.session_tasks
2536            .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
2537            .await
2538            .map_err(|e| anyhow::anyhow!("Failed to terminate Derive data session tasks: {e}"))?;
2539        Ok(())
2540    }
2541
2542    async fn join_pending_tasks(&self) -> anyhow::Result<()> {
2543        self.pending_tasks.begin_shutdown();
2544        self.pending_tasks
2545            .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
2546            .await
2547            .map_err(|e| anyhow::anyhow!("Failed to terminate Derive data tasks: {e}"))?;
2548        Ok(())
2549    }
2550}
2551
2552fn cache_instrument(
2553    instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
2554    instrument: &InstrumentAny,
2555) {
2556    instruments.insert(instrument.id(), instrument.clone());
2557}
2558
2559fn process_ticker_quote(
2560    msg: &DeriveTickerMsg,
2561    price_precision: u8,
2562    size_precision: u8,
2563    ts_init: UnixNanos,
2564    quote_cache: &mut QuoteCache,
2565) -> anyhow::Result<Option<QuoteTick>> {
2566    let quote = parse_ticker_quote(msg, price_precision, size_precision, ts_init)?;
2567    if quote.bid_price.is_zero()
2568        || quote.ask_price.is_zero()
2569        || quote.bid_size.is_zero()
2570        || quote.ask_size.is_zero()
2571    {
2572        // Tickers are complete snapshots; a zero side withdraws cached liquidity
2573        quote_cache.remove(&quote.instrument_id);
2574
2575        log::debug!(
2576            "Skipping Derive ticker quote with an empty side for {}",
2577            msg.data.instrument_name(),
2578        );
2579        return Ok(None);
2580    }
2581
2582    quote_cache.insert(quote.instrument_id, quote);
2583    Ok(Some(quote))
2584}
2585
2586fn channel_is_active(
2587    channels: &AtomicMap<InstrumentId, String>,
2588    instrument_id: InstrumentId,
2589    channel: &str,
2590) -> bool {
2591    channels
2592        .get_cloned(&instrument_id)
2593        .is_some_and(|active_channel| active_channel == channel)
2594}
2595
2596fn orderbook_channel_parts(channel: &str) -> anyhow::Result<(String, String, String)> {
2597    let rest = channel
2598        .strip_prefix("orderbook.")
2599        .ok_or_else(|| anyhow::anyhow!("invalid Derive orderbook channel `{channel}`"))?;
2600    let mut parts = rest.rsplitn(3, '.');
2601    let depth = parts
2602        .next()
2603        .filter(|value| !value.is_empty())
2604        .ok_or_else(|| anyhow::anyhow!("invalid Derive orderbook channel `{channel}`"))?;
2605    let group = parts
2606        .next()
2607        .filter(|value| !value.is_empty())
2608        .ok_or_else(|| anyhow::anyhow!("invalid Derive orderbook channel `{channel}`"))?;
2609    let instrument_name = parts
2610        .next()
2611        .filter(|value| !value.is_empty())
2612        .ok_or_else(|| anyhow::anyhow!("invalid Derive orderbook channel `{channel}`"))?;
2613
2614    Ok((
2615        instrument_name.to_string(),
2616        group.to_string(),
2617        depth.to_string(),
2618    ))
2619}
2620
2621fn ticker_channel_parts(channel: &str) -> anyhow::Result<(String, String)> {
2622    let rest = channel
2623        .strip_prefix("ticker_slim.")
2624        .or_else(|| channel.strip_prefix("ticker."))
2625        .ok_or_else(|| anyhow::anyhow!("invalid Derive ticker channel `{channel}`"))?;
2626    let (instrument_name, interval) = rest
2627        .rsplit_once('.')
2628        .ok_or_else(|| anyhow::anyhow!("invalid Derive ticker channel `{channel}`"))?;
2629    anyhow::ensure!(
2630        !instrument_name.is_empty() && !interval.is_empty(),
2631        "invalid Derive ticker channel `{channel}`"
2632    );
2633
2634    Ok((instrument_name.to_string(), interval.to_string()))
2635}
2636
2637fn orderbook_group(params: &Option<Params>) -> anyhow::Result<String> {
2638    let group = params
2639        .as_ref()
2640        .and_then(|p| {
2641            p.get_str("group")
2642                .map(ToOwned::to_owned)
2643                .or_else(|| p.get_u64("group").map(|value| value.to_string()))
2644        })
2645        .unwrap_or_else(|| DEFAULT_ORDERBOOK_GROUP.to_string());
2646
2647    DeriveOrderbookGroup::from_str(&group)
2648        .with_context(|| format!("invalid Derive orderbook group `{group}`"))?;
2649    Ok(group)
2650}
2651
2652fn orderbook_depth(depth: Option<usize>, params: &Option<Params>) -> anyhow::Result<String> {
2653    let depth = depth
2654        .map(|value| value.to_string())
2655        .or_else(|| {
2656            params.as_ref().and_then(|p| {
2657                p.get_str("depth")
2658                    .map(ToOwned::to_owned)
2659                    .or_else(|| p.get_u64("depth").map(|value| value.to_string()))
2660            })
2661        })
2662        .unwrap_or_else(|| DEFAULT_ORDERBOOK_DEPTH.to_string());
2663
2664    DeriveOrderbookDepth::from_str(&depth)
2665        .with_context(|| format!("invalid Derive orderbook depth `{depth}`"))?;
2666    Ok(depth)
2667}
2668
2669fn ticker_interval(params: &Option<Params>) -> anyhow::Result<String> {
2670    let interval = params
2671        .as_ref()
2672        .and_then(|p| {
2673            p.get_str("interval")
2674                .map(ToOwned::to_owned)
2675                .or_else(|| p.get_u64("interval").map(|value| value.to_string()))
2676        })
2677        .unwrap_or_else(|| DEFAULT_TICKER_INTERVAL.to_string());
2678
2679    DeriveTickerInterval::from_str(&interval)
2680        .with_context(|| format!("invalid Derive ticker interval `{interval}`"))?;
2681    Ok(interval)
2682}
2683
2684fn trade_channel(instrument: &InstrumentAny) -> anyhow::Result<String> {
2685    let instrument_type = derive_instrument_type(instrument)?.to_string();
2686    let instrument_id = instrument.id();
2687    let currency = currency_from_instrument_id(&instrument_id)?;
2688    Ok(trades_channel(&instrument_type, currency))
2689}
2690
2691fn derive_instrument_type(instrument: &InstrumentAny) -> anyhow::Result<DeriveInstrumentType> {
2692    match instrument {
2693        InstrumentAny::CryptoPerpetual(_) => Ok(DeriveInstrumentType::Perp),
2694        InstrumentAny::CryptoOption(_) => Ok(DeriveInstrumentType::Option),
2695        InstrumentAny::CurrencyPair(_) => Ok(DeriveInstrumentType::Erc20),
2696        other => anyhow::bail!("unsupported Derive instrument type for trades: {other:?}"),
2697    }
2698}
2699
2700fn currency_from_instrument_id(instrument_id: &InstrumentId) -> anyhow::Result<&str> {
2701    anyhow::ensure!(
2702        instrument_id.venue == *DERIVE_VENUE,
2703        "instrument ID `{instrument_id}` is not for venue {}",
2704        DERIVE_VENUE.as_str(),
2705    );
2706
2707    instrument_id
2708        .symbol
2709        .as_str()
2710        .split_once('-')
2711        .and_then(|(currency, _)| (!currency.is_empty()).then_some(currency))
2712        .ok_or_else(|| anyhow::anyhow!("cannot derive currency from {instrument_id}"))
2713}
2714
2715// Caps the rendered JSON at ~512 bytes for log grep-ability and backs the
2716// slice off to a UTF-8 char boundary so a multi-byte codepoint near the cap
2717// can never produce a panicking slice.
2718fn truncated_payload_snippet(raw: &str) -> String {
2719    const MAX_LEN: usize = 512;
2720
2721    if raw.len() <= MAX_LEN {
2722        return raw.to_string();
2723    }
2724
2725    let mut end = MAX_LEN;
2726    while end > 0 && !raw.is_char_boundary(end) {
2727        end -= 1;
2728    }
2729
2730    format!("{}...(truncated)", &raw[..end])
2731}
2732
2733#[cfg(test)]
2734mod tests {
2735    use std::{path::PathBuf, time::Duration};
2736
2737    use nautilus_common::{live::runner::replace_data_event_sender, testing::wait_until_async};
2738    use nautilus_core::{UUID4, UnixNanos};
2739    use nautilus_model::{
2740        enums::AggressorSide,
2741        identifiers::InstrumentId,
2742        types::{Price, Quantity},
2743    };
2744    use nautilus_network::websocket::TransportBackend;
2745    use rstest::rstest;
2746    use rust_decimal_macros::dec;
2747    use serde_json::{Value, json};
2748
2749    use super::*;
2750    use crate::{
2751        client_logging,
2752        common::{
2753            consts::DERIVE_CLIENT_ID, enums::DeriveEnvironment, parse::parse_derive_instrument_any,
2754            urls,
2755        },
2756        http::models::DeriveInstrument,
2757        websocket::{DeriveWsFrame, WsSubscriptionPayload, parse_orderbook_msg, parse_ticker_msg},
2758    };
2759
2760    fn data_path() -> PathBuf {
2761        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("test_data")
2762    }
2763
2764    fn load_json(filename: &str) -> Value {
2765        let content = std::fs::read_to_string(data_path().join(filename))
2766            .unwrap_or_else(|_| panic!("failed to read {filename}"));
2767        serde_json::from_str(&content).expect("invalid json")
2768    }
2769
2770    #[tokio::test]
2771    async fn test_data_task_log_omits_provider_error() {
2772        let _capture = client_logging::capture_logs().await;
2773        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
2774        replace_data_event_sender(sender);
2775        let client =
2776            DeriveDataClient::new(*DERIVE_CLIENT_ID, DeriveDataClientConfig::default()).unwrap();
2777        client.spawn_task("Synthetic subscription", async {
2778            anyhow::bail!("SYN-secret-provider-error")
2779        });
2780
2781        wait_until_async(
2782            || async {
2783                client_logging::messages().iter().any(|(level, message)| {
2784                    *level == log::Level::Warn && message == "Synthetic subscription failed"
2785                })
2786            },
2787            Duration::from_secs(2),
2788        )
2789        .await;
2790
2791        client.join_pending_tasks().await.unwrap();
2792        let messages = client_logging::messages();
2793        assert!(messages.contains(&(
2794            log::Level::Warn,
2795            "Synthetic subscription failed".to_string()
2796        )));
2797        assert!(
2798            !messages
2799                .iter()
2800                .any(|(_, message)| message.contains("SYN-secret"))
2801        );
2802    }
2803
2804    #[tokio::test]
2805    async fn test_session_failure_log_omits_provider_reason() {
2806        let _capture = client_logging::capture_logs().await;
2807        let (ctx, _receiver) = make_ctx(None);
2808        DeriveDataClient::handle_ws_message(
2809            DeriveWsMessage::SessionRecoveryFailed("SYN-secret-provider-reason".to_string()),
2810            &ctx,
2811        );
2812        assert_eq!(
2813            client_logging::messages()
2814                .into_iter()
2815                .filter(|(_, message)| {
2816                    message.starts_with("Derive WebSocket session recovery failed")
2817                })
2818                .collect::<Vec<_>>(),
2819            vec![(
2820                log::Level::Error,
2821                "Derive WebSocket session recovery failed".to_string()
2822            )]
2823        );
2824    }
2825
2826    #[tokio::test]
2827    async fn test_public_stream_closure_marks_disconnected() {
2828        let (sender, mut events) = tokio::sync::mpsc::unbounded_channel();
2829        replace_data_event_sender(sender);
2830        let client =
2831            DeriveDataClient::new(*DERIVE_CLIENT_ID, DeriveDataClientConfig::default()).unwrap();
2832        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
2833        client.is_connected.store(true, Ordering::Release);
2834        client.spawn_stream_task(rx).unwrap();
2835        assert!(client.is_connected());
2836        drop(tx);
2837        client.session_tasks.begin_shutdown();
2838        client
2839            .session_tasks
2840            .finish_shutdown(Duration::from_secs(1), Duration::from_secs(1))
2841            .await
2842            .unwrap();
2843        assert!(!client.is_connected());
2844        assert!(events.try_recv().is_err());
2845    }
2846
2847    #[rstest]
2848    fn test_truncated_payload_snippet_short_payload_is_unchanged() {
2849        let short = json!({"ok": true}).to_string();
2850        assert_eq!(truncated_payload_snippet(&short), r#"{"ok":true}"#);
2851    }
2852
2853    #[rstest]
2854    fn test_truncated_payload_snippet_truncates_long_ascii_payload() {
2855        let big = json!({"msg": "x".repeat(1024)}).to_string();
2856        let snippet = truncated_payload_snippet(&big);
2857        assert!(snippet.ends_with("...(truncated)"));
2858        // Body before the suffix capped near the 512-byte target.
2859        assert!(snippet.len() <= 512 + "...(truncated)".len());
2860    }
2861
2862    #[rstest]
2863    fn test_truncated_payload_snippet_handles_multibyte_at_boundary() {
2864        // Construct a payload where MAX_LEN (512) lands *inside* a UTF-8
2865        // codepoint. `{"a":"x` is 7 bytes (odd-aligned), each 'e' with acute
2866        // accent (U+00E9) is 2 bytes, and 7 + 252*2 = 511, so byte 512 falls
2867        // inside the 253rd accented char. Under the old fixed-byte slice
2868        // `&raw[..512]` would panic with "byte index 512 is not a char
2869        // boundary"; with the backoff it must return a valid `String`
2870        // strictly shorter than 512 bytes for the body.
2871        let value: String = format!("x{}", "\u{00E9}".repeat(1024));
2872        let big = json!({"a": value});
2873        let raw = big.to_string();
2874        assert!(
2875            !raw.is_char_boundary(512),
2876            "test premise: 512 must be mid-codepoint",
2877        );
2878
2879        let snippet = truncated_payload_snippet(&raw);
2880        assert!(snippet.ends_with("...(truncated)"));
2881        let body_len = snippet.len() - "...(truncated)".len();
2882        assert!(body_len <= 512);
2883    }
2884
2885    fn subscription_payload(channel: &str, data: &Value) -> WsSubscriptionPayload {
2886        let frame = json!({
2887            "jsonrpc": "2.0",
2888            "method": "subscription",
2889            "params": {
2890                "channel": channel,
2891                "data": data
2892            }
2893        });
2894
2895        match DeriveWsFrame::parse(&frame.to_string()).unwrap() {
2896            DeriveWsFrame::Subscription(payload) => payload,
2897            other => panic!("expected subscription frame, was {other:?}"),
2898        }
2899    }
2900
2901    fn make_ctx(
2902        instrument: Option<InstrumentAny>,
2903    ) -> (
2904        WsMessageContext,
2905        tokio::sync::mpsc::UnboundedReceiver<DataEvent>,
2906    ) {
2907        let (data_sender, data_rx) = tokio::sync::mpsc::unbounded_channel();
2908        let instruments = Arc::new(AtomicMap::new());
2909        if let Some(instrument) = instrument {
2910            cache_instrument(&instruments, &instrument);
2911        }
2912
2913        (
2914            WsMessageContext {
2915                clock: get_atomic_clock_realtime(),
2916                data_sender: data_sender.into(),
2917                instruments,
2918                active_book_delta_channels: Arc::new(AtomicMap::new()),
2919                active_book_depth_channels: Arc::new(AtomicMap::new()),
2920                active_ticker_channels: Arc::new(AtomicMap::new()),
2921                active_quote_subs: Arc::new(AtomicSet::new()),
2922                active_trade_subs: Arc::new(AtomicSet::new()),
2923                active_mark_subs: Arc::new(AtomicSet::new()),
2924                active_index_subs: Arc::new(AtomicSet::new()),
2925                active_funding_subs: Arc::new(AtomicSet::new()),
2926                active_greeks_subs: Arc::new(AtomicSet::new()),
2927                subscription_lock: Arc::new(Mutex::new(())),
2928                book_sync: {
2929                    let tracker = BookSyncTracker::default();
2930                    let id = InstrumentId::from("ETH-PERP.DERIVE");
2931                    tracker.record_subscription("orderbook.ETH-PERP.1.10", id);
2932                    tracker
2933                        .write("orderbook.ETH-PERP.1.10")
2934                        .unwrap()
2935                        .gate
2936                        .open();
2937                    tracker
2938                },
2939                quote_cache: Arc::new(Mutex::new(QuoteCache::new())),
2940                trades_emitted: Arc::new(Mutex::new(FifoCache::new())),
2941            },
2942            data_rx,
2943        )
2944    }
2945
2946    fn perp_instrument() -> InstrumentAny {
2947        parse_derive_instrument_any(&perp_definition("ETH-PERP", "ETH"), UnixNanos::from(1))
2948            .unwrap()
2949            .unwrap()
2950    }
2951
2952    fn btc_perp_instrument() -> InstrumentAny {
2953        parse_derive_instrument_any(&perp_definition("BTC-PERP", "BTC"), UnixNanos::from(1))
2954            .unwrap()
2955            .unwrap()
2956    }
2957
2958    fn perp_definition(name: &str, currency: &str) -> DeriveInstrument {
2959        let mut value = load_json("perps/instrument_eth.json");
2960        value["base_currency"] = json!(currency);
2961        value["instrument_name"] = json!(name);
2962        value["perp_details"]["index"] = json!(format!("{currency}-USD"));
2963
2964        serde_json::from_value(value).unwrap()
2965    }
2966
2967    fn option_instrument() -> InstrumentAny {
2968        let definition: DeriveInstrument =
2969            serde_json::from_value(load_json("options/instrument_eth.json")).unwrap();
2970        parse_derive_instrument_any(&definition, UnixNanos::from(1))
2971            .unwrap()
2972            .unwrap()
2973    }
2974
2975    fn spot_instrument() -> InstrumentAny {
2976        let definition: DeriveInstrument =
2977            serde_json::from_value(load_json("spot/instrument_eth.json")).unwrap();
2978        parse_derive_instrument_any(&definition, UnixNanos::from(1))
2979            .unwrap()
2980            .unwrap()
2981    }
2982
2983    fn ticker_json(timestamp: i64) -> Value {
2984        let mut value = load_json("perps/ws_ticker_eth.json");
2985        value["timestamp"] = json!(timestamp);
2986        value
2987    }
2988
2989    fn option_ticker_json(timestamp: i64) -> Value {
2990        let mut value = load_json("options/http_ticker_eth_snapshot.json");
2991        value["timestamp"] = json!(timestamp);
2992        value
2993    }
2994
2995    fn spot_ticker_slim_json() -> Value {
2996        load_json("spot/ws_ticker_slim_eth.json")
2997    }
2998
2999    fn orderbook_json() -> Value {
3000        load_json("perps/ws_orderbook_eth.json")
3001    }
3002
3003    fn trade_json(instrument_name: &str, trade_id: &str) -> Value {
3004        let mut value = load_json("perps/ws_trade_eth.json");
3005        value["instrument_name"] = json!(instrument_name);
3006        value["trade_id"] = json!(trade_id);
3007        value
3008    }
3009
3010    #[rstest]
3011    fn test_handle_ticker_subscription_emits_quote_with_instrument_precision() {
3012        let instrument = perp_instrument();
3013        let instrument_id = instrument.id();
3014        let (ctx, mut rx) = make_ctx(Some(instrument));
3015        ctx.active_ticker_channels
3016            .insert(instrument_id, "ticker_slim.ETH-PERP.1000".to_string());
3017        ctx.active_quote_subs.insert(instrument_id);
3018        let payload = subscription_payload(
3019            "ticker_slim.ETH-PERP.1000",
3020            &json!({
3021                "timestamp": 1_700_000_000_010_i64,
3022                "instrument_ticker": ticker_json(1_700_000_000_000)
3023            }),
3024        );
3025
3026        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3027
3028        match rx.try_recv().unwrap() {
3029            DataEvent::Data(Data::Quote(quote)) => {
3030                assert_eq!(quote.instrument_id, instrument_id);
3031                assert_eq!(quote.bid_price, Price::from("3500.00"));
3032                assert_eq!(quote.ask_price, Price::from("3501.00"));
3033                assert_eq!(quote.bid_size, Quantity::from("1.000"));
3034                assert_eq!(quote.ask_size, Quantity::from("2.000"));
3035                assert_eq!(quote.bid_price.precision, 2);
3036                assert_eq!(quote.bid_size.precision, 3);
3037            }
3038            other => panic!("expected quote data event, was {other:?}"),
3039        }
3040    }
3041
3042    #[rstest]
3043    fn test_handle_ticker_partial_quote_without_cache_emits_no_quote() {
3044        let instrument = spot_instrument();
3045        let instrument_id = instrument.id();
3046        let channel = "ticker_slim.ETH-USDC.1000";
3047        let (ctx, mut rx) = make_ctx(Some(instrument));
3048        install_ticker(&ctx, instrument_id, channel);
3049        ctx.active_quote_subs.insert(instrument_id);
3050        let payload = subscription_payload(channel, &spot_ticker_slim_json());
3051
3052        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3053
3054        assert!(rx.try_recv().is_err());
3055    }
3056
3057    #[rstest]
3058    fn test_handle_ticker_empty_side_clears_cached_quote() {
3059        let instrument = spot_instrument();
3060        let instrument_id = instrument.id();
3061        let channel = "ticker_slim.ETH-USDC.1000";
3062        let (ctx, mut rx) = make_ctx(Some(instrument));
3063
3064        let cached_quote = QuoteTick::new(
3065            instrument_id,
3066            Price::from("0.1"),
3067            Price::from("0.3"),
3068            Quantity::from("10.00"),
3069            Quantity::from("20.00"),
3070            UnixNanos::from(1),
3071            UnixNanos::from(1),
3072        );
3073        ctx.quote_cache.lock().insert(instrument_id, cached_quote);
3074        install_ticker(&ctx, instrument_id, channel);
3075        ctx.active_quote_subs.insert(instrument_id);
3076        let payload = subscription_payload(channel, &spot_ticker_slim_json());
3077
3078        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3079
3080        assert!(rx.try_recv().is_err());
3081        assert_eq!(ctx.quote_cache.lock().get(&instrument_id), None);
3082    }
3083
3084    #[rstest]
3085    #[case::bid_price("0", "0.4", "45", "67")]
3086    #[case::bid_size("0.2", "0.4", "0", "67")]
3087    #[case::ask_price("0.2", "0", "45", "67")]
3088    #[case::ask_size("0.2", "0.4", "45", "0")]
3089    #[case::both_empty("0", "0", "0", "0")]
3090    fn test_process_ticker_empty_sides_and_recovery(
3091        #[case] bid_price: &str,
3092        #[case] ask_price: &str,
3093        #[case] bid_size: &str,
3094        #[case] ask_size: &str,
3095        #[values(false, true)] populated: bool,
3096        #[values(false, true)] slim: bool,
3097    ) {
3098        let (instrument, channel, mut data) = if slim {
3099            (
3100                spot_instrument(),
3101                "ticker_slim.ETH-USDC.1000",
3102                spot_ticker_slim_json(),
3103            )
3104        } else {
3105            (perp_instrument(), "ticker.ETH-PERP.1000", ticker_json(1234))
3106        };
3107
3108        let instrument_id = instrument.id();
3109        let mut quote_cache = QuoteCache::new();
3110        let ts_init = UnixNanos::from(9_876_543_210);
3111
3112        let expected = QuoteTick::new(
3113            instrument_id,
3114            Price::from_decimal_dp(dec!(0.2), instrument.price_precision()).unwrap(),
3115            Price::from_decimal_dp(dec!(0.4), instrument.price_precision()).unwrap(),
3116            Quantity::from_decimal_dp(dec!(45), instrument.size_precision()).unwrap(),
3117            Quantity::from_decimal_dp(dec!(67), instrument.size_precision()).unwrap(),
3118            UnixNanos::from(1_238_000_000),
3119            ts_init,
3120        );
3121
3122        let unrelated_quote = QuoteTick {
3123            instrument_id: InstrumentId::from("BTC-PERP.DERIVE"),
3124            ..expected
3125        };
3126
3127        quote_cache.insert(unrelated_quote.instrument_id, unrelated_quote);
3128        if populated {
3129            quote_cache.insert(instrument_id, expected);
3130        }
3131
3132        let keys = if slim {
3133            ["b", "a", "B", "A"]
3134        } else {
3135            [
3136                "best_bid_price",
3137                "best_ask_price",
3138                "best_bid_amount",
3139                "best_ask_amount",
3140            ]
3141        };
3142
3143        for (values, timestamp, expected_result) in [
3144            ([bid_price, ask_price, bid_size, ask_size], 1235, None),
3145            (["0.6", "0", "89", "0"], 1236, None),
3146            (["0", "0.8", "0", "91"], 1237, None),
3147            (["0.2", "0.4", "45", "67"], 1238, Some(expected)),
3148        ] {
3149            data["timestamp"] = json!(timestamp);
3150            let snapshot = if slim {
3151                &mut data["instrument_ticker"]
3152            } else {
3153                &mut data
3154            };
3155
3156            for (key, value) in keys.into_iter().zip(values) {
3157                snapshot[key] = json!(value);
3158            }
3159
3160            let payload = subscription_payload(channel, &data);
3161            let msg = parse_ticker_msg(&payload).unwrap();
3162            let result = process_ticker_quote(
3163                &msg,
3164                instrument.price_precision(),
3165                instrument.size_precision(),
3166                ts_init,
3167                &mut quote_cache,
3168            )
3169            .unwrap();
3170
3171            assert_eq!(result, expected_result);
3172            assert_eq!(quote_cache.get(&instrument_id), expected_result.as_ref());
3173            assert_eq!(
3174                quote_cache.get(&unrelated_quote.instrument_id),
3175                Some(&unrelated_quote)
3176            );
3177        }
3178    }
3179
3180    #[rstest]
3181    #[case::testnet_spot(DeriveEnvironment::Testnet, "ETH-USDC")]
3182    #[case::testnet_perp(DeriveEnvironment::Testnet, "ETH-PERP")]
3183    #[case::mainnet_spot(DeriveEnvironment::Mainnet, "ETH-USDC")]
3184    #[case::mainnet_perp(DeriveEnvironment::Mainnet, "ETH-PERP")]
3185    #[tokio::test]
3186    #[ignore = "requires Derive public network access; mainnet requires an isolated NZ route"]
3187    async fn test_live_ticker_complete_snapshots(
3188        #[case] environment: DeriveEnvironment,
3189        #[case] instrument_name: &str,
3190    ) {
3191        let http =
3192            DeriveHttpClient::new(urls::rest_url(environment), Some(15), None, None).unwrap();
3193        let definition = http.get_instrument(instrument_name).await.unwrap();
3194        let instrument = parse_derive_instrument_any(&definition, UnixNanos::from(1))
3195            .unwrap()
3196            .unwrap();
3197        let instrument_id = instrument.id();
3198        let mut ws_client =
3199            DeriveWebSocketClient::new(None, environment, TransportBackend::default(), None);
3200        ws_client.connect().await.unwrap();
3201        let mut events = ws_client.take_event_receiver().unwrap();
3202        let validation = async {
3203            ws_client
3204                .subscribe_ticker(instrument_name, "1000")
3205                .await
3206                .unwrap();
3207            let mut quote_cache = QuoteCache::new();
3208
3209            let seed = QuoteTick::new(
3210                instrument_id,
3211                Price::from("0.1"),
3212                Price::from("0.3"),
3213                Quantity::from("10.00"),
3214                Quantity::from("20.00"),
3215                UnixNanos::from(1),
3216                UnixNanos::from(1),
3217            );
3218            quote_cache.insert(instrument_id, seed);
3219            let mut published = 0;
3220            let mut suppressed = 0;
3221
3222            for _ in 0..3 {
3223                let payload = loop {
3224                    if let DeriveWsMessage::Subscription(payload) = events.recv().await.unwrap() {
3225                        break payload;
3226                    }
3227                };
3228
3229                let msg = parse_ticker_msg(&payload).unwrap();
3230                let ts_init = UnixNanos::from(9_876_543_210);
3231                let expected = parse_ticker_quote(
3232                    &msg,
3233                    instrument.price_precision(),
3234                    instrument.size_precision(),
3235                    ts_init,
3236                )
3237                .unwrap();
3238                let result = process_ticker_quote(
3239                    &msg,
3240                    instrument.price_precision(),
3241                    instrument.size_precision(),
3242                    ts_init,
3243                    &mut quote_cache,
3244                )
3245                .unwrap();
3246
3247                if expected.bid_price.is_zero()
3248                    || expected.ask_price.is_zero()
3249                    || expected.bid_size.is_zero()
3250                    || expected.ask_size.is_zero()
3251                {
3252                    assert_eq!(result, None);
3253                    assert_eq!(quote_cache.get(&instrument_id), None);
3254                    suppressed += 1;
3255                } else {
3256                    assert_eq!(result, Some(expected));
3257                    assert_eq!(quote_cache.get(&instrument_id), Some(&expected));
3258                    assert_eq!(expected.instrument_id, instrument_id);
3259                    assert_eq!(
3260                        expected.ts_event,
3261                        UnixNanos::from(msg.data.timestamp() as u64 * 1_000_000)
3262                    );
3263                    assert_eq!(expected.ts_init, ts_init);
3264                    published += 1;
3265                }
3266            }
3267
3268            ws_client
3269                .unsubscribe_ticker(instrument_name, "1000")
3270                .await
3271                .unwrap();
3272            (published, suppressed)
3273        };
3274
3275        let result = tokio::time::timeout(Duration::from_secs(30), validation).await;
3276        tokio::time::timeout(Duration::from_secs(5), ws_client.disconnect())
3277            .await
3278            .expect("ticker validation shutdown exceeds five seconds")
3279            .unwrap();
3280        let (published, suppressed) = result.expect("ticker validation exceeds 30 seconds");
3281        println!(
3282            "Derive {environment:?} {instrument_name}: {published} complete quotes, {suppressed} empty snapshots suppressed; unsubscribed and disconnected"
3283        );
3284    }
3285
3286    #[rstest]
3287    fn test_handle_reconnected_clears_quote_cache() {
3288        let instrument = spot_instrument();
3289        let instrument_id = instrument.id();
3290        let channel = "ticker_slim.ETH-USDC.1000";
3291        let (ctx, mut rx) = make_ctx(Some(instrument));
3292
3293        let cached_quote = QuoteTick::new(
3294            instrument_id,
3295            Price::from("0.1"),
3296            Price::from("0.3"),
3297            Quantity::from("10.00"),
3298            Quantity::from("20.00"),
3299            UnixNanos::from(1),
3300            UnixNanos::from(1),
3301        );
3302        ctx.quote_cache.lock().insert(instrument_id, cached_quote);
3303        install_ticker(&ctx, instrument_id, channel);
3304        ctx.active_quote_subs.insert(instrument_id);
3305
3306        DeriveDataClient::handle_ws_message(DeriveWsMessage::Reconnected, &ctx);
3307
3308        assert_eq!(ctx.quote_cache.lock().get(&instrument_id), None);
3309
3310        let payload = subscription_payload(channel, &spot_ticker_slim_json());
3311        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3312
3313        assert!(rx.try_recv().is_err());
3314    }
3315
3316    #[rstest]
3317    #[tokio::test]
3318    async fn test_session_recovery_failure_marks_client_disconnected() {
3319        let _capture = client_logging::capture_logs().await;
3320        let (data_tx, _data_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3321        replace_data_event_sender(data_tx);
3322
3323        let config = DeriveDataClientConfig {
3324            environment: DeriveEnvironment::Mainnet,
3325            ..Default::default()
3326        };
3327
3328        let client = DeriveDataClient::new(*DERIVE_CLIENT_ID, config).unwrap();
3329        let (ws_tx, ws_rx) = tokio::sync::mpsc::unbounded_channel();
3330        client.is_connected.store(true, Ordering::Release);
3331        client.spawn_stream_task(ws_rx).unwrap();
3332
3333        ws_tx
3334            .send(DeriveWsMessage::SessionRecoveryFailed(
3335                "subscription replay failed".to_string(),
3336            ))
3337            .unwrap();
3338        wait_until_async(|| async { !client.is_connected() }, Duration::from_secs(2)).await;
3339
3340        assert!(!client.is_connected());
3341
3342        client.cancellation_token.cancel();
3343        client.join_session_tasks().await.unwrap();
3344    }
3345
3346    #[rstest]
3347    fn test_handle_orderbook_subscription_emits_snapshot_deltas() {
3348        let instrument = perp_instrument();
3349        let instrument_id = instrument.id();
3350        let (ctx, mut rx) = make_ctx(Some(instrument));
3351        ctx.active_book_delta_channels
3352            .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
3353        let payload = subscription_payload("orderbook.ETH-PERP.1.10", &orderbook_json());
3354
3355        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3356
3357        match rx.try_recv().unwrap() {
3358            DataEvent::Data(Data::BookDeltas(deltas)) => {
3359                assert_eq!(deltas.instrument_id, instrument_id);
3360                assert_eq!(deltas.deltas.len(), 3);
3361                assert_eq!(deltas.deltas[1].order.price, Price::from("3500.00"));
3362                assert_eq!(deltas.deltas[1].order.size, Quantity::from("1.000"));
3363                assert_eq!(deltas.deltas[2].order.price, Price::from("3501.00"));
3364                assert_eq!(deltas.deltas[2].order.size, Quantity::from("2.000"));
3365            }
3366            other => panic!("expected deltas data event, was {other:?}"),
3367        }
3368    }
3369
3370    #[rstest]
3371    fn test_handle_orderbook_subscription_emits_for_depth_subscription() {
3372        let instrument = perp_instrument();
3373        let instrument_id = instrument.id();
3374        let (ctx, mut rx) = make_ctx(Some(instrument));
3375        ctx.active_book_depth_channels
3376            .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
3377        let payload = subscription_payload("orderbook.ETH-PERP.1.10", &orderbook_json());
3378
3379        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3380
3381        match rx.try_recv().unwrap() {
3382            DataEvent::Data(Data::BookDepth(depth)) => {
3383                assert_eq!(depth.instrument_id, instrument_id);
3384                assert_eq!(depth.bids[0].price, Price::from("3500.00"));
3385                assert_eq!(depth.bids[0].size, Quantity::from("1.000"));
3386                assert_eq!(depth.asks[0].price, Price::from("3501.00"));
3387                assert_eq!(depth.asks[0].size, Quantity::from("2.000"));
3388            }
3389            other => panic!("expected depth data event, was {other:?}"),
3390        }
3391    }
3392
3393    #[rstest]
3394    #[case::null_price(0, Value::Null)]
3395    #[case::empty_price(0, json!(""))]
3396    #[case::null_amount(1, Value::Null)]
3397    #[case::empty_amount(1, json!(""))]
3398    fn test_orderbook_missing_numeric_values_preserve_recovery_and_watermark(
3399        #[case] field: usize,
3400        #[case] value: Value,
3401        #[values("bids", "asks")] side: &str,
3402        #[values(false, true)] depth: bool,
3403    ) {
3404        let instrument = perp_instrument();
3405        let instrument_id = instrument.id();
3406        let price_precision = instrument.price_precision();
3407        let size_precision = instrument.size_precision();
3408        let (ctx, mut rx) = make_ctx(Some(instrument));
3409        let channel = "orderbook.ETH-PERP.1.10";
3410
3411        let channels = if depth {
3412            &ctx.active_book_depth_channels
3413        } else {
3414            &ctx.active_book_delta_channels
3415        };
3416
3417        channels.insert(instrument_id, channel.to_string());
3418
3419        let assert_snapshot = |event, msg: &DeriveOrderbookMsg| match event {
3420            DataEvent::Data(Data::BookDeltas(deltas)) if !depth => {
3421                let expected =
3422                    parse_orderbook_deltas(msg, price_precision, size_precision, deltas.ts_init)
3423                        .unwrap();
3424                assert_eq!(*deltas, expected);
3425            }
3426            DataEvent::Data(Data::BookDepth(snapshot)) if depth => {
3427                let expected =
3428                    parse_orderbook_depth(msg, price_precision, size_precision, snapshot.ts_init)
3429                        .unwrap();
3430                assert_eq!(*snapshot, expected);
3431            }
3432            other => panic!("expected book data event, was {other:?}"),
3433        };
3434
3435        let initial = orderbook_json();
3436        let timestamp = initial["timestamp"].as_i64().unwrap();
3437        let write = ctx.book_sync.write(channel).unwrap();
3438        let payload = subscription_payload(channel, &initial);
3439        let msg = parse_orderbook_msg(&payload).unwrap();
3440        DeriveDataClient::handle_ws_message(
3441            DeriveWsMessage::BookSubscription {
3442                payload,
3443                generation: write.generation.clone(),
3444            },
3445            &ctx,
3446        );
3447
3448        assert_snapshot(rx.try_recv().unwrap(), &msg);
3449
3450        let mut malformed = initial.clone();
3451        malformed["timestamp"] = json!(timestamp + 2);
3452        malformed[side][0][field] = value;
3453        let payload = subscription_payload(channel, &malformed);
3454        DeriveDataClient::handle_ws_message(
3455            DeriveWsMessage::BookSubscription {
3456                payload,
3457                generation: write.generation,
3458            },
3459            &ctx,
3460        );
3461
3462        assert!(rx.try_recv().is_err());
3463        let recoveries = ctx.book_sync.recoveries(Duration::ZERO, Instant::now());
3464        assert_eq!(recoveries.len(), 1);
3465        let recovery = &recoveries[0].2;
3466        let write = ctx
3467            .book_sync
3468            .begin_recovery_write(channel, recovery, recovery.gate.clone())
3469            .unwrap();
3470        write.gate.open();
3471        let payload = subscription_payload(channel, &malformed);
3472        DeriveDataClient::handle_ws_message(
3473            DeriveWsMessage::BookSubscription {
3474                payload,
3475                generation: write.generation.clone(),
3476            },
3477            &ctx,
3478        );
3479
3480        assert!(rx.try_recv().is_err());
3481        assert!(recovery.is_running());
3482        assert!(!recovery.is_accepted());
3483
3484        let mut valid = initial;
3485        valid["timestamp"] = json!(timestamp + 1);
3486        let payload = subscription_payload(channel, &valid);
3487        let msg = parse_orderbook_msg(&payload).unwrap();
3488        DeriveDataClient::handle_ws_message(
3489            DeriveWsMessage::BookSubscription {
3490                payload,
3491                generation: write.generation,
3492            },
3493            &ctx,
3494        );
3495
3496        assert_snapshot(rx.try_recv().unwrap(), &msg);
3497        assert!(recovery.is_accepted());
3498        assert!(!recovery.is_running());
3499        assert!(rx.try_recv().is_err());
3500        assert!(
3501            ctx.book_sync
3502                .recoveries(Duration::ZERO, Instant::now())
3503                .is_empty()
3504        );
3505    }
3506
3507    #[rstest]
3508    fn test_orderbook_frame_ignored_for_inactive_channel() {
3509        let instrument = perp_instrument();
3510        let instrument_id = instrument.id();
3511        let (ctx, mut rx) = make_ctx(Some(instrument));
3512        ctx.active_book_delta_channels
3513            .insert(instrument_id, "orderbook.ETH-PERP.1.20".to_string());
3514        let payload = subscription_payload("orderbook.ETH-PERP.1.10", &orderbook_json());
3515
3516        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3517
3518        assert!(rx.try_recv().is_err());
3519    }
3520
3521    #[rstest]
3522    fn test_handle_trades_subscription_filters_and_emits_active_instrument() {
3523        let instrument = perp_instrument();
3524        let other = btc_perp_instrument();
3525        let instrument_id = instrument.id();
3526        let (ctx, mut rx) = make_ctx(Some(instrument));
3527        cache_instrument(&ctx.instruments, &other);
3528        ctx.active_trade_subs.insert(instrument_id);
3529        let payload = subscription_payload(
3530            "trades.perp.ETH",
3531            &json!([
3532                trade_json("ETH-PERP", "trade-1"),
3533                trade_json("BTC-PERP", "trade-2")
3534            ]),
3535        );
3536
3537        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3538
3539        match rx.try_recv().unwrap() {
3540            DataEvent::Data(Data::Trade(trade)) => {
3541                assert_eq!(trade.instrument_id, instrument_id);
3542                assert_eq!(trade.trade_id.to_string(), "trade-1");
3543                assert_eq!(trade.price, Price::from("3500.00"));
3544                assert_eq!(trade.size, Quantity::from("1.000"));
3545            }
3546            other => panic!("expected trade data event, was {other:?}"),
3547        }
3548
3549        assert!(rx.try_recv().is_err());
3550    }
3551
3552    #[rstest]
3553    fn test_public_trade_replay_after_failed_delivery() {
3554        let instrument = perp_instrument();
3555        let instrument_id = instrument.id();
3556        let (mut ctx, rx) = make_ctx(Some(instrument));
3557        ctx.clock = Box::leak(Box::new(AtomicTime::new(false, UnixNanos::from(42))));
3558        ctx.active_trade_subs.insert(instrument_id);
3559        drop(rx);
3560        let payload = subscription_payload(
3561            "trades.perp.ETH",
3562            &json!([trade_json("ETH-PERP", "delivery-replay")]),
3563        );
3564        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload.clone()), &ctx);
3565        let (sender, mut rx) = tokio::sync::mpsc::unbounded_channel();
3566        ctx.data_sender = sender.into();
3567
3568        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload.clone()), &ctx);
3569        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3570
3571        let DataEvent::Data(Data::Trade(trade)) =
3572            rx.try_recv().expect("undelivered trade must replay")
3573        else {
3574            panic!("expected trade data event");
3575        };
3576
3577        assert_eq!(trade.instrument_id, instrument_id);
3578        assert_eq!(trade.trade_id, TradeId::from("delivery-replay"));
3579        assert_eq!(trade.price, Price::from("3500.00"));
3580        assert_eq!(trade.size, Quantity::from("1.000"));
3581        assert_eq!(trade.aggressor_side, AggressorSide::Buy);
3582        assert_eq!(trade.ts_event, UnixNanos::from(1_700_000_000_001_000_000));
3583        assert_eq!(trade.ts_init, UnixNanos::from(42));
3584        assert!(rx.try_recv().is_err());
3585    }
3586
3587    #[rstest]
3588    fn test_public_trades_continue_after_invalid_id() {
3589        let instrument = perp_instrument();
3590        let instrument_id = instrument.id();
3591        let (ctx, mut rx) = make_ctx(Some(instrument));
3592        ctx.active_trade_subs.insert(instrument_id);
3593        let payload = subscription_payload(
3594            "trades.perp.ETH",
3595            &json!([
3596                trade_json("ETH-PERP", "1234567890123456789012345678901234567"),
3597                trade_json("ETH-PERP", "valid-first")
3598            ]),
3599        );
3600
3601        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3602        let first_event = rx
3603            .try_recv()
3604            .expect("valid row in the same frame must emit");
3605        assert!(rx.try_recv().is_err());
3606        let payload = subscription_payload(
3607            "trades.perp.ETH",
3608            &json!([
3609                trade_json("ETH-PERP", "valid-first"),
3610                trade_json("ETH-PERP", "valid-next")
3611            ]),
3612        );
3613        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3614
3615        for (event, expected_id) in [
3616            (first_event, "valid-first"),
3617            (rx.try_recv().unwrap(), "valid-next"),
3618        ] {
3619            let DataEvent::Data(Data::Trade(trade)) = event else {
3620                panic!("expected trade data event");
3621            };
3622
3623            assert_eq!(trade.instrument_id, instrument_id);
3624            assert_eq!(trade.trade_id.to_string(), expected_id);
3625            assert_eq!(trade.price, Price::from("3500.00"));
3626            assert_eq!(trade.size, Quantity::from("1.000"));
3627            assert_eq!(trade.aggressor_side, AggressorSide::Buy);
3628            assert_eq!(trade.ts_event, UnixNanos::from(1_700_000_000_001_000_000));
3629        }
3630
3631        assert!(rx.try_recv().is_err());
3632    }
3633
3634    #[rstest]
3635    #[case("perps/http_public_trades_result_eth_paired_maker_first.json")]
3636    #[case("perps/http_public_trades_result_eth_paired_taker_first.json")]
3637    fn test_v3_public_paired_rows_deduplicate_across_frames(#[case] filename: &str) {
3638        let instrument = perp_instrument();
3639        let instrument_id = instrument.id();
3640        let (ctx, mut rx) = make_ctx(Some(instrument));
3641        ctx.active_trade_subs.insert(instrument_id);
3642        let rows = load_json(filename)["trades"].as_array().unwrap().clone();
3643        for row in &rows {
3644            let payload = subscription_payload("trades.perp.ETH", &json!([row]));
3645            DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3646        }
3647
3648        DeriveDataClient::handle_ws_message(DeriveWsMessage::Reconnected, &ctx);
3649        let payload = subscription_payload("trades.perp.ETH", &json!(rows));
3650        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3651
3652        match rx.try_recv().unwrap() {
3653            DataEvent::Data(Data::Trade(trade)) => {
3654                assert_eq!(trade.instrument_id, instrument_id);
3655                assert_eq!(trade.aggressor_side, AggressorSide::Buy);
3656                assert_eq!(trade.trade_id, TradeId::new("pub-pair-1"));
3657                assert_eq!(trade.price, Price::from("3500.00"));
3658                assert_eq!(trade.size, Quantity::from("0.250"));
3659                assert_eq!(trade.ts_event, UnixNanos::from(1_700_000_000_000_000_000));
3660            }
3661            other => panic!("expected trade data event, was {other:?}"),
3662        }
3663
3664        assert!(rx.try_recv().is_err());
3665    }
3666
3667    #[rstest]
3668    fn test_handle_subscription_without_cached_instrument_emits_no_event() {
3669        let (ctx, mut rx) = make_ctx(None);
3670        let instrument_id = InstrumentId::from("ETH-PERP.DERIVE");
3671        ctx.active_ticker_channels
3672            .insert(instrument_id, "ticker_slim.ETH-PERP.1000".to_string());
3673        ctx.active_quote_subs.insert(instrument_id);
3674        let payload =
3675            subscription_payload("ticker_slim.ETH-PERP.1000", &ticker_json(1_700_000_000_000));
3676
3677        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3678
3679        assert!(rx.try_recv().is_err());
3680    }
3681
3682    #[rstest]
3683    fn test_ticker_frame_ignored_without_quote_subscription() {
3684        let instrument = perp_instrument();
3685        let (ctx, mut rx) = make_ctx(Some(instrument));
3686        let payload =
3687            subscription_payload("ticker_slim.ETH-PERP.1000", &ticker_json(1_700_000_000_000));
3688
3689        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3690
3691        assert!(rx.try_recv().is_err());
3692    }
3693
3694    #[rstest]
3695    fn test_ticker_frame_ignored_for_inactive_channel() {
3696        let instrument = perp_instrument();
3697        let instrument_id = instrument.id();
3698        let (ctx, mut rx) = make_ctx(Some(instrument));
3699        ctx.active_ticker_channels
3700            .insert(instrument_id, "ticker_slim.ETH-PERP.100".to_string());
3701        ctx.active_quote_subs.insert(instrument_id);
3702        let payload =
3703            subscription_payload("ticker_slim.ETH-PERP.1000", &ticker_json(1_700_000_000_000));
3704
3705        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3706
3707        assert!(rx.try_recv().is_err());
3708    }
3709
3710    #[rstest]
3711    fn test_trade_channel_uses_instrument_type_and_currency() {
3712        let instrument = perp_instrument();
3713
3714        assert_eq!(trade_channel(&instrument).unwrap(), "trades.perp.ETH");
3715    }
3716
3717    #[rstest]
3718    fn test_trade_channel_uses_erc20_for_spot() {
3719        let instrument = spot_instrument();
3720
3721        assert_eq!(
3722            derive_instrument_type(&instrument).unwrap(),
3723            DeriveInstrumentType::Erc20
3724        );
3725        assert_eq!(trade_channel(&instrument).unwrap(), "trades.erc20.ETH");
3726    }
3727
3728    #[rstest]
3729    fn test_param_defaults_match_derive_public_channels() {
3730        assert_eq!(orderbook_group(&None).unwrap(), DEFAULT_ORDERBOOK_GROUP);
3731        assert_eq!(
3732            orderbook_depth(None, &None).unwrap(),
3733            DEFAULT_ORDERBOOK_DEPTH
3734        );
3735        assert_eq!(ticker_interval(&None).unwrap(), DEFAULT_TICKER_INTERVAL);
3736    }
3737
3738    #[rstest]
3739    fn test_orderbook_channel_parts_splits_from_right() {
3740        assert_eq!(
3741            orderbook_channel_parts("orderbook.ETH.TEST-PERP.10.100").unwrap(),
3742            (
3743                "ETH.TEST-PERP".to_string(),
3744                "10".to_string(),
3745                "100".to_string()
3746            )
3747        );
3748    }
3749
3750    #[rstest]
3751    fn test_ticker_channel_parts_splits_from_right() {
3752        assert_eq!(
3753            ticker_channel_parts("ticker_slim.ETH.TEST-PERP.1000").unwrap(),
3754            ("ETH.TEST-PERP".to_string(), "1000".to_string())
3755        );
3756    }
3757
3758    #[rstest]
3759    fn test_ticker_channel_parts_accepts_legacy_ticker_channel() {
3760        assert_eq!(
3761            ticker_channel_parts("ticker.ETH.TEST-PERP.1000").unwrap(),
3762            ("ETH.TEST-PERP".to_string(), "1000".to_string())
3763        );
3764    }
3765
3766    #[rstest]
3767    fn test_stale_generation_rollback_preserves_resubscription() {
3768        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3769        replace_data_event_sender(tx);
3770
3771        let client = DeriveDataClient::new(
3772            *DERIVE_CLIENT_ID,
3773            DeriveDataClientConfig {
3774                environment: DeriveEnvironment::Mainnet,
3775                ..Default::default()
3776            },
3777        )
3778        .unwrap();
3779
3780        let lifecycle = client.subscription_lifecycle();
3781        let instrument_id = InstrumentId::from("ETH-PERP.DERIVE");
3782
3783        let owner = ChannelOwner::Ticker {
3784            instrument_id,
3785            feed: TickerFeed::Quote,
3786        };
3787
3788        let channel = "ticker_slim.ETH-PERP.1000".to_string();
3789
3790        let first_generation = lifecycle.activate(owner, Some(&channel)).unwrap();
3791        lifecycle.remove(owner).unwrap();
3792        let second_generation = lifecycle.activate(owner, Some(&channel)).unwrap();
3793
3794        assert!(!lifecycle.rollback(owner, first_generation));
3795        assert!(lifecycle.is_current_channel(owner, second_generation, &channel));
3796        assert!(client.active_quote_subs.contains(&instrument_id));
3797        assert!(channel_is_active(
3798            &client.active_ticker_channels,
3799            instrument_id,
3800            &channel,
3801        ));
3802    }
3803
3804    #[rstest]
3805    fn test_subscription_registry_clear_does_not_reuse_generation() {
3806        let registry = ChannelSubscriptionRegistry::default();
3807        let owner = ChannelOwner::Trades(InstrumentId::from("ETH-20260627-3500-C.DERIVE"));
3808        let first_generation = registry.state.lock().activate(owner, None).unwrap();
3809
3810        registry.clear();
3811        let second_generation = registry.state.lock().activate(owner, None).unwrap();
3812
3813        assert!(second_generation > first_generation);
3814    }
3815
3816    #[rstest]
3817    fn test_subscription_registry_clear_preserves_inflight_transition() {
3818        let registry = ChannelSubscriptionRegistry::default();
3819        let first = registry.transition("ticker_slim.ETH-PERP.1000");
3820
3821        registry.clear();
3822        let second = registry.transition("ticker_slim.ETH-PERP.1000");
3823
3824        assert!(Arc::ptr_eq(&first, &second));
3825    }
3826
3827    #[rstest]
3828    fn test_subscription_transition_registry_collects_inactive_channels() {
3829        let registry = ChannelSubscriptionRegistry::default();
3830        let live = registry.transition("ticker_slim.ETH-PERP.1000");
3831        for index in 0..TRANSITION_GC_THRESHOLD + 16 {
3832            drop(registry.transition(&format!("ticker_slim.ETH-OPTION-{index}.1000")));
3833        }
3834
3835        let same_live = registry.transition("ticker_slim.ETH-PERP.1000");
3836
3837        assert!(Arc::ptr_eq(&live, &same_live));
3838        assert!(registry.transitions.len() < TRANSITION_GC_THRESHOLD);
3839    }
3840
3841    #[rstest]
3842    fn test_ambiguous_cleanup_retains_survivor_for_reconnect() {
3843        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3844        replace_data_event_sender(tx);
3845
3846        let client = DeriveDataClient::new(
3847            *DERIVE_CLIENT_ID,
3848            DeriveDataClientConfig {
3849                environment: DeriveEnvironment::Mainnet,
3850                ..Default::default()
3851            },
3852        )
3853        .unwrap();
3854
3855        let lifecycle = client.subscription_lifecycle();
3856        let instrument_id = InstrumentId::from("ETH-20260627-3600-C.DERIVE");
3857        let owner = ChannelOwner::Trades(instrument_id);
3858        lifecycle.activate(owner, Some("trades.option.ETH"));
3859
3860        let error = DeriveWsError::Timeout {
3861            method: "unsubscribe".to_string(),
3862        };
3863
3864        let ws = client.ws_handle();
3865
3866        let retained = retain_channel_for_reconnect(&ws, "trades.option.ETH", true, Some(&error));
3867
3868        assert!(retained);
3869        assert!(lifecycle.is_active(owner));
3870        assert!(client.active_trade_subs.contains(&instrument_id));
3871        assert!(ws.has_subscription("trades.option.ETH"));
3872    }
3873
3874    #[rstest]
3875    fn test_explicit_cleanup_rejection_preserves_surviving_channel_owner() {
3876        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3877        replace_data_event_sender(tx);
3878
3879        let client = DeriveDataClient::new(
3880            *DERIVE_CLIENT_ID,
3881            DeriveDataClientConfig {
3882                environment: DeriveEnvironment::Mainnet,
3883                ..Default::default()
3884            },
3885        )
3886        .unwrap();
3887
3888        let lifecycle = client.subscription_lifecycle();
3889        let instrument_id = InstrumentId::from("ETH-20260627-3600-C.DERIVE");
3890        let owner = ChannelOwner::Trades(instrument_id);
3891        lifecycle.activate(owner, Some("trades.option.ETH"));
3892
3893        let error = DeriveWsError::JsonRpc {
3894            code: -32603,
3895            message: "not subscribed".to_string(),
3896            data: None,
3897        };
3898
3899        let ws = client.ws_handle();
3900        ws.remember_subscription("trades.option.ETH");
3901
3902        let retained = retain_channel_for_reconnect(&ws, "trades.option.ETH", true, Some(&error));
3903
3904        assert!(!retained);
3905        assert!(lifecycle.is_active(owner));
3906        assert!(client.active_trade_subs.contains(&instrument_id));
3907        assert!(!ws.has_subscription("trades.option.ETH"));
3908    }
3909
3910    #[rstest]
3911    fn test_ambiguous_cleanup_without_survivor_is_not_replayed() {
3912        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3913        replace_data_event_sender(tx);
3914
3915        let client = DeriveDataClient::new(
3916            *DERIVE_CLIENT_ID,
3917            DeriveDataClientConfig {
3918                environment: DeriveEnvironment::Mainnet,
3919                ..Default::default()
3920            },
3921        )
3922        .unwrap();
3923
3924        let ws = client.ws_handle();
3925        ws.remember_subscription("trades.option.ETH");
3926
3927        let error = DeriveWsError::Timeout {
3928            method: "unsubscribe".to_string(),
3929        };
3930
3931        let retained = retain_channel_for_reconnect(&ws, "trades.option.ETH", false, Some(&error));
3932
3933        assert!(!retained);
3934        assert!(!ws.has_subscription("trades.option.ETH"));
3935    }
3936
3937    #[rstest]
3938    fn test_unsubscribe_quotes_prunes_cached_quote() {
3939        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3940        replace_data_event_sender(tx);
3941
3942        let mut client = DeriveDataClient::new(
3943            *DERIVE_CLIENT_ID,
3944            DeriveDataClientConfig {
3945                environment: DeriveEnvironment::Mainnet,
3946                ..Default::default()
3947            },
3948        )
3949        .unwrap();
3950
3951        let instrument_id = InstrumentId::from("ETH-PERP.DERIVE");
3952
3953        let owner = ChannelOwner::Ticker {
3954            instrument_id,
3955            feed: TickerFeed::Quote,
3956        };
3957
3958        client
3959            .subscription_lifecycle()
3960            .activate(owner, Some("ticker_slim.ETH-PERP.1000"));
3961        client.quote_cache.lock().insert(
3962            instrument_id,
3963            QuoteTick::new(
3964                instrument_id,
3965                Price::from("3500.00"),
3966                Price::from("3501.00"),
3967                Quantity::from("1.000"),
3968                Quantity::from("2.000"),
3969                UnixNanos::from(1),
3970                UnixNanos::from(1),
3971            ),
3972        );
3973
3974        let command = UnsubscribeQuotes::new(
3975            instrument_id,
3976            Some(*DERIVE_CLIENT_ID),
3977            None,
3978            UUID4::new(),
3979            UnixNanos::default(),
3980            None,
3981            None,
3982        );
3983
3984        client.unsubscribe_quotes(&command).unwrap();
3985
3986        assert!(!client.quote_cache.lock().contains(&instrument_id));
3987    }
3988
3989    #[rstest]
3990    #[tokio::test]
3991    async fn test_unsubscribe_trades_uses_recorded_channel_without_cached_instrument() {
3992        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3993        replace_data_event_sender(tx);
3994
3995        let mut client = DeriveDataClient::new(
3996            *DERIVE_CLIENT_ID,
3997            DeriveDataClientConfig {
3998                environment: DeriveEnvironment::Mainnet,
3999                ..Default::default()
4000            },
4001        )
4002        .unwrap();
4003
4004        let instrument_id = InstrumentId::from("ETH-20260627-3500-C.DERIVE");
4005        let owner = ChannelOwner::Trades(instrument_id);
4006        let lifecycle = client.subscription_lifecycle();
4007        let generation = lifecycle.activate(owner, None).unwrap();
4008        assert!(lifecycle.attach_channel(owner, generation, "trades.option.ETH".to_string(),));
4009
4010        let command = UnsubscribeTrades::new(
4011            instrument_id,
4012            Some(*DERIVE_CLIENT_ID),
4013            None,
4014            UUID4::new(),
4015            UnixNanos::default(),
4016            None,
4017            None,
4018        );
4019
4020        client.unsubscribe_trades(&command).unwrap();
4021        wait_until_async(
4022            || {
4023                let registry = Arc::clone(&client.channel_subscriptions);
4024                async move { registry.transitions.contains_key("trades.option.ETH") }
4025            },
4026            Duration::from_secs(1),
4027        )
4028        .await;
4029
4030        assert!(!lifecycle.is_active(owner));
4031        assert!(!client.active_trade_subs.contains(&instrument_id));
4032    }
4033
4034    fn perp_ticker_payload(instrument_id: InstrumentId) -> WsSubscriptionPayload {
4035        let channel = "ticker_slim.ETH-PERP.1000";
4036        let payload = subscription_payload(
4037            channel,
4038            &json!({
4039                "timestamp": 1_700_000_000_010_i64,
4040                "instrument_ticker": ticker_json(1_700_000_000_000)
4041            }),
4042        );
4043        assert_eq!(payload.channel, channel);
4044        let _ = instrument_id;
4045        payload
4046    }
4047
4048    fn install_ticker(ctx: &WsMessageContext, instrument_id: InstrumentId, channel: &str) {
4049        ctx.active_ticker_channels
4050            .insert(instrument_id, channel.to_string());
4051    }
4052
4053    #[rstest]
4054    fn test_ticker_emits_mark_price_when_mark_subscribed() {
4055        let instrument = perp_instrument();
4056        let instrument_id = instrument.id();
4057        let (ctx, mut rx) = make_ctx(Some(instrument));
4058        install_ticker(&ctx, instrument_id, "ticker_slim.ETH-PERP.1000");
4059        ctx.active_mark_subs.insert(instrument_id);
4060
4061        DeriveDataClient::handle_ws_message(
4062            DeriveWsMessage::Subscription(perp_ticker_payload(instrument_id)),
4063            &ctx,
4064        );
4065
4066        match rx.try_recv().unwrap() {
4067            DataEvent::Data(Data::MarkPrice(mark)) => {
4068                assert_eq!(mark.instrument_id, instrument_id);
4069                assert_eq!(mark.value, Price::from("3500.50"));
4070            }
4071            other => panic!("expected MarkPriceUpdate, was {other:?}"),
4072        }
4073
4074        assert!(rx.try_recv().is_err());
4075    }
4076
4077    #[rstest]
4078    fn test_ticker_emits_index_price_when_index_subscribed() {
4079        let instrument = perp_instrument();
4080        let instrument_id = instrument.id();
4081        let (ctx, mut rx) = make_ctx(Some(instrument));
4082        install_ticker(&ctx, instrument_id, "ticker_slim.ETH-PERP.1000");
4083        ctx.active_index_subs.insert(instrument_id);
4084
4085        DeriveDataClient::handle_ws_message(
4086            DeriveWsMessage::Subscription(perp_ticker_payload(instrument_id)),
4087            &ctx,
4088        );
4089
4090        match rx.try_recv().unwrap() {
4091            DataEvent::Data(Data::IndexPrice(index)) => {
4092                assert_eq!(index.instrument_id, instrument_id);
4093                assert_eq!(index.value, Price::from("3500.00"));
4094            }
4095            other => panic!("expected IndexPriceUpdate, was {other:?}"),
4096        }
4097
4098        assert!(rx.try_recv().is_err());
4099    }
4100
4101    #[rstest]
4102    fn test_ticker_emits_funding_rate_for_perp_when_subscribed() {
4103        let instrument = perp_instrument();
4104        let instrument_id = instrument.id();
4105        let (ctx, mut rx) = make_ctx(Some(instrument));
4106        install_ticker(&ctx, instrument_id, "ticker_slim.ETH-PERP.1000");
4107        ctx.active_funding_subs.insert(instrument_id);
4108
4109        DeriveDataClient::handle_ws_message(
4110            DeriveWsMessage::Subscription(perp_ticker_payload(instrument_id)),
4111            &ctx,
4112        );
4113
4114        match rx.try_recv().unwrap() {
4115            DataEvent::FundingRate(update) => {
4116                assert_eq!(update.instrument_id, instrument_id);
4117                assert_eq!(update.rate, "0.0002".parse().unwrap());
4118            }
4119            other => panic!("expected FundingRateUpdate, was {other:?}"),
4120        }
4121
4122        assert!(rx.try_recv().is_err());
4123    }
4124
4125    #[rstest]
4126    fn test_ticker_skips_funding_rate_when_not_perp() {
4127        let instrument = option_instrument();
4128        let instrument_id = instrument.id();
4129        let channel = format!("ticker_slim.{}.1000", instrument_id.symbol.as_str());
4130        let (ctx, mut rx) = make_ctx(Some(instrument));
4131        install_ticker(&ctx, instrument_id, &channel);
4132        ctx.active_funding_subs.insert(instrument_id);
4133
4134        let mut option_data = option_ticker_json(1_700_000_000_000);
4135        option_data["instrument_name"] = json!(instrument_id.symbol.as_str());
4136        let payload = subscription_payload(
4137            &channel,
4138            &json!({
4139                "timestamp": 1_700_000_000_010_i64,
4140                "instrument_ticker": option_data
4141            }),
4142        );
4143
4144        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
4145
4146        assert!(rx.try_recv().is_err());
4147    }
4148
4149    #[rstest]
4150    fn test_ticker_emits_option_greeks_when_subscribed() {
4151        let instrument = option_instrument();
4152        let instrument_id = instrument.id();
4153        let channel = format!("ticker_slim.{}.1000", instrument_id.symbol.as_str());
4154        let (ctx, mut rx) = make_ctx(Some(instrument));
4155        install_ticker(&ctx, instrument_id, &channel);
4156        ctx.active_greeks_subs.insert(instrument_id);
4157
4158        let mut option_data = option_ticker_json(1_700_000_000_000);
4159        option_data["instrument_name"] = json!(instrument_id.symbol.as_str());
4160        let payload = subscription_payload(
4161            &channel,
4162            &json!({
4163                "timestamp": 1_700_000_000_010_i64,
4164                "instrument_ticker": option_data
4165            }),
4166        );
4167
4168        DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
4169
4170        match rx.try_recv().unwrap() {
4171            DataEvent::OptionGreeks(greeks) => {
4172                assert_eq!(greeks.instrument_id, instrument_id);
4173                assert!((greeks.greeks.delta - 0.55).abs() < 1e-9);
4174                assert!((greeks.greeks.gamma - 0.0008).abs() < 1e-9);
4175                assert!((greeks.greeks.vega - 4.5).abs() < 1e-9);
4176                assert!((greeks.greeks.theta + 2.1).abs() < 1e-9);
4177                assert!((greeks.greeks.rho - 1.2).abs() < 1e-9);
4178                assert_eq!(greeks.mark_iv, Some(0.60));
4179                assert_eq!(greeks.bid_iv, Some(0.58));
4180                assert_eq!(greeks.ask_iv, Some(0.62));
4181                assert_eq!(greeks.underlying_price, Some(3505.0));
4182                assert_eq!(greeks.open_interest, Some(1000.0));
4183            }
4184            other => panic!("expected OptionGreeks, was {other:?}"),
4185        }
4186
4187        assert!(rx.try_recv().is_err());
4188    }
4189
4190    #[rstest]
4191    fn test_ticker_emits_all_subscribed_feeds_in_one_frame() {
4192        let instrument = perp_instrument();
4193        let instrument_id = instrument.id();
4194        let (ctx, mut rx) = make_ctx(Some(instrument));
4195        install_ticker(&ctx, instrument_id, "ticker_slim.ETH-PERP.1000");
4196        ctx.active_quote_subs.insert(instrument_id);
4197        ctx.active_mark_subs.insert(instrument_id);
4198        ctx.active_index_subs.insert(instrument_id);
4199        ctx.active_funding_subs.insert(instrument_id);
4200
4201        DeriveDataClient::handle_ws_message(
4202            DeriveWsMessage::Subscription(perp_ticker_payload(instrument_id)),
4203            &ctx,
4204        );
4205
4206        let mut quote = None;
4207        let mut mark = None;
4208        let mut index = None;
4209        let mut funding = None;
4210
4211        while let Ok(event) = rx.try_recv() {
4212            match event {
4213                DataEvent::Data(Data::Quote(q)) => {
4214                    assert!(quote.replace(q).is_none(), "duplicate Quote emission");
4215                }
4216                DataEvent::Data(Data::MarkPrice(m)) => {
4217                    assert!(
4218                        mark.replace(m).is_none(),
4219                        "duplicate MarkPriceUpdate emission"
4220                    );
4221                }
4222                DataEvent::Data(Data::IndexPrice(i)) => {
4223                    assert!(
4224                        index.replace(i).is_none(),
4225                        "duplicate IndexPriceUpdate emission"
4226                    );
4227                }
4228                DataEvent::FundingRate(f) => {
4229                    assert!(
4230                        funding.replace(f).is_none(),
4231                        "duplicate FundingRate emission"
4232                    );
4233                }
4234                other => panic!("unexpected event: {other:?}"),
4235            }
4236        }
4237
4238        let quote = quote.expect("Quote event missing");
4239        let mark = mark.expect("MarkPriceUpdate missing");
4240        let index = index.expect("IndexPriceUpdate missing");
4241        let funding = funding.expect("FundingRateUpdate missing");
4242
4243        assert_eq!(quote.instrument_id, instrument_id);
4244        assert_eq!(quote.bid_price, Price::from("3500.00"));
4245        assert_eq!(quote.ask_price, Price::from("3501.00"));
4246        assert_eq!(mark.instrument_id, instrument_id);
4247        assert_eq!(mark.value, Price::from("3500.50"));
4248        assert_eq!(index.instrument_id, instrument_id);
4249        assert_eq!(index.value, Price::from("3500.00"));
4250        assert_eq!(funding.instrument_id, instrument_id);
4251        assert_eq!(funding.rate, "0.0002".parse().unwrap());
4252    }
4253
4254    #[rstest]
4255    fn test_reset_clears_all_subscription_state() {
4256        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
4257        replace_data_event_sender(tx);
4258
4259        let config = DeriveDataClientConfig {
4260            environment: DeriveEnvironment::Mainnet,
4261            ..Default::default()
4262        };
4263
4264        let mut client = DeriveDataClient::new(*DERIVE_CLIENT_ID, config).unwrap();
4265        let instrument = perp_instrument();
4266        let instrument_id = instrument.id();
4267        cache_instrument(&client.instruments, &instrument);
4268
4269        client
4270            .active_book_delta_channels
4271            .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
4272        client
4273            .active_book_depth_channels
4274            .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
4275        client
4276            .active_ticker_channels
4277            .insert(instrument_id, "ticker_slim.ETH-PERP.1000".to_string());
4278        client.active_quote_subs.insert(instrument_id);
4279        client.active_trade_subs.insert(instrument_id);
4280        client
4281            .channel_subscriptions
4282            .state
4283            .lock()
4284            .activate(ChannelOwner::Trades(instrument_id), Some("trades.perp.ETH"));
4285        client.channel_subscriptions.transition("trades.perp.ETH");
4286        client.active_mark_subs.insert(instrument_id);
4287        client.active_index_subs.insert(instrument_id);
4288        client.active_funding_subs.insert(instrument_id);
4289        client.active_greeks_subs.insert(instrument_id);
4290
4291        client.reset().unwrap();
4292
4293        assert!(!client.instruments.contains_key(&instrument_id));
4294        assert!(
4295            !client
4296                .active_book_delta_channels
4297                .contains_key(&instrument_id)
4298        );
4299        assert!(
4300            !client
4301                .active_book_depth_channels
4302                .contains_key(&instrument_id)
4303        );
4304        assert!(!client.active_ticker_channels.contains_key(&instrument_id));
4305        assert!(!client.active_quote_subs.contains(&instrument_id));
4306        assert!(!client.active_trade_subs.contains(&instrument_id));
4307        assert!(client.channel_subscriptions.state.lock().owners.is_empty());
4308
4309        // Sync reset cannot await canceled tasks, so their channel locks stay
4310        // stable until the next async teardown joins them.
4311        assert_eq!(client.channel_subscriptions.transitions.len(), 1);
4312        assert!(!client.active_mark_subs.contains(&instrument_id));
4313        assert!(!client.active_index_subs.contains(&instrument_id));
4314        assert!(!client.active_funding_subs.contains(&instrument_id));
4315        assert!(!client.active_greeks_subs.contains(&instrument_id));
4316        assert!(!client.is_connected());
4317    }
4318
4319    #[tokio::test]
4320    async fn test_disconnect_clears_subscription_state() {
4321        // Regression: aborting in-flight subscribe tasks during disconnect()
4322        // skips their on-error rollback, so any `active_*` entries staged
4323        // before spawn would leak across reconnect and silently suppress
4324        // the next subscribe. disconnect() must clear the same local
4325        // subscription maps that reset() does.
4326        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
4327        replace_data_event_sender(tx);
4328
4329        let config = DeriveDataClientConfig {
4330            environment: DeriveEnvironment::Mainnet,
4331            ..Default::default()
4332        };
4333
4334        let mut client = DeriveDataClient::new(*DERIVE_CLIENT_ID, config).unwrap();
4335        let instrument = perp_instrument();
4336        let instrument_id = instrument.id();
4337        cache_instrument(&client.instruments, &instrument);
4338
4339        client
4340            .active_book_delta_channels
4341            .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
4342        client
4343            .active_book_depth_channels
4344            .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
4345        client
4346            .active_ticker_channels
4347            .insert(instrument_id, "ticker_slim.ETH-PERP.1000".to_string());
4348        client.active_quote_subs.insert(instrument_id);
4349        client.active_trade_subs.insert(instrument_id);
4350        client
4351            .channel_subscriptions
4352            .state
4353            .lock()
4354            .activate(ChannelOwner::Trades(instrument_id), Some("trades.perp.ETH"));
4355        client.channel_subscriptions.transition("trades.perp.ETH");
4356        client.active_mark_subs.insert(instrument_id);
4357        client.active_index_subs.insert(instrument_id);
4358        client.active_funding_subs.insert(instrument_id);
4359        client.active_greeks_subs.insert(instrument_id);
4360        client.is_connected.store(true, Ordering::Relaxed);
4361        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
4362        let (drop_tx, mut drop_rx) = tokio::sync::oneshot::channel::<()>();
4363        client
4364            .pending_tasks
4365            .spawn(async move {
4366                let _drop_tx = drop_tx;
4367                let _ = started_tx.send(());
4368                std::future::pending::<()>().await;
4369            })
4370            .unwrap();
4371
4372        started_rx.await.unwrap();
4373
4374        client.disconnect().await.unwrap();
4375
4376        // Instrument cache must survive disconnect (reset() is the path
4377        // that wipes it); only the subscription maps are cleared.
4378        assert!(client.instruments.contains_key(&instrument_id));
4379        assert!(
4380            !client
4381                .active_book_delta_channels
4382                .contains_key(&instrument_id)
4383        );
4384        assert!(
4385            !client
4386                .active_book_depth_channels
4387                .contains_key(&instrument_id)
4388        );
4389        assert!(!client.active_ticker_channels.contains_key(&instrument_id));
4390        assert!(!client.active_quote_subs.contains(&instrument_id));
4391        assert!(!client.active_trade_subs.contains(&instrument_id));
4392        assert!(client.channel_subscriptions.state.lock().owners.is_empty());
4393        assert!(client.channel_subscriptions.transitions.is_empty());
4394        assert!(!client.active_mark_subs.contains(&instrument_id));
4395        assert!(!client.active_index_subs.contains(&instrument_id));
4396        assert!(!client.active_funding_subs.contains(&instrument_id));
4397        assert!(!client.active_greeks_subs.contains(&instrument_id));
4398        assert!(!client.is_connected());
4399        assert_eq!(
4400            drop_rx.try_recv(),
4401            Err(tokio::sync::oneshot::error::TryRecvError::Closed),
4402        );
4403    }
4404
4405    #[tokio::test]
4406    async fn test_task_group_unregisters_finished_tasks_before_drain() {
4407        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
4408        replace_data_event_sender(tx);
4409
4410        let config = DeriveDataClientConfig {
4411            environment: DeriveEnvironment::Mainnet,
4412            ..Default::default()
4413        };
4414
4415        let client = DeriveDataClient::new(*DERIVE_CLIENT_ID, config).unwrap();
4416        for _ in 0..100 {
4417            client.spawn_task("test_noop", async { Ok(()) });
4418        }
4419
4420        wait_until_async(
4421            || async { client.pending_tasks.all_finished() },
4422            Duration::from_secs(2),
4423        )
4424        .await;
4425
4426        assert!(client.pending_tasks.is_empty());
4427        client.pending_tasks.begin_shutdown();
4428        client
4429            .pending_tasks
4430            .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
4431            .await
4432            .unwrap();
4433        assert!(client.pending_tasks.is_empty());
4434    }
4435}
nautilus_derive/data.rs

nautilus_derive/
data.rs
