sim_lib_journal/
memory.rs

1use crate::{Admission, JournalBackend, JournalError, JournalHead, Lease, StoredState};
2use std::sync::{Mutex, PoisonError};
3
4#[derive(Default)]
5struct Inner {
6    fence: u64,
7    state: StoredState,
8}
9
10/// Deterministic law-reference backend. It is deliberately not durable.
11#[derive(Default)]
12pub struct MemoryBackend {
13    inner: Mutex<Inner>,
14}
15
16impl MemoryBackend {
17    pub fn new() -> Self {
18        Self::default()
19    }
20}
21
22impl JournalBackend for MemoryBackend {
23    fn acquire_lease(&self) -> Result<Lease, JournalError> {
24        let mut inner = self
25            .inner
26            .lock()
27            .map_err(|_: PoisonError<_>| JournalError::Backend("memory lock poisoned".into()))?;
28        inner.fence = inner
29            .fence
30            .checked_add(1)
31            .ok_or_else(|| JournalError::Backend("fence exhausted".into()))?;
32        Ok(Lease { fence: inner.fence })
33    }
34    fn read_state(&self) -> Result<StoredState, JournalError> {
35        Ok(self
36            .inner
37            .lock()
38            .map_err(|_: PoisonError<_>| JournalError::Backend("memory lock poisoned".into()))?
39            .state
40            .clone())
41    }
42    fn admit(&self, admission: Admission) -> Result<JournalHead, JournalError> {
43        let mut inner = self
44            .inner
45            .lock()
46            .map_err(|_: PoisonError<_>| JournalError::Backend("memory lock poisoned".into()))?;
47        if admission.fence != inner.fence {
48            return Err(JournalError::StaleLease);
49        }
50        if admission.expected != inner.state.head {
51            // A caller may retry an exactly committed batch after losing its
52            // acknowledgement. This is the sole stale-head exception.
53            let exact = admission
54                .entries
55                .iter()
56                .all(|entry| inner.state.entries.get(&entry.sequence) == Some(entry))
57                && !admission.entries.is_empty();
58            if exact {
59                return inner.state.head.clone().ok_or(JournalError::WrongHead);
60            }
61            if admission
62                .entries
63                .iter()
64                .any(|entry| inner.state.entries.contains_key(&entry.sequence))
65            {
66                return Err(JournalError::ConflictingDelivery);
67            }
68            return Err(JournalError::WrongHead);
69        }
70        for object in admission.objects {
71            match inner.state.objects.get(&object.id) {
72                Some(bytes) if bytes != &object.bytes => {
73                    return Err(JournalError::ConflictingObject);
74                }
75                Some(_) => {}
76                None => {
77                    inner.state.objects.insert(object.id, object.bytes);
78                }
79            }
80        }
81        for entry in admission.entries {
82            match inner.state.entries.get(&entry.sequence) {
83                Some(existing) if existing == &entry => {}
84                Some(_) => return Err(JournalError::ConflictingDelivery),
85                None => {
86                    inner.state.entries.insert(entry.sequence, entry);
87                }
88            }
89        }
90        let entry = inner
91            .state
92            .entries
93            .last_key_value()
94            .ok_or(JournalError::EmptyBatch)?
95            .1;
96        let head = JournalHead {
97            sequence: entry.sequence,
98            entry: entry.id.clone(),
99        };
100        inner.state.head = Some(head.clone());
101        Ok(head)
102    }
103}
sim_lib_journal/memory.rs

sim_lib_journal/
memory.rs
