krabka_raft/controller/
metadata_fetch.rs

1//! Committed `__cluster_metadata` reads for observers: the local record-batch
2//! slice this node serves out of its own log, and the one-shot outbound fetch a
3//! broker-only observer issues against a controller listener.
4
5use {
6    super::{ControllerHandle, connection_options},
7    crate::{
8        error::RaftError,
9        kraft::MetadataFetchSlice,
10        types::NodeId,
11        wire::{
12            API_KEY_METADATA_FETCH, KrabkaMetadataFetchRequest, KrabkaMetadataFetchResponse,
13            PRIVATE_UNSUPPORTED_VERSION, private_request_version,
14        },
15    },
16    krabka_units::prelude::{ByteSize, ByteSizeExt as _},
17    std::net::SocketAddr,
18};
19
20impl ControllerHandle {
21    /// Read committed `__cluster_metadata` entries starting at `fetch_offset`,
22    /// encoded as Kafka record batches for an observer.
23    #[must_use]
24    pub async fn metadata_records(
25        &self,
26        fetch_offset: u64,
27        max_size: ByteSize,
28    ) -> MetadataFetchSlice {
29        let off = i64::try_from(fetch_offset).unwrap_or(i64::MAX);
30        self.engine
31            .metadata_fetch(off, max_size, None)
32            .await
33            .unwrap_or(MetadataFetchSlice {
34                records: bytes::Bytes::new(),
35                snapshot_id: None,
36                log_start_offset: 0,
37                high_watermark: 0,
38                quorum_high_watermark: 0,
39            })
40    }
41
42    /// Dial a controller-listener `addr` and issue one `API_KEY_METADATA_FETCH`.
43    /// Used by broker-only observers to pull committed `__cluster_metadata`.
44    ///
45    /// # Errors
46    /// - [`RaftError::Network`] if the dial or request fails.
47    /// - [`RaftError::Protocol`] if the response cannot be decoded.
48    pub async fn fetch_metadata_from(
49        &self,
50        addr: SocketAddr,
51        fetch_offset: u64,
52        max_size: ByteSize,
53    ) -> Result<KrabkaMetadataFetchResponse, RaftError> {
54        let req = KrabkaMetadataFetchRequest {
55            fetch_offset: i64::try_from(fetch_offset).unwrap_or(i64::MAX),
56            // `max_bytes` is the KIP-595-shaped `int32` on the Krabka observer
57            // wire; the quantity converts here and nowhere deeper.
58            max_bytes: max_size.bytes_i32(),
59            // A one-shot read, not a replica: the leader tracks nothing for it.
60            replica_id: -1,
61            replica_directory_id: uuid::Uuid::nil(),
62        };
63        // Negotiated from `krabka.version`; every level this build knows
64        // gives v0, the body below.
65        let version = private_request_version(
66            krabka_metadata::PrivateRpc::MetadataFetch,
67            Some(&self.current_image()),
68        );
69        let mut body = Vec::with_capacity(32);
70        req.encode_v0(&mut body);
71
72        let opts = connection_options(
73            &self.client_id,
74            self.client_dispatch_queue_capacity,
75            self.client_frame_max,
76        );
77        let conn = self
78            .dialer
79            .dial(NodeId(1), &addr.to_string(), opts)
80            .await
81            .map_err(RaftError::Network)?;
82        let resp_body = conn
83            .raw_request(API_KEY_METADATA_FETCH, version, bytes::Bytes::from(body))
84            .await
85            .map_err(RaftError::Network)?;
86        conn.close();
87
88        let mut cur: &[u8] = &resp_body;
89        let response =
90            KrabkaMetadataFetchResponse::decode_v0(&mut cur).map_err(RaftError::Protocol)?;
91        if response.error_code == PRIVATE_UNSUPPORTED_VERSION {
92            return Err(RaftError::UnsupportedPrivateVersion {
93                api_key: API_KEY_METADATA_FETCH,
94                version,
95            });
96        }
97        Ok(response)
98    }
99}
100
101#[cfg(test)]
102mod tests {
103    use {
104        super::*,
105        crate::{
106            config::ControllerConfig,
107            controller::{
108                Controller,
109                test_support::{
110                    FAST_ELECTION_TIMEOUT, RecordingDialer, TEST_OP_TIMEOUT, UNBOUNDED_FETCH,
111                    bootstrap_controller, committable_topic_record, submit_change_with_timeout,
112                    wait_for_leader,
113                },
114            },
115        },
116        krabka_metadata::{MetadataImage, MetadataRecord, from_kraft_value},
117        krabka_protocol::records::RecordBatch,
118        krabka_units::prelude::{TimeExt as _, mebibytes},
119        std::sync::{Arc, Mutex},
120        tempfile::TempDir,
121        tokio::time,
122    };
123
124    fn contains_topic(records: &[u8], name: &str) -> bool {
125        let image = MetadataImage::new(uuid::Uuid::nil());
126        let mut buf = records;
127        let mut found = false;
128        while !buf.is_empty() {
129            let batch = RecordBatch::decode(&mut buf).expect("decode");
130            if batch.attributes.is_control_batch() {
131                continue;
132            }
133            for r in &batch.records {
134                let Some(value) = r.value.as_ref() else {
135                    continue;
136                };
137                if let Ok(MetadataRecord::V1Topic(t)) = from_kraft_value(value, &image)
138                    && t.name == name
139                {
140                    found = true;
141                }
142            }
143        }
144        found
145    }
146
147    /// A bootstrapped controller with one committed topic, and the data
148    /// directory it must outlive.
149    struct SeededController {
150        dir: TempDir,
151        ctrl: ControllerHandle,
152    }
153
154    async fn committed_topic(name: &str, label: &str) -> SeededController {
155        let (dir, ctrl) = bootstrap_controller("bootstrap").await;
156        wait_for_leader(&ctrl).await;
157        submit_change_with_timeout(&ctrl, vec![committable_topic_record(name)], label)
158            .await
159            .expect("submit");
160        SeededController { dir, ctrl }
161    }
162
163    #[tokio::test]
164    async fn metadata_records_serves_committed_topic() {
165        let SeededController { dir: _dir, ctrl } =
166            committed_topic("t", "metadata_records seed").await;
167
168        let slice = time::timeout(
169            TEST_OP_TIMEOUT.to_std(),
170            ctrl.metadata_records(0, UNBOUNDED_FETCH),
171        )
172        .await
173        .expect("metadata_records timed out");
174        assert2::assert!(slice.high_watermark >= 1);
175        assert2::assert!(contains_topic(&slice.records, "t"));
176        ctrl.shutdown().await;
177    }
178
179    #[tokio::test]
180    async fn fetch_metadata_from_returns_committed_records() {
181        let SeededController { dir: _dir, ctrl } =
182            committed_topic("fetched", "fetch_metadata seed").await;
183
184        let addr = ctrl.controller_bound_addr();
185        let resp = time::timeout(
186            TEST_OP_TIMEOUT.to_std(),
187            ctrl.fetch_metadata_from(addr, 0, mebibytes(1)),
188        )
189        .await
190        .expect("fetch_metadata_from timed out")
191        .expect("fetch");
192        assert2::assert!(resp.error_code == 0);
193        assert2::assert!(resp.high_watermark >= 1);
194
195        assert2::assert!(contains_topic(&resp.records, "fetched"));
196        ctrl.shutdown().await;
197    }
198
199    #[tokio::test]
200    async fn fetch_metadata_from_passes_configured_client_id_to_dialer() {
201        let dir = TempDir::new().unwrap();
202        let client_ids = Arc::new(Mutex::new(Vec::new()));
203        let recorded_options = Arc::new(Mutex::new(Vec::new()));
204        let dialer = RecordingDialer {
205            client_ids: Arc::clone(&client_ids),
206            options: Arc::clone(&recorded_options),
207        };
208        let mut cfg = ControllerConfig::for_tests(NodeId(1), dir.path().to_path_buf());
209        cfg.election_timeout = FAST_ELECTION_TIMEOUT;
210        cfg.client_id = "metadata-fetch-client".into();
211        cfg.client_dispatch_queue_capacity =
212            krabka_client_core::ConnectionDispatchQueueCapacity::new(7).unwrap();
213        cfg.client_frame_max = krabka_client_core::ClientFrameMax::try_from(mebibytes(8)).unwrap();
214        cfg.dialer = Some(Arc::new(dialer));
215
216        let ctrl = Controller::start(cfg).await.expect("bootstrap");
217        wait_for_leader(&ctrl).await;
218        submit_change_with_timeout(
219            &ctrl,
220            vec![committable_topic_record("client-id-check")],
221            "client-id fetch seed",
222        )
223        .await
224        .expect("submit");
225
226        let resp = time::timeout(
227            TEST_OP_TIMEOUT.to_std(),
228            ctrl.fetch_metadata_from(ctrl.controller_bound_addr(), 0, mebibytes(1)),
229        )
230        .await
231        .expect("fetch_metadata_from timed out")
232        .expect("fetch");
233
234        assert2::assert!(resp.error_code == 0);
235        assert2::assert!(client_ids.lock().unwrap().as_slice() == ["metadata-fetch-client"]);
236        {
237            let opts = recorded_options.lock().unwrap().clone();
238            let first_opt = &opts[0];
239            assert2::assert!(opts.len() == 1);
240            let expected_cap = krabka_client_core::ConnectionDispatchQueueCapacity::new(7).unwrap();
241            assert2::assert!(first_opt.dispatch_queue_capacity == expected_cap);
242            let expected_frame =
243                krabka_client_core::ClientFrameMax::try_from(mebibytes(8)).unwrap();
244            assert2::assert!(first_opt.frame_max == expected_frame);
245        }
246        ctrl.shutdown().await;
247    }
248}
krabka_raft/controller/metadata_fetch.rs

krabka_raft/controller/
metadata_fetch.rs
