use crate::hash::{FNV_OFFSET, fnv64};
use serde::{Deserialize, Serialize};
const COMMIT_MARKER: u64 = 0x4943_4D45_534E_4150;
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CommittedGenerationBytes {
pub(crate) generation: u64,
pub(crate) commit_marker: u64,
pub(crate) checksum: u64,
#[serde(with = "opaque_payload")]
pub(crate) payload: Vec<u8>,
}
mod opaque_payload {
use crate::constants::MAX_COMMITTED_PAYLOAD_BYTES;
use serde::{Deserializer, Serialize, Serializer, de::Visitor};
pub fn serialize<S: Serializer>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
if bytes.len() > MAX_COMMITTED_PAYLOAD_BYTES {
return Err(serde::ser::Error::custom(
"ledger byte string exceeds payload bound",
));
}
if serializer.is_human_readable() {
bytes.serialize(serializer)
} else {
serializer.serialize_bytes(bytes)
}
}
pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Vec<u8>, D::Error> {
struct Bytes;
impl Visitor<'_> for Bytes {
type Value = Vec<u8>;
fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("a bounded opaque ledger byte string")
}
fn visit_bytes<E: serde::de::Error>(self, bytes: &[u8]) -> Result<Self::Value, E> {
if bytes.len() > MAX_COMMITTED_PAYLOAD_BYTES {
return Err(E::custom("ledger byte string exceeds payload bound"));
}
Ok(bytes.to_vec())
}
fn visit_byte_buf<E: serde::de::Error>(self, bytes: Vec<u8>) -> Result<Self::Value, E> {
if bytes.len() > MAX_COMMITTED_PAYLOAD_BYTES {
return Err(E::custom("ledger byte string exceeds payload bound"));
}
Ok(bytes)
}
}
if deserializer.is_human_readable() {
return crate::cbor::deserialize_bounded_vec::<D, u8, MAX_COMMITTED_PAYLOAD_BYTES>(
deserializer,
);
}
deserializer.deserialize_byte_buf(Bytes)
}
}
impl CommittedGenerationBytes {
#[must_use]
pub fn new(generation: u64, payload: Vec<u8>) -> Self {
let mut record = Self {
generation,
commit_marker: COMMIT_MARKER,
checksum: 0,
payload,
};
record.checksum = generation_checksum(&record);
record
}
#[must_use]
pub const fn generation(&self) -> u64 {
self.generation
}
#[must_use]
#[cfg(test)]
pub const fn commit_marker(&self) -> u64 {
self.commit_marker
}
#[must_use]
#[cfg(test)]
pub const fn checksum(&self) -> u64 {
self.checksum
}
#[must_use]
pub fn payload(&self) -> &[u8] {
&self.payload
}
#[must_use]
pub fn validates(&self) -> bool {
self.commit_marker == COMMIT_MARKER && self.checksum == generation_checksum(self)
}
}
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SnapshotCommitStore {
#[serde(deserialize_with = "crate::cbor::deserialize_present_option")]
pub(crate) current: Option<CommittedGenerationBytes>,
}
impl SnapshotCommitStore {
pub const fn current(&self) -> Option<&CommittedGenerationBytes> {
self.current.as_ref()
}
pub fn authoritative(&self) -> Result<&CommittedGenerationBytes, CommitRecoveryError> {
let current = self
.current()
.ok_or(CommitRecoveryError::NoValidGeneration)?;
if !current.validates() {
return Err(CommitRecoveryError::InvalidSnapshot {
generation: current.generation(),
});
}
Ok(current)
}
pub(crate) fn authoritative_with_diagnostic(
&self,
) -> (
Result<&CommittedGenerationBytes, CommitRecoveryError>,
CommitStoreDiagnostic,
) {
let recovery = self.authoritative();
let current = match self.current() {
None => CommitSnapshotDiagnostic::Empty,
Some(record) if recovery.is_ok() => CommitSnapshotDiagnostic::Valid {
generation: record.generation(),
},
Some(record) => CommitSnapshotDiagnostic::Invalid {
generation: record.generation(),
},
};
let diagnostic = CommitStoreDiagnostic {
current,
recovery: recovery.map(CommittedGenerationBytes::generation),
};
(recovery, diagnostic)
}
pub fn diagnostic(&self) -> CommitStoreDiagnostic {
self.authoritative_with_diagnostic().1
}
pub fn commit_payload_at_generation(
&mut self,
generation: u64,
payload: Vec<u8>,
) -> Result<&CommittedGenerationBytes, CommitRecoveryError> {
if payload.len() > crate::constants::MAX_COMMITTED_PAYLOAD_BYTES {
return Err(CommitRecoveryError::PayloadTooLarge {
len: payload.len(),
limit: crate::constants::MAX_COMMITTED_PAYLOAD_BYTES,
});
}
match self.authoritative() {
Ok(prior) => {
let prior_generation = prior.generation();
let expected = prior_generation.checked_add(1).ok_or(
CommitRecoveryError::GenerationOverflow {
generation: prior_generation,
},
)?;
if generation != expected {
return Err(CommitRecoveryError::UnexpectedGeneration {
expected,
actual: generation,
});
}
}
Err(CommitRecoveryError::NoValidGeneration) => (),
Err(error) => return Err(error),
}
Ok(self
.current
.insert(CommittedGenerationBytes::new(generation, payload)))
}
#[cfg(test)]
pub fn write_corrupt_snapshot(&mut self, generation: u64, payload: Vec<u8>) {
let mut corrupt = CommittedGenerationBytes::new(generation, payload);
corrupt.checksum ^= 1;
self.current = Some(corrupt);
}
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CommitStoreDiagnostic {
pub current: CommitSnapshotDiagnostic,
pub recovery: Result<u64, CommitRecoveryError>,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum CommitSnapshotDiagnostic {
Empty,
Valid {
generation: u64,
},
Invalid {
generation: u64,
},
}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, Deserialize, Eq, thiserror::Error, PartialEq, Serialize)]
pub enum CommitRecoveryError {
#[error("committed payload length {len} exceeds byte limit {limit}")]
PayloadTooLarge {
len: usize,
limit: usize,
},
#[error("no committed ledger generation is present")]
NoValidGeneration,
#[error("current ledger snapshot validation failed at generation {generation}")]
InvalidSnapshot {
generation: u64,
},
#[error("committed ledger generation {generation} cannot be advanced without overflow")]
GenerationOverflow {
generation: u64,
},
#[error("expected committed ledger generation {expected}, got {actual}")]
UnexpectedGeneration {
expected: u64,
actual: u64,
},
}
fn generation_checksum(generation: &CommittedGenerationBytes) -> u64 {
let header = [
generation.generation.to_le_bytes(),
generation.commit_marker.to_le_bytes(),
(generation.payload.len() as u64).to_le_bytes(),
];
let hash = fnv64(FNV_OFFSET, header.as_flattened());
fnv64(hash, &generation.payload)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn rejected_successors_and_oversized_payloads_preserve_current_and_allow_retry() {
let limit = crate::constants::MAX_COMMITTED_PAYLOAD_BYTES;
for populated in [false, true] {
let mut store = SnapshotCommitStore::default();
if populated {
store.commit_payload_at_generation(7, vec![1]).unwrap();
}
let before = store.clone();
let generation = if populated { 8 } else { 0 };
assert_eq!(
store.commit_payload_at_generation(generation, vec![0; limit + 1]),
Err(CommitRecoveryError::PayloadTooLarge {
len: limit + 1,
limit
})
);
assert_eq!(store, before);
if populated {
assert_eq!(
store.commit_payload_at_generation(9, vec![2]),
Err(CommitRecoveryError::UnexpectedGeneration {
expected: 8,
actual: 9
})
);
assert_eq!(store, before);
}
store
.commit_payload_at_generation(generation, vec![3])
.unwrap();
assert_eq!(store.authoritative().unwrap().generation(), generation);
let bytes = crate::test_cbor::to_vec(&store).unwrap();
assert_eq!(
crate::cbor::from_slice_exact::<SnapshotCommitStore>(&bytes).unwrap(),
store
);
}
}
#[test]
fn commits_replace_the_snapshot_and_diagnostics_share_recovery() {
let mut store = SnapshotCommitStore::default();
assert_eq!(
store.diagnostic(),
CommitStoreDiagnostic {
current: CommitSnapshotDiagnostic::Empty,
recovery: Err(CommitRecoveryError::NoValidGeneration),
}
);
for generation in 0..4 {
let payload = vec![u8::try_from(generation).unwrap(); 4];
store
.commit_payload_at_generation(generation, payload.clone())
.unwrap();
let record = store.current().unwrap();
assert_eq!(record.payload(), payload);
assert_eq!(record.commit_marker(), COMMIT_MARKER);
assert_eq!(record.checksum(), generation_checksum(record));
let diagnostic = store.diagnostic();
assert_eq!(
diagnostic,
CommitStoreDiagnostic {
current: CommitSnapshotDiagnostic::Valid { generation },
recovery: Ok(generation),
}
);
let bytes = crate::test_cbor::to_vec(&diagnostic).unwrap();
assert_eq!(
crate::test_cbor::from_slice::<CommitStoreDiagnostic>(&bytes).unwrap(),
diagnostic
);
}
}
#[test]
fn corrupt_current_fails_recovery_and_commit_without_mutation() {
for field in ["checksum", "marker", "payload", "generation"] {
let mut store = SnapshotCommitStore::default();
store
.commit_payload_at_generation(7, vec![1, 2, 3])
.unwrap();
let current = store.current.as_mut().unwrap();
match field {
"checksum" => current.checksum ^= 1,
"marker" => {
current.commit_marker ^= 1;
current.checksum = generation_checksum(current);
}
"payload" => current.payload[0] ^= 1,
_ => current.generation ^= 1,
}
let generation = current.generation();
let before = store.clone();
let expected = CommitRecoveryError::InvalidSnapshot { generation };
assert_eq!(store.authoritative(), Err(expected));
assert_eq!(
store.diagnostic(),
CommitStoreDiagnostic {
current: CommitSnapshotDiagnostic::Invalid { generation },
recovery: Err(expected),
}
);
assert_eq!(
store.commit_payload_at_generation(8, vec![4]),
Err(expected)
);
assert_eq!(store, before);
}
}
#[test]
fn physical_generation_overflow_preserves_current() {
let mut store = SnapshotCommitStore::default();
store
.commit_payload_at_generation(u64::MAX, vec![1])
.unwrap();
let before = store.clone();
assert_eq!(
store.commit_payload_at_generation(0, vec![2]),
Err(CommitRecoveryError::GenerationOverflow {
generation: u64::MAX
})
);
assert_eq!(store, before);
}
#[test]
fn maximum_payload_writer_reader_agree() {
let mut store = SnapshotCommitStore::default();
store
.commit_payload_at_generation(0, vec![0; crate::constants::MAX_COMMITTED_PAYLOAD_BYTES])
.unwrap();
let bytes = crate::test_cbor::to_vec(&store).unwrap();
assert!(bytes.len() <= crate::constants::MAX_LEDGER_RECORD_BYTES);
let decoded: SnapshotCommitStore = crate::cbor::from_slice_exact(&bytes).unwrap();
assert_eq!(decoded, store);
assert!(decoded.authoritative().is_ok());
let oversized = CommittedGenerationBytes::new(
0,
vec![0; crate::constants::MAX_COMMITTED_PAYLOAD_BYTES + 1],
);
assert!(crate::test_cbor::to_vec(&oversized).is_err());
}
#[test]
fn payload_uses_binary_bytes_and_human_readable_arrays() {
let record = CommittedGenerationBytes::new(7, vec![0, 24, 255]);
let bytes = crate::test_cbor::to_vec(&record).unwrap();
let value: ciborium::Value = crate::cbor::from_slice_exact(&bytes).unwrap();
let ciborium::Value::Map(fields) = value else {
panic!("record map")
};
let (_, payload) = fields
.iter()
.find(|(key, _)| key.as_text() == Some("payload"))
.unwrap();
assert_eq!(*payload, ciborium::Value::Bytes(vec![0, 24, 255]));
assert_eq!(
crate::cbor::from_slice_exact::<CommittedGenerationBytes>(&bytes).unwrap(),
record
);
let json = serde_json::to_value(&record).unwrap();
assert_eq!(json["payload"], serde_json::json!([0, 24, 255]));
assert_eq!(
serde_json::from_value::<CommittedGenerationBytes>(json).unwrap(),
record
);
}
}