nautilus_coinapi/book/
sync.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Adapter-local order book synchronization state for CoinAPI.
17//!
18//! [`BookSyncTracker`] applies the CoinAPI sequence rule to the shared per-book [`BookSync`]
19//! lifecycle and converts accepted messages into deltas. A snapshot lists the whole book, so it is
20//! accepted at any sequence; an update is accepted only at the sequence after the last accepted
21//! message. A book's position is that last sequence.
22//!
23//! CoinAPI restarts a pair's sequence at one whenever it restarts the pair or the connection, and
24//! a managed `OrderBook` keeps its sequence as a high-water mark across snapshots. Emitted deltas
25//! therefore carry the venue sequence plus a per-book offset, set at each accepted snapshot so the
26//! emitted sequence keeps rising across restarts, unsubscribes, and reconnects. Gap detection uses
27//! the venue sequence.
28//!
29//! Conversion happens under the tracker's lock, so an L3 book's `OrderIndex` changes only with
30//! its accepted position. A message that fails conversion leaves the book out of sync, as a gap
31//! does. This module performs state transitions; `super::recovery` runs the asynchronous
32//! scope changes and retries.
33
34use std::sync::{
35    Arc,
36    atomic::{AtomicBool, Ordering},
37};
38
39use ahash::AHashMap;
40use nautilus_common::live::dst::time::Instant;
41use nautilus_core::AtomicMap;
42use nautilus_live::book::{
43    BookSequenceOutcome, BookSyncSignal, BookSyncSignalKind,
44    snapshot::SnapshotGate,
45    sync::{BookPhase, BookSync},
46};
47use nautilus_model::{
48    data::OrderBookDeltas, identifiers::InstrumentId, instruments::InstrumentAny,
49};
50use parking_lot::Mutex;
51use tokio_util::sync::CancellationToken;
52
53use super::{CoinApiBookError, orders::OrderIndex, recovery::BookRecovery};
54use crate::websocket::{
55    messages::{Book, BookL3, DataType, Subscription},
56    parse::parse_book_deltas,
57};
58
59/// A `book` or `book_l3` message routed to its book.
60#[derive(Debug, Clone, Copy)]
61pub enum BookFrame<'a> {
62    Levels(&'a Book),
63    Orders(&'a BookL3),
64}
65
66impl BookFrame<'_> {
67    fn sequence(self) -> u64 {
68        match self {
69            Self::Levels(book) => book.sequence,
70            Self::Orders(book) => book.sequence,
71        }
72    }
73
74    fn is_snapshot(self) -> bool {
75        match self {
76            Self::Levels(book) => book.is_snapshot == Some(true),
77            Self::Orders(book) => book.is_snapshot,
78        }
79    }
80}
81
82#[derive(Debug, Clone)]
83pub(crate) struct BookSubscription {
84    pub(crate) data_type: DataType,
85    pub(crate) depth: Option<usize>,
86    pub(crate) active: Arc<AtomicBool>,
87}
88
89// Registration and retirement share one writer lock; market-data reads use immutable snapshots
90#[derive(Debug, Default)]
91pub(crate) struct BookSubscriptions {
92    pub(crate) entries: AtomicMap<InstrumentId, BookSubscription>,
93    pub(crate) writes: Mutex<()>,
94    pub(crate) sync: BookSyncTracker,
95}
96
97impl BookSubscriptions {
98    pub(crate) fn clear(&self) {
99        let _write = self.writes.lock();
100        self.sync.clear();
101        self.entries.store(AHashMap::new());
102    }
103
104    pub(crate) fn retire(&self) {
105        let _write = self.writes.lock();
106        let mut entries = (**self.entries.load()).clone();
107        entries.retain(|instrument_id, registration| {
108            if registration.active.load(Ordering::Relaxed) {
109                self.sync.remove(*instrument_id);
110                false
111            } else {
112                true
113            }
114        });
115
116        self.entries.store(entries);
117    }
118}
119
120#[derive(Debug)]
121struct BookState {
122    subscription: Subscription,
123    sync: BookSync<CoinApiBookError, u64>,
124    orders: OrderIndex,
125    // Added to the venue sequence of each delta emitted in the current snapshot episode
126    offset: u64,
127}
128
129/// Sequence, snapshot, and recovery state of each subscribed book.
130///
131/// A book accepts messages only after [`Self::record_subscription`] registers it; [`Self::apply`]
132/// suppresses messages for any other book.
133#[derive(Debug, Clone, Default)]
134pub struct BookSyncTracker {
135    books: Arc<Mutex<AHashMap<InstrumentId, BookState>>>,
136    // Last emitted sequence of each book, kept across subscriptions and connections
137    emitted: Arc<Mutex<AHashMap<InstrumentId, u64>>>,
138    // Last venue and emitted sequence of each depth stream
139    depths: Arc<Mutex<AHashMap<InstrumentId, (u64, u64)>>>,
140}
141
142impl BookSyncTracker {
143    /// Replaces the book with one waiting for its first snapshot behind `gate`.
144    pub fn record_subscription(
145        &self,
146        instrument_id: InstrumentId,
147        subscription: Subscription,
148        now: Instant,
149        gate: SnapshotGate,
150    ) -> CancellationToken {
151        let mut sync = BookSync::new(now);
152        let cancel = sync.expect_snapshot(gate);
153
154        let state = BookState {
155            subscription,
156            sync,
157            orders: OrderIndex::default(),
158            offset: 0,
159        };
160
161        self.books.lock().insert(instrument_id, state);
162        cancel
163    }
164
165    pub(crate) fn remove(&self, instrument_id: InstrumentId) {
166        self.books.lock().remove(&instrument_id);
167    }
168
169    pub(crate) fn clear(&self) {
170        self.books.lock().clear();
171    }
172
173    /// Validates a message and converts it once accepted.
174    ///
175    /// Returns [`BookSequenceOutcome::Recover`] when the caller must start a recovery, and deltas
176    /// only for an accepted message that changes the book.
177    pub fn apply(
178        &self,
179        instrument_id: InstrumentId,
180        frame: BookFrame<'_>,
181        instrument: &InstrumentAny,
182        now: Instant,
183    ) -> (BookSequenceOutcome, Option<OrderBookDeltas>) {
184        let mut books = self.books.lock();
185
186        let Some(state) = books.get_mut(&instrument_id) else {
187            return (BookSequenceOutcome::Suppress, None);
188        };
189
190        // Frames before the subscribing write belong to a scope that did not include the book
191        if state.sync.is_send_pending() {
192            return (BookSequenceOutcome::Suppress, None);
193        }
194
195        let sequence = frame.sequence();
196        if frame.is_snapshot() {
197            let converted = match frame {
198                BookFrame::Levels(book) => parse_book_deltas(book, instrument).map(|d| (None, d)),
199                BookFrame::Orders(book) => OrderIndex::snapshot(book, instrument)
200                    .map(|(orders, deltas)| (Some(orders), Some(deltas))),
201            };
202
203            let (orders, deltas) = match converted {
204                Ok(converted) => converted,
205                Err(e) => return (invalid(state, instrument_id, sequence, &e), None),
206            };
207
208            if !state.sync.accept_snapshot(sequence, now) {
209                return (BookSequenceOutcome::Suppress, None);
210            }
211
212            if let Some(orders) = orders {
213                state.orders = orders;
214            }
215
216            // The snapshot's emitted sequence follows the book's last one, even after a restart
217            let mut emitted = self.emitted.lock();
218            let last = emitted.get(&instrument_id).copied().unwrap_or(0);
219            state.offset = (last + 1).saturating_sub(sequence);
220            emitted.insert(instrument_id, state.offset + sequence);
221            return (
222                BookSequenceOutcome::Accept,
223                deltas.map(|deltas| offset(deltas, state.offset)),
224            );
225        }
226
227        // An out-of-sync book suppresses updates while an owner restores it, and asks for one
228        // otherwise
229        match *state.sync.phase() {
230            BookPhase::Synced(position) if position.checked_add(1) == Some(sequence) => {}
231            BookPhase::Synced(_) | BookPhase::Waiting | BookPhase::Recovering => {
232                return (gap(state, instrument_id, sequence), None);
233            }
234        }
235
236        let converted = match frame {
237            BookFrame::Levels(book) => parse_book_deltas(book, instrument),
238            BookFrame::Orders(book) => state.orders.update(book, instrument),
239        };
240
241        match converted {
242            Ok(deltas) => {
243                state.sync.advance(sequence, now);
244                self.emitted
245                    .lock()
246                    .insert(instrument_id, state.offset + sequence);
247                (
248                    BookSequenceOutcome::Accept,
249                    deltas.map(|deltas| offset(deltas, state.offset)),
250                )
251            }
252            Err(e) => (invalid(state, instrument_id, sequence, &e), None),
253        }
254    }
255
256    /// Returns the emitted sequence of a depth snapshot at venue `sequence`.
257    ///
258    /// Each depth message lists complete top levels, so the stream advances by the venue step and
259    /// continues from its last emitted sequence when CoinAPI restarts the venue sequence.
260    pub fn depth_sequence(&self, instrument_id: InstrumentId, sequence: u64) -> u64 {
261        let mut depths = self.depths.lock();
262
263        let emitted = match depths.get(&instrument_id) {
264            Some(&(venue, emitted)) if sequence > venue => emitted + (sequence - venue),
265            Some(&(_, emitted)) => emitted + 1,
266            None => sequence,
267        };
268
269        depths.insert(instrument_id, (sequence, emitted));
270        emitted
271    }
272
273    /// Restarts every book after its connection reconnects, arming a snapshot `deadline` when
274    /// given, and returns how many deadlines were armed.
275    pub fn reset_on_reconnect(&self, deadline: Option<Instant>) -> usize {
276        let mut armed = 0;
277
278        for state in self.books.lock().values_mut() {
279            state.sync.reset_on_reconnect();
280
281            if deadline.is_some_and(|deadline| state.sync.arm_deadline(deadline)) {
282                armed += 1;
283            }
284        }
285
286        armed
287    }
288
289    pub(crate) fn take_expired_snapshots(&self, now: Instant) -> Vec<BookSyncSignal> {
290        let mut expired = self
291            .books
292            .lock()
293            .iter_mut()
294            .filter_map(|(instrument_id, state)| {
295                state.sync.take_expired(now).then_some(instrument_id)
296            })
297            .map(|instrument_id| BookSyncSignal {
298                instrument_id: *instrument_id,
299                kind: BookSyncSignalKind::SnapshotMissing,
300            })
301            .collect::<Vec<_>>();
302
303        // Sort by instrument; the book map iterates in per-process hash order
304        expired.sort_by_key(|signal| signal.instrument_id);
305        expired
306    }
307
308    /// Claims a recovery for a subscription whose initial snapshot did not arrive, unless the
309    /// subscription was replaced or removed.
310    pub(crate) fn claim_subscription_recovery(
311        &self,
312        instrument_id: InstrumentId,
313        cancel: &CancellationToken,
314    ) -> Option<(Arc<BookRecovery>, Subscription)> {
315        let mut books = self.books.lock();
316
317        if cancel.is_cancelled() {
318            return None;
319        }
320
321        let state = books.get_mut(&instrument_id)?;
322        let recovery = state.sync.claim()?;
323        Some((recovery, state.subscription))
324    }
325
326    pub(crate) fn claim_recovery(
327        &self,
328        instrument_id: InstrumentId,
329    ) -> Option<(Arc<BookRecovery>, Subscription)> {
330        let mut books = self.books.lock();
331        let state = books
332            .get_mut(&instrument_id)
333            .filter(|state| !state.sync.is_send_pending())?;
334        let recovery = state.sync.claim()?;
335        Some((recovery, state.subscription))
336    }
337}
338
339fn offset(mut deltas: OrderBookDeltas, offset: u64) -> OrderBookDeltas {
340    for delta in &mut deltas.deltas {
341        delta.sequence += offset;
342    }
343
344    deltas.sequence += offset;
345    deltas
346}
347
348fn gap(state: &mut BookState, instrument_id: InstrumentId, sequence: u64) -> BookSequenceOutcome {
349    let position = state.sync.position().copied();
350    let outcome = state.sync.gap();
351
352    if outcome == BookSequenceOutcome::Recover {
353        log::warn!(
354            "Book sequence gap for {instrument_id}: last sequence {position:?}, received \
355             {sequence}; requesting a fresh snapshot"
356        );
357    }
358
359    outcome
360}
361
362fn invalid(
363    state: &mut BookState,
364    instrument_id: InstrumentId,
365    sequence: u64,
366    error: &anyhow::Error,
367) -> BookSequenceOutcome {
368    log::warn!("Invalid book message for {instrument_id} at sequence {sequence}: {error:#}");
369    state.sync.gap()
370}
371
372#[cfg(test)]
373mod tests {
374    use nautilus_common::live::dst::time::Duration;
375    use nautilus_core::UnixNanos;
376    use nautilus_model::{
377        enums::BookAction,
378        identifiers::InstrumentId,
379        instruments::CurrencyPair,
380        types::{Currency, Price, Quantity},
381    };
382    use rstest::rstest;
383    use ustr::Ustr;
384
385    use super::*;
386    use crate::websocket::messages::{Channel, DataType};
387
388    fn instrument_id() -> InstrumentId {
389        InstrumentId::from("BTC-USD.COINBASE")
390    }
391
392    fn instrument() -> InstrumentAny {
393        InstrumentAny::CurrencyPair(
394            CurrencyPair::builder()
395                .instrument_id(instrument_id())
396                .raw_symbol("BTC-USD".into())
397                .base_currency(Currency::BTC())
398                .quote_currency(Currency::USD())
399                .price_precision(2)
400                .size_precision(8)
401                .price_increment(Price::from("0.01"))
402                .size_increment(Quantity::from("0.00000001"))
403                .ts_event(UnixNanos::default())
404                .ts_init(UnixNanos::default())
405                .build()
406                .unwrap(),
407        )
408    }
409
410    fn subscription(data_type: DataType) -> Subscription {
411        Subscription {
412            channel: Channel {
413                data_type,
414                period_id: None,
415            },
416            symbol_id: Ustr::from("COINBASE_SPOT_BTC_USD"),
417        }
418    }
419
420    fn levels(sequence: u64, is_snapshot: bool) -> Book {
421        let mut book: Book =
422            serde_json::from_str(include_str!("../../test_data/ws_book.json")).unwrap();
423        book.sequence = sequence;
424        book.is_snapshot = Some(is_snapshot);
425        book
426    }
427
428    fn orders(sequence: u64, is_snapshot: bool) -> BookL3 {
429        let mut book: BookL3 =
430            serde_json::from_str(include_str!("../../test_data/ws_book_l3.json")).unwrap();
431        book.sequence = sequence;
432        book.is_snapshot = is_snapshot;
433        book
434    }
435
436    // A tracker whose book's subscribing write has completed
437    fn subscribed(data_type: DataType, now: Instant) -> BookSyncTracker {
438        let tracker = BookSyncTracker::default();
439        tracker.record_subscription(
440            instrument_id(),
441            subscription(data_type),
442            now,
443            SnapshotGate::default(),
444        );
445        tracker
446    }
447
448    fn apply(
449        tracker: &BookSyncTracker,
450        book: &Book,
451        now: Instant,
452    ) -> (BookSequenceOutcome, Option<Vec<BookAction>>) {
453        let (outcome, deltas) =
454            tracker.apply(instrument_id(), BookFrame::Levels(book), &instrument(), now);
455        let actions = deltas.map(|deltas| deltas.deltas.iter().map(|d| d.action).collect());
456        (outcome, actions)
457    }
458
459    #[rstest]
460    #[case(true)]
461    #[case(false)]
462    fn retirement_removes_active_and_preserves_pending(#[case] active: bool) {
463        let subscriptions = BookSubscriptions::default();
464        subscriptions.entries.insert(
465            instrument_id(),
466            BookSubscription {
467                data_type: DataType::Book,
468                depth: None,
469                active: Arc::new(AtomicBool::new(active)),
470            },
471        );
472
473        let cancel = subscriptions.sync.record_subscription(
474            instrument_id(),
475            subscription(DataType::Book),
476            Instant::now(),
477            SnapshotGate::default(),
478        );
479        subscriptions.retire();
480
481        assert_eq!(
482            subscriptions.entries.contains_key(&instrument_id()),
483            !active
484        );
485        assert_eq!(cancel.is_cancelled(), active);
486    }
487
488    #[rstest]
489    fn retirement_keeps_replacement_registration_when_old_command_commits() {
490        let subscriptions = BookSubscriptions::default();
491        let old = Arc::new(AtomicBool::new(false));
492        let replacement = Arc::new(AtomicBool::new(false));
493        subscriptions.entries.insert(
494            instrument_id(),
495            BookSubscription {
496                data_type: DataType::Book20,
497                depth: Some(10),
498                active: Arc::clone(&replacement),
499            },
500        );
501
502        old.store(true, Ordering::Relaxed);
503        subscriptions.retire();
504
505        let registration = subscriptions.entries.get_cloned(&instrument_id()).unwrap();
506        assert_eq!(registration.data_type, DataType::Book20);
507        assert_eq!(registration.depth, Some(10));
508        assert!(!registration.active.load(Ordering::Relaxed));
509        assert!(Arc::ptr_eq(&registration.active, &replacement));
510    }
511
512    #[rstest]
513    fn snapshot_then_linked_updates_are_accepted() {
514        let now = Instant::now();
515        let tracker = subscribed(DataType::Book, now);
516
517        let snapshot = apply(&tracker, &levels(1, true), now);
518        let update = apply(&tracker, &levels(2, false), now);
519
520        assert_eq!(
521            snapshot,
522            (
523                BookSequenceOutcome::Accept,
524                Some(vec![
525                    BookAction::Clear,
526                    BookAction::Add,
527                    BookAction::Add,
528                    BookAction::Add,
529                    BookAction::Add,
530                ])
531            )
532        );
533        assert_eq!(
534            update,
535            (
536                BookSequenceOutcome::Accept,
537                Some(vec![BookAction::Update; 4])
538            )
539        );
540    }
541
542    // A snapshot lists the whole book, so it resynchronizes at any sequence, including after a skip
543    #[rstest]
544    fn gap_recovers_once_claimed_and_snapshot_completes_recovery() {
545        let now = Instant::now();
546        let tracker = subscribed(DataType::Book, now);
547        apply(&tracker, &levels(1, true), now);
548
549        let gap = apply(&tracker, &levels(3, false), now);
550        let unowned = apply(&tracker, &levels(4, false), now);
551        let (recovery, claimed) = tracker.claim_recovery(instrument_id()).unwrap();
552        assert!(recovery.begin_replacement());
553        let owned = apply(&tracker, &levels(5, false), now);
554        let during_write = apply(&tracker, &levels(1, true), now);
555        recovery.gate.open();
556        let restarted = apply(&tracker, &levels(1, true), now);
557        let linked = apply(&tracker, &levels(2, false), now);
558
559        assert_eq!(gap, (BookSequenceOutcome::Recover, None));
560        assert_eq!(unowned, (BookSequenceOutcome::Recover, None));
561        assert_eq!(claimed, subscription(DataType::Book));
562        assert_eq!(owned, (BookSequenceOutcome::Suppress, None));
563        assert_eq!(during_write, (BookSequenceOutcome::Suppress, None));
564        assert_eq!(restarted.0, BookSequenceOutcome::Accept);
565        assert!(recovery.is_accepted());
566        assert_eq!(linked.0, BookSequenceOutcome::Accept);
567    }
568
569    // CoinAPI restarts sequences at one, so emitted sequences continue from the last emitted one,
570    // including after the book is subscribed again
571    #[rstest]
572    fn emitted_sequences_keep_rising_across_restarts_and_resubscribes() {
573        let now = Instant::now();
574        let tracker = subscribed(DataType::Book, now);
575        let emitted = |frame: &Book| {
576            tracker
577                .apply(
578                    instrument_id(),
579                    BookFrame::Levels(frame),
580                    &instrument(),
581                    now,
582                )
583                .1
584                .map(|deltas| {
585                    let sequences: Vec<u64> = deltas.deltas.iter().map(|d| d.sequence).collect();
586                    assert!(
587                        sequences
588                            .iter()
589                            .all(|sequence| *sequence == deltas.sequence)
590                    );
591                    deltas.sequence
592                })
593        };
594
595        let first = emitted(&levels(1, true));
596        let second = emitted(&levels(2, false));
597        let doubled = emitted(&levels(2, true));
598        let restarted = emitted(&levels(1, true));
599        let linked = emitted(&levels(2, false));
600        tracker.record_subscription(
601            instrument_id(),
602            subscription(DataType::Book),
603            now,
604            SnapshotGate::default(),
605        );
606        let resubscribed = emitted(&levels(1, true));
607
608        assert_eq!(
609            [first, second, doubled, restarted, linked, resubscribed],
610            [Some(1), Some(2), Some(3), Some(4), Some(5), Some(6)]
611        );
612    }
613
614    #[rstest]
615    fn depth_sequences_keep_rising_when_the_venue_restarts() {
616        let tracker = BookSyncTracker::default();
617        let emitted: Vec<u64> = [7, 9, 1, 2, 5]
618            .into_iter()
619            .map(|sequence| tracker.depth_sequence(instrument_id(), sequence))
620            .collect();
621
622        assert_eq!(emitted, [7, 9, 10, 11, 14]);
623    }
624
625    #[rstest]
626    fn update_before_snapshot_requests_recovery() {
627        let now = Instant::now();
628        let tracker = subscribed(DataType::Book, now);
629
630        let result = apply(&tracker, &levels(7, false), now);
631
632        assert_eq!(result, (BookSequenceOutcome::Recover, None));
633    }
634
635    #[rstest]
636    fn frames_wait_for_subscribing_write() {
637        let now = Instant::now();
638        let tracker = BookSyncTracker::default();
639        let gate = SnapshotGate::default();
640        gate.lock().close();
641        let cancel = tracker.record_subscription(
642            instrument_id(),
643            subscription(DataType::Book),
644            now,
645            gate.clone(),
646        );
647
648        let update = apply(&tracker, &levels(7, false), now);
649        let snapshot = apply(&tracker, &levels(1, true), now);
650        let refused = tracker.claim_recovery(instrument_id());
651        gate.open();
652        let accepted = apply(&tracker, &levels(1, true), now);
653
654        assert_eq!(update, (BookSequenceOutcome::Suppress, None));
655        assert_eq!(snapshot, (BookSequenceOutcome::Suppress, None));
656        assert!(refused.is_none());
657        assert_eq!(accepted.0, BookSequenceOutcome::Accept);
658        assert!(cancel.is_cancelled());
659    }
660
661    #[rstest]
662    fn invalid_messages_request_recovery_without_output() {
663        let now = Instant::now();
664        let tracker = subscribed(DataType::BookL3, now);
665        let mut aggregated = orders(1, true);
666        aggregated.bids[1].update_type = None;
667        aggregated
668            .asks
669            .iter_mut()
670            .for_each(|entry| entry.update_type = None);
671
672        let (snapshot, snapshot_deltas) = tracker.apply(
673            instrument_id(),
674            BookFrame::Orders(&aggregated),
675            &instrument(),
676            now,
677        );
678
679        assert_eq!(snapshot, BookSequenceOutcome::Recover);
680        assert_eq!(snapshot_deltas, None);
681    }
682
683    // The index follows the accepted snapshot, so a later ADD for a listed order is inconsistent
684    #[rstest]
685    fn l3_updates_apply_to_orders_from_snapshot() {
686        let now = Instant::now();
687        let tracker = subscribed(DataType::BookL3, now);
688        let mut snapshot = orders(1, true);
689        snapshot.bids.truncate(1);
690        let mut delete = orders(2, false);
691        delete.asks.remove(0);
692        delete.bids.clear();
693        let mut add_listed = orders(3, false);
694        add_listed.asks.truncate(1);
695        add_listed.bids.clear();
696
697        let apply = |book: &BookL3| {
698            let (outcome, deltas) =
699                tracker.apply(instrument_id(), BookFrame::Orders(book), &instrument(), now);
700            let actions: Option<Vec<BookAction>> =
701                deltas.map(|deltas| deltas.deltas.iter().map(|d| d.action).collect());
702            (outcome, actions)
703        };
704
705        let accepted = apply(&snapshot);
706        let deleted = apply(&delete);
707        let inconsistent = apply(&add_listed);
708
709        assert_eq!(
710            accepted,
711            (
712                BookSequenceOutcome::Accept,
713                Some(vec![
714                    BookAction::Clear,
715                    BookAction::Add,
716                    BookAction::Add,
717                    BookAction::Add
718                ])
719            )
720        );
721        assert_eq!(
722            deleted,
723            (BookSequenceOutcome::Accept, Some(vec![BookAction::Delete]))
724        );
725        assert_eq!(inconsistent, (BookSequenceOutcome::Recover, None));
726    }
727
728    #[rstest]
729    fn reconnect_waits_for_replayed_snapshot_until_deadline() {
730        let now = Instant::now();
731        let tracker = subscribed(DataType::Book, now);
732        apply(&tracker, &levels(1, true), now);
733        apply(&tracker, &levels(2, false), now);
734
735        let armed = tracker.reset_on_reconnect(Some(now + Duration::from_secs(10)));
736        let stale = apply(&tracker, &levels(3, false), now);
737        let early = tracker.take_expired_snapshots(now + Duration::from_secs(9));
738        let expired = tracker.take_expired_snapshots(now + Duration::from_secs(10));
739        let unowned = apply(&tracker, &levels(3, false), now);
740
741        assert_eq!(armed, 1);
742        assert_eq!(stale, (BookSequenceOutcome::Suppress, None));
743        assert!(early.is_empty());
744        assert_eq!(
745            expired,
746            vec![BookSyncSignal {
747                instrument_id: instrument_id(),
748                kind: BookSyncSignalKind::SnapshotMissing,
749            }]
750        );
751        assert_eq!(unowned, (BookSequenceOutcome::Recover, None));
752    }
753
754    #[rstest]
755    fn removed_book_suppresses_frames_and_cancels_recovery() {
756        let now = Instant::now();
757        let tracker = subscribed(DataType::Book, now);
758        let (recovery, _) = tracker.claim_recovery(instrument_id()).unwrap();
759
760        tracker.remove(instrument_id());
761        let result = apply(&tracker, &levels(1, true), now);
762
763        assert!(recovery.cancellation.is_cancelled());
764        assert_eq!(result, (BookSequenceOutcome::Suppress, None));
765    }
766}
nautilus_coinapi/book/sync.rs

nautilus_coinapi/book/
sync.rs
