mlua_swarm/store/operator_session/
inmemory.rs

1//! `InMemoryOperatorSessionStore` — a process-volatile
2//! [`OperatorSessionStore`] used as the default when no store path is
3//! configured. Byte-for-byte the pre-persistence behaviour: sessions die
4//! with the process.
5
6use super::{
7    Inner, OperatorSessionRecord, OperatorSessionStore, OperatorSessionStoreError, SessionId,
8    SharedInner,
9};
10use async_trait::async_trait;
11use std::sync::Mutex;
12
13/// Process-volatile [`OperatorSessionStore`] default backend.
14#[derive(Default)]
15pub struct InMemoryOperatorSessionStore {
16    inner: SharedInner,
17}
18
19impl InMemoryOperatorSessionStore {
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 OperatorSessionStore for InMemoryOperatorSessionStore {
30    fn name(&self) -> &str {
31        "in-memory"
32    }
33
34    async fn put(&self, record: OperatorSessionRecord) -> Result<(), OperatorSessionStoreError> {
35        let mut inner = self.inner.lock().unwrap();
36        if !inner.records.contains_key(&record.sid) {
37            inner.order.push(record.sid.clone());
38        }
39        inner.records.insert(record.sid.clone(), record);
40        Ok(())
41    }
42
43    async fn delete(&self, sid: &SessionId) -> Result<(), OperatorSessionStoreError> {
44        let mut inner = self.inner.lock().unwrap();
45        if inner.records.remove(sid).is_none() {
46            return Err(OperatorSessionStoreError::NotFound(sid.clone()));
47        }
48        inner.order.retain(|s| s != sid);
49        Ok(())
50    }
51
52    /// Unfiltered by design (see the trait's contract): this backend's map
53    /// is the whole of what it stores, so handing back what is under the
54    /// key is exactly "the row as stored".
55    async fn get(
56        &self,
57        sid: &SessionId,
58    ) -> Result<Option<OperatorSessionRecord>, OperatorSessionStoreError> {
59        let inner = self.inner.lock().unwrap();
60        Ok(inner.records.get(sid).cloned())
61    }
62
63    /// Expired rows are dropped from the answer **and** from the map (see
64    /// the trait's contract). This backend can do both under the one lock
65    /// it already takes, so the removal is atomic with the read that
66    /// judged it.
67    async fn list(&self) -> Result<Vec<OperatorSessionRecord>, OperatorSessionStoreError> {
68        let mut inner = self.inner.lock().unwrap();
69        let mut records: Vec<OperatorSessionRecord> = inner
70            .order
71            .iter()
72            .filter_map(|sid| inner.records.get(sid).cloned())
73            .collect();
74        records.sort_by_key(|r| r.joined_at_secs);
75        let (live, expired) = super::partition_expired(records, super::expiry_now(), "in-memory");
76        for sid in expired {
77            inner.records.remove(&sid);
78            inner.order.retain(|s| s != &sid);
79        }
80        Ok(live)
81    }
82}
83
84// ──────────────────────────────────────────────────────────────────────────
85// tests
86// ──────────────────────────────────────────────────────────────────────────
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91
92    /// A live record joined at `joined_at_secs`.
93    ///
94    /// The join time stays a small literal — the tests below are about
95    /// ordering by it — while the access clock is set to now, because
96    /// `list()` deletes what the 24h horizon has expired and a record last accessed
97    /// at second 100 of 1970 is expired by any real clock.
98    fn mk(sid: &str, joined_at_secs: u64) -> OperatorSessionRecord {
99        OperatorSessionRecord {
100            sid: SessionId::parse(sid).unwrap(),
101            token_digest: OperatorSessionRecord::digest_of(&format!("bearer-{sid}")),
102            capability_manifest: None,
103            joined_at_secs,
104            last_access_secs: super::super::expiry_now(),
105            desc: None,
106            observed: Vec::new(),
107            observed_total: 0,
108        }
109    }
110
111    /// The in-memory backend holds live records, so the 記名 an `Assign`
112    /// wrote is what the next `list()` reports — no encode/decode in
113    /// between to lose it.
114    #[tokio::test]
115    async fn the_kimei_round_trips() {
116        let s = InMemoryOperatorSessionStore::new();
117        let mut rec = mk("S-1", 100);
118        rec.desc = Some("rewriting the seat resolver in mlua-swarm-server".to_string());
119        rec.record_observed(super::super::ObservedAssignment::new(
120            "R-1".to_string(),
121            "phase-a-op".to_string(),
122            Some("resolve issue #10".to_string()),
123            Some("/repo".to_string()),
124            None,
125            None,
126            140,
127        ));
128        s.put(rec).await.unwrap();
129
130        let list = s.list().await.unwrap();
131        assert_eq!(
132            list[0].desc.as_deref(),
133            Some("rewriting the seat resolver in mlua-swarm-server")
134        );
135        assert_eq!(list[0].observed.len(), 1);
136        assert_eq!(list[0].observed_total, 1);
137        assert_eq!(list[0].last_activity_secs(), 140);
138    }
139
140    #[tokio::test]
141    async fn put_then_list() {
142        let s = InMemoryOperatorSessionStore::new();
143        s.put(mk("S-1", 100)).await.unwrap();
144        s.put(mk("S-2", 50)).await.unwrap();
145        let list = s.list().await.unwrap();
146        let sids: Vec<_> = list.iter().map(|r| r.sid.to_string()).collect();
147        assert_eq!(sids, vec!["S-2", "S-1"], "ascending by joined_at_secs");
148    }
149
150    #[tokio::test]
151    async fn put_is_upsert() {
152        let s = InMemoryOperatorSessionStore::new();
153        s.put(mk("S-1", 100)).await.unwrap();
154        let mut updated = mk("S-1", 100);
155        updated.desc = Some("the same session, re-put".to_string());
156        s.put(updated).await.unwrap();
157        let list = s.list().await.unwrap();
158        assert_eq!(list.len(), 1);
159        assert_eq!(list[0].desc.as_deref(), Some("the same session, re-put"));
160    }
161
162    #[tokio::test]
163    async fn verify_bearer_accepts_only_the_minted_bearer() {
164        let record = mk("S-1", 100);
165        assert!(record.verify_bearer("bearer-S-1"));
166        assert!(!record.verify_bearer("bearer-S-2"));
167        assert!(!record.verify_bearer(""));
168        // The stored value is the digest, not the bearer.
169        assert_ne!(record.token_digest, "bearer-S-1");
170        assert_eq!(record.token_digest.len(), 64, "hex SHA-256");
171    }
172
173    #[tokio::test]
174    async fn delete_removes_and_missing_is_not_found() {
175        let s = InMemoryOperatorSessionStore::new();
176        s.put(mk("S-1", 100)).await.unwrap();
177        s.delete(&SessionId::parse("S-1").unwrap()).await.unwrap();
178        assert!(s.list().await.unwrap().is_empty());
179        let err = s
180            .delete(&SessionId::parse("S-1").unwrap())
181            .await
182            .unwrap_err();
183        assert!(matches!(err, OperatorSessionStoreError::NotFound(_)));
184    }
185
186    #[tokio::test]
187    async fn name_is_in_memory() {
188        assert_eq!(InMemoryOperatorSessionStore::new().name(), "in-memory");
189    }
190}
mlua_swarm/store/operator_session/inmemory.rs

mlua_swarm/store/operator_session/
inmemory.rs
