harn_vm/event_log/
memory.rs

1use std::collections::{HashMap, VecDeque};
2use std::sync::{Arc, Mutex};
3
4use futures::stream::BoxStream;
5
6use super::util::{prepare_event_after, stream_from_broadcast, BroadcastMap};
7use super::{
8    require_expected_topic_head, AppendHeadExpectation, AppendOutcome, CompactReport, ConsumerId,
9    EventId, EventLog, EventLogBackendKind, EventLogDescription, LogError, LogEvent, Topic,
10};
11
12#[derive(Default)]
13struct MemoryState {
14    topics: HashMap<String, VecDeque<(EventId, LogEvent)>>,
15    latest: HashMap<String, EventId>,
16    consumers: HashMap<(String, String), EventId>,
17}
18
19pub struct MemoryEventLog {
20    state: Mutex<MemoryState>,
21    pub(super) broadcasts: BroadcastMap,
22    pub(super) queue_depth: usize,
23}
24
25impl MemoryEventLog {
26    pub fn new(queue_depth: usize) -> Self {
27        Self {
28            state: Mutex::new(MemoryState::default()),
29            broadcasts: BroadcastMap::default(),
30            queue_depth: queue_depth.max(1),
31        }
32    }
33
34    fn state(&self) -> Result<std::sync::MutexGuard<'_, MemoryState>, LogError> {
35        self.state
36            .lock()
37            .map_err(|_| LogError::Io("memory event log state poisoned".to_string()))
38    }
39
40    pub(super) async fn topics(&self) -> Result<Vec<Topic>, LogError> {
41        let state = self.state()?;
42        let mut topics = state
43            .topics
44            .keys()
45            .map(|topic| Topic::new(topic.clone()))
46            .collect::<Result<Vec<_>, _>>()?;
47        topics.sort_by(|left, right| left.as_str().cmp(right.as_str()));
48        Ok(topics)
49    }
50
51    pub(super) async fn append_idempotent_by_header(
52        &self,
53        topic: &Topic,
54        header: &str,
55        value: &str,
56        event: LogEvent,
57    ) -> Result<AppendOutcome, LogError> {
58        self.append_idempotent_by_header_with_expectation(
59            topic,
60            header,
61            value,
62            AppendHeadExpectation::Any,
63            event,
64        )
65        .await
66    }
67
68    pub(super) async fn append_idempotent_chained_by_header(
69        &self,
70        topic: &Topic,
71        header: &str,
72        value: &str,
73        expected_head: Option<&str>,
74        event: LogEvent,
75    ) -> Result<AppendOutcome, LogError> {
76        self.append_idempotent_by_header_with_expectation(
77            topic,
78            header,
79            value,
80            AppendHeadExpectation::Exact(expected_head),
81            event,
82        )
83        .await
84    }
85
86    async fn append_idempotent_by_header_with_expectation(
87        &self,
88        topic: &Topic,
89        header: &str,
90        value: &str,
91        expectation: AppendHeadExpectation<'_>,
92        event: LogEvent,
93    ) -> Result<AppendOutcome, LogError> {
94        let mut state = self.state()?;
95        if let Some((event_id, existing)) = state
96            .topics
97            .get(topic.as_str())
98            .into_iter()
99            .flat_map(|events| events.iter())
100            .find(|(_, event)| {
101                event
102                    .headers
103                    .get(header)
104                    .is_some_and(|found| found == value)
105            })
106        {
107            return Ok(AppendOutcome {
108                event_id: *event_id,
109                event: existing.clone(),
110                inserted: false,
111            });
112        }
113
114        let event_id = state.latest.get(topic.as_str()).copied().unwrap_or(0) + 1;
115        let previous = state
116            .topics
117            .get(topic.as_str())
118            .and_then(|events| events.back())
119            .map(|(previous_id, previous_event)| (*previous_id, previous_event));
120        require_expected_topic_head(topic, previous, expectation)?;
121        let event = prepare_event_after(topic, event_id, previous, event)?;
122        state.latest.insert(topic.as_str().to_string(), event_id);
123        state
124            .topics
125            .entry(topic.as_str().to_string())
126            .or_default()
127            .push_back((event_id, event.clone()));
128        drop(state);
129        self.broadcasts
130            .publish(topic, self.queue_depth, (event_id, event.clone()));
131        Ok(AppendOutcome {
132            event_id,
133            event,
134            inserted: true,
135        })
136    }
137
138    /// Read counterpart of [`Self::append_idempotent_by_header`]. The in-memory
139    /// backend has no header index, so this scans the topic — acceptable for a
140    /// dev/test backend (SQLite is the durable default).
141    pub(super) async fn read_idempotent_by_header(
142        &self,
143        topic: &Topic,
144        header: &str,
145        value: &str,
146    ) -> Result<Option<(EventId, LogEvent)>, LogError> {
147        let state = self.state()?;
148        Ok(state
149            .topics
150            .get(topic.as_str())
151            .into_iter()
152            .flat_map(|events| events.iter())
153            .find(|(_, event)| {
154                event
155                    .headers
156                    .get(header)
157                    .is_some_and(|found| found == value)
158            })
159            .map(|(event_id, event)| (*event_id, event.clone())))
160    }
161}
162
163impl EventLog for MemoryEventLog {
164    fn describe(&self) -> EventLogDescription {
165        EventLogDescription {
166            backend: EventLogBackendKind::Memory,
167            location: None,
168            size_bytes: None,
169            queue_depth: self.queue_depth,
170        }
171    }
172
173    async fn append(&self, topic: &Topic, event: LogEvent) -> Result<EventId, LogError> {
174        let mut state = self.state()?;
175        let event_id = state.latest.get(topic.as_str()).copied().unwrap_or(0) + 1;
176        let previous = state
177            .topics
178            .get(topic.as_str())
179            .and_then(|events| events.back())
180            .map(|(previous_id, previous_event)| (*previous_id, previous_event));
181        let event = prepare_event_after(topic, event_id, previous, event)?;
182        state.latest.insert(topic.as_str().to_string(), event_id);
183        state
184            .topics
185            .entry(topic.as_str().to_string())
186            .or_default()
187            .push_back((event_id, event.clone()));
188        drop(state);
189        self.broadcasts
190            .publish(topic, self.queue_depth, (event_id, event));
191        Ok(event_id)
192    }
193
194    async fn flush(&self) -> Result<(), LogError> {
195        Ok(())
196    }
197
198    async fn read_range(
199        &self,
200        topic: &Topic,
201        from: Option<EventId>,
202        limit: usize,
203    ) -> Result<Vec<(EventId, LogEvent)>, LogError> {
204        let from = from.unwrap_or(0);
205        let state = self.state()?;
206        let events = state
207            .topics
208            .get(topic.as_str())
209            .into_iter()
210            .flat_map(|events| events.iter())
211            .filter(|(event_id, _)| *event_id > from)
212            .take(limit)
213            .map(|(event_id, event)| (*event_id, event.clone()))
214            .collect();
215        Ok(events)
216    }
217
218    async fn subscribe(
219        self: Arc<Self>,
220        topic: &Topic,
221        from: Option<EventId>,
222    ) -> Result<BoxStream<'static, Result<(EventId, LogEvent), LogError>>, LogError> {
223        let rx = self.broadcasts.subscribe(topic, self.queue_depth);
224        let history = self.read_range(topic, from, usize::MAX).await?;
225        Ok(stream_from_broadcast(history, from, rx, self.queue_depth))
226    }
227
228    async fn ack(
229        &self,
230        topic: &Topic,
231        consumer: &ConsumerId,
232        up_to: EventId,
233    ) -> Result<(), LogError> {
234        let mut state = self.state()?;
235        state.consumers.insert(
236            (topic.as_str().to_string(), consumer.as_str().to_string()),
237            up_to,
238        );
239        Ok(())
240    }
241
242    async fn consumer_cursor(
243        &self,
244        topic: &Topic,
245        consumer: &ConsumerId,
246    ) -> Result<Option<EventId>, LogError> {
247        let state = self.state()?;
248        Ok(state
249            .consumers
250            .get(&(topic.as_str().to_string(), consumer.as_str().to_string()))
251            .copied())
252    }
253
254    async fn latest(&self, topic: &Topic) -> Result<Option<EventId>, LogError> {
255        let state = self.state()?;
256        Ok(state.latest.get(topic.as_str()).copied())
257    }
258
259    async fn compact(&self, topic: &Topic, before: EventId) -> Result<CompactReport, LogError> {
260        let mut state = self.state()?;
261        let Some(events) = state.topics.get_mut(topic.as_str()) else {
262            return Ok(CompactReport::default());
263        };
264        let removed = events
265            .iter()
266            .take_while(|(event_id, _)| *event_id <= before)
267            .count();
268        for _ in 0..removed {
269            events.pop_front();
270        }
271        Ok(CompactReport {
272            removed,
273            remaining: events.len(),
274            latest: state.latest.get(topic.as_str()).copied(),
275            checkpointed: false,
276        })
277    }
278}
harn_vm/event_log/memory.rs

harn_vm/event_log/
memory.rs
