nautilus_live/book/
snapshot.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//! Subscription write gates and pending snapshot lifetimes.
17//!
18//! - [`SnapshotGate`] prevents snapshot acceptance while a replacement subscription is being sent.
19//! - [`PendingSnapshot`] owns the cancellation token, gate, and optional deadline for a pending
20//!   snapshot. Removing it cancels the associated wait.
21//! - [`snapshot_expired`] waits for a snapshot deadline or cancellation. A zero timeout disables
22//!   monitoring.
23//!
24//! # Snapshot Lifecycle
25//!
26//! Gates start open. The adapter closes a gate before a guarded write and opens it only after the
27//! intended connection confirms that write. Snapshot acceptance checks the gate under its lock.
28//!
29//! An adapter can keep an absolute deadline for monitoring or await [`snapshot_expired`] after
30//! write confirmation. Accepting a snapshot or retiring its subscription drops the pending owner,
31//! cancelling obsolete work.
32//!
33//! # Adapters
34//!
35//! The adapter verifies sequence validity and any subscription-generation correlation. The gate
36//! alone cannot identify which request or connection produced a snapshot.
37//!
38//! Scheduling waits and starting recovery remain adapter responsibilities.
39
40use std::sync::Arc;
41
42use nautilus_common::live::dst::time::{self, Duration, Instant};
43use parking_lot::{Mutex, MutexGuard};
44use tokio_util::sync::CancellationToken;
45
46/// Coordinates subscription sends with snapshot acceptance.
47///
48/// Clones share one gate, which is initially open.
49#[derive(Debug, Clone, Default)]
50pub struct SnapshotGate {
51    closed: Arc<Mutex<bool>>,
52}
53
54impl SnapshotGate {
55    /// Opens the gate after the subscription write completes.
56    pub fn open(&self) {
57        *self.closed.lock() = false;
58    }
59
60    /// Locks snapshot acceptance against a replacement send.
61    #[must_use]
62    pub fn lock(&self) -> SnapshotGateGuard<'_> {
63        SnapshotGateGuard(self.closed.lock())
64    }
65}
66
67/// Locked snapshot acceptance gate.
68#[derive(Debug)]
69pub struct SnapshotGateGuard<'a>(MutexGuard<'a, bool>);
70
71impl SnapshotGateGuard<'_> {
72    /// Suppresses snapshots until the replacement write completes.
73    pub fn close(&mut self) {
74        *self.0 = true;
75    }
76
77    /// Returns whether snapshot acceptance is suppressed.
78    #[must_use]
79    pub fn is_closed(&self) -> bool {
80        *self.0
81    }
82}
83
84/// Pending snapshot ownership, cancelled on acceptance, replacement, or removal.
85#[derive(Debug)]
86pub struct PendingSnapshot {
87    pub deadline: Option<Instant>,
88    pub cancel: CancellationToken,
89    pub gate: SnapshotGate,
90}
91
92impl Drop for PendingSnapshot {
93    fn drop(&mut self) {
94        self.cancel.cancel();
95    }
96}
97
98/// Returns whether the snapshot deadline expires before cancellation.
99///
100/// A zero timeout disables monitoring. Adapters cancel the token when accepting a snapshot
101/// or retiring its subscription, so obsolete waits cannot start replacement work.
102pub async fn snapshot_expired(cancel: &CancellationToken, timeout: Duration) -> bool {
103    !timeout.is_zero() && time::timeout(timeout, cancel.cancelled()).await.is_err()
104}
nautilus_live/book/snapshot.rs

nautilus_live/book/
snapshot.rs
