krabka_remote_storage_topic/
log.rs

1//! [`MetadataEventLog`]: the publish/subscribe seam between the
2//! [`TopicBasedRemoteLogMetadataManager`](crate::TopicBasedRemoteLogMetadataManager)
3//! and the underlying durable event store.
4//!
5//! The in-process implementation, [`InProcessMetadataEventLog`], is an
6//! in-memory broadcast-channel fixture that unit tests use. It is also a
7//! single-process model for the multi-broker case, because multiple manager
8//! instances that share the same fixture observe each other's writes. The
9//! production Kafka-backed adapter implements the same trait.
10//!
11//! [`MetadataEventLog::subscribe`] does not consume every partition from
12//! offset 0. It takes an explicit [`PartitionStart`] assignment, which is a
13//! subset of partitions, each with its own start offset. It returns an
14//! [`AssignmentHandle`] that can mutate the live assignment at runtime.
15
16use {
17    crate::error::MetadataLogError,
18    async_trait::async_trait,
19    bytes::Bytes,
20    futures_util::stream::Stream,
21    std::{pin::Pin, sync::Arc},
22};
23
24mod in_process;
25
26pub use self::in_process::InProcessMetadataEventLog;
27
28/// One event read from the metadata log.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct MetadataEventRecord {
31    /// Metadata-topic partition the event came from.
32    pub partition: i32,
33    /// Offset within that partition.
34    pub offset: i64,
35    /// Record key. Compacted metadata topics use it to identify superseded
36    /// state; ordinary append-only metadata events leave it unset.
37    pub key: Option<Bytes>,
38    /// Encoded event payload. See [`crate::serde`].
39    pub payload: Bytes,
40    /// Whether Kafka carried a null value for this key.
41    pub tombstone: bool,
42}
43
44/// One keyed record of a [`MetadataEventLog::publish_keyed_group`] call.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct KeyedEvent {
47    /// Metadata-topic partition to append to.
48    pub partition: i32,
49    /// Compaction key.
50    pub key: Bytes,
51    /// Value, or `None` for a tombstone.
52    pub event: Option<Bytes>,
53}
54
55/// Boxed event stream the [`MetadataEventLog`] hands to subscribers.
56pub type MetadataEventStream = Pin<Box<dyn Stream<Item = MetadataEventRecord> + Send + 'static>>;
57
58/// Receives each record of [`MetadataEventLog::visit_range`]. An error stops
59/// the read, and the read returns it.
60pub type RangeVisitor<'a> =
61    dyn FnMut(MetadataEventRecord) -> Result<(), MetadataLogError> + Send + 'a;
62
63/// One partition to consume and the offset to begin at (inclusive).
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65pub struct PartitionStart {
66    /// Metadata-topic partition to consume.
67    pub partition: i32,
68    /// First offset to deliver (inclusive). `0` replays from the start.
69    pub start_offset: i64,
70}
71
72impl PartitionStart {
73    /// Consume one partition beginning at the inclusive offset.
74    #[must_use]
75    pub const fn new(partition: i32, start_offset: i64) -> Self {
76        Self {
77            partition,
78            start_offset,
79        }
80    }
81}
82
83/// Runtime control over a live [`MetadataEventLog`] subscription's assigned
84/// partition set. [`MetadataEventLog::subscribe`] returns it together with
85/// the stream.
86pub trait AssignmentHandle: Send + Sync {
87    /// Begin to consume `start.partition` from `start.start_offset`. This
88    /// method does nothing if the partition is already assigned. A
89    /// newly-added partition emits its backlog from `start_offset` into the
90    /// existing stream, and then live records.
91    fn add(&self, start: PartitionStart);
92    /// Stop the consumption of `partition` and stop the emission of its
93    /// events. This method does nothing if the partition is not currently
94    /// assigned.
95    fn remove(&self, partition: i32);
96    /// Current assigned partition set (unordered).
97    fn assigned(&self) -> Vec<i32>;
98}
99
100/// Publish/subscribe transport that backs the `__remote_log_metadata`
101/// topic.
102///
103/// Implementations must guarantee:
104///
105/// - `publish(p, _)` resolves to a monotonically-increasing offset
106///   within partition `p`, and the assigned offset is also what every
107///   subscriber observes for that record.
108/// - The stream returned by `subscribe` replays each assigned
109///   partition's backlog from its `start_offset` and then forwards new
110///   records as they are published for currently-assigned partitions.
111///   Subscribers attached after some records were already published
112///   still see the history at/after their start offset.
113/// - Records are delivered in publish order on a per-partition basis.
114#[async_trait]
115pub trait MetadataEventLog: Send + Sync {
116    /// Number of partitions the log holds. It is stable for the lifetime of
117    /// the log. The manager hashes user partitions into
118    /// `[0, partition_count())`.
119    fn partition_count(&self) -> i32;
120
121    /// Append `event` to `partition`. Resolves to the assigned offset.
122    ///
123    /// # Errors
124    ///
125    /// Returns [`MetadataLogError`] if the transport refused the write. That
126    /// happens when the partition is out of range, or when the log has been
127    /// closed.
128    async fn publish(&self, partition: i32, event: Bytes) -> Result<i64, MetadataLogError>;
129
130    /// Append a keyed value or tombstone to a compacted metadata topic.
131    ///
132    /// The default rejects keyed publication. A transport used for compaction
133    /// must override this method.
134    async fn publish_keyed(
135        &self,
136        _partition: i32,
137        _key: Bytes,
138        _event: Option<Bytes>,
139    ) -> Result<i64, MetadataLogError> {
140        Err(MetadataLogError::Other(
141            "metadata log does not support keyed publication".into(),
142        ))
143    }
144
145    /// Append a group of keyed records and wait until every one of them is
146    /// acknowledged. Resolves to their offsets, in the order given.
147    ///
148    /// Records for the same partition are appended in the order given. A
149    /// transport that pipelines writes submits the whole group before it
150    /// waits, so the group costs one acknowledgment round trip rather than
151    /// one per record. It still waits for every submitted record before it
152    /// returns, so none is left in flight after an error.
153    ///
154    /// The default publishes the records one at a time through
155    /// [`Self::publish_keyed`] and stops at the first failure.
156    ///
157    /// # Errors
158    ///
159    /// Returns the first [`MetadataLogError`] that any record met. Any subset
160    /// of the group may then have been appended.
161    async fn publish_keyed_group(
162        &self,
163        events: Vec<KeyedEvent>,
164    ) -> Result<Vec<i64>, MetadataLogError> {
165        let mut offsets = Vec::with_capacity(events.len());
166        for KeyedEvent {
167            partition,
168            key,
169            event,
170        } in events
171        {
172            offsets.push(self.publish_keyed(partition, key, event).await?);
173        }
174        Ok(offsets)
175    }
176
177    /// Start to consume the given partitions, each from its start offset,
178    /// which is inclusive. Returns the event stream and a handle to mutate
179    /// the live assignment.
180    ///
181    /// The stream replays each assigned partition's backlog from its
182    /// `start_offset`, then forwards live appends for the currently
183    /// assigned partitions. Records are delivered in publish order on
184    /// a per-partition basis.
185    fn subscribe(
186        &self,
187        assignment: Vec<PartitionStart>,
188    ) -> (MetadataEventStream, Arc<dyn AssignmentHandle>);
189
190    /// Earliest retained offset for each partition, indexed by partition.
191    ///
192    /// Append-only implementations retain offset zero. Compacted transports
193    /// override this so a replay does not wait forever on deleted offsets.
194    async fn low_water_marks(&self) -> Result<Vec<i64>, MetadataLogError> {
195        // A negative count yields no partitions, as an empty range would.
196        Ok(vec![
197            0;
198            usize::try_from(self.partition_count()).unwrap_or(0)
199        ])
200    }
201
202    /// One past the highest written offset for each partition,
203    /// indexed by partition.
204    ///
205    /// # Errors
206    ///
207    /// Returns [`MetadataLogError`] only on an underlying store failure. An
208    /// empty partition is `0`, not an error.
209    async fn high_water_marks(&self) -> Result<Vec<i64>, MetadataLogError>;
210
211    /// Pass every retained record of `partition` whose offset lies in
212    /// `[start, end)` to `visit`, in offset order.
213    ///
214    /// The read neither subscribes nor writes, so a reader that holds only
215    /// `READ` and `DESCRIBE` on the topic can use it. A transport hands
216    /// records over as it fetches them and holds at most one fetched page, so
217    /// the range can be larger than memory. A compacted transport steps over
218    /// the offsets compaction removed and still returns once it reaches `end`.
219    /// An empty range visits nothing.
220    ///
221    /// A networked transport retries transient failures, such as a leader
222    /// change, without a bound of its own. Bound the read with a deadline.
223    ///
224    /// The default reports that the transport does not serve range reads.
225    ///
226    /// # Errors
227    ///
228    /// Returns [`MetadataLogError::PartitionOutOfRange`] for a partition
229    /// outside the log, the first error that `visit` returns, and
230    /// [`MetadataLogError`] when the store fails or stops short of `end`.
231    async fn visit_range(
232        &self,
233        partition: i32,
234        start: i64,
235        end: i64,
236        visit: &mut RangeVisitor<'_>,
237    ) -> Result<(), MetadataLogError> {
238        let _ = (partition, start, end, visit);
239        Err(MetadataLogError::Other(
240            "metadata log does not serve range reads".into(),
241        ))
242    }
243}
244
245#[cfg(test)]
246mod tests {
247    use {
248        super::*,
249        std::sync::atomic::{AtomicUsize, Ordering},
250    };
251
252    struct AppendOnlyLog(AtomicUsize);
253
254    #[async_trait]
255    impl MetadataEventLog for AppendOnlyLog {
256        fn partition_count(&self) -> i32 { 1 }
257
258        async fn publish(&self, _partition: i32, _event: Bytes) -> Result<i64, MetadataLogError> {
259            self.0.fetch_add(1, Ordering::Relaxed);
260            Ok(7)
261        }
262
263        fn subscribe(
264            &self,
265            _assignment: Vec<PartitionStart>,
266        ) -> (MetadataEventStream, Arc<dyn AssignmentHandle>) {
267            unreachable!("not used by these tests")
268        }
269
270        async fn high_water_marks(&self) -> Result<Vec<i64>, MetadataLogError> { Ok(vec![0]) }
271    }
272
273    #[tokio::test]
274    async fn default_keyed_publish_rejects_values_and_tombstones_without_appending() {
275        let log = AppendOnlyLog(AtomicUsize::new(0));
276        assert2::assert!(log.partition_count() == 1);
277        assert2::assert!(log.high_water_marks().await.unwrap() == vec![0]);
278        assert2::assert!(
279            log.publish(0, Bytes::from_static(b"append-only"))
280                .await
281                .unwrap()
282                == 7
283        );
284        let value_error = log
285            .publish_keyed(
286                0,
287                Bytes::from_static(b"key"),
288                Some(Bytes::from_static(b"value")),
289            )
290            .await
291            .unwrap_err();
292        assert2::assert!(
293            value_error
294                .to_string()
295                .contains("does not support keyed publication")
296        );
297
298        let error = log
299            .publish_keyed(0, Bytes::from_static(b"key"), None)
300            .await
301            .unwrap_err();
302        assert2::assert!(
303            error
304                .to_string()
305                .contains("does not support keyed publication")
306        );
307        assert2::assert!(log.0.load(Ordering::Relaxed) == 1);
308    }
309
310    #[tokio::test]
311    async fn default_range_read_refuses_without_visiting() {
312        let mut visits = 0;
313        let error = AppendOnlyLog(AtomicUsize::new(0))
314            .visit_range(0, 0, 1, &mut |_record| {
315                visits += 1;
316                Ok(())
317            })
318            .await
319            .unwrap_err();
320
321        assert2::assert!(
322            error.to_string() == "metadata log error: metadata log does not serve range reads"
323        );
324        assert2::assert!(visits == 0);
325    }
326}
krabka_remote_storage_topic/log.rs

krabka_remote_storage_topic/
log.rs
