nautilus_common/cache/
position.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//! Position snapshot storage for the platform [`Cache`].
17//!
18//! Three mechanisms share the "position snapshot" name and live here together:
19//!
20//! - The NETTING archive ([`Cache::snapshot_position`]), which preserves each closed position
21//!   cycle before its ID is reused, and backs cross-cycle realized PnL.
22//! - The durable correction boundary ([`Cache::snapshot_position_encoded`],
23//!   [`Cache::restore_snapshot_blob`]), which produces and restores the encoded frames an event
24//!   store anchors.
25//! - The routine state snapshot ([`Cache::snapshot_position_state`]), which writes position state
26//!   to the backing database, defaulting to open positions.
27
28use std::{cell::OnceCell, str::FromStr};
29
30use ahash::AHashSet;
31use bytes::Bytes;
32use nautilus_core::{UUID4, UnixNanos};
33use nautilus_model::{
34    identifiers::{AccountId, InstrumentId, PositionId},
35    position::Position,
36    types::Money,
37};
38
39use super::Cache;
40
41/// Cache-owned reference to a snapshot blob.
42///
43/// The cache writes and later fetches the blob; external systems persist this opaque reference
44/// and may hash the bytes before recording a durable anchor.
45#[derive(Clone, Debug, PartialEq, Eq)]
46pub struct CacheSnapshotRef {
47    /// Opaque cache-owned snapshot location.
48    pub blob_ref: String,
49    /// Snapshot bytes stored under [`Self::blob_ref`].
50    pub blob: Bytes,
51}
52
53impl CacheSnapshotRef {
54    /// Creates a new [`CacheSnapshotRef`].
55    #[must_use]
56    pub fn new(blob_ref: impl Into<String>, blob: impl Into<Bytes>) -> Self {
57        Self {
58            blob_ref: blob_ref.into(),
59            blob: blob.into(),
60        }
61    }
62}
63
64/// One frame in a position's NETTING snapshot history.
65///
66/// A frame keeps the archived position and encodes it only when a consumer asks for the bytes,
67/// so a run with no durable snapshot sink never pays the encode on the order path. A frame
68/// restored from durable bytes keeps those exact bytes, since anchors record their content hash.
69#[derive(Debug)]
70pub(super) struct PositionSnapshotFrame {
71    position: Position,
72    encoded: OnceCell<Bytes>,
73}
74
75impl PositionSnapshotFrame {
76    fn new(position: Position, encoded: Option<Bytes>) -> Self {
77        Self {
78            position,
79            encoded: encoded.map_or_else(OnceCell::new, OnceCell::from),
80        }
81    }
82
83    fn encoded(&self) -> anyhow::Result<Bytes> {
84        if let Some(encoded) = self.encoded.get() {
85            return Ok(encoded.clone());
86        }
87
88        let encoded = Bytes::from(serde_json::to_vec(&self.position)?);
89        let _ = self.encoded.set(encoded.clone());
90
91        Ok(encoded)
92    }
93}
94
95impl Cache {
96    /// Creates a snapshot of the `position` by cloning it, assigning a new ID, and storing it
97    /// in the position snapshots.
98    ///
99    /// The copy excludes `replay_events` and `fill_voids`, which no snapshot consumer reads,
100    /// so snapshot size stays independent of the fills applied to the position ID. The copy
101    /// encodes only when a consumer asks for the bytes, so this call stays off the encode path
102    /// unless a backing database has to persist the frame.
103    ///
104    /// # Errors
105    ///
106    /// Returns an error if serializing or storing the position snapshot fails.
107    pub fn snapshot_position(&mut self, position: &Position) -> anyhow::Result<()> {
108        let (blob_ref, snapshot) = self.build_position_snapshot(position);
109
110        let encoded = if self.database.is_some() {
111            Some(self.persist_position_snapshot(&blob_ref, &snapshot)?)
112        } else {
113            None
114        };
115        self.store_position_snapshot(position.id, snapshot, encoded);
116
117        Ok(())
118    }
119
120    /// Creates a snapshot of the `position` and returns its encoded cache-owned reference.
121    ///
122    /// Behaves as [`Self::snapshot_position`] but encodes the frame eagerly, for callers that
123    /// record the bytes or their content hash against a durable anchor.
124    ///
125    /// # Errors
126    ///
127    /// Returns an error if serializing or storing the position snapshot fails.
128    pub fn snapshot_position_encoded(
129        &mut self,
130        position: &Position,
131    ) -> anyhow::Result<CacheSnapshotRef> {
132        let (blob_ref, snapshot) = self.build_position_snapshot(position);
133        let encoded = self.persist_position_snapshot(&blob_ref, &snapshot)?;
134
135        self.store_position_snapshot(position.id, snapshot, Some(encoded.clone()));
136
137        Ok(CacheSnapshotRef::new(blob_ref, encoded))
138    }
139
140    /// Replaces every NETTING archive frame held for `position` with the cycles a correction
141    /// rebuilt, worth `closed_cycles_pnl`.
142    ///
143    /// A correction that reaches an earlier cycle moves the boundaries the existing frames
144    /// describe, so they cannot be reconciled and are settled into one frame instead. Pass
145    /// `None` when the corrected history never goes flat, which leaves no archived cycle at all.
146    /// As with [`Self::purge_position`], the durable `cache://position-snapshots/...` entries
147    /// stay in general cache state.
148    ///
149    /// Requires `closed_cycles_pnl` to account for every frame held, since this removes all of
150    /// them. The position's replay log must therefore span every archived cycle for the ID. Any
151    /// future retention cap on the log has to preserve that at trim time, either by folding the
152    /// trimmed cycles' realized PnL into a baseline the rebuild adds to its banked total, or by
153    /// purging the frames those cycles produced in the same operation. Settling cannot detect the
154    /// shortfall, because frames carry no cycle identity to match against the retained log.
155    ///
156    /// Known limitation, shared with [`Self::purge_position`]: frame indices restart from zero,
157    /// so a later cycle can overwrite bytes an event store anchor already recorded, failing its
158    /// content-hash check on restore. Frames need an identity independent of their vector
159    /// position to fix it.
160    pub fn settle_position_snapshots(
161        &mut self,
162        position: &Position,
163        closed_cycles_pnl: Option<Money>,
164    ) {
165        self.position_snapshots.remove(&position.id);
166        self.bump_position_snapshot_revision(position.id);
167
168        if let Some(closed_cycles_pnl) = closed_cycles_pnl {
169            let (_, mut settled) = self.build_position_snapshot(position);
170            settled.realized_pnl = Some(closed_cycles_pnl);
171            self.store_position_snapshot(position.id, settled, None);
172        }
173    }
174
175    /// Records that the frames held for `position_id` were replaced rather than appended to.
176    ///
177    /// Consumers cache per-position aggregates keyed off the frame count, which settling and
178    /// purging can leave unchanged while the frames behind it differ.
179    pub(super) fn bump_position_snapshot_revision(&mut self, position_id: PositionId) {
180        *self
181            .position_snapshot_revisions
182            .entry(position_id)
183            .or_default() += 1;
184    }
185
186    fn build_position_snapshot(&self, position: &Position) -> (String, Position) {
187        let position_id = position.id;
188
189        let mut copied_position = position.clone();
190        let new_id = format!("{}-{}", position_id.as_str(), UUID4::new());
191        copied_position.id = PositionId::new(new_id);
192        copied_position.replay_events.clear();
193        copied_position.fill_voids.clear();
194
195        let blob_ref = format!(
196            "cache://position-snapshots/{}/{}",
197            position_id.as_str(),
198            self.position_snapshot_count(&position_id),
199        );
200
201        (blob_ref, copied_position)
202    }
203
204    fn persist_position_snapshot(
205        &mut self,
206        blob_ref: &str,
207        snapshot: &Position,
208    ) -> anyhow::Result<Bytes> {
209        let encoded = Bytes::from(serde_json::to_vec(snapshot)?);
210        self.add(blob_ref, encoded.clone())?;
211
212        Ok(encoded)
213    }
214
215    /// Stores the frame after any persist step, so a failed write does not advance the count.
216    fn store_position_snapshot(
217        &mut self,
218        position_id: PositionId,
219        snapshot: Position,
220        encoded: Option<Bytes>,
221    ) {
222        log::debug!("Snapshot {snapshot}");
223
224        self.position_snapshots
225            .entry(position_id)
226            .or_default()
227            .push(PositionSnapshotFrame::new(snapshot, encoded));
228    }
229
230    fn position_snapshot_frame(&self, blob_ref: &str) -> Option<&PositionSnapshotFrame> {
231        let (position_id, snapshot_index) = parse_position_snapshot_blob_ref(blob_ref).ok()?;
232
233        self.position_snapshots
234            .get(&position_id)
235            .and_then(|frames| frames.get(snapshot_index))
236    }
237
238    /// Loads the cache-owned snapshot blob stored under `blob_ref`.
239    ///
240    /// The cache first checks in-memory snapshot state. When the blob is not present and a
241    /// database adapter exists, the generic cache entries are loaded and checked for the same
242    /// opaque reference.
243    ///
244    /// # Errors
245    ///
246    /// Returns an error if loading generic cache entries from the backing database fails.
247    pub fn load_snapshot_blob(&mut self, blob_ref: &str) -> anyhow::Result<Option<Bytes>> {
248        if let Some(blob) = self.snapshot_blob(blob_ref) {
249            return Ok(Some(blob));
250        }
251
252        if self.database.is_some() {
253            self.cache_general()?;
254        }
255
256        Ok(self.snapshot_blob(blob_ref))
257    }
258
259    /// Restores the cache-owned snapshot blob stored under `blob_ref`.
260    ///
261    /// Only cache-owned `cache://position-snapshots/...` blobs are currently supported.
262    ///
263    /// # Errors
264    ///
265    /// Returns an error if the blob reference is unsupported, malformed, skips earlier
266    /// snapshot frames, conflicts with an existing frame, or does not decode to the expected
267    /// position snapshot.
268    pub fn restore_snapshot_blob(&mut self, blob_ref: &str, blob: Bytes) -> anyhow::Result<()> {
269        let (position_id, snapshot_index) = parse_position_snapshot_blob_ref(blob_ref)?;
270        let restored = decode_position_snapshot_blob(&position_id, blob.as_ref())?;
271
272        let frames = self.position_snapshots.entry(position_id).or_default();
273        match frames.get(snapshot_index) {
274            Some(existing) if existing.encoded()? == blob => {}
275            Some(_) => {
276                anyhow::bail!(
277                    "position snapshot frame {snapshot_index} for {position_id} already exists with different bytes"
278                );
279            }
280            None if frames.len() == snapshot_index => {
281                frames.push(PositionSnapshotFrame::new(restored, Some(blob.clone())));
282            }
283            None => {
284                anyhow::bail!(
285                    "position snapshot blob_ref {blob_ref} skips missing frame {}",
286                    frames.len()
287                );
288            }
289        }
290
291        self.general.insert(blob_ref.to_string(), blob);
292        Ok(())
293    }
294
295    fn snapshot_blob(&self, blob_ref: &str) -> Option<Bytes> {
296        if let Some(blob) = self.general.get(blob_ref) {
297            return Some(blob.clone());
298        }
299
300        self.position_snapshot_frame(blob_ref)?
301            .encoded()
302            .inspect_err(|e| log::warn!("Failed to encode position snapshot {blob_ref}: {e}"))
303            .ok()
304    }
305
306    /// Creates a snapshot of the `position` state in the database.
307    ///
308    /// # Errors
309    ///
310    /// Returns an error if snapshotting the position state fails.
311    pub fn snapshot_position_state(
312        &mut self,
313        position: &Position,
314        ts_snapshot: UnixNanos,
315        unrealized_pnl: Option<Money>,
316        open_only: Option<bool>,
317    ) -> anyhow::Result<()> {
318        let open_only = open_only.unwrap_or(true);
319
320        if open_only && !position.is_open() {
321            return Ok(());
322        }
323
324        if let Some(database) = &mut self.database {
325            database
326                .snapshot_position_state(position, ts_snapshot, unrealized_pnl)
327                .map_err(|e| {
328                    log::error!(
329                        "Failed to snapshot position state for {}: {e:?}",
330                        position.id
331                    );
332                    e
333                })?;
334        } else {
335            log::warn!(
336                "Cannot snapshot position state for {} (no database configured)",
337                position.id
338            );
339        }
340
341        Ok(())
342    }
343
344    /// Gets the serialized position snapshot frames for the `position_id`.
345    ///
346    /// Each element in the returned vector is one JSON-encoded [`Position`] snapshot,
347    /// in the order they were taken. Frames that fail to serialize are skipped with a warning.
348    #[must_use]
349    pub fn position_snapshot_bytes(&self, position_id: &PositionId) -> Option<Vec<Vec<u8>>> {
350        self.position_snapshots.get(position_id).map(|frames| {
351            frames
352                .iter()
353                .filter_map(|frame| match frame.encoded() {
354                    Ok(encoded) => Some(encoded.to_vec()),
355                    Err(e) => {
356                        log::warn!("Failed to encode position snapshot: {e}");
357                        None
358                    }
359                })
360                .collect()
361        })
362    }
363
364    /// Returns the number of stored snapshot frames for the `position_id`.
365    ///
366    /// Returns `0` when no frames are stored. Does not allocate or copy frame bytes.
367    #[must_use]
368    pub fn position_snapshot_count(&self, position_id: &PositionId) -> usize {
369        self.position_snapshots.get(position_id).map_or(0, Vec::len)
370    }
371
372    /// Returns how many times the frames stored for the `position_id` were replaced.
373    ///
374    /// Pair this with [`Self::position_snapshot_count`] to detect frame changes: settling or
375    /// purging can replace the frames without moving the count, so the count alone is not
376    /// enough to tell whether cached per-position aggregates are still current.
377    #[must_use]
378    pub fn position_snapshot_revision(&self, position_id: &PositionId) -> u64 {
379        self.position_snapshot_revisions
380            .get(position_id)
381            .copied()
382            .unwrap_or(0)
383    }
384
385    /// Returns all position snapshots with the given optional filters.
386    ///
387    /// When `position_id` is `Some`, only snapshots for that position are returned.
388    /// When `account_id` is `Some`, snapshots are filtered to that account.
389    #[must_use]
390    pub fn position_snapshots(
391        &self,
392        position_id: Option<&PositionId>,
393        account_id: Option<&AccountId>,
394    ) -> Vec<Position> {
395        let frames: Box<dyn Iterator<Item = &PositionSnapshotFrame> + '_> = match position_id {
396            Some(pid) => match self.position_snapshots.get(pid) {
397                Some(v) => Box::new(v.iter()),
398                None => Box::new(std::iter::empty()),
399            },
400            None => Box::new(self.position_snapshots.values().flat_map(|v| v.iter())),
401        };
402
403        let mut results: Vec<Position> = frames.map(|frame| frame.position.clone()).collect();
404
405        if let Some(aid) = account_id {
406            results.retain(|p| p.account_id == *aid);
407        }
408
409        results
410    }
411
412    /// Returns position snapshots for `position_id` starting from the `skip`th frame.
413    ///
414    /// Use this to read only newly appended snapshots when the caller already processed
415    /// earlier frames. Returns an empty vector when at most `skip` frames are stored.
416    #[must_use]
417    pub fn position_snapshots_from(&self, position_id: &PositionId, skip: usize) -> Vec<Position> {
418        let Some(frames) = self.position_snapshots.get(position_id) else {
419            return Vec::new();
420        };
421
422        frames
423            .iter()
424            .skip(skip)
425            .map(|frame| frame.position.clone())
426            .collect()
427    }
428
429    /// Gets position snapshot IDs for the `instrument_id`.
430    #[must_use]
431    pub fn position_snapshot_ids(&self, instrument_id: &InstrumentId) -> AHashSet<PositionId> {
432        self.position_snapshots
433            .keys()
434            .filter(|position_id| {
435                self.positions
436                    .get(position_id)
437                    .is_some_and(|position| position.borrow().instrument_id == *instrument_id)
438            })
439            .copied()
440            .collect()
441    }
442}
443
444fn parse_position_snapshot_blob_ref(blob_ref: &str) -> anyhow::Result<(PositionId, usize)> {
445    let Some(rest) = blob_ref.strip_prefix("cache://position-snapshots/") else {
446        anyhow::bail!("unsupported cache snapshot blob_ref {blob_ref}");
447    };
448
449    let Some((position_id, snapshot_index)) = rest.rsplit_once('/') else {
450        anyhow::bail!("malformed position snapshot blob_ref {blob_ref}");
451    };
452
453    if position_id.is_empty() {
454        anyhow::bail!("position snapshot blob_ref {blob_ref} has empty position id");
455    }
456
457    let snapshot_index = snapshot_index.parse::<usize>().map_err(|e| {
458        anyhow::anyhow!("position snapshot blob_ref {blob_ref} has invalid frame index: {e}")
459    })?;
460
461    Ok((PositionId::new(position_id), snapshot_index))
462}
463
464fn decode_position_snapshot_blob(
465    position_id: &PositionId,
466    blob: &[u8],
467) -> anyhow::Result<Position> {
468    let snapshot = serde_json::from_slice::<Position>(blob)?;
469    let expected_prefix = format!("{}-", position_id.as_str());
470
471    let Some(snapshot_uuid) = snapshot.id.as_str().strip_prefix(&expected_prefix) else {
472        anyhow::bail!(
473            "position snapshot id {} does not match blob_ref position {position_id}",
474            snapshot.id
475        );
476    };
477
478    if UUID4::from_str(snapshot_uuid).is_err() {
479        anyhow::bail!(
480            "position snapshot id {} does not match blob_ref position {position_id}",
481            snapshot.id
482        );
483    }
484
485    Ok(snapshot)
486}
nautilus_common/cache/position.rs

nautilus_common/cache/
position.rs
