1use std::{
19 num::NonZeroUsize,
20 str::FromStr,
21 sync::{
22 Arc, Weak,
23 atomic::{AtomicBool, Ordering},
24 },
25 time::Duration,
26};
27
28use ahash::{AHashMap, AHashSet};
29use anyhow::Context;
30use async_trait::async_trait;
31use dashmap::DashMap;
32use nautilus_common::{
33 cache::{InstrumentLookupError, quote::QuoteCache},
34 clients::DataClient,
35 live::runner::get_data_event_sender,
36 messages::{
37 DataEvent,
38 data::{
39 BarsResponse, DataResponse, ForwardPricesResponse, FundingRatesResponse,
40 InstrumentResponse, InstrumentsResponse, QuotesResponse, RequestBars,
41 RequestForwardPrices, RequestFundingRates, RequestInstrument, RequestInstruments,
42 RequestQuotes, RequestTrades, SubscribeBookDeltas, SubscribeBookDepth10,
43 SubscribeFundingRates, SubscribeIndexPrices, SubscribeMarkPrices,
44 SubscribeOptionGreeks, SubscribeQuotes, SubscribeTrades, TradesResponse,
45 UnsubscribeBookDeltas, UnsubscribeBookDepth10, UnsubscribeFundingRates,
46 UnsubscribeIndexPrices, UnsubscribeMarkPrices, UnsubscribeOptionGreeks,
47 UnsubscribeQuotes, UnsubscribeTrades,
48 },
49 },
50 providers::InstrumentProvider,
51};
52use nautilus_core::{
53 AtomicMap, AtomicSet, Params, UnixNanos,
54 datetime::{NANOSECONDS_IN_SECOND, datetime_to_unix_nanos},
55 time::{AtomicTime, get_atomic_clock_realtime},
56};
57use nautilus_live::{
58 SocketControl,
59 task::{TaskGroup, TaskGroupGuard},
60};
61use nautilus_model::{
62 data::{Bar, Data, ForwardPrice, QuoteTick},
63 enums::{AggregationSource, BookType, PriceType},
64 identifiers::{ClientId, InstrumentId, Venue},
65 instruments::{Instrument, InstrumentAny},
66 types::{Price, Quantity},
67};
68use parking_lot::Mutex;
69use tokio_util::sync::CancellationToken;
70
71use crate::{
72 common::{
73 consts::{
74 DERIVE_CANDLES_DEFAULT_LIMIT, DERIVE_CANDLES_MAX_PAGES, DERIVE_TRADES_PAGE_SIZE,
75 DERIVE_VENUE,
76 },
77 enums::{
78 DeriveInstrumentType, DeriveOrderbookDepth, DeriveOrderbookGroup, DeriveTickerInterval,
79 },
80 parse::{format_instrument_id, format_venue_symbol, parse_derive_instrument_any},
81 },
82 config::DeriveDataClientConfig,
83 http::DeriveHttpClient,
84 providers::{
85 DeriveInstrumentProvider, fetch_instrument_definitions, parse_instrument_definitions,
86 },
87 websocket::{
88 DEFAULT_ORDERBOOK_DEPTH, DEFAULT_ORDERBOOK_GROUP, DEFAULT_TICKER_INTERVAL,
89 DerivePublicWsData, DeriveTickerMsg, DeriveWebSocketClient,
90 DeriveWebSocketSubscriptionHandle, DeriveWsError, DeriveWsMessage, WsMessageContext,
91 bar_spec_to_derive_period, orderbook_channel, parse_candle_record, parse_funding_rate,
92 parse_funding_rate_history_record, parse_index_price, parse_mark_price,
93 parse_option_greeks, parse_orderbook_deltas, parse_orderbook_depth10, parse_public_ws_data,
94 parse_ticker_quote, parse_ticker_quote_from_rest, parse_trade_tick,
95 parse_trade_tick_from_rest, ticker_channel, ticker_ts_event, trades_channel,
96 },
97};
98
99#[derive(Debug)]
101pub struct DeriveDataClient {
102 client_id: ClientId,
103 config: DeriveDataClientConfig,
104 http_client: DeriveHttpClient,
105 provider: DeriveInstrumentProvider,
106 ws_client: DeriveWebSocketClient,
107 is_connected: Arc<AtomicBool>,
108 cancellation_token: CancellationToken,
109 session_tasks: TaskGroup,
110 pending_tasks: TaskGroup,
111 shutdown_errors: Vec<String>,
112 data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
113 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
114 active_book_delta_channels: Arc<AtomicMap<InstrumentId, String>>,
115 active_book_depth10_channels: Arc<AtomicMap<InstrumentId, String>>,
116 active_ticker_channels: Arc<AtomicMap<InstrumentId, String>>,
117 active_quote_subs: Arc<AtomicSet<InstrumentId>>,
118 active_trade_subs: Arc<AtomicSet<InstrumentId>>,
119 active_mark_subs: Arc<AtomicSet<InstrumentId>>,
120 active_index_subs: Arc<AtomicSet<InstrumentId>>,
121 active_funding_subs: Arc<AtomicSet<InstrumentId>>,
122 active_greeks_subs: Arc<AtomicSet<InstrumentId>>,
123 channel_subscriptions: Arc<ChannelSubscriptionRegistry>,
124 subscription_lock: Arc<Mutex<()>>,
125 quote_cache: Arc<Mutex<QuoteCache>>,
126 clock: &'static AtomicTime,
127}
128
129impl DeriveDataClient {
130 pub fn new(client_id: ClientId, config: DeriveDataClientConfig) -> anyhow::Result<Self> {
136 let clock = get_atomic_clock_realtime();
137 let data_sender = get_data_event_sender();
138 let http_client = DeriveHttpClient::new(
139 config.rest_url(),
140 Some(config.http_timeout_secs),
141 config.proxy_url.clone(),
142 None,
143 )?;
144 let provider = DeriveInstrumentProvider::with_expired(
145 http_client.clone(),
146 config.currencies.clone(),
147 config.include_expired,
148 );
149 let mut ws_client = DeriveWebSocketClient::new(
150 Some(config.ws_url()),
151 config.environment,
152 config.transport_backend,
153 config.proxy_url.clone(),
154 )
155 .with_socket_control(SocketControl::new(
156 client_id,
157 Some(*DERIVE_VENUE),
158 "derive-data-streams",
159 ));
160
161 if let Some(secs) = config.ws_timeout_secs {
162 ws_client.set_request_timeout(Duration::from_secs(secs));
163 }
164
165 let session_tasks = TaskGroup::new();
166 let pending_tasks = TaskGroup::new();
167
168 Ok(Self {
169 client_id,
170 config,
171 http_client,
172 provider,
173 ws_client,
174 is_connected: Arc::new(AtomicBool::new(false)),
175 cancellation_token: CancellationToken::new(),
176 session_tasks,
177 pending_tasks,
178 shutdown_errors: Vec::new(),
179 data_sender,
180 instruments: Arc::new(AtomicMap::new()),
181 active_book_delta_channels: Arc::new(AtomicMap::new()),
182 active_book_depth10_channels: Arc::new(AtomicMap::new()),
183 active_ticker_channels: Arc::new(AtomicMap::new()),
184 active_quote_subs: Arc::new(AtomicSet::new()),
185 active_trade_subs: Arc::new(AtomicSet::new()),
186 active_mark_subs: Arc::new(AtomicSet::new()),
187 active_index_subs: Arc::new(AtomicSet::new()),
188 active_funding_subs: Arc::new(AtomicSet::new()),
189 active_greeks_subs: Arc::new(AtomicSet::new()),
190 channel_subscriptions: Arc::new(ChannelSubscriptionRegistry::default()),
191 subscription_lock: Arc::new(Mutex::new(())),
192 quote_cache: Arc::new(Mutex::new(QuoteCache::new())),
193 clock,
194 })
195 }
196
197 fn spawn_task<F>(&self, description: &'static str, fut: F)
200 where
201 F: std::future::Future<Output = anyhow::Result<()>> + Send + 'static,
202 {
203 let future = async move {
204 if let Err(e) = fut.await {
205 log::warn!("{description} failed: {e:?}");
206 }
207 };
208
209 if let Err(e) = self.pending_tasks.spawn(future) {
210 log::warn!("Skipping Derive {description} after shutdown began: {e}");
211 }
212 }
213
214 fn abort_pending_tasks(&self) {
215 self.pending_tasks.begin_shutdown();
216 }
217
218 fn abort_session_tasks(&self) {
219 self.session_tasks.begin_shutdown();
220 self.ws_client.begin_shutdown();
221 }
222
223 fn clear_subscription_state(&self) {
228 let _guard = self.subscription_lock.lock();
229 self.channel_subscriptions.clear();
230 self.active_book_delta_channels.store(AHashMap::new());
231 self.active_book_depth10_channels.store(AHashMap::new());
232 self.active_ticker_channels.store(AHashMap::new());
233 self.active_quote_subs.store(AHashSet::new());
234 self.active_trade_subs.store(AHashSet::new());
235 self.active_mark_subs.store(AHashSet::new());
236 self.active_index_subs.store(AHashSet::new());
237 self.active_funding_subs.store(AHashSet::new());
238 self.active_greeks_subs.store(AHashSet::new());
239 self.quote_cache.lock().clear();
240 }
241
242 async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
243 self.cancellation_token.cancel();
244 self.abort_session_tasks();
245 self.abort_pending_tasks();
246
247 if let Err(e) = self.ws_client.disconnect().await {
248 self.shutdown_errors
249 .push(format!("Derive WebSocket shutdown failed: {e}"));
250 }
251 let (session_result, pending_result) =
252 tokio::join!(self.join_session_tasks(), self.join_pending_tasks());
253 self.clear_subscription_state();
254 self.channel_subscriptions.clear_transitions();
255 self.is_connected.store(false, Ordering::Release);
256
257 if let Err(e) = session_result {
258 self.shutdown_errors.push(e.to_string());
259 }
260
261 if let Err(e) = pending_result {
262 self.shutdown_errors.push(e.to_string());
263 }
264
265 if !self.shutdown_errors.is_empty() {
266 anyhow::bail!(std::mem::take(&mut self.shutdown_errors).join("; "));
267 }
268 Ok(())
269 }
270
271 fn spawn_stream_task(
272 &self,
273 mut rx: tokio::sync::mpsc::UnboundedReceiver<DeriveWsMessage>,
274 ) -> anyhow::Result<()> {
275 let ctx = WsMessageContext {
276 clock: self.clock,
277 data_sender: self.data_sender.clone(),
278 instruments: Arc::clone(&self.instruments),
279 active_book_delta_channels: Arc::clone(&self.active_book_delta_channels),
280 active_book_depth10_channels: Arc::clone(&self.active_book_depth10_channels),
281 active_ticker_channels: Arc::clone(&self.active_ticker_channels),
282 active_quote_subs: Arc::clone(&self.active_quote_subs),
283 active_trade_subs: Arc::clone(&self.active_trade_subs),
284 active_mark_subs: Arc::clone(&self.active_mark_subs),
285 active_index_subs: Arc::clone(&self.active_index_subs),
286 active_funding_subs: Arc::clone(&self.active_funding_subs),
287 active_greeks_subs: Arc::clone(&self.active_greeks_subs),
288 subscription_lock: Arc::clone(&self.subscription_lock),
289 quote_cache: Arc::clone(&self.quote_cache),
290 };
291 let cancellation = self.cancellation_token.clone();
292 let is_connected = Arc::clone(&self.is_connected);
293
294 self.session_tasks.spawn(async move {
295 loop {
296 tokio::select! {
297 maybe_msg = rx.recv() => {
298 match maybe_msg {
299 Some(msg) => {
300 if matches!(&msg, DeriveWsMessage::SessionRecoveryFailed(_)) {
301 is_connected.store(false, Ordering::Release);
302 }
303 Self::handle_ws_message(msg, &ctx);
304 }
305 None => {
306 log::debug!("Derive WebSocket data stream ended");
307 break;
308 }
309 }
310 }
311 () = cancellation.cancelled() => {
312 log::debug!("Derive WebSocket data stream task cancelled");
313 break;
314 }
315 }
316 }
317 })?;
318
319 Ok(())
320 }
321
322 fn handle_ws_message(message: DeriveWsMessage, ctx: &WsMessageContext) {
323 match message {
324 DeriveWsMessage::Subscription(payload) => match parse_public_ws_data(&payload) {
325 Ok(data) => Self::handle_public_ws_data(data, ctx),
326 Err(e) => {
327 let snippet = truncated_payload_snippet(payload.data.get());
331 log::warn!(
332 "Failed to parse Derive public WS data on channel `{}`: {e}; payload: {snippet}",
333 payload.channel,
334 );
335 }
336 },
337 DeriveWsMessage::Reconnected => {
338 let _guard = ctx.subscription_lock.lock();
339 ctx.quote_cache.lock().clear();
340 log::info!("Derive WebSocket reconnected");
341 }
342 DeriveWsMessage::SessionRecoveryFailed(reason) => {
343 log::error!("Derive WebSocket session recovery failed: {reason}");
344 }
345 DeriveWsMessage::Authenticated => log::debug!("Derive WebSocket authenticated"),
346 }
347 }
348
349 fn handle_public_ws_data(data: DerivePublicWsData, ctx: &WsMessageContext) {
350 let _guard = ctx.subscription_lock.lock();
353
354 match data {
355 DerivePublicWsData::Orderbook(msg) => {
356 let instrument_id = msg.data.instrument_id();
357 let channel = msg.channel.as_str();
358 let deltas_active =
359 channel_is_active(&ctx.active_book_delta_channels, instrument_id, channel);
360 let depth10_active =
361 channel_is_active(&ctx.active_book_depth10_channels, instrument_id, channel);
362
363 if !deltas_active && !depth10_active {
364 return;
365 }
366
367 let Some(instrument) = ctx.instruments.get_cloned(&instrument_id) else {
368 log::warn!("Orderbook message received for unknown instrument {instrument_id}");
369 return;
370 };
371
372 let ts_init = ctx.clock.get_time_ns();
373
374 if deltas_active {
375 match parse_orderbook_deltas(
376 &msg,
377 instrument.price_precision(),
378 instrument.size_precision(),
379 ts_init,
380 ) {
381 Ok(deltas) => {
382 Self::send_data(ctx, Data::Deltas(Box::new(deltas)));
383 }
384 Err(e) => log::warn!("Failed to parse Derive orderbook deltas: {e}"),
385 }
386 }
387
388 if depth10_active {
389 match parse_orderbook_depth10(
390 &msg,
391 instrument.price_precision(),
392 instrument.size_precision(),
393 ts_init,
394 ) {
395 Ok(depth) => Self::send_data(ctx, Data::Depth10(Box::new(depth))),
396 Err(e) => log::warn!("Failed to parse Derive orderbook depth10: {e}"),
397 }
398 }
399 }
400 DerivePublicWsData::Trades(msg) => {
401 let ts_init = ctx.clock.get_time_ns();
402
403 for trade in &msg.trades {
404 let instrument_id = format_instrument_id(trade.instrument_name.as_str());
405
406 if !ctx.active_trade_subs.contains(&instrument_id) {
407 continue;
408 }
409
410 let Some(instrument) = ctx.instruments.get_cloned(&instrument_id) else {
411 log::warn!("Trade message received for unknown instrument {instrument_id}");
412 continue;
413 };
414
415 match parse_trade_tick(
416 trade,
417 instrument.price_precision(),
418 instrument.size_precision(),
419 ts_init,
420 ) {
421 Ok(tick) => Self::send_data(ctx, Data::Trade(tick)),
422 Err(e) => log::warn!("Failed to parse Derive trade tick: {e}"),
423 }
424 }
425 }
426 DerivePublicWsData::Ticker(msg) => {
427 let instrument_id = msg.data.instrument_id();
428
429 if !channel_is_active(
430 &ctx.active_ticker_channels,
431 instrument_id,
432 msg.channel.as_str(),
433 ) {
434 return;
435 }
436
437 let Some(instrument) = ctx.instruments.get_cloned(&instrument_id) else {
438 log::warn!("Ticker message received for unknown instrument {instrument_id}");
439 return;
440 };
441
442 let ts_init = ctx.clock.get_time_ns();
443 let price_precision = instrument.price_precision();
444
445 if ctx.active_quote_subs.contains(&instrument_id) {
446 let mut quote_cache = ctx.quote_cache.lock();
447
448 match process_ticker_quote(
449 &msg,
450 price_precision,
451 instrument.size_precision(),
452 ts_init,
453 &mut quote_cache,
454 ) {
455 Ok(Some(quote)) => Self::send_data(ctx, Data::Quote(quote)),
456 Ok(None) => {}
457 Err(e) => log::warn!("Failed to parse Derive ticker quote: {e}"),
458 }
459 }
460
461 if ctx.active_mark_subs.contains(&instrument_id) {
462 match parse_mark_price(&msg, price_precision, ts_init) {
463 Ok(Some(update)) => Self::send_data(ctx, Data::MarkPrice(update)),
464 Ok(None) => {}
465 Err(e) => log::warn!("Failed to parse Derive mark price: {e}"),
466 }
467 }
468
469 if ctx.active_index_subs.contains(&instrument_id) {
470 match parse_index_price(&msg, price_precision, ts_init) {
471 Ok(Some(update)) => Self::send_data(ctx, Data::IndexPrice(update)),
472 Ok(None) => {}
473 Err(e) => log::warn!("Failed to parse Derive index price: {e}"),
474 }
475 }
476
477 if ctx.active_funding_subs.contains(&instrument_id) {
478 match parse_funding_rate(&msg, ts_init) {
479 Ok(Some(update)) => {
480 if let Err(e) = ctx.data_sender.send(DataEvent::FundingRate(update)) {
481 log::error!("Failed to send Derive funding rate: {e}");
482 }
483 }
484 Ok(None) => {}
485 Err(e) => log::warn!("Failed to parse Derive funding rate: {e}"),
486 }
487 }
488
489 if ctx.active_greeks_subs.contains(&instrument_id) {
490 match parse_option_greeks(&msg, ts_init) {
491 Ok(Some(greeks)) => {
492 if let Err(e) = ctx.data_sender.send(DataEvent::OptionGreeks(greeks)) {
493 log::error!("Failed to send Derive option greeks: {e}");
494 }
495 }
496 Ok(None) => {}
497 Err(e) => log::warn!("Failed to parse Derive option greeks: {e}"),
498 }
499 }
500 }
501 }
502 }
503
504 fn send_data(ctx: &WsMessageContext, data: Data) {
505 if let Err(e) = ctx.data_sender.send(DataEvent::Data(data)) {
506 log::error!("Failed to send Derive data event: {e}");
507 }
508 }
509
510 fn cache_provider_instruments(&self) {
511 let instruments = self
512 .provider
513 .store()
514 .get_all()
515 .values()
516 .cloned()
517 .collect::<Vec<_>>();
518
519 for instrument in instruments {
520 self.cache_instrument(&instrument);
521 if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
522 log::warn!("Failed to send Derive instrument: {e}");
523 }
524 }
525 }
526
527 fn cache_instrument(&self, instrument: &InstrumentAny) {
528 cache_instrument(&self.instruments, instrument);
529 }
530
531 fn prepare_subscribe(&self, instrument_id: InstrumentId) -> anyhow::Result<bool> {
532 if self.instruments.contains_key(&instrument_id) {
533 return Ok(false);
534 }
535
536 if !self.config.auto_load_missing_instruments {
537 anyhow::bail!(
538 "Instrument {instrument_id} not found and `auto_load_missing_instruments` is disabled"
539 );
540 }
541 Ok(true)
542 }
543
544 async fn lazy_load_instrument(
545 http_client: DeriveHttpClient,
546 instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
547 instrument_id: InstrumentId,
548 include_expired: bool,
549 ) -> anyhow::Result<()> {
550 let currency = currency_from_instrument_id(&instrument_id)?;
551 let definitions = fetch_instrument_definitions(&http_client, currency, include_expired)
552 .await
553 .with_context(|| format!("failed to lazy-load Derive instruments for {currency}"))?;
554 let mut found = false;
555
556 for instrument in parse_instrument_definitions(definitions) {
557 if instrument.id() == instrument_id {
558 found = true;
559 }
560 cache_instrument(&instruments, &instrument);
561 }
562
563 if !found {
564 anyhow::bail!(InstrumentLookupError::not_found(instrument_id));
565 }
566
567 Ok(())
568 }
569
570 fn ws_handle(&self) -> DeriveWebSocketSubscriptionHandle {
571 self.ws_client.subscription_handle()
572 }
573
574 fn subscription_lifecycle(&self) -> SubscriptionLifecycle {
575 SubscriptionLifecycle {
576 registry: Arc::clone(&self.channel_subscriptions),
577 lock: Arc::clone(&self.subscription_lock),
578 dispatch: SubscriptionDispatchState {
579 active_book_delta_channels: Arc::clone(&self.active_book_delta_channels),
580 active_book_depth10_channels: Arc::clone(&self.active_book_depth10_channels),
581 active_ticker_channels: Arc::clone(&self.active_ticker_channels),
582 active_quote_subs: Arc::clone(&self.active_quote_subs),
583 active_trade_subs: Arc::clone(&self.active_trade_subs),
584 active_mark_subs: Arc::clone(&self.active_mark_subs),
585 active_index_subs: Arc::clone(&self.active_index_subs),
586 active_funding_subs: Arc::clone(&self.active_funding_subs),
587 active_greeks_subs: Arc::clone(&self.active_greeks_subs),
588 quote_cache: Arc::clone(&self.quote_cache),
589 },
590 }
591 }
592
593 fn subscribe_ticker_feed(
594 &self,
595 instrument_id: InstrumentId,
596 params: &Option<Params>,
597 feed: TickerFeed,
598 label: &'static str,
599 ) -> anyhow::Result<()> {
600 let owner = ChannelOwner::Ticker {
601 instrument_id,
602 feed,
603 };
604 let lifecycle = self.subscription_lifecycle();
605 if lifecycle.is_active(owner) {
606 return Ok(());
607 }
608
609 let channel = match self.active_ticker_channels.get_cloned(&instrument_id) {
610 Some(channel) => channel,
611 None => {
612 let instrument_name = format_venue_symbol(&instrument_id)?.to_string();
613 let interval = ticker_interval(params)?;
614 ticker_channel(&instrument_name, &interval)
615 }
616 };
617 let request = ChannelRequest::from_channel(&channel)?;
618 let needs_load = self.prepare_subscribe(instrument_id)?;
619 let Some(generation) = lifecycle.activate(owner, Some(&channel)) else {
620 return Ok(());
621 };
622 let ws = self.ws_handle();
623 let http_client = self.http_client.clone();
624 let include_expired = self.config.include_expired;
625 let instruments = Arc::clone(&self.instruments);
626
627 self.spawn_task("subscribe_ticker_feed", async move {
628 if needs_load
629 && let Err(e) = Self::lazy_load_instrument(
630 http_client,
631 instruments,
632 instrument_id,
633 include_expired,
634 )
635 .await
636 {
637 lifecycle.rollback(owner, generation);
638 log::error!("Lazy-load failed for {instrument_id} ({label}): {e}");
639 return Ok(());
640 }
641
642 run_channel_subscribe(lifecycle, owner, generation, request, ws).await
643 });
644
645 Ok(())
646 }
647
648 fn unsubscribe_channel_owner(&self, owner: ChannelOwner) -> anyhow::Result<()> {
649 let lifecycle = self.subscription_lifecycle();
650 let Some(removed) = lifecycle.remove(owner) else {
651 return Ok(());
652 };
653
654 if !removed.channel_empty {
655 return Ok(());
656 }
657 let Some(channel) = removed.channel else {
658 return Ok(());
659 };
660 let request = ChannelRequest::from_channel(&channel)?;
661 let ws = self.ws_handle();
662
663 self.spawn_task("unsubscribe_channel", async move {
664 run_channel_unsubscribe(lifecycle, request, ws).await
665 });
666 Ok(())
667 }
668}
669
670#[async_trait(?Send)]
671impl DataClient for DeriveDataClient {
672 fn client_id(&self) -> ClientId {
673 self.client_id
674 }
675
676 fn venue(&self) -> Option<Venue> {
677 Some(*DERIVE_VENUE)
678 }
679
680 fn start(&mut self) -> anyhow::Result<()> {
681 log::info!("Starting Derive data client: {}", self.client_id);
682 Ok(())
683 }
684
685 fn stop(&mut self) -> anyhow::Result<()> {
686 log::info!("Stopping Derive data client: {}", self.client_id);
687 self.cancellation_token.cancel();
688 self.abort_session_tasks();
689 self.abort_pending_tasks();
690 self.is_connected.store(false, Ordering::Relaxed);
691 Ok(())
692 }
693
694 fn reset(&mut self) -> anyhow::Result<()> {
695 log::info!("Resetting Derive data client: {}", self.client_id);
696 self.cancellation_token.cancel();
697
698 self.abort_session_tasks();
699 self.abort_pending_tasks();
700 self.is_connected.store(false, Ordering::Relaxed);
701
702 self.instruments.store(AHashMap::new());
705 self.clear_subscription_state();
706 self.provider.store_mut().clear();
707 Ok(())
708 }
709
710 fn dispose(&mut self) -> anyhow::Result<()> {
711 log::debug!("Disposing Derive data client: {}", self.client_id);
712 self.stop()
713 }
714
715 fn is_connected(&self) -> bool {
716 self.is_connected.load(Ordering::SeqCst)
717 }
718
719 fn is_disconnected(&self) -> bool {
720 !self.is_connected()
721 }
722
723 async fn connect(&mut self) -> anyhow::Result<()> {
724 if self.is_connected()
725 && !self.cancellation_token.is_cancelled()
726 && self.session_tasks.is_open()
727 && self.pending_tasks.is_open()
728 {
729 return Ok(());
730 }
731
732 if self.cancellation_token.is_cancelled()
734 || !self.session_tasks.is_open()
735 || !self.pending_tasks.is_open()
736 {
737 self.teardown_partial_connect().await?;
738 self.cancellation_token = CancellationToken::new();
739 self.session_tasks.start_generation().map_err(|e| {
740 anyhow::anyhow!("Failed to start Derive data session generation: {e}")
741 })?;
742 self.pending_tasks
743 .start_generation()
744 .map_err(|e| anyhow::anyhow!("Failed to start Derive data task generation: {e}"))?;
745 }
746 let cancellation_token = self.cancellation_token.clone();
747 let ws_shutdown = self.ws_client.shutdown_handle();
748 let setup_guard =
749 TaskGroupGuard::new(&[&self.session_tasks, &self.pending_tasks], move || {
750 cancellation_token.cancel();
751 ws_shutdown.begin_shutdown();
752 });
753
754 if !self.config.currencies.is_empty() {
755 self.provider
756 .load_all(None)
757 .await
758 .context("failed to load Derive instruments")?;
759 self.cache_provider_instruments();
760 }
761
762 self.ws_client
763 .connect()
764 .await
765 .context("failed to connect Derive WebSocket")?;
766 let session_result = self
767 .ws_client
768 .take_event_receiver()
769 .ok_or_else(|| anyhow::anyhow!("Derive WebSocket event receiver not initialized"))
770 .and_then(|rx| self.spawn_stream_task(rx));
771
772 if let Err(e) = session_result {
773 if let Err(teardown_error) = self.teardown_partial_connect().await {
774 return Err(e.context(format!(
775 "Derive data startup teardown failed: {teardown_error}"
776 )));
777 }
778 return Err(e);
779 }
780
781 self.is_connected.store(true, Ordering::Release);
782 setup_guard.disarm();
783 log::info!(
784 "Connected Derive data client ({:?})",
785 self.config.environment
786 );
787 Ok(())
788 }
789
790 async fn disconnect(&mut self) -> anyhow::Result<()> {
791 self.teardown_partial_connect().await?;
792 log::info!("Disconnected Derive data client");
793 Ok(())
794 }
795
796 fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
797 if cmd.book_type != BookType::L2_MBP {
798 anyhow::bail!("Derive only supports L2_MBP order book deltas");
799 }
800
801 let instrument_id = cmd.instrument_id;
802 let owner = ChannelOwner::BookDeltas(instrument_id);
803 let lifecycle = self.subscription_lifecycle();
804 if lifecycle.is_active(owner) {
805 return Ok(());
806 }
807
808 let instrument_name = format_venue_symbol(&instrument_id)?.to_string();
809 let group = orderbook_group(&cmd.params)?;
810 let depth = orderbook_depth(cmd.depth.map(|d| d.get()), &cmd.params)?;
811 let channel = orderbook_channel(&instrument_name, &group, &depth);
812 let request = ChannelRequest::from_channel(&channel)?;
813 let needs_load = self.prepare_subscribe(instrument_id)?;
814 let Some(generation) = lifecycle.activate(owner, Some(&channel)) else {
815 return Ok(());
816 };
817 let ws = self.ws_handle();
818 let http_client = self.http_client.clone();
819 let include_expired = self.config.include_expired;
820 let instruments = Arc::clone(&self.instruments);
821
822 self.spawn_task("subscribe_book_deltas", async move {
823 if needs_load
824 && let Err(e) = Self::lazy_load_instrument(
825 http_client,
826 instruments,
827 instrument_id,
828 include_expired,
829 )
830 .await
831 {
832 lifecycle.rollback(owner, generation);
833 log::error!("Lazy-load failed for {instrument_id} (book deltas): {e}");
834 return Ok(());
835 }
836
837 run_channel_subscribe(lifecycle, owner, generation, request, ws).await
838 });
839
840 Ok(())
841 }
842
843 fn subscribe_book_depth10(&mut self, cmd: SubscribeBookDepth10) -> anyhow::Result<()> {
844 if cmd.book_type != BookType::L2_MBP {
845 anyhow::bail!("Derive only supports L2_MBP order book depth");
846 }
847
848 let instrument_id = cmd.instrument_id;
849 let owner = ChannelOwner::BookDepth10(instrument_id);
850 let lifecycle = self.subscription_lifecycle();
851 if lifecycle.is_active(owner) {
852 return Ok(());
853 }
854
855 let instrument_name = format_venue_symbol(&instrument_id)?.to_string();
856 let group = orderbook_group(&cmd.params)?;
857 let depth = DeriveOrderbookDepth::D10.to_string();
858 let channel = orderbook_channel(&instrument_name, &group, &depth);
859 let request = ChannelRequest::from_channel(&channel)?;
860 let needs_load = self.prepare_subscribe(instrument_id)?;
861 let Some(generation) = lifecycle.activate(owner, Some(&channel)) else {
862 return Ok(());
863 };
864 let ws = self.ws_handle();
865 let http_client = self.http_client.clone();
866 let include_expired = self.config.include_expired;
867 let instruments = Arc::clone(&self.instruments);
868
869 self.spawn_task("subscribe_book_depth10", async move {
870 if needs_load
871 && let Err(e) = Self::lazy_load_instrument(
872 http_client,
873 instruments,
874 instrument_id,
875 include_expired,
876 )
877 .await
878 {
879 lifecycle.rollback(owner, generation);
880 log::error!("Lazy-load failed for {instrument_id} (book depth10): {e}");
881 return Ok(());
882 }
883
884 run_channel_subscribe(lifecycle, owner, generation, request, ws).await
885 });
886
887 Ok(())
888 }
889
890 fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
891 self.subscribe_ticker_feed(cmd.instrument_id, &cmd.params, TickerFeed::Quote, "quotes")
892 }
893
894 fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
895 let instrument_id = cmd.instrument_id;
896 let owner = ChannelOwner::Trades(instrument_id);
897 let lifecycle = self.subscription_lifecycle();
898 if lifecycle.is_active(owner) {
899 return Ok(());
900 }
901
902 let needs_load = self.prepare_subscribe(instrument_id)?;
903 let Some(generation) = lifecycle.activate(owner, None) else {
904 return Ok(());
905 };
906 let ws = self.ws_handle();
907 let http_client = self.http_client.clone();
908 let include_expired = self.config.include_expired;
909 let instruments = Arc::clone(&self.instruments);
910
911 self.spawn_task("subscribe_trades", async move {
912 if needs_load
913 && let Err(e) = Self::lazy_load_instrument(
914 http_client,
915 Arc::clone(&instruments),
916 instrument_id,
917 include_expired,
918 )
919 .await
920 {
921 lifecycle.rollback(owner, generation);
922 log::error!("Lazy-load failed for {instrument_id} (trades): {e}");
923 return Ok(());
924 }
925
926 if !lifecycle.is_current(owner, generation) {
927 return Ok(());
928 }
929
930 let Some(instrument) = instruments.get_cloned(&instrument_id) else {
931 lifecycle.rollback(owner, generation);
932 log::error!("Instrument {instrument_id} not found for Derive trades");
933 return Ok(());
934 };
935 let channel = match trade_channel(&instrument) {
936 Ok(channel) => channel,
937 Err(e) => {
938 lifecycle.rollback(owner, generation);
939 log::error!("Failed to resolve Derive trades channel: {e}");
940 return Ok(());
941 }
942 };
943 let request = match ChannelRequest::from_channel(&channel) {
944 Ok(request) => request,
945 Err(e) => {
946 lifecycle.rollback(owner, generation);
947 log::error!("Invalid Derive trades channel `{channel}`: {e}");
948 return Ok(());
949 }
950 };
951
952 if !lifecycle.attach_channel(owner, generation, channel) {
953 return Ok(());
954 }
955
956 run_channel_subscribe(lifecycle, owner, generation, request, ws).await
957 });
958
959 Ok(())
960 }
961
962 fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
963 self.subscribe_ticker_feed(
964 cmd.instrument_id,
965 &cmd.params,
966 TickerFeed::Mark,
967 "mark prices",
968 )
969 }
970
971 fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
972 self.subscribe_ticker_feed(
973 cmd.instrument_id,
974 &cmd.params,
975 TickerFeed::Index,
976 "index prices",
977 )
978 }
979
980 fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
981 self.subscribe_ticker_feed(
982 cmd.instrument_id,
983 &cmd.params,
984 TickerFeed::Funding,
985 "funding rates",
986 )
987 }
988
989 fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
990 self.subscribe_ticker_feed(
991 cmd.instrument_id,
992 &cmd.params,
993 TickerFeed::Greeks,
994 "option greeks",
995 )
996 }
997
998 fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
999 self.unsubscribe_channel_owner(ChannelOwner::BookDeltas(cmd.instrument_id))
1000 }
1001
1002 fn unsubscribe_book_depth10(&mut self, cmd: &UnsubscribeBookDepth10) -> anyhow::Result<()> {
1003 self.unsubscribe_channel_owner(ChannelOwner::BookDepth10(cmd.instrument_id))
1004 }
1005
1006 fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
1007 self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1008 instrument_id: cmd.instrument_id,
1009 feed: TickerFeed::Quote,
1010 })
1011 }
1012
1013 fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
1014 self.unsubscribe_channel_owner(ChannelOwner::Trades(cmd.instrument_id))
1015 }
1016
1017 fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
1018 self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1019 instrument_id: cmd.instrument_id,
1020 feed: TickerFeed::Mark,
1021 })
1022 }
1023
1024 fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
1025 self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1026 instrument_id: cmd.instrument_id,
1027 feed: TickerFeed::Index,
1028 })
1029 }
1030
1031 fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
1032 self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1033 instrument_id: cmd.instrument_id,
1034 feed: TickerFeed::Funding,
1035 })
1036 }
1037
1038 fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
1039 self.unsubscribe_channel_owner(ChannelOwner::Ticker {
1040 instrument_id: cmd.instrument_id,
1041 feed: TickerFeed::Greeks,
1042 })
1043 }
1044
1045 fn request_quotes(&self, request: RequestQuotes) -> anyhow::Result<()> {
1046 let instrument_id = request.instrument_id;
1048 let instrument = self
1049 .instruments
1050 .get_cloned(&instrument_id)
1051 .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1052 let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1053 let price_precision = instrument.price_precision();
1054 let size_precision = instrument.size_precision();
1055
1056 let http_client = self.http_client.clone();
1057 let sender = self.data_sender.clone();
1058 let clock = self.clock;
1059 let client_id = request.client_id.unwrap_or(self.client_id);
1060 let request_id = request.request_id;
1061 let params = request.params;
1062 let start_nanos = datetime_to_unix_nanos(request.start);
1063 let end_nanos = datetime_to_unix_nanos(request.end);
1064
1065 self.spawn_task("request_quotes", async move {
1066 let ticker = match http_client.get_ticker(&venue_symbol).await {
1067 Ok(ticker) => ticker,
1068 Err(e) => {
1069 log::error!("Failed to fetch Derive ticker for {instrument_id}: {e:?}");
1070 return Ok(());
1071 }
1072 };
1073
1074 let ts_init = clock.get_time_ns();
1075 let quotes = match parse_ticker_quote_from_rest(
1076 &ticker,
1077 price_precision,
1078 size_precision,
1079 ts_init,
1080 ) {
1081 Ok(quote) => {
1082 let within_start = start_nanos.is_none_or(|nanos| quote.ts_event >= nanos);
1084 let within_end = end_nanos.is_none_or(|nanos| quote.ts_event <= nanos);
1085 if within_start && within_end {
1086 vec![quote]
1087 } else {
1088 Vec::new()
1089 }
1090 }
1091 Err(e) => {
1092 log::warn!("Failed to parse Derive ticker for {instrument_id}: {e}");
1093 Vec::new()
1094 }
1095 };
1096
1097 let response = DataResponse::Quotes(QuotesResponse::new(
1098 request_id,
1099 client_id,
1100 instrument_id,
1101 quotes,
1102 start_nanos,
1103 end_nanos,
1104 clock.get_time_ns(),
1105 params,
1106 ));
1107
1108 if let Err(e) = sender.send(DataEvent::Response(response)) {
1109 log::error!("Failed to send Derive quotes response: {e}");
1110 }
1111 Ok(())
1112 });
1113
1114 Ok(())
1115 }
1116
1117 fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
1118 let instrument_id = request.instrument_id;
1119 let instrument = self
1120 .instruments
1121 .get_cloned(&instrument_id)
1122 .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1123 let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1124 let price_precision = instrument.price_precision();
1125 let size_precision = instrument.size_precision();
1126
1127 let http_client = self.http_client.clone();
1128 let sender = self.data_sender.clone();
1129 let clock = self.clock;
1130 let client_id = request.client_id.unwrap_or(self.client_id);
1131 let request_id = request.request_id;
1132 let params = request.params;
1133 let start = request.start;
1134 let end = request.end;
1135 let limit = request.limit.map(NonZeroUsize::get);
1136 let start_nanos = datetime_to_unix_nanos(start);
1137 let end_nanos = datetime_to_unix_nanos(end);
1138 let from_timestamp = start.map(|dt| dt.as_millisecond());
1139 let to_timestamp = Some(match end {
1140 Some(dt) => dt.as_millisecond(),
1141 None => i64::try_from(clock.get_time_ms())
1142 .context("Derive current time exceeds i64 milliseconds")?,
1143 });
1144
1145 self.spawn_task("request_trades", async move {
1146 let page_size = limit.map_or(DERIVE_TRADES_PAGE_SIZE, |cap| {
1150 cap.min(DERIVE_TRADES_PAGE_SIZE as usize) as u32
1151 });
1152 let mut trades = Vec::new();
1153 let mut seen_trade_ids = AHashSet::new();
1154 let mut page = 1u32;
1155
1156 loop {
1157 let result = match http_client
1158 .get_trade_history(&venue_symbol, from_timestamp, to_timestamp, page, page_size)
1159 .await
1160 {
1161 Ok(result) => result,
1162 Err(e) => {
1163 log::error!("Failed to fetch Derive trades for {instrument_id}: {e:?}");
1164 return Ok(());
1165 }
1166 };
1167
1168 if result.trades.is_empty() {
1169 break;
1170 }
1171
1172 let num_pages = result.pagination.num_pages;
1173 let ts_init = clock.get_time_ns();
1174
1175 for trade in &result.trades {
1176 match parse_trade_tick_from_rest(
1177 trade,
1178 price_precision,
1179 size_precision,
1180 ts_init,
1181 ) {
1182 Ok(tick) if seen_trade_ids.insert(tick.trade_id) => trades.push(tick),
1183 Ok(_) => {}
1184 Err(e) => log::warn!(
1185 "Failed to parse Derive trade {} for {instrument_id}: {e}",
1186 trade.trade_id,
1187 ),
1188 }
1189 }
1190
1191 if let Some(cap) = limit
1192 && trades.len() >= cap
1193 {
1194 break;
1195 }
1196
1197 if (page as i64) >= num_pages {
1198 break;
1199 }
1200 page += 1;
1201 }
1202
1203 trades.sort_by_key(|trade| trade.ts_event);
1204 if let Some(cap) = limit
1205 && trades.len() > cap
1206 {
1207 trades.drain(..trades.len() - cap);
1208 }
1209
1210 let response = DataResponse::Trades(TradesResponse::new(
1211 request_id,
1212 client_id,
1213 instrument_id,
1214 trades,
1215 start_nanos,
1216 end_nanos,
1217 clock.get_time_ns(),
1218 params,
1219 ));
1220
1221 if let Err(e) = sender.send(DataEvent::Response(response)) {
1222 log::error!("Failed to send Derive trades response: {e}");
1223 }
1224 Ok(())
1225 });
1226
1227 Ok(())
1228 }
1229
1230 fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
1231 let instrument_id = request.instrument_id;
1232 let instrument = self
1233 .instruments
1234 .get_cloned(&instrument_id)
1235 .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1236 anyhow::ensure!(
1237 matches!(instrument, InstrumentAny::CryptoPerpetual(_)),
1238 "Funding rates are only available for Derive perpetual instruments (got {instrument_id})",
1239 );
1240 let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1241
1242 let http_client = self.http_client.clone();
1243 let sender = self.data_sender.clone();
1244 let clock = self.clock;
1245 let client_id = request.client_id.unwrap_or(self.client_id);
1246 let request_id = request.request_id;
1247 let params = request.params;
1248 let start = request.start;
1249 let end = request.end;
1250 let limit = request.limit.map(NonZeroUsize::get);
1251 let start_nanos = datetime_to_unix_nanos(start);
1252 let end_nanos = datetime_to_unix_nanos(end);
1253 let start_ms = start.map(|dt| dt.as_millisecond());
1254 let end_ms = end.map(|dt| dt.as_millisecond());
1255
1256 self.spawn_task("request_funding_rates", async move {
1257 let result = match http_client
1258 .get_funding_rate_history(&venue_symbol, start_ms, end_ms, None)
1259 .await
1260 {
1261 Ok(result) => result,
1262 Err(e) => {
1263 log::error!(
1264 "Failed to fetch Derive funding rate history for {instrument_id}: {e:?}",
1265 );
1266 return Ok(());
1267 }
1268 };
1269
1270 let ts_init = clock.get_time_ns();
1271 let mut updates = Vec::with_capacity(result.funding_rate_history.len());
1272
1273 for record in &result.funding_rate_history {
1274 match parse_funding_rate_history_record(record, instrument_id, None, ts_init) {
1275 Ok(update) => updates.push(update),
1276 Err(e) => log::warn!(
1277 "Failed to parse Derive funding rate record for {instrument_id} at {}: {e}",
1278 record.timestamp,
1279 ),
1280 }
1281 }
1282
1283 updates.sort_by_key(|update| update.ts_event);
1284 if let Some(cap) = limit
1285 && updates.len() > cap
1286 {
1287 updates.drain(..updates.len() - cap);
1288 }
1289
1290 let response = DataResponse::FundingRates(FundingRatesResponse::new(
1291 request_id,
1292 client_id,
1293 instrument_id,
1294 updates,
1295 start_nanos,
1296 end_nanos,
1297 clock.get_time_ns(),
1298 params,
1299 ));
1300
1301 if let Err(e) = sender.send(DataEvent::Response(response)) {
1302 log::error!("Failed to send Derive funding rates response: {e}");
1303 }
1304 Ok(())
1305 });
1306
1307 Ok(())
1308 }
1309
1310 fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
1311 let bar_type = request.bar_type;
1312 anyhow::ensure!(
1313 bar_type.aggregation_source() == AggregationSource::External,
1314 "Derive only supports EXTERNAL aggregation source (got {bar_type})",
1315 );
1316 let spec = bar_type.spec();
1317 anyhow::ensure!(
1318 spec.price_type == PriceType::Last,
1319 "Derive candles are trade-based; only PriceType::Last is supported (got {bar_type})",
1320 );
1321
1322 let instrument_id = bar_type.instrument_id();
1323 let instrument = self
1324 .instruments
1325 .get_cloned(&instrument_id)
1326 .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1327 let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1328 let price_precision = instrument.price_precision();
1329 let size_precision = instrument.size_precision();
1330
1331 let period = bar_spec_to_derive_period(spec.aggregation, spec.step.get() as u64)
1332 .with_context(|| format!("unsupported Derive bar spec for {bar_type}"))?;
1333
1334 let http_client = self.http_client.clone();
1335 let sender = self.data_sender.clone();
1336 let clock = self.clock;
1337 let client_id = request.client_id.unwrap_or(self.client_id);
1338 let request_id = request.request_id;
1339 let params = request.params;
1340 let start = request.start;
1341 let end = request.end;
1342 let limit = request.limit.map(NonZeroUsize::get);
1343 let start_nanos = datetime_to_unix_nanos(start);
1344 let end_nanos = datetime_to_unix_nanos(end);
1345
1346 let request_time = clock.get_time_ns();
1349 let now_secs = (request_time.as_u64() / NANOSECONDS_IN_SECOND) as i64;
1350 let end_ts = end.map_or(now_secs, |dt| dt.as_second());
1351 let default_span = i64::from(period) * limit.unwrap_or(DERIVE_CANDLES_DEFAULT_LIMIT) as i64;
1352 let start_ts = start.map_or(end_ts - default_span, |dt| dt.as_second());
1353
1354 self.spawn_task("request_bars", async move {
1355 let mut seen_timestamps: AHashSet<i64> = AHashSet::new();
1358 let mut pages: Vec<Vec<Bar>> = Vec::new();
1359 let mut total_bars = 0usize;
1360 let mut current_end = end_ts;
1361 let mut page_count = 0;
1362
1363 loop {
1364 page_count += 1;
1365
1366 let mut records = match http_client
1367 .get_candles(&venue_symbol, start_ts, current_end, period)
1368 .await
1369 {
1370 Ok(records) => records,
1371 Err(e) => {
1372 log::error!("Failed to fetch Derive candles for {bar_type}: {e:?}");
1373 return Ok(());
1374 }
1375 };
1376
1377 if records.is_empty() {
1378 break;
1379 }
1380
1381 records.sort_by_key(|r| r.timestamp_bucket);
1382
1383 let has_new = records
1384 .iter()
1385 .any(|r| !seen_timestamps.contains(&r.timestamp_bucket));
1386
1387 if !has_new {
1388 break;
1389 }
1390
1391 let ts_init = clock.get_time_ns();
1392 let mut page_bars = Vec::with_capacity(records.len());
1393 let mut earliest_ts: Option<i64> = None;
1394
1395 for record in &records {
1396 let bucket = record.timestamp_bucket;
1397 if earliest_ts.is_none_or(|ts| bucket < ts) {
1398 earliest_ts = Some(bucket);
1399 }
1400
1401 if seen_timestamps.contains(&bucket) {
1402 continue;
1403 }
1404
1405 match parse_candle_record(
1406 record,
1407 bar_type,
1408 price_precision,
1409 size_precision,
1410 ts_init,
1411 ) {
1412 Ok(bar) => {
1413 seen_timestamps.insert(bucket);
1414
1415 if bar.ts_event <= request_time {
1416 page_bars.push(bar);
1417 }
1418 }
1419 Err(e) => log::warn!(
1420 "Failed to parse Derive candle for {bar_type} at {bucket}: {e}",
1421 ),
1422 }
1423 }
1424
1425 total_bars += page_bars.len();
1426 pages.push(page_bars);
1427
1428 if let Some(cap) = limit
1429 && total_bars >= cap
1430 {
1431 break;
1432 }
1433
1434 let Some(earliest) = earliest_ts else {
1435 break;
1436 };
1437
1438 if earliest <= start_ts {
1439 break;
1440 }
1441
1442 current_end = earliest - 1;
1443
1444 if page_count >= DERIVE_CANDLES_MAX_PAGES {
1445 log::warn!(
1446 "Derive bars pagination hit safety cap of {DERIVE_CANDLES_MAX_PAGES} pages for {bar_type}",
1447 );
1448 break;
1449 }
1450 }
1451
1452 let mut bars: Vec<Bar> = Vec::with_capacity(total_bars);
1453 for page in pages.into_iter().rev() {
1454 bars.extend(page);
1455 }
1456
1457 if let Some(cap) = limit
1458 && bars.len() > cap
1459 {
1460 let drop_count = bars.len() - cap;
1461 bars.drain(..drop_count);
1462 }
1463
1464 let response = DataResponse::Bars(BarsResponse::new(
1465 request_id,
1466 client_id,
1467 bar_type,
1468 bars,
1469 start_nanos,
1470 end_nanos,
1471 clock.get_time_ns(),
1472 params,
1473 ));
1474
1475 if let Err(e) = sender.send(DataEvent::Response(response)) {
1476 log::error!("Failed to send Derive bars response: {e}");
1477 }
1478 Ok(())
1479 });
1480
1481 Ok(())
1482 }
1483
1484 fn request_forward_prices(&self, request: RequestForwardPrices) -> anyhow::Result<()> {
1485 let Some(instrument_id) = request.instrument_id else {
1491 anyhow::bail!(
1492 "Derive request_forward_prices requires an `instrument_id`; bulk fetch is not supported",
1493 );
1494 };
1495 let instrument = self
1496 .instruments
1497 .get_cloned(&instrument_id)
1498 .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
1499 anyhow::ensure!(
1500 matches!(instrument, InstrumentAny::CryptoOption(_)),
1501 "Derive forward prices are only meaningful for options (got {instrument_id})",
1502 );
1503 let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1504
1505 let http_client = self.http_client.clone();
1506 let sender = self.data_sender.clone();
1507 let clock = self.clock;
1508 let client_id = request.client_id.unwrap_or(self.client_id);
1509 let request_id = request.request_id;
1510 let venue = request.venue;
1511 let underlying = request.underlying;
1512 let params = request.params;
1513
1514 self.spawn_task("request_forward_prices", async move {
1515 let forwards: Vec<ForwardPrice> = match http_client.get_ticker(&venue_symbol).await {
1520 Ok(ticker) => match ticker.option_pricing.as_ref() {
1521 Some(pricing) => match ticker_ts_event(ticker.timestamp) {
1522 Ok(ts_event) => vec![ForwardPrice::new(
1523 instrument_id,
1524 pricing.forward_price,
1525 Some(underlying.to_string()),
1526 ts_event,
1527 clock.get_time_ns(),
1528 )],
1529 Err(e) => {
1530 log::warn!(
1531 "Derive ticker for {instrument_id} has an invalid timestamp: {e:?}; emitting empty forward prices",
1532 );
1533 Vec::new()
1534 }
1535 },
1536 None => {
1537 log::warn!(
1538 "Derive ticker for {instrument_id} has no option_pricing; emitting empty forward prices",
1539 );
1540 Vec::new()
1541 }
1542 },
1543 Err(e) => {
1544 log::error!(
1545 "Failed to fetch Derive ticker for {instrument_id}: {e:?}; emitting empty forward prices",
1546 );
1547 Vec::new()
1548 }
1549 };
1550
1551 let response = DataResponse::ForwardPrices(ForwardPricesResponse::new(
1552 request_id,
1553 client_id,
1554 venue,
1555 forwards,
1556 clock.get_time_ns(),
1557 params,
1558 ));
1559
1560 if let Err(e) = sender.send(DataEvent::Response(response)) {
1561 log::error!("Failed to send Derive forward prices response: {e}");
1562 }
1563 Ok(())
1564 });
1565
1566 Ok(())
1567 }
1568
1569 fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
1570 let currencies = self.config.currencies.clone();
1571 if currencies.is_empty() {
1572 anyhow::bail!(
1573 "Derive request_instruments requires at least one configured currency \
1574 (DeriveDataClientConfig::currencies)"
1575 );
1576 }
1577
1578 let http_client = self.http_client.clone();
1579 let include_expired = self.config.include_expired;
1580 let instruments_cache = Arc::clone(&self.instruments);
1581 let sender = self.data_sender.clone();
1582 let clock = self.clock;
1583 let venue = self.venue().unwrap_or(*DERIVE_VENUE);
1584 let client_id = request.client_id.unwrap_or(self.client_id);
1585 let request_id = request.request_id;
1586 let start_nanos = datetime_to_unix_nanos(request.start);
1587 let end_nanos = datetime_to_unix_nanos(request.end);
1588 let params = request.params;
1589
1590 self.spawn_task("request_instruments", async move {
1591 let mut all_instruments = Vec::new();
1592
1593 for currency in currencies {
1594 match fetch_instrument_definitions(&http_client, ¤cy, include_expired).await {
1595 Ok(definitions) => {
1596 for instrument in parse_instrument_definitions(definitions) {
1597 cache_instrument(&instruments_cache, &instrument);
1598 all_instruments.push(instrument);
1599 }
1600 }
1601 Err(e) => {
1602 log::error!("Failed to fetch Derive instruments for {currency}: {e:?}");
1603 }
1604 }
1605 }
1606
1607 let response = DataResponse::Instruments(InstrumentsResponse::new(
1608 request_id,
1609 client_id,
1610 venue,
1611 all_instruments,
1612 start_nanos,
1613 end_nanos,
1614 clock.get_time_ns(),
1615 params,
1616 ));
1617
1618 if let Err(e) = sender.send(DataEvent::Response(response)) {
1619 log::error!("Failed to send Derive instruments response: {e}");
1620 }
1621 Ok(())
1622 });
1623
1624 Ok(())
1625 }
1626
1627 fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
1628 let instrument_id = request.instrument_id;
1629 let venue_symbol = format_venue_symbol(&instrument_id)?.to_string();
1630
1631 let http_client = self.http_client.clone();
1632 let instruments_cache = Arc::clone(&self.instruments);
1633 let sender = self.data_sender.clone();
1634 let clock = self.clock;
1635 let client_id = request.client_id.unwrap_or(self.client_id);
1636 let request_id = request.request_id;
1637 let start_nanos = datetime_to_unix_nanos(request.start);
1638 let end_nanos = datetime_to_unix_nanos(request.end);
1639 let params = request.params;
1640
1641 self.spawn_task("request_instrument", async move {
1642 let definition = match http_client.get_instrument(&venue_symbol).await {
1643 Ok(definition) => definition,
1644 Err(e) => {
1645 log::error!("Failed to fetch Derive instrument {instrument_id}: {e:?}");
1646 return Ok(());
1647 }
1648 };
1649
1650 let ts_init = clock.get_time_ns();
1651 let instrument = match parse_derive_instrument_any(&definition, ts_init) {
1652 Ok(Some(instrument)) => instrument,
1653 Ok(None) => {
1654 log::warn!(
1655 "Derive instrument {instrument_id} resolved to an unsupported type ({:?})",
1656 definition.instrument_type,
1657 );
1658 return Ok(());
1659 }
1660 Err(e) => {
1661 log::error!("Failed to parse Derive instrument {instrument_id}: {e}");
1662 return Ok(());
1663 }
1664 };
1665
1666 cache_instrument(&instruments_cache, &instrument);
1667
1668 let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
1669 request_id,
1670 client_id,
1671 instrument.id(),
1672 instrument,
1673 start_nanos,
1674 end_nanos,
1675 clock.get_time_ns(),
1676 params,
1677 )));
1678
1679 if let Err(e) = sender.send(DataEvent::Response(response)) {
1680 log::error!("Failed to send Derive instrument response: {e}");
1681 }
1682 Ok(())
1683 });
1684
1685 Ok(())
1686 }
1687}
1688
1689#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1690enum TickerFeed {
1691 Quote,
1692 Mark,
1693 Index,
1694 Funding,
1695 Greeks,
1696}
1697
1698#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1699enum ChannelOwner {
1700 BookDeltas(InstrumentId),
1701 BookDepth10(InstrumentId),
1702 Ticker {
1703 instrument_id: InstrumentId,
1704 feed: TickerFeed,
1705 },
1706 Trades(InstrumentId),
1707}
1708
1709#[derive(Debug)]
1710struct OwnedChannel {
1711 generation: u64,
1712 channel: Option<String>,
1713}
1714
1715#[derive(Debug, Default)]
1716struct ChannelSubscriptionState {
1717 next_generation: u64,
1718 owners: AHashMap<ChannelOwner, OwnedChannel>,
1719 channels: AHashMap<String, AHashSet<ChannelOwner>>,
1720}
1721
1722#[derive(Debug, Default)]
1723struct ChannelSubscriptionRegistry {
1724 state: Mutex<ChannelSubscriptionState>,
1725 transitions: DashMap<String, Weak<tokio::sync::Mutex<()>>>,
1726}
1727
1728const TRANSITION_GC_THRESHOLD: usize = 256;
1729
1730#[derive(Debug)]
1731struct RemovedSubscription {
1732 channel: Option<String>,
1733 channel_empty: bool,
1734}
1735
1736#[derive(Debug, Clone)]
1737struct SubscriptionDispatchState {
1738 active_book_delta_channels: Arc<AtomicMap<InstrumentId, String>>,
1739 active_book_depth10_channels: Arc<AtomicMap<InstrumentId, String>>,
1740 active_ticker_channels: Arc<AtomicMap<InstrumentId, String>>,
1741 active_quote_subs: Arc<AtomicSet<InstrumentId>>,
1742 active_trade_subs: Arc<AtomicSet<InstrumentId>>,
1743 active_mark_subs: Arc<AtomicSet<InstrumentId>>,
1744 active_index_subs: Arc<AtomicSet<InstrumentId>>,
1745 active_funding_subs: Arc<AtomicSet<InstrumentId>>,
1746 active_greeks_subs: Arc<AtomicSet<InstrumentId>>,
1747 quote_cache: Arc<Mutex<QuoteCache>>,
1748}
1749
1750#[derive(Debug, Clone)]
1751struct SubscriptionLifecycle {
1752 registry: Arc<ChannelSubscriptionRegistry>,
1753 lock: Arc<Mutex<()>>,
1754 dispatch: SubscriptionDispatchState,
1755}
1756
1757#[derive(Debug)]
1758enum ChannelRequest {
1759 Orderbook {
1760 channel: String,
1761 instrument_name: String,
1762 group: String,
1763 depth: String,
1764 },
1765 Ticker {
1766 channel: String,
1767 instrument_name: String,
1768 interval: String,
1769 },
1770 Trades {
1771 channel: String,
1772 instrument_type: String,
1773 currency: String,
1774 },
1775}
1776
1777async fn run_channel_subscribe(
1778 lifecycle: SubscriptionLifecycle,
1779 owner: ChannelOwner,
1780 generation: u64,
1781 request: ChannelRequest,
1782 ws: DeriveWebSocketSubscriptionHandle,
1783) -> anyhow::Result<()> {
1784 let channel = request.channel().to_string();
1785 let transition = lifecycle.registry.transition(&channel);
1786 let _guard = transition.lock().await;
1787
1788 if !lifecycle.is_current_channel(owner, generation, &channel) || ws.has_subscription(&channel) {
1789 return Ok(());
1790 }
1791
1792 if let Err(e) = request.subscribe(&ws).await {
1793 lifecycle.rollback(owner, generation);
1794 log::error!("Failed to subscribe to Derive channel `{channel}`: {e}");
1795 let cleanup_error = match request.unsubscribe(&ws).await {
1796 Ok(()) => None,
1797 Err(cleanup_error) => {
1798 log::error!(
1799 "Failed to clean up uncertain Derive channel `{channel}`: {cleanup_error}",
1800 );
1801 Some(cleanup_error)
1802 }
1803 };
1804
1805 if retain_channel_for_reconnect(
1806 &ws,
1807 &channel,
1808 lifecycle.has_owners(&channel),
1809 cleanup_error.as_ref(),
1810 ) {
1811 log::error!(
1812 "Derive channel `{channel}` remains uncertain and will replay on reconnect",
1813 );
1814 }
1815 return Ok(());
1816 }
1817
1818 if !lifecycle.has_owners(&channel)
1819 && let Err(e) = request.unsubscribe(&ws).await
1820 {
1821 log::error!("Failed to clean up stale Derive channel `{channel}`: {e}");
1822 ws.forget_subscription(&channel);
1823 }
1824 Ok(())
1825}
1826
1827async fn run_channel_unsubscribe(
1828 lifecycle: SubscriptionLifecycle,
1829 request: ChannelRequest,
1830 ws: DeriveWebSocketSubscriptionHandle,
1831) -> anyhow::Result<()> {
1832 let channel = request.channel().to_string();
1833 let transition = lifecycle.registry.transition(&channel);
1834 let _guard = transition.lock().await;
1835
1836 if lifecycle.has_owners(&channel) || !ws.has_subscription(&channel) {
1837 return Ok(());
1838 }
1839
1840 if let Err(e) = request.unsubscribe(&ws).await {
1841 log::error!("Failed to unsubscribe from Derive channel `{channel}`: {e}");
1842 ws.forget_subscription(&channel);
1843 }
1844 Ok(())
1845}
1846
1847impl SubscriptionLifecycle {
1848 fn is_active(&self, owner: ChannelOwner) -> bool {
1849 let _guard = self.lock.lock();
1850 self.registry.state.lock().owners.contains_key(&owner)
1851 }
1852
1853 fn activate(&self, owner: ChannelOwner, channel: Option<&str>) -> Option<u64> {
1854 let _guard = self.lock.lock();
1855 let mut state = self.registry.state.lock();
1856 let generation = state.activate(owner, channel)?;
1857 self.dispatch.activate(owner, channel);
1858 Some(generation)
1859 }
1860
1861 fn attach_channel(&self, owner: ChannelOwner, generation: u64, channel: String) -> bool {
1862 let _guard = self.lock.lock();
1863 self.registry
1864 .state
1865 .lock()
1866 .attach_channel(owner, generation, channel)
1867 }
1868
1869 fn is_current(&self, owner: ChannelOwner, generation: u64) -> bool {
1870 let _guard = self.lock.lock();
1871 self.registry.state.lock().is_current(owner, generation)
1872 }
1873
1874 fn is_current_channel(&self, owner: ChannelOwner, generation: u64, channel: &str) -> bool {
1875 let _guard = self.lock.lock();
1876 self.registry
1877 .state
1878 .lock()
1879 .is_current_channel(owner, generation, channel)
1880 }
1881
1882 fn rollback(&self, owner: ChannelOwner, generation: u64) -> bool {
1883 let _guard = self.lock.lock();
1884 let mut state = self.registry.state.lock();
1885 let Some(removed) = state.remove_if_generation(owner, generation) else {
1886 return false;
1887 };
1888 self.dispatch.deactivate(owner, removed.channel_empty);
1889 true
1890 }
1891
1892 fn remove(&self, owner: ChannelOwner) -> Option<RemovedSubscription> {
1893 let _guard = self.lock.lock();
1894 let mut state = self.registry.state.lock();
1895 let removed = state.remove(owner)?;
1896 self.dispatch.deactivate(owner, removed.channel_empty);
1897 Some(removed)
1898 }
1899
1900 fn has_owners(&self, channel: &str) -> bool {
1901 let _guard = self.lock.lock();
1902 self.registry.state.lock().has_owners(channel)
1903 }
1904}
1905
1906impl SubscriptionDispatchState {
1907 fn activate(&self, owner: ChannelOwner, channel: Option<&str>) {
1908 match owner {
1909 ChannelOwner::BookDeltas(instrument_id) => {
1910 self.active_book_delta_channels.insert(
1911 instrument_id,
1912 channel.expect("book channel present").to_string(),
1913 );
1914 }
1915 ChannelOwner::BookDepth10(instrument_id) => {
1916 self.active_book_depth10_channels.insert(
1917 instrument_id,
1918 channel.expect("book channel present").to_string(),
1919 );
1920 }
1921 ChannelOwner::Ticker {
1922 instrument_id,
1923 feed,
1924 } => {
1925 self.active_ticker_channels.insert(
1926 instrument_id,
1927 channel.expect("ticker channel present").to_string(),
1928 );
1929 self.ticker_subscriptions(feed).insert(instrument_id);
1930 }
1931 ChannelOwner::Trades(instrument_id) => {
1932 self.active_trade_subs.insert(instrument_id);
1933 }
1934 }
1935 }
1936
1937 fn deactivate(&self, owner: ChannelOwner, channel_empty: bool) {
1938 match owner {
1939 ChannelOwner::BookDeltas(instrument_id) => {
1940 self.active_book_delta_channels.remove(&instrument_id);
1941 }
1942 ChannelOwner::BookDepth10(instrument_id) => {
1943 self.active_book_depth10_channels.remove(&instrument_id);
1944 }
1945 ChannelOwner::Ticker {
1946 instrument_id,
1947 feed,
1948 } => {
1949 self.ticker_subscriptions(feed).remove(&instrument_id);
1950 if feed == TickerFeed::Quote {
1951 self.quote_cache.lock().remove(&instrument_id);
1952 }
1953
1954 if channel_empty {
1955 self.active_ticker_channels.remove(&instrument_id);
1956 }
1957 }
1958 ChannelOwner::Trades(instrument_id) => {
1959 self.active_trade_subs.remove(&instrument_id);
1960 }
1961 }
1962 }
1963
1964 fn ticker_subscriptions(&self, feed: TickerFeed) -> &AtomicSet<InstrumentId> {
1965 match feed {
1966 TickerFeed::Quote => &self.active_quote_subs,
1967 TickerFeed::Mark => &self.active_mark_subs,
1968 TickerFeed::Index => &self.active_index_subs,
1969 TickerFeed::Funding => &self.active_funding_subs,
1970 TickerFeed::Greeks => &self.active_greeks_subs,
1971 }
1972 }
1973}
1974
1975impl ChannelSubscriptionRegistry {
1976 fn transition(&self, channel: &str) -> Arc<tokio::sync::Mutex<()>> {
1977 if self.transitions.len() >= TRANSITION_GC_THRESHOLD {
1978 self.transitions
1979 .retain(|_, transition| transition.strong_count() > 0);
1980 }
1981
1982 let mut entry = self.transitions.entry(channel.to_string()).or_default();
1983 if let Some(transition) = entry.value().upgrade() {
1984 return transition;
1985 }
1986
1987 let transition = Arc::new(tokio::sync::Mutex::new(()));
1988 *entry.value_mut() = Arc::downgrade(&transition);
1989 transition
1990 }
1991
1992 fn clear(&self) {
1993 let mut state = self.state.lock();
1994 let next_generation = state.next_generation;
1995 *state = ChannelSubscriptionState {
1996 next_generation,
1997 ..Default::default()
1998 };
1999 }
2000
2001 fn clear_transitions(&self) {
2002 self.transitions.clear();
2003 }
2004}
2005
2006impl ChannelSubscriptionState {
2007 fn activate(&mut self, owner: ChannelOwner, channel: Option<&str>) -> Option<u64> {
2008 if self.owners.contains_key(&owner) {
2009 return None;
2010 }
2011
2012 self.next_generation = self
2013 .next_generation
2014 .checked_add(1)
2015 .expect("subscription generation overflow");
2016 let generation = self.next_generation;
2017
2018 if let Some(channel) = channel {
2019 self.channels
2020 .entry(channel.to_string())
2021 .or_default()
2022 .insert(owner);
2023 }
2024 self.owners.insert(
2025 owner,
2026 OwnedChannel {
2027 generation,
2028 channel: channel.map(ToOwned::to_owned),
2029 },
2030 );
2031 Some(generation)
2032 }
2033
2034 fn attach_channel(&mut self, owner: ChannelOwner, generation: u64, channel: String) -> bool {
2035 let Some(owned) = self.owners.get(&owner) else {
2036 return false;
2037 };
2038
2039 if owned.generation != generation {
2040 return false;
2041 }
2042
2043 if let Some(active_channel) = &owned.channel {
2044 return active_channel == &channel;
2045 }
2046
2047 self.owners.get_mut(&owner).expect("owner present").channel = Some(channel.clone());
2048 self.channels.entry(channel).or_default().insert(owner);
2049 true
2050 }
2051
2052 fn is_current(&self, owner: ChannelOwner, generation: u64) -> bool {
2053 self.owners
2054 .get(&owner)
2055 .is_some_and(|owned| owned.generation == generation)
2056 }
2057
2058 fn is_current_channel(&self, owner: ChannelOwner, generation: u64, channel: &str) -> bool {
2059 self.owners.get(&owner).is_some_and(|owned| {
2060 owned.generation == generation && owned.channel.as_deref() == Some(channel)
2061 })
2062 }
2063
2064 fn remove_if_generation(
2065 &mut self,
2066 owner: ChannelOwner,
2067 generation: u64,
2068 ) -> Option<RemovedSubscription> {
2069 if !self.is_current(owner, generation) {
2070 return None;
2071 }
2072 self.remove(owner)
2073 }
2074
2075 fn remove(&mut self, owner: ChannelOwner) -> Option<RemovedSubscription> {
2076 let owned = self.owners.remove(&owner)?;
2077 let channel_empty = owned.channel.as_ref().is_some_and(|channel| {
2078 let Some(owners) = self.channels.get_mut(channel) else {
2079 return true;
2080 };
2081 owners.remove(&owner);
2082 owners.is_empty()
2083 });
2084
2085 if channel_empty && let Some(channel) = &owned.channel {
2086 self.channels.remove(channel);
2087 }
2088
2089 Some(RemovedSubscription {
2090 channel: owned.channel,
2091 channel_empty,
2092 })
2093 }
2094
2095 fn has_owners(&self, channel: &str) -> bool {
2096 self.channels
2097 .get(channel)
2098 .is_some_and(|owners| !owners.is_empty())
2099 }
2100}
2101
2102impl ChannelRequest {
2103 fn from_channel(channel: &str) -> anyhow::Result<Self> {
2104 if channel.starts_with("orderbook.") {
2105 let (instrument_name, group, depth) = orderbook_channel_parts(channel)?;
2106 return Ok(Self::Orderbook {
2107 channel: channel.to_string(),
2108 instrument_name,
2109 group,
2110 depth,
2111 });
2112 }
2113
2114 if channel.starts_with("ticker_slim.") || channel.starts_with("ticker.") {
2115 let (instrument_name, interval) = ticker_channel_parts(channel)?;
2116 return Ok(Self::Ticker {
2117 channel: channel.to_string(),
2118 instrument_name,
2119 interval,
2120 });
2121 }
2122
2123 if let Some((instrument_type, currency)) = channel
2124 .strip_prefix("trades.")
2125 .and_then(|value| value.split_once('.'))
2126 {
2127 return Ok(Self::Trades {
2128 channel: channel.to_string(),
2129 instrument_type: instrument_type.to_string(),
2130 currency: currency.to_string(),
2131 });
2132 }
2133 anyhow::bail!("invalid Derive subscription channel `{channel}`")
2134 }
2135
2136 fn channel(&self) -> &str {
2137 match self {
2138 Self::Orderbook { channel, .. }
2139 | Self::Ticker { channel, .. }
2140 | Self::Trades { channel, .. } => channel,
2141 }
2142 }
2143
2144 async fn subscribe(&self, ws: &DeriveWebSocketSubscriptionHandle) -> Result<(), DeriveWsError> {
2145 match self {
2146 Self::Orderbook {
2147 instrument_name,
2148 group,
2149 depth,
2150 ..
2151 } => {
2152 ws.subscribe_orderbook(instrument_name, group, depth)
2153 .await?;
2154 }
2155 Self::Ticker {
2156 instrument_name,
2157 interval,
2158 ..
2159 } => ws.subscribe_ticker(instrument_name, interval).await?,
2160 Self::Trades {
2161 instrument_type,
2162 currency,
2163 ..
2164 } => ws.subscribe_trades(instrument_type, currency).await?,
2165 }
2166 Ok(())
2167 }
2168
2169 async fn unsubscribe(
2170 &self,
2171 ws: &DeriveWebSocketSubscriptionHandle,
2172 ) -> Result<(), DeriveWsError> {
2173 match self {
2174 Self::Orderbook {
2175 instrument_name,
2176 group,
2177 depth,
2178 ..
2179 } => {
2180 ws.unsubscribe_orderbook(instrument_name, group, depth)
2181 .await?;
2182 }
2183 Self::Ticker {
2184 instrument_name,
2185 interval,
2186 ..
2187 } => ws.unsubscribe_ticker(instrument_name, interval).await?,
2188 Self::Trades {
2189 instrument_type,
2190 currency,
2191 ..
2192 } => ws.unsubscribe_trades(instrument_type, currency).await?,
2193 }
2194 Ok(())
2195 }
2196}
2197
2198fn retain_channel_for_reconnect(
2199 ws: &DeriveWebSocketSubscriptionHandle,
2200 channel: &str,
2201 has_surviving_owner: bool,
2202 cleanup_error: Option<&DeriveWsError>,
2203) -> bool {
2204 let replay = has_surviving_owner
2205 && cleanup_error.is_some_and(|e| {
2206 matches!(
2207 e,
2208 DeriveWsError::Transport(_)
2209 | DeriveWsError::RequestCancelled { .. }
2210 | DeriveWsError::Timeout { .. }
2211 | DeriveWsError::NotConnected
2212 )
2213 });
2214
2215 if replay {
2216 ws.remember_subscription(channel);
2217 } else {
2218 ws.forget_subscription(channel);
2219 }
2220 replay
2221}
2222
2223impl DeriveDataClient {
2224 async fn join_session_tasks(&self) -> anyhow::Result<()> {
2225 self.session_tasks.begin_shutdown();
2226 self.session_tasks
2227 .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
2228 .await
2229 .map_err(|e| anyhow::anyhow!("Failed to terminate Derive data session tasks: {e}"))?;
2230 Ok(())
2231 }
2232
2233 async fn join_pending_tasks(&self) -> anyhow::Result<()> {
2234 self.pending_tasks.begin_shutdown();
2235 self.pending_tasks
2236 .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
2237 .await
2238 .map_err(|e| anyhow::anyhow!("Failed to terminate Derive data tasks: {e}"))?;
2239 Ok(())
2240 }
2241}
2242
2243fn cache_instrument(
2244 instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
2245 instrument: &InstrumentAny,
2246) {
2247 instruments.insert(instrument.id(), instrument.clone());
2248}
2249
2250fn process_ticker_quote(
2251 msg: &DeriveTickerMsg,
2252 price_precision: u8,
2253 size_precision: u8,
2254 ts_init: UnixNanos,
2255 quote_cache: &mut QuoteCache,
2256) -> anyhow::Result<Option<QuoteTick>> {
2257 let quote = parse_ticker_quote(msg, price_precision, size_precision, ts_init)?;
2258 let (bid_price, bid_size) = quote_side(quote.bid_price, quote.bid_size);
2259 let (ask_price, ask_size) = quote_side(quote.ask_price, quote.ask_size);
2260
2261 match quote_cache.process(
2262 quote.instrument_id,
2263 bid_price,
2264 ask_price,
2265 bid_size,
2266 ask_size,
2267 quote.ts_event,
2268 quote.ts_init,
2269 ) {
2270 Ok(quote) => Ok(Some(quote)),
2271 Err(e) => {
2272 log::debug!(
2273 "Skipping partial Derive ticker quote for {}: {e}",
2274 msg.data.instrument_name(),
2275 );
2276 Ok(None)
2277 }
2278 }
2279}
2280
2281fn quote_side(price: Price, size: Quantity) -> (Option<Price>, Option<Quantity>) {
2282 if price.is_zero() || size.is_zero() {
2283 (None, None)
2284 } else {
2285 (Some(price), Some(size))
2286 }
2287}
2288
2289fn channel_is_active(
2290 channels: &AtomicMap<InstrumentId, String>,
2291 instrument_id: InstrumentId,
2292 channel: &str,
2293) -> bool {
2294 channels
2295 .get_cloned(&instrument_id)
2296 .is_some_and(|active_channel| active_channel == channel)
2297}
2298
2299fn orderbook_channel_parts(channel: &str) -> anyhow::Result<(String, String, String)> {
2300 let rest = channel
2301 .strip_prefix("orderbook.")
2302 .ok_or_else(|| anyhow::anyhow!("invalid Derive orderbook channel `{channel}`"))?;
2303 let mut parts = rest.rsplitn(3, '.');
2304 let depth = parts
2305 .next()
2306 .filter(|value| !value.is_empty())
2307 .ok_or_else(|| anyhow::anyhow!("invalid Derive orderbook channel `{channel}`"))?;
2308 let group = parts
2309 .next()
2310 .filter(|value| !value.is_empty())
2311 .ok_or_else(|| anyhow::anyhow!("invalid Derive orderbook channel `{channel}`"))?;
2312 let instrument_name = parts
2313 .next()
2314 .filter(|value| !value.is_empty())
2315 .ok_or_else(|| anyhow::anyhow!("invalid Derive orderbook channel `{channel}`"))?;
2316
2317 Ok((
2318 instrument_name.to_string(),
2319 group.to_string(),
2320 depth.to_string(),
2321 ))
2322}
2323
2324fn ticker_channel_parts(channel: &str) -> anyhow::Result<(String, String)> {
2325 let rest = channel
2326 .strip_prefix("ticker_slim.")
2327 .or_else(|| channel.strip_prefix("ticker."))
2328 .ok_or_else(|| anyhow::anyhow!("invalid Derive ticker channel `{channel}`"))?;
2329 let (instrument_name, interval) = rest
2330 .rsplit_once('.')
2331 .ok_or_else(|| anyhow::anyhow!("invalid Derive ticker channel `{channel}`"))?;
2332 anyhow::ensure!(
2333 !instrument_name.is_empty() && !interval.is_empty(),
2334 "invalid Derive ticker channel `{channel}`"
2335 );
2336
2337 Ok((instrument_name.to_string(), interval.to_string()))
2338}
2339
2340fn orderbook_group(params: &Option<Params>) -> anyhow::Result<String> {
2341 let group = params
2342 .as_ref()
2343 .and_then(|p| {
2344 p.get_str("group")
2345 .map(ToOwned::to_owned)
2346 .or_else(|| p.get_u64("group").map(|value| value.to_string()))
2347 })
2348 .unwrap_or_else(|| DEFAULT_ORDERBOOK_GROUP.to_string());
2349
2350 DeriveOrderbookGroup::from_str(&group)
2351 .with_context(|| format!("invalid Derive orderbook group `{group}`"))?;
2352 Ok(group)
2353}
2354
2355fn orderbook_depth(depth: Option<usize>, params: &Option<Params>) -> anyhow::Result<String> {
2356 let depth = depth
2357 .map(|value| value.to_string())
2358 .or_else(|| {
2359 params.as_ref().and_then(|p| {
2360 p.get_str("depth")
2361 .map(ToOwned::to_owned)
2362 .or_else(|| p.get_u64("depth").map(|value| value.to_string()))
2363 })
2364 })
2365 .unwrap_or_else(|| DEFAULT_ORDERBOOK_DEPTH.to_string());
2366
2367 DeriveOrderbookDepth::from_str(&depth)
2368 .with_context(|| format!("invalid Derive orderbook depth `{depth}`"))?;
2369 Ok(depth)
2370}
2371
2372fn ticker_interval(params: &Option<Params>) -> anyhow::Result<String> {
2373 let interval = params
2374 .as_ref()
2375 .and_then(|p| {
2376 p.get_str("interval")
2377 .map(ToOwned::to_owned)
2378 .or_else(|| p.get_u64("interval").map(|value| value.to_string()))
2379 })
2380 .unwrap_or_else(|| DEFAULT_TICKER_INTERVAL.to_string());
2381
2382 DeriveTickerInterval::from_str(&interval)
2383 .with_context(|| format!("invalid Derive ticker interval `{interval}`"))?;
2384 Ok(interval)
2385}
2386
2387fn trade_channel(instrument: &InstrumentAny) -> anyhow::Result<String> {
2388 let instrument_type = derive_instrument_type(instrument)?.to_string();
2389 let instrument_id = instrument.id();
2390 let currency = currency_from_instrument_id(&instrument_id)?;
2391 Ok(trades_channel(&instrument_type, currency))
2392}
2393
2394fn derive_instrument_type(instrument: &InstrumentAny) -> anyhow::Result<DeriveInstrumentType> {
2395 match instrument {
2396 InstrumentAny::CryptoPerpetual(_) => Ok(DeriveInstrumentType::Perp),
2397 InstrumentAny::CryptoOption(_) => Ok(DeriveInstrumentType::Option),
2398 InstrumentAny::CurrencyPair(_) => Ok(DeriveInstrumentType::Erc20),
2399 other => anyhow::bail!("unsupported Derive instrument type for trades: {other:?}"),
2400 }
2401}
2402
2403fn currency_from_instrument_id(instrument_id: &InstrumentId) -> anyhow::Result<&str> {
2404 anyhow::ensure!(
2405 instrument_id.venue == *DERIVE_VENUE,
2406 "instrument ID `{instrument_id}` is not for venue {}",
2407 DERIVE_VENUE.as_str(),
2408 );
2409
2410 instrument_id
2411 .symbol
2412 .as_str()
2413 .split_once('-')
2414 .and_then(|(currency, _)| (!currency.is_empty()).then_some(currency))
2415 .ok_or_else(|| anyhow::anyhow!("cannot derive currency from {instrument_id}"))
2416}
2417
2418fn truncated_payload_snippet(raw: &str) -> String {
2422 const MAX_LEN: usize = 512;
2423 if raw.len() <= MAX_LEN {
2424 return raw.to_string();
2425 }
2426 let mut end = MAX_LEN;
2427 while end > 0 && !raw.is_char_boundary(end) {
2428 end -= 1;
2429 }
2430 format!("{}...(truncated)", &raw[..end])
2431}
2432
2433#[cfg(test)]
2434mod tests {
2435 use std::{path::PathBuf, time::Duration};
2436
2437 use nautilus_common::{live::runner::replace_data_event_sender, testing::wait_until_async};
2438 use nautilus_core::{UUID4, UnixNanos};
2439 use nautilus_model::{
2440 identifiers::InstrumentId,
2441 types::{Price, Quantity},
2442 };
2443 use rstest::rstest;
2444 use serde_json::{Value, json};
2445
2446 use super::*;
2447 use crate::{
2448 common::{
2449 consts::DERIVE_CLIENT_ID, enums::DeriveEnvironment, parse::parse_derive_instrument_any,
2450 },
2451 http::models::DeriveInstrument,
2452 websocket::{DeriveWsFrame, WsSubscriptionPayload},
2453 };
2454
2455 fn data_path() -> PathBuf {
2456 PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("test_data")
2457 }
2458
2459 fn load_json(filename: &str) -> Value {
2460 let content = std::fs::read_to_string(data_path().join(filename))
2461 .unwrap_or_else(|_| panic!("failed to read {filename}"));
2462 serde_json::from_str(&content).expect("invalid json")
2463 }
2464
2465 #[rstest]
2466 fn test_truncated_payload_snippet_short_payload_is_unchanged() {
2467 let short = json!({"ok": true}).to_string();
2468 assert_eq!(truncated_payload_snippet(&short), r#"{"ok":true}"#);
2469 }
2470
2471 #[rstest]
2472 fn test_truncated_payload_snippet_truncates_long_ascii_payload() {
2473 let big = json!({"msg": "x".repeat(1024)}).to_string();
2474 let snippet = truncated_payload_snippet(&big);
2475 assert!(snippet.ends_with("...(truncated)"));
2476 assert!(snippet.len() <= 512 + "...(truncated)".len());
2478 }
2479
2480 #[rstest]
2481 fn test_truncated_payload_snippet_handles_multibyte_at_boundary() {
2482 let value: String = format!("x{}", "\u{00E9}".repeat(1024));
2490 let big = json!({"a": value});
2491 let raw = big.to_string();
2492 assert!(
2493 !raw.is_char_boundary(512),
2494 "test premise: 512 must be mid-codepoint",
2495 );
2496
2497 let snippet = truncated_payload_snippet(&raw);
2498 assert!(snippet.ends_with("...(truncated)"));
2499 let body_len = snippet.len() - "...(truncated)".len();
2500 assert!(body_len <= 512);
2501 }
2502
2503 fn subscription_payload(channel: &str, data: &Value) -> WsSubscriptionPayload {
2504 let frame = json!({
2505 "jsonrpc": "2.0",
2506 "method": "subscription",
2507 "params": {
2508 "channel": channel,
2509 "data": data
2510 }
2511 });
2512
2513 match DeriveWsFrame::parse(&frame.to_string()).unwrap() {
2514 DeriveWsFrame::Subscription(payload) => payload,
2515 other => panic!("expected subscription frame, was {other:?}"),
2516 }
2517 }
2518
2519 fn make_ctx(
2520 instrument: Option<InstrumentAny>,
2521 ) -> (
2522 WsMessageContext,
2523 tokio::sync::mpsc::UnboundedReceiver<DataEvent>,
2524 ) {
2525 let (data_sender, data_rx) = tokio::sync::mpsc::unbounded_channel();
2526 let instruments = Arc::new(AtomicMap::new());
2527 if let Some(instrument) = instrument {
2528 cache_instrument(&instruments, &instrument);
2529 }
2530
2531 (
2532 WsMessageContext {
2533 clock: get_atomic_clock_realtime(),
2534 data_sender,
2535 instruments,
2536 active_book_delta_channels: Arc::new(AtomicMap::new()),
2537 active_book_depth10_channels: Arc::new(AtomicMap::new()),
2538 active_ticker_channels: Arc::new(AtomicMap::new()),
2539 active_quote_subs: Arc::new(AtomicSet::new()),
2540 active_trade_subs: Arc::new(AtomicSet::new()),
2541 active_mark_subs: Arc::new(AtomicSet::new()),
2542 active_index_subs: Arc::new(AtomicSet::new()),
2543 active_funding_subs: Arc::new(AtomicSet::new()),
2544 active_greeks_subs: Arc::new(AtomicSet::new()),
2545 subscription_lock: Arc::new(Mutex::new(())),
2546 quote_cache: Arc::new(Mutex::new(QuoteCache::new())),
2547 },
2548 data_rx,
2549 )
2550 }
2551
2552 fn perp_instrument() -> InstrumentAny {
2553 parse_derive_instrument_any(&perp_definition("ETH-PERP", "ETH"), UnixNanos::from(1))
2554 .unwrap()
2555 .unwrap()
2556 }
2557
2558 fn btc_perp_instrument() -> InstrumentAny {
2559 parse_derive_instrument_any(&perp_definition("BTC-PERP", "BTC"), UnixNanos::from(1))
2560 .unwrap()
2561 .unwrap()
2562 }
2563
2564 fn perp_definition(name: &str, currency: &str) -> DeriveInstrument {
2565 let mut value = load_json("perps/instrument_eth.json");
2566 value["base_currency"] = json!(currency);
2567 value["instrument_name"] = json!(name);
2568 value["perp_details"]["index"] = json!(format!("{currency}-USD"));
2569
2570 serde_json::from_value(value).unwrap()
2571 }
2572
2573 fn option_instrument() -> InstrumentAny {
2574 let definition: DeriveInstrument =
2575 serde_json::from_value(load_json("options/instrument_eth.json")).unwrap();
2576 parse_derive_instrument_any(&definition, UnixNanos::from(1))
2577 .unwrap()
2578 .unwrap()
2579 }
2580
2581 fn spot_instrument() -> InstrumentAny {
2582 let definition: DeriveInstrument =
2583 serde_json::from_value(load_json("spot/instrument_eth.json")).unwrap();
2584 parse_derive_instrument_any(&definition, UnixNanos::from(1))
2585 .unwrap()
2586 .unwrap()
2587 }
2588
2589 fn ticker_json(timestamp: i64) -> Value {
2590 let mut value = load_json("perps/ws_ticker_eth.json");
2591 value["timestamp"] = json!(timestamp);
2592 value
2593 }
2594
2595 fn option_ticker_json(timestamp: i64) -> Value {
2596 let mut value = load_json("options/http_ticker_eth_snapshot.json");
2597 value["timestamp"] = json!(timestamp);
2598 value
2599 }
2600
2601 fn spot_ticker_slim_json() -> Value {
2602 load_json("spot/ws_ticker_slim_eth.json")
2603 }
2604
2605 fn orderbook_json() -> Value {
2606 load_json("perps/ws_orderbook_eth.json")
2607 }
2608
2609 fn trade_json(instrument_name: &str, trade_id: &str) -> Value {
2610 let mut value = load_json("perps/ws_trade_eth.json");
2611 value["instrument_name"] = json!(instrument_name);
2612 value["trade_id"] = json!(trade_id);
2613 value
2614 }
2615
2616 #[rstest]
2617 fn test_handle_ticker_subscription_emits_quote_with_instrument_precision() {
2618 let instrument = perp_instrument();
2619 let instrument_id = instrument.id();
2620 let (ctx, mut rx) = make_ctx(Some(instrument));
2621 ctx.active_ticker_channels
2622 .insert(instrument_id, "ticker_slim.ETH-PERP.1000".to_string());
2623 ctx.active_quote_subs.insert(instrument_id);
2624 let payload = subscription_payload(
2625 "ticker_slim.ETH-PERP.1000",
2626 &json!({
2627 "timestamp": 1_700_000_000_010_i64,
2628 "instrument_ticker": ticker_json(1_700_000_000_000)
2629 }),
2630 );
2631
2632 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2633
2634 match rx.try_recv().unwrap() {
2635 DataEvent::Data(Data::Quote(quote)) => {
2636 assert_eq!(quote.instrument_id, instrument_id);
2637 assert_eq!(quote.bid_price, Price::from("3500.00"));
2638 assert_eq!(quote.ask_price, Price::from("3501.00"));
2639 assert_eq!(quote.bid_size, Quantity::from("1.000"));
2640 assert_eq!(quote.ask_size, Quantity::from("2.000"));
2641 assert_eq!(quote.bid_price.precision, 2);
2642 assert_eq!(quote.bid_size.precision, 3);
2643 }
2644 other => panic!("expected quote data event, was {other:?}"),
2645 }
2646 }
2647
2648 #[rstest]
2649 fn test_handle_ticker_partial_quote_without_cache_emits_no_quote() {
2650 let instrument = spot_instrument();
2651 let instrument_id = instrument.id();
2652 let channel = "ticker_slim.ETH-USDC.1000";
2653 let (ctx, mut rx) = make_ctx(Some(instrument));
2654 install_ticker(&ctx, instrument_id, channel);
2655 ctx.active_quote_subs.insert(instrument_id);
2656 let payload = subscription_payload(channel, &spot_ticker_slim_json());
2657
2658 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2659
2660 assert!(rx.try_recv().is_err());
2661 }
2662
2663 #[rstest]
2664 fn test_handle_ticker_partial_quote_uses_cached_side() {
2665 let instrument = spot_instrument();
2666 let instrument_id = instrument.id();
2667 let channel = "ticker_slim.ETH-USDC.1000";
2668 let (ctx, mut rx) = make_ctx(Some(instrument));
2669 let cached_quote = QuoteTick::new(
2670 instrument_id,
2671 Price::from("0.1"),
2672 Price::from("0.3"),
2673 Quantity::from("10.00"),
2674 Quantity::from("20.00"),
2675 UnixNanos::from(1),
2676 UnixNanos::from(1),
2677 );
2678 ctx.quote_cache.lock().insert(instrument_id, cached_quote);
2679 install_ticker(&ctx, instrument_id, channel);
2680 ctx.active_quote_subs.insert(instrument_id);
2681 let payload = subscription_payload(channel, &spot_ticker_slim_json());
2682
2683 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2684
2685 match rx.try_recv().unwrap() {
2686 DataEvent::Data(Data::Quote(quote)) => {
2687 assert_eq!(quote.instrument_id, instrument_id);
2688 assert_eq!(quote.bid_price, Price::from("0.2"));
2689 assert_eq!(quote.ask_price, Price::from("0.3"));
2690 assert_eq!(quote.bid_size, Quantity::from("45.00"));
2691 assert_eq!(quote.ask_size, Quantity::from("20.00"));
2692 }
2693 other => panic!("expected quote data event, was {other:?}"),
2694 }
2695 }
2696
2697 #[rstest]
2698 fn test_handle_reconnected_clears_quote_cache() {
2699 let instrument = spot_instrument();
2700 let instrument_id = instrument.id();
2701 let channel = "ticker_slim.ETH-USDC.1000";
2702 let (ctx, mut rx) = make_ctx(Some(instrument));
2703 let cached_quote = QuoteTick::new(
2704 instrument_id,
2705 Price::from("0.1"),
2706 Price::from("0.3"),
2707 Quantity::from("10.00"),
2708 Quantity::from("20.00"),
2709 UnixNanos::from(1),
2710 UnixNanos::from(1),
2711 );
2712 ctx.quote_cache.lock().insert(instrument_id, cached_quote);
2713 install_ticker(&ctx, instrument_id, channel);
2714 ctx.active_quote_subs.insert(instrument_id);
2715
2716 DeriveDataClient::handle_ws_message(DeriveWsMessage::Reconnected, &ctx);
2717 let payload = subscription_payload(channel, &spot_ticker_slim_json());
2718 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2719
2720 assert!(rx.try_recv().is_err());
2721 }
2722
2723 #[rstest]
2724 #[tokio::test]
2725 async fn test_session_recovery_failure_marks_client_disconnected() {
2726 let (data_tx, _data_rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
2727 replace_data_event_sender(data_tx);
2728 let config = DeriveDataClientConfig {
2729 environment: DeriveEnvironment::Mainnet,
2730 ..Default::default()
2731 };
2732 let client = DeriveDataClient::new(*DERIVE_CLIENT_ID, config).unwrap();
2733 let (ws_tx, ws_rx) = tokio::sync::mpsc::unbounded_channel();
2734 client.is_connected.store(true, Ordering::Release);
2735 client.spawn_stream_task(ws_rx).unwrap();
2736
2737 ws_tx
2738 .send(DeriveWsMessage::SessionRecoveryFailed(
2739 "subscription replay failed".to_string(),
2740 ))
2741 .unwrap();
2742 wait_until_async(|| async { !client.is_connected() }, Duration::from_secs(2)).await;
2743
2744 assert!(!client.is_connected());
2745
2746 client.cancellation_token.cancel();
2747 }
2748
2749 #[rstest]
2750 fn test_handle_orderbook_subscription_emits_snapshot_deltas() {
2751 let instrument = perp_instrument();
2752 let instrument_id = instrument.id();
2753 let (ctx, mut rx) = make_ctx(Some(instrument));
2754 ctx.active_book_delta_channels
2755 .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
2756 let payload = subscription_payload("orderbook.ETH-PERP.1.10", &orderbook_json());
2757
2758 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2759
2760 match rx.try_recv().unwrap() {
2761 DataEvent::Data(Data::Deltas(deltas)) => {
2762 assert_eq!(deltas.instrument_id, instrument_id);
2763 assert_eq!(deltas.deltas.len(), 3);
2764 assert_eq!(deltas.deltas[1].order.price, Price::from("3500.00"));
2765 assert_eq!(deltas.deltas[1].order.size, Quantity::from("1.000"));
2766 assert_eq!(deltas.deltas[2].order.price, Price::from("3501.00"));
2767 assert_eq!(deltas.deltas[2].order.size, Quantity::from("2.000"));
2768 }
2769 other => panic!("expected deltas data event, was {other:?}"),
2770 }
2771 }
2772
2773 #[rstest]
2774 fn test_handle_orderbook_subscription_emits_for_depth10_subscription() {
2775 let instrument = perp_instrument();
2776 let instrument_id = instrument.id();
2777 let (ctx, mut rx) = make_ctx(Some(instrument));
2778 ctx.active_book_depth10_channels
2779 .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
2780 let payload = subscription_payload("orderbook.ETH-PERP.1.10", &orderbook_json());
2781
2782 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2783
2784 match rx.try_recv().unwrap() {
2785 DataEvent::Data(Data::Depth10(depth)) => {
2786 assert_eq!(depth.instrument_id, instrument_id);
2787 assert_eq!(depth.bids[0].price, Price::from("3500.00"));
2788 assert_eq!(depth.bids[0].size, Quantity::from("1.000"));
2789 assert_eq!(depth.asks[0].price, Price::from("3501.00"));
2790 assert_eq!(depth.asks[0].size, Quantity::from("2.000"));
2791 }
2792 other => panic!("expected depth10 data event, was {other:?}"),
2793 }
2794 }
2795
2796 #[rstest]
2797 fn test_orderbook_frame_ignored_for_inactive_channel() {
2798 let instrument = perp_instrument();
2799 let instrument_id = instrument.id();
2800 let (ctx, mut rx) = make_ctx(Some(instrument));
2801 ctx.active_book_delta_channels
2802 .insert(instrument_id, "orderbook.ETH-PERP.1.20".to_string());
2803 let payload = subscription_payload("orderbook.ETH-PERP.1.10", &orderbook_json());
2804
2805 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2806
2807 assert!(rx.try_recv().is_err());
2808 }
2809
2810 #[rstest]
2811 fn test_handle_trades_subscription_filters_and_emits_active_instrument() {
2812 let instrument = perp_instrument();
2813 let other = btc_perp_instrument();
2814 let instrument_id = instrument.id();
2815 let (ctx, mut rx) = make_ctx(Some(instrument));
2816 cache_instrument(&ctx.instruments, &other);
2817 ctx.active_trade_subs.insert(instrument_id);
2818 let payload = subscription_payload(
2819 "trades.perp.ETH",
2820 &json!([
2821 trade_json("ETH-PERP", "trade-1"),
2822 trade_json("BTC-PERP", "trade-2")
2823 ]),
2824 );
2825
2826 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2827
2828 match rx.try_recv().unwrap() {
2829 DataEvent::Data(Data::Trade(trade)) => {
2830 assert_eq!(trade.instrument_id, instrument_id);
2831 assert_eq!(trade.trade_id.to_string(), "trade-1");
2832 assert_eq!(trade.price, Price::from("3500.00"));
2833 assert_eq!(trade.size, Quantity::from("1.000"));
2834 }
2835 other => panic!("expected trade data event, was {other:?}"),
2836 }
2837 assert!(rx.try_recv().is_err());
2838 }
2839
2840 #[rstest]
2841 fn test_handle_subscription_without_cached_instrument_emits_no_event() {
2842 let (ctx, mut rx) = make_ctx(None);
2843 let instrument_id = InstrumentId::from("ETH-PERP.DERIVE");
2844 ctx.active_ticker_channels
2845 .insert(instrument_id, "ticker_slim.ETH-PERP.1000".to_string());
2846 ctx.active_quote_subs.insert(instrument_id);
2847 let payload =
2848 subscription_payload("ticker_slim.ETH-PERP.1000", &ticker_json(1_700_000_000_000));
2849
2850 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2851
2852 assert!(rx.try_recv().is_err());
2853 }
2854
2855 #[rstest]
2856 fn test_ticker_frame_ignored_without_quote_subscription() {
2857 let instrument = perp_instrument();
2858 let (ctx, mut rx) = make_ctx(Some(instrument));
2859 let payload =
2860 subscription_payload("ticker_slim.ETH-PERP.1000", &ticker_json(1_700_000_000_000));
2861
2862 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2863
2864 assert!(rx.try_recv().is_err());
2865 }
2866
2867 #[rstest]
2868 fn test_ticker_frame_ignored_for_inactive_channel() {
2869 let instrument = perp_instrument();
2870 let instrument_id = instrument.id();
2871 let (ctx, mut rx) = make_ctx(Some(instrument));
2872 ctx.active_ticker_channels
2873 .insert(instrument_id, "ticker_slim.ETH-PERP.100".to_string());
2874 ctx.active_quote_subs.insert(instrument_id);
2875 let payload =
2876 subscription_payload("ticker_slim.ETH-PERP.1000", &ticker_json(1_700_000_000_000));
2877
2878 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
2879
2880 assert!(rx.try_recv().is_err());
2881 }
2882
2883 #[rstest]
2884 fn test_trade_channel_uses_instrument_type_and_currency() {
2885 let instrument = perp_instrument();
2886
2887 assert_eq!(trade_channel(&instrument).unwrap(), "trades.perp.ETH");
2888 }
2889
2890 #[rstest]
2891 fn test_trade_channel_uses_erc20_for_spot() {
2892 let instrument = spot_instrument();
2893
2894 assert_eq!(
2895 derive_instrument_type(&instrument).unwrap(),
2896 DeriveInstrumentType::Erc20
2897 );
2898 assert_eq!(trade_channel(&instrument).unwrap(), "trades.erc20.ETH");
2899 }
2900
2901 #[rstest]
2902 fn test_param_defaults_match_derive_public_channels() {
2903 assert_eq!(orderbook_group(&None).unwrap(), DEFAULT_ORDERBOOK_GROUP);
2904 assert_eq!(
2905 orderbook_depth(None, &None).unwrap(),
2906 DEFAULT_ORDERBOOK_DEPTH
2907 );
2908 assert_eq!(ticker_interval(&None).unwrap(), DEFAULT_TICKER_INTERVAL);
2909 }
2910
2911 #[rstest]
2912 fn test_orderbook_channel_parts_splits_from_right() {
2913 assert_eq!(
2914 orderbook_channel_parts("orderbook.ETH.TEST-PERP.10.100").unwrap(),
2915 (
2916 "ETH.TEST-PERP".to_string(),
2917 "10".to_string(),
2918 "100".to_string()
2919 )
2920 );
2921 }
2922
2923 #[rstest]
2924 fn test_ticker_channel_parts_splits_from_right() {
2925 assert_eq!(
2926 ticker_channel_parts("ticker_slim.ETH.TEST-PERP.1000").unwrap(),
2927 ("ETH.TEST-PERP".to_string(), "1000".to_string())
2928 );
2929 }
2930
2931 #[rstest]
2932 fn test_ticker_channel_parts_accepts_legacy_ticker_channel() {
2933 assert_eq!(
2934 ticker_channel_parts("ticker.ETH.TEST-PERP.1000").unwrap(),
2935 ("ETH.TEST-PERP".to_string(), "1000".to_string())
2936 );
2937 }
2938
2939 #[rstest]
2940 fn test_stale_generation_rollback_preserves_resubscription() {
2941 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
2942 replace_data_event_sender(tx);
2943 let client = DeriveDataClient::new(
2944 *DERIVE_CLIENT_ID,
2945 DeriveDataClientConfig {
2946 environment: DeriveEnvironment::Mainnet,
2947 ..Default::default()
2948 },
2949 )
2950 .unwrap();
2951 let lifecycle = client.subscription_lifecycle();
2952 let instrument_id = InstrumentId::from("ETH-PERP.DERIVE");
2953 let owner = ChannelOwner::Ticker {
2954 instrument_id,
2955 feed: TickerFeed::Quote,
2956 };
2957 let channel = "ticker_slim.ETH-PERP.1000".to_string();
2958
2959 let first_generation = lifecycle.activate(owner, Some(&channel)).unwrap();
2960 lifecycle.remove(owner).unwrap();
2961 let second_generation = lifecycle.activate(owner, Some(&channel)).unwrap();
2962
2963 assert!(!lifecycle.rollback(owner, first_generation));
2964 assert!(lifecycle.is_current_channel(owner, second_generation, &channel));
2965 assert!(client.active_quote_subs.contains(&instrument_id));
2966 assert!(channel_is_active(
2967 &client.active_ticker_channels,
2968 instrument_id,
2969 &channel,
2970 ));
2971 }
2972
2973 #[rstest]
2974 fn test_subscription_registry_clear_does_not_reuse_generation() {
2975 let registry = ChannelSubscriptionRegistry::default();
2976 let owner = ChannelOwner::Trades(InstrumentId::from("ETH-20260627-3500-C.DERIVE"));
2977 let first_generation = registry.state.lock().activate(owner, None).unwrap();
2978
2979 registry.clear();
2980 let second_generation = registry.state.lock().activate(owner, None).unwrap();
2981
2982 assert!(second_generation > first_generation);
2983 }
2984
2985 #[rstest]
2986 fn test_subscription_registry_clear_preserves_inflight_transition() {
2987 let registry = ChannelSubscriptionRegistry::default();
2988 let first = registry.transition("ticker_slim.ETH-PERP.1000");
2989
2990 registry.clear();
2991 let second = registry.transition("ticker_slim.ETH-PERP.1000");
2992
2993 assert!(Arc::ptr_eq(&first, &second));
2994 }
2995
2996 #[rstest]
2997 fn test_subscription_transition_registry_collects_inactive_channels() {
2998 let registry = ChannelSubscriptionRegistry::default();
2999 let live = registry.transition("ticker_slim.ETH-PERP.1000");
3000
3001 for index in 0..TRANSITION_GC_THRESHOLD + 16 {
3002 drop(registry.transition(&format!("ticker_slim.ETH-OPTION-{index}.1000")));
3003 }
3004 let same_live = registry.transition("ticker_slim.ETH-PERP.1000");
3005
3006 assert!(Arc::ptr_eq(&live, &same_live));
3007 assert!(registry.transitions.len() < TRANSITION_GC_THRESHOLD);
3008 }
3009
3010 #[rstest]
3011 fn test_ambiguous_cleanup_retains_survivor_for_reconnect() {
3012 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3013 replace_data_event_sender(tx);
3014 let client = DeriveDataClient::new(
3015 *DERIVE_CLIENT_ID,
3016 DeriveDataClientConfig {
3017 environment: DeriveEnvironment::Mainnet,
3018 ..Default::default()
3019 },
3020 )
3021 .unwrap();
3022 let lifecycle = client.subscription_lifecycle();
3023 let instrument_id = InstrumentId::from("ETH-20260627-3600-C.DERIVE");
3024 let owner = ChannelOwner::Trades(instrument_id);
3025 lifecycle.activate(owner, Some("trades.option.ETH"));
3026 let error = DeriveWsError::Timeout {
3027 method: "unsubscribe".to_string(),
3028 };
3029 let ws = client.ws_handle();
3030
3031 let retained = retain_channel_for_reconnect(&ws, "trades.option.ETH", true, Some(&error));
3032
3033 assert!(retained);
3034 assert!(lifecycle.is_active(owner));
3035 assert!(client.active_trade_subs.contains(&instrument_id));
3036 assert!(ws.has_subscription("trades.option.ETH"));
3037 }
3038
3039 #[rstest]
3040 fn test_explicit_cleanup_rejection_preserves_surviving_channel_owner() {
3041 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3042 replace_data_event_sender(tx);
3043 let client = DeriveDataClient::new(
3044 *DERIVE_CLIENT_ID,
3045 DeriveDataClientConfig {
3046 environment: DeriveEnvironment::Mainnet,
3047 ..Default::default()
3048 },
3049 )
3050 .unwrap();
3051 let lifecycle = client.subscription_lifecycle();
3052 let instrument_id = InstrumentId::from("ETH-20260627-3600-C.DERIVE");
3053 let owner = ChannelOwner::Trades(instrument_id);
3054 lifecycle.activate(owner, Some("trades.option.ETH"));
3055 let error = DeriveWsError::JsonRpc {
3056 code: -32603,
3057 message: "not subscribed".to_string(),
3058 data: None,
3059 };
3060
3061 let ws = client.ws_handle();
3062 ws.remember_subscription("trades.option.ETH");
3063
3064 let retained = retain_channel_for_reconnect(&ws, "trades.option.ETH", true, Some(&error));
3065
3066 assert!(!retained);
3067 assert!(lifecycle.is_active(owner));
3068 assert!(client.active_trade_subs.contains(&instrument_id));
3069 assert!(!ws.has_subscription("trades.option.ETH"));
3070 }
3071
3072 #[rstest]
3073 fn test_ambiguous_cleanup_without_survivor_is_not_replayed() {
3074 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3075 replace_data_event_sender(tx);
3076 let client = DeriveDataClient::new(
3077 *DERIVE_CLIENT_ID,
3078 DeriveDataClientConfig {
3079 environment: DeriveEnvironment::Mainnet,
3080 ..Default::default()
3081 },
3082 )
3083 .unwrap();
3084 let ws = client.ws_handle();
3085 ws.remember_subscription("trades.option.ETH");
3086 let error = DeriveWsError::Timeout {
3087 method: "unsubscribe".to_string(),
3088 };
3089
3090 let retained = retain_channel_for_reconnect(&ws, "trades.option.ETH", false, Some(&error));
3091
3092 assert!(!retained);
3093 assert!(!ws.has_subscription("trades.option.ETH"));
3094 }
3095
3096 #[rstest]
3097 fn test_unsubscribe_quotes_prunes_cached_quote() {
3098 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3099 replace_data_event_sender(tx);
3100 let mut client = DeriveDataClient::new(
3101 *DERIVE_CLIENT_ID,
3102 DeriveDataClientConfig {
3103 environment: DeriveEnvironment::Mainnet,
3104 ..Default::default()
3105 },
3106 )
3107 .unwrap();
3108 let instrument_id = InstrumentId::from("ETH-PERP.DERIVE");
3109 let owner = ChannelOwner::Ticker {
3110 instrument_id,
3111 feed: TickerFeed::Quote,
3112 };
3113 client
3114 .subscription_lifecycle()
3115 .activate(owner, Some("ticker_slim.ETH-PERP.1000"));
3116 client.quote_cache.lock().insert(
3117 instrument_id,
3118 QuoteTick::new(
3119 instrument_id,
3120 Price::from("3500.00"),
3121 Price::from("3501.00"),
3122 Quantity::from("1.000"),
3123 Quantity::from("2.000"),
3124 UnixNanos::from(1),
3125 UnixNanos::from(1),
3126 ),
3127 );
3128 let command = UnsubscribeQuotes::new(
3129 instrument_id,
3130 Some(*DERIVE_CLIENT_ID),
3131 None,
3132 UUID4::new(),
3133 UnixNanos::default(),
3134 None,
3135 None,
3136 );
3137
3138 client.unsubscribe_quotes(&command).unwrap();
3139
3140 assert!(!client.quote_cache.lock().contains(&instrument_id));
3141 }
3142
3143 #[rstest]
3144 #[tokio::test]
3145 async fn test_unsubscribe_trades_uses_recorded_channel_without_cached_instrument() {
3146 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3147 replace_data_event_sender(tx);
3148 let mut client = DeriveDataClient::new(
3149 *DERIVE_CLIENT_ID,
3150 DeriveDataClientConfig {
3151 environment: DeriveEnvironment::Mainnet,
3152 ..Default::default()
3153 },
3154 )
3155 .unwrap();
3156 let instrument_id = InstrumentId::from("ETH-20260627-3500-C.DERIVE");
3157 let owner = ChannelOwner::Trades(instrument_id);
3158 let lifecycle = client.subscription_lifecycle();
3159 let generation = lifecycle.activate(owner, None).unwrap();
3160 assert!(lifecycle.attach_channel(owner, generation, "trades.option.ETH".to_string(),));
3161 let command = UnsubscribeTrades::new(
3162 instrument_id,
3163 Some(*DERIVE_CLIENT_ID),
3164 None,
3165 UUID4::new(),
3166 UnixNanos::default(),
3167 None,
3168 None,
3169 );
3170
3171 client.unsubscribe_trades(&command).unwrap();
3172
3173 wait_until_async(
3174 || {
3175 let registry = Arc::clone(&client.channel_subscriptions);
3176 async move { registry.transitions.contains_key("trades.option.ETH") }
3177 },
3178 Duration::from_secs(1),
3179 )
3180 .await;
3181
3182 assert!(!lifecycle.is_active(owner));
3183 assert!(!client.active_trade_subs.contains(&instrument_id));
3184 }
3185
3186 fn perp_ticker_payload(instrument_id: InstrumentId) -> WsSubscriptionPayload {
3187 let channel = "ticker_slim.ETH-PERP.1000";
3188 let payload = subscription_payload(
3189 channel,
3190 &json!({
3191 "timestamp": 1_700_000_000_010_i64,
3192 "instrument_ticker": ticker_json(1_700_000_000_000)
3193 }),
3194 );
3195 assert_eq!(payload.channel.as_str(), channel);
3196 let _ = instrument_id;
3197 payload
3198 }
3199
3200 fn install_ticker(ctx: &WsMessageContext, instrument_id: InstrumentId, channel: &str) {
3201 ctx.active_ticker_channels
3202 .insert(instrument_id, channel.to_string());
3203 }
3204
3205 #[rstest]
3206 fn test_ticker_emits_mark_price_when_mark_subscribed() {
3207 let instrument = perp_instrument();
3208 let instrument_id = instrument.id();
3209 let (ctx, mut rx) = make_ctx(Some(instrument));
3210 install_ticker(&ctx, instrument_id, "ticker_slim.ETH-PERP.1000");
3211 ctx.active_mark_subs.insert(instrument_id);
3212
3213 DeriveDataClient::handle_ws_message(
3214 DeriveWsMessage::Subscription(perp_ticker_payload(instrument_id)),
3215 &ctx,
3216 );
3217
3218 match rx.try_recv().unwrap() {
3219 DataEvent::Data(Data::MarkPrice(mark)) => {
3220 assert_eq!(mark.instrument_id, instrument_id);
3221 assert_eq!(mark.value, Price::from("3500.50"));
3222 }
3223 other => panic!("expected MarkPriceUpdate, was {other:?}"),
3224 }
3225 assert!(rx.try_recv().is_err());
3226 }
3227
3228 #[rstest]
3229 fn test_ticker_emits_index_price_when_index_subscribed() {
3230 let instrument = perp_instrument();
3231 let instrument_id = instrument.id();
3232 let (ctx, mut rx) = make_ctx(Some(instrument));
3233 install_ticker(&ctx, instrument_id, "ticker_slim.ETH-PERP.1000");
3234 ctx.active_index_subs.insert(instrument_id);
3235
3236 DeriveDataClient::handle_ws_message(
3237 DeriveWsMessage::Subscription(perp_ticker_payload(instrument_id)),
3238 &ctx,
3239 );
3240
3241 match rx.try_recv().unwrap() {
3242 DataEvent::Data(Data::IndexPrice(index)) => {
3243 assert_eq!(index.instrument_id, instrument_id);
3244 assert_eq!(index.value, Price::from("3500.00"));
3245 }
3246 other => panic!("expected IndexPriceUpdate, was {other:?}"),
3247 }
3248 assert!(rx.try_recv().is_err());
3249 }
3250
3251 #[rstest]
3252 fn test_ticker_emits_funding_rate_for_perp_when_subscribed() {
3253 let instrument = perp_instrument();
3254 let instrument_id = instrument.id();
3255 let (ctx, mut rx) = make_ctx(Some(instrument));
3256 install_ticker(&ctx, instrument_id, "ticker_slim.ETH-PERP.1000");
3257 ctx.active_funding_subs.insert(instrument_id);
3258
3259 DeriveDataClient::handle_ws_message(
3260 DeriveWsMessage::Subscription(perp_ticker_payload(instrument_id)),
3261 &ctx,
3262 );
3263
3264 match rx.try_recv().unwrap() {
3265 DataEvent::FundingRate(update) => {
3266 assert_eq!(update.instrument_id, instrument_id);
3267 assert_eq!(update.rate, "0.0002".parse().unwrap());
3268 }
3269 other => panic!("expected FundingRateUpdate, was {other:?}"),
3270 }
3271 assert!(rx.try_recv().is_err());
3272 }
3273
3274 #[rstest]
3275 fn test_ticker_skips_funding_rate_when_not_perp() {
3276 let instrument = option_instrument();
3277 let instrument_id = instrument.id();
3278 let channel = format!("ticker_slim.{}.1000", instrument_id.symbol.as_str());
3279 let (ctx, mut rx) = make_ctx(Some(instrument));
3280 install_ticker(&ctx, instrument_id, &channel);
3281 ctx.active_funding_subs.insert(instrument_id);
3282
3283 let mut option_data = option_ticker_json(1_700_000_000_000);
3284 option_data["instrument_name"] = json!(instrument_id.symbol.as_str());
3285 let payload = subscription_payload(
3286 &channel,
3287 &json!({
3288 "timestamp": 1_700_000_000_010_i64,
3289 "instrument_ticker": option_data
3290 }),
3291 );
3292
3293 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3294
3295 assert!(rx.try_recv().is_err());
3296 }
3297
3298 #[rstest]
3299 fn test_ticker_emits_option_greeks_when_subscribed() {
3300 let instrument = option_instrument();
3301 let instrument_id = instrument.id();
3302 let channel = format!("ticker_slim.{}.1000", instrument_id.symbol.as_str());
3303 let (ctx, mut rx) = make_ctx(Some(instrument));
3304 install_ticker(&ctx, instrument_id, &channel);
3305 ctx.active_greeks_subs.insert(instrument_id);
3306
3307 let mut option_data = option_ticker_json(1_700_000_000_000);
3308 option_data["instrument_name"] = json!(instrument_id.symbol.as_str());
3309 let payload = subscription_payload(
3310 &channel,
3311 &json!({
3312 "timestamp": 1_700_000_000_010_i64,
3313 "instrument_ticker": option_data
3314 }),
3315 );
3316
3317 DeriveDataClient::handle_ws_message(DeriveWsMessage::Subscription(payload), &ctx);
3318
3319 match rx.try_recv().unwrap() {
3320 DataEvent::OptionGreeks(greeks) => {
3321 assert_eq!(greeks.instrument_id, instrument_id);
3322 assert!((greeks.greeks.delta - 0.55).abs() < 1e-9);
3323 assert!((greeks.greeks.gamma - 0.0008).abs() < 1e-9);
3324 assert!((greeks.greeks.vega - 4.5).abs() < 1e-9);
3325 assert!((greeks.greeks.theta + 2.1).abs() < 1e-9);
3326 assert!((greeks.greeks.rho - 1.2).abs() < 1e-9);
3327 assert_eq!(greeks.mark_iv, Some(0.60));
3328 assert_eq!(greeks.bid_iv, Some(0.58));
3329 assert_eq!(greeks.ask_iv, Some(0.62));
3330 assert_eq!(greeks.underlying_price, Some(3505.0));
3331 assert_eq!(greeks.open_interest, Some(1000.0));
3332 }
3333 other => panic!("expected OptionGreeks, was {other:?}"),
3334 }
3335 assert!(rx.try_recv().is_err());
3336 }
3337
3338 #[rstest]
3339 fn test_ticker_emits_all_subscribed_feeds_in_one_frame() {
3340 let instrument = perp_instrument();
3341 let instrument_id = instrument.id();
3342 let (ctx, mut rx) = make_ctx(Some(instrument));
3343 install_ticker(&ctx, instrument_id, "ticker_slim.ETH-PERP.1000");
3344 ctx.active_quote_subs.insert(instrument_id);
3345 ctx.active_mark_subs.insert(instrument_id);
3346 ctx.active_index_subs.insert(instrument_id);
3347 ctx.active_funding_subs.insert(instrument_id);
3348
3349 DeriveDataClient::handle_ws_message(
3350 DeriveWsMessage::Subscription(perp_ticker_payload(instrument_id)),
3351 &ctx,
3352 );
3353
3354 let mut quote = None;
3355 let mut mark = None;
3356 let mut index = None;
3357 let mut funding = None;
3358
3359 while let Ok(event) = rx.try_recv() {
3360 match event {
3361 DataEvent::Data(Data::Quote(q)) => {
3362 assert!(quote.replace(q).is_none(), "duplicate Quote emission");
3363 }
3364 DataEvent::Data(Data::MarkPrice(m)) => {
3365 assert!(
3366 mark.replace(m).is_none(),
3367 "duplicate MarkPriceUpdate emission"
3368 );
3369 }
3370 DataEvent::Data(Data::IndexPrice(i)) => {
3371 assert!(
3372 index.replace(i).is_none(),
3373 "duplicate IndexPriceUpdate emission"
3374 );
3375 }
3376 DataEvent::FundingRate(f) => {
3377 assert!(
3378 funding.replace(f).is_none(),
3379 "duplicate FundingRate emission"
3380 );
3381 }
3382 other => panic!("unexpected event: {other:?}"),
3383 }
3384 }
3385
3386 let quote = quote.expect("Quote event missing");
3387 let mark = mark.expect("MarkPriceUpdate missing");
3388 let index = index.expect("IndexPriceUpdate missing");
3389 let funding = funding.expect("FundingRateUpdate missing");
3390
3391 assert_eq!(quote.instrument_id, instrument_id);
3392 assert_eq!(quote.bid_price, Price::from("3500.00"));
3393 assert_eq!(quote.ask_price, Price::from("3501.00"));
3394 assert_eq!(mark.instrument_id, instrument_id);
3395 assert_eq!(mark.value, Price::from("3500.50"));
3396 assert_eq!(index.instrument_id, instrument_id);
3397 assert_eq!(index.value, Price::from("3500.00"));
3398 assert_eq!(funding.instrument_id, instrument_id);
3399 assert_eq!(funding.rate, "0.0002".parse().unwrap());
3400 }
3401
3402 #[rstest]
3403 fn test_reset_clears_all_subscription_state() {
3404 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3405 replace_data_event_sender(tx);
3406
3407 let config = DeriveDataClientConfig {
3408 environment: DeriveEnvironment::Mainnet,
3409 ..Default::default()
3410 };
3411 let mut client = DeriveDataClient::new(*DERIVE_CLIENT_ID, config).unwrap();
3412 let instrument = perp_instrument();
3413 let instrument_id = instrument.id();
3414 cache_instrument(&client.instruments, &instrument);
3415
3416 client
3417 .active_book_delta_channels
3418 .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
3419 client
3420 .active_book_depth10_channels
3421 .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
3422 client
3423 .active_ticker_channels
3424 .insert(instrument_id, "ticker_slim.ETH-PERP.1000".to_string());
3425 client.active_quote_subs.insert(instrument_id);
3426 client.active_trade_subs.insert(instrument_id);
3427 client
3428 .channel_subscriptions
3429 .state
3430 .lock()
3431 .activate(ChannelOwner::Trades(instrument_id), Some("trades.perp.ETH"));
3432 client.channel_subscriptions.transition("trades.perp.ETH");
3433 client.active_mark_subs.insert(instrument_id);
3434 client.active_index_subs.insert(instrument_id);
3435 client.active_funding_subs.insert(instrument_id);
3436 client.active_greeks_subs.insert(instrument_id);
3437
3438 client.reset().unwrap();
3439
3440 assert!(!client.instruments.contains_key(&instrument_id));
3441 assert!(
3442 !client
3443 .active_book_delta_channels
3444 .contains_key(&instrument_id)
3445 );
3446 assert!(
3447 !client
3448 .active_book_depth10_channels
3449 .contains_key(&instrument_id)
3450 );
3451 assert!(!client.active_ticker_channels.contains_key(&instrument_id));
3452 assert!(!client.active_quote_subs.contains(&instrument_id));
3453 assert!(!client.active_trade_subs.contains(&instrument_id));
3454 assert!(client.channel_subscriptions.state.lock().owners.is_empty());
3455
3456 assert_eq!(client.channel_subscriptions.transitions.len(), 1);
3459 assert!(!client.active_mark_subs.contains(&instrument_id));
3460 assert!(!client.active_index_subs.contains(&instrument_id));
3461 assert!(!client.active_funding_subs.contains(&instrument_id));
3462 assert!(!client.active_greeks_subs.contains(&instrument_id));
3463 assert!(!client.is_connected());
3464 }
3465
3466 #[tokio::test]
3467 async fn test_disconnect_clears_subscription_state() {
3468 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3474 replace_data_event_sender(tx);
3475
3476 let config = DeriveDataClientConfig {
3477 environment: DeriveEnvironment::Mainnet,
3478 ..Default::default()
3479 };
3480 let mut client = DeriveDataClient::new(*DERIVE_CLIENT_ID, config).unwrap();
3481 let instrument = perp_instrument();
3482 let instrument_id = instrument.id();
3483 cache_instrument(&client.instruments, &instrument);
3484
3485 client
3486 .active_book_delta_channels
3487 .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
3488 client
3489 .active_book_depth10_channels
3490 .insert(instrument_id, "orderbook.ETH-PERP.1.10".to_string());
3491 client
3492 .active_ticker_channels
3493 .insert(instrument_id, "ticker_slim.ETH-PERP.1000".to_string());
3494 client.active_quote_subs.insert(instrument_id);
3495 client.active_trade_subs.insert(instrument_id);
3496 client
3497 .channel_subscriptions
3498 .state
3499 .lock()
3500 .activate(ChannelOwner::Trades(instrument_id), Some("trades.perp.ETH"));
3501 client.channel_subscriptions.transition("trades.perp.ETH");
3502 client.active_mark_subs.insert(instrument_id);
3503 client.active_index_subs.insert(instrument_id);
3504 client.active_funding_subs.insert(instrument_id);
3505 client.active_greeks_subs.insert(instrument_id);
3506 client.is_connected.store(true, Ordering::Relaxed);
3507 let (started_tx, started_rx) = tokio::sync::oneshot::channel();
3508 let (drop_tx, mut drop_rx) = tokio::sync::oneshot::channel::<()>();
3509 client
3510 .pending_tasks
3511 .spawn(async move {
3512 let _drop_tx = drop_tx;
3513 let _ = started_tx.send(());
3514 std::future::pending::<()>().await;
3515 })
3516 .unwrap();
3517 started_rx.await.unwrap();
3518
3519 client.disconnect().await.unwrap();
3520
3521 assert!(client.instruments.contains_key(&instrument_id));
3524 assert!(
3525 !client
3526 .active_book_delta_channels
3527 .contains_key(&instrument_id)
3528 );
3529 assert!(
3530 !client
3531 .active_book_depth10_channels
3532 .contains_key(&instrument_id)
3533 );
3534 assert!(!client.active_ticker_channels.contains_key(&instrument_id));
3535 assert!(!client.active_quote_subs.contains(&instrument_id));
3536 assert!(!client.active_trade_subs.contains(&instrument_id));
3537 assert!(client.channel_subscriptions.state.lock().owners.is_empty());
3538 assert!(client.channel_subscriptions.transitions.is_empty());
3539 assert!(!client.active_mark_subs.contains(&instrument_id));
3540 assert!(!client.active_index_subs.contains(&instrument_id));
3541 assert!(!client.active_funding_subs.contains(&instrument_id));
3542 assert!(!client.active_greeks_subs.contains(&instrument_id));
3543 assert!(!client.is_connected());
3544 assert_eq!(
3545 drop_rx.try_recv(),
3546 Err(tokio::sync::oneshot::error::TryRecvError::Closed),
3547 );
3548 }
3549
3550 #[tokio::test]
3551 async fn test_task_group_unregisters_finished_tasks_before_drain() {
3552 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
3553 replace_data_event_sender(tx);
3554
3555 let config = DeriveDataClientConfig {
3556 environment: DeriveEnvironment::Mainnet,
3557 ..Default::default()
3558 };
3559 let client = DeriveDataClient::new(*DERIVE_CLIENT_ID, config).unwrap();
3560
3561 for _ in 0..100 {
3562 client.spawn_task("test_noop", async { Ok(()) });
3563 }
3564
3565 wait_until_async(
3566 || async { client.pending_tasks.all_finished() },
3567 Duration::from_secs(2),
3568 )
3569 .await;
3570
3571 assert!(client.pending_tasks.is_empty());
3572 client.pending_tasks.begin_shutdown();
3573 client
3574 .pending_tasks
3575 .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
3576 .await
3577 .unwrap();
3578 assert!(client.pending_tasks.is_empty());
3579 }
3580}