truefix_store/lib.rs
1//! `truefix-store` — pluggable persistence of sequence numbers and sent messages.
2//!
3//! The [`MessageStore`] trait abstracts the storage needed for resend (FR-G2): the next inbound
4//! and outbound sequence numbers, and the bytes of each sent message keyed by sequence number.
5//! Implementations: [`MemoryStore`], [`FileStore`], [`CachedFileStore`], [`NoopStore`], and a
6//! SQL-backed store behind the `sql` feature.
7//!
8//! Audit 006 additions: [`MessageStore::contains`] exposes duplicate-sequence detection without
9//! changing `save()`'s existing overwrite-on-conflict contract (NEW-137);
10//! [`FileStoreOptions::max_body_records`] bounds a file-backed store's body-log index while
11//! preserving resend/recovery semantics (NEW-108); Mongo/SQL/MSSQL `reset()` is atomic across
12//! messages and sequence numbers (NEW-116); and `NoopStore` tracks sequence numbers in memory for
13//! the session lifetime (NEW-118).
14//!
15//! Feature 012 (audit 007) additions / **migration note**: [`StoreConfig`] gained a `Custom(Arc<dyn
16//! MessageStore>)` variant (NEW-158), letting a caller supply a backend not on the built-in list
17//! through the same config surface `Engine::start` resolves (see `truefix::Engine::start_with_overrides`).
18//! This is additive to the variant set, but `StoreConfig` can no longer derive `Debug` (`dyn
19//! MessageStore` isn't `Debug`) -- it now has a hand-written `Debug` impl with the same output for
20//! every pre-existing variant, so this only breaks code that exhaustively `match`es `StoreConfig`
21//! without a `_ =>` arm.
22//!
23//! Design: `specs/001-fix-engine-parity/`, `specs/012-audit-007-remediation/`.
24#![cfg_attr(
25 not(test),
26 deny(
27 clippy::unwrap_used,
28 clippy::expect_used,
29 clippy::panic,
30 clippy::indexing_slicing
31 )
32)]
33
34mod file;
35mod memory;
36#[cfg(feature = "mongodb")]
37mod mongo;
38#[cfg(feature = "mssql")]
39mod mssql;
40mod noop;
41#[cfg(feature = "redb")]
42mod redb;
43#[cfg(feature = "sql")]
44mod sql;
45
46use std::path::PathBuf;
47use std::sync::Arc;
48
49use async_trait::async_trait;
50use thiserror::Error;
51
52pub use file::{CachedFileStore, FileStore, FileStoreOptions};
53pub use memory::MemoryStore;
54#[cfg(feature = "mongodb")]
55pub use mongo::{MongoStore, MongoStoreConfig};
56#[cfg(feature = "mssql")]
57pub use mssql::{MssqlStore, MssqlStoreConfig};
58pub use noop::NoopStore;
59#[cfg(feature = "redb")]
60pub use redb::{RedbStore, RedbStoreConfig};
61#[cfg(feature = "sql")]
62pub use sql::{SqlPoolOptions, SqlStore, SqlStoreConfig};
63
64/// An error from a message store.
65#[derive(Debug, Error)]
66pub enum StoreError {
67 /// An underlying I/O error.
68 #[error("store I/O error: {0}")]
69 Io(String),
70 /// A backend (e.g. SQL) error.
71 #[error("store backend error: {0}")]
72 Backend(String),
73}
74
75/// Persistence for sequence numbers and outbound messages, sufficient to satisfy resend
76/// across restarts (FR-G2, SC-006).
77#[async_trait]
78pub trait MessageStore: Send + Sync {
79 /// The next outbound (sender) sequence number.
80 async fn next_sender_seq(&self) -> Result<u64, StoreError>;
81 /// The next inbound (target) sequence number expected.
82 async fn next_target_seq(&self) -> Result<u64, StoreError>;
83 /// Set the next outbound sequence number.
84 async fn set_next_sender_seq(&self, seq: u64) -> Result<(), StoreError>;
85 /// Set the next inbound sequence number.
86 async fn set_next_target_seq(&self, seq: u64) -> Result<(), StoreError>;
87 /// Persist an outbound message's bytes under its sequence number.
88 async fn save(&self, seq: u64, message: &[u8]) -> Result<(), StoreError>;
89 /// Retrieve stored messages whose sequence numbers fall in `[begin, end]`, in order.
90 async fn get(&self, begin: u64, end: u64) -> Result<Vec<(u64, Vec<u8>)>, StoreError>;
91 /// Reset: clear stored messages and set both sequence numbers back to 1.
92 async fn reset(&self) -> Result<(), StoreError>;
93 /// Whether this store detected and recovered from corruption when it was opened
94 /// (`ForceResendWhenCorruptedStore`). Backends with no corruption-detection concept (e.g.
95 /// `MemoryStore`, `NoopStore`, SQL backends relying on the database's own durability) default
96 /// to `false`.
97 fn was_corrupted(&self) -> bool {
98 false
99 }
100 /// The wall-clock time this store was created (or last reset), if the backend tracks one
101 /// (GAP-38/FR-017, feature 005). `None` for backends with no such concept (e.g.
102 /// `MemoryStore`, `NoopStore`) or when not yet recorded.
103 async fn creation_time(&self) -> Result<Option<time::OffsetDateTime>, StoreError> {
104 Ok(None)
105 }
106 /// Atomically persist an outbound message and advance the sender sequence past it
107 /// (GAP-39/FR-018, feature 005): closes the crash window where `save()` succeeds but a
108 /// subsequent `set_next_sender_seq()` doesn't (or vice versa), which could otherwise
109 /// double-send or skip a sequence number on restart. The default implementation preserves
110 /// today's behavior (two independent calls) for any backend that doesn't override it.
111 async fn save_and_advance_sender(&self, seq: u64, message: &[u8]) -> Result<(), StoreError> {
112 self.save(seq, message).await?;
113 self.set_next_sender_seq(seq + 1).await
114 }
115 /// Whether a message is already saved under `seq` (NEW-137, audit 006): a way for callers to
116 /// observe duplicate sequence usage without changing `save()`'s existing overwrite-on-conflict
117 /// contract (deliberately kept separate rather than having `save()` itself return an
118 /// overwrote-something flag, to avoid a breaking signature change across every backend).
119 /// Default `Ok(false)` for any backend that doesn't override it.
120 async fn contains(&self, seq: u64) -> Result<bool, StoreError> {
121 let _ = seq;
122 Ok(false)
123 }
124}
125
126/// NEW-158 (feature 012): lets a `StoreConfig::Custom(Arc<dyn MessageStore>)` be handed to
127/// `build_store` and used exactly like any built-in backend, mirroring `truefix_log`'s
128/// `impl Log for Box<dyn Log>`.
129#[async_trait]
130impl MessageStore for Arc<dyn MessageStore> {
131 async fn next_sender_seq(&self) -> Result<u64, StoreError> {
132 (**self).next_sender_seq().await
133 }
134 async fn next_target_seq(&self) -> Result<u64, StoreError> {
135 (**self).next_target_seq().await
136 }
137 async fn set_next_sender_seq(&self, seq: u64) -> Result<(), StoreError> {
138 (**self).set_next_sender_seq(seq).await
139 }
140 async fn set_next_target_seq(&self, seq: u64) -> Result<(), StoreError> {
141 (**self).set_next_target_seq(seq).await
142 }
143 async fn save(&self, seq: u64, message: &[u8]) -> Result<(), StoreError> {
144 (**self).save(seq, message).await
145 }
146 async fn get(&self, begin: u64, end: u64) -> Result<Vec<(u64, Vec<u8>)>, StoreError> {
147 (**self).get(begin, end).await
148 }
149 async fn reset(&self) -> Result<(), StoreError> {
150 (**self).reset().await
151 }
152 fn was_corrupted(&self) -> bool {
153 (**self).was_corrupted()
154 }
155 async fn creation_time(&self) -> Result<Option<time::OffsetDateTime>, StoreError> {
156 (**self).creation_time().await
157 }
158 async fn save_and_advance_sender(&self, seq: u64, message: &[u8]) -> Result<(), StoreError> {
159 (**self).save_and_advance_sender(seq, message).await
160 }
161 async fn contains(&self, seq: u64) -> Result<bool, StoreError> {
162 (**self).contains(seq).await
163 }
164}
165
166/// Which store backend to construct.
167#[derive(Clone)]
168pub enum StoreConfig {
169 /// Volatile in-memory store.
170 Memory,
171 /// File-backed store in a directory.
172 File {
173 /// Directory holding the store files.
174 dir: PathBuf,
175 /// `FileStoreSync` (FR-025).
176 options: FileStoreOptions,
177 },
178 /// File-backed store that also caches messages in memory.
179 CachedFile {
180 /// Directory holding the store files.
181 dir: PathBuf,
182 /// `FileStoreSync`/`FileStoreMaxCachedMsgs` (FR-025).
183 options: FileStoreOptions,
184 },
185 /// No-op store (never persists; sequence numbers stay at 1).
186 Noop,
187 /// SQL-backed store (PostgreSQL/MySQL/SQLite; requires the `sql` feature).
188 #[cfg(feature = "sql")]
189 Sql {
190 /// Database URL.
191 url: String,
192 /// `JdbcStoreSessionsTableName` (US8, feature 005, FR-021); `None` uses
193 /// `SqlStoreConfig::new`'s default (`"seqnums"`).
194 sessions_table: Option<String>,
195 /// `JdbcStoreMessagesTableName` (US8, feature 005, FR-021); `None` uses
196 /// `SqlStoreConfig::new`'s default (`"messages"`).
197 messages_table: Option<String>,
198 /// `JdbcSessionIdDefaultPropertyValue` (US8, feature 005, FR-021); `None` uses
199 /// `SqlStoreConfig::new`'s default (`"default"`).
200 session_id: Option<String>,
201 /// `Jdbc*Connection*` pool-tuning keys (US8, feature 005, FR-021); `None` uses
202 /// `SqlPoolOptions::default()`.
203 pool: Option<SqlPoolOptions>,
204 },
205 /// MSSQL-backed store (requires the `mssql` feature; FR-020).
206 #[cfg(feature = "mssql")]
207 Mssql {
208 /// Database URL (`mssql://user:password@host[:port]/database`).
209 url: String,
210 /// `JdbcStoreSessionsTableName` (US8, feature 005, FR-021); `None` uses
211 /// `MssqlStoreConfig::new`'s default (`"seqnums"`).
212 sessions_table: Option<String>,
213 /// `JdbcStoreMessagesTableName` (US8, feature 005, FR-021); `None` uses
214 /// `MssqlStoreConfig::new`'s default (`"messages"`).
215 messages_table: Option<String>,
216 /// `JdbcSessionIdDefaultPropertyValue` (US8, feature 005, FR-021); `None` uses
217 /// `MssqlStoreConfig::new`'s default (`"default"`).
218 session_id: Option<String>,
219 },
220 /// Embedded transactional store via `redb` (requires the `redb` feature; US5, feature 004,
221 /// FR-006), a modern replacement for QuickFIX/J's obsolete `SleepycatStore`.
222 #[cfg(feature = "redb")]
223 Redb {
224 /// Path to the `redb` database file.
225 path: PathBuf,
226 },
227 /// MongoDB-backed store (requires the `mongodb` feature; US6, feature 004), matching
228 /// QuickFIX/Go's MongoDB option.
229 #[cfg(feature = "mongodb")]
230 Mongo {
231 /// MongoDB connection URI (`mongodb://...`).
232 uri: String,
233 },
234 /// NEW-158 (feature 012): a caller-supplied `MessageStore` implementation, for backends not
235 /// on the built-in list (or a proprietary audit-log sink) -- previously only reachable via
236 /// the lower-level `truefix::transport::Services::store` escape hatch; this widens the
237 /// `.cfg`-adjacent config surface itself to express the same thing (FR-006). `Arc` (not
238 /// `Box`) so `StoreConfig` keeps deriving `Clone`.
239 Custom(Arc<dyn MessageStore>),
240}
241
242impl std::fmt::Debug for StoreConfig {
243 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
244 match self {
245 Self::Memory => write!(f, "Memory"),
246 Self::File { dir, options } => f
247 .debug_struct("File")
248 .field("dir", dir)
249 .field("options", options)
250 .finish(),
251 Self::CachedFile { dir, options } => f
252 .debug_struct("CachedFile")
253 .field("dir", dir)
254 .field("options", options)
255 .finish(),
256 Self::Noop => write!(f, "Noop"),
257 #[cfg(feature = "sql")]
258 Self::Sql { url, .. } => f.debug_struct("Sql").field("url", url).finish(),
259 #[cfg(feature = "mssql")]
260 Self::Mssql { url, .. } => f.debug_struct("Mssql").field("url", url).finish(),
261 #[cfg(feature = "redb")]
262 Self::Redb { path } => f.debug_struct("Redb").field("path", path).finish(),
263 #[cfg(feature = "mongodb")]
264 Self::Mongo { uri } => f.debug_struct("Mongo").field("uri", uri).finish(),
265 // `dyn MessageStore` isn't `Debug`; this variant only ever comes from programmatic
266 // construction, so a placeholder is sufficient for diagnostics.
267 Self::Custom(_) => f
268 .debug_tuple("Custom")
269 .field(&"<dyn MessageStore>")
270 .finish(),
271 }
272 }
273}
274
275/// Build a boxed [`MessageStore`] from a [`StoreConfig`].
276pub async fn build_store(config: &StoreConfig) -> Result<Box<dyn MessageStore>, StoreError> {
277 Ok(match config {
278 StoreConfig::Memory => Box::new(MemoryStore::new()),
279 StoreConfig::File { dir, options } => {
280 Box::new(FileStore::open_with_options(dir, *options)?)
281 }
282 StoreConfig::CachedFile { dir, options } => {
283 Box::new(CachedFileStore::open_with_options(dir, *options)?)
284 }
285 StoreConfig::Noop => Box::new(NoopStore::new()),
286 #[cfg(feature = "sql")]
287 StoreConfig::Sql {
288 url,
289 sessions_table,
290 messages_table,
291 session_id,
292 pool,
293 } => {
294 let defaults = SqlStoreConfig::new(url);
295 Box::new(
296 SqlStore::connect_with_config(SqlStoreConfig {
297 sessions_table: sessions_table.clone().unwrap_or(defaults.sessions_table),
298 messages_table: messages_table.clone().unwrap_or(defaults.messages_table),
299 session_id: session_id.clone().unwrap_or(defaults.session_id),
300 pool: pool.unwrap_or(defaults.pool),
301 ..defaults
302 })
303 .await?,
304 )
305 }
306 #[cfg(feature = "mssql")]
307 StoreConfig::Mssql {
308 url,
309 sessions_table,
310 messages_table,
311 session_id,
312 } => {
313 let defaults = MssqlStoreConfig::new(url);
314 Box::new(
315 MssqlStore::connect_with_config(MssqlStoreConfig {
316 sessions_table: sessions_table.clone().unwrap_or(defaults.sessions_table),
317 messages_table: messages_table.clone().unwrap_or(defaults.messages_table),
318 session_id: session_id.clone().unwrap_or(defaults.session_id),
319 ..defaults
320 })
321 .await?,
322 )
323 }
324 #[cfg(feature = "redb")]
325 StoreConfig::Redb { path } => Box::new(RedbStore::connect(path).await?),
326 #[cfg(feature = "mongodb")]
327 StoreConfig::Mongo { uri } => Box::new(MongoStore::connect(uri).await?),
328 StoreConfig::Custom(store) => Box::new(store.clone()),
329 })
330}