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