sim_lib_journal/
memory.rs

1use crate::{
2    Admission, JournalBackend, JournalError, JournalHead, JournalObject, Lease, StoredDatumRef,
3    StoredState,
4};
5use sim_kernel::{ContentId, Datum};
6use std::sync::{Mutex, PoisonError};
7
8#[derive(Default)]
9struct Inner {
10    fence: u64,
11    state: StoredState,
12}
13
14/// Deterministic law-reference backend. It is deliberately not durable.
15#[derive(Default)]
16pub struct MemoryBackend {
17    inner: Mutex<Inner>,
18}
19
20impl MemoryBackend {
21    pub fn new() -> Self {
22        Self::default()
23    }
24}
25
26impl JournalBackend for MemoryBackend {
27    fn acquire_lease(&self) -> Result<Lease, JournalError> {
28        let mut inner = self
29            .inner
30            .lock()
31            .map_err(|_: PoisonError<_>| JournalError::Backend("memory lock poisoned".into()))?;
32        inner.fence = inner
33            .fence
34            .checked_add(1)
35            .ok_or_else(|| JournalError::Backend("fence exhausted".into()))?;
36        Ok(Lease { fence: inner.fence })
37    }
38    fn read_state(&self) -> Result<StoredState, JournalError> {
39        Ok(self
40            .inner
41            .lock()
42            .map_err(|_: PoisonError<_>| JournalError::Backend("memory lock poisoned".into()))?
43            .state
44            .clone())
45    }
46    fn admit(&self, admission: Admission) -> Result<JournalHead, JournalError> {
47        let mut inner = self
48            .inner
49            .lock()
50            .map_err(|_: PoisonError<_>| JournalError::Backend("memory lock poisoned".into()))?;
51        if admission.fence != inner.fence {
52            return Err(JournalError::StaleLease);
53        }
54        if admission.expected != inner.state.head {
55            // A caller may retry an exactly committed batch after losing its
56            // acknowledgement. This is the sole stale-head exception.
57            let exact = admission
58                .entries
59                .iter()
60                .all(|entry| inner.state.entries.get(&entry.sequence) == Some(entry))
61                && !admission.entries.is_empty();
62            if exact {
63                return inner.state.head.clone().ok_or(JournalError::WrongHead);
64            }
65            if admission
66                .entries
67                .iter()
68                .any(|entry| inner.state.entries.contains_key(&entry.sequence))
69            {
70                return Err(JournalError::ConflictingDelivery);
71            }
72            return Err(JournalError::WrongHead);
73        }
74        for object in admission.objects {
75            match inner.state.objects.get(&object.id) {
76                Some(bytes) if bytes != &object.bytes => {
77                    return Err(JournalError::ConflictingObject);
78                }
79                Some(_) => {}
80                None => {
81                    inner
82                        .state
83                        .datums
84                        .insert(object.id.clone(), object.datum().clone());
85                    inner.state.objects.insert(object.id, object.bytes);
86                }
87            }
88        }
89        for entry in admission.entries {
90            match inner.state.entries.get(&entry.sequence) {
91                Some(existing) if existing == &entry => {}
92                Some(_) => return Err(JournalError::ConflictingDelivery),
93                None => {
94                    inner.state.entries.insert(entry.sequence, entry);
95                }
96            }
97        }
98        let entry = inner
99            .state
100            .entries
101            .last_key_value()
102            .ok_or(JournalError::EmptyBatch)?
103            .1;
104        let head = JournalHead {
105            sequence: entry.sequence,
106            entry: entry.id.clone(),
107        };
108        inner.state.head = Some(head.clone());
109        Ok(head)
110    }
111
112    fn put_datum(&self, object: JournalObject) -> Result<StoredDatumRef, JournalError> {
113        object.verify()?;
114        let storage_bytes = object.storage_bytes()?;
115        let reference = StoredDatumRef {
116            meaning: object.id.clone(),
117            storage: crate::object::storage_id(&storage_bytes),
118        };
119        let mut inner = self
120            .inner
121            .lock()
122            .map_err(|_: PoisonError<_>| JournalError::Backend("memory lock poisoned".into()))?;
123        match inner.state.datums.get(&object.id) {
124            Some(value) if value != object.datum() => return Err(JournalError::ConflictingObject),
125            _ => {
126                let datum = object.datum().clone();
127                inner.state.objects.insert(object.id.clone(), object.bytes);
128                inner.state.datums.insert(object.id, datum);
129            }
130        }
131        Ok(reference)
132    }
133
134    fn get_datum(&self, meaning: &ContentId) -> Result<Datum, JournalError> {
135        self.inner
136            .lock()
137            .map_err(|_: PoisonError<_>| JournalError::Backend("memory lock poisoned".into()))?
138            .state
139            .datums
140            .get(meaning)
141            .cloned()
142            .ok_or_else(|| JournalError::MissingSemanticObject(meaning.clone()))
143    }
144
145    fn rebuild_datum_index(&self) -> Result<Vec<StoredDatumRef>, JournalError> {
146        let inner = self
147            .inner
148            .lock()
149            .map_err(|_: PoisonError<_>| JournalError::Backend("memory lock poisoned".into()))?;
150        inner
151            .state
152            .datums
153            .iter()
154            .map(|(meaning, datum)| {
155                let object = JournalObject::from_datum(datum.clone())?;
156                let bytes = object.storage_bytes()?;
157                Ok(StoredDatumRef {
158                    meaning: meaning.clone(),
159                    storage: crate::object::storage_id(&bytes),
160                })
161            })
162            .collect()
163    }
164}
sim_lib_journal/memory.rs

sim_lib_journal/
memory.rs
