use super::restore_checkpoint::{CheckpointError, CheckpointPositions, CheckpointSchema};
const MAGIC: [u8; 4] = *b"LSCK";
const VERSION: (u16, u16) = (1, 0);
const POSITIONS_TAG: u16 = 1;
const LIVE_TAG: u16 = 2;
const FIXED_POSITIONS_BYTES: usize = 12 * 8 + 1 + 2;
const DIGEST_BYTES: usize = 32;
pub(super) fn encode<const N: usize>(
positions: CheckpointPositions,
root: &[u8; N],
live: &[u8],
) -> Result<Vec<u8>, CheckpointError> {
let root_width = u16::try_from(N).map_err(|_| CheckpointError::LengthOverflow)?;
let position_len = u32::try_from(
FIXED_POSITIONS_BYTES
.checked_add(N)
.ok_or(CheckpointError::LengthOverflow)?,
)
.map_err(|_| CheckpointError::LengthOverflow)?;
let live_len = u32::try_from(live.len()).map_err(|_| CheckpointError::LengthOverflow)?;
let mut output = Vec::new();
output.extend_from_slice(&MAGIC);
output.extend_from_slice(&VERSION.0.to_be_bytes());
output.extend_from_slice(&VERSION.1.to_be_bytes());
output.extend_from_slice(&POSITIONS_TAG.to_be_bytes());
output.extend_from_slice(&position_len.to_be_bytes());
for value in [
positions.conversation_id,
positions.covered_log_sequence,
positions.next_event_ordinal,
positions.next_log_sequence,
positions.next_seq,
positions.next_order,
positions.next_participant,
positions.extension_sequence,
positions.highest_delivery_seq,
positions.observer_progress,
positions.merged_base_sequence,
positions.merged_extension_sequence,
] {
output.extend_from_slice(&value.to_be_bytes());
}
output.push(match positions.schema {
CheckpointSchema::V2Prefix => 2,
CheckpointSchema::V3Suffix => 3,
CheckpointSchema::V4Suffix => 4,
});
output.extend_from_slice(&root_width.to_be_bytes());
output.extend_from_slice(root);
output.extend_from_slice(&LIVE_TAG.to_be_bytes());
output.extend_from_slice(&live_len.to_be_bytes());
output.extend_from_slice(live);
let digest = blake3::hash(&output);
output.extend_from_slice(digest.as_bytes());
Ok(output)
}
pub(super) fn decode<const N: usize>(
input: &[u8],
expected_conversation: u64,
) -> Result<(CheckpointPositions, [u8; N], &[u8]), CheckpointError> {
let mut reader = Reader::new(input);
if reader.take(4)? != MAGIC {
return Err(CheckpointError::InvalidMagic);
}
let major = reader.u16()?;
let minor = reader.u16()?;
if (major, minor) != VERSION {
return Err(CheckpointError::UnsupportedCodec { major, minor });
}
let position_bytes = reader.section(POSITIONS_TAG)?;
let mut position_reader = Reader::new(position_bytes);
let positions = CheckpointPositions {
conversation_id: position_reader.u64()?,
covered_log_sequence: position_reader.u64()?,
next_event_ordinal: position_reader.u64()?,
next_log_sequence: position_reader.u64()?,
next_seq: position_reader.u64()?,
next_order: position_reader.u64()?,
next_participant: position_reader.u64()?,
extension_sequence: position_reader.u64()?,
highest_delivery_seq: position_reader.u64()?,
observer_progress: position_reader.u64()?,
merged_base_sequence: position_reader.u64()?,
merged_extension_sequence: position_reader.u64()?,
schema: match position_reader.array::<1>()?[0] {
2 => CheckpointSchema::V2Prefix,
3 => CheckpointSchema::V3Suffix,
4 => CheckpointSchema::V4Suffix,
tag => return Err(CheckpointError::Schema { tag }),
},
};
let actual_width = usize::from(position_reader.u16()?);
if actual_width != N || N == 0 {
return Err(CheckpointError::RootWidth {
expected: N,
actual: actual_width,
});
}
let root = position_reader.array::<N>()?;
if !position_reader.finished() {
return Err(CheckpointError::Section { tag: POSITIONS_TAG });
}
let live = reader.section(LIVE_TAG)?;
let digest_offset = reader.at;
let digest = reader.array::<DIGEST_BYTES>()?;
if !reader.finished() {
return Err(CheckpointError::Section { tag: LIVE_TAG });
}
if blake3::hash(&input[..digest_offset]).as_bytes() != &digest {
return Err(CheckpointError::Integrity);
}
if positions.conversation_id != expected_conversation {
return Err(CheckpointError::ForeignConversation {
expected: expected_conversation,
actual: positions.conversation_id,
});
}
positions.validate()?;
Ok((positions, root, live))
}
struct Reader<'a> {
bytes: &'a [u8],
at: usize,
}
impl<'a> Reader<'a> {
const fn new(bytes: &'a [u8]) -> Self {
Self { bytes, at: 0 }
}
fn take(&mut self, count: usize) -> Result<&'a [u8], CheckpointError> {
let end = self
.at
.checked_add(count)
.ok_or(CheckpointError::LengthOverflow)?;
let bytes = self
.bytes
.get(self.at..end)
.ok_or(CheckpointError::Truncated {
required: end,
available: self.bytes.len(),
})?;
self.at = end;
Ok(bytes)
}
fn array<const N: usize>(&mut self) -> Result<[u8; N], CheckpointError> {
self.take(N)?
.try_into()
.map_err(|_| CheckpointError::LengthOverflow)
}
fn u16(&mut self) -> Result<u16, CheckpointError> {
self.array().map(u16::from_be_bytes)
}
fn u64(&mut self) -> Result<u64, CheckpointError> {
self.array().map(u64::from_be_bytes)
}
fn section(&mut self, expected: u16) -> Result<&'a [u8], CheckpointError> {
if self.u16()? != expected {
return Err(CheckpointError::Section { tag: expected });
}
let count = usize::try_from(u32::from_be_bytes(self.array()?))
.map_err(|_| CheckpointError::LengthOverflow)?;
self.take(count)
}
const fn finished(&self) -> bool {
self.at == self.bytes.len()
}
}