krabka_remote_storage_topic/
swappable.rs

1//! [`SwappableRlmm`]: a [`RemoteLogMetadataManager`] facade whose backing
2//! implementation can be replaced atomically after construction.
3//!
4//! It lets [`crate::TopicBasedRemoteLogMetadataManager`] take over from
5//! [`crate::NotReadyRlmm`] *after* the
6//! broker's listener accept loop is serving, so the manager's loopback
7//! `AdminClient` call to provision `__remote_log_metadata` has a server to
8//! talk to. Until a caller calls [`SwappableRlmm::swap`], every RLMM method
9//! delegates to the placeholder. After the swap, every call delegates to the
10//! new implementation.
11//!
12//! ## Invariant during the swap window
13//!
14//! Trait calls between construction and the first `swap` go to the
15//! placeholder, which is normally [`crate::NotReadyRlmm`]. Once `swap` returns,
16//! every later call uses the new implementation. Each trait method snapshots
17//! the inner [`Arc`] under a read lock and releases the lock before it
18//! delegates. A call already in progress can finish on the old implementation
19//! after the swap. Separate calls within a larger operation can therefore use
20//! different implementations.
21
22use {
23    krabka_ids::LeaderEpoch,
24    krabka_remote_storage::{
25        RemoteLogMetadataManager, RemoteLogSegmentMetadata, RemoteLogSegmentMetadataUpdate,
26        RemotePartitionDeleteMetadata, RemoteStorageError, TopicIdPartition,
27    },
28    std::sync::{Arc, PoisonError, RwLock},
29};
30
31/// Hot-swappable [`RemoteLogMetadataManager`] wrapper.
32pub struct SwappableRlmm {
33    inner: RwLock<Arc<dyn RemoteLogMetadataManager>>,
34}
35
36impl SwappableRlmm {
37    /// Construct the facade with an initial backing implementation. The
38    /// first [`Self::swap`], if there is one, replaces it.
39    #[must_use]
40    pub fn new(initial: Arc<dyn RemoteLogMetadataManager>) -> Self {
41        Self {
42            inner: RwLock::new(initial),
43        }
44    }
45
46    /// Replace the backing implementation. The next trait call
47    /// observes the new one.
48    pub fn swap(&self, new: Arc<dyn RemoteLogMetadataManager>) {
49        *self.inner.write().unwrap_or_else(PoisonError::into_inner) = new;
50    }
51
52    /// Snapshot the current inner [`Arc`] and do not hold the read lock past
53    /// the call site. Every trait method below calls this.
54    fn current(&self) -> Arc<dyn RemoteLogMetadataManager> {
55        self.inner
56            .read()
57            .unwrap_or_else(PoisonError::into_inner)
58            .clone()
59    }
60}
61
62impl RemoteLogMetadataManager for SwappableRlmm {
63    fn add_remote_log_segment_metadata(
64        &self,
65        metadata: RemoteLogSegmentMetadata,
66    ) -> Result<(), RemoteStorageError> {
67        self.current().add_remote_log_segment_metadata(metadata)
68    }
69
70    fn update_remote_log_segment_metadata(
71        &self,
72        update: RemoteLogSegmentMetadataUpdate,
73    ) -> Result<(), RemoteStorageError> {
74        self.current().update_remote_log_segment_metadata(update)
75    }
76
77    fn remote_log_segment_metadata(
78        &self,
79        topic_id_partition: &TopicIdPartition,
80        leader_epoch: LeaderEpoch,
81        offset: i64,
82    ) -> Result<Option<RemoteLogSegmentMetadata>, RemoteStorageError> {
83        self.current()
84            .remote_log_segment_metadata(topic_id_partition, leader_epoch, offset)
85    }
86
87    fn highest_offset_for_epoch(
88        &self,
89        topic_id_partition: &TopicIdPartition,
90        leader_epoch: LeaderEpoch,
91    ) -> Result<Option<i64>, RemoteStorageError> {
92        self.current()
93            .highest_offset_for_epoch(topic_id_partition, leader_epoch)
94    }
95
96    fn list_remote_log_segments(
97        &self,
98        topic_id_partition: &TopicIdPartition,
99    ) -> Result<Vec<RemoteLogSegmentMetadata>, RemoteStorageError> {
100        self.current().list_remote_log_segments(topic_id_partition)
101    }
102
103    fn list_remote_log_segments_by_epoch(
104        &self,
105        topic_id_partition: &TopicIdPartition,
106        leader_epoch: LeaderEpoch,
107    ) -> Result<Vec<RemoteLogSegmentMetadata>, RemoteStorageError> {
108        self.current()
109            .list_remote_log_segments_by_epoch(topic_id_partition, leader_epoch)
110    }
111
112    fn put_remote_partition_delete_metadata(
113        &self,
114        metadata: RemotePartitionDeleteMetadata,
115    ) -> Result<(), RemoteStorageError> {
116        self.current()
117            .put_remote_partition_delete_metadata(metadata)
118    }
119
120    fn caught_up(&self, topic_id_partition: &TopicIdPartition) -> Result<bool, RemoteStorageError> {
121        self.current().caught_up(topic_id_partition)
122    }
123}
124
125#[cfg(test)]
126mod tests {
127    use {
128        super::*,
129        crate::{
130            NotReadyRlmm,
131            manager::test_support::{started, tp},
132        },
133        assert2::assert,
134        krabka_remote_storage::InmemoryRemoteLogMetadataManager,
135    };
136
137    #[test]
138    fn delegates_to_initial_then_to_swapped() {
139        let first: Arc<dyn RemoteLogMetadataManager> =
140            Arc::new(InmemoryRemoteLogMetadataManager::new());
141        let swap = SwappableRlmm::new(Arc::clone(&first));
142
143        // Write through the facade lands in `first`.
144        swap.add_remote_log_segment_metadata(started(10, 0, 99))
145            .unwrap();
146        assert!(first.list_remote_log_segments(&tp()).unwrap().len() == 1);
147
148        // Swap in a fresh empty backing impl; reads now show the new
149        // (empty) one, not the original.
150        let second: Arc<dyn RemoteLogMetadataManager> =
151            Arc::new(InmemoryRemoteLogMetadataManager::new());
152        swap.swap(Arc::clone(&second));
153        assert!(swap.list_remote_log_segments(&tp()).unwrap().is_empty());
154        // The previous `first` is undisturbed — the facade just stopped
155        // pointing at it.
156        assert!(first.list_remote_log_segments(&tp()).unwrap().len() == 1);
157
158        // Writes after the swap go to `second`.
159        swap.add_remote_log_segment_metadata(started(11, 100, 199))
160            .unwrap();
161        assert!(second.list_remote_log_segments(&tp()).unwrap().len() == 1);
162        assert!(first.list_remote_log_segments(&tp()).unwrap().len() == 1);
163    }
164
165    #[test]
166    fn caught_up_delegates_to_the_current_inner() {
167        let swap = SwappableRlmm::new(Arc::new(NotReadyRlmm::new()));
168        assert!(let Err(RemoteStorageError::NotReady { .. }) = swap.caught_up(&tp()));
169        swap.swap(Arc::new(InmemoryRemoteLogMetadataManager::new()));
170        assert!(let Ok(true) = swap.caught_up(&tp()));
171    }
172}
krabka_remote_storage_topic/swappable.rs

krabka_remote_storage_topic/
swappable.rs
