nautilus_testkit/
cache.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//! Stateful cache database test double.
17
18use std::sync::Arc;
19
20use ahash::AHashMap;
21use bytes::Bytes;
22use indexmap::IndexMap;
23use nautilus_common::{
24    cache::database::{CacheDatabaseAdapter, CacheMap},
25    signal::Signal,
26};
27use nautilus_core::UnixNanos;
28use nautilus_model::{
29    accounts::AccountAny,
30    data::{
31        Bar, CustomData, DataType, FundingRateUpdate, QuoteTick, TradeTick,
32        greeks::{GreeksData, YieldCurveData},
33    },
34    events::{OrderEventAny, OrderSnapshot, position::snapshot::PositionSnapshot},
35    identifiers::{
36        AccountId, ActorId, ClientId, ClientOrderId, InstrumentId, PositionId, StrategyId,
37        VenueOrderId,
38    },
39    instruments::{InstrumentAny, SyntheticInstrument},
40    orderbook::OrderBook,
41    orders::OrderAny,
42    position::Position,
43    types::{Currency, Money},
44};
45use parking_lot::Mutex;
46use ustr::Ustr;
47
48#[expect(
49    clippy::struct_excessive_bools,
50    reason = "independent switches cover lifecycle persistence failure modes"
51)]
52#[derive(Debug, Default)]
53struct TestCacheDatabaseState {
54    actors: AHashMap<ActorId, AHashMap<String, Bytes>>,
55    strategies: AHashMap<StrategyId, AHashMap<String, Bytes>>,
56    events: Vec<String>,
57    fail_load_actor: bool,
58    fail_load_strategy: bool,
59    fail_update_actor: bool,
60    fail_update_strategy: bool,
61    fail_update_position: bool,
62}
63
64/// Shared control and observation handle for [`TestCacheDatabase`].
65#[derive(Clone, Debug, Default)]
66pub struct TestCacheDatabaseControl {
67    state: Arc<Mutex<TestCacheDatabaseState>>,
68}
69
70impl TestCacheDatabaseControl {
71    /// Creates an adapter and its shared control handle.
72    #[must_use]
73    pub fn create() -> (TestCacheDatabase, Self) {
74        let control = Self::default();
75        (
76            TestCacheDatabase {
77                control: control.clone(),
78            },
79            control,
80        )
81    }
82
83    /// Records an event in the shared lifecycle log.
84    pub fn record(&self, event: impl Into<String>) {
85        self.state.lock().events.push(event.into());
86    }
87
88    /// Returns the recorded lifecycle events.
89    #[must_use]
90    pub fn events(&self) -> Vec<String> {
91        self.state.lock().events.clone()
92    }
93
94    /// Seeds actor state for a later load.
95    pub fn set_actor_state(&self, actor_id: ActorId, state: &IndexMap<String, Vec<u8>>) {
96        self.state
97            .lock()
98            .actors
99            .insert(actor_id, encode_state(state));
100    }
101
102    /// Seeds strategy state for a later load.
103    pub fn set_strategy_state(&self, strategy_id: StrategyId, state: &IndexMap<String, Vec<u8>>) {
104        self.state
105            .lock()
106            .strategies
107            .insert(strategy_id, encode_state(state));
108    }
109
110    /// Returns persisted actor state.
111    #[must_use]
112    pub fn actor_state(&self, actor_id: &ActorId) -> Option<IndexMap<String, Vec<u8>>> {
113        self.state
114            .lock()
115            .actors
116            .get(actor_id)
117            .cloned()
118            .map(decode_state)
119    }
120
121    /// Returns persisted strategy state.
122    #[must_use]
123    pub fn strategy_state(&self, strategy_id: &StrategyId) -> Option<IndexMap<String, Vec<u8>>> {
124        self.state
125            .lock()
126            .strategies
127            .get(strategy_id)
128            .cloned()
129            .map(decode_state)
130    }
131
132    /// Configures actor loads to fail.
133    pub fn set_fail_load_actor(&self, fail: bool) {
134        self.state.lock().fail_load_actor = fail;
135    }
136
137    /// Configures strategy loads to fail.
138    pub fn set_fail_load_strategy(&self, fail: bool) {
139        self.state.lock().fail_load_strategy = fail;
140    }
141
142    /// Configures actor updates to fail.
143    pub fn set_fail_update_actor(&self, fail: bool) {
144        self.state.lock().fail_update_actor = fail;
145    }
146
147    /// Configures strategy updates to fail.
148    pub fn set_fail_update_strategy(&self, fail: bool) {
149        self.state.lock().fail_update_strategy = fail;
150    }
151
152    /// Configures position updates to fail.
153    pub fn set_fail_update_position(&self, fail: bool) {
154        self.state.lock().fail_update_position = fail;
155    }
156}
157
158/// Stateful cache database adapter for lifecycle tests.
159#[derive(Debug)]
160pub struct TestCacheDatabase {
161    control: TestCacheDatabaseControl,
162}
163
164#[async_trait::async_trait]
165impl CacheDatabaseAdapter for TestCacheDatabase {
166    fn close(&mut self) -> anyhow::Result<()> {
167        self.control.record("database.close");
168        Ok(())
169    }
170
171    fn flush(&mut self) -> anyhow::Result<()> {
172        Ok(())
173    }
174
175    async fn load_all(&self) -> anyhow::Result<CacheMap> {
176        Ok(CacheMap::default())
177    }
178
179    fn load(&self) -> anyhow::Result<AHashMap<String, Bytes>> {
180        Ok(AHashMap::new())
181    }
182
183    async fn load_currencies(&self) -> anyhow::Result<AHashMap<Ustr, Currency>> {
184        Ok(AHashMap::new())
185    }
186
187    async fn load_instruments(&self) -> anyhow::Result<AHashMap<InstrumentId, InstrumentAny>> {
188        Ok(AHashMap::new())
189    }
190
191    async fn load_synthetics(&self) -> anyhow::Result<AHashMap<InstrumentId, SyntheticInstrument>> {
192        Ok(AHashMap::new())
193    }
194
195    async fn load_accounts(&self) -> anyhow::Result<AHashMap<AccountId, AccountAny>> {
196        Ok(AHashMap::new())
197    }
198
199    async fn load_orders(&self) -> anyhow::Result<AHashMap<ClientOrderId, OrderAny>> {
200        Ok(AHashMap::new())
201    }
202
203    async fn load_positions(&self) -> anyhow::Result<AHashMap<PositionId, Position>> {
204        Ok(AHashMap::new())
205    }
206
207    fn load_index_order_position(&self) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>> {
208        Ok(AHashMap::new())
209    }
210
211    fn load_index_order_client(&self) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>> {
212        Ok(AHashMap::new())
213    }
214
215    async fn load_currency(&self, _code: &Ustr) -> anyhow::Result<Option<Currency>> {
216        Ok(None)
217    }
218
219    async fn load_instrument(
220        &self,
221        _instrument_id: &InstrumentId,
222    ) -> anyhow::Result<Option<InstrumentAny>> {
223        Ok(None)
224    }
225
226    async fn load_synthetic(
227        &self,
228        _instrument_id: &InstrumentId,
229    ) -> anyhow::Result<Option<SyntheticInstrument>> {
230        Ok(None)
231    }
232
233    async fn load_account(&self, _account_id: &AccountId) -> anyhow::Result<Option<AccountAny>> {
234        Ok(None)
235    }
236
237    async fn load_order(
238        &self,
239        _client_order_id: &ClientOrderId,
240    ) -> anyhow::Result<Option<OrderAny>> {
241        Ok(None)
242    }
243
244    async fn load_position(&self, _position_id: &PositionId) -> anyhow::Result<Option<Position>> {
245        Ok(None)
246    }
247
248    fn load_actor(&self, actor_id: &ActorId) -> anyhow::Result<AHashMap<String, Bytes>> {
249        self.control.record(format!("actor.load:{actor_id}"));
250        let state = self.control.state.lock();
251        if state.fail_load_actor {
252            anyhow::bail!("test actor load failure");
253        }
254        Ok(state.actors.get(actor_id).cloned().unwrap_or_default())
255    }
256
257    fn load_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<AHashMap<String, Bytes>> {
258        self.control.record(format!("strategy.load:{strategy_id}"));
259        let state = self.control.state.lock();
260        if state.fail_load_strategy {
261            anyhow::bail!("test strategy load failure");
262        }
263        Ok(state
264            .strategies
265            .get(strategy_id)
266            .cloned()
267            .unwrap_or_default())
268    }
269
270    fn load_signals(&self, _name: &str) -> anyhow::Result<Vec<Signal>> {
271        Ok(Vec::new())
272    }
273
274    fn load_custom_data(&self, _data_type: &DataType) -> anyhow::Result<Vec<CustomData>> {
275        Ok(Vec::new())
276    }
277
278    fn load_order_snapshot(
279        &self,
280        _client_order_id: &ClientOrderId,
281    ) -> anyhow::Result<Option<OrderSnapshot>> {
282        Ok(None)
283    }
284
285    fn load_position_snapshot(
286        &self,
287        _position_id: &PositionId,
288    ) -> anyhow::Result<Option<PositionSnapshot>> {
289        Ok(None)
290    }
291
292    fn load_quotes(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<QuoteTick>> {
293        Ok(Vec::new())
294    }
295
296    fn load_trades(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<TradeTick>> {
297        Ok(Vec::new())
298    }
299
300    fn load_funding_rates(
301        &self,
302        _instrument_id: &InstrumentId,
303    ) -> anyhow::Result<Vec<FundingRateUpdate>> {
304        Ok(Vec::new())
305    }
306
307    fn load_bars(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<Bar>> {
308        Ok(Vec::new())
309    }
310
311    fn add(&self, _key: String, _value: Bytes) -> anyhow::Result<()> {
312        Ok(())
313    }
314
315    fn add_currency(&self, _currency: &Currency) -> anyhow::Result<()> {
316        Ok(())
317    }
318
319    fn add_instrument(&self, _instrument: &InstrumentAny) -> anyhow::Result<()> {
320        Ok(())
321    }
322
323    fn add_synthetic(&self, _synthetic: &SyntheticInstrument) -> anyhow::Result<()> {
324        Ok(())
325    }
326
327    fn add_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
328        Ok(())
329    }
330
331    fn add_order(&self, _order: &OrderAny, _client_id: Option<ClientId>) -> anyhow::Result<()> {
332        Ok(())
333    }
334
335    fn add_order_snapshot(&self, _snapshot: &OrderSnapshot) -> anyhow::Result<()> {
336        Ok(())
337    }
338
339    fn add_position(&self, _position: &Position) -> anyhow::Result<()> {
340        Ok(())
341    }
342
343    fn add_position_snapshot(&self, _snapshot: &PositionSnapshot) -> anyhow::Result<()> {
344        Ok(())
345    }
346
347    fn add_order_book(&self, _order_book: &OrderBook) -> anyhow::Result<()> {
348        Ok(())
349    }
350
351    fn add_signal(&self, _signal: &Signal) -> anyhow::Result<()> {
352        Ok(())
353    }
354
355    fn add_custom_data(&self, _data: &CustomData) -> anyhow::Result<()> {
356        Ok(())
357    }
358
359    fn add_quote(&self, _quote: &QuoteTick) -> anyhow::Result<()> {
360        Ok(())
361    }
362
363    fn add_trade(&self, _trade: &TradeTick) -> anyhow::Result<()> {
364        Ok(())
365    }
366
367    fn add_funding_rate(&self, _funding_rate: &FundingRateUpdate) -> anyhow::Result<()> {
368        Ok(())
369    }
370
371    fn add_bar(&self, _bar: &Bar) -> anyhow::Result<()> {
372        Ok(())
373    }
374
375    fn add_greeks(&self, _greeks: &GreeksData) -> anyhow::Result<()> {
376        Ok(())
377    }
378
379    fn add_yield_curve(&self, _yield_curve: &YieldCurveData) -> anyhow::Result<()> {
380        Ok(())
381    }
382
383    fn delete_actor(&self, _actor_id: &ActorId) -> anyhow::Result<()> {
384        Ok(())
385    }
386
387    fn delete_strategy(&self, _component_id: &StrategyId) -> anyhow::Result<()> {
388        Ok(())
389    }
390
391    fn delete_order(&self, _client_order_id: &ClientOrderId) -> anyhow::Result<()> {
392        Ok(())
393    }
394
395    fn delete_position(&self, _position_id: &PositionId) -> anyhow::Result<()> {
396        Ok(())
397    }
398
399    fn delete_account_event(&self, _account_id: &AccountId, _event_id: &str) -> anyhow::Result<()> {
400        Ok(())
401    }
402
403    fn index_venue_order_id(
404        &self,
405        _client_order_id: ClientOrderId,
406        _venue_order_id: VenueOrderId,
407    ) -> anyhow::Result<()> {
408        Ok(())
409    }
410
411    fn index_order_position(
412        &self,
413        _client_order_id: ClientOrderId,
414        _position_id: PositionId,
415    ) -> anyhow::Result<()> {
416        Ok(())
417    }
418
419    fn update_actor(
420        &self,
421        actor_id: &ActorId,
422        actor_state: &AHashMap<String, Bytes>,
423    ) -> anyhow::Result<()> {
424        self.control.record(format!("actor.update:{actor_id}"));
425        let mut state = self.control.state.lock();
426        if state.fail_update_actor {
427            anyhow::bail!("test actor update failure");
428        }
429        state.actors.insert(*actor_id, actor_state.clone());
430        Ok(())
431    }
432
433    fn update_strategy(
434        &self,
435        strategy_id: &StrategyId,
436        strategy_state: &AHashMap<String, Bytes>,
437    ) -> anyhow::Result<()> {
438        self.control
439            .record(format!("strategy.update:{strategy_id}"));
440        let mut state = self.control.state.lock();
441        if state.fail_update_strategy {
442            anyhow::bail!("test strategy update failure");
443        }
444        state
445            .strategies
446            .insert(*strategy_id, strategy_state.clone());
447        Ok(())
448    }
449
450    fn update_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
451        Ok(())
452    }
453
454    fn update_order(&self, _order_event: &OrderEventAny) -> anyhow::Result<()> {
455        Ok(())
456    }
457
458    fn update_position(&self, _position: &Position) -> anyhow::Result<()> {
459        if self.control.state.lock().fail_update_position {
460            anyhow::bail!("test position update failure");
461        }
462        Ok(())
463    }
464
465    fn snapshot_order_state(&self, _order: &OrderAny) -> anyhow::Result<()> {
466        Ok(())
467    }
468
469    fn snapshot_position_state(
470        &self,
471        _position: &Position,
472        _ts_snapshot: UnixNanos,
473        _unrealized_pnl: Option<Money>,
474    ) -> anyhow::Result<()> {
475        Ok(())
476    }
477
478    fn heartbeat(&self, _timestamp: UnixNanos) -> anyhow::Result<()> {
479        Ok(())
480    }
481}
482
483fn decode_state(state: AHashMap<String, Bytes>) -> IndexMap<String, Vec<u8>> {
484    state
485        .into_iter()
486        .map(|(key, value)| (key, value.to_vec()))
487        .collect()
488}
489
490fn encode_state(state: &IndexMap<String, Vec<u8>>) -> AHashMap<String, Bytes> {
491    state
492        .iter()
493        .map(|(key, value)| (key.clone(), Bytes::copy_from_slice(value)))
494        .collect()
495}
nautilus_testkit/cache.rs

nautilus_testkit/
cache.rs
