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

nautilus_event_store/writer/
halt.rs
