mlua_swarm/store/run/
inmemory.rs

1//! `InMemoryRunStore` — a process-volatile `RunStore` used by the current
2//! default.
3
4use super::{
5    Inner, RunId, RunRecord, RunStatus, RunStore, RunStoreError, SharedInner, StepEntry, TaskId,
6};
7use async_trait::async_trait;
8use std::sync::Mutex;
9
10/// Process-volatile [`RunStore`] used as the current default. Entries are
11/// lost on restart; persistent backends (SQLite / Git / mini-app / …) are
12/// future carries.
13#[derive(Default)]
14pub struct InMemoryRunStore {
15    inner: SharedInner,
16}
17
18impl InMemoryRunStore {
19    /// Create an empty store.
20    pub fn new() -> Self {
21        Self {
22            inner: Mutex::new(Inner::default()),
23        }
24    }
25}
26
27#[async_trait]
28impl RunStore for InMemoryRunStore {
29    fn name(&self) -> &str {
30        "in-memory"
31    }
32
33    async fn create(&self, record: RunRecord) -> Result<(), RunStoreError> {
34        let mut inner = self.inner.lock().unwrap();
35        if inner.records.contains_key(&record.id) {
36            return Err(RunStoreError::Duplicate(record.id));
37        }
38        inner.order.push(record.id.clone());
39        inner.records.insert(record.id.clone(), record);
40        Ok(())
41    }
42
43    async fn get(&self, id: &RunId) -> Result<RunRecord, RunStoreError> {
44        let inner = self.inner.lock().unwrap();
45        inner
46            .records
47            .get(id)
48            .cloned()
49            .ok_or_else(|| RunStoreError::NotFound(id.clone()))
50    }
51
52    async fn list_by_task(&self, task_id: &TaskId) -> Result<Vec<RunRecord>, RunStoreError> {
53        let inner = self.inner.lock().unwrap();
54        let mut records: Vec<RunRecord> = inner
55            .order
56            .iter()
57            .filter_map(|id| inner.records.get(id).cloned())
58            .filter(|r| &r.task_id == task_id)
59            .collect();
60        records.sort_by_key(|r| r.created_at);
61        Ok(records)
62    }
63
64    async fn append_step_entry(&self, id: &RunId, entry: StepEntry) -> Result<(), RunStoreError> {
65        let mut inner = self.inner.lock().unwrap();
66        let record = inner
67            .records
68            .get_mut(id)
69            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
70        record.step_entries.push(entry);
71        record.updated_at = crate::types::now_unix();
72        Ok(())
73    }
74
75    async fn update_status(&self, id: &RunId, status: RunStatus) -> Result<(), RunStoreError> {
76        let mut inner = self.inner.lock().unwrap();
77        let record = inner
78            .records
79            .get_mut(id)
80            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
81        record.status = status;
82        record.updated_at = crate::types::now_unix();
83        Ok(())
84    }
85
86    async fn set_result(
87        &self,
88        id: &RunId,
89        result_ref: serde_json::Value,
90    ) -> Result<(), RunStoreError> {
91        let mut inner = self.inner.lock().unwrap();
92        let record = inner
93            .records
94            .get_mut(id)
95            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
96        record.result_ref = Some(result_ref);
97        record.updated_at = crate::types::now_unix();
98        Ok(())
99    }
100}
101
102// ──────────────────────────────────────────────────────────────────────────
103// tests
104// ──────────────────────────────────────────────────────────────────────────
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109    use serde_json::json;
110
111    fn mk(id: &str, task_id: &str, created_at: u64) -> RunRecord {
112        RunRecord {
113            id: RunId::parse(id).unwrap(),
114            task_id: TaskId::parse(task_id).unwrap(),
115            status: RunStatus::Pending,
116            step_entries: vec![],
117            operator_sid: None,
118            result_ref: None,
119            created_at,
120            updated_at: created_at,
121        }
122    }
123
124    #[tokio::test]
125    async fn create_then_get() {
126        let s = InMemoryRunStore::new();
127        s.create(mk("R-1", "T-1", 100)).await.unwrap();
128        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
129        assert_eq!(got.task_id, TaskId::parse("T-1").unwrap());
130        assert_eq!(got.status, RunStatus::Pending);
131        assert!(got.step_entries.is_empty());
132    }
133
134    #[tokio::test]
135    async fn duplicate_create_rejected() {
136        let s = InMemoryRunStore::new();
137        s.create(mk("R-1", "T-1", 100)).await.unwrap();
138        let err = s.create(mk("R-1", "T-1", 200)).await.unwrap_err();
139        assert!(matches!(err, RunStoreError::Duplicate(_)));
140    }
141
142    #[tokio::test]
143    async fn get_missing_returns_not_found() {
144        let s = InMemoryRunStore::new();
145        let err = s.get(&RunId::parse("R-nope").unwrap()).await.unwrap_err();
146        assert!(matches!(err, RunStoreError::NotFound(_)));
147    }
148
149    #[tokio::test]
150    async fn list_by_task_filters_and_orders_ascending() {
151        let s = InMemoryRunStore::new();
152        s.create(mk("R-1", "T-1", 300)).await.unwrap();
153        s.create(mk("R-2", "T-2", 50)).await.unwrap();
154        s.create(mk("R-3", "T-1", 100)).await.unwrap();
155        let list = s
156            .list_by_task(&TaskId::parse("T-1").unwrap())
157            .await
158            .unwrap();
159        let ids: Vec<_> = list.iter().map(|r| r.id.to_string()).collect();
160        assert_eq!(ids, vec!["R-3", "R-1"]);
161    }
162
163    #[tokio::test]
164    async fn append_step_entry_accumulates_in_order() {
165        let s = InMemoryRunStore::new();
166        s.create(mk("R-1", "T-1", 100)).await.unwrap();
167        s.append_step_entry(
168            &RunId::parse("R-1").unwrap(),
169            StepEntry {
170                step_id: crate::types::StepId::parse("ST-1").unwrap(),
171                step_ref: Some("step-a".into()),
172                status: Some("dispatched".into()),
173                at: 101,
174            },
175        )
176        .await
177        .unwrap();
178        s.append_step_entry(
179            &RunId::parse("R-1").unwrap(),
180            StepEntry {
181                step_id: crate::types::StepId::parse("ST-2").unwrap(),
182                step_ref: Some("step-b".into()),
183                status: Some("passed".into()),
184                at: 102,
185            },
186        )
187        .await
188        .unwrap();
189        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
190        assert_eq!(got.step_entries.len(), 2);
191        assert_eq!(got.step_entries[0].step_ref, Some("step-a".into()));
192        assert_eq!(got.step_entries[1].step_ref, Some("step-b".into()));
193        assert!(got.updated_at >= got.created_at);
194    }
195
196    #[tokio::test]
197    async fn append_step_entry_unknown_run_fails() {
198        let s = InMemoryRunStore::new();
199        let err = s
200            .append_step_entry(
201                &RunId::parse("R-nope").unwrap(),
202                StepEntry {
203                    step_id: crate::types::StepId::parse("ST-1").unwrap(),
204                    step_ref: None,
205                    status: None,
206                    at: 1,
207                },
208            )
209            .await
210            .unwrap_err();
211        assert!(matches!(err, RunStoreError::NotFound(_)));
212    }
213
214    #[tokio::test]
215    async fn update_status_persists() {
216        let s = InMemoryRunStore::new();
217        s.create(mk("R-1", "T-1", 100)).await.unwrap();
218        s.update_status(&RunId::parse("R-1").unwrap(), RunStatus::Running)
219            .await
220            .unwrap();
221        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
222        assert_eq!(got.status, RunStatus::Running);
223    }
224
225    #[tokio::test]
226    async fn set_result_persists() {
227        let s = InMemoryRunStore::new();
228        s.create(mk("R-1", "T-1", 100)).await.unwrap();
229        s.set_result(&RunId::parse("R-1").unwrap(), json!({"ok": true}))
230            .await
231            .unwrap();
232        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
233        assert_eq!(got.result_ref, Some(json!({"ok": true})));
234    }
235
236    #[tokio::test]
237    async fn name_is_in_memory() {
238        assert_eq!(InMemoryRunStore::new().name(), "in-memory");
239    }
240}
mlua_swarm/store/run/inmemory.rs

mlua_swarm/store/run/
inmemory.rs
