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