krabka_remote_storage/diskless/capture/
mod.rs

1//! Portable disaster-recovery capture state and validation.
2
3mod authentication;
4mod projection;
5
6pub use projection::WalCaptureProjection;
7use {
8    super::WalIndexEntry,
9    crate::worm::{SegmentManifest, Sha256Digest},
10    serde::{Deserialize, Serialize},
11    std::collections::HashSet,
12    uuid::Uuid,
13};
14
15/// Synthetic expected-head name used for a signed capture boundary.
16pub const CAPTURE_HEAD_NAME: &str = "diskless-capture";
17
18const CAPTURE_STATE_KEY: &str = "__krabka_diskless_capture_state";
19
20/// One live capture range and its authoritative object.
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct CapturedWalRange {
23    pub object_key: String,
24    pub entry: WalIndexEntry,
25}
26
27/// Captured committed state for one data partition.
28#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
29pub struct DisklessPartitionCapture {
30    pub topic: String,
31    pub topic_id: Uuid,
32    pub partition: i32,
33    pub delete_floor: i64,
34    pub recovery_cutoff: i64,
35    pub ranges: Vec<CapturedWalRange>,
36}
37
38/// Portable committed diskless-WAL capture.
39#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
40pub struct DisklessWalCapture {
41    pub format_version: u16,
42    pub captured_at_ms: u64,
43    pub source_cutoffs: Vec<i64>,
44    pub partitions: Vec<DisklessPartitionCapture>,
45    /// Digest of the controller checkpoint captured beside this boundary.
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub metadata_snapshot_sha256: Option<Sha256Digest>,
48    /// Digest of the RLMM cache snapshot captured beside this boundary.
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub rlmm_snapshot_sha256: Option<Sha256Digest>,
51    /// Digest of the consumer-group offsets captured beside this boundary.
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    pub group_offsets_sha256: Option<Sha256Digest>,
54    /// Optional signed WORM boundary covering this capture and its WAL objects.
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub authentication: Option<SegmentManifest>,
57}
58
59/// Just the version of a capture, read before the rest of it.
60#[derive(Deserialize)]
61struct DisklessWalCaptureVersion {
62    format_version: Option<serde_json::Value>,
63}
64
65impl DisklessWalCapture {
66    /// Version of `diskless-wal-index.json`, its required top-level
67    /// `"format_version"` field.
68    ///
69    /// Part of the 1.x on-disk contract: a 1.x restore reads every capture
70    /// that any earlier 1.x build wrote.
71    pub const FORMAT_VERSION: u16 = 1;
72
73    /// Decode JSON and reject a missing or unknown capture version.
74    ///
75    /// The version is read on its own first, so a capture of another version
76    /// is reported as that, and not as whatever field its shape lacks.
77    ///
78    /// # Errors
79    /// Returns an error for malformed JSON, a missing or unsupported version,
80    /// or invalid state.
81    pub fn from_slice(bytes: &[u8]) -> Result<Self, String> {
82        let probe: DisklessWalCaptureVersion =
83            serde_json::from_slice(bytes).map_err(|error| error.to_string())?;
84        match probe.format_version {
85            None => {
86                return Err(
87                    "diskless WAL capture has no format_version: it predates krabka \
88                     1.0, which this build does not read"
89                        .to_owned(),
90                );
91            }
92            Some(version) if version != Self::FORMAT_VERSION => {
93                return Err(format!(
94                    "unsupported diskless WAL capture format version {version}: this build \
95                     reads version {}",
96                    Self::FORMAT_VERSION
97                ));
98            }
99            Some(_) => {}
100        }
101        let capture: Self = serde_json::from_slice(bytes).map_err(|error| error.to_string())?;
102        capture.validate()?;
103        Ok(capture)
104    }
105
106    /// Validate all state that controls restore selection and boundaries.
107    ///
108    /// # Errors
109    /// Returns an error for invalid partitions, ranges, floors, or cutoffs.
110    pub fn validate(&self) -> Result<(), String> {
111        if self.source_cutoffs.iter().any(|cutoff| *cutoff < 0) {
112            return Err("negative diskless WAL source cutoff".to_owned());
113        }
114        let mut partitions = HashSet::new();
115        for partition in &self.partitions {
116            if !valid_topic_name(&partition.topic)
117                || partition.partition < 0
118                || partition.delete_floor < 0
119                || partition.recovery_cutoff < partition.delete_floor
120                || !partitions.insert((partition.topic_id, partition.partition))
121            {
122                return Err(format!(
123                    "invalid or duplicate diskless capture partition {}-{}",
124                    partition.topic, partition.partition
125                ));
126            }
127            let mut previous_last = None;
128            let mut covered_through = partition.delete_floor;
129            for range in &partition.ranges {
130                if range.object_key.is_empty()
131                    || range.object_key == CAPTURE_STATE_KEY
132                    || range.entry.topic_id != partition.topic_id
133                    || range.entry.partition != partition.partition
134                    || range.entry.first_offset < 0
135                    || range.entry.last_offset < range.entry.first_offset
136                    || range.entry.byte_len == 0
137                    || previous_last.is_some_and(|last| last >= range.entry.first_offset)
138                {
139                    return Err(format!(
140                        "invalid or unordered diskless WAL range for {}-{}",
141                        partition.topic, partition.partition
142                    ));
143                }
144                if range.entry.last_offset >= partition.delete_floor {
145                    if range.entry.first_offset > covered_through {
146                        return Err(format!(
147                            "diskless WAL ranges for {}-{} have a gap at {covered_through}",
148                            partition.topic, partition.partition
149                        ));
150                    }
151                    covered_through = range
152                        .entry
153                        .last_offset
154                        .checked_add(1)
155                        .ok_or_else(|| "diskless WAL recovery cutoff overflow".to_owned())?;
156                }
157                previous_last = Some(range.entry.last_offset);
158            }
159            let observed_cutoff = previous_last
160                .map(|last| {
161                    last.checked_add(1)
162                        .ok_or_else(|| "diskless WAL recovery cutoff overflow".to_owned())
163                })
164                .transpose()?
165                .unwrap_or(partition.delete_floor)
166                .max(partition.delete_floor);
167            if observed_cutoff != partition.recovery_cutoff {
168                return Err(format!(
169                    "diskless WAL ranges for {}-{} end at {observed_cutoff}, capture declares {}",
170                    partition.topic, partition.partition, partition.recovery_cutoff
171                ));
172            }
173        }
174        Ok(())
175    }
176}
177
178fn valid_topic_name(name: &str) -> bool {
179    !name.is_empty()
180        && name.len() <= 249
181        && name != "."
182        && name != ".."
183        && name
184            .bytes()
185            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
186}
krabka_remote_storage/diskless/capture/mod.rs

krabka_remote_storage/diskless/capture/
mod.rs
