krabka_restore/
verify.rs

1//! Integrity of one archived segment, and the facts it yields.
2//!
3//! This module owns the decision that a segment is safe to rehydrate. It
4//! fetches the segment's artifacts, walks the `.log` batch by batch with
5//! `krabka_protocol::records::validate_one_v2_batch`, and checks framing and
6//! CRC without decoding records. A batch whose declared length overruns the
7//! object is a truncated segment; a batch whose CRC disagrees with its body is
8//! a checksum mismatch, reported with the object key and the byte position. It
9//! also checks that the copy is not torn: a segment that carries a log but no
10//! time index was archived only in part. From the batch headers it derives the
11//! two facts the rest of the pipeline needs and the archive metadata cannot be
12//! trusted for, the segment's true `end_offset` and `max_timestamp_ms`, and it
13//! returns the verified log bytes so the segment is fetched exactly once.
14//!
15//! This file holds the orchestration and the size caps every fetch is held to.
16//! The per-artifact checks live beside it: [`self::log_walk`] for the `.log`
17//! pass, [`self::index`] for the sparse sidecar indexes, [`self::snapshot`] for
18//! the producer-state snapshot, and [`self::leader_epoch`] for the
19//! leader-epoch checkpoint.
20
21use {
22    crate::authenticity,
23    bytes::Bytes,
24    krabka_ids::{LeaderEpoch, Offset},
25    krabka_object_store::{ObjectOps, ObjectStoreError},
26    krabka_remote_storage::{ObjectEntry, TopicIdPartition},
27    object_store::path::Path,
28    std::collections::BTreeMap,
29    uuid::Uuid,
30};
31
32mod index;
33mod leader_epoch;
34mod log_walk;
35mod snapshot;
36
37#[cfg(test)]
38mod tests;
39
40use {
41    self::{
42        index::{validate_offset_index, validate_time_index, validate_txn_index},
43        leader_epoch::parse_leader_epoch_checkpoint,
44        log_walk::walk_log,
45        snapshot::validate_producer_snapshot,
46    },
47    crate::{
48        backend::ArchiveStore,
49        discover::{ArchiveObject, SegmentInventory},
50        error::RestoreError,
51    },
52};
53
54/// What verification established about one segment.
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub struct SegmentFacts {
57    /// The per-copy segment id.
58    pub segment_id: Uuid,
59    /// First offset the segment holds.
60    pub base_offset: Offset,
61    /// Last offset the segment holds, derived from the batch headers.
62    pub end_offset: Offset,
63    /// Highest record timestamp in the segment, derived from the batch
64    /// headers. It is `-1` for a segment with no timestamped record.
65    pub max_timestamp_ms: i64,
66    /// Batches the segment holds.
67    pub batches: u64,
68    /// Records the batch headers account for.
69    pub records: u64,
70    /// Size of the verified `.log`, in bytes.
71    pub log_bytes: u64,
72    /// The offset each leader epoch starts at, from the leader-epoch
73    /// checkpoint. The target partition needs it to answer `OffsetForLeaderEpoch`.
74    pub leader_epochs: Vec<LeaderEpochStart>,
75}
76
77/// The offset a leader epoch starts at, from a leader-epoch checkpoint row.
78#[derive(Debug, Clone, Copy, PartialEq, Eq)]
79pub struct LeaderEpochStart {
80    /// The leader epoch.
81    pub epoch: LeaderEpoch,
82    /// The first offset written under `epoch`.
83    pub start_offset: Offset,
84}
85
86/// A segment that passed verification, with the bytes that passed it.
87#[derive(Debug, Clone, PartialEq, Eq)]
88pub struct VerifiedSegment {
89    /// What the verification established.
90    pub facts: SegmentFacts,
91    /// The verified `.log` bytes.
92    pub log: Bytes,
93}
94
95/// Guard against an unbounded read of a corrupt or hostile `.log` object.
96/// Kafka's default `segment.bytes` is 1 GiB, and an operator can raise it, so
97/// this cap is a generous multiple of that default rather than the exact
98/// configured value, which this offline tool never sees.
99pub const MAX_LOG_BYTES: u64 = 8 * 1024 * 1024 * 1024;
100
101/// Guard for the sparse `.index` and `.timeindex` sidecars. An entry lands
102/// only every `index.interval.bytes` (4 KiB by default), so even a segment at
103/// the [`MAX_LOG_BYTES`] cap produces a sidecar many orders of magnitude
104/// smaller than this.
105const MAX_INDEX_BYTES: u64 = 256 * 1024 * 1024;
106
107/// Guard for the `.txnindex` sidecar: one entry per aborted transaction.
108const MAX_TXN_INDEX_BYTES: u64 = MAX_INDEX_BYTES;
109
110/// Guard for the `.snapshot` producer-state sidecar: 46 bytes per producer.
111const MAX_SNAPSHOT_BYTES: u64 = 64 * 1024 * 1024;
112
113/// Guard for the `.leader_epoch_checkpoint` text sidecar: one short line per
114/// leader change over the segment's lifetime.
115const MAX_LEADER_EPOCH_BYTES: u64 = 16 * 1024 * 1024;
116
117/// Fetch and verify one segment.
118///
119/// # Errors
120///
121/// Returns [`RestoreError::TornCopy`] when a required artifact is absent,
122/// [`RestoreError::TruncatedSegment`] when the log ends inside a batch,
123/// [`RestoreError::ChecksumMismatch`] when a batch CRC disagrees with its
124/// body, and [`RestoreError::ObjectStore`] when a fetch fails.
125pub async fn verify_segment(
126    store: &ArchiveStore,
127    partition: &TopicIdPartition,
128    segment: &SegmentInventory,
129) -> Result<VerifiedSegment, RestoreError> {
130    verify_segment_inner(store, partition, segment, None).await
131}
132
133/// Fetch and verify one segment against signed WORM object claims.
134///
135/// # Errors
136///
137/// Returns [`RestoreError::Authenticity`] when an artifact is absent from the
138/// authenticated manifest set or the bytes fetched now differ from its signed
139/// size or digest. Other failures are the same as [`verify_segment`].
140pub async fn verify_segment_authenticated(
141    store: &ArchiveStore,
142    partition: &TopicIdPartition,
143    segment: &SegmentInventory,
144    objects: &BTreeMap<String, ObjectEntry>,
145) -> Result<VerifiedSegment, RestoreError> {
146    verify_segment_inner(store, partition, segment, Some(objects)).await
147}
148
149async fn verify_segment_inner(
150    store: &ArchiveStore,
151    partition: &TopicIdPartition,
152    segment: &SegmentInventory,
153    objects: Option<&BTreeMap<String, ObjectEntry>>,
154) -> Result<VerifiedSegment, RestoreError> {
155    // Checked in the order the broker's copy path writes the artifacts, so a
156    // torn copy is reported by the first one actually missing.
157    let log = require_artifact(segment.log.as_ref(), ".log", partition, segment)?;
158    let offset_index =
159        require_artifact(segment.offset_index.as_ref(), ".index", partition, segment)?;
160    let time_index = require_artifact(
161        segment.time_index.as_ref(),
162        ".timeindex",
163        partition,
164        segment,
165    )?;
166    let producer_snapshot = require_artifact(
167        segment.producer_snapshot.as_ref(),
168        ".snapshot",
169        partition,
170        segment,
171    )?;
172    let leader_epoch = require_artifact(
173        segment.leader_epoch.as_ref(),
174        ".leader_epoch_checkpoint",
175        partition,
176        segment,
177    )?;
178
179    let ops = store.ops();
180    // The `.log` is fetched exactly once, here; every later stage reads it
181    // from the `Bytes` this function returns, not from the archive again.
182    let log_bytes = fetch_capped(ops, &log.key, MAX_LOG_BYTES, objects).await?;
183    let offset_index_bytes = fetch_capped(ops, &offset_index.key, MAX_INDEX_BYTES, objects).await?;
184    let time_index_bytes = fetch_capped(ops, &time_index.key, MAX_INDEX_BYTES, objects).await?;
185    let producer_snapshot_bytes =
186        fetch_capped(ops, &producer_snapshot.key, MAX_SNAPSHOT_BYTES, objects).await?;
187    let leader_epoch_bytes =
188        fetch_capped(ops, &leader_epoch.key, MAX_LEADER_EPOCH_BYTES, objects).await?;
189    let transaction_index = match &segment.transaction_index {
190        Some(artifact) => Some((
191            artifact.key.clone(),
192            fetch_capped(ops, &artifact.key, MAX_TXN_INDEX_BYTES, objects).await?,
193        )),
194        None => None,
195    };
196
197    let walked = walk_log(&log.key, segment, &log_bytes)?;
198    let log_bytes_len = u64::try_from(log_bytes.len()).unwrap_or(u64::MAX);
199
200    validate_offset_index(
201        &offset_index.key,
202        &offset_index_bytes,
203        segment.base_offset,
204        walked.end_offset,
205        log_bytes_len,
206    )?;
207    validate_time_index(
208        &time_index.key,
209        &time_index_bytes,
210        segment.base_offset,
211        walked.end_offset,
212    )?;
213    if let Some((key, bytes)) = &transaction_index {
214        validate_txn_index(key, bytes, segment.base_offset, walked.end_offset)?;
215    }
216    let snapshot_offset =
217        walked
218            .end_offset
219            .0
220            .checked_add(1)
221            .ok_or_else(|| RestoreError::TruncatedSegment {
222                key: producer_snapshot.key.to_string(),
223                position: 0,
224                declared: u64::MAX,
225                available: u64::MAX,
226            })?;
227    validate_producer_snapshot(
228        &producer_snapshot.key,
229        &producer_snapshot_bytes,
230        snapshot_offset,
231    )?;
232    let leader_epochs = parse_leader_epoch_checkpoint(
233        &leader_epoch.key,
234        &leader_epoch_bytes,
235        segment.base_offset,
236        walked.end_offset,
237    )?;
238
239    Ok(VerifiedSegment {
240        facts: SegmentFacts {
241            segment_id: segment.segment_id,
242            base_offset: segment.base_offset,
243            end_offset: walked.end_offset,
244            max_timestamp_ms: walked.max_timestamp_ms,
245            batches: walked.batches,
246            records: walked.records,
247            log_bytes: log_bytes_len,
248            leader_epochs,
249        },
250        log: log_bytes,
251    })
252}
253
254/// Look up one mandatory artifact, or report the torn copy it reveals.
255fn require_artifact<'a>(
256    artifact: Option<&'a ArchiveObject>,
257    name: &str,
258    partition: &TopicIdPartition,
259    segment: &SegmentInventory,
260) -> Result<&'a ArchiveObject, RestoreError> {
261    artifact.ok_or_else(|| RestoreError::TornCopy {
262        topic: partition.topic.clone(),
263        partition: partition.partition,
264        segment_id: segment.segment_id,
265        artifact: name.to_owned(),
266    })
267}
268
269/// Fetch a whole object, refusing it before buffering any bytes if it exceeds
270/// `max_bytes`. [`ObjectOps::get_capped`] binds its read to backend metadata
271/// when available and enforces the limit while streaming the response.
272pub async fn fetch_capped(
273    ops: &dyn ObjectOps,
274    key: &Path,
275    max_bytes: u64,
276    authenticated: Option<&BTreeMap<String, ObjectEntry>>,
277) -> Result<Bytes, RestoreError> {
278    let expected = authenticity::object_claim(key, authenticated)?;
279    let meta = ops.head(key).await?;
280    authenticity::verify_object_size(key, expected, meta.size)?;
281    if meta.size > max_bytes {
282        return Err(ObjectStoreError::TooLarge {
283            key: key.clone(),
284            size: meta.size,
285            max_bytes,
286        }
287        .into());
288    }
289    let bytes = ops.get_capped(key, max_bytes).await?;
290    authenticity::verify_object_bytes(key, expected, &bytes)?;
291    Ok(bytes)
292}
293
294/// Best-effort `i64` → `u64` for a diagnostic [`RestoreError`] field. A Kafka
295/// offset or timestamp is never negative in a well-formed segment; this only
296/// has to render something sensible for corrupt input, never panic, and never
297/// allocate.
298fn offset_as_u64(value: i64) -> u64 { u64::try_from(value).unwrap_or(0) }
krabka_restore/verify.rs

krabka_restore/
verify.rs
