nautilus_live/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//! Synchronization lifecycle of one subscribed book.
17//!
18//! [`BookSync`] holds the state an adapter tracks per book: its phase, the time of its last
19//! accepted frame, a pending snapshot wait, and its recovery episode. It performs no I/O and takes
20//! `&mut self`, so an adapter keeps it under its existing state lock or owning task.
21//!
22//! # Sync Contract
23//!
24//! - Output happens only in [`BookPhase::Synced`], which starts with an accepted snapshot.
25//! - A book out of sync needs an owner: a running recovery or an armed snapshot deadline.
26//!   [`BookSync::gap`] returns [`BookSequenceOutcome::Recover`] when it has neither, and the adapter
27//!   then claims a recovery once it has what the replacement needs.
28//! - The adapter passes every incremental update for an unsynced book through [`BookSync::gap`],
29//!   so a book whose recovery could not start asks again on its next update.
30//! - [`BookSync::stale`] reports a book that stops advancing without a running recovery.
31//!
32//! # Adapters
33//!
34//! The adapter keeps its venue sequence rule, wire commands, snapshot parsing, and buffers, and
35//! stores the venue position of the last accepted frame as `P`.
36
37use std::sync::Arc;
38
39use nautilus_common::live::dst::time::{Duration, Instant};
40use tokio_util::sync::CancellationToken;
41
42use super::{
43    BookSequenceOutcome,
44    recovery::{BookRecovery, BookRecoveryOutcome, BookRecoveryState},
45    snapshot::{PendingSnapshot, SnapshotGate},
46};
47
48/// Synchronization phase of one book.
49#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum BookPhase<P> {
51    /// Subscribed and waiting for the first snapshot.
52    Waiting,
53    /// A snapshot was accepted; holds the venue position of the last accepted frame.
54    Synced(P),
55    /// Output is suppressed until a snapshot restores the book.
56    Recovering,
57}
58
59/// Synchronization state of one subscribed book.
60#[derive(Debug)]
61pub struct BookSync<E, P = ()> {
62    phase: BookPhase<P>,
63    last_update: Instant,
64    pending: Option<PendingSnapshot>,
65    recovery: BookRecoveryState<E>,
66}
67
68impl<E: Clone, P> BookSync<E, P> {
69    /// Creates a book waiting for its first snapshot.
70    #[must_use]
71    pub fn new(now: Instant) -> Self {
72        Self {
73            phase: BookPhase::Waiting,
74            last_update: now,
75            pending: None,
76            recovery: BookRecoveryState::default(),
77        }
78    }
79
80    /// Returns the synchronization phase.
81    #[must_use]
82    pub fn phase(&self) -> &BookPhase<P> {
83        &self.phase
84    }
85
86    /// Returns the venue position of the last accepted frame while synced.
87    #[must_use]
88    pub fn position(&self) -> Option<&P> {
89        match &self.phase {
90            BookPhase::Synced(position) => Some(position),
91            BookPhase::Waiting | BookPhase::Recovering => None,
92        }
93    }
94
95    /// Returns whether `recovery` is the book's current episode.
96    #[must_use]
97    pub fn is_current(&self, recovery: &Arc<BookRecovery<E>>) -> bool {
98        self.recovery
99            .current()
100            .is_some_and(|current| Arc::ptr_eq(current, recovery))
101    }
102
103    /// Returns whether a snapshot wait or deadline is pending.
104    #[must_use]
105    pub fn has_pending_snapshot(&self) -> bool {
106        self.pending.is_some()
107    }
108
109    /// Returns whether a guarded subscription write is still in flight.
110    #[must_use]
111    pub fn is_send_pending(&self) -> bool {
112        self.pending
113            .as_ref()
114            .is_some_and(|pending| pending.gate.lock().is_closed())
115    }
116
117    /// Records the initial snapshot wait for a subscription whose write `gate` guards.
118    ///
119    /// Accepting a snapshot, claiming a recovery, or removing the book cancels the returned token.
120    pub fn expect_snapshot(&mut self, gate: SnapshotGate) -> CancellationToken {
121        let cancel = CancellationToken::new();
122        self.pending = Some(PendingSnapshot {
123            deadline: None,
124            cancel: cancel.clone(),
125            gate,
126        });
127
128        cancel
129    }
130
131    /// Accepts a snapshot at `position` unless a subscription write or recovery gate blocks it.
132    pub fn accept_snapshot(&mut self, position: P, now: Instant) -> bool {
133        if self.is_send_pending() {
134            return false;
135        }
136
137        if let Some(recovery) = self.recovery.current()
138            && recovery.is_running()
139            && !recovery.accept()
140        {
141            return false;
142        }
143
144        self.phase = BookPhase::Synced(position);
145        self.pending = None;
146        self.last_update = now;
147        true
148    }
149
150    /// Advances a synced book to `position`, returning `false` when the book is not synced.
151    pub fn advance(&mut self, position: P, now: Instant) -> bool {
152        if !matches!(self.phase, BookPhase::Synced(_)) {
153            return false;
154        }
155
156        self.phase = BookPhase::Synced(position);
157        self.last_update = now;
158        true
159    }
160
161    /// Marks the book out of sync after a gap, an incremental while unsynced, or an invalid
162    /// snapshot.
163    ///
164    /// Returns [`BookSequenceOutcome::Recover`] when neither a running recovery nor an armed
165    /// snapshot deadline owns the book; the adapter then claims a recovery. The result grants no
166    /// ownership, so later calls return `Recover` again until the adapter claims one.
167    pub fn gap(&mut self) -> BookSequenceOutcome {
168        self.phase = BookPhase::Recovering;
169
170        let deadline_armed = self
171            .pending
172            .as_ref()
173            .is_some_and(|pending| pending.deadline.is_some());
174
175        if self.recovery.is_running() || deadline_armed {
176            BookSequenceOutcome::Suppress
177        } else {
178            BookSequenceOutcome::Recover
179        }
180    }
181
182    /// Claims a recovery episode, refusing while another is running.
183    ///
184    /// The claimed episode owns the book and retires any pending snapshot wait. Cancel it if its
185    /// task cannot start, or it keeps the book owned with nothing running.
186    pub fn claim(&mut self) -> Option<Arc<BookRecovery<E>>> {
187        let recovery = self.recovery.claim()?;
188        self.phase = BookPhase::Recovering;
189        self.pending = None;
190        Some(recovery)
191    }
192
193    /// Delivers `error` to the running recovery, returning `false` when none is running.
194    pub fn reject(&self, error: E) -> bool {
195        match self.recovery.current() {
196            Some(recovery) if recovery.is_running() => {
197                recovery
198                    .outcome
199                    .send_replace(BookRecoveryOutcome::Rejected(error));
200                true
201            }
202            _ => false,
203        }
204    }
205
206    /// Restarts synchronization after the book's connection reconnects.
207    ///
208    /// An initial subscription write still in flight keeps its snapshot wait, since reconnect
209    /// replay may not include it yet. A running recovery keeps running with its budget;
210    /// [`BookRecoveryState::reset_on_reconnect`] describes when the reconnect wakes it.
211    pub fn reset_on_reconnect(&mut self) {
212        if !self.has_initial_wait() {
213            self.pending = None;
214        }
215
216        self.phase = BookPhase::Recovering;
217        self.recovery.reset_on_reconnect();
218    }
219
220    /// Arms a snapshot deadline, returning `false` while an initial snapshot wait is in flight.
221    pub fn arm_deadline(&mut self, deadline: Instant) -> bool {
222        if self.has_initial_wait() {
223            return false;
224        }
225
226        self.pending = Some(PendingSnapshot {
227            deadline: Some(deadline),
228            cancel: CancellationToken::new(),
229            gate: SnapshotGate::default(),
230        });
231
232        true
233    }
234
235    /// Removes an expired snapshot deadline, returning whether one expired.
236    pub fn take_expired(&mut self, now: Instant) -> bool {
237        let expired = self
238            .pending
239            .as_ref()
240            .is_some_and(|pending| pending.deadline.is_some_and(|deadline| deadline <= now));
241
242        if expired {
243            self.pending = None;
244        }
245
246        expired
247    }
248
249    /// Returns how long the book has gone without an accepted frame once that exceeds
250    /// `threshold`, re-arming the window so a dead feed is reported once per threshold.
251    ///
252    /// A book owned by a running recovery is not reported; the recovery logs its own attempts.
253    pub fn stale(&mut self, threshold: Duration, now: Instant) -> Option<Duration> {
254        if self.recovery.is_running() {
255            return None;
256        }
257
258        let elapsed = now.checked_duration_since(self.last_update)?;
259
260        if elapsed <= threshold {
261            return None;
262        }
263
264        self.last_update = now;
265        Some(elapsed)
266    }
267
268    fn has_initial_wait(&self) -> bool {
269        self.pending
270            .as_ref()
271            .is_some_and(|pending| pending.deadline.is_none() && !pending.cancel.is_cancelled())
272    }
273}
274
275#[cfg(test)]
276mod tests {
277    use nautilus_network::error::SendError;
278    use proptest::prelude::*;
279    use rstest::rstest;
280
281    use super::*;
282
283    type Book = BookSync<SendError, u64>;
284
285    fn synced(now: Instant) -> Book {
286        let mut book = Book::new(now);
287        assert!(book.accept_snapshot(1, now));
288        book
289    }
290
291    #[rstest]
292    fn new_book_waits_and_rejects_incrementals() {
293        let now = Instant::now();
294        let mut book = Book::new(now);
295
296        let advanced = book.advance(2, now);
297
298        assert_eq!(*book.phase(), BookPhase::Waiting);
299        assert!(!advanced);
300        assert_eq!(book.position(), None);
301    }
302
303    #[rstest]
304    fn snapshot_syncs_and_incrementals_advance() {
305        let now = Instant::now();
306        let mut book = Book::new(now);
307
308        let accepted = book.accept_snapshot(10, now);
309        let advanced = book.advance(11, now);
310
311        assert!(accepted);
312        assert!(advanced);
313        assert_eq!(*book.phase(), BookPhase::Synced(11));
314    }
315
316    #[rstest]
317    fn closed_subscription_gate_blocks_snapshot() {
318        let now = Instant::now();
319        let mut book = Book::new(now);
320        let gate = SnapshotGate::default();
321        gate.lock().close();
322        let cancel = book.expect_snapshot(gate.clone());
323
324        let while_sending = book.accept_snapshot(1, now);
325        gate.open();
326        let after_send = book.accept_snapshot(1, now);
327
328        assert!(!while_sending);
329        assert!(after_send);
330        assert!(cancel.is_cancelled());
331        assert!(!book.has_pending_snapshot());
332    }
333
334    #[rstest]
335    fn gap_requests_recovery_until_claimed() {
336        let now = Instant::now();
337        let mut book = synced(now);
338
339        let first = book.gap();
340        let again = book.gap();
341        let recovery = book.claim().unwrap();
342        let owned = book.gap();
343
344        assert_eq!(first, BookSequenceOutcome::Recover);
345        assert_eq!(again, BookSequenceOutcome::Recover);
346        assert_eq!(owned, BookSequenceOutcome::Suppress);
347        assert!(recovery.is_running());
348        assert_eq!(*book.phase(), BookPhase::Recovering);
349    }
350
351    #[rstest]
352    fn armed_deadline_owns_book_after_reconnect() {
353        let now = Instant::now();
354        let mut book = synced(now);
355        book.reset_on_reconnect();
356        let armed = book.arm_deadline(now + Duration::from_secs(10));
357
358        let outcome = book.gap();
359        let early = book.take_expired(now + Duration::from_secs(9));
360        let expired = book.take_expired(now + Duration::from_secs(10));
361        let unowned = book.gap();
362
363        assert!(armed);
364        assert_eq!(outcome, BookSequenceOutcome::Suppress);
365        assert!(!early);
366        assert!(expired);
367        assert_eq!(unowned, BookSequenceOutcome::Recover);
368    }
369
370    #[rstest]
371    fn recovery_gate_blocks_snapshot_until_write_confirmed() {
372        let now = Instant::now();
373        let mut book = synced(now);
374        let recovery = book.claim().unwrap();
375        assert!(recovery.begin_replacement());
376
377        let during_write = book.accept_snapshot(5, now);
378        recovery.gate.open();
379        let after_write = book.accept_snapshot(5, now);
380
381        assert!(!during_write);
382        assert!(after_write);
383        assert!(recovery.is_accepted());
384        assert_eq!(*book.phase(), BookPhase::Synced(5));
385    }
386
387    #[rstest]
388    fn reconnect_keeps_initial_send_wait() {
389        let now = Instant::now();
390        let mut book = Book::new(now);
391        let gate = SnapshotGate::default();
392        gate.lock().close();
393        let cancel = book.expect_snapshot(gate);
394
395        book.reset_on_reconnect();
396        let armed = book.arm_deadline(now + Duration::from_secs(10));
397
398        assert!(!armed);
399        assert!(!cancel.is_cancelled());
400        assert!(book.is_send_pending());
401    }
402
403    #[rstest]
404    fn claim_makes_new_episode_current() {
405        let now = Instant::now();
406        let mut book = synced(now);
407        let first = book.claim().unwrap();
408        assert!(first.accept());
409        let second = book.claim().unwrap();
410
411        assert!(!book.is_current(&first));
412        assert!(book.is_current(&second));
413    }
414
415    #[rstest]
416    fn reject_reaches_only_running_recovery() {
417        let now = Instant::now();
418        let mut book = synced(now);
419        let unowned = book.reject(SendError::Closed);
420        let recovery = book.claim().unwrap();
421
422        let delivered = book.reject(SendError::Closed);
423
424        assert!(!unowned);
425        assert!(delivered);
426        assert!(matches!(
427            *recovery.outcome.borrow(),
428            BookRecoveryOutcome::Rejected(SendError::Closed)
429        ));
430    }
431
432    #[rstest]
433    fn stale_skips_running_recovery_and_rearms() {
434        let now = Instant::now();
435        let threshold = Duration::from_secs(5);
436        let mut book = synced(now);
437
438        let fresh = book.stale(threshold, now + Duration::from_secs(5));
439        let first = book.stale(threshold, now + Duration::from_secs(6));
440        let rearmed = book.stale(threshold, now + Duration::from_secs(7));
441        let _recovery = book.claim().unwrap();
442        let owned = book.stale(threshold, now + Duration::from_secs(60));
443
444        assert_eq!(fresh, None);
445        assert_eq!(first, Some(Duration::from_secs(6)));
446        assert_eq!(rearmed, None);
447        assert_eq!(owned, None);
448    }
449
450    #[derive(Debug, Clone)]
451    enum Op {
452        Subscribe { sending: bool },
453        ConfirmSend,
454        Snapshot,
455        Advance,
456        Gap,
457        Claim,
458        BeginReplacement,
459        OpenRecoveryGate,
460        Reject,
461        CancelRecovery,
462        Reconnect,
463        Arm(u8),
464        Expire,
465        Tick(u8),
466        Stale(u8),
467    }
468
469    fn op() -> impl Strategy<Value = Op> {
470        prop_oneof![
471            any::<bool>().prop_map(|sending| Op::Subscribe { sending }),
472            Just(Op::ConfirmSend),
473            Just(Op::Snapshot),
474            Just(Op::Advance),
475            Just(Op::Gap),
476            Just(Op::Claim),
477            Just(Op::BeginReplacement),
478            Just(Op::OpenRecoveryGate),
479            Just(Op::Reject),
480            Just(Op::CancelRecovery),
481            Just(Op::Reconnect),
482            (1u8..30).prop_map(Op::Arm),
483            Just(Op::Expire),
484            (0u8..30).prop_map(Op::Tick),
485            (1u8..30).prop_map(Op::Stale),
486        ]
487    }
488
489    // Tracks what the contract says the book must do, independent of its internal fields
490    #[derive(Debug, Default)]
491    struct Model {
492        send_gate: Option<SnapshotGate>,
493        initial_wait: Option<CancellationToken>,
494        deadline: Option<Instant>,
495        recovery: Option<Arc<BookRecovery<SendError>>>,
496        synced: bool,
497    }
498
499    impl Model {
500        fn running(&self) -> Option<&Arc<BookRecovery<SendError>>> {
501            self.recovery.as_ref().filter(|r| r.is_running())
502        }
503
504        fn sending(&self) -> bool {
505            self.initial_wait.is_some()
506                && self
507                    .send_gate
508                    .as_ref()
509                    .is_some_and(|g| g.lock().is_closed())
510        }
511
512        fn clear_pending(&mut self) {
513            if let Some(cancel) = self.initial_wait.take() {
514                assert!(cancel.is_cancelled(), "dropped initial wait must cancel");
515            }
516
517            self.deadline = None;
518        }
519    }
520
521    proptest! {
522        #[rstest]
523        fn schedule_keeps_sync_contract(ops in prop::collection::vec(op(), 1..80)) {
524            let mut now = Instant::now();
525            let mut book = Book::new(now);
526            let mut model = Model::default();
527            let mut position = 0u64;
528            let mut last_accepted = now;
529
530            for op in ops {
531                match op {
532                    Op::Subscribe { sending } => {
533                        // A resubscribe replaces the book, cancelling the previous episode
534                        let previous = model.recovery.take();
535                        let gate = SnapshotGate::default();
536                        if sending {
537                            gate.lock().close();
538                        }
539                        book = Book::new(now);
540                        let cancel = book.expect_snapshot(gate.clone());
541                        model = Model {
542                            send_gate: Some(gate),
543                            initial_wait: Some(cancel),
544                            ..Model::default()
545                        };
546                        last_accepted = now;
547
548                        if let Some(previous) = previous {
549                            prop_assert!(previous.cancellation.is_cancelled());
550                        }
551                        prop_assert_eq!(book.is_send_pending(), sending);
552                    }
553                    Op::ConfirmSend => {
554                        if let Some(gate) = &model.send_gate {
555                            gate.open();
556                        }
557                    }
558                    Op::Snapshot => {
559                        position += 1;
560                        let blocked = model.sending()
561                            || model.running().is_some_and(|r| r.gate.lock().is_closed());
562                        let accepted = book.accept_snapshot(position, now);
563
564                        prop_assert_eq!(accepted, !blocked);
565                        if accepted {
566                            model.clear_pending();
567                            model.synced = true;
568                            last_accepted = now;
569                            prop_assert!(!book.has_pending_snapshot());
570                            prop_assert!(model.running().is_none());
571                        }
572                    }
573                    Op::Advance => {
574                        position += 1;
575                        let advanced = book.advance(position, now);
576
577                        // Incremental output only while synced
578                        prop_assert_eq!(advanced, model.synced);
579                        if advanced {
580                            last_accepted = now;
581                        }
582                    }
583                    Op::Gap => {
584                        let owned = model.running().is_some() || model.deadline.is_some();
585                        let outcome = book.gap();
586                        model.synced = false;
587
588                        // An out-of-sync book always has an owner or asks for one
589                        prop_assert_eq!(*book.phase(), BookPhase::Recovering);
590                        prop_assert_eq!(outcome == BookSequenceOutcome::Recover, !owned);
591                    }
592                    Op::Claim => {
593                        let expected = model.running().is_none();
594                        let claimed = book.claim();
595
596                        // At most one running owner, and claiming retires any pending wait
597                        prop_assert_eq!(claimed.is_some(), expected);
598                        if let Some(recovery) = claimed {
599                            if let Some(previous) = model.recovery.replace(recovery) {
600                                prop_assert!(previous.cancellation.is_cancelled());
601                            }
602                            model.clear_pending();
603                            model.synced = false;
604                            prop_assert!(!book.has_pending_snapshot());
605                        }
606                        prop_assert!(model.running().is_some_and(|recovery| book.is_current(recovery)));
607                    }
608                    Op::BeginReplacement => {
609                        if let Some(recovery) = model.running() {
610                            prop_assert!(recovery.begin_replacement());
611                        }
612                    }
613                    Op::OpenRecoveryGate => {
614                        if let Some(recovery) = model.running() {
615                            recovery.gate.open();
616                        }
617                    }
618                    Op::Reject => {
619                        let delivered = book.reject(SendError::Closed);
620
621                        prop_assert_eq!(delivered, model.running().is_some());
622                    }
623                    Op::CancelRecovery => {
624                        if let Some(recovery) = model.running() {
625                            recovery.cancellation.cancel();
626                        }
627
628                        // Rejection reaches only a running recovery, so nothing receives it
629                        prop_assert!(!book.reject(SendError::Closed));
630                    }
631                    Op::Reconnect => {
632                        let running = model.running().cloned();
633                        let waiting = model.initial_wait.clone();
634                        book.reset_on_reconnect();
635                        model.synced = false;
636                        model.deadline = None;
637
638                        // Reconnect keeps a running recovery and an initial send in flight
639                        prop_assert_eq!(*book.phase(), BookPhase::Recovering);
640                        match &running {
641                            Some(recovery) => prop_assert!(recovery.is_running() && book.is_current(recovery)),
642                            None => prop_assert!(!book.reject(SendError::Closed)),
643                        }
644
645                        if let Some(waiting) = waiting {
646                            prop_assert!(!waiting.is_cancelled());
647                            prop_assert!(book.has_pending_snapshot());
648                        }
649                    }
650                    Op::Arm(secs) => {
651                        let deadline = now + Duration::from_secs(u64::from(secs));
652                        let armed = book.arm_deadline(deadline);
653
654                        // An initial send in flight keeps its wait
655                        prop_assert_eq!(armed, model.initial_wait.is_none());
656                        if armed {
657                            model.deadline = Some(deadline);
658                        }
659                    }
660                    Op::Expire => {
661                        let expected = model.deadline.is_some_and(|d| d <= now);
662                        let expired = book.take_expired(now);
663
664                        prop_assert_eq!(expired, expected);
665                        if expired {
666                            model.deadline = None;
667                        }
668                    }
669                    Op::Tick(secs) => {
670                        now += Duration::from_secs(u64::from(secs));
671                    }
672                    Op::Stale(secs) => {
673                        let threshold = Duration::from_secs(u64::from(secs));
674                        let reported = book.stale(threshold, now);
675
676                        // Every book without a running owner is reported once past the threshold
677                        if model.running().is_some() {
678                            prop_assert_eq!(reported, None);
679                        } else {
680                            let elapsed = now.duration_since(last_accepted);
681                            prop_assert_eq!(reported, (elapsed > threshold).then_some(elapsed));
682                            if reported.is_some() {
683                                last_accepted = now;
684                            }
685                        }
686                    }
687                }
688            }
689        }
690    }
691}
nautilus_live/book/sync.rs

nautilus_live/book/
sync.rs
