arete_server/snapshot/
envelope.rs

1//! On-disk snapshot format: a JSON header (readable with `head`/`jq` after
2//! skipping 12 bytes) followed by a zstd-compressed JSON payload.
3//!
4//! Layout: `b"ARSNAP01" | u32-le header_len | header JSON | zstd(payload JSON)`.
5
6use anyhow::{bail, Context, Result};
7use arete_interpreter::snapshot::VmSnapshot;
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use std::collections::BTreeMap;
11
12const MAGIC: &[u8; 8] = b"ARSNAP01";
13const ZSTD_LEVEL: i32 = 3;
14/// Guards against decompression bombs from a corrupted or hostile store.
15const MAX_PAYLOAD_BYTES: usize = 1 << 30;
16
17/// A versioned, normalized description of data that a snapshot persists.
18#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
19pub struct SnapshotContract {
20    pub schema: String,
21    pub hash: String,
22}
23
24/// Plain-JSON metadata used to validate a snapshot before decompressing it.
25#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct SnapshotHeader {
27    /// [`arete_interpreter::snapshot::SNAPSHOT_FORMAT_VERSION`] at write time.
28    pub format_version: u32,
29    /// Fingerprint of the compiled `MultiEntityBytecode`. A mismatch means the
30    /// stack's logic changed; the snapshot is discarded (cold start).
31    pub bytecode_hash: String,
32    /// Durable entity structure, independent of handler opcode ordering.
33    /// Absent on snapshots written before contract-aware restore.
34    #[serde(default, skip_serializing_if = "Option::is_none")]
35    pub state_contract: Option<SnapshotContract>,
36    /// Runtime view/cache structure. Kept separate so a future migration can
37    /// make an explicit decision about rebuilding projections.
38    #[serde(default, skip_serializing_if = "Option::is_none")]
39    pub projection_contract: Option<SnapshotContract>,
40    pub program_ids: Vec<String>,
41    /// Highest slot among mutation batches the projector had applied when the
42    /// VM was dumped. Safe `from_slot` resume point.
43    pub resume_watermark: u64,
44    /// Slot-subscription tip at dump time (diagnostics / staleness clamping).
45    pub observed_slot: u64,
46    pub created_at_epoch_ms: u64,
47    /// Entity-name -> row count, for logging and debugging.
48    #[serde(default)]
49    pub entry_counts: BTreeMap<String, u64>,
50    /// What wrote this snapshot. A shutdown snapshot is taken at a consistency
51    /// cut with nothing in flight, so its offsets are exactly what was
52    /// published; a periodic one can be behind by up to the write interval.
53    /// Absent in snapshots written before the distinction mattered, which are
54    /// treated as the unclean case.
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub trigger: Option<crate::snapshot::SnapshotTrigger>,
57}
58
59/// The compressed body of a snapshot.
60#[derive(Debug, Clone, Serialize, Deserialize)]
61pub struct SnapshotPayload {
62    pub vm: VmSnapshot,
63    /// Per view id: `(entity_key, entity)` pairs, most-recently-used first.
64    pub entity_cache: Vec<(String, Vec<(String, Value)>)>,
65    /// Explicit account-lifetime ordering. `None` identifies a snapshot from
66    /// before account write_version was separated from instruction txn_index.
67    #[serde(default)]
68    pub entity_lifetimes: Option<crate::cache::EntityLifetimes>,
69    /// Legacy `_seq`-based deletion barriers, read only for migration.
70    #[serde(default)]
71    pub entity_tombstones: crate::cache::EntityTombstones,
72    /// Retained event tape per view. Absent in snapshots written before
73    /// replayable subscriptions existed, which restore with an empty tape.
74    #[serde(default)]
75    pub journal: crate::journal::JournalSnapshot,
76}
77
78pub fn encode(header: &SnapshotHeader, payload: &SnapshotPayload) -> Result<Vec<u8>> {
79    let header_json = serde_json::to_vec(header).context("serialize snapshot header")?;
80    let payload_json = serde_json::to_vec(payload).context("serialize snapshot payload")?;
81    let compressed =
82        zstd::encode_all(payload_json.as_slice(), ZSTD_LEVEL).context("compress snapshot")?;
83
84    let mut bytes = Vec::with_capacity(MAGIC.len() + 4 + header_json.len() + compressed.len());
85    bytes.extend_from_slice(MAGIC);
86    bytes.extend_from_slice(&(header_json.len() as u32).to_le_bytes());
87    bytes.extend_from_slice(&header_json);
88    bytes.extend_from_slice(&compressed);
89    Ok(bytes)
90}
91
92/// Parse only the header, without touching the compressed payload.
93pub fn decode_header(bytes: &[u8]) -> Result<SnapshotHeader> {
94    if bytes.len() < MAGIC.len() + 4 {
95        bail!("snapshot blob truncated ({} bytes)", bytes.len());
96    }
97    if &bytes[..MAGIC.len()] != MAGIC {
98        bail!("snapshot blob has wrong magic");
99    }
100    let header_len =
101        u32::from_le_bytes(bytes[MAGIC.len()..MAGIC.len() + 4].try_into().unwrap()) as usize;
102    let header_start = MAGIC.len() + 4;
103    let header_end = header_start
104        .checked_add(header_len)
105        .filter(|end| *end <= bytes.len())
106        .context("snapshot header length out of bounds")?;
107    serde_json::from_slice(&bytes[header_start..header_end]).context("parse snapshot header")
108}
109
110pub fn decode_payload(bytes: &[u8]) -> Result<SnapshotPayload> {
111    // Re-derive the payload offset the same way decode_header does.
112    let header_len =
113        u32::from_le_bytes(bytes[MAGIC.len()..MAGIC.len() + 4].try_into().unwrap()) as usize;
114    let payload_start = MAGIC.len() + 4 + header_len;
115
116    use std::io::Read;
117    let decoder = zstd::Decoder::new(&bytes[payload_start..])?;
118    let mut payload_json = Vec::new();
119    decoder
120        .take(MAX_PAYLOAD_BYTES as u64 + 1)
121        .read_to_end(&mut payload_json)
122        .context("decompress snapshot payload")?;
123    if payload_json.len() > MAX_PAYLOAD_BYTES {
124        bail!("snapshot payload exceeds {} bytes", MAX_PAYLOAD_BYTES);
125    }
126    serde_json::from_slice(&payload_json).context("parse snapshot payload")
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132
133    fn sample() -> (SnapshotHeader, SnapshotPayload) {
134        let header = SnapshotHeader {
135            format_version: arete_interpreter::snapshot::SNAPSHOT_FORMAT_VERSION,
136            bytecode_hash: "abc123".to_string(),
137            state_contract: Some(SnapshotContract {
138                schema: "arete.snapshot-state-contract/v1".to_string(),
139                hash: "state123".to_string(),
140            }),
141            projection_contract: Some(SnapshotContract {
142                schema: "arete.snapshot-projection-contract/v1".to_string(),
143                hash: "projection123".to_string(),
144            }),
145            program_ids: vec!["Program111".to_string()],
146            resume_watermark: 42,
147            observed_slot: 50,
148            created_at_epoch_ms: 1_000,
149            trigger: Some(crate::snapshot::SnapshotTrigger::Shutdown),
150            entry_counts: BTreeMap::new(),
151        };
152        let payload = SnapshotPayload {
153            vm: VmSnapshot::default(),
154            entity_lifetimes: Some(vec![(
155                "tokens/list".to_string(),
156                "key1".to_string(),
157                crate::cache::EntityLifetimeCheckpoint {
158                    account_position: Some(arete_interpreter::AccountPosition::new(
159                        9_007_199_254_740_993,
160                        u64::MAX,
161                    )),
162                    source_seq: None,
163                    source_sequences: Default::default(),
164                    deleted: false,
165                },
166            )]),
167            entity_tombstones: Default::default(),
168            entity_cache: vec![(
169                "tokens/list".to_string(),
170                vec![("key1".to_string(), serde_json::json!({"id": 1}))],
171            )],
172            journal: crate::journal::JournalSnapshot::default(),
173        };
174        (header, payload)
175    }
176
177    #[test]
178    fn round_trips_header_and_payload() {
179        let (header, payload) = sample();
180        let bytes = encode(&header, &payload).unwrap();
181
182        let decoded_header = decode_header(&bytes).unwrap();
183        assert_eq!(decoded_header.bytecode_hash, "abc123");
184        assert_eq!(decoded_header.resume_watermark, 42);
185
186        let decoded_payload = decode_payload(&bytes).unwrap();
187        assert_eq!(decoded_payload.entity_cache.len(), 1);
188        assert_eq!(decoded_payload.entity_cache[0].0, "tokens/list");
189        let lifetimes = decoded_payload.entity_lifetimes.unwrap();
190        let checkpoint = &lifetimes[0].2;
191        assert_eq!(
192            checkpoint.account_position,
193            Some(arete_interpreter::AccountPosition::new(
194                9_007_199_254_740_993,
195                u64::MAX
196            ))
197        );
198    }
199
200    #[test]
201    fn rejects_truncated_and_corrupt_blobs() {
202        let (header, payload) = sample();
203        let bytes = encode(&header, &payload).unwrap();
204
205        assert!(decode_header(&bytes[..4]).is_err());
206        assert!(decode_header(&[0u8; 32]).is_err());
207
208        let mut truncated = bytes.clone();
209        truncated.truncate(bytes.len() - 5);
210        assert!(decode_payload(&truncated).is_err());
211
212        let mut corrupted = bytes.clone();
213        let last = corrupted.len() - 1;
214        corrupted[last] ^= 0xFF;
215        assert!(decode_payload(&corrupted).is_err());
216    }
217}
arete_server/snapshot/envelope.rs

arete_server/snapshot/
envelope.rs
