nautilus_event_store/
hash.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//! Canonical entry hashing for the event store.
17//!
18//! Every captured entry carries a 32-byte BLAKE3 hash computed from a domain-separated,
19//! length-prefixed serialization of `(seq, ts_init, ts_publish, topic, payload_type, payload,
20//! headers)`. The hash is recomputed on every read; a mismatch quarantines the run.
21//!
22//! BLAKE3 is the SPEC default for the integrity-first reading: it is fast enough for the
23//! capture path and outruns xxhash3 on the verifier scan over GiB-scale runs while remaining
24//! a cryptographic hash for tamper detection.
25
26use bytes::Bytes;
27use nautilus_core::{UUID4, UnixNanos};
28use serde::{Deserialize, Serialize};
29
30use crate::headers::Headers;
31
32const HASH_DOMAIN: &[u8] = b"nautilus-event-store/entry/v1";
33
34/// The 32-byte canonical hash of an event store entry.
35#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
36pub struct EntryHash(pub [u8; 32]);
37
38impl EntryHash {
39    /// Returns the hash as a borrowed byte slice.
40    #[must_use]
41    pub const fn as_bytes(&self) -> &[u8; 32] {
42        &self.0
43    }
44
45    /// Returns the hash as a lowercase hexadecimal string.
46    #[must_use]
47    pub fn to_hex(&self) -> String {
48        let mut out = String::with_capacity(64);
49
50        for byte in self.0 {
51            out.push(nibble_to_hex(byte >> 4));
52            out.push(nibble_to_hex(byte & 0x0F));
53        }
54        out
55    }
56}
57
58const fn nibble_to_hex(nibble: u8) -> char {
59    match nibble {
60        0..=9 => (b'0' + nibble) as char,
61        10..=15 => (b'a' + nibble - 10) as char,
62        _ => unreachable!(),
63    }
64}
65
66/// Computes the canonical hash of an event store entry.
67///
68/// The hash is domain-separated by a fixed crate-internal prefix and uses big-endian
69/// fixed-width framing for every variable-length field so the output depends only on the
70/// logical content of the entry, not on the host endianness or the runtime serialization
71/// format.
72#[must_use]
73pub fn compute_entry_hash(
74    seq: u64,
75    ts_init: UnixNanos,
76    ts_publish: UnixNanos,
77    topic: &str,
78    payload_type: &str,
79    payload: &Bytes,
80    headers: &Headers,
81) -> EntryHash {
82    let mut hasher = blake3::Hasher::new();
83    hasher.update(HASH_DOMAIN);
84    hasher.update(&seq.to_be_bytes());
85    hasher.update(&ts_init.as_u64().to_be_bytes());
86    hasher.update(&ts_publish.as_u64().to_be_bytes());
87    write_str(&mut hasher, topic);
88    write_str(&mut hasher, payload_type);
89    write_bytes(&mut hasher, payload);
90    write_optional_uuid(&mut hasher, headers.correlation_id.as_ref());
91    write_optional_uuid(&mut hasher, headers.causation_id.as_ref());
92    EntryHash(*hasher.finalize().as_bytes())
93}
94
95fn write_str(hasher: &mut blake3::Hasher, value: &str) {
96    let bytes = value.as_bytes();
97    hasher.update(&(bytes.len() as u64).to_be_bytes());
98    hasher.update(bytes);
99}
100
101fn write_bytes(hasher: &mut blake3::Hasher, value: &Bytes) {
102    hasher.update(&(value.len() as u64).to_be_bytes());
103    hasher.update(value);
104}
105
106fn write_optional_uuid(hasher: &mut blake3::Hasher, value: Option<&UUID4>) {
107    match value {
108        Some(uuid) => {
109            hasher.update(&[1u8]);
110            hasher.update(&uuid.as_bytes());
111        }
112        None => {
113            hasher.update(&[0u8]);
114        }
115    }
116}
117
118#[cfg(test)]
119mod tests {
120    use rstest::rstest;
121
122    use super::*;
123
124    // Canonical hash inputs used as the baseline across sensitivity tests.
125    #[derive(Clone)]
126    struct HashInput {
127        seq: u64,
128        ts_init: UnixNanos,
129        ts_publish: UnixNanos,
130        topic: String,
131        payload_type: String,
132        payload: Bytes,
133        headers: Headers,
134    }
135
136    fn baseline() -> HashInput {
137        HashInput {
138            seq: 42,
139            ts_init: UnixNanos::from(1_700_000_000_000_000_000),
140            ts_publish: UnixNanos::from(1_700_000_000_000_000_001),
141            topic: "exec.command".to_string(),
142            payload_type: "SubmitOrder".to_string(),
143            payload: Bytes::from_static(b"\x01\x02\x03"),
144            headers: Headers::empty(),
145        }
146    }
147
148    fn hash_of(input: &HashInput) -> EntryHash {
149        compute_entry_hash(
150            input.seq,
151            input.ts_init,
152            input.ts_publish,
153            &input.topic,
154            &input.payload_type,
155            &input.payload,
156            &input.headers,
157        )
158    }
159
160    #[rstest]
161    fn hash_is_deterministic() {
162        let input = baseline();
163
164        assert_eq!(hash_of(&input), hash_of(&input));
165    }
166
167    #[rstest]
168    #[case::seq(|i: &mut HashInput| i.seq = 99)]
169    #[case::ts_init(|i: &mut HashInput| i.ts_init = UnixNanos::from(1))]
170    #[case::ts_publish(|i: &mut HashInput| i.ts_publish = UnixNanos::from(1))]
171    #[case::topic(|i: &mut HashInput| i.topic = "other".to_string())]
172    #[case::payload_type(|i: &mut HashInput| i.payload_type = "Other".to_string())]
173    #[case::payload(|i: &mut HashInput| i.payload = Bytes::from_static(b"\xFF"))]
174    #[case::correlation_id(|i: &mut HashInput| i.headers.correlation_id = Some(UUID4::new()))]
175    #[case::causation_id(|i: &mut HashInput| i.headers.causation_id = Some(UUID4::new()))]
176    fn every_input_field_affects_hash(#[case] mutate: fn(&mut HashInput)) {
177        let input = baseline();
178        let mut mutated = input.clone();
179        mutate(&mut mutated);
180
181        assert_ne!(hash_of(&input), hash_of(&mutated));
182    }
183
184    #[rstest]
185    fn hash_separates_topic_from_payload_type() {
186        // Length-prefixed framing must prevent (topic="ab", payload_type="c") from
187        // colliding with (topic="a", payload_type="bc"); without the prefix, both
188        // would hash the same flattened byte stream.
189        let payload = Bytes::from_static(b"x");
190        let a = compute_entry_hash(
191            1,
192            UnixNanos::from(0),
193            UnixNanos::from(0),
194            "ab",
195            "c",
196            &payload,
197            &Headers::empty(),
198        );
199        let b = compute_entry_hash(
200            1,
201            UnixNanos::from(0),
202            UnixNanos::from(0),
203            "a",
204            "bc",
205            &payload,
206            &Headers::empty(),
207        );
208
209        assert_ne!(a, b);
210    }
211
212    #[rstest]
213    fn compute_entry_hash_known_vector() {
214        // Pinning the BLAKE3 wire format. Any change to the domain prefix, write
215        // order, or endianness of the framing flips this expected value.
216        let input = baseline();
217
218        assert_eq!(
219            hash_of(&input).to_hex(),
220            "06b08d9615241ccdee4c21303e8d5a21682ceb085eb4eaf170365c700836e620",
221        );
222    }
223
224    #[rstest]
225    fn compute_entry_hash_known_vector_populated_headers() {
226        // Pinning the BLAKE3 wire format with distinct correlation_id and causation_id
227        // values. The empty-headers vector above is insensitive to write-order changes
228        // between the two header fields (both write a single 0x00 None marker); this
229        // vector flips if the writer ever swaps the two fields' positions.
230        let correlation = UUID4::from_bytes([
231            0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E,
232            0x0F, 0x10,
233        ]);
234        let causation = UUID4::from_bytes([
235            0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1A, 0x1B, 0x1C, 0x1D, 0x1E,
236            0x1F, 0x20,
237        ]);
238        let mut input = baseline();
239        input.headers = Headers {
240            correlation_id: Some(correlation),
241            causation_id: Some(causation),
242        };
243
244        assert_eq!(
245            hash_of(&input).to_hex(),
246            "69be87a947cbfb61dd445908ae5825ada0b679bb3c9dd7e8cb14dcc0baf74eaa",
247        );
248    }
249
250    #[rstest]
251    fn to_hex_is_lowercase_big_endian_nibbles() {
252        // Pinning EntryHash::to_hex's output: high nibble first, lowercase, fixed
253        // 64 chars. Catches nibble-swap or uppercase regressions that the prior
254        // length-and-charclass check let through.
255        let hash = EntryHash([0xABu8; 32]);
256
257        assert_eq!(hash.to_hex(), "ab".repeat(32));
258    }
259}
nautilus_event_store/hash.rs

nautilus_event_store/
hash.rs
