sim_lib_journal/
host.rs

1use crate::{
2    Admission, JournalBackend, JournalEntry, JournalError, JournalHead, Lease, StoredState,
3};
4use sim_kernel::{ContentId, Symbol};
5use sim_storage_port::{HostDirErrorKind, HostDirPort, NeverCancel};
6use std::{collections::BTreeMap, sync::Arc};
7
8/// Evidence supplied by the concrete Table/Dir binding at construction.
9#[derive(Clone, Copy, Debug, Eq, PartialEq)]
10pub struct BackendCapabilities {
11    /// State-leaf compare-exchange is linearizable across coordinators.
12    pub linearizable_cas: bool,
13    /// Successful immutable-leaf writes carry the binding's durability receipt.
14    pub durable_publish: bool,
15}
16
17/// Stable crash-injection boundaries in the publication protocol.
18#[derive(Clone, Copy, Debug, Eq, PartialEq)]
19pub enum Failpoint {
20    BeforeObjectPublish,
21    AfterObjectPublish,
22    AfterDurabilityReceipt,
23    BeforeCas,
24    AfterCas,
25    BeforeAcknowledgement,
26}
27
28impl Failpoint {
29    fn label(self) -> &'static str {
30        match self {
31            Self::BeforeObjectPublish => "before-object-publish",
32            Self::AfterObjectPublish => "after-object-publish",
33            Self::AfterDurabilityReceipt => "after-durability-receipt",
34            Self::BeforeCas => "before-cas",
35            Self::AfterCas => "after-cas",
36            Self::BeforeAcknowledgement => "before-acknowledgement",
37        }
38    }
39}
40
41type FailHook = Arc<dyn Fn(Failpoint) -> bool + Send + Sync>;
42
43/// Crash-durable journal composition over the canonical host-backed Table/Dir port.
44///
45/// `state` contains both fence and head. Its one compare-exchange is the only
46/// publication linearization point. Objects and entries are immutable leaves.
47pub struct HostDirJournalBackend {
48    port: Arc<dyn HostDirPort>,
49    capabilities: BackendCapabilities,
50    work_bound: usize,
51    fail: Option<FailHook>,
52}
53
54impl HostDirJournalBackend {
55    /// Opens a binding. Read operations remain available when write safety is absent.
56    pub fn open(
57        port: Arc<dyn HostDirPort>,
58        capabilities: BackendCapabilities,
59        work_bound: usize,
60    ) -> Result<Self, JournalError> {
61        if work_bound == 0 {
62            return Err(JournalError::WorkBoundExceeded);
63        }
64        let backend = Self {
65            port,
66            capabilities,
67            work_bound,
68            fail: None,
69        };
70        backend.read_state()?; // full chain and content-closure verification before exposure
71        Ok(backend)
72    }
73
74    /// Installs a deterministic test hook. Returning true simulates process death.
75    pub fn with_failpoint_hook(
76        mut self,
77        hook: impl Fn(Failpoint) -> bool + Send + Sync + 'static,
78    ) -> Self {
79        self.fail = Some(Arc::new(hook));
80        self
81    }
82
83    /// Reports the binding evidence used to admit or refuse write mode.
84    pub fn capabilities(&self) -> BackendCapabilities {
85        self.capabilities
86    }
87
88    fn trip(&self, point: Failpoint) -> Result<(), JournalError> {
89        if self.fail.as_ref().is_some_and(|hook| hook(point)) {
90            Err(JournalError::InjectedCrash(point.label()))
91        } else {
92            Ok(())
93        }
94    }
95
96    fn write_capable(&self) -> Result<(), JournalError> {
97        if !self.capabilities.linearizable_cas {
98            return Err(JournalError::WriteRefused(
99                "linearizable table/cas unavailable",
100            ));
101        }
102        if !self.capabilities.durable_publish {
103            return Err(JournalError::WriteRefused(
104                "storage durability receipt unavailable",
105            ));
106        }
107        Ok(())
108    }
109
110    fn ensure_layout(&self) -> Result<(), JournalError> {
111        for path in [["objects"], ["entries"], ["temporary"]] {
112            self.port
113                .create_dir(&path.map(str::to_owned))
114                .map_err(port_error)?;
115        }
116        Ok(())
117    }
118
119    fn state_bytes(&self) -> Result<Option<Vec<u8>>, JournalError> {
120        match self.port.read(&["state".into()]) {
121            Ok(bytes) => Ok(Some(bytes)),
122            Err(e) if e.kind == HostDirErrorKind::NotFound => Ok(None),
123            Err(e) => Err(port_error(e)),
124        }
125    }
126
127    fn put_immutable(&self, path: &[String], bytes: &[u8]) -> Result<(), JournalError> {
128        let outcome = self
129            .port
130            .compare_exchange(path, None, Some(bytes), &NeverCancel)
131            .map_err(port_error)?;
132        if outcome.exchanged {
133            return Ok(());
134        }
135        if outcome.observed.as_deref() == Some(bytes) {
136            Ok(())
137        } else {
138            Err(JournalError::ConflictingObject)
139        }
140    }
141
142    fn load(&self, path: &[String]) -> Result<Vec<u8>, JournalError> {
143        self.port.read(path).map_err(port_error)
144    }
145}
146
147impl JournalBackend for HostDirJournalBackend {
148    fn acquire_lease(&self) -> Result<Lease, JournalError> {
149        self.write_capable()?;
150        self.ensure_layout()?;
151        loop {
152            let observed = self.state_bytes()?;
153            let (fence, head) = observed
154                .as_deref()
155                .map(decode_state)
156                .transpose()?
157                .unwrap_or((0, None));
158            let next = fence
159                .checked_add(1)
160                .ok_or_else(|| JournalError::Backend("fence exhausted".into()))?;
161            let replacement = encode_state(next, head.as_ref());
162            let result = self
163                .port
164                .compare_exchange(
165                    &["state".into()],
166                    observed.as_deref(),
167                    Some(&replacement),
168                    &NeverCancel,
169                )
170                .map_err(port_error)?;
171            if result.exchanged {
172                return Ok(Lease { fence: next });
173            }
174        }
175    }
176
177    fn read_state(&self) -> Result<StoredState, JournalError> {
178        let (_, head) = self
179            .state_bytes()?
180            .as_deref()
181            .map(decode_state)
182            .transpose()?
183            .unwrap_or((0, None));
184        let Some(head) = head else {
185            return Ok(StoredState::default());
186        };
187        let needed = (head.sequence as usize)
188            .checked_add(1)
189            .ok_or(JournalError::WorkBoundExceeded)?;
190        if needed > self.work_bound {
191            return Err(JournalError::WorkBoundExceeded);
192        }
193        let mut entries = BTreeMap::new();
194        let mut objects = BTreeMap::new();
195        for sequence in 0..=head.sequence {
196            let bytes = self.load(&entry_path(sequence))?;
197            let entry = decode_entry(&bytes)?;
198            if entry.sequence != sequence {
199                return Err(JournalError::CorruptState("entry location"));
200            }
201            for id in &entry.payloads {
202                if !objects.contains_key(id) {
203                    if entries.len() + objects.len() >= self.work_bound {
204                        return Err(JournalError::WorkBoundExceeded);
205                    }
206                    objects.insert(id.clone(), self.load(&object_path(id))?);
207                }
208            }
209            entries.insert(sequence, entry);
210        }
211        let state = StoredState {
212            objects,
213            entries,
214            head: Some(head),
215        };
216        crate::verify::verify_state(&state)?;
217        Ok(state)
218    }
219
220    fn admit(&self, admission: Admission) -> Result<JournalHead, JournalError> {
221        self.write_capable()?;
222        self.ensure_layout()?;
223        let observed = self.state_bytes()?;
224        let (fence, head) = observed
225            .as_deref()
226            .map(decode_state)
227            .transpose()?
228            .unwrap_or((0, None));
229        if fence != admission.fence {
230            return Err(JournalError::StaleLease);
231        }
232        if head != admission.expected {
233            return Err(JournalError::WrongHead);
234        }
235        self.trip(Failpoint::BeforeObjectPublish)?;
236        for object in &admission.objects {
237            object.verify()?;
238            self.put_immutable(&object_path(&object.id), &object.bytes)?;
239        }
240        self.trip(Failpoint::AfterObjectPublish)?;
241        for entry in &admission.entries {
242            self.put_immutable(&entry_path(entry.sequence), &encode_entry(entry))?;
243        }
244        self.trip(Failpoint::AfterDurabilityReceipt)?;
245        let last = admission.entries.last().ok_or(JournalError::EmptyBatch)?;
246        let new_head = JournalHead {
247            sequence: last.sequence,
248            entry: last.id.clone(),
249        };
250        self.trip(Failpoint::BeforeCas)?;
251        let replacement = encode_state(fence, Some(&new_head));
252        let result = self
253            .port
254            .compare_exchange(
255                &["state".into()],
256                observed.as_deref(),
257                Some(&replacement),
258                &NeverCancel,
259            )
260            .map_err(port_error)?;
261        if !result.exchanged {
262            return Err(JournalError::WrongHead);
263        }
264        self.trip(Failpoint::AfterCas)?;
265        self.trip(Failpoint::BeforeAcknowledgement)?;
266        Ok(new_head)
267    }
268}
269
270fn port_error(error: sim_storage_port::HostDirError) -> JournalError {
271    JournalError::Backend(error.to_string())
272}
273fn entry_path(sequence: u64) -> Vec<String> {
274    vec!["entries".into(), format!("{sequence:016x}")]
275}
276fn object_path(id: &ContentId) -> Vec<String> {
277    vec!["objects".into(), hex(&id.bytes)]
278}
279fn hex(bytes: &[u8]) -> String {
280    bytes.iter().map(|b| format!("{b:02x}")).collect()
281}
282
283fn encode_state(fence: u64, head: Option<&JournalHead>) -> Vec<u8> {
284    let mut out = b"SIMJSTATE1".to_vec();
285    out.extend(fence.to_be_bytes());
286    match head {
287        Some(h) => {
288            out.push(1);
289            out.extend(h.sequence.to_be_bytes());
290            put_id(&mut out, &h.entry);
291        }
292        None => out.push(0),
293    }
294    out
295}
296fn decode_state(bytes: &[u8]) -> Result<(u64, Option<JournalHead>), JournalError> {
297    let mut c = Cursor::new(bytes);
298    if c.take(10)? != b"SIMJSTATE1" {
299        return Err(JournalError::CorruptState("state format"));
300    }
301    let fence = c.u64()?;
302    let head = match c.byte()? {
303        0 => None,
304        1 => Some(JournalHead {
305            sequence: c.u64()?,
306            entry: c.id()?,
307        }),
308        _ => return Err(JournalError::CorruptState("state tag")),
309    };
310    c.end()?;
311    Ok((fence, head))
312}
313fn encode_entry(entry: &JournalEntry) -> Vec<u8> {
314    let mut out = b"SIMJENTRY1".to_vec();
315    put_id(&mut out, &entry.id);
316    out.extend(entry.sequence.to_be_bytes());
317    match &entry.previous {
318        Some(id) => {
319            out.push(1);
320            put_id(&mut out, id)
321        }
322        None => out.push(0),
323    };
324    put_text(&mut out, &entry.kind.as_qualified_str());
325    out.extend((entry.payloads.len() as u32).to_be_bytes());
326    for id in &entry.payloads {
327        put_id(&mut out, id);
328    }
329    out
330}
331fn decode_entry(bytes: &[u8]) -> Result<JournalEntry, JournalError> {
332    let mut c = Cursor::new(bytes);
333    if c.take(10)? != b"SIMJENTRY1" {
334        return Err(JournalError::CorruptState("entry format"));
335    }
336    let id = c.id()?;
337    let sequence = c.u64()?;
338    let previous = match c.byte()? {
339        0 => None,
340        1 => Some(c.id()?),
341        _ => return Err(JournalError::CorruptState("entry tag")),
342    };
343    let kind = parse_symbol(&c.text()?)?;
344    let count = c.u32()? as usize;
345    let mut payloads = Vec::with_capacity(count);
346    for _ in 0..count {
347        payloads.push(c.id()?);
348    }
349    c.end()?;
350    Ok(JournalEntry {
351        id,
352        sequence,
353        previous,
354        kind,
355        payloads,
356    })
357}
358fn put_id(out: &mut Vec<u8>, id: &ContentId) {
359    put_text(out, &id.algorithm.as_qualified_str());
360    out.extend(id.bytes)
361}
362fn put_text(out: &mut Vec<u8>, text: &str) {
363    out.extend((text.len() as u32).to_be_bytes());
364    out.extend(text.as_bytes())
365}
366fn parse_symbol(text: &str) -> Result<Symbol, JournalError> {
367    match text.split_once('/') {
368        Some((n, v)) if !n.is_empty() && !v.is_empty() => Ok(Symbol::qualified(n, v)),
369        None => Symbol::checked(text).map_err(|_| JournalError::CorruptState("symbol")),
370        _ => Err(JournalError::CorruptState("symbol")),
371    }
372}
373struct Cursor<'a> {
374    bytes: &'a [u8],
375    at: usize,
376}
377impl<'a> Cursor<'a> {
378    fn new(bytes: &'a [u8]) -> Self {
379        Self { bytes, at: 0 }
380    }
381    fn take(&mut self, n: usize) -> Result<&'a [u8], JournalError> {
382        let end = self
383            .at
384            .checked_add(n)
385            .ok_or(JournalError::CorruptState("length"))?;
386        let value = self
387            .bytes
388            .get(self.at..end)
389            .ok_or(JournalError::CorruptState("truncated"))?;
390        self.at = end;
391        Ok(value)
392    }
393    fn byte(&mut self) -> Result<u8, JournalError> {
394        Ok(self.take(1)?[0])
395    }
396    fn u32(&mut self) -> Result<u32, JournalError> {
397        Ok(u32::from_be_bytes(self.take(4)?.try_into().unwrap()))
398    }
399    fn u64(&mut self) -> Result<u64, JournalError> {
400        Ok(u64::from_be_bytes(self.take(8)?.try_into().unwrap()))
401    }
402    fn text(&mut self) -> Result<String, JournalError> {
403        let n = self.u32()? as usize;
404        String::from_utf8(self.take(n)?.to_vec()).map_err(|_| JournalError::CorruptState("utf8"))
405    }
406    fn id(&mut self) -> Result<ContentId, JournalError> {
407        let algorithm = parse_symbol(&self.text()?)?;
408        let bytes = self.take(32)?.try_into().unwrap();
409        Ok(ContentId::from_bytes(algorithm, bytes))
410    }
411    fn end(&self) -> Result<(), JournalError> {
412        if self.at == self.bytes.len() {
413            Ok(())
414        } else {
415            Err(JournalError::CorruptState("trailing bytes"))
416        }
417    }
418}
sim_lib_journal/host.rs

sim_lib_journal/
host.rs
