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

appcore_sync/sync/
snapshot.rs
