mlua_swarm/store/run/
inmemory.rs

1//! `InMemoryRunStore` — a process-volatile `RunStore` used by the current
2//! default.
3
4use super::{
5    Assignee, DegradationEntry, Inner, RunId, RunListFilter, RunRecord, RunStatus, RunStore,
6    RunStoreError, SharedInner, StepEntry, TaskId, VacateOutcome,
7};
8use async_trait::async_trait;
9use std::sync::Mutex;
10
11/// Process-volatile [`RunStore`] used as the current default. Entries are
12/// lost on restart; persistent backends (SQLite / Git / mini-app / …) are
13/// future carries.
14#[derive(Default)]
15pub struct InMemoryRunStore {
16    inner: SharedInner,
17}
18
19impl InMemoryRunStore {
20    /// Create an empty store.
21    pub fn new() -> Self {
22        Self {
23            inner: Mutex::new(Inner::default()),
24        }
25    }
26}
27
28#[async_trait]
29impl RunStore for InMemoryRunStore {
30    fn name(&self) -> &str {
31        "in-memory"
32    }
33
34    async fn create(&self, record: RunRecord) -> Result<(), RunStoreError> {
35        // **A2** on the way in — see `RunStore::create`. Checked before the
36        // lock so a rejected record never touches the map.
37        record.validate_assignment_generations()?;
38        let mut inner = self.inner.lock().unwrap();
39        if inner.records.contains_key(&record.id) {
40            return Err(RunStoreError::Duplicate(record.id));
41        }
42        inner.order.push(record.id.clone());
43        inner.records.insert(record.id.clone(), record);
44        Ok(())
45    }
46
47    async fn get(&self, id: &RunId) -> Result<RunRecord, RunStoreError> {
48        let inner = self.inner.lock().unwrap();
49        inner
50            .records
51            .get(id)
52            .cloned()
53            .ok_or_else(|| RunStoreError::NotFound(id.clone()))
54    }
55
56    async fn list_by_task(&self, task_id: &TaskId) -> Result<Vec<RunRecord>, RunStoreError> {
57        let inner = self.inner.lock().unwrap();
58        let mut records: Vec<RunRecord> = inner
59            .order
60            .iter()
61            .filter_map(|id| inner.records.get(id).cloned())
62            .filter(|r| &r.task_id == task_id)
63            .collect();
64        records.sort_by_key(|r| r.created_at);
65        Ok(records)
66    }
67
68    async fn append_step_entry(&self, id: &RunId, entry: StepEntry) -> Result<(), RunStoreError> {
69        let mut inner = self.inner.lock().unwrap();
70        let record = inner
71            .records
72            .get_mut(id)
73            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
74        record.step_entries.push(entry);
75        record.updated_at = crate::types::now_unix();
76        Ok(())
77    }
78
79    async fn append_degradation(
80        &self,
81        id: &RunId,
82        entry: DegradationEntry,
83    ) -> Result<(), RunStoreError> {
84        let mut inner = self.inner.lock().unwrap();
85        let record = inner
86            .records
87            .get_mut(id)
88            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
89        record.degradations.push(entry);
90        record.updated_at = crate::types::now_unix();
91        Ok(())
92    }
93
94    async fn update_status(&self, id: &RunId, status: RunStatus) -> Result<(), RunStoreError> {
95        let mut inner = self.inner.lock().unwrap();
96        let record = inner
97            .records
98            .get_mut(id)
99            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
100        record.status = status;
101        record.updated_at = crate::types::now_unix();
102        Ok(())
103    }
104
105    async fn try_transition(
106        &self,
107        id: &RunId,
108        from: RunStatus,
109        to: RunStatus,
110    ) -> Result<bool, RunStoreError> {
111        let mut inner = self.inner.lock().unwrap();
112        // Held under the single `inner` mutex, so the read + compare + set
113        // is atomic against any other appender/transition. An absent row or
114        // a status mismatch both report `false` (the caller's race signal),
115        // not an error.
116        match inner.records.get_mut(id) {
117            Some(record) if record.status == from => {
118                record.status = to;
119                record.updated_at = crate::types::now_unix();
120                Ok(true)
121            }
122            _ => Ok(false),
123        }
124    }
125
126    async fn acquire_assignee(
127        &self,
128        id: &RunId,
129        slot: &str,
130        op: &str,
131        desc: &str,
132    ) -> Result<(u64, Option<Assignee>), RunStoreError> {
133        // A9 (and the slot's own requirement): refuse before taking the
134        // lock — a rejected acquire must not burn a generation.
135        if slot.is_empty() {
136            return Err(RunStoreError::AssigneeSlotRequired);
137        }
138        if desc.trim().is_empty() {
139            return Err(RunStoreError::AssigneeDescRequired);
140        }
141        let mut inner = self.inner.lock().unwrap();
142        let record = inner
143            .records
144            .get_mut(id)
145            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
146        // A4: the bump is unconditional (an Assign to the incumbent still
147        // advances the counter) and Run-wide (a different slot advances the
148        // same counter). Held under the single `inner` mutex with no await
149        // in between, so concurrent acquires cannot read the same
150        // generation. A8: no precondition on the incumbent.
151        record.next_generation += 1;
152        // Q3: `insert` returns the previous holder OF THIS SLOT, moved out
153        // whole — it is handed back with its `gen` intact (A3), never
154        // rewritten. Other slots' entries are not read or touched.
155        let previous = record.current.insert(
156            slot.to_string(),
157            Assignee {
158                op: op.to_string(),
159                desc: desc.to_string(),
160                gen: record.next_generation,
161            },
162        );
163        record.updated_at = crate::types::now_unix();
164        Ok((record.next_generation, previous))
165    }
166
167    async fn vacate_assignee(
168        &self,
169        id: &RunId,
170        slot: &str,
171        expected_gen: u64,
172    ) -> Result<VacateOutcome, RunStoreError> {
173        if slot.is_empty() {
174            return Err(RunStoreError::AssigneeSlotRequired);
175        }
176        let mut inner = self.inner.lock().unwrap();
177        let record = inner
178            .records
179            .get_mut(id)
180            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
181        // The compare and the removal are both under the single `inner`
182        // mutex with no await between them, so an acquire cannot slip in
183        // after the generation matched and before the holder is dropped.
184        match record.current.get(slot) {
185            Some(held) if held.gen == expected_gen => {}
186            other => {
187                return Ok(VacateOutcome::Stale {
188                    current: other.cloned(),
189                });
190            }
191        }
192        // Only this slot's key leaves the map — a Vacant is per seat. The
193        // `else` cannot be reached while the lock is held (the key was just
194        // matched), and it is written as data rather than a panic: nothing
195        // was removed, so nothing was released.
196        let Some(released) = record.current.remove(slot) else {
197            return Ok(VacateOutcome::Stale { current: None });
198        };
199        // A4: a release that happens bumps `G` exactly like Assign does; it
200        // just mints no Assignee, so the next acquire continues from the
201        // bumped value. A Stale answer returned above wrote nothing.
202        record.next_generation += 1;
203        record.updated_at = crate::types::now_unix();
204        Ok(VacateOutcome::Released {
205            generation: record.next_generation,
206            released,
207        })
208    }
209
210    async fn set_result(
211        &self,
212        id: &RunId,
213        result_ref: serde_json::Value,
214    ) -> Result<(), RunStoreError> {
215        let mut inner = self.inner.lock().unwrap();
216        let record = inner
217            .records
218            .get_mut(id)
219            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
220        record.result_ref = Some(result_ref);
221        record.updated_at = crate::types::now_unix();
222        Ok(())
223    }
224
225    async fn set_input_json(&self, id: &RunId, input_json: String) -> Result<(), RunStoreError> {
226        let mut inner = self.inner.lock().unwrap();
227        let record = inner
228            .records
229            .get_mut(id)
230            .ok_or_else(|| RunStoreError::NotFound(id.clone()))?;
231        record.input_json = Some(input_json);
232        record.updated_at = crate::types::now_unix();
233        Ok(())
234    }
235
236    async fn list_running(&self) -> Result<Vec<RunRecord>, RunStoreError> {
237        let inner = self.inner.lock().unwrap();
238        let records: Vec<RunRecord> = inner
239            .order
240            .iter()
241            .filter_map(|id| inner.records.get(id).cloned())
242            .filter(|r| r.status == RunStatus::Running)
243            .collect();
244        Ok(records)
245    }
246
247    async fn list(&self, filter: &RunListFilter) -> Result<Vec<RunRecord>, RunStoreError> {
248        let inner = self.inner.lock().unwrap();
249        let mut records: Vec<RunRecord> = inner
250            .order
251            .iter()
252            .filter_map(|id| inner.records.get(id).cloned())
253            .filter(|r| {
254                filter
255                    .task_id
256                    .as_ref()
257                    .map(|t| &r.task_id == t)
258                    .unwrap_or(true)
259                    && filter.status.map(|s| r.status == s).unwrap_or(true)
260            })
261            .collect();
262        // Newest-first; `order` index breaks `created_at` ties stably
263        // (later insertion sorts first within the same second).
264        records.reverse();
265        records.sort_by_key(|r| std::cmp::Reverse(r.created_at));
266        let offset = filter.offset.unwrap_or(0);
267        let records: Vec<RunRecord> = records
268            .into_iter()
269            .skip(offset)
270            .take(filter.limit.unwrap_or(usize::MAX))
271            .collect();
272        Ok(records)
273    }
274
275    async fn delete(&self, id: &RunId) -> Result<(), RunStoreError> {
276        let mut inner = self.inner.lock().unwrap();
277        if inner.records.remove(id).is_none() {
278            return Err(RunStoreError::NotFound(id.clone()));
279        }
280        inner.order.retain(|r| r != id);
281        Ok(())
282    }
283}
284
285// ──────────────────────────────────────────────────────────────────────────
286// tests
287// ──────────────────────────────────────────────────────────────────────────
288
289#[cfg(test)]
290mod tests {
291    use super::*;
292    use serde_json::json;
293
294    fn mk(id: &str, task_id: &str, created_at: u64) -> RunRecord {
295        RunRecord {
296            id: RunId::parse(id).unwrap(),
297            task_id: TaskId::parse(task_id).unwrap(),
298            status: RunStatus::Pending,
299            step_entries: vec![],
300            degradations: vec![],
301            operator_sid: None,
302            current: Default::default(),
303            next_generation: 0,
304            result_ref: None,
305            input_json: None,
306            created_at,
307            updated_at: created_at,
308        }
309    }
310
311    fn mk_degradation(tool: &str, at: u64) -> DegradationEntry {
312        DegradationEntry {
313            tool: tool.to_string(),
314            error: "boom".to_string(),
315            fallback: "cached-default".to_string(),
316            note: None,
317            step_ref: Some("worker".to_string()),
318            attempt: Some(1),
319            at,
320        }
321    }
322
323    #[tokio::test]
324    async fn create_then_get() {
325        let s = InMemoryRunStore::new();
326        s.create(mk("R-1", "T-1", 100)).await.unwrap();
327        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
328        assert_eq!(got.task_id, TaskId::parse("T-1").unwrap());
329        assert_eq!(got.status, RunStatus::Pending);
330        assert!(got.step_entries.is_empty());
331    }
332
333    /// **A2** at the one door that takes a caller-built record. Seeding
334    /// `current[slot].gen = 99` against `next_generation = 0` would be
335    /// permanent: the next acquire stamps generation 1, *below* the
336    /// incumbent, and ordering two holders by `gen` — the reason `G` is
337    /// Run-wide — silently inverts. So it is refused, and nothing is
338    /// stored.
339    #[tokio::test]
340    async fn create_rejects_a_holder_generation_above_the_counter() {
341        let s = InMemoryRunStore::new();
342        let mut record = mk("R-1", "T-1", 100);
343        record.current.insert(
344            SLOT_A.to_string(),
345            Assignee {
346                op: "S-seeded".to_string(),
347                desc: "seeded straight into the record".to_string(),
348                gen: 99,
349            },
350        );
351
352        let err = s.create(record).await.unwrap_err();
353        match err {
354            RunStoreError::AssigneeGenerationAhead {
355                slot,
356                gen,
357                next_generation,
358            } => {
359                assert_eq!(slot, SLOT_A);
360                assert_eq!(gen, 99);
361                assert_eq!(next_generation, 0);
362            }
363            other => panic!("got: {other:?}"),
364        }
365        assert!(
366            matches!(
367                s.get(&RunId::parse("R-1").unwrap()).await.unwrap_err(),
368                RunStoreError::NotFound(_)
369            ),
370            "a refused create must leave no row behind"
371        );
372    }
373
374    /// The boundary is `>`, not `>=`: a record whose holder was stamped at
375    /// exactly `G` satisfies `a.gen ≤ G` and is the normal shape of a Run
376    /// that has been assigned once, so round-tripping one through `create`
377    /// must keep working.
378    #[tokio::test]
379    async fn create_accepts_a_holder_generation_equal_to_the_counter() {
380        let s = InMemoryRunStore::new();
381        let mut record = mk("R-1", "T-1", 100);
382        record.next_generation = 1;
383        record.current.insert(
384            SLOT_A.to_string(),
385            Assignee {
386                op: "S-a1".to_string(),
387                desc: "stamped at G".to_string(),
388                gen: 1,
389            },
390        );
391
392        s.create(record).await.unwrap();
393        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
394        assert_eq!(got.current[SLOT_A].gen, 1);
395    }
396
397    #[tokio::test]
398    async fn duplicate_create_rejected() {
399        let s = InMemoryRunStore::new();
400        s.create(mk("R-1", "T-1", 100)).await.unwrap();
401        let err = s.create(mk("R-1", "T-1", 200)).await.unwrap_err();
402        assert!(matches!(err, RunStoreError::Duplicate(_)));
403    }
404
405    #[tokio::test]
406    async fn get_missing_returns_not_found() {
407        let s = InMemoryRunStore::new();
408        let err = s.get(&RunId::parse("R-nope").unwrap()).await.unwrap_err();
409        assert!(matches!(err, RunStoreError::NotFound(_)));
410    }
411
412    #[tokio::test]
413    async fn list_by_task_filters_and_orders_ascending() {
414        let s = InMemoryRunStore::new();
415        s.create(mk("R-1", "T-1", 300)).await.unwrap();
416        s.create(mk("R-2", "T-2", 50)).await.unwrap();
417        s.create(mk("R-3", "T-1", 100)).await.unwrap();
418        let list = s
419            .list_by_task(&TaskId::parse("T-1").unwrap())
420            .await
421            .unwrap();
422        let ids: Vec<_> = list.iter().map(|r| r.id.to_string()).collect();
423        assert_eq!(ids, vec!["R-3", "R-1"]);
424    }
425
426    #[tokio::test]
427    async fn append_step_entry_accumulates_in_order() {
428        let s = InMemoryRunStore::new();
429        s.create(mk("R-1", "T-1", 100)).await.unwrap();
430        s.append_step_entry(
431            &RunId::parse("R-1").unwrap(),
432            StepEntry::basic(
433                crate::types::StepId::parse("ST-1").unwrap(),
434                Some("step-a".into()),
435                Some("dispatched".into()),
436                None,
437                101,
438            ),
439        )
440        .await
441        .unwrap();
442        s.append_step_entry(
443            &RunId::parse("R-1").unwrap(),
444            StepEntry::basic(
445                crate::types::StepId::parse("ST-2").unwrap(),
446                Some("step-b".into()),
447                Some("passed".into()),
448                None,
449                102,
450            ),
451        )
452        .await
453        .unwrap();
454        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
455        assert_eq!(got.step_entries.len(), 2);
456        assert_eq!(got.step_entries[0].step_ref, Some("step-a".into()));
457        assert_eq!(got.step_entries[1].step_ref, Some("step-b".into()));
458        assert!(got.updated_at >= got.created_at);
459    }
460
461    #[tokio::test]
462    async fn append_degradation_accumulates_in_order() {
463        let s = InMemoryRunStore::new();
464        s.create(mk("R-1", "T-1", 100)).await.unwrap();
465        s.append_degradation(
466            &RunId::parse("R-1").unwrap(),
467            mk_degradation("web_search", 101),
468        )
469        .await
470        .unwrap();
471        s.append_degradation(
472            &RunId::parse("R-1").unwrap(),
473            mk_degradation("code_exec", 102),
474        )
475        .await
476        .unwrap();
477        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
478        assert_eq!(got.degradations.len(), 2);
479        assert_eq!(got.degradations[0].tool, "web_search");
480        assert_eq!(got.degradations[1].tool, "code_exec");
481        assert!(got.updated_at >= got.created_at);
482    }
483
484    #[tokio::test]
485    async fn append_degradation_unknown_run_fails() {
486        let s = InMemoryRunStore::new();
487        let err = s
488            .append_degradation(
489                &RunId::parse("R-nope").unwrap(),
490                mk_degradation("web_search", 1),
491            )
492            .await
493            .unwrap_err();
494        assert!(matches!(err, RunStoreError::NotFound(_)));
495    }
496
497    #[tokio::test]
498    async fn append_degradation_bumps_updated_at() {
499        let s = InMemoryRunStore::new();
500        s.create(mk("R-1", "T-1", 100)).await.unwrap();
501        s.append_degradation(
502            &RunId::parse("R-1").unwrap(),
503            mk_degradation("web_search", 200),
504        )
505        .await
506        .unwrap();
507        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
508        assert!(got.updated_at > 100);
509    }
510
511    #[tokio::test]
512    async fn append_step_entry_unknown_run_fails() {
513        let s = InMemoryRunStore::new();
514        let err = s
515            .append_step_entry(
516                &RunId::parse("R-nope").unwrap(),
517                StepEntry::basic(
518                    crate::types::StepId::parse("ST-1").unwrap(),
519                    None,
520                    None,
521                    None,
522                    1,
523                ),
524            )
525            .await
526            .unwrap_err();
527        assert!(matches!(err, RunStoreError::NotFound(_)));
528    }
529
530    #[tokio::test]
531    async fn update_status_persists() {
532        let s = InMemoryRunStore::new();
533        s.create(mk("R-1", "T-1", 100)).await.unwrap();
534        s.update_status(&RunId::parse("R-1").unwrap(), RunStatus::Running)
535            .await
536            .unwrap();
537        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
538        assert_eq!(got.status, RunStatus::Running);
539    }
540
541    #[tokio::test]
542    async fn set_result_persists() {
543        let s = InMemoryRunStore::new();
544        s.create(mk("R-1", "T-1", 100)).await.unwrap();
545        s.set_result(&RunId::parse("R-1").unwrap(), json!({"ok": true}))
546            .await
547            .unwrap();
548        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
549        assert_eq!(got.result_ref, Some(json!({"ok": true})));
550    }
551
552    #[tokio::test]
553    async fn name_is_in_memory() {
554        assert_eq!(InMemoryRunStore::new().name(), "in-memory");
555    }
556
557    #[tokio::test]
558    async fn list_running_filters_by_status() {
559        let s = InMemoryRunStore::new();
560        s.create(mk("R-1", "T-1", 100)).await.unwrap();
561        s.create(mk("R-2", "T-2", 200)).await.unwrap();
562        s.create(mk("R-3", "T-3", 300)).await.unwrap();
563        s.update_status(&RunId::parse("R-2").unwrap(), RunStatus::Running)
564            .await
565            .unwrap();
566        s.update_status(&RunId::parse("R-3").unwrap(), RunStatus::Done)
567            .await
568            .unwrap();
569        let running = s.list_running().await.unwrap();
570        assert_eq!(running.len(), 1);
571        assert_eq!(running[0].id, RunId::parse("R-2").unwrap());
572        assert_eq!(running[0].status, RunStatus::Running);
573    }
574
575    #[tokio::test]
576    async fn try_transition_flips_on_match_and_is_idempotent_under_race() {
577        let s = InMemoryRunStore::new();
578        s.create(mk("R-1", "T-1", 100)).await.unwrap();
579        s.update_status(&RunId::parse("R-1").unwrap(), RunStatus::Interrupted)
580            .await
581            .unwrap();
582
583        // First CAS matches `Interrupted` and flips to `Running`.
584        let first = s
585            .try_transition(
586                &RunId::parse("R-1").unwrap(),
587                RunStatus::Interrupted,
588                RunStatus::Running,
589            )
590            .await
591            .unwrap();
592        assert!(first, "first CAS must flip Interrupted -> Running");
593        assert_eq!(
594            s.get(&RunId::parse("R-1").unwrap()).await.unwrap().status,
595            RunStatus::Running
596        );
597
598        // Second CAS (a racing double-resume) no longer sees `Interrupted`
599        // and must report `false` without touching the row.
600        let second = s
601            .try_transition(
602                &RunId::parse("R-1").unwrap(),
603                RunStatus::Interrupted,
604                RunStatus::Running,
605            )
606            .await
607            .unwrap();
608        assert!(!second, "second CAS must not flip a now-Running row");
609    }
610
611    #[tokio::test]
612    async fn try_transition_absent_run_reports_false() {
613        let s = InMemoryRunStore::new();
614        let flipped = s
615            .try_transition(
616                &RunId::parse("R-nope").unwrap(),
617                RunStatus::Interrupted,
618                RunStatus::Running,
619            )
620            .await
621            .unwrap();
622        assert!(!flipped, "an absent Run must report false, not error");
623    }
624
625    // ── assignment axis (model §4.3) ──────────────────────────────────
626
627    /// The two slots (Blueprint-declared Operator seats) the tests below
628    /// assign to — the shipped per-lane alias shape, where a Blueprint
629    /// declares one seat per phase.
630    const SLOT_A: &str = "phase-a-op";
631    const SLOT_B: &str = "phase-b-op";
632
633    /// A4: a launched Run starts with every slot Vacant and `G == 0` — the
634    /// counter is not pre-advanced by the launch itself.
635    #[tokio::test]
636    async fn launch_starts_vacant_at_generation_zero() {
637        let s = InMemoryRunStore::new();
638        s.create(mk("R-1", "T-1", 100)).await.unwrap();
639        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
640        assert!(got.current.is_empty(), "no slot is held at launch");
641        assert_eq!(got.next_generation, 0);
642    }
643
644    /// A4: every event advances `G` by one, and the FIRST Assign lands on
645    /// `1`. A8: re-acquiring for the incumbent still succeeds and still
646    /// advances — the counter tracks events, not state changes.
647    #[tokio::test]
648    async fn acquire_advances_generation_even_for_the_same_op() {
649        let s = InMemoryRunStore::new();
650        s.create(mk("R-1", "T-1", 100)).await.unwrap();
651        let id = RunId::parse("R-1").unwrap();
652
653        let (gen, previous) = s
654            .acquire_assignee(&id, SLOT_A, "S-a1", "first hold")
655            .await
656            .unwrap();
657        assert_eq!(gen, 1, "the first Assign stamps generation 1");
658        assert_eq!(previous, None);
659
660        let (gen, previous) = s
661            .acquire_assignee(&id, SLOT_A, "S-a1", "same holder, new event")
662            .await
663            .unwrap();
664        assert_eq!(gen, 2, "A4: a repeat Assign for the same op still bumps");
665        assert_eq!(previous.expect("displaced holder").gen, 1);
666
667        let got = s.get(&id).await.unwrap();
668        assert_eq!(got.next_generation, 2);
669        assert_eq!(
670            got.current.len(),
671            1,
672            "A1: re-assigning a slot leaves it with exactly one holder"
673        );
674        assert_eq!(got.current[SLOT_A].gen, 2);
675    }
676
677    /// The slots are independent: assigning one leaves every other Vacant.
678    #[tokio::test]
679    async fn assigning_one_slot_leaves_the_others_vacant() {
680        let s = InMemoryRunStore::new();
681        s.create(mk("R-1", "T-1", 100)).await.unwrap();
682        let id = RunId::parse("R-1").unwrap();
683
684        s.acquire_assignee(&id, SLOT_A, "S-a1", "holds phase a")
685            .await
686            .unwrap();
687
688        let got = s.get(&id).await.unwrap();
689        assert_eq!(got.current[SLOT_A].op, "S-a1");
690        assert!(
691            !got.current.contains_key(SLOT_B),
692            "an unassigned slot has no entry — that absence IS its Vacant"
693        );
694    }
695
696    /// A4 is Run-wide, not per slot: interleaved assignments to two slots
697    /// walk ONE counter, so any two holders can be ordered by `gen`.
698    #[tokio::test]
699    async fn the_generation_counter_is_shared_across_slots() {
700        let s = InMemoryRunStore::new();
701        s.create(mk("R-1", "T-1", 100)).await.unwrap();
702        let id = RunId::parse("R-1").unwrap();
703
704        let (first, _) = s
705            .acquire_assignee(&id, SLOT_A, "S-a1", "holds phase a")
706            .await
707            .unwrap();
708        let (second, _) = s
709            .acquire_assignee(&id, SLOT_B, "S-b2", "holds phase b")
710            .await
711            .unwrap();
712        let (third, _) = s
713            .acquire_assignee(&id, SLOT_A, "S-a3", "takes over phase a")
714            .await
715            .unwrap();
716
717        assert_eq!(
718            (first, second, third),
719            (1, 2, 3),
720            "a second slot does not start its own counter at 1"
721        );
722
723        let got = s.get(&id).await.unwrap();
724        assert_eq!(got.next_generation, 3);
725        assert_eq!(got.current[SLOT_A].gen, 3);
726        assert_eq!(got.current[SLOT_B].gen, 2);
727        assert!(
728            got.current[SLOT_A].gen > got.current[SLOT_B].gen,
729            "holders of different slots stay comparable by gen"
730        );
731    }
732
733    /// A4 (Vacant side): releasing bumps `G` too, so the next Assign picks
734    /// up from the bumped value rather than reusing the released one.
735    #[tokio::test]
736    async fn vacate_advances_generation_and_clears_the_holder() {
737        let s = InMemoryRunStore::new();
738        s.create(mk("R-1", "T-1", 100)).await.unwrap();
739        let id = RunId::parse("R-1").unwrap();
740
741        s.acquire_assignee(&id, SLOT_A, "S-a1", "first hold")
742            .await
743            .unwrap();
744        let outcome = s.vacate_assignee(&id, SLOT_A, 1).await.unwrap();
745        assert_eq!(
746            outcome,
747            VacateOutcome::Released {
748                generation: 2,
749                released: Assignee {
750                    op: "S-a1".into(),
751                    desc: "first hold".into(),
752                    gen: 1,
753                },
754            },
755            "A4: a Vacant that happens is an event and advances G"
756        );
757
758        let got = s.get(&id).await.unwrap();
759        assert!(
760            !got.current.contains_key(SLOT_A),
761            "R2: the Run stays, the holder does not"
762        );
763        assert_eq!(got.next_generation, 2);
764
765        let (gen, previous) = s
766            .acquire_assignee(&id, SLOT_A, "S-b2", "after release")
767            .await
768            .unwrap();
769        assert_eq!(gen, 3, "the next Assign continues from the bumped counter");
770        assert_eq!(
771            previous, None,
772            "nothing was displaced — the slot was Vacant"
773        );
774    }
775
776    /// A Vacant applies to the named slot only — the other seats keep the
777    /// holders they had.
778    #[tokio::test]
779    async fn vacate_releases_only_the_named_slot() {
780        let s = InMemoryRunStore::new();
781        s.create(mk("R-1", "T-1", 100)).await.unwrap();
782        let id = RunId::parse("R-1").unwrap();
783
784        s.acquire_assignee(&id, SLOT_A, "S-a1", "holds phase a")
785            .await
786            .unwrap();
787        s.acquire_assignee(&id, SLOT_B, "S-b2", "holds phase b")
788            .await
789            .unwrap();
790
791        let outcome = s.vacate_assignee(&id, SLOT_A, 1).await.unwrap();
792        assert!(
793            matches!(&outcome, VacateOutcome::Released { released, .. } if released.op == "S-a1"),
794            "got: {outcome:?}"
795        );
796
797        let got = s.get(&id).await.unwrap();
798        assert!(!got.current.contains_key(SLOT_A));
799        assert_eq!(
800            got.current[SLOT_B].op, "S-b2",
801            "vacating one seat must not empty another"
802        );
803    }
804
805    /// An already-Vacant slot holds no generation, so no release can match
806    /// it: the call is refused as stale and writes nothing — not even the
807    /// counter bump the unconditional verb used to make.
808    #[tokio::test]
809    async fn vacate_on_a_vacant_run_is_stale_and_writes_nothing() {
810        let s = InMemoryRunStore::new();
811        s.create(mk("R-1", "T-1", 100)).await.unwrap();
812        let id = RunId::parse("R-1").unwrap();
813        let outcome = s.vacate_assignee(&id, SLOT_A, 1).await.unwrap();
814        assert_eq!(outcome, VacateOutcome::Stale { current: None });
815        assert_eq!(
816            s.get(&id).await.unwrap().next_generation,
817            0,
818            "a release that did not release is not an assignment event"
819        );
820    }
821
822    /// The defect this verb exists for: a release issued against a
823    /// generation the seat no longer holds must leave the current holder
824    /// exactly where it is. A7 and O8's cascade both read a holder, await,
825    /// and only then release — an acquire landing in that window must win.
826    #[tokio::test]
827    async fn a_stale_release_does_not_disturb_the_current_holder() {
828        let s = InMemoryRunStore::new();
829        s.create(mk("R-1", "T-1", 100)).await.unwrap();
830        let id = RunId::parse("R-1").unwrap();
831
832        // What the releasing caller read.
833        let (observed_gen, _) = s
834            .acquire_assignee(&id, SLOT_A, "S-away", "the holder that went quiet")
835            .await
836            .unwrap();
837        // What landed while it was deciding (A8: acquire never excludes).
838        s.acquire_assignee(&id, SLOT_A, "S-fresh", "took the seat mid-decision")
839            .await
840            .unwrap();
841
842        let outcome = s.vacate_assignee(&id, SLOT_A, observed_gen).await.unwrap();
843        assert_eq!(
844            outcome,
845            VacateOutcome::Stale {
846                current: Some(Assignee {
847                    op: "S-fresh".into(),
848                    desc: "took the seat mid-decision".into(),
849                    gen: 2,
850                }),
851            },
852            "the stale reader is told who holds the seat now, and nothing is released"
853        );
854
855        let got = s.get(&id).await.unwrap();
856        assert_eq!(
857            got.current[SLOT_A].op, "S-fresh",
858            "the newer holder stands — A8 already decided this contest"
859        );
860        assert_eq!(got.current[SLOT_A].gen, 2);
861        assert_eq!(
862            got.next_generation, 2,
863            "the refused release burned no generation"
864        );
865    }
866
867    /// A3 / Q3: an acquire mints a NEW `Assignee`; a handle taken before it
868    /// still reads its original generation afterwards, and the displaced
869    /// holder is handed back with that same stamp.
870    #[tokio::test]
871    async fn acquire_never_rewrites_the_incumbent_assignee() {
872        let s = InMemoryRunStore::new();
873        s.create(mk("R-1", "T-1", 100)).await.unwrap();
874        let id = RunId::parse("R-1").unwrap();
875
876        s.acquire_assignee(&id, SLOT_A, "S-a1", "first hold")
877            .await
878            .unwrap();
879        let held_before = s.get(&id).await.unwrap().current[SLOT_A].clone();
880        assert_eq!(held_before.gen, 1);
881
882        let (_, displaced) = s
883            .acquire_assignee(&id, SLOT_A, "S-b2", "takeover")
884            .await
885            .unwrap();
886
887        assert_eq!(
888            held_before.gen, 1,
889            "A3: gen is immutable for the lifetime of an instance"
890        );
891        assert_eq!(
892            displaced.expect("displaced holder"),
893            held_before,
894            "Q3: the displaced instance is returned as-is, not mutated"
895        );
896    }
897
898    /// A8: acquire has no precondition on the slot's incumbent — the later
899    /// caller wins outright, no exclusion, no rejection.
900    #[tokio::test]
901    async fn acquire_displaces_a_live_holder() {
902        let s = InMemoryRunStore::new();
903        s.create(mk("R-1", "T-1", 100)).await.unwrap();
904        let id = RunId::parse("R-1").unwrap();
905
906        s.acquire_assignee(&id, SLOT_A, "S-a1", "first hold")
907            .await
908            .unwrap();
909        let (gen, displaced) = s
910            .acquire_assignee(&id, SLOT_A, "S-b2", "takeover")
911            .await
912            .unwrap();
913
914        assert_eq!(gen, 2);
915        assert_eq!(displaced.expect("displaced holder").op, "S-a1");
916        let got = s.get(&id).await.unwrap();
917        assert_eq!(got.current[SLOT_A].op, "S-b2", "last writer wins");
918        assert_eq!(got.current.len(), 1, "A1: still one holder for that slot");
919    }
920
921    /// A9: `desc` is mandatory, and so is the slot. A rejected acquire must
922    /// not have burned a generation or disturbed the incumbent.
923    #[tokio::test]
924    async fn acquire_rejects_a_missing_desc_without_side_effects() {
925        let s = InMemoryRunStore::new();
926        s.create(mk("R-1", "T-1", 100)).await.unwrap();
927        let id = RunId::parse("R-1").unwrap();
928        s.acquire_assignee(&id, SLOT_A, "S-a1", "first hold")
929            .await
930            .unwrap();
931
932        for blank in ["", "   "] {
933            let err = s
934                .acquire_assignee(&id, SLOT_A, "S-b2", blank)
935                .await
936                .unwrap_err();
937            assert!(
938                matches!(err, RunStoreError::AssigneeDescRequired),
939                "got: {err:?}"
940            );
941        }
942
943        let err = s
944            .acquire_assignee(&id, "", "S-b2", "no slot named")
945            .await
946            .unwrap_err();
947        assert!(
948            matches!(err, RunStoreError::AssigneeSlotRequired),
949            "got: {err:?}"
950        );
951        let err = s.vacate_assignee(&id, "", 1).await.unwrap_err();
952        assert!(
953            matches!(err, RunStoreError::AssigneeSlotRequired),
954            "got: {err:?}"
955        );
956
957        let got = s.get(&id).await.unwrap();
958        assert_eq!(got.next_generation, 1, "a refused event is not an event");
959        assert_eq!(got.current[SLOT_A].op, "S-a1");
960    }
961
962    /// Concurrent acquires must never read the same `G`, so no two callers
963    /// can be told they hold the same generation — including when they name
964    /// different slots, since the counter is Run-wide.
965    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
966    async fn concurrent_acquires_hand_out_distinct_generations() {
967        let s = std::sync::Arc::new(InMemoryRunStore::new());
968        s.create(mk("R-1", "T-1", 100)).await.unwrap();
969
970        let mut handles = Vec::new();
971        for i in 0..8u32 {
972            let s = s.clone();
973            let slot = if i % 2 == 0 { SLOT_A } else { SLOT_B };
974            handles.push(tokio::spawn(async move {
975                s.acquire_assignee(
976                    &RunId::parse("R-1").unwrap(),
977                    slot,
978                    &format!("S-{i}"),
979                    "concurrent hold",
980                )
981                .await
982                .unwrap()
983                .0
984            }));
985        }
986        let mut generations = Vec::new();
987        for h in handles {
988            generations.push(h.await.unwrap());
989        }
990        generations.sort_unstable();
991        assert_eq!(generations, (1..=8).collect::<Vec<u64>>());
992        assert_eq!(
993            s.get(&RunId::parse("R-1").unwrap())
994                .await
995                .unwrap()
996                .next_generation,
997            8
998        );
999    }
1000
1001    #[tokio::test]
1002    async fn assignment_on_an_unknown_run_fails() {
1003        let s = InMemoryRunStore::new();
1004        let missing = RunId::parse("R-nope").unwrap();
1005        let err = s
1006            .acquire_assignee(&missing, SLOT_A, "S-a1", "hold")
1007            .await
1008            .unwrap_err();
1009        assert!(matches!(err, RunStoreError::NotFound(_)), "got: {err:?}");
1010        let err = s.vacate_assignee(&missing, SLOT_A, 1).await.unwrap_err();
1011        assert!(matches!(err, RunStoreError::NotFound(_)), "got: {err:?}");
1012    }
1013
1014    #[tokio::test]
1015    async fn input_json_roundtrips_through_create_get() {
1016        let s = InMemoryRunStore::new();
1017        let mut rec = mk("R-1", "T-1", 100);
1018        rec.input_json = Some(r#"{"blueprint":"snapshot"}"#.to_string());
1019        s.create(rec).await.unwrap();
1020        let got = s.get(&RunId::parse("R-1").unwrap()).await.unwrap();
1021        assert_eq!(
1022            got.input_json.as_deref(),
1023            Some(r#"{"blueprint":"snapshot"}"#)
1024        );
1025    }
1026}
mlua_swarm/store/run/inmemory.rs

mlua_swarm/store/run/
inmemory.rs
