nautilus_event_store/
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//! Snapshot anchors recorded by the event store.
17//!
18//! Cache snapshots remain owned by the cache backing store. The event store records only
19//! the durable log high-watermark the snapshot covers plus enough metadata for restore to
20//! fetch the blob and validate its content hash before tail replay resumes at
21//! `seq > high_watermark`.
22
23use serde::{Deserialize, Serialize};
24
25use crate::error::EventStoreError;
26
27/// A pointer from a cache snapshot blob to a durable event-store high-watermark.
28///
29/// `blob_ref` and `content_hash` are intentionally opaque strings. The cache owns the
30/// storage backend and hash algorithm; the event store only persists the metadata needed
31/// to find the blob and prove the fetched bytes match the snapshot that was anchored.
32#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
33pub struct SnapshotAnchor {
34    /// Largest event-store `seq` covered by the snapshot.
35    pub high_watermark: u64,
36    /// Cache-owned reference to the snapshot blob.
37    pub blob_ref: String,
38    /// Cache-owned content hash for the snapshot blob.
39    pub content_hash: String,
40}
41
42impl SnapshotAnchor {
43    /// Creates a new [`SnapshotAnchor`].
44    #[must_use]
45    pub fn new(
46        high_watermark: u64,
47        blob_ref: impl Into<String>,
48        content_hash: impl Into<String>,
49    ) -> Self {
50        Self {
51            high_watermark,
52            blob_ref: blob_ref.into(),
53            content_hash: content_hash.into(),
54        }
55    }
56}
57
58/// Computes the default cache snapshot content hash recorded in a [`SnapshotAnchor`].
59#[must_use]
60pub fn compute_snapshot_content_hash(bytes: &[u8]) -> String {
61    format!("blake3:{}", blake3::hash(bytes).to_hex())
62}
63
64pub(crate) fn validate_new_anchor(
65    anchor: &SnapshotAnchor,
66    durable_high_watermark: u64,
67    latest: Option<&SnapshotAnchor>,
68) -> Result<(), EventStoreError> {
69    if anchor.high_watermark > durable_high_watermark {
70        return Err(EventStoreError::Backend(format!(
71            "snapshot anchor high_watermark {} exceeds durable high_watermark {}",
72            anchor.high_watermark, durable_high_watermark,
73        )));
74    }
75
76    if let Some(latest) = latest
77        && anchor.high_watermark < latest.high_watermark
78    {
79        return Err(EventStoreError::Backend(format!(
80            "snapshot anchor high_watermark {} is older than latest anchor {}",
81            anchor.high_watermark, latest.high_watermark,
82        )));
83    }
84
85    Ok(())
86}
87
88#[cfg(test)]
89mod tests {
90    use rstest::rstest;
91
92    use super::*;
93
94    #[rstest]
95    fn anchor_new_sets_all_fields() {
96        let anchor = SnapshotAnchor::new(7, "cache://snapshots/run-1/7", "blake3:abc");
97
98        assert_eq!(anchor.high_watermark, 7);
99        assert_eq!(anchor.blob_ref, "cache://snapshots/run-1/7");
100        assert_eq!(anchor.content_hash, "blake3:abc");
101    }
102
103    #[rstest]
104    fn compute_snapshot_content_hash_prefixes_blake3_digest() {
105        assert_eq!(
106            compute_snapshot_content_hash(b"snapshot"),
107            format!("blake3:{}", blake3::hash(b"snapshot").to_hex()),
108        );
109    }
110
111    #[rstest]
112    fn validate_rejects_anchor_past_durable_watermark() {
113        let anchor = SnapshotAnchor::new(8, "blob", "hash");
114        let err = validate_new_anchor(&anchor, 7, None).expect_err("must reject");
115
116        match err {
117            EventStoreError::Backend(msg) => {
118                assert!(
119                    msg.contains("exceeds durable high_watermark"),
120                    "msg was: {msg}",
121                );
122            }
123            other => panic!("expected Backend, was {other:?}"),
124        }
125    }
126
127    #[rstest]
128    fn validate_rejects_anchor_older_than_latest() {
129        let latest = SnapshotAnchor::new(9, "latest", "hash-latest");
130        let anchor = SnapshotAnchor::new(8, "older", "hash-older");
131        let err = validate_new_anchor(&anchor, 10, Some(&latest)).expect_err("must reject");
132
133        match err {
134            EventStoreError::Backend(msg) => {
135                assert!(msg.contains("older than latest anchor"), "msg was: {msg}");
136            }
137            other => panic!("expected Backend, was {other:?}"),
138        }
139    }
140
141    #[rstest]
142    fn validate_accepts_equal_or_newer_anchor() {
143        let latest = SnapshotAnchor::new(9, "latest", "hash-latest");
144        let same = SnapshotAnchor::new(9, "same", "hash-same");
145        let newer = SnapshotAnchor::new(10, "newer", "hash-newer");
146
147        validate_new_anchor(&same, 10, Some(&latest)).expect("same hwm accepted");
148        validate_new_anchor(&newer, 10, Some(&latest)).expect("newer hwm accepted");
149    }
150}
nautilus_event_store/snapshot.rs

nautilus_event_store/
snapshot.rs
