nautilus_system/
event_store.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//! Kernel-facing seam for run-lifecycle event-sourcing.
17//!
18//! The [`KernelEventStore`] trait is the surface [`crate::kernel::NautilusKernel`] uses to wire
19//! a durable event-sourcing session into its boot, snapshot, and seal flow. The concrete
20//! implementation lives in `nautilus-event-store` so that crate can be developed and versioned
21//! independently of `nautilus-system`; callers inject an implementation through the builder
22//! (see [`crate::builder::NautilusKernelBuilder::with_event_store`]).
23
24use std::{cell::RefCell, fmt::Debug, path::PathBuf, rc::Rc, time::Duration};
25
26use indexmap::IndexMap;
27use nautilus_common::{cache::Cache, clock::Clock, enums::Environment};
28use nautilus_core::{UUID4, UnixNanos};
29use nautilus_execution::engine::SnapshotAnchorer;
30use serde::{Deserialize, Serialize};
31
32/// Factory closure invoked by the kernel to construct an injected event-store implementation.
33///
34/// Receives the kernel's instance id and clock so the resulting [`KernelEventStore`]
35/// implementation scans the same on-disk run directory the kernel later passes to
36/// `restore_parent_cache`/`open`, and stamps lifecycle timestamps against the same time
37/// source the kernel uses.
38pub type EventStoreFactory = Box<
39    dyn FnOnce(UUID4, Rc<RefCell<dyn Clock>>) -> anyhow::Result<Box<dyn KernelEventStore>>
40        + 'static,
41>;
42
43/// The component manifest captured into the event-store `RunStarted` entry.
44///
45/// Replay binds actors, strategies, algorithms, subscriptions, and command endpoints from
46/// this manifest without consulting external configuration.
47#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
48pub struct RegisteredComponents {
49    /// Registered actor ids and their config hashes.
50    pub actors: IndexMap<String, String>,
51    /// Registered strategy ids and their config hashes.
52    pub strategies: IndexMap<String, String>,
53    /// Registered algorithm ids and their config hashes.
54    pub algorithms: IndexMap<String, String>,
55    /// Subscription bindings active at run start.
56    pub subscriptions: Vec<String>,
57    /// Endpoint registrations active at run start.
58    pub endpoints: Vec<String>,
59}
60
61/// Kernel-facing seam for event-sourcing lifecycle integration.
62///
63/// `NautilusKernel` drives the open/restore/seal sequence through this trait so the concrete
64/// event-store machinery (writers, readers, bus tap, redb backend) lives outside
65/// `nautilus-system`. Implementations are typically built by the caller and injected via
66/// [`crate::builder::NautilusKernelBuilder::with_event_store`].
67pub trait KernelEventStore: Debug {
68    /// Restores cache state from a configured replay source or recovered parent run.
69    ///
70    /// Implementations may open a sealed replay source, validate its snapshot anchor, and
71    /// replay the tail directly into `cache`. The kernel calls this once before [`Self::open`].
72    ///
73    /// # Errors
74    ///
75    /// Returns an error when the source reader, snapshot restore, decode, or cache apply
76    /// step fails.
77    fn restore_parent_cache(&mut self, instance_id: UUID4, cache: &mut Cache)
78    -> anyhow::Result<()>;
79
80    /// Opens a fresh run for the current kernel session.
81    ///
82    /// `components` carries the registered manifest written to the run's `RunStarted` entry.
83    /// `environment` selects the clock source the implementation uses to stamp publish
84    /// timestamps. Idempotency across reset/rerun is the implementation's responsibility.
85    ///
86    /// # Errors
87    ///
88    /// Returns an error when opening the new run, spawning the writer, or blocking on the
89    /// initial entry ack fails.
90    fn open(
91        &mut self,
92        instance_id: UUID4,
93        components: &RegisteredComponents,
94        environment: Environment,
95    ) -> anyhow::Result<()>;
96
97    /// Returns a snapshot anchorer for the currently open run, when capture is active.
98    ///
99    /// The execution engine installs the returned callback so position snapshots commit a
100    /// matching anchor entry against the durable high-watermark.
101    fn snapshot_anchorer(&self) -> Option<SnapshotAnchorer>;
102
103    /// Seals the open run by writing the terminal entry and updating the manifest.
104    ///
105    /// Idempotent: a closed or absent session is a no-op. Halted sessions defer the seal to
106    /// the next-boot recovery sweep.
107    fn seal(&mut self, ts_init: UnixNanos);
108
109    /// Returns the run id of the currently open run, when capture is active.
110    fn run_id(&self) -> Option<&str>;
111
112    /// Returns the configured replay source or recovered parent run id, when present.
113    fn parent_run_id(&self) -> Option<&str>;
114
115    /// Returns whether the current config enables event-store replay.
116    ///
117    /// Event-store replay restores cache state and opens a child run for inspection. The kernel
118    /// promotes this config state to runtime state only after restore and open both succeed.
119    fn is_event_store_replay_configured(&self) -> bool {
120        false
121    }
122
123    /// Returns whether the implementation has signaled a fail-stop condition.
124    fn is_halted(&self) -> bool;
125}
126
127/// How the supervisor prunes sealed run files.
128///
129/// The kernel records the choice in the manifest's `feature_flags`; actual retention
130/// enforcement is performed by a separate supervisor process and is out of scope for
131/// the kernel boot path.
132#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
133pub enum RetentionMode {
134    /// Keep every sealed run; never reclaim.
135    #[default]
136    Full,
137    /// Keep at most `keep_last` sealed runs; the supervisor reclaims older files.
138    Bounded {
139        /// The number of sealed runs to retain.
140        keep_last: usize,
141    },
142    /// Keep the manifest plus a snapshot anchor and the tail since the anchor; older
143    /// entries reclaim once a newer anchor is durable.
144    SnapshotAnchored,
145}
146
147/// Per-run identification data the kernel populates from build metadata.
148///
149/// The kernel records what is available at run start; downstream binaries refine these
150/// values when their build-time wiring populates them. Defaults are placeholders so the
151/// kernel can boot before the binary-hash and crate-versions wiring is finalized.
152#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
153pub struct RunIdentity {
154    /// A hex-encoded hash of the trader binary.
155    pub binary_hash: String,
156    /// The entry payload schema version.
157    pub schema_version: u32,
158    /// A hex-encoded hash of `Cargo.lock` or an equivalent crate version manifest.
159    pub crate_versions: String,
160    /// The active Cargo features for the trader binary.
161    pub feature_flags: Vec<String>,
162    /// Per-adapter version stamp keyed by adapter name.
163    pub adapter_versions: IndexMap<String, String>,
164    /// A hex-encoded hash of the kernel configuration.
165    pub config_hash: String,
166    /// The deterministic seed, populated when the run executes under a seeded mode.
167    pub seed: Option<u64>,
168}
169
170/// The id of a captured run: `<start_ts_init>-<short_uuid>`, sortable by start time.
171///
172/// The runtime constructs this from the kernel's start timestamp plus a fresh `UUID4` so
173/// the representation stays stable across processes and platforms.
174pub type RunId = String;
175
176/// Default maximum interval between data-marker cursor snapshots when no entry boundary occurs.
177pub const DEFAULT_DATA_MARKER_SAFETY_FLUSH_INTERVAL: Duration = Duration::from_secs(1);
178/// Default capacity of the data-marker writer's bounded submit channel.
179pub const DEFAULT_DATA_MARKER_CHANNEL_CAPACITY: usize = 10_000;
180
181/// Market-data class enabled for data-marker capture.
182#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
183pub enum DataMarkerClass {
184    /// Order-book delta stream.
185    BookDeltas,
186    /// Order-book depth snapshot stream.
187    #[serde(alias = "BookDepth10")]
188    BookDepth,
189    /// Quote (level-1 bid/ask) stream.
190    Quote,
191    /// Trade (last sale) stream.
192    Trade,
193    /// Bar (OHLCV aggregate) stream.
194    Bar,
195}
196
197impl DataMarkerClass {
198    /// All builtin data-marker classes in canonical order.
199    pub const ALL: [Self; 5] = [
200        Self::BookDeltas,
201        Self::BookDepth,
202        Self::Quote,
203        Self::Trade,
204        Self::Bar,
205    ];
206}
207
208/// Opt-in data-marker sidecar settings for an event-store run.
209#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
210pub struct DataMarkerConfig {
211    /// Market-data classes captured into marker cursors.
212    #[serde(default = "default_data_marker_classes")]
213    pub classes: Vec<DataMarkerClass>,
214    /// Maximum interval between cursor snapshots when data advances without entry submissions.
215    #[serde(default = "default_data_marker_safety_flush_interval")]
216    pub safety_flush_interval: Duration,
217    /// Capacity of the marker writer's bounded submit channel.
218    #[serde(default = "default_data_marker_channel_capacity")]
219    pub channel_capacity: usize,
220    /// Instrument identifiers that emit one high-fidelity marker per observed data message.
221    #[serde(default)]
222    pub high_fidelity: Vec<String>,
223}
224
225impl Default for DataMarkerConfig {
226    fn default() -> Self {
227        Self {
228            classes: default_data_marker_classes(),
229            safety_flush_interval: DEFAULT_DATA_MARKER_SAFETY_FLUSH_INTERVAL,
230            channel_capacity: DEFAULT_DATA_MARKER_CHANNEL_CAPACITY,
231            high_fidelity: Vec::new(),
232        }
233    }
234}
235
236fn default_data_marker_classes() -> Vec<DataMarkerClass> {
237    DataMarkerClass::ALL.to_vec()
238}
239
240const fn default_data_marker_safety_flush_interval() -> Duration {
241    DEFAULT_DATA_MARKER_SAFETY_FLUSH_INTERVAL
242}
243
244const fn default_data_marker_channel_capacity() -> usize {
245    DEFAULT_DATA_MARKER_CHANNEL_CAPACITY
246}
247
248/// Configuration for the kernel-managed event store run lifecycle.
249#[derive(Clone, Debug, Serialize, Deserialize)]
250pub struct EventStoreConfig {
251    /// Root directory; the backend creates `<base_dir>/<instance_id>/<run_id>.redb`.
252    pub base_dir: PathBuf,
253    /// Stable identification for this trader instance and binary.
254    pub identity: RunIdentity,
255    /// How the supervisor reclaims sealed run files.
256    pub retention: RetentionMode,
257    /// Sealed run to restore cache state from before opening a fresh run.
258    ///
259    /// When set, this enables event-store replay: the kernel restores cache state from this run,
260    /// records it as the parent link for the fresh child run, and then skips engines, clients,
261    /// trader startup, and live reconciliation. Quarantined runs are rejected.
262    pub replay_from_run_id: Option<RunId>,
263    /// Data-marker sidecar settings. `None` disables marker capture for the run.
264    #[serde(default)]
265    pub data_markers: Option<DataMarkerConfig>,
266    /// Capacity of the writer's bounded submit channel.
267    pub channel_capacity: usize,
268    /// Maximum entries collected before the writer forces a commit.
269    pub max_batch_entries: usize,
270    /// Maximum time a batch may accumulate before the writer forces a commit.
271    pub max_batch_latency: Duration,
272    /// Submit-side stall ceiling that triggers writer fail-stop.
273    pub halt_threshold: Duration,
274    /// Maximum time to wait for the `RunStarted` entry to durably commit before the
275    /// kernel surfaces an event-store boot error.
276    pub run_started_timeout: Duration,
277}
278
279impl Default for EventStoreConfig {
280    fn default() -> Self {
281        Self {
282            base_dir: PathBuf::new(),
283            identity: RunIdentity::default(),
284            retention: RetentionMode::default(),
285            replay_from_run_id: None,
286            data_markers: None,
287            channel_capacity: 10_000,
288            max_batch_entries: 100,
289            max_batch_latency: Duration::from_millis(5),
290            halt_threshold: Duration::from_millis(250),
291            run_started_timeout: Duration::from_secs(5),
292        }
293    }
294}
295
296#[cfg(test)]
297mod tests {
298    use rstest::rstest;
299
300    use super::*;
301
302    #[rstest]
303    fn event_store_config_serde_roundtrip() {
304        let config = EventStoreConfig::default();
305        let json = serde_json::to_string(&config).expect("serialize");
306        let restored: EventStoreConfig = serde_json::from_str(&json).expect("deserialize");
307
308        assert_eq!(restored.channel_capacity, config.channel_capacity);
309        assert_eq!(restored.max_batch_entries, config.max_batch_entries);
310        assert_eq!(restored.max_batch_latency, config.max_batch_latency);
311        assert_eq!(restored.halt_threshold, config.halt_threshold);
312        assert_eq!(restored.run_started_timeout, config.run_started_timeout);
313        assert_eq!(restored.base_dir, config.base_dir);
314        assert_eq!(restored.retention, config.retention);
315        assert_eq!(restored.replay_from_run_id, config.replay_from_run_id);
316        assert_eq!(restored.identity, config.identity);
317        assert_eq!(restored.data_markers, config.data_markers);
318    }
319
320    #[rstest]
321    fn data_marker_config_serde_roundtrip() {
322        let config = EventStoreConfig {
323            data_markers: Some(DataMarkerConfig {
324                classes: vec![DataMarkerClass::Quote, DataMarkerClass::BookDeltas],
325                safety_flush_interval: Duration::from_millis(250),
326                channel_capacity: 512,
327                high_fidelity: vec!["ETHUSDT-PERP.BINANCE".to_string()],
328            }),
329            ..Default::default()
330        };
331        let json = serde_json::to_string(&config).expect("serialize");
332        let restored: EventStoreConfig = serde_json::from_str(&json).expect("deserialize");
333
334        assert_eq!(restored.data_markers, config.data_markers);
335    }
336
337    #[rstest]
338    fn data_marker_class_serde_accepts_legacy_depth10_spelling() {
339        assert_eq!(
340            serde_json::from_str::<DataMarkerClass>(r#""BookDepth10""#).unwrap(),
341            DataMarkerClass::BookDepth
342        );
343        assert_eq!(
344            serde_json::from_str::<DataMarkerClass>(r#""BookDepth""#).unwrap(),
345            DataMarkerClass::BookDepth
346        );
347        assert_eq!(
348            serde_json::to_string(&DataMarkerClass::BookDepth).unwrap(),
349            r#""BookDepth""#
350        );
351    }
352
353    #[rstest]
354    fn retention_mode_serde_roundtrip() {
355        for mode in [
356            RetentionMode::Full,
357            RetentionMode::Bounded { keep_last: 5 },
358            RetentionMode::SnapshotAnchored,
359        ] {
360            let json = serde_json::to_string(&mode).expect("serialize");
361            let restored: RetentionMode = serde_json::from_str(&json).expect("deserialize");
362            assert_eq!(restored, mode);
363        }
364    }
365
366    #[rstest]
367    fn event_store_config_default_values() {
368        let config = EventStoreConfig::default();
369
370        assert_eq!(config.channel_capacity, 10_000);
371        assert_eq!(config.max_batch_entries, 100);
372        assert_eq!(config.max_batch_latency, Duration::from_millis(5));
373        assert_eq!(config.halt_threshold, Duration::from_millis(250));
374        assert_eq!(config.run_started_timeout, Duration::from_secs(5));
375        assert_eq!(config.base_dir, PathBuf::new());
376        assert_eq!(config.retention, RetentionMode::Full);
377        assert!(config.replay_from_run_id.is_none());
378        assert_eq!(config.identity, RunIdentity::default());
379        assert!(config.data_markers.is_none());
380    }
381
382    #[rstest]
383    fn data_markers_default_is_none() {
384        let config = EventStoreConfig::default();
385
386        assert!(config.data_markers.is_none());
387    }
388}
nautilus_system/event_store.rs

nautilus_system/
event_store.rs
