salamander/
retention.rs

1//! Authoritative engine-core retention anchor publication and validation.
2
3use std::fs::{File, OpenOptions};
4use std::io::{Read, Write};
5use std::path::{Path, PathBuf};
6
7use serde::{Deserialize, Serialize};
8
9use crate::branch::BranchCatalog;
10use crate::stream::StreamCatalog;
11use crate::{Result, SalamanderError};
12
13const MAGIC: &[u8; 8] = b"SLMRETN1";
14pub(crate) const FORMAT_VERSION: u32 = 5;
15const MAX_ANCHOR_BYTES: usize = 256 * 1024 * 1024;
16/// Maximum opaque bootstrap payload accepted for one branch or consumer.
17pub const MAX_BOOTSTRAP_BYTES: usize = 64 * 1024 * 1024;
18
19/// Verified metadata for one authoritative engine-core retention anchor.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct RetentionAnchorInfo {
22    /// On-disk anchor format.
23    pub format_version: u32,
24    /// Database whose core catalogs are captured.
25    pub database_id: [u8; 16],
26    /// Effective retention floor represented by the anchor.
27    pub floor: u64,
28    /// Exact durable head represented by the anchor.
29    pub head: u64,
30    /// Encoded file length.
31    pub bytes: u64,
32    /// CRC32C of the encoded catalog payload.
33    pub checksum: u32,
34    /// Projection checkpoints promoted into this anchor.
35    pub projection_checkpoints: usize,
36    /// Branch bootstrap payloads promoted into this anchor.
37    pub branch_bootstraps: usize,
38    /// Consumer bootstrap payloads promoted into this anchor.
39    pub consumer_bootstraps: usize,
40    /// Total opaque bootstrap payload bytes carried by the anchor.
41    pub bootstrap_bytes: u64,
42    /// Engine system metadata records captured by the anchor.
43    pub system_records: usize,
44}
45
46/// Verified projection checkpoint coverage promoted by a retention anchor.
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
48pub struct RetentionProjectionCoverage {
49    /// Durable projection registration name.
50    pub name: String,
51    /// Descriptor fingerprint captured by every referenced checkpoint.
52    pub descriptor_fingerprint: [u8; 16],
53    /// Branch whose projection state is captured.
54    pub branch_id: [u8; 16],
55    /// Exclusive projection cursor represented by the checkpoints.
56    pub cursor: u64,
57    /// Immutable, checksummed snapshot files promoted by the anchor.
58    pub snapshot_ids: Vec<String>,
59}
60
61/// Opaque application bootstrap for one branch at an effective floor.
62#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
63pub struct RetentionBranchBootstrap {
64    /// Branch covered by the application checkpoint.
65    pub branch_id: [u8; 16],
66    /// Effective floor for which the checkpoint was created.
67    pub floor: u64,
68    /// CRC32C of `checkpoint`.
69    pub checksum: u32,
70    /// Opaque application bytes; the engine never interprets them.
71    pub checkpoint: Vec<u8>,
72}
73
74/// Opaque external-consumer bootstrap at an effective floor.
75#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
76pub struct RetentionConsumerBootstrap {
77    /// Stable durable consumer identifier.
78    pub consumer_id: String,
79    /// Effective floor for which the checkpoint was created.
80    pub floor: u64,
81    /// CRC32C of `checkpoint`.
82    pub checksum: u32,
83    /// Opaque consumer bytes; the engine never interprets them.
84    pub checkpoint: Vec<u8>,
85    /// Stable identity used when fetching this checkpoint.
86    #[serde(default)]
87    pub checkpoint_id: [u8; 16],
88    /// Exact feed scope for which this checkpoint is valid.
89    #[serde(default)]
90    pub scope: RetentionFeedScope,
91    /// Application-defined checkpoint codec identifier.
92    #[serde(default = "default_bootstrap_codec")]
93    pub codec: String,
94    /// Application-defined checkpoint codec version.
95    #[serde(default = "default_bootstrap_codec_version")]
96    pub codec_version: u32,
97}
98
99fn default_bootstrap_codec() -> String {
100    "opaque".into()
101}
102
103fn default_bootstrap_codec_version() -> u32 {
104    1
105}
106
107/// Envelope-only feed scope bound to a consumer bootstrap.
108#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
109pub struct RetentionFeedScope {
110    /// Selected branches; empty means all branches.
111    pub branches: Vec<[u8; 16]>,
112    /// Selected streams; empty means all streams.
113    pub streams: Vec<[u8; 16]>,
114    /// Selected event types; empty means all event types.
115    pub event_types: Vec<String>,
116}
117
118#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
119pub(crate) struct AnchoredSystemRecord {
120    pub event_type: String,
121    pub payload: Vec<u8>,
122}
123
124#[derive(Serialize, Deserialize)]
125pub(crate) struct CoreRetentionAnchor {
126    pub format_version: u32,
127    pub database_id: [u8; 16],
128    pub floor: u64,
129    pub head: u64,
130    pub stream_catalog: StreamCatalog,
131    pub branch_catalog: BranchCatalog,
132    #[serde(default)]
133    pub projection_coverage: Vec<RetentionProjectionCoverage>,
134    #[serde(default)]
135    pub branch_bootstraps: Vec<RetentionBranchBootstrap>,
136    #[serde(default)]
137    pub consumer_bootstraps: Vec<RetentionConsumerBootstrap>,
138    #[serde(default)]
139    pub system_records: Vec<AnchoredSystemRecord>,
140}
141
142pub(crate) fn publish(root: &Path, anchor: CoreRetentionAnchor) -> Result<RetentionAnchorInfo> {
143    validate_bootstraps(
144        anchor.floor,
145        &anchor.branch_bootstraps,
146        &anchor.consumer_bootstraps,
147    )?;
148    if let Ok(Some((existing, info))) = load(root) {
149        if validate_identity(&existing, anchor.database_id, anchor.floor, anchor.head).is_ok()
150            && existing.projection_coverage == anchor.projection_coverage
151            && existing.branch_bootstraps == anchor.branch_bootstraps
152            && existing.consumer_bootstraps == anchor.consumer_bootstraps
153        {
154            return Ok(info);
155        }
156    }
157    let payload = bincode::serialize(&anchor)
158        .map_err(|error| SalamanderError::Serialization(error.to_string()))?;
159    if payload.len() > MAX_ANCHOR_BYTES {
160        return Err(SalamanderError::ResourceLimit {
161            resource: "retention anchor bytes",
162            actual: payload.len() as u64,
163            maximum: MAX_ANCHOR_BYTES as u64,
164        });
165    }
166    let checksum = crc32c::crc32c(&payload);
167    let mut encoded = Vec::with_capacity(16 + payload.len());
168    encoded.extend_from_slice(MAGIC);
169    encoded.extend_from_slice(&(payload.len() as u32).to_le_bytes());
170    encoded.extend_from_slice(&checksum.to_le_bytes());
171    encoded.extend_from_slice(&payload);
172
173    let dir = anchor_dir(root);
174    std::fs::create_dir_all(&dir)?;
175    let temporary = dir.join("core.anchor.tmp");
176    crash_point("before_anchor_publish");
177    {
178        let mut file = OpenOptions::new()
179            .create(true)
180            .truncate(true)
181            .write(true)
182            .open(&temporary)?;
183        file.write_all(&encoded)?;
184        file.sync_all()?;
185    }
186    crash_point("after_anchor_fsync");
187    let (verified, info) = decode(&temporary)?;
188    validate_identity(&verified, anchor.database_id, anchor.floor, anchor.head)?;
189    if anchor_path(root).exists() {
190        std::fs::remove_file(anchor_path(root))?;
191    }
192    std::fs::rename(&temporary, anchor_path(root))?;
193    sync_dir(&dir)?;
194    crash_point("after_anchor_publish");
195    Ok(info)
196}
197
198pub(crate) fn crash_point(name: &str) {
199    if std::env::var_os("SALAMANDER_RETENTION_CRASH_AT").is_some_and(|value| value == name) {
200        std::process::abort();
201    }
202}
203
204pub(crate) fn load(root: &Path) -> Result<Option<(CoreRetentionAnchor, RetentionAnchorInfo)>> {
205    let path = anchor_path(root);
206    match decode(&path) {
207        Ok(value) => Ok(Some(value)),
208        Err(SalamanderError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
209        Err(error) => Err(error),
210    }
211}
212
213pub(crate) fn validate_identity(
214    anchor: &CoreRetentionAnchor,
215    database_id: [u8; 16],
216    floor: u64,
217    head: u64,
218) -> Result<()> {
219    if anchor.format_version != FORMAT_VERSION {
220        return Err(corrupt(format!(
221            "unsupported retention anchor format {}",
222            anchor.format_version
223        )));
224    }
225    if anchor.database_id != database_id {
226        return Err(corrupt("retention anchor belongs to another database"));
227    }
228    if anchor.floor != floor {
229        return Err(corrupt(format!(
230            "retention anchor floor {} does not match required floor {floor}",
231            anchor.floor
232        )));
233    }
234    if anchor.head != head {
235        return Err(corrupt(format!(
236            "retention anchor head {} does not match durable head {head}",
237            anchor.head
238        )));
239    }
240    Ok(())
241}
242
243fn decode(path: &Path) -> Result<(CoreRetentionAnchor, RetentionAnchorInfo)> {
244    let length = std::fs::metadata(path)?.len() as usize;
245    if !(16..=16 + MAX_ANCHOR_BYTES).contains(&length) {
246        return Err(corrupt("retention anchor length is invalid"));
247    }
248    let mut file = File::open(path)?;
249    let mut prefix = [0u8; 16];
250    file.read_exact(&mut prefix)?;
251    if &prefix[..8] != MAGIC {
252        return Err(corrupt("retention anchor magic is invalid"));
253    }
254    let payload_len = u32::from_le_bytes(prefix[8..12].try_into().unwrap()) as usize;
255    if payload_len == 0 || payload_len > MAX_ANCHOR_BYTES || payload_len + 16 != length {
256        return Err(corrupt("retention anchor payload length is invalid"));
257    }
258    let expected_checksum = u32::from_le_bytes(prefix[12..16].try_into().unwrap());
259    let mut payload = vec![0; payload_len];
260    file.read_exact(&mut payload)?;
261    if crc32c::crc32c(&payload) != expected_checksum {
262        return Err(corrupt("retention anchor checksum mismatch"));
263    }
264    let anchor: CoreRetentionAnchor = bincode::deserialize(&payload)
265        .map_err(|error| corrupt(format!("retention anchor decode: {error}")))?;
266    validate_bootstraps(
267        anchor.floor,
268        &anchor.branch_bootstraps,
269        &anchor.consumer_bootstraps,
270    )?;
271    let info = RetentionAnchorInfo {
272        format_version: anchor.format_version,
273        database_id: anchor.database_id,
274        floor: anchor.floor,
275        head: anchor.head,
276        bytes: length as u64,
277        checksum: expected_checksum,
278        projection_checkpoints: anchor
279            .projection_coverage
280            .iter()
281            .map(|coverage| coverage.snapshot_ids.len())
282            .sum(),
283        branch_bootstraps: anchor.branch_bootstraps.len(),
284        consumer_bootstraps: anchor.consumer_bootstraps.len(),
285        bootstrap_bytes: anchor
286            .branch_bootstraps
287            .iter()
288            .map(|item| item.checkpoint.len() as u64)
289            .chain(
290                anchor
291                    .consumer_bootstraps
292                    .iter()
293                    .map(|item| item.checkpoint.len() as u64),
294            )
295            .sum(),
296        system_records: anchor.system_records.len(),
297    };
298    Ok((anchor, info))
299}
300
301fn validate_bootstraps(
302    floor: u64,
303    branches: &[RetentionBranchBootstrap],
304    consumers: &[RetentionConsumerBootstrap],
305) -> Result<()> {
306    let mut branch_ids = std::collections::BTreeSet::new();
307    for item in branches {
308        if item.floor != floor
309            || item.checkpoint.len() > MAX_BOOTSTRAP_BYTES
310            || crc32c::crc32c(&item.checkpoint) != item.checksum
311            || !branch_ids.insert(item.branch_id)
312        {
313            return Err(corrupt("invalid or duplicate branch bootstrap coverage"));
314        }
315    }
316    let mut consumer_ids = std::collections::BTreeSet::new();
317    let mut checkpoint_ids = std::collections::BTreeSet::new();
318    for item in consumers {
319        if item.floor != floor
320            || item.checkpoint.len() > MAX_BOOTSTRAP_BYTES
321            || crc32c::crc32c(&item.checkpoint) != item.checksum
322            || !consumer_ids.insert(item.consumer_id.as_str())
323            || item.checkpoint_id == [0; 16]
324            || !checkpoint_ids.insert(item.checkpoint_id)
325            || item.codec.is_empty()
326            || item.codec.len() > 128
327        {
328            return Err(corrupt("invalid or duplicate consumer bootstrap coverage"));
329        }
330    }
331    Ok(())
332}
333
334fn anchor_dir(root: &Path) -> PathBuf {
335    root.join("retention")
336}
337
338fn anchor_path(root: &Path) -> PathBuf {
339    anchor_dir(root).join("core.anchor")
340}
341
342fn corrupt(reason: impl Into<String>) -> SalamanderError {
343    SalamanderError::InvalidFormat(reason.into())
344}
345
346#[cfg(unix)]
347fn sync_dir(dir: &Path) -> Result<()> {
348    File::open(dir)?.sync_all()?;
349    Ok(())
350}
351
352#[cfg(not(unix))]
353fn sync_dir(_dir: &Path) -> Result<()> {
354    Ok(())
355}
salamander/retention.rs

salamander/
retention.rs
