krabka_audit/log/
handle.rs

1//! The cloneable `AuditLog` handle that broker code calls to record events.
2
3use {
4    crate::{
5        event::AuditEvent,
6        sink::AuditError,
7        spool::{PendingLosses, Spool},
8    },
9    arc_swap::ArcSwapOption,
10    serde::Deserialize,
11    std::sync::{
12        Arc,
13        atomic::{AtomicU64, Ordering},
14    },
15    tokio::sync::{mpsc, oneshot},
16};
17
18/// Availability policy for privileged audit records.
19#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)]
20#[serde(rename_all = "kebab-case")]
21pub enum AuditMode {
22    /// Keep serving when audit storage is unavailable and mark the loss later.
23    #[default]
24    FailOpen,
25    /// Refuse a privileged action unless its write-ahead record is durable.
26    FailClosed,
27}
28
29pub(super) struct AuditMessage {
30    pub(super) event: AuditEvent,
31    pub(super) acknowledgement: Option<oneshot::Sender<Result<(), AuditError>>>,
32}
33
34/// Receiver half of an [`AuditLog`].
35///
36/// Tests can receive plain events; [`super::AuditWriter`] consumes the full
37/// message so it can acknowledge durable fail-closed writes.
38pub struct AuditReceiver {
39    receiver: mpsc::Receiver<AuditMessage>,
40    pending_losses: Arc<PendingLosses>,
41}
42
43impl AuditReceiver {
44    pub async fn recv(&mut self) -> Option<AuditEvent> {
45        self.receiver.recv().await.map(|message| message.event)
46    }
47
48    /// Receive an event without waiting.
49    ///
50    /// # Errors
51    /// Returns `Empty` when no event is ready and `Disconnected` when every
52    /// sender has closed.
53    pub fn try_recv(&mut self) -> Result<AuditEvent, mpsc::error::TryRecvError> {
54        self.receiver.try_recv().map(|message| message.event)
55    }
56
57    pub(super) async fn recv_message(&mut self) -> Option<AuditMessage> {
58        self.receiver.recv().await
59    }
60
61    pub(super) fn pending_losses(&self) -> Arc<PendingLosses> { Arc::clone(&self.pending_losses) }
62}
63
64/// Cloneable, cheap handle that broker code calls to record events.
65#[derive(Debug)]
66pub struct AuditLog {
67    tx: ArcSwapOption<mpsc::Sender<AuditMessage>>,
68    mode: AuditMode,
69    dropped: AtomicU64,
70    pending_losses: Arc<PendingLosses>,
71}
72
73impl AuditLog {
74    /// Create a fail-open log.
75    #[must_use]
76    pub fn new(capacity: usize) -> (Arc<Self>, AuditReceiver) {
77        Self::new_with_mode(capacity, AuditMode::FailOpen)
78    }
79
80    /// Create an enabled log with an explicit privileged-record policy.
81    #[must_use]
82    pub fn new_with_mode(capacity: usize, mode: AuditMode) -> (Arc<Self>, AuditReceiver) {
83        Self::new_with_losses(capacity, mode, PendingLosses::memory())
84    }
85
86    /// Create an enabled log backed by the spool's durable loss counter.
87    #[must_use]
88    pub fn new_with_mode_and_spool(
89        capacity: usize,
90        mode: AuditMode,
91        spool: &Spool,
92    ) -> (Arc<Self>, AuditReceiver) {
93        Self::new_with_losses(capacity, mode, spool.pending_losses())
94    }
95
96    fn new_with_losses(
97        capacity: usize,
98        mode: AuditMode,
99        pending_losses: Arc<PendingLosses>,
100    ) -> (Arc<Self>, AuditReceiver) {
101        let (tx, receiver) = mpsc::channel(capacity);
102        (
103            Arc::new(Self {
104                tx: ArcSwapOption::new(Some(Arc::new(tx))),
105                mode,
106                dropped: AtomicU64::new(0),
107                pending_losses: Arc::clone(&pending_losses),
108            }),
109            AuditReceiver {
110                receiver,
111                pending_losses,
112            },
113        )
114    }
115
116    /// A no-op log for a disabled audit subsystem.
117    #[must_use]
118    pub fn disabled() -> Arc<Self> {
119        Arc::new(Self {
120            tx: ArcSwapOption::new(None),
121            mode: AuditMode::FailOpen,
122            dropped: AtomicU64::new(0),
123            pending_losses: PendingLosses::memory(),
124        })
125    }
126
127    /// Record an ordinary fail-open event without blocking.
128    pub fn emit(&self, event: AuditEvent) {
129        let Some(tx) = self.tx.load_full() else {
130            return;
131        };
132        if tx
133            .try_send(AuditMessage {
134                event,
135                acknowledgement: None,
136            })
137            .is_err()
138        {
139            self.dropped.fetch_add(1, Ordering::Relaxed);
140            self.pending_losses.add(1);
141            tracing::warn!("audit event dropped (queue full or writer stopped)");
142        }
143    }
144
145    /// Durably record a privileged write-ahead event in fail-closed mode.
146    ///
147    /// Fail-open mode retains the non-blocking behavior of [`Self::emit`].
148    /// Fail-closed mode refuses immediately under queue backpressure, then
149    /// waits until the writer confirms that the sink or spool accepted the
150    /// record durably.
151    ///
152    /// # Errors
153    /// Returns an unavailable error if the queue is full, the writer stopped,
154    /// or neither the sink nor spool can durably store the record.
155    pub async fn emit_required(&self, event: AuditEvent) -> Result<(), AuditError> {
156        if self.mode == AuditMode::FailOpen {
157            self.emit(event);
158            return Ok(());
159        }
160        let Some(tx) = self.tx.load_full() else {
161            self.dropped.fetch_add(1, Ordering::Relaxed);
162            return Err(AuditError::Unavailable(
163                "audit writer is not running".into(),
164            ));
165        };
166        let (acknowledgement, result) = oneshot::channel();
167        tx.try_send(AuditMessage {
168            event,
169            acknowledgement: Some(acknowledgement),
170        })
171        .map_err(|error| {
172            self.dropped.fetch_add(1, Ordering::Relaxed);
173            AuditError::Unavailable(match error {
174                mpsc::error::TrySendError::Full(_) => "audit queue is full".into(),
175                mpsc::error::TrySendError::Closed(_) => "audit writer is not running".into(),
176            })
177        })?;
178        result.await.map_err(|_| {
179            self.dropped.fetch_add(1, Ordering::Relaxed);
180            AuditError::Unavailable("audit writer stopped before durable acknowledgement".into())
181        })?
182    }
183
184    /// Whether privileged actions require durable write-ahead audit records.
185    #[must_use]
186    pub fn mode(&self) -> AuditMode { self.mode }
187
188    /// Close the event stream for every clone of this handle.
189    pub fn close(&self) { self.tx.store(None); }
190
191    /// Count events lost or refused because audit processing was unavailable.
192    #[must_use]
193    pub fn dropped(&self) -> u64 { self.dropped.load(Ordering::Relaxed) }
194}
195
196#[cfg(test)]
197mod tests {
198    use {
199        super::*,
200        crate::{
201            NodeId,
202            log::test_support::{emit_lifecycle, life},
203            spool::FILE_HEADER_LEN,
204        },
205        assert2::check,
206        krabka_units::prelude,
207        std::fs,
208    };
209
210    #[tokio::test]
211    async fn close_ends_stream_for_every_log_clone_after_queued_events() {
212        let (log, mut rx) = AuditLog::new(16);
213        let clone = Arc::clone(&log);
214        log.emit(life(NodeId(1)));
215
216        clone.close();
217        log.emit(life(NodeId(2)));
218
219        check!(rx.recv().await == Some(life(NodeId(1))));
220        check!(rx.recv().await.is_none());
221        check!(log.dropped() == 0);
222    }
223
224    #[test]
225    fn disabled_log_drops_without_panicking() {
226        let log = AuditLog::disabled();
227        log.emit(life(NodeId(1)));
228        check!(log.dropped() == 0);
229    }
230
231    #[tokio::test]
232    async fn full_queue_increments_dropped_and_pending_loss() {
233        let (log, rx) = AuditLog::new(1);
234        for i in 0..10 {
235            log.emit(life(NodeId(i)));
236        }
237        check!(log.dropped() == 9);
238        check!(rx.pending_losses.count() == 9);
239    }
240
241    #[test]
242    fn full_queue_does_not_persist_losses_on_the_emit_thread() {
243        let dir = tempfile::tempdir().unwrap();
244        let spool = Spool::open(dir.path(), prelude::bytes(0)).unwrap();
245        let (log, _rx) = AuditLog::new_with_mode_and_spool(1, AuditMode::FailOpen, &spool);
246        emit_lifecycle(&log, &[NodeId(1), NodeId(2)]);
247
248        let state = fs::read(dir.path().join("audit.losses")).unwrap();
249        let count = &state[FILE_HEADER_LEN + 8..];
250        check!(u64::from_be_bytes(count.try_into().unwrap()) == 0);
251        check!(log.dropped() == 1);
252    }
253
254    #[tokio::test]
255    async fn fail_closed_refuses_queue_backpressure() {
256        let (log, _rx) = AuditLog::new_with_mode(1, AuditMode::FailClosed);
257        log.emit(life(NodeId(0)));
258        let error = log.emit_required(life(NodeId(1))).await.unwrap_err();
259        check!(error.to_string().contains("queue is full"));
260    }
261
262    #[test]
263    fn mode_returns_configured_mode() {
264        let (closed, _rx) = AuditLog::new_with_mode(1, AuditMode::FailClosed);
265        check!(closed.mode() == AuditMode::FailClosed);
266        let (open, _rx) = AuditLog::new_with_mode(1, AuditMode::FailOpen);
267        check!(open.mode() == AuditMode::FailOpen);
268    }
269}
krabka_audit/log/handle.rs

krabka_audit/log/
handle.rs
