appcore_sync/sync/
snapshot.rs

1//! Versioned replication snapshot contract and integrity validation.
2
3use crate::sync::codec::bytes_to_hex;
4use crate::sync::error::{SyncError, SyncResult};
5use crate::sync::log::{replication_record_hash, validate_record_size, ReplicationRecord};
6use serde::{Deserialize, Serialize};
7use sha2::{Digest, Sha256};
8use std::collections::HashSet;
9
10/// Stable format version for portable replication snapshots.
11pub const SYNC_SNAPSHOT_FORMAT_V1: u16 = 1;
12
13/// Sequence-addressed event stored in a replication snapshot.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15pub struct ReplicationSnapshotRecord {
16    /// Source replication sequence.
17    pub sequence: u64,
18    /// Opaque serialized event bytes.
19    pub payload: Vec<u8>,
20}
21
22/// Integrity-protected portable image of a replication log.
23#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
24pub struct ReplicationSnapshot {
25    /// Snapshot encoding version.
26    pub format_version: u16,
27    /// One-based index of the final record.
28    pub last_index: u64,
29    /// Replication records in log order.
30    pub records: Vec<ReplicationSnapshotRecord>,
31    /// SHA-256 checksum over version, index, sequences, and payloads.
32    pub checksum: String,
33}
34
35pub(super) fn snapshot_from_records(records: &[ReplicationRecord]) -> ReplicationSnapshot {
36    let records = records
37        .iter()
38        .map(|record| ReplicationSnapshotRecord {
39            sequence: record.sequence,
40            payload: record.payload.clone(),
41        })
42        .collect::<Vec<_>>();
43    let last_index = records.len() as u64;
44    let checksum = snapshot_checksum(SYNC_SNAPSHOT_FORMAT_V1, last_index, &records);
45    ReplicationSnapshot {
46        format_version: SYNC_SNAPSHOT_FORMAT_V1,
47        last_index,
48        records,
49        checksum,
50    }
51}
52
53pub(super) fn validate_snapshot(
54    snapshot: &ReplicationSnapshot,
55) -> SyncResult<Vec<ReplicationRecord>> {
56    if snapshot.format_version != SYNC_SNAPSHOT_FORMAT_V1 {
57        return Err(SyncError::InvalidSnapshot("unsupported format version"));
58    }
59    if snapshot.last_index != snapshot.records.len() as u64 {
60        return Err(SyncError::InvalidSnapshot("last index mismatch"));
61    }
62    if snapshot.checksum
63        != snapshot_checksum(
64            snapshot.format_version,
65            snapshot.last_index,
66            &snapshot.records,
67        )
68    {
69        return Err(SyncError::InvalidSnapshot("checksum mismatch"));
70    }
71    snapshot_records(snapshot)
72}
73
74fn snapshot_records(snapshot: &ReplicationSnapshot) -> SyncResult<Vec<ReplicationRecord>> {
75    let mut sequences = HashSet::new();
76    let mut records = Vec::with_capacity(snapshot.records.len());
77    let mut previous_hash = String::new();
78    for (offset, record) in snapshot.records.iter().enumerate() {
79        validate_record_size(&record.payload)?;
80        if record.sequence > 0 && !sequences.insert(record.sequence) {
81            return Err(SyncError::InvalidSnapshot("duplicate sequence"));
82        }
83        let record_hash = replication_record_hash(&previous_hash, record.sequence, &record.payload);
84        records.push(ReplicationRecord {
85            index: offset + 1,
86            sequence: record.sequence,
87            payload: record.payload.clone(),
88            previous_hash,
89            record_hash: record_hash.clone(),
90        });
91        previous_hash = record_hash;
92    }
93    Ok(records)
94}
95
96fn snapshot_checksum(
97    format_version: u16,
98    last_index: u64,
99    records: &[ReplicationSnapshotRecord],
100) -> String {
101    let mut hasher = Sha256::new();
102    hasher.update(format_version.to_be_bytes());
103    hasher.update(last_index.to_be_bytes());
104    hasher.update((records.len() as u64).to_be_bytes());
105    for record in records {
106        hasher.update(record.sequence.to_be_bytes());
107        hasher.update((record.payload.len() as u64).to_be_bytes());
108        hasher.update(&record.payload);
109    }
110    bytes_to_hex(&hasher.finalize())
111}
appcore_sync/sync/snapshot.rs

appcore_sync/sync/
snapshot.rs
