krabka_remote_storage/metadata/
segment_metadata.rs

1//! The remote-segment metadata record and the update that advances it.
2//!
3//! [`RemoteLogSegmentMetadata`] is what the metadata SPI stores for one
4//! segment, [`RemoteLogSegmentDetails`] groups the fields its constructor
5//! validates, [`RemoteLogSegmentMetadataUpdate`] is the lifecycle update that
6//! [`RemoteLogSegmentMetadata::with_update`] applies, and [`CustomMetadata`]
7//! is the opaque payload a storage manager may attach at copy time.
8
9use {
10    crate::{
11        error::RemoteStorageError,
12        metadata::{RemoteLogSegmentId, RemoteLogSegmentState},
13    },
14    krabka_ids::LeaderEpoch,
15    std::collections::BTreeMap,
16};
17
18#[derive(derive_more::From, derive_more::Into)]
19/// Opaque bytes that an [`RemoteStorageManager`](crate::RemoteStorageManager)
20/// can return from `copy_log_segment_data`, for example an object-store key
21/// or a version id. The caller sends the bytes back on every later call for
22/// that segment.
23#[derive(Debug, Clone, PartialEq, Eq, Hash)]
24pub struct CustomMetadata(pub Vec<u8>);
25
26/// Metadata for one segment that is stored, or is being stored, in the
27/// remote tier.
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub struct RemoteLogSegmentMetadata {
30    remote_log_segment_id: RemoteLogSegmentId,
31    start_offset: i64,
32    end_offset: i64,
33    max_timestamp_ms: i64,
34    broker_id: i32,
35    event_timestamp_ms: i64,
36    segment_size_in_bytes: i32,
37    custom_metadata: Option<CustomMetadata>,
38    state: RemoteLogSegmentState,
39    segment_leader_epochs: BTreeMap<LeaderEpoch, i64>,
40    /// KIP-405 `txnIndexEmpty`: `true` when the segment carries no transaction
41    /// index. Serialized as tagged field (tag 0) in the JVM record format.
42    /// Defaults to `false`.
43    txn_index_empty: bool,
44}
45
46/// Size, lifecycle state, and leader-epoch offsets for a remote log segment.
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct RemoteLogSegmentDetails {
49    segment_size_in_bytes: i32,
50    state: RemoteLogSegmentState,
51    segment_leader_epochs: BTreeMap<LeaderEpoch, i64>,
52}
53
54impl RemoteLogSegmentDetails {
55    #[must_use]
56    pub fn new(
57        segment_size_in_bytes: i32,
58        state: RemoteLogSegmentState,
59        segment_leader_epochs: BTreeMap<LeaderEpoch, i64>,
60    ) -> Self {
61        Self {
62            segment_size_in_bytes,
63            state,
64            segment_leader_epochs,
65        }
66    }
67}
68
69impl RemoteLogSegmentMetadata {
70    /// Constructs a [`RemoteLogSegmentMetadata`].
71    ///
72    /// # Errors
73    ///
74    /// Returns [`RemoteStorageError::InvalidArgument`] when
75    /// `segment_leader_epochs` is empty, `end_offset < start_offset`, or
76    /// `segment_size_in_bytes < 0`.
77    pub fn new(
78        remote_log_segment_id: RemoteLogSegmentId,
79        start_offset: i64,
80        end_offset: i64,
81        max_timestamp_ms: i64,
82        broker_id: i32,
83        event_timestamp_ms: i64,
84        details: RemoteLogSegmentDetails,
85    ) -> Result<Self, RemoteStorageError> {
86        let RemoteLogSegmentDetails {
87            segment_size_in_bytes,
88            state,
89            segment_leader_epochs,
90        } = details;
91        if segment_leader_epochs.is_empty() {
92            return Err(RemoteStorageError::InvalidArgument(
93                "segment_leader_epochs must not be empty".into(),
94            ));
95        }
96        if end_offset < start_offset {
97            return Err(RemoteStorageError::InvalidArgument(format!(
98                "end_offset ({end_offset}) < start_offset ({start_offset})"
99            )));
100        }
101        if segment_size_in_bytes < 0 {
102            return Err(RemoteStorageError::InvalidArgument(format!(
103                "segment_size_in_bytes ({segment_size_in_bytes}) must be >= 0"
104            )));
105        }
106        Ok(Self {
107            remote_log_segment_id,
108            start_offset,
109            end_offset,
110            max_timestamp_ms,
111            broker_id,
112            event_timestamp_ms,
113            segment_size_in_bytes,
114            custom_metadata: None,
115            state,
116            segment_leader_epochs,
117            txn_index_empty: false,
118        })
119    }
120
121    /// Applies a [`RemoteLogSegmentMetadataUpdate`] and returns the updated
122    /// copy. The update advances `state`, refreshes `event_timestamp_ms`
123    /// and `broker_id`, and replaces `custom_metadata` when the update
124    /// carries `Some`.
125    ///
126    /// # Errors
127    ///
128    /// Returns [`RemoteStorageError::InvalidArgument`] if the update's
129    /// segment id does not match, or
130    /// [`RemoteStorageError::InvalidSegmentTransition`] if the state change
131    /// is not permitted from the current state.
132    pub fn with_update(
133        &self,
134        update: &RemoteLogSegmentMetadataUpdate,
135    ) -> Result<Self, RemoteStorageError> {
136        if update.remote_log_segment_id != self.remote_log_segment_id {
137            return Err(RemoteStorageError::InvalidArgument(
138                "update segment id does not match metadata segment id".into(),
139            ));
140        }
141        if !self.state.is_valid_transition(update.state) {
142            return Err(RemoteStorageError::InvalidSegmentTransition {
143                id: self.remote_log_segment_id.clone(),
144                from: self.state,
145                to: update.state,
146            });
147        }
148        let mut next = self.clone();
149        next.state = update.state;
150        next.event_timestamp_ms = update.event_timestamp_ms;
151        next.broker_id = update.broker_id;
152        if update.custom_metadata.is_some() {
153            next.custom_metadata.clone_from(&update.custom_metadata);
154        }
155        Ok(next)
156    }
157
158    /// The segment's unique id.
159    #[must_use]
160    pub fn remote_log_segment_id(&self) -> &RemoteLogSegmentId { &self.remote_log_segment_id }
161
162    /// First offset (inclusive) covered by this segment.
163    #[must_use]
164    pub fn start_offset(&self) -> i64 { self.start_offset }
165
166    /// Last offset (inclusive) covered by this segment.
167    #[must_use]
168    pub fn end_offset(&self) -> i64 { self.end_offset }
169
170    /// Highest record timestamp in this segment.
171    #[must_use]
172    pub fn max_timestamp_ms(&self) -> i64 { self.max_timestamp_ms }
173
174    /// Id of the broker that produced this metadata.
175    #[must_use]
176    pub fn broker_id(&self) -> i32 { self.broker_id }
177
178    /// Wall-clock time the latest event for this segment was created.
179    #[must_use]
180    pub fn event_timestamp_ms(&self) -> i64 { self.event_timestamp_ms }
181
182    /// Size of the `.log` data in bytes.
183    #[must_use]
184    pub fn segment_size_in_bytes(&self) -> i32 { self.segment_size_in_bytes }
185
186    /// Opaque metadata the [`RemoteStorageManager`](crate::RemoteStorageManager)
187    /// attached at copy time, if any.
188    #[must_use]
189    pub fn custom_metadata(&self) -> Option<&CustomMetadata> { self.custom_metadata.as_ref() }
190
191    /// Current lifecycle state.
192    #[must_use]
193    pub fn state(&self) -> RemoteLogSegmentState { self.state }
194
195    /// Map of leader epoch → first offset that epoch contributed to this
196    /// segment.
197    #[must_use]
198    pub fn segment_leader_epochs(&self) -> &BTreeMap<LeaderEpoch, i64> {
199        &self.segment_leader_epochs
200    }
201
202    /// The first offset of `epoch` and the earliest offset belonging to a later epoch.
203    #[must_use]
204    pub fn epoch_bounds(&self, epoch: LeaderEpoch) -> (Option<i64>, Option<i64>) {
205        let start = self.segment_leader_epochs.get(&epoch).copied();
206        let next = self
207            .segment_leader_epochs
208            .iter()
209            .filter(|(candidate, _)| **candidate > epoch)
210            .map(|(_, start)| *start)
211            .min();
212        (start, next)
213    }
214
215    /// Attaches custom metadata in builder style. RSM copy paths use it when
216    /// they produce a key before they record `CopySegmentFinished`.
217    #[must_use]
218    pub fn with_custom_metadata(mut self, custom: CustomMetadata) -> Self {
219        self.custom_metadata = Some(custom);
220        self
221    }
222
223    /// `true` if the segment has no transaction index (KIP-405 `txnIndexEmpty`).
224    /// Defaults to `false`. Serialized as the JVM record's tagged field (tag 0).
225    #[must_use]
226    pub fn txn_index_empty(&self) -> bool { self.txn_index_empty }
227
228    /// Builder-style setter for [`Self::txn_index_empty`].
229    #[must_use]
230    pub fn with_txn_index_empty(mut self, empty: bool) -> Self {
231        self.txn_index_empty = empty;
232        self
233    }
234}
235
236/// An update to an existing [`RemoteLogSegmentMetadata`]'s lifecycle state.
237#[derive(Debug, Clone, PartialEq, Eq)]
238pub struct RemoteLogSegmentMetadataUpdate {
239    /// The segment being updated.
240    pub remote_log_segment_id: RemoteLogSegmentId,
241    /// Wall-clock time of this update.
242    pub event_timestamp_ms: i64,
243    /// New custom metadata, when the update introduces or changes it.
244    pub custom_metadata: Option<CustomMetadata>,
245    /// The new lifecycle state.
246    pub state: RemoteLogSegmentState,
247    /// Broker that produced this update.
248    pub broker_id: i32,
249}
250
251#[cfg(test)]
252mod tests {
253    use {
254        super::*,
255        crate::metadata::{TopicIdPartition, test_topic_partition as tp},
256        assert2::{assert, check},
257        uuid::Uuid,
258    };
259
260    fn seg_id() -> RemoteLogSegmentId { RemoteLogSegmentId::new(tp(), Uuid::from_u128(99)) }
261
262    fn epochs() -> BTreeMap<LeaderEpoch, i64> {
263        maplit::btreemap! {LeaderEpoch(0) => 0}
264    }
265
266    fn metadata_range(
267        start: i64,
268        end: i64,
269        leader_epochs: BTreeMap<LeaderEpoch, i64>,
270    ) -> Result<RemoteLogSegmentMetadata, RemoteStorageError> {
271        RemoteLogSegmentMetadata::new(
272            seg_id(),
273            start,
274            end,
275            123,
276            1,
277            456,
278            RemoteLogSegmentDetails::new(
279                1024,
280                RemoteLogSegmentState::CopySegmentStarted,
281                leader_epochs,
282            ),
283        )
284    }
285
286    fn started_metadata() -> RemoteLogSegmentMetadata { metadata_range(0, 10, epochs()).unwrap() }
287
288    #[test]
289    fn accessors_return_constructed_values() {
290        // max_timestamp_ms / segment_size_in_bytes accessors were never read
291        // back in the suite; pin them to distinct non-default values.
292        let md = RemoteLogSegmentMetadata::new(
293            seg_id(),
294            0,
295            100,
296            777, // max_timestamp_ms
297            5,
298            888,
299            RemoteLogSegmentDetails::new(
300                4096, // segment_size_in_bytes
301                RemoteLogSegmentState::CopySegmentStarted,
302                epochs(),
303            ),
304        )
305        .unwrap();
306        assert!(md.max_timestamp_ms() == 777);
307        assert!(md.segment_size_in_bytes() == 4096);
308    }
309
310    #[test]
311    fn metadata_rejects_empty_leader_epochs() {
312        let err = metadata_range(0, 10, BTreeMap::new()).unwrap_err();
313        assert!(matches!(err, RemoteStorageError::InvalidArgument(_)));
314    }
315
316    #[test]
317    fn metadata_rejects_end_before_start() {
318        let err = metadata_range(10, 5, epochs()).unwrap_err();
319        assert!(matches!(err, RemoteStorageError::InvalidArgument(_)));
320    }
321
322    #[test]
323    fn with_update_advances_state_and_fields() {
324        let started = started_metadata();
325        let update = RemoteLogSegmentMetadataUpdate {
326            remote_log_segment_id: seg_id(),
327            event_timestamp_ms: 789,
328            custom_metadata: Some(CustomMetadata(vec![1, 2, 3])),
329            state: RemoteLogSegmentState::CopySegmentFinished,
330            broker_id: 2,
331        };
332        let finished = started.with_update(&update).unwrap();
333        check!(finished.state() == RemoteLogSegmentState::CopySegmentFinished);
334        check!(finished.event_timestamp_ms() == 789);
335        check!(finished.broker_id() == 2);
336        check!(finished.custom_metadata() == Some(&CustomMetadata(vec![1, 2, 3])));
337        // Untouched fields survive.
338        check!(finished.start_offset() == 0);
339        check!(finished.end_offset() == 10);
340    }
341
342    #[test]
343    fn with_update_keeps_custom_metadata_when_update_omits_it() {
344        let started = started_metadata().with_custom_metadata(CustomMetadata(vec![9]));
345        let update = RemoteLogSegmentMetadataUpdate {
346            remote_log_segment_id: seg_id(),
347            event_timestamp_ms: 789,
348            custom_metadata: None,
349            state: RemoteLogSegmentState::CopySegmentFinished,
350            broker_id: 2,
351        };
352        let finished = started.with_update(&update).unwrap();
353        assert!(finished.custom_metadata() == Some(&CustomMetadata(vec![9])));
354    }
355
356    #[test]
357    fn with_update_rejects_invalid_transition() {
358        let started = started_metadata();
359        let update = RemoteLogSegmentMetadataUpdate {
360            remote_log_segment_id: seg_id(),
361            event_timestamp_ms: 789,
362            custom_metadata: None,
363            state: RemoteLogSegmentState::DeleteSegmentFinished,
364            broker_id: 2,
365        };
366        let err = started.with_update(&update).unwrap_err();
367        assert!(matches!(
368            err,
369            RemoteStorageError::InvalidSegmentTransition { .. }
370        ));
371    }
372
373    #[test]
374    fn with_update_rejects_mismatched_id() {
375        let started = started_metadata();
376        let other = RemoteLogSegmentId::new(tp(), Uuid::from_u128(1234));
377        let update = RemoteLogSegmentMetadataUpdate {
378            remote_log_segment_id: other,
379            event_timestamp_ms: 789,
380            custom_metadata: None,
381            state: RemoteLogSegmentState::CopySegmentFinished,
382            broker_id: 2,
383        };
384        let err = started.with_update(&update).unwrap_err();
385        assert!(matches!(err, RemoteStorageError::InvalidArgument(_)));
386    }
387
388    #[test]
389    fn txn_index_empty_defaults_false_and_is_settable() {
390        let md = RemoteLogSegmentMetadata::new(
391            RemoteLogSegmentId::new(
392                TopicIdPartition::new(Uuid::from_u128(1), "t", 0),
393                Uuid::from_u128(2),
394            ),
395            0,
396            9,
397            9,
398            1,
399            100,
400            RemoteLogSegmentDetails::new(
401                1024,
402                RemoteLogSegmentState::CopySegmentStarted,
403                maplit::btreemap! {LeaderEpoch(0) => 0},
404            ),
405        )
406        .unwrap();
407        assert!(!md.txn_index_empty());
408        let md = md.with_txn_index_empty(true);
409        assert!(md.txn_index_empty());
410    }
411}
krabka_remote_storage/metadata/segment_metadata.rs

krabka_remote_storage/metadata/
segment_metadata.rs
