nautilus_event_store/
entry.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//! The captured event store entry.
17
18use bytes::Bytes;
19use nautilus_common::msgbus::{self, MStr};
20use nautilus_core::UnixNanos;
21use serde::{Deserialize, Serialize};
22use ustr::Ustr;
23
24use crate::{hash::EntryHash, headers::Headers, wire};
25
26/// A bus topic identifier captured for an event store entry.
27///
28/// Reuses `nautilus_common::msgbus::MStr<Topic>` so captured rows carry the same
29/// phantom-typed handle the bus uses; the writer copies the handle rather than the
30/// string and the type system rejects accidentally captured patterns or endpoints.
31pub type Topic = MStr<msgbus::Topic>;
32
33/// A canonical payload type identifier captured for an event store entry.
34///
35/// Identifies which encoder produced [`EventStoreEntry::payload`]. The bus capture adapter
36/// allow-list maps Rust message types to these tags; the reader uses the tag to dispatch
37/// the matching decoder. Unlike [`Topic`] this is not a bus type, so we keep the plain
38/// `Ustr` and avoid taking on the bus's `MStr` discipline for a free-form encoder tag.
39pub type PayloadType = Ustr;
40
41/// One captured row in the event store: a state-affecting bus message plus metadata.
42///
43/// The fields cover the SPEC's logical entry contract:
44///
45/// - `seq` is the per-run monotonic sequence assigned by the writer at commit time. It is
46///   the replay-order authority.
47/// - `ts_init` is the domain timestamp; strictly monotonic and unique system-wide via the
48///   shared `AtomicTime`.
49/// - `ts_publish` records when the writer received the entry (and, when populated by the
50///   bus, when the bus accepted it for fanout). Neither timestamp orders replay.
51/// - `topic` and `payload_type` describe the logical message identity; the writer commits
52///   them verbatim.
53/// - `payload` carries the canonical bytes produced by the registered encoder.
54/// - `headers` carry first-class correlation metadata.
55/// - `entry_hash` is the canonical hash recomputed on every read; mismatch quarantines the
56///   run.
57#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
58pub struct EventStoreEntry {
59    /// The canonical hash over every preceding field.
60    pub entry_hash: EntryHash,
61    /// The per-run monotonic sequence; replay-order authority.
62    pub seq: u64,
63    /// First-class correlation headers.
64    pub headers: Headers,
65    /// The logical bus topic this entry was captured on.
66    pub topic: Topic,
67    /// The canonical payload type tag chosen by the bus capture adapter.
68    pub payload_type: PayloadType,
69    /// The canonical encoded payload bytes.
70    pub payload: Bytes,
71    /// The domain timestamp from `AtomicTime`.
72    #[serde(with = "wire::nanos_as_u64")]
73    pub ts_init: UnixNanos,
74    /// The bus-accepted or writer-receive timestamp.
75    #[serde(with = "wire::nanos_as_u64")]
76    pub ts_publish: UnixNanos,
77}
78
79impl EventStoreEntry {
80    /// Creates a new [`EventStoreEntry`] with all fields supplied by the writer.
81    #[must_use]
82    #[expect(
83        clippy::too_many_arguments,
84        reason = "entry constructor takes the full SPEC envelope"
85    )]
86    pub fn new(
87        entry_hash: EntryHash,
88        seq: u64,
89        headers: Headers,
90        topic: Topic,
91        payload_type: PayloadType,
92        payload: Bytes,
93        ts_init: UnixNanos,
94        ts_publish: UnixNanos,
95    ) -> Self {
96        Self {
97            entry_hash,
98            seq,
99            headers,
100            topic,
101            payload_type,
102            payload,
103            ts_init,
104            ts_publish,
105        }
106    }
107
108    /// Recomputes the canonical hash for this entry from its current fields.
109    ///
110    /// Backends call this on every read to validate `entry_hash`; mismatch quarantines the
111    /// run.
112    #[must_use]
113    pub fn recompute_hash(&self) -> EntryHash {
114        crate::hash::compute_entry_hash(
115            self.seq,
116            self.ts_init,
117            self.ts_publish,
118            self.topic.as_ref(),
119            self.payload_type.as_str(),
120            &self.payload,
121            &self.headers,
122        )
123    }
124}
125
126#[cfg(test)]
127mod tests {
128    use rstest::rstest;
129
130    use super::*;
131    use crate::hash::compute_entry_hash;
132
133    fn entry() -> EventStoreEntry {
134        let topic: Topic = "exec.command".into();
135        let payload_type = Ustr::from("SubmitOrder");
136        let payload = Bytes::from_static(b"\x01\x02\x03");
137        let headers = Headers::empty();
138        let hash = compute_entry_hash(
139            1,
140            UnixNanos::from(10),
141            UnixNanos::from(11),
142            topic.as_ref(),
143            payload_type.as_str(),
144            &payload,
145            &headers,
146        );
147        EventStoreEntry::new(
148            hash,
149            1,
150            headers,
151            topic,
152            payload_type,
153            payload,
154            UnixNanos::from(10),
155            UnixNanos::from(11),
156        )
157    }
158
159    #[rstest]
160    fn recompute_matches_stored_hash() {
161        let e = entry();
162
163        assert_eq!(e.recompute_hash(), e.entry_hash);
164    }
165
166    #[rstest]
167    fn tampered_payload_breaks_hash_check() {
168        let mut e = entry();
169        e.payload = Bytes::from_static(b"\x01\x02\x04");
170
171        assert_ne!(e.recompute_hash(), e.entry_hash);
172    }
173}
nautilus_event_store/entry.rs

nautilus_event_store/
entry.rs
