subetha_cxc/
event_state_log.rs

1//! `EventStateLog<Event, State>` - event-sourced state with
2//! materialized view.
3//!
4//! Composes [`SharedRing`] (the durable event log)
5//! with [`SharedCell`] (the materialized current
6//! state). Producers `emit` events onto the ring; consumers
7//! `drain_and_fold` to advance the materialized view; any process
8//! can `read_current` at constant-time for the latest state
9//! snapshot.
10//!
11//! # Architectural pattern
12//!
13//! This is the CQRS / event-sourcing shape used by Kafka +
14//! materialized views, EventStore + projections, Akka Persistence -
15//! lifted to shared memory at lock-free MMF cost. The ring file IS
16//! the durable event log (flush() syncs to disk); the cell IS the
17//! current-state cache.
18//!
19//! # Two files per log
20//!
21//! - `<base>.events.bin` - the SharedRing
22//! - `<base>.state.bin`  - the SharedCell holding State
23//!
24//! Pass the BASE PATH (without extension) to `create` / `open`;
25//! the wrapper appends the extensions.
26
27use std::marker::PhantomData;
28use std::mem::size_of;
29use std::path::{Path, PathBuf};
30use std::sync::Arc;
31
32use crate::shared_cell::{SharedCell, SharedCellError, PAYLOAD_BYTES as CELL_PAYLOAD};
33use crate::shared_ring::{RingError, SharedRing, PAYLOAD_BYTES as RING_PAYLOAD};
34
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
36pub enum EventLogError {
37    Cell(SharedCellError),
38    Ring(RingError),
39    EventTooLarge,
40    StateTooLarge,
41}
42
43impl From<SharedCellError> for EventLogError {
44    fn from(e: SharedCellError) -> Self { Self::Cell(e) }
45}
46impl From<RingError> for EventLogError {
47    fn from(e: RingError) -> Self { Self::Ring(e) }
48}
49
50fn events_path(base: &Path) -> PathBuf {
51    let mut p = base.to_path_buf();
52    let stem = p.file_name().unwrap().to_string_lossy().to_string();
53    p.set_file_name(format!("{stem}.events.bin"));
54    p
55}
56
57fn state_path(base: &Path) -> PathBuf {
58    let mut p = base.to_path_buf();
59    let stem = p.file_name().unwrap().to_string_lossy().to_string();
60    p.set_file_name(format!("{stem}.state.bin"));
61    p
62}
63
64pub struct EventStateLog<Event: Copy + 'static, State: Copy + 'static> {
65    ring: Arc<SharedRing>,
66    state: Arc<SharedCell<State>>,
67    _phantom_e: PhantomData<Event>,
68    header_sidecar: subetha_core::HandshakeHeader,
69    ring_sidecar: Box<subetha_core::ObservationRing>,
70}
71
72impl<Event: Copy + Send + Sync + 'static, State: Copy + Send + Sync + 'static>
73    subetha_sidecar::AdaptiveInstance for EventStateLog<Event, State>
74{
75    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
76    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
77    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
78        Box::new(subetha_sidecar::NoMigrationPolicy)
79    }
80}
81
82impl<Event: Copy + 'static, State: Copy + 'static> EventStateLog<Event, State> {
83    pub fn create(
84        base_path: impl AsRef<Path>,
85        ring_capacity: usize,
86        initial_state: State,
87    ) -> Result<Self, EventLogError> {
88        if size_of::<Event>() > RING_PAYLOAD { return Err(EventLogError::EventTooLarge); }
89        if size_of::<State>() > CELL_PAYLOAD { return Err(EventLogError::StateTooLarge); }
90        let base = base_path.as_ref();
91        let ring = Arc::new(SharedRing::create(events_path(base), ring_capacity)?);
92        let state = Arc::new(SharedCell::<State>::create(state_path(base))?);
93        state.set(initial_state);
94        Ok(Self {
95            ring, state, _phantom_e: PhantomData,
96            header_sidecar: subetha_core::HandshakeHeader::new(),
97            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
98        })
99    }
100
101    pub fn open(
102        base_path: impl AsRef<Path>,
103        ring_capacity: usize,
104    ) -> Result<Self, EventLogError> {
105        if size_of::<Event>() > RING_PAYLOAD { return Err(EventLogError::EventTooLarge); }
106        if size_of::<State>() > CELL_PAYLOAD { return Err(EventLogError::StateTooLarge); }
107        let base = base_path.as_ref();
108        let ring = Arc::new(SharedRing::open(events_path(base), ring_capacity)?);
109        let state = Arc::new(SharedCell::<State>::open(state_path(base))?);
110        Ok(Self {
111            ring, state, _phantom_e: PhantomData,
112            header_sidecar: subetha_core::HandshakeHeader::new(),
113            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
114        })
115    }
116
117    /// Push an event onto the durable log. Returns `Err(Ring(Full))`
118    /// when the ring is full; callers should drain or apply
119    /// backpressure.
120    pub fn emit(&self, event: Event) -> Result<(), EventLogError> {
121        let bytes: [u8; RING_PAYLOAD] = {
122            let mut buf = [0u8; RING_PAYLOAD];
123            // SAFETY: Event is Copy + Sized, size_of::<Event>() <= RING_PAYLOAD
124            // (checked at create/open). We memcpy the event's bytes
125            // into the ring slot's payload region.
126            unsafe {
127                std::ptr::copy_nonoverlapping(
128                    &event as *const Event as *const u8,
129                    buf.as_mut_ptr(),
130                    size_of::<Event>(),
131                );
132            }
133            buf
134        };
135        let r = self.ring.try_push(&bytes);
136        self.ring_sidecar.push_op(
137            crate::sidecar_ops::event_log::OP_EMIT,
138            if r.is_err() { 1 } else { 0 },
139        );
140        r?;
141        Ok(())
142    }
143
144    /// Drain all pending events from the log and apply each to the
145    /// current state via `fold`. The state cell is updated after
146    /// all events are folded. Returns the number of events applied.
147    pub fn drain_and_fold<F: FnMut(&mut State, &Event)>(
148        &self,
149        mut fold: F,
150    ) -> usize {
151        let mut state = self.state.get();
152        let mut count = 0;
153        let mut buf = [0u8; RING_PAYLOAD];
154        loop {
155            match self.ring.try_pop(&mut buf) {
156                Ok(_) => {
157                    // SAFETY: bytes were pushed by emit() with the
158                    // same Event layout.
159                    let event: Event = unsafe {
160                        std::ptr::read_unaligned(buf.as_ptr() as *const Event)
161                    };
162                    fold(&mut state, &event);
163                    count += 1;
164                }
165                Err(RingError::Empty) => break,
166                Err(_) => break,
167            }
168        }
169        if count > 0 {
170            self.state.set(state);
171        }
172        self.ring_sidecar
173            .push_op(crate::sidecar_ops::event_log::OP_DRAIN_FOLD, 0);
174        count
175    }
176
177    /// Read the current materialized state. O(1) - one SeqLock cell read.
178    pub fn read_current(&self) -> State {
179        let s = self.state.get();
180        self.ring_sidecar
181            .push_op(crate::sidecar_ops::event_log::OP_READ_CURRENT, 0);
182        s
183    }
184
185    /// Approximate number of events waiting in the log.
186    pub fn pending_events(&self) -> usize {
187        self.ring.approx_len()
188    }
189
190    /// Force-set the materialized state (e.g., for checkpoint restore).
191    pub fn set_state(&self, state: State) {
192        self.state.set(state);
193    }
194
195    /// Sync both files to disk.
196    pub fn flush(&self) -> Result<(), EventLogError> {
197        self.ring.flush()?;
198        self.state.flush()?;
199        Ok(())
200    }
201
202    /// Non-blocking flush of both files. Delegates to the ring and
203    /// the state cell's flush_async.
204    /// Note: Windows is only partially async (sync to page cache,
205    /// not to disk).
206    pub fn flush_async(&self) -> Result<(), EventLogError> {
207        self.ring.flush_async()?;
208        self.state.flush_async()?;
209        Ok(())
210    }
211}
212
213#[cfg(test)]
214mod tests {
215    use super::*;
216    use std::thread;
217
218    fn tmp_base(name: &str) -> PathBuf {
219        let mut p = std::env::temp_dir();
220        let pid = std::process::id();
221        p.push(format!("subetha-eventlog-{name}-{pid}"));
222        p
223    }
224
225    fn cleanup(base: &Path) {
226        std::fs::remove_file(events_path(base)).ok();
227        std::fs::remove_file(state_path(base)).ok();
228    }
229
230    #[test]
231    fn emit_drain_fold_round_trip() {
232        let base = tmp_base("rt");
233        let log: EventStateLog<u32, u32> = EventStateLog::create(&base, 16, 0).unwrap();
234        log.emit(10).unwrap();
235        log.emit(20).unwrap();
236        log.emit(30).unwrap();
237        assert_eq!(log.pending_events(), 3);
238        let n = log.drain_and_fold(|s, e| *s += *e);
239        assert_eq!(n, 3);
240        assert_eq!(log.read_current(), 60);
241        assert_eq!(log.pending_events(), 0);
242        cleanup(&base);
243    }
244
245    #[test]
246    fn initial_state_is_visible_before_any_emit() {
247        let base = tmp_base("initial");
248        let log: EventStateLog<u32, u64> = EventStateLog::create(&base, 8, 1234).unwrap();
249        assert_eq!(log.read_current(), 1234);
250        cleanup(&base);
251    }
252
253    #[test]
254    fn cross_handle_emit_and_drain() {
255        let base = tmp_base("cross-handle");
256        let producer: EventStateLog<u32, u32> = EventStateLog::create(&base, 16, 0).unwrap();
257        let consumer: EventStateLog<u32, u32> = EventStateLog::open(&base, 16).unwrap();
258        producer.emit(5).unwrap();
259        producer.emit(7).unwrap();
260        producer.emit(11).unwrap();
261        // Consumer drains and folds independently of producer.
262        let n = consumer.drain_and_fold(|s, e| *s += *e);
263        assert_eq!(n, 3);
264        assert_eq!(consumer.read_current(), 23);
265        // Producer reads the consumer's update.
266        assert_eq!(producer.read_current(), 23);
267        cleanup(&base);
268    }
269
270    #[test]
271    fn drain_with_empty_log_returns_zero() {
272        let base = tmp_base("empty-drain");
273        let log: EventStateLog<u32, u32> = EventStateLog::create(&base, 8, 100).unwrap();
274        let n = log.drain_and_fold(|_, _| panic!("must not run on empty"));
275        assert_eq!(n, 0);
276        assert_eq!(log.read_current(), 100);  // unchanged
277        cleanup(&base);
278    }
279
280    #[test]
281    fn full_ring_returns_error_on_emit() {
282        let base = tmp_base("full");
283        let log: EventStateLog<u32, u32> = EventStateLog::create(&base, 4, 0).unwrap();
284        for i in 0..4u32 { log.emit(i).unwrap(); }
285        match log.emit(99) {
286            Err(EventLogError::Ring(RingError::Full)) => {}
287            other => panic!("expected Ring(Full), got {other:?}"),
288        }
289        cleanup(&base);
290    }
291
292    #[test]
293    fn concurrent_producers_drain_correctly() {
294        let base = tmp_base("concurrent");
295        let log: Arc<EventStateLog<u32, u64>>
296            = Arc::new(EventStateLog::create(&base, 256, 0).unwrap());
297        let n_threads = 4;
298        let per_thread = 50u32;
299        let total = (n_threads as u32) * per_thread;
300        let mut handles = vec![];
301        for _ in 0..n_threads {
302            let log = log.clone();
303            handles.push(thread::spawn(move || {
304                for i in 0..per_thread {
305                    while log.emit(i).is_err() {
306                        std::hint::spin_loop();
307                    }
308                }
309            }));
310        }
311        for h in handles { h.join().unwrap(); }
312        // Single-threaded drain to count.
313        let n = log.drain_and_fold(|s, e| *s += *e as u64);
314        assert_eq!(n, total as usize);
315        // Sum = N * sum(0..per_thread)
316        let expected = (n_threads as u64) * (0u64..per_thread as u64).sum::<u64>();
317        assert_eq!(log.read_current(), expected);
318        cleanup(&base);
319    }
320
321    #[test]
322    fn disk_persistence_state_survives_reopen() {
323        let base = tmp_base("disk");
324        {
325            let log: EventStateLog<u32, u32> = EventStateLog::create(&base, 8, 0).unwrap();
326            log.emit(10).unwrap();
327            log.emit(20).unwrap();
328            log.drain_and_fold(|s, e| *s += *e);
329            log.flush().unwrap();
330        }
331        let log2: EventStateLog<u32, u32> = EventStateLog::open(&base, 8).unwrap();
332        assert_eq!(log2.read_current(), 30);
333        cleanup(&base);
334    }
335
336    #[test]
337    fn set_state_overrides_for_checkpoint_restore() {
338        let base = tmp_base("set-state");
339        let log: EventStateLog<u32, u32> = EventStateLog::create(&base, 8, 0).unwrap();
340        log.emit(5).unwrap();
341        log.drain_and_fold(|s, e| *s += *e);
342        assert_eq!(log.read_current(), 5);
343        log.set_state(9999);
344        assert_eq!(log.read_current(), 9999);
345        cleanup(&base);
346    }
347
348    #[test]
349    fn struct_event_and_state_round_trip() {
350        #[derive(Clone, Copy, Debug, PartialEq)]
351        #[repr(C)]
352        struct Event { delta_x: i32, delta_y: i32 }
353        #[derive(Clone, Copy, Debug, PartialEq)]
354        #[repr(C)]
355        struct State { x: i32, y: i32 }
356        let base = tmp_base("struct");
357        let log: EventStateLog<Event, State> = EventStateLog::create(
358            &base, 16, State { x: 0, y: 0 },
359        ).unwrap();
360        log.emit(Event { delta_x: 3, delta_y: 4 }).unwrap();
361        log.emit(Event { delta_x: -1, delta_y: 2 }).unwrap();
362        let n = log.drain_and_fold(|s, e| {
363            s.x += e.delta_x;
364            s.y += e.delta_y;
365        });
366        assert_eq!(n, 2);
367        assert_eq!(log.read_current(), State { x: 2, y: 6 });
368        cleanup(&base);
369    }
370}
subetha_cxc/event_state_log.rs

subetha_cxc/
event_state_log.rs
