krabka_remote_storage_topic/
not_ready.rs

1//! [`NotReadyRlmm`]: a fail-closed [`RemoteLogMetadataManager`] placeholder.
2//!
3//! This type is the [`SwappableRlmm`](crate::SwappableRlmm) placeholder during
4//! the topic-backed bootstrap window. Every method returns
5//! [`RemoteStorageError::NotReady`] immediately:
6//!
7//! - The copy task's first call, `add_remote_log_segment_metadata`, fails with
8//!   `NotReady` before it copies any data. Nothing is therefore tiered with
9//!   non-durable metadata, and no orphaned RSM objects accumulate in the
10//!   remote store while the real manager still initialises.
11//! - Remote reads return a retryable `NotReady` error rather than a definitive
12//!   `Ok(None)`, so callers retry until the real manager swaps in and catches
13//!   up to the high-water mark.
14//!
15//! This placeholder is stricter than
16//! [`krabka_remote_storage::InmemoryRemoteLogMetadataManager`], which silently
17//! accepts writes that are then lost when the real manager replaces it.
18
19use {
20    krabka_ids::LeaderEpoch,
21    krabka_remote_storage::{
22        RemoteLogMetadataManager, RemoteLogSegmentMetadata, RemoteLogSegmentMetadataUpdate,
23        RemotePartitionDeleteMetadata, RemoteStorageError, TopicIdPartition,
24    },
25};
26
27/// A [`RemoteLogMetadataManager`] that returns
28/// [`RemoteStorageError::NotReady`] for every method.
29///
30/// Construct it with [`NotReadyRlmm::new`] and pass it as the initial
31/// placeholder to [`crate::SwappableRlmm::new`].
32#[derive(Default)]
33pub struct NotReadyRlmm;
34
35impl NotReadyRlmm {
36    /// Construct a new `NotReadyRlmm`.
37    #[must_use]
38    pub fn new() -> Self { Self }
39}
40
41impl RemoteLogMetadataManager for NotReadyRlmm {
42    fn add_remote_log_segment_metadata(
43        &self,
44        metadata: RemoteLogSegmentMetadata,
45    ) -> Result<(), RemoteStorageError> {
46        let partition = metadata
47            .remote_log_segment_id()
48            .topic_id_partition
49            .partition;
50        Err(RemoteStorageError::NotReady { partition })
51    }
52
53    fn update_remote_log_segment_metadata(
54        &self,
55        update: RemoteLogSegmentMetadataUpdate,
56    ) -> Result<(), RemoteStorageError> {
57        let partition = update.remote_log_segment_id.topic_id_partition.partition;
58        Err(RemoteStorageError::NotReady { partition })
59    }
60
61    fn remote_log_segment_metadata(
62        &self,
63        topic_id_partition: &TopicIdPartition,
64        _leader_epoch: LeaderEpoch,
65        _offset: i64,
66    ) -> Result<Option<RemoteLogSegmentMetadata>, RemoteStorageError> {
67        Err(RemoteStorageError::NotReady {
68            partition: topic_id_partition.partition,
69        })
70    }
71
72    fn highest_offset_for_epoch(
73        &self,
74        topic_id_partition: &TopicIdPartition,
75        _leader_epoch: LeaderEpoch,
76    ) -> Result<Option<i64>, RemoteStorageError> {
77        Err(RemoteStorageError::NotReady {
78            partition: topic_id_partition.partition,
79        })
80    }
81
82    fn list_remote_log_segments(
83        &self,
84        topic_id_partition: &TopicIdPartition,
85    ) -> Result<Vec<RemoteLogSegmentMetadata>, RemoteStorageError> {
86        Err(RemoteStorageError::NotReady {
87            partition: topic_id_partition.partition,
88        })
89    }
90
91    fn list_remote_log_segments_by_epoch(
92        &self,
93        topic_id_partition: &TopicIdPartition,
94        _leader_epoch: LeaderEpoch,
95    ) -> Result<Vec<RemoteLogSegmentMetadata>, RemoteStorageError> {
96        Err(RemoteStorageError::NotReady {
97            partition: topic_id_partition.partition,
98        })
99    }
100
101    fn put_remote_partition_delete_metadata(
102        &self,
103        metadata: RemotePartitionDeleteMetadata,
104    ) -> Result<(), RemoteStorageError> {
105        let partition = metadata.topic_id_partition.partition;
106        Err(RemoteStorageError::NotReady { partition })
107    }
108
109    fn caught_up(&self, topic_id_partition: &TopicIdPartition) -> Result<bool, RemoteStorageError> {
110        Err(RemoteStorageError::NotReady {
111            partition: topic_id_partition.partition,
112        })
113    }
114}
115
116#[cfg(test)]
117mod tests {
118    use {
119        super::*,
120        assert2::assert,
121        krabka_remote_storage::{
122            RemoteLogSegmentId, RemoteLogSegmentState, RemotePartitionDeleteState,
123        },
124        uuid::Uuid,
125    };
126
127    krabka_macros::remote_started_segment!(synthetic_started_segment, krabka_remote_storage);
128
129    fn tp() -> TopicIdPartition { TopicIdPartition::new(Uuid::from_u128(1), "orders", 3) }
130
131    fn seg_id(id: u128) -> RemoteLogSegmentId { RemoteLogSegmentId::new(tp(), Uuid::from_u128(id)) }
132
133    #[test]
134    fn reads_return_not_ready_with_partition() {
135        let m = NotReadyRlmm::new();
136        let err = m
137            .remote_log_segment_metadata(&tp(), LeaderEpoch(0), 0)
138            .unwrap_err();
139        assert!(matches!(err, RemoteStorageError::NotReady { partition: 3 }));
140
141        let err = m.list_remote_log_segments(&tp()).unwrap_err();
142        assert!(matches!(err, RemoteStorageError::NotReady { partition: 3 }));
143
144        let err = m
145            .list_remote_log_segments_by_epoch(&tp(), LeaderEpoch(0))
146            .unwrap_err();
147        assert!(matches!(err, RemoteStorageError::NotReady { partition: 3 }));
148
149        let err = m
150            .highest_offset_for_epoch(&tp(), LeaderEpoch(0))
151            .unwrap_err();
152        assert!(matches!(err, RemoteStorageError::NotReady { partition: 3 }));
153
154        let err = m.caught_up(&tp()).unwrap_err();
155        assert!(matches!(err, RemoteStorageError::NotReady { partition: 3 }));
156    }
157
158    #[test]
159    fn add_returns_not_ready_with_partition() {
160        let m = NotReadyRlmm::new();
161        let md = synthetic_started_segment(seg_id(10), 0, 99, 100);
162        let err = m.add_remote_log_segment_metadata(md).unwrap_err();
163        assert!(matches!(err, RemoteStorageError::NotReady { partition: 3 }));
164    }
165
166    #[test]
167    fn update_returns_not_ready_with_partition() {
168        let m = NotReadyRlmm::new();
169        let upd = RemoteLogSegmentMetadataUpdate {
170            remote_log_segment_id: seg_id(11),
171            event_timestamp_ms: 1,
172            custom_metadata: None,
173            state: RemoteLogSegmentState::CopySegmentFinished,
174            broker_id: 0,
175        };
176        let err = m.update_remote_log_segment_metadata(upd).unwrap_err();
177        assert!(matches!(err, RemoteStorageError::NotReady { partition: 3 }));
178    }
179
180    #[test]
181    fn put_partition_delete_returns_not_ready_with_partition() {
182        let m = NotReadyRlmm::new();
183        let del = RemotePartitionDeleteMetadata {
184            topic_id_partition: tp(),
185            state: RemotePartitionDeleteState::DeletePartitionMarked,
186            event_timestamp_ms: 1,
187            broker_id: 0,
188        };
189        let err = m.put_remote_partition_delete_metadata(del).unwrap_err();
190        assert!(matches!(err, RemoteStorageError::NotReady { partition: 3 }));
191    }
192}
krabka_remote_storage_topic/not_ready.rs

krabka_remote_storage_topic/
not_ready.rs
