sim_lib_journal/
native_codec.rs

1use crate::{JournalEntry, JournalError, JournalHead, StoredDatumRef};
2use sha2::{Digest, Sha256};
3use sim_kernel::{ContentId, Datum, Symbol};
4use sim_storage_port::HostDirPort;
5
6/// Evidence supplied by the concrete Table/Dir binding at construction.
7#[derive(Clone, Copy, Debug, Eq, PartialEq)]
8pub struct BackendCapabilities {
9    /// State-leaf compare-exchange is linearizable across coordinators.
10    pub linearizable_cas: bool,
11    /// Successful immutable-leaf writes carry the binding's durability receipt.
12    pub durable_publish: bool,
13}
14
15/// Native physical state version. It never enters a semantic entry identity.
16#[derive(Clone, Copy, Debug, Eq, PartialEq)]
17pub enum NativeFormatId {
18    /// Retained `SIMJSTATE1`/`SIMJENTRY1` prefix.
19    V1,
20    /// Canonical semantic entries in isolated content-addressed leaves.
21    V2,
22}
23
24/// Backend-allocated routing token for one v2 entry namespace.
25#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
26pub struct EntryNamespace(pub String);
27
28/// Exact physical locator for a committed v2 entry leaf.
29#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
30pub struct EntryLocation {
31    pub namespace: EntryNamespace,
32    pub sequence: u64,
33    pub entry: ContentId,
34    pub storage: ContentId,
35}
36
37/// Durable proof reference for the immutable v1 committed prefix.
38#[derive(Clone, Debug, Eq, PartialEq)]
39pub struct VerifiedNativePrefixRef {
40    pub entries: u64,
41    pub physical_head: JournalHead,
42    pub canonical_head: JournalHead,
43    pub descriptor: ContentId,
44}
45
46/// The one physical state selected by the native state CAS.
47#[derive(Clone, Debug, Eq, PartialEq)]
48pub struct NativeStateEnvelope {
49    pub format: NativeFormatId,
50    pub fence: u64,
51    pub namespace: EntryNamespace,
52    pub prefix: Option<VerifiedNativePrefixRef>,
53    pub head: Option<JournalHead>,
54    pub head_location: Option<EntryLocation>,
55}
56
57/// Stable crash-injection boundaries in upgrade and append publication.
58#[derive(Clone, Copy, Debug, Eq, PartialEq)]
59pub enum Failpoint {
60    BeforePrefixDescriptor,
61    AfterPrefixDescriptor,
62    AfterNamespaceReservation,
63    BeforeFormatCas,
64    AfterFormatCas,
65    BeforeObjectPublish,
66    AfterObjectPublish,
67    AfterDurabilityReceipt,
68    BeforeCas,
69    AfterCas,
70    BeforeAcknowledgement,
71}
72
73impl Failpoint {
74    pub(crate) fn label(self) -> &'static str {
75        match self {
76            Self::BeforePrefixDescriptor => "before-prefix-descriptor",
77            Self::AfterPrefixDescriptor => "after-prefix-descriptor",
78            Self::AfterNamespaceReservation => "after-namespace-reservation",
79            Self::BeforeFormatCas => "before-format-cas",
80            Self::AfterFormatCas => "after-format-cas",
81            Self::BeforeObjectPublish => "before-object-publish",
82            Self::AfterObjectPublish => "after-object-publish",
83            Self::AfterDurabilityReceipt => "after-durability-receipt",
84            Self::BeforeCas => "before-cas",
85            Self::AfterCas => "after-cas",
86            Self::BeforeAcknowledgement => "before-acknowledgement",
87        }
88    }
89}
90
91#[derive(Clone)]
92pub(crate) struct PhysicalEntry {
93    pub(crate) entry: JournalEntry,
94    pub(crate) previous_location: Option<EntryLocation>,
95    pub(crate) payloads: Vec<StoredDatumRef>,
96}
97
98#[derive(Clone)]
99pub(crate) struct V1Entry {
100    pub(crate) id: ContentId,
101    pub(crate) sequence: u64,
102    pub(crate) previous: Option<ContentId>,
103    pub(crate) kind: Symbol,
104    pub(crate) payloads: Vec<ContentId>,
105}
106
107impl V1Entry {
108    pub(crate) fn canonical_id(&self) -> ContentId {
109        let mut hasher = Sha256::new();
110        hasher.update(b"sim-journal-entry-v1\0");
111        hasher.update(self.sequence.to_be_bytes());
112        put_v1_optional_id_hash(&mut hasher, self.previous.as_ref());
113        put_v1_symbol_hash(&mut hasher, &self.kind);
114        hasher.update((self.payloads.len() as u64).to_be_bytes());
115        for id in &self.payloads {
116            put_v1_id_hash(&mut hasher, id);
117        }
118        ContentId::from_bytes(
119            Symbol::qualified("journal", "sha256-entry-v1"),
120            hasher.finalize().into(),
121        )
122    }
123}
124
125pub(crate) fn encode_envelope(value: &NativeStateEnvelope) -> Vec<u8> {
126    let mut out = b"SIMJSTATE2".to_vec();
127    out.extend(value.fence.to_be_bytes());
128    put_text(&mut out, &value.namespace.0);
129    match &value.prefix {
130        Some(prefix) => {
131            out.push(1);
132            out.extend(prefix.entries.to_be_bytes());
133            put_head(&mut out, &prefix.physical_head);
134            put_head(&mut out, &prefix.canonical_head);
135            put_id(&mut out, &prefix.descriptor);
136        }
137        None => out.push(0),
138    }
139    put_optional_head(&mut out, value.head.as_ref());
140    put_optional_location(&mut out, value.head_location.as_ref());
141    out
142}
143
144pub(crate) fn decode_envelope(bytes: &[u8]) -> Result<NativeStateEnvelope, JournalError> {
145    let mut cursor = Cursor::new(bytes);
146    if cursor.take(10)? != b"SIMJSTATE2" {
147        return Err(JournalError::CorruptState("state format"));
148    }
149    let fence = cursor.u64()?;
150    let namespace = EntryNamespace(cursor.text()?);
151    let prefix = match cursor.byte()? {
152        0 => None,
153        1 => Some(VerifiedNativePrefixRef {
154            entries: cursor.u64()?,
155            physical_head: cursor.head()?,
156            canonical_head: cursor.head()?,
157            descriptor: cursor.id()?,
158        }),
159        _ => return Err(JournalError::CorruptState("prefix tag")),
160    };
161    let head = cursor.optional_head()?;
162    let head_location = cursor.optional_location()?;
163    cursor.end()?;
164    if prefix.is_none() && head.is_some() && head_location.is_none() {
165        return Err(JournalError::CorruptState("head without locator"));
166    }
167    Ok(NativeStateEnvelope {
168        format: NativeFormatId::V2,
169        fence,
170        namespace,
171        prefix,
172        head,
173        head_location,
174    })
175}
176
177pub(crate) fn encode_descriptor(old_state: &[u8], canonical_head: &JournalHead) -> Vec<u8> {
178    let mut out = b"SIMJPREFIX1".to_vec();
179    out.extend((old_state.len() as u64).to_be_bytes());
180    out.extend(old_state);
181    put_head(&mut out, canonical_head);
182    out
183}
184
185pub(crate) fn decode_descriptor(bytes: &[u8]) -> Result<(Vec<u8>, JournalHead), JournalError> {
186    let mut cursor = Cursor::new(bytes);
187    if cursor.take(11)? != b"SIMJPREFIX1" {
188        return Err(JournalError::CorruptState("prefix descriptor"));
189    }
190    let len = usize::try_from(cursor.u64()?).map_err(|_| JournalError::WorkBoundExceeded)?;
191    let state = cursor.take(len)?.to_vec();
192    let canonical_head = cursor.head()?;
193    cursor.end()?;
194    Ok((state, canonical_head))
195}
196
197pub(crate) fn encode_v2_entry(value: &PhysicalEntry) -> Vec<u8> {
198    let mut out = b"SIMJENTRY2".to_vec();
199    let datum = crate::datum_codec::encode(&value.entry.canonical_datum())
200        .expect("canonical entry datum encodes");
201    out.extend((datum.len() as u64).to_be_bytes());
202    out.extend(datum);
203    put_optional_location(&mut out, value.previous_location.as_ref());
204    out.extend((value.payloads.len() as u64).to_be_bytes());
205    for reference in &value.payloads {
206        put_id(&mut out, &reference.meaning);
207        put_id(&mut out, &reference.storage);
208    }
209    out
210}
211
212pub(crate) fn decode_v2_entry(bytes: &[u8]) -> Result<PhysicalEntry, JournalError> {
213    let mut cursor = Cursor::new(bytes);
214    if cursor.take(10)? != b"SIMJENTRY2" {
215        return Err(JournalError::CorruptState("entry format"));
216    }
217    let datum_len = usize::try_from(cursor.u64()?).map_err(|_| JournalError::WorkBoundExceeded)?;
218    let datum = crate::datum_codec::decode(cursor.take(datum_len)?)?;
219    let entry = entry_from_datum(datum)?;
220    let previous_location = cursor.optional_location()?;
221    let count = usize::try_from(cursor.u64()?).map_err(|_| JournalError::WorkBoundExceeded)?;
222    let mut payloads = Vec::with_capacity(count);
223    for _ in 0..count {
224        payloads.push(StoredDatumRef {
225            meaning: cursor.id()?,
226            storage: cursor.id()?,
227        });
228    }
229    cursor.end()?;
230    Ok(PhysicalEntry {
231        entry,
232        previous_location,
233        payloads,
234    })
235}
236
237pub(crate) fn entry_from_datum(datum: Datum) -> Result<JournalEntry, JournalError> {
238    let id = datum.content_id().map_err(|_| JournalError::CorruptEntry)?;
239    let Datum::Node { tag, fields } = datum else {
240        return Err(JournalError::CorruptEntry);
241    };
242    if tag != Symbol::qualified("journal", "entry-v2") || fields.len() != 4 {
243        return Err(JournalError::CorruptEntry);
244    }
245    let field = |name: &str| {
246        fields
247            .iter()
248            .find(|(field, _)| *field == Symbol::new(name))
249            .map(|(_, value)| value)
250            .ok_or(JournalError::CorruptEntry)
251    };
252    let sequence = match field("sequence")? {
253        Datum::Number(number)
254            if number.domain == Symbol::qualified("numbers", "u64")
255                && number
256                    .canonical
257                    .parse::<u64>()
258                    .ok()
259                    .is_some_and(|value| value.to_string() == number.canonical) =>
260        {
261            number
262                .canonical
263                .parse()
264                .map_err(|_| JournalError::CorruptEntry)?
265        }
266        _ => return Err(JournalError::CorruptEntry),
267    };
268    let previous = match field("previous")? {
269        Datum::Nil => None,
270        value => Some(id_from_datum(value)?),
271    };
272    let kind = match field("kind")? {
273        Datum::Symbol(kind) => kind.clone(),
274        _ => return Err(JournalError::CorruptEntry),
275    };
276    let payloads = match field("payloads")? {
277        Datum::Vector(values) => values.iter().map(id_from_datum).collect::<Result<_, _>>()?,
278        _ => return Err(JournalError::CorruptEntry),
279    };
280    let entry = JournalEntry {
281        id,
282        sequence,
283        previous,
284        kind,
285        payloads,
286    };
287    if entry.canonical_id()? != entry.id {
288        return Err(JournalError::CorruptEntry);
289    }
290    Ok(entry)
291}
292
293pub(crate) fn id_from_datum(value: &Datum) -> Result<ContentId, JournalError> {
294    let Datum::Node { tag, fields } = value else {
295        return Err(JournalError::CorruptEntry);
296    };
297    if *tag != Symbol::qualified("journal", "content-id-v1") || fields.len() != 2 {
298        return Err(JournalError::CorruptEntry);
299    }
300    let algorithm = fields
301        .iter()
302        .find_map(|(name, value)| (*name == Symbol::new("algorithm")).then_some(value));
303    let digest = fields
304        .iter()
305        .find_map(|(name, value)| (*name == Symbol::new("digest")).then_some(value));
306    let (Some(Datum::Symbol(algorithm)), Some(Datum::Bytes(bytes))) = (algorithm, digest) else {
307        return Err(JournalError::CorruptEntry);
308    };
309    let digest: [u8; 32] = bytes
310        .as_slice()
311        .try_into()
312        .map_err(|_| JournalError::CorruptEntry)?;
313    Ok(ContentId::from_bytes(algorithm.clone(), digest))
314}
315
316pub(crate) fn decode_v1_state(bytes: &[u8]) -> Result<(u64, Option<JournalHead>), JournalError> {
317    let mut cursor = Cursor::new(bytes);
318    if cursor.take(10)? != b"SIMJSTATE1" {
319        return Err(JournalError::CorruptState("state format"));
320    }
321    let fence = cursor.u64()?;
322    let head = cursor.optional_head()?;
323    cursor.end()?;
324    Ok((fence, head))
325}
326
327pub(crate) fn decode_v1_entry(bytes: &[u8]) -> Result<V1Entry, JournalError> {
328    let mut cursor = Cursor::new(bytes);
329    if cursor.take(10)? != b"SIMJENTRY1" {
330        return Err(JournalError::CorruptState("entry format"));
331    }
332    let id = cursor.id()?;
333    let sequence = cursor.u64()?;
334    let previous = match cursor.byte()? {
335        0 => None,
336        1 => Some(cursor.id()?),
337        _ => return Err(JournalError::CorruptState("entry tag")),
338    };
339    let kind = parse_symbol(&cursor.text()?)?;
340    let count = cursor.u32()? as usize;
341    let mut payloads = Vec::with_capacity(count);
342    for _ in 0..count {
343        payloads.push(cursor.id()?);
344    }
345    cursor.end()?;
346    Ok(V1Entry {
347        id,
348        sequence,
349        previous,
350        kind,
351        payloads,
352    })
353}
354
355pub(crate) fn v1_object_id(bytes: &[u8]) -> ContentId {
356    ContentId::from_bytes(
357        Symbol::qualified("journal", "sha256-bytes-v1"),
358        Sha256::digest(bytes).into(),
359    )
360}
361
362pub(crate) fn put_v1_optional_id_hash(hasher: &mut Sha256, id: Option<&ContentId>) {
363    match id {
364        Some(id) => {
365            hasher.update([1]);
366            put_v1_id_hash(hasher, id);
367        }
368        None => hasher.update([0]),
369    }
370}
371
372pub(crate) fn put_v1_id_hash(hasher: &mut Sha256, id: &ContentId) {
373    put_v1_symbol_hash(hasher, &id.algorithm);
374    hasher.update(id.bytes);
375}
376
377pub(crate) fn put_v1_symbol_hash(hasher: &mut Sha256, symbol: &Symbol) {
378    let text = symbol.as_qualified_str();
379    hasher.update((text.len() as u64).to_be_bytes());
380    hasher.update(text.as_bytes());
381}
382
383pub(crate) fn namespace_root(namespace: &EntryNamespace) -> Vec<String> {
384    vec!["namespaces-v2".into(), namespace.0.clone()]
385}
386
387pub(crate) fn ensure_entry_dirs(
388    port: &dyn HostDirPort,
389    location: &EntryLocation,
390) -> Result<(), JournalError> {
391    let root = [namespace_root(&location.namespace), vec!["entries".into()]].concat();
392    let sequence = [root, vec![format!("{:016x}", location.sequence)]].concat();
393    port.create_dir(&sequence).map_err(port_error)?;
394    port.create_dir(&[sequence, vec![id_key(&location.entry)]].concat())
395        .map_err(port_error)
396}
397
398pub(crate) fn entry_path(location: &EntryLocation) -> Vec<String> {
399    vec![
400        "namespaces-v2".into(),
401        location.namespace.0.clone(),
402        "entries".into(),
403        format!("{:016x}", location.sequence),
404        id_key(&location.entry),
405        id_key(&location.storage),
406    ]
407}
408
409pub(crate) fn object_path(reference: &StoredDatumRef) -> Vec<String> {
410    vec![
411        "objects-v2".into(),
412        id_key(&reference.meaning),
413        id_key(&reference.storage),
414    ]
415}
416
417pub(crate) fn descriptor_path(id: &ContentId) -> Vec<String> {
418    vec!["compat-v1".into(), id_key(id)]
419}
420
421pub(crate) fn v1_entry_path(sequence: u64) -> Vec<String> {
422    vec!["entries".into(), format!("{sequence:016x}")]
423}
424
425pub(crate) fn v1_object_path(id: &ContentId) -> Vec<String> {
426    vec!["objects".into(), hex(&id.bytes)]
427}
428
429pub(crate) fn id_key(id: &ContentId) -> String {
430    format!(
431        "{}-{}",
432        hex(id.algorithm.as_qualified_str().as_bytes()),
433        hex(&id.bytes)
434    )
435}
436
437pub(crate) fn parse_id_key(text: &str) -> Result<ContentId, JournalError> {
438    let (algorithm, digest) = text
439        .split_once('-')
440        .ok_or(JournalError::CorruptState("id path"))?;
441    let algorithm = String::from_utf8(unhex(algorithm)?)
442        .map_err(|_| JournalError::CorruptState("id path utf8"))?;
443    let digest = unhex(digest)?;
444    let bytes: [u8; 32] = digest
445        .try_into()
446        .map_err(|_| JournalError::CorruptState("id path digest"))?;
447    Ok(ContentId::from_bytes(parse_symbol(&algorithm)?, bytes))
448}
449
450pub(crate) fn hex(bytes: &[u8]) -> String {
451    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
452}
453
454pub(crate) fn unhex(text: &str) -> Result<Vec<u8>, JournalError> {
455    if !text.len().is_multiple_of(2) {
456        return Err(JournalError::CorruptState("hex path"));
457    }
458    text.as_bytes()
459        .chunks_exact(2)
460        .map(|pair| {
461            let high = (pair[0] as char)
462                .to_digit(16)
463                .ok_or(JournalError::CorruptState("hex path"))?;
464            let low = (pair[1] as char)
465                .to_digit(16)
466                .ok_or(JournalError::CorruptState("hex path"))?;
467            Ok(((high << 4) | low) as u8)
468        })
469        .collect()
470}
471
472pub(crate) fn put_head(out: &mut Vec<u8>, head: &JournalHead) {
473    out.extend(head.sequence.to_be_bytes());
474    put_id(out, &head.entry);
475}
476
477pub(crate) fn put_optional_head(out: &mut Vec<u8>, head: Option<&JournalHead>) {
478    match head {
479        Some(head) => {
480            out.push(1);
481            put_head(out, head);
482        }
483        None => out.push(0),
484    }
485}
486
487pub(crate) fn put_optional_location(out: &mut Vec<u8>, location: Option<&EntryLocation>) {
488    match location {
489        Some(location) => {
490            out.push(1);
491            put_text(out, &location.namespace.0);
492            out.extend(location.sequence.to_be_bytes());
493            put_id(out, &location.entry);
494            put_id(out, &location.storage);
495        }
496        None => out.push(0),
497    }
498}
499
500pub(crate) fn put_id(out: &mut Vec<u8>, id: &ContentId) {
501    put_text(out, &id.algorithm.as_qualified_str());
502    out.extend(id.bytes);
503}
504
505pub(crate) fn put_text(out: &mut Vec<u8>, text: &str) {
506    out.extend((text.len() as u32).to_be_bytes());
507    out.extend(text.as_bytes());
508}
509
510pub(crate) fn parse_symbol(text: &str) -> Result<Symbol, JournalError> {
511    match text.split_once('/') {
512        Some((namespace, name)) if !namespace.is_empty() && !name.is_empty() => {
513            Ok(Symbol::qualified(namespace, name))
514        }
515        None => Symbol::checked(text).map_err(|_| JournalError::CorruptState("symbol")),
516        _ => Err(JournalError::CorruptState("symbol")),
517    }
518}
519
520pub(crate) fn port_error(error: sim_storage_port::HostDirError) -> JournalError {
521    JournalError::Backend(error.to_string())
522}
523
524struct Cursor<'a> {
525    bytes: &'a [u8],
526    at: usize,
527}
528
529impl<'a> Cursor<'a> {
530    fn new(bytes: &'a [u8]) -> Self {
531        Self { bytes, at: 0 }
532    }
533    fn take(&mut self, len: usize) -> Result<&'a [u8], JournalError> {
534        let end = self
535            .at
536            .checked_add(len)
537            .ok_or(JournalError::CorruptState("length"))?;
538        let value = self
539            .bytes
540            .get(self.at..end)
541            .ok_or(JournalError::CorruptState("truncated"))?;
542        self.at = end;
543        Ok(value)
544    }
545    fn byte(&mut self) -> Result<u8, JournalError> {
546        Ok(self.take(1)?[0])
547    }
548    fn u32(&mut self) -> Result<u32, JournalError> {
549        Ok(u32::from_be_bytes(
550            self.take(4)?
551                .try_into()
552                .map_err(|_| JournalError::CorruptState("u32"))?,
553        ))
554    }
555    fn u64(&mut self) -> Result<u64, JournalError> {
556        Ok(u64::from_be_bytes(
557            self.take(8)?
558                .try_into()
559                .map_err(|_| JournalError::CorruptState("u64"))?,
560        ))
561    }
562    fn text(&mut self) -> Result<String, JournalError> {
563        let len = self.u32()? as usize;
564        String::from_utf8(self.take(len)?.to_vec()).map_err(|_| JournalError::CorruptState("utf8"))
565    }
566    fn id(&mut self) -> Result<ContentId, JournalError> {
567        let algorithm = parse_symbol(&self.text()?)?;
568        let bytes = self
569            .take(32)?
570            .try_into()
571            .map_err(|_| JournalError::CorruptState("id"))?;
572        Ok(ContentId::from_bytes(algorithm, bytes))
573    }
574    fn head(&mut self) -> Result<JournalHead, JournalError> {
575        Ok(JournalHead {
576            sequence: self.u64()?,
577            entry: self.id()?,
578        })
579    }
580    fn optional_head(&mut self) -> Result<Option<JournalHead>, JournalError> {
581        match self.byte()? {
582            0 => Ok(None),
583            1 => Ok(Some(self.head()?)),
584            _ => Err(JournalError::CorruptState("head tag")),
585        }
586    }
587    fn optional_location(&mut self) -> Result<Option<EntryLocation>, JournalError> {
588        match self.byte()? {
589            0 => Ok(None),
590            1 => Ok(Some(EntryLocation {
591                namespace: EntryNamespace(self.text()?),
592                sequence: self.u64()?,
593                entry: self.id()?,
594                storage: self.id()?,
595            })),
596            _ => Err(JournalError::CorruptState("location tag")),
597        }
598    }
599    fn end(&self) -> Result<(), JournalError> {
600        if self.at == self.bytes.len() {
601            Ok(())
602        } else {
603            Err(JournalError::CorruptState("trailing bytes"))
604        }
605    }
606}
sim_lib_journal/native_codec.rs

sim_lib_journal/
native_codec.rs
