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 CommitStoreDiagnostic,
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 let diagnostic = CommitStoreDiagnostic {
192 current,
193 recovery: recovery.map(CommittedGenerationBytes::generation),
194 };
195 (recovery, diagnostic)
196 }
197
198 pub fn diagnostic(&self) -> CommitStoreDiagnostic {
200 self.authoritative_with_diagnostic().1
201 }
202
203 pub fn commit_payload_at_generation(
206 &mut self,
207 generation: u64,
208 payload: Vec<u8>,
209 ) -> Result<&CommittedGenerationBytes, CommitRecoveryError> {
210 if payload.len() > crate::constants::MAX_COMMITTED_PAYLOAD_BYTES {
211 return Err(CommitRecoveryError::PayloadTooLarge {
212 len: payload.len(),
213 limit: crate::constants::MAX_COMMITTED_PAYLOAD_BYTES,
214 });
215 }
216 match self.authoritative() {
217 Ok(prior) => {
218 let prior_generation = prior.generation();
219 let expected = prior_generation.checked_add(1).ok_or(
220 CommitRecoveryError::GenerationOverflow {
221 generation: prior_generation,
222 },
223 )?;
224 if generation != expected {
225 return Err(CommitRecoveryError::UnexpectedGeneration {
226 expected,
227 actual: generation,
228 });
229 }
230 }
231 Err(CommitRecoveryError::NoValidGeneration) => (),
233 Err(error) => return Err(error),
234 }
235 Ok(self
236 .current
237 .insert(CommittedGenerationBytes::new(generation, payload)))
238 }
239
240 #[cfg(test)]
241 pub fn write_corrupt_snapshot(&mut self, generation: u64, payload: Vec<u8>) {
242 let mut corrupt = CommittedGenerationBytes::new(generation, payload);
243 corrupt.checksum ^= 1;
244 self.current = Some(corrupt);
245 }
246}
247
248#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
256#[serde(deny_unknown_fields)]
257pub struct CommitStoreDiagnostic {
258 pub current: CommitSnapshotDiagnostic,
260 pub recovery: Result<u64, CommitRecoveryError>,
262}
263
264#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
271pub enum CommitSnapshotDiagnostic {
272 Empty,
274 Valid {
276 generation: u64,
278 },
279 Invalid {
281 generation: u64,
283 },
284}
285
286#[non_exhaustive]
293#[derive(Clone, Copy, Debug, Deserialize, Eq, thiserror::Error, PartialEq, Serialize)]
294pub enum CommitRecoveryError {
295 #[error("committed payload length {len} exceeds byte limit {limit}")]
297 PayloadTooLarge {
298 len: usize,
300 limit: usize,
302 },
303 #[error("no committed ledger generation is present")]
305 NoValidGeneration,
306 #[error("current ledger snapshot validation failed at generation {generation}")]
308 InvalidSnapshot {
309 generation: u64,
311 },
312 #[error("committed ledger generation {generation} cannot be advanced without overflow")]
314 GenerationOverflow {
315 generation: u64,
317 },
318 #[error("expected committed ledger generation {expected}, got {actual}")]
320 UnexpectedGeneration {
321 expected: u64,
323 actual: u64,
325 },
326}
327
328fn generation_checksum(generation: &CommittedGenerationBytes) -> u64 {
329 let header = [
330 generation.generation.to_le_bytes(),
331 generation.commit_marker.to_le_bytes(),
332 (generation.payload.len() as u64).to_le_bytes(),
333 ];
334 let hash = fnv64(FNV_OFFSET, header.as_flattened());
335 fnv64(hash, &generation.payload)
336}
337
338#[cfg(test)]
339mod tests {
340 use super::*;
341
342 #[test]
343 fn rejected_successors_and_oversized_payloads_preserve_current_and_allow_retry() {
344 let limit = crate::constants::MAX_COMMITTED_PAYLOAD_BYTES;
345 for populated in [false, true] {
346 let mut store = SnapshotCommitStore::default();
347 if populated {
348 store.commit_payload_at_generation(7, vec![1]).unwrap();
349 }
350 let before = store.clone();
351 let generation = if populated { 8 } else { 0 };
352 assert_eq!(
353 store.commit_payload_at_generation(generation, vec![0; limit + 1]),
354 Err(CommitRecoveryError::PayloadTooLarge {
355 len: limit + 1,
356 limit
357 })
358 );
359 assert_eq!(store, before);
360 if populated {
361 assert_eq!(
362 store.commit_payload_at_generation(9, vec![2]),
363 Err(CommitRecoveryError::UnexpectedGeneration {
364 expected: 8,
365 actual: 9
366 })
367 );
368 assert_eq!(store, before);
369 }
370 store
371 .commit_payload_at_generation(generation, vec![3])
372 .unwrap();
373 assert_eq!(store.authoritative().unwrap().generation(), generation);
374 let bytes = crate::test_cbor::to_vec(&store).unwrap();
375 assert_eq!(
376 crate::cbor::from_slice_exact::<SnapshotCommitStore>(&bytes).unwrap(),
377 store
378 );
379 }
380 }
381
382 #[test]
383 fn commits_replace_the_snapshot_and_diagnostics_share_recovery() {
384 let mut store = SnapshotCommitStore::default();
385 assert_eq!(
386 store.diagnostic(),
387 CommitStoreDiagnostic {
388 current: CommitSnapshotDiagnostic::Empty,
389 recovery: Err(CommitRecoveryError::NoValidGeneration),
390 }
391 );
392 for generation in 0..4 {
393 let payload = vec![u8::try_from(generation).unwrap(); 4];
394 store
395 .commit_payload_at_generation(generation, payload.clone())
396 .unwrap();
397 let record = store.current().unwrap();
398 assert_eq!(record.payload(), payload);
399 assert_eq!(record.commit_marker(), COMMIT_MARKER);
400 assert_eq!(record.checksum(), generation_checksum(record));
401 let diagnostic = store.diagnostic();
402 assert_eq!(
403 diagnostic,
404 CommitStoreDiagnostic {
405 current: CommitSnapshotDiagnostic::Valid { generation },
406 recovery: Ok(generation),
407 }
408 );
409 let bytes = crate::test_cbor::to_vec(&diagnostic).unwrap();
410 assert_eq!(
411 crate::test_cbor::from_slice::<CommitStoreDiagnostic>(&bytes).unwrap(),
412 diagnostic
413 );
414 }
415 }
416
417 #[test]
418 fn corrupt_current_fails_recovery_and_commit_without_mutation() {
419 for field in ["checksum", "marker", "payload", "generation"] {
420 let mut store = SnapshotCommitStore::default();
421 store
422 .commit_payload_at_generation(7, vec![1, 2, 3])
423 .unwrap();
424 let current = store.current.as_mut().unwrap();
425 match field {
426 "checksum" => current.checksum ^= 1,
427 "marker" => {
428 current.commit_marker ^= 1;
429 current.checksum = generation_checksum(current);
430 }
431 "payload" => current.payload[0] ^= 1,
432 _ => current.generation ^= 1,
433 }
434 let generation = current.generation();
435 let before = store.clone();
436 let expected = CommitRecoveryError::InvalidSnapshot { generation };
437 assert_eq!(store.authoritative(), Err(expected));
438 assert_eq!(
439 store.diagnostic(),
440 CommitStoreDiagnostic {
441 current: CommitSnapshotDiagnostic::Invalid { generation },
442 recovery: Err(expected),
443 }
444 );
445 assert_eq!(
446 store.commit_payload_at_generation(8, vec![4]),
447 Err(expected)
448 );
449 assert_eq!(store, before);
450 }
451 }
452
453 #[test]
454 fn physical_generation_overflow_preserves_current() {
455 let mut store = SnapshotCommitStore::default();
456 store
457 .commit_payload_at_generation(u64::MAX, vec![1])
458 .unwrap();
459 let before = store.clone();
460 assert_eq!(
461 store.commit_payload_at_generation(0, vec![2]),
462 Err(CommitRecoveryError::GenerationOverflow {
463 generation: u64::MAX
464 })
465 );
466 assert_eq!(store, before);
467 }
468
469 #[test]
470 fn maximum_payload_writer_reader_agree() {
471 let mut store = SnapshotCommitStore::default();
472 store
473 .commit_payload_at_generation(0, vec![0; crate::constants::MAX_COMMITTED_PAYLOAD_BYTES])
474 .unwrap();
475 let bytes = crate::test_cbor::to_vec(&store).unwrap();
476 assert!(bytes.len() <= crate::constants::MAX_LEDGER_RECORD_BYTES);
477 let decoded: SnapshotCommitStore = crate::cbor::from_slice_exact(&bytes).unwrap();
478 assert_eq!(decoded, store);
479 assert!(decoded.authoritative().is_ok());
480 let oversized = CommittedGenerationBytes::new(
481 0,
482 vec![0; crate::constants::MAX_COMMITTED_PAYLOAD_BYTES + 1],
483 );
484 assert!(crate::test_cbor::to_vec(&oversized).is_err());
485 }
486
487 #[test]
488 fn payload_uses_binary_bytes_and_human_readable_arrays() {
489 let record = CommittedGenerationBytes::new(7, vec![0, 24, 255]);
490 let bytes = crate::test_cbor::to_vec(&record).unwrap();
491 let value: ciborium::Value = crate::cbor::from_slice_exact(&bytes).unwrap();
492 let ciborium::Value::Map(fields) = value else {
493 panic!("record map")
494 };
495 let (_, payload) = fields
496 .iter()
497 .find(|(key, _)| key.as_text() == Some("payload"))
498 .unwrap();
499 assert_eq!(*payload, ciborium::Value::Bytes(vec![0, 24, 255]));
500 assert_eq!(
501 crate::cbor::from_slice_exact::<CommittedGenerationBytes>(&bytes).unwrap(),
502 record
503 );
504 let json = serde_json::to_value(&record).unwrap();
505 assert_eq!(json["payload"], serde_json::json!([0, 24, 255]));
506 assert_eq!(
507 serde_json::from_value::<CommittedGenerationBytes>(json).unwrap(),
508 record
509 );
510 }
511}