nautilus_event_store/writer/
halt.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//! Fail-stop signaling for the event store writer.
17//!
18//! The kernel registers a [`HaltCallback`] when constructing the writer. The writer invokes
19//! the callback once on the first unrecoverable condition: a submit-side backpressure stall
20//! that exceeded the configured threshold, or a backend error that breaks the audit
21//! contract (disk pressure, corruption, surface I/O failure). The callback is the kernel's
22//! signal to fail-stop trading; it is fired exactly once before the writer ceases to accept
23//! further entries.
24
25use std::{sync::Arc, time::Duration};
26
27use crate::error::EventStoreError;
28
29/// Reason a writer requested kernel halt.
30#[derive(Clone, Debug)]
31pub enum HaltReason {
32    /// A submit blocked longer than the configured halt threshold while waiting for the
33    /// writer thread to drain the channel.
34    BackpressureStall {
35        /// How long the submit blocked before signaling halt.
36        stalled_for: Duration,
37        /// The configured threshold the stall exceeded.
38        threshold: Duration,
39    },
40    /// The backend rejected a commit because of disk pressure (ENOSPC, `RLIMIT_FSIZE`,
41    /// quota). The audit contract requires fail-stop rather than dropping entries.
42    BackendDisk(String),
43    /// The backend reported structural corruption.
44    BackendCorrupted(String),
45    /// The backend returned an unclassified error that the writer cannot retry past.
46    BackendError(String),
47}
48
49impl HaltReason {
50    /// Maps a backend [`EventStoreError`] onto the matching halt reason.
51    #[must_use]
52    pub fn from_backend_error(err: &EventStoreError) -> Self {
53        match err {
54            EventStoreError::Disk(msg) => Self::BackendDisk(msg.clone()),
55            EventStoreError::Corrupted(msg) => Self::BackendCorrupted(msg.clone()),
56            other => Self::BackendError(other.to_string()),
57        }
58    }
59}
60
61/// Callback invoked once on the first unrecoverable writer condition.
62///
63/// Cloneable so submit, the writer thread, and tests can share the same fail-stop sink.
64pub type HaltCallback = Arc<dyn Fn(HaltReason) + Send + Sync + 'static>;
65
66/// Returns a [`HaltCallback`] that performs no action.
67///
68/// Useful for tests and for writers operating under simulation where halt is observed
69/// via the test harness rather than the kernel.
70#[must_use]
71pub fn noop_halt() -> HaltCallback {
72    Arc::new(|_reason| ())
73}
74
75#[cfg(test)]
76mod tests {
77    use std::sync::Mutex;
78
79    use rstest::rstest;
80
81    use super::*;
82
83    #[rstest]
84    fn from_backend_error_classifies_disk_as_disk() {
85        let err = EventStoreError::Disk("ENOSPC".to_string());
86        let reason = HaltReason::from_backend_error(&err);
87
88        match reason {
89            HaltReason::BackendDisk(msg) => assert!(msg.contains("ENOSPC"), "msg was: {msg}"),
90            other => panic!("expected BackendDisk, was {other:?}"),
91        }
92    }
93
94    #[rstest]
95    fn from_backend_error_classifies_corrupted_as_corrupted() {
96        let err = EventStoreError::Corrupted("bad page".to_string());
97        let reason = HaltReason::from_backend_error(&err);
98
99        match reason {
100            HaltReason::BackendCorrupted(msg) => {
101                assert!(msg.contains("bad page"), "msg was: {msg}");
102            }
103            other => panic!("expected BackendCorrupted, was {other:?}"),
104        }
105    }
106
107    #[rstest]
108    fn from_backend_error_classifies_other_variants_as_error() {
109        let err = EventStoreError::Closed;
110        let reason = HaltReason::from_backend_error(&err);
111
112        match reason {
113            HaltReason::BackendError(_) => {}
114            other => panic!("expected BackendError, was {other:?}"),
115        }
116    }
117
118    #[rstest]
119    fn noop_halt_does_not_panic() {
120        let halt = noop_halt();
121        halt(HaltReason::BackendDisk("test".to_string()));
122    }
123
124    #[rstest]
125    fn callback_runs_on_invocation() {
126        let captured: Arc<Mutex<Option<HaltReason>>> = Arc::new(Mutex::new(None));
127        let captured_for_cb = Arc::clone(&captured);
128        let halt: HaltCallback = Arc::new(move |reason| {
129            *captured_for_cb.lock().expect("lock") = Some(reason);
130        });
131
132        halt(HaltReason::BackendDisk("stall".to_string()));
133
134        match captured.lock().expect("lock").take() {
135            Some(HaltReason::BackendDisk(msg)) => assert_eq!(msg, "stall"),
136            other => panic!("expected BackendDisk, was {other:?}"),
137        }
138    }
139}
nautilus_event_store/writer/halt.rs

nautilus_event_store/writer/
halt.rs
