use super::super::restore_checkpoint::{CheckpointError, CheckpointPositions};
const MAGIC: &[u8; 5] = b"LCAO\x01";
const DIGEST_DOMAIN: &[u8] = b"liminal:checkpoint-owner:v1\0";
#[derive(Debug, thiserror::Error)]
enum OwnerBodyError {
#[error("invalid checkpoint owner field {field}")]
Field { field: &'static str },
}
pub(in crate::server::participant::production) fn invalid(field: &'static str) -> CheckpointError {
CheckpointError::LiveState {
source: Box::new(OwnerBodyError::Field { field }),
}
}
#[derive(Debug)]
pub(in crate::server::participant::production) struct CheckpointOwnerBody<'a> {
pub conversation_id: u64,
pub next_order: u64,
pub next_seq: u64,
pub next_participant: u64,
pub next_log_sequence: u64,
pub observer_progress: u64,
pub closed: bool,
pub lifecycle_capsule: &'a [u8],
pub debt_dispatch: &'a [u8],
pub outbox: &'a [u8],
pub slots: &'a [u8],
pub pending_specific_fates: &'a [u8],
pub prepared_ordinary_finalizers: &'a [u8],
pub fate_occurrences: &'a [u8],
pub retired: &'a [u8],
pub tokens: &'a [u8],
pub committed_admissions: &'a [u8],
pub observer_progress_witnesses: &'a [u8],
pub terminal_validation: &'a [u8],
}
impl<'a> CheckpointOwnerBody<'a> {
pub(in crate::server::participant::production) fn encode_canonical(
&self,
) -> Result<Vec<u8>, CheckpointError> {
let mut encoder = CheckpointBodyEncoder::new();
encoder.raw(MAGIC);
for value in [
self.conversation_id,
self.next_order,
self.next_seq,
self.next_participant,
self.next_log_sequence,
self.observer_progress,
] {
encoder.u64(value);
}
encoder.u8(u8::from(self.closed));
for (field, bytes) in [
("lifecycle capsule", self.lifecycle_capsule),
("debt dispatch", self.debt_dispatch),
("outbox", self.outbox),
("slots", self.slots),
("pending specific fates", self.pending_specific_fates),
(
"prepared ordinary finalizers",
self.prepared_ordinary_finalizers,
),
("fate occurrences", self.fate_occurrences),
("retired", self.retired),
("tokens", self.tokens),
("committed admissions", self.committed_admissions),
(
"observer progress witnesses",
self.observer_progress_witnesses,
),
("terminal validation", self.terminal_validation),
] {
encoder.bytes(field, bytes)?;
}
let mut body = encoder.finish();
let mut hasher = blake3::Hasher::new();
hasher.update(DIGEST_DOMAIN);
hasher.update(&body);
body.extend_from_slice(hasher.finalize().as_bytes());
Ok(body)
}
pub(in crate::server::participant::production) fn decode_canonical(
bytes: &'a [u8],
) -> Result<Self, CheckpointError> {
let end = bytes
.len()
.checked_sub(blake3::OUT_LEN)
.ok_or_else(|| invalid("owner digest"))?;
let framed = bytes.get(..end).ok_or_else(|| invalid("owner body"))?;
let digest = bytes.get(end..).ok_or_else(|| invalid("owner digest"))?;
let mut hasher = blake3::Hasher::new();
hasher.update(DIGEST_DOMAIN);
hasher.update(framed);
if hasher.finalize().as_bytes() != digest {
return Err(invalid("owner digest"));
}
let mut decoder = CheckpointBodyDecoder::new(framed);
if decoder.take("owner version", MAGIC.len())? != MAGIC {
return Err(invalid("owner version"));
}
let body = Self {
conversation_id: decoder.u64("conversation id")?,
next_order: decoder.u64("next order")?,
next_seq: decoder.u64("next delivery sequence")?,
next_participant: decoder.u64("next participant")?,
next_log_sequence: decoder.u64("next log sequence")?,
observer_progress: decoder.u64("observer progress")?,
closed: match decoder.u8("closed")? {
0 => false,
1 => true,
_ => return Err(invalid("closed")),
},
lifecycle_capsule: decoder.bytes("lifecycle capsule")?,
debt_dispatch: decoder.bytes("debt dispatch")?,
outbox: decoder.bytes("outbox")?,
slots: decoder.bytes("slots")?,
pending_specific_fates: decoder.bytes("pending specific fates")?,
prepared_ordinary_finalizers: decoder.bytes("prepared ordinary finalizers")?,
fate_occurrences: decoder.bytes("fate occurrences")?,
retired: decoder.bytes("retired")?,
tokens: decoder.bytes("tokens")?,
committed_admissions: decoder.bytes("committed admissions")?,
observer_progress_witnesses: decoder.bytes("observer progress witnesses")?,
terminal_validation: decoder.bytes("terminal validation")?,
};
decoder.finish()?;
Ok(body)
}
pub(in crate::server::participant::production) fn validate_positions(
&self,
positions: &CheckpointPositions,
) -> Result<(), CheckpointError> {
positions.validate()?;
if self.conversation_id != positions.conversation_id {
return Err(CheckpointError::ForeignConversation {
expected: positions.conversation_id,
actual: self.conversation_id,
});
}
for (field, actual, expected) in [
("next order", self.next_order, positions.next_order),
("next delivery sequence", self.next_seq, positions.next_seq),
(
"next participant",
self.next_participant,
positions.next_participant,
),
(
"next log sequence",
self.next_log_sequence,
positions.next_log_sequence,
),
(
"observer progress",
self.observer_progress,
positions.observer_progress,
),
] {
if actual != expected {
return Err(invalid(field));
}
}
Ok(())
}
}
pub(in crate::server::participant::production) struct CheckpointBodyEncoder {
bytes: Vec<u8>,
}
impl CheckpointBodyEncoder {
pub(in crate::server::participant::production) const fn new() -> Self {
Self { bytes: Vec::new() }
}
pub(in crate::server::participant::production) fn raw(&mut self, bytes: &[u8]) {
self.bytes.extend_from_slice(bytes);
}
pub(in crate::server::participant::production) fn u8(&mut self, value: u8) {
self.bytes.push(value);
}
pub(in crate::server::participant::production) fn u64(&mut self, value: u64) {
self.raw(&value.to_le_bytes());
}
pub(in crate::server::participant::production) fn count(
&mut self,
field: &'static str,
count: usize,
) -> Result<(), CheckpointError> {
let count = u32::try_from(count).map_err(|_| invalid(field))?;
self.raw(&count.to_le_bytes());
Ok(())
}
pub(in crate::server::participant::production) fn bytes(
&mut self,
field: &'static str,
bytes: &[u8],
) -> Result<(), CheckpointError> {
self.count(field, bytes.len())?;
self.raw(bytes);
Ok(())
}
pub(in crate::server::participant::production) fn finish(self) -> Vec<u8> {
self.bytes
}
}
pub(in crate::server::participant::production) struct CheckpointBodyDecoder<'a> {
bytes: &'a [u8],
}
impl<'a> CheckpointBodyDecoder<'a> {
pub(in crate::server::participant::production) const fn is_empty(&self) -> bool {
self.bytes.is_empty()
}
pub(in crate::server::participant::production) const fn new(bytes: &'a [u8]) -> Self {
Self { bytes }
}
pub(in crate::server::participant::production) fn take(
&mut self,
field: &'static str,
count: usize,
) -> Result<&'a [u8], CheckpointError> {
let head = self.bytes.get(..count).ok_or_else(|| invalid(field))?;
self.bytes = self.bytes.get(count..).ok_or_else(|| invalid(field))?;
Ok(head)
}
pub(in crate::server::participant::production) fn u8(
&mut self,
field: &'static str,
) -> Result<u8, CheckpointError> {
self.take(field, 1)?
.first()
.copied()
.ok_or_else(|| invalid(field))
}
pub(in crate::server::participant::production) fn u64(
&mut self,
field: &'static str,
) -> Result<u64, CheckpointError> {
let bytes: [u8; 8] = self
.take(field, 8)?
.try_into()
.map_err(|_| invalid(field))?;
Ok(u64::from_le_bytes(bytes))
}
pub(in crate::server::participant::production) fn count(
&mut self,
field: &'static str,
minimum_width: usize,
) -> Result<usize, CheckpointError> {
let bytes: [u8; 4] = self
.take(field, 4)?
.try_into()
.map_err(|_| invalid(field))?;
let count = usize::try_from(u32::from_le_bytes(bytes)).map_err(|_| invalid(field))?;
if minimum_width == 0 || count > self.bytes.len() / minimum_width {
return Err(invalid(field));
}
Ok(count)
}
pub(in crate::server::participant::production) fn bytes(
&mut self,
field: &'static str,
) -> Result<&'a [u8], CheckpointError> {
let length = self.count(field, 1)?;
self.take(field, length)
}
pub(in crate::server::participant::production) fn finish(self) -> Result<(), CheckpointError> {
if self.bytes.is_empty() {
Ok(())
} else {
Err(invalid("trailing owner bytes"))
}
}
}