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

nautilus_testkit/
cache.rs
