1use crate::hash::{FNV_OFFSET, fnv64};
2use serde::{Deserialize, Serialize};
3
4const COMMIT_MARKER: u64 = 0x4943_4D45_534E_4150;
6
7#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
21#[serde(deny_unknown_fields)]
22pub struct CommittedGenerationBytes {
23 pub(crate) generation: u64,
25 pub(crate) commit_marker: u64,
27 pub(crate) checksum: u64,
29 #[serde(with = "opaque_payload")]
31 pub(crate) payload: Vec<u8>,
32}
33
34mod opaque_payload {
37 use crate::constants::MAX_COMMITTED_PAYLOAD_BYTES;
38 use serde::{Deserializer, Serialize, Serializer, de::Visitor};
39
40 pub fn serialize<S: Serializer>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
41 if bytes.len() > MAX_COMMITTED_PAYLOAD_BYTES {
42 return Err(serde::ser::Error::custom(
43 "ledger byte string exceeds payload bound",
44 ));
45 }
46 if serializer.is_human_readable() {
47 bytes.serialize(serializer)
48 } else {
49 serializer.serialize_bytes(bytes)
50 }
51 }
52
53 pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Vec<u8>, D::Error> {
54 struct Bytes;
55 impl Visitor<'_> for Bytes {
56 type Value = Vec<u8>;
57
58 fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59 f.write_str("a bounded opaque ledger byte string")
60 }
61
62 fn visit_bytes<E: serde::de::Error>(self, bytes: &[u8]) -> Result<Self::Value, E> {
63 if bytes.len() > MAX_COMMITTED_PAYLOAD_BYTES {
64 return Err(E::custom("ledger byte string exceeds payload bound"));
65 }
66 Ok(bytes.to_vec())
67 }
68
69 fn visit_byte_buf<E: serde::de::Error>(self, bytes: Vec<u8>) -> Result<Self::Value, E> {
70 if bytes.len() > MAX_COMMITTED_PAYLOAD_BYTES {
71 return Err(E::custom("ledger byte string exceeds payload bound"));
72 }
73 Ok(bytes)
74 }
75 }
76 if deserializer.is_human_readable() {
77 return crate::cbor::deserialize_bounded_vec::<D, u8, MAX_COMMITTED_PAYLOAD_BYTES>(
78 deserializer,
79 );
80 }
81 deserializer.deserialize_byte_buf(Bytes)
82 }
83}
84
85impl CommittedGenerationBytes {
86 #[must_use]
88 pub fn new(generation: u64, payload: Vec<u8>) -> Self {
89 let mut record = Self {
90 generation,
91 commit_marker: COMMIT_MARKER,
92 checksum: 0,
93 payload,
94 };
95 record.checksum = generation_checksum(&record);
96 record
97 }
98
99 #[must_use]
101 pub const fn generation(&self) -> u64 {
102 self.generation
103 }
104
105 #[must_use]
111 #[cfg(test)]
112 pub const fn commit_marker(&self) -> u64 {
113 self.commit_marker
114 }
115
116 #[must_use]
121 #[cfg(test)]
122 pub const fn checksum(&self) -> u64 {
123 self.checksum
124 }
125
126 #[must_use]
128 pub fn payload(&self) -> &[u8] {
129 &self.payload
130 }
131
132 #[must_use]
134 pub fn validates(&self) -> bool {
135 self.commit_marker == COMMIT_MARKER && self.checksum == generation_checksum(self)
136 }
137}
138
139#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
150#[serde(deny_unknown_fields)]
151pub struct SnapshotCommitStore {
152 #[serde(deserialize_with = "crate::cbor::deserialize_present_option")]
153 pub(crate) current: Option<CommittedGenerationBytes>,
154}
155
156impl SnapshotCommitStore {
157 pub const fn current(&self) -> Option<&CommittedGenerationBytes> {
159 self.current.as_ref()
160 }
161
162 pub fn authoritative(&self) -> Result<&CommittedGenerationBytes, CommitRecoveryError> {
164 let current = self
165 .current()
166 .ok_or(CommitRecoveryError::NoValidGeneration)?;
167 if !current.validates() {
168 return Err(CommitRecoveryError::InvalidSnapshot {
169 generation: current.generation(),
170 });
171 }
172 Ok(current)
173 }
174
175 pub(crate) fn authoritative_with_diagnostic(
176 &self,
177 ) -> (
178 Result<&CommittedGenerationBytes, CommitRecoveryError>,
179 CommitSnapshotDiagnostic,
180 ) {
181 let recovery = self.authoritative();
182 let current = match self.current() {
183 None => CommitSnapshotDiagnostic::Empty,
184 Some(record) if recovery.is_ok() => CommitSnapshotDiagnostic::Valid {
185 generation: record.generation(),
186 },
187 Some(record) => CommitSnapshotDiagnostic::Invalid {
188 generation: record.generation(),
189 },
190 };
191 (recovery, current)
192 }
193
194 pub fn diagnostic(&self) -> CommitSnapshotDiagnostic {
196 self.authoritative_with_diagnostic().1
197 }
198
199 pub fn commit_payload_at_generation(
202 &mut self,
203 generation: u64,
204 payload: Vec<u8>,
205 ) -> Result<&CommittedGenerationBytes, CommitRecoveryError> {
206 if payload.len() > crate::constants::MAX_COMMITTED_PAYLOAD_BYTES {
207 return Err(CommitRecoveryError::PayloadTooLarge {
208 len: payload.len(),
209 limit: crate::constants::MAX_COMMITTED_PAYLOAD_BYTES,
210 });
211 }
212 match self.authoritative() {
213 Ok(prior) => {
214 let prior_generation = prior.generation();
215 let expected = prior_generation.checked_add(1).ok_or(
216 CommitRecoveryError::GenerationOverflow {
217 generation: prior_generation,
218 },
219 )?;
220 if generation != expected {
221 return Err(CommitRecoveryError::UnexpectedGeneration {
222 expected,
223 actual: generation,
224 });
225 }
226 }
227 Err(CommitRecoveryError::NoValidGeneration) => (),
229 Err(error) => return Err(error),
230 }
231 Ok(self
232 .current
233 .insert(CommittedGenerationBytes::new(generation, payload)))
234 }
235
236 #[cfg(test)]
237 pub fn write_corrupt_snapshot(&mut self, generation: u64, payload: Vec<u8>) {
238 let mut corrupt = CommittedGenerationBytes::new(generation, payload);
239 corrupt.checksum ^= 1;
240 self.current = Some(corrupt);
241 }
242}
243
244#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
251pub enum CommitSnapshotDiagnostic {
252 Empty,
254 Valid {
256 generation: u64,
258 },
259 Invalid {
261 generation: u64,
263 },
264}
265
266#[non_exhaustive]
273#[derive(Clone, Copy, Debug, Deserialize, Eq, thiserror::Error, PartialEq, Serialize)]
274pub enum CommitRecoveryError {
275 #[error("committed payload length {len} exceeds byte limit {limit}")]
277 PayloadTooLarge {
278 len: usize,
280 limit: usize,
282 },
283 #[error("no committed ledger generation is present")]
285 NoValidGeneration,
286 #[error("current ledger snapshot validation failed at generation {generation}")]
288 InvalidSnapshot {
289 generation: u64,
291 },
292 #[error("committed ledger generation {generation} cannot be advanced without overflow")]
294 GenerationOverflow {
295 generation: u64,
297 },
298 #[error("expected committed ledger generation {expected}, got {actual}")]
300 UnexpectedGeneration {
301 expected: u64,
303 actual: u64,
305 },
306}
307
308fn generation_checksum(generation: &CommittedGenerationBytes) -> u64 {
309 let header = [
310 generation.generation.to_le_bytes(),
311 generation.commit_marker.to_le_bytes(),
312 (generation.payload.len() as u64).to_le_bytes(),
313 ];
314 let hash = fnv64(FNV_OFFSET, header.as_flattened());
315 fnv64(hash, &generation.payload)
316}
317
318#[cfg(test)]
319mod tests {
320 use super::*;
321
322 #[test]
323 fn rejected_successors_and_oversized_payloads_preserve_current_and_allow_retry() {
324 let limit = crate::constants::MAX_COMMITTED_PAYLOAD_BYTES;
325 for populated in [false, true] {
326 let mut store = SnapshotCommitStore::default();
327 if populated {
328 store.commit_payload_at_generation(7, vec![1]).unwrap();
329 }
330 let before = store.clone();
331 let generation = if populated { 8 } else { 0 };
332 assert_eq!(
333 store.commit_payload_at_generation(generation, vec![0; limit + 1]),
334 Err(CommitRecoveryError::PayloadTooLarge {
335 len: limit + 1,
336 limit
337 })
338 );
339 assert_eq!(store, before);
340 if populated {
341 assert_eq!(
342 store.commit_payload_at_generation(9, vec![2]),
343 Err(CommitRecoveryError::UnexpectedGeneration {
344 expected: 8,
345 actual: 9
346 })
347 );
348 assert_eq!(store, before);
349 }
350 store
351 .commit_payload_at_generation(generation, vec![3])
352 .unwrap();
353 assert_eq!(store.authoritative().unwrap().generation(), generation);
354 let bytes = crate::test_cbor::to_vec(&store).unwrap();
355 assert_eq!(
356 crate::cbor::from_slice_exact::<SnapshotCommitStore>(&bytes).unwrap(),
357 store
358 );
359 }
360 }
361
362 #[test]
363 fn commits_replace_the_snapshot_and_diagnostics_share_recovery() {
364 let mut store = SnapshotCommitStore::default();
365 assert_eq!(store.diagnostic(), CommitSnapshotDiagnostic::Empty);
366 for generation in 0..4 {
367 let payload = vec![u8::try_from(generation).unwrap(); 4];
368 store
369 .commit_payload_at_generation(generation, payload.clone())
370 .unwrap();
371 let record = store.current().unwrap();
372 assert_eq!(record.payload(), payload);
373 assert_eq!(record.commit_marker(), COMMIT_MARKER);
374 assert_eq!(record.checksum(), generation_checksum(record));
375 let diagnostic = store.diagnostic();
376 assert_eq!(diagnostic, CommitSnapshotDiagnostic::Valid { generation });
377 let bytes = crate::test_cbor::to_vec(&diagnostic).unwrap();
378 assert_eq!(
379 crate::test_cbor::from_slice::<CommitSnapshotDiagnostic>(&bytes).unwrap(),
380 diagnostic
381 );
382 }
383 }
384
385 #[test]
386 fn corrupt_current_fails_recovery_and_commit_without_mutation() {
387 for field in ["checksum", "marker", "payload", "generation"] {
388 let mut store = SnapshotCommitStore::default();
389 store
390 .commit_payload_at_generation(7, vec![1, 2, 3])
391 .unwrap();
392 let current = store.current.as_mut().unwrap();
393 match field {
394 "checksum" => current.checksum ^= 1,
395 "marker" => {
396 current.commit_marker ^= 1;
397 current.checksum = generation_checksum(current);
398 }
399 "payload" => current.payload[0] ^= 1,
400 _ => current.generation ^= 1,
401 }
402 let generation = current.generation();
403 let before = store.clone();
404 let expected = CommitRecoveryError::InvalidSnapshot { generation };
405 assert_eq!(store.authoritative(), Err(expected));
406 assert_eq!(
407 store.diagnostic(),
408 CommitSnapshotDiagnostic::Invalid { generation }
409 );
410 assert_eq!(
411 store.commit_payload_at_generation(8, vec![4]),
412 Err(expected)
413 );
414 assert_eq!(store, before);
415 }
416 }
417
418 #[test]
419 fn physical_generation_overflow_preserves_current() {
420 let mut store = SnapshotCommitStore::default();
421 store
422 .commit_payload_at_generation(u64::MAX, vec![1])
423 .unwrap();
424 let before = store.clone();
425 assert_eq!(
426 store.commit_payload_at_generation(0, vec![2]),
427 Err(CommitRecoveryError::GenerationOverflow {
428 generation: u64::MAX
429 })
430 );
431 assert_eq!(store, before);
432 }
433
434 #[test]
435 fn maximum_payload_writer_reader_agree() {
436 let mut store = SnapshotCommitStore::default();
437 store
438 .commit_payload_at_generation(0, vec![0; crate::constants::MAX_COMMITTED_PAYLOAD_BYTES])
439 .unwrap();
440 let bytes = crate::test_cbor::to_vec(&store).unwrap();
441 assert!(bytes.len() <= crate::constants::MAX_LEDGER_RECORD_BYTES);
442 let decoded: SnapshotCommitStore = crate::cbor::from_slice_exact(&bytes).unwrap();
443 assert_eq!(decoded, store);
444 assert!(decoded.authoritative().is_ok());
445 let oversized = CommittedGenerationBytes::new(
446 0,
447 vec![0; crate::constants::MAX_COMMITTED_PAYLOAD_BYTES + 1],
448 );
449 assert!(crate::test_cbor::to_vec(&oversized).is_err());
450 }
451
452 #[test]
453 fn payload_uses_binary_bytes_and_human_readable_arrays() {
454 let record = CommittedGenerationBytes::new(7, vec![0, 24, 255]);
455 let bytes = crate::test_cbor::to_vec(&record).unwrap();
456 let value: ciborium::Value = crate::cbor::from_slice_exact(&bytes).unwrap();
457 let ciborium::Value::Map(fields) = value else {
458 panic!("record map")
459 };
460 let (_, payload) = fields
461 .iter()
462 .find(|(key, _)| key.as_text() == Some("payload"))
463 .unwrap();
464 assert_eq!(*payload, ciborium::Value::Bytes(vec![0, 24, 255]));
465 assert_eq!(
466 crate::cbor::from_slice_exact::<CommittedGenerationBytes>(&bytes).unwrap(),
467 record
468 );
469 let json = serde_json::to_value(&record).unwrap();
470 assert_eq!(json["payload"], serde_json::json!([0, 24, 255]));
471 assert_eq!(
472 serde_json::from_value::<CommittedGenerationBytes>(json).unwrap(),
473 record
474 );
475 }
476}