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