appcore_sync/sync/
snapshot.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: snapshot.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/23 23:50:45 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/23 23:50:45 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Versioned replication snapshot contract and integrity validation.
12
13use crate::sync::codec::bytes_to_hex;
14use crate::sync::error::{SyncError, SyncResult};
15use crate::sync::log::{
16    replication_record_hash, validate_record_count, validate_record_size, ReplicationRecord,
17};
18use serde::{Deserialize, Serialize};
19use sha2::{Digest, Sha256};
20use std::collections::HashSet;
21
22/// Stable format version for portable replication snapshots.
23pub const SYNC_SNAPSHOT_FORMAT_V1: u16 = 1;
24
25/// Sequence-addressed event stored in a replication snapshot.
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27pub struct ReplicationSnapshotRecord {
28    /// Source replication sequence.
29    pub sequence: u64,
30    /// Opaque serialized event bytes.
31    pub payload: Vec<u8>,
32}
33
34/// Integrity-protected portable image of a replication log.
35#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
36pub struct ReplicationSnapshot {
37    /// Snapshot encoding version.
38    pub format_version: u16,
39    /// One-based index of the final record.
40    pub last_index: u64,
41    /// Replication records in log order.
42    pub records: Vec<ReplicationSnapshotRecord>,
43    /// SHA-256 checksum over version, index, sequences, and payloads.
44    pub checksum: String,
45}
46
47impl ReplicationSnapshot {
48    /// Creates a validated snapshot while taking ownership of every payload.
49    ///
50    /// Payload allocations are moved into the snapshot instead of cloned. The
51    /// same record-count, record-size, sequence-uniqueness and checksum
52    /// contracts used by durable replication logs are applied.
53    pub fn try_from_records(records: impl IntoIterator<Item = (u64, Vec<u8>)>) -> SyncResult<Self> {
54        snapshot_from_payloads(records.into_iter().map(Ok))
55    }
56
57    /// Validates this snapshot without cloning record payloads.
58    pub fn validate(&self) -> SyncResult<()> {
59        validate_snapshot_contract(self)
60    }
61}
62
63pub(super) fn snapshot_from_records(records: &[ReplicationRecord]) -> ReplicationSnapshot {
64    let records = records
65        .iter()
66        .map(|record| ReplicationSnapshotRecord {
67            sequence: record.sequence,
68            payload: record.payload.clone(),
69        })
70        .collect::<Vec<_>>();
71    let last_index = records.len() as u64;
72    let checksum = snapshot_checksum(SYNC_SNAPSHOT_FORMAT_V1, last_index, &records);
73    ReplicationSnapshot {
74        format_version: SYNC_SNAPSHOT_FORMAT_V1,
75        last_index,
76        records,
77        checksum,
78    }
79}
80
81pub(super) fn validate_snapshot(
82    snapshot: &ReplicationSnapshot,
83) -> SyncResult<Vec<ReplicationRecord>> {
84    validate_snapshot_contract(snapshot)?;
85    snapshot_records(snapshot)
86}
87
88pub(super) fn validate_snapshot_contract(snapshot: &ReplicationSnapshot) -> SyncResult<()> {
89    if snapshot.format_version != SYNC_SNAPSHOT_FORMAT_V1 {
90        return Err(SyncError::InvalidSnapshot("unsupported format version"));
91    }
92    if snapshot.last_index != snapshot.records.len() as u64 {
93        return Err(SyncError::InvalidSnapshot("last index mismatch"));
94    }
95    validate_record_count(snapshot.records.len())?;
96    if snapshot.checksum
97        != snapshot_checksum(
98            snapshot.format_version,
99            snapshot.last_index,
100            &snapshot.records,
101        )
102    {
103        return Err(SyncError::InvalidSnapshot("checksum mismatch"));
104    }
105    let mut sequences = HashSet::new();
106    for record in &snapshot.records {
107        validate_record_size(&record.payload)?;
108        if record.sequence > 0 && !sequences.insert(record.sequence) {
109            return Err(SyncError::InvalidSnapshot("duplicate sequence"));
110        }
111    }
112    Ok(())
113}
114
115pub(super) fn snapshot_from_payloads(
116    records: impl Iterator<Item = SyncResult<(u64, Vec<u8>)>>,
117) -> SyncResult<ReplicationSnapshot> {
118    let mut output = Vec::new();
119    let mut sequences = HashSet::new();
120    for record in records {
121        validate_record_count(output.len().saturating_add(1))?;
122        let (sequence, payload) = record?;
123        validate_record_size(&payload)?;
124        if sequence > 0 && !sequences.insert(sequence) {
125            return Err(SyncError::InvalidSnapshot("duplicate sequence"));
126        }
127        output.push(ReplicationSnapshotRecord { sequence, payload });
128    }
129    let last_index = output.len() as u64;
130    let checksum = snapshot_checksum(SYNC_SNAPSHOT_FORMAT_V1, last_index, &output);
131    Ok(ReplicationSnapshot {
132        format_version: SYNC_SNAPSHOT_FORMAT_V1,
133        last_index,
134        records: output,
135        checksum,
136    })
137}
138
139fn snapshot_records(snapshot: &ReplicationSnapshot) -> SyncResult<Vec<ReplicationRecord>> {
140    let mut sequences = HashSet::new();
141    let mut records = Vec::with_capacity(snapshot.records.len());
142    let mut previous_hash = String::new();
143    for (offset, record) in snapshot.records.iter().enumerate() {
144        validate_record_size(&record.payload)?;
145        if record.sequence > 0 && !sequences.insert(record.sequence) {
146            return Err(SyncError::InvalidSnapshot("duplicate sequence"));
147        }
148        let record_hash = replication_record_hash(&previous_hash, record.sequence, &record.payload);
149        records.push(ReplicationRecord {
150            index: offset + 1,
151            sequence: record.sequence,
152            payload: record.payload.clone(),
153            previous_hash,
154            record_hash: record_hash.clone(),
155        });
156        previous_hash = record_hash;
157    }
158    Ok(records)
159}
160
161pub(super) fn into_replication_records(
162    snapshot: ReplicationSnapshot,
163) -> SyncResult<Vec<ReplicationRecord>> {
164    validate_snapshot_contract(&snapshot)?;
165    let mut sequences = HashSet::new();
166    let mut records = Vec::with_capacity(snapshot.records.len());
167    let mut previous_hash = String::new();
168    for (offset, record) in snapshot.records.into_iter().enumerate() {
169        if record.sequence > 0 && !sequences.insert(record.sequence) {
170            return Err(SyncError::InvalidSnapshot("duplicate sequence"));
171        }
172        validate_record_size(&record.payload)?;
173        let record_hash = replication_record_hash(&previous_hash, record.sequence, &record.payload);
174        records.push(ReplicationRecord {
175            index: offset + 1,
176            sequence: record.sequence,
177            payload: record.payload,
178            previous_hash,
179            record_hash: record_hash.clone(),
180        });
181        previous_hash = record_hash;
182    }
183    Ok(records)
184}
185
186fn snapshot_checksum(
187    format_version: u16,
188    last_index: u64,
189    records: &[ReplicationSnapshotRecord],
190) -> String {
191    let mut hasher = Sha256::new();
192    hasher.update(format_version.to_be_bytes());
193    hasher.update(last_index.to_be_bytes());
194    hasher.update((records.len() as u64).to_be_bytes());
195    for record in records {
196        hasher.update(record.sequence.to_be_bytes());
197        hasher.update((record.payload.len() as u64).to_be_bytes());
198        hasher.update(&record.payload);
199    }
200    bytes_to_hex(&hasher.finalize())
201}
appcore_sync/sync/snapshot.rs

appcore_sync/sync/
snapshot.rs
