krabka_remote_storage_topic/
serde.rs

1//! Binary wire format for `__remote_log_metadata` events.
2//!
3//! This module encodes records through
4//! [`krabka_protocol::RemoteLogMetadataRecord`], which produces bytes that are
5//! byte-identical to the JVM `RemoteLogMetadataSerde`. That is the
6//! `AbstractApiMessageSerde` envelope, with all three header fields written as
7//! unsigned varints: `frameVersion(uvarint)=1 | apiKey(uvarint) |
8//! apiVersion(uvarint) | flexible-message-body`.
9//!
10//! apiKey mapping:
11//! - 0 = `RemoteLogSegmentMetadataRecord`  → [`MetadataEvent::AddSegment`]
12//! - 1 = `RemoteLogSegmentMetadataUpdateRecord` → [`MetadataEvent::UpdateSegment`]
13//! - 2 = `RemotePartitionDeleteMetadataRecord` → [`MetadataEvent::PartitionDelete`]
14//!
15//! One submodule per record, plus the scalar conversions they share:
16//! `segment_metadata` for apiKey 0, `segment_update` for apiKey 1,
17//! `partition_delete` for apiKey 2, and `primitives` for the UUID, state-byte
18//! and custom-metadata conversions all three use. The unrelated varint and
19//! cursor helpers that `snapshot.rs` frames its own envelope with live in
20//! `reader`.
21
22use {
23    crate::error::CodecError,
24    bytes::Bytes,
25    krabka_protocol::RemoteLogMetadataRecord,
26    krabka_remote_storage::{
27        RemoteLogSegmentMetadata, RemoteLogSegmentMetadataUpdate, RemotePartitionDeleteMetadata,
28    },
29};
30
31mod partition_delete;
32mod primitives;
33mod reader;
34mod segment_metadata;
35mod segment_update;
36#[cfg(test)]
37mod tests;
38
39pub(crate) use self::reader::{Reader, read_uvarint, write_uvarint};
40use self::{
41    partition_delete::{from_proto_partition_delete, to_proto_partition_delete},
42    segment_metadata::{from_proto_add, to_proto_add},
43    segment_update::{from_proto_update, to_proto_update},
44};
45
46/// One of the three event variants the topic carries.
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub enum MetadataEvent {
49    /// A new segment starts to copy (`CopySegmentStarted`).
50    AddSegment(RemoteLogSegmentMetadata),
51    /// Lifecycle transition for an existing segment.
52    UpdateSegment(RemoteLogSegmentMetadataUpdate),
53    /// Partition-delete lifecycle.
54    PartitionDelete(RemotePartitionDeleteMetadata),
55}
56
57impl MetadataEvent {
58    /// Encode this event into freshly-allocated [`Bytes`] with the JVM
59    /// `RemoteLogMetadataSerde` wire format.
60    ///
61    /// # Errors
62    ///
63    /// Returns [`CodecError::Protocol`] if the generated codec rejects the
64    /// record. That cannot happen for version 0, which every record uses.
65    pub fn encode(&self) -> Result<Bytes, CodecError> {
66        let record = match self {
67            Self::AddSegment(md) => RemoteLogMetadataRecord::SegmentMetadata(to_proto_add(md)),
68            Self::UpdateSegment(u) => {
69                RemoteLogMetadataRecord::SegmentMetadataUpdate(to_proto_update(u))
70            }
71            Self::PartitionDelete(d) => {
72                RemoteLogMetadataRecord::PartitionDelete(to_proto_partition_delete(d))
73            }
74        };
75        record
76            .encode_value()
77            .map_err(|error| CodecError::Protocol(error.to_string()))
78    }
79
80    /// Decode an event from `bytes`.
81    ///
82    /// # Errors
83    ///
84    /// Returns [`CodecError`] for any malformed input: a truncated envelope,
85    /// an unknown or unsupported apiKey, an out-of-range state byte, or a
86    /// domain invariant violation that [`RemoteLogSegmentMetadata::new`]
87    /// reports. apiKey 3, `SegmentMetadataSnapshot`, never appears on the
88    /// topic.
89    pub fn decode(bytes: &[u8]) -> Result<Self, CodecError> {
90        let record = RemoteLogMetadataRecord::decode_value(bytes)
91            .map_err(|e| CodecError::Protocol(e.to_string()))?;
92        match record {
93            RemoteLogMetadataRecord::SegmentMetadata(r) => Ok(Self::AddSegment(from_proto_add(r)?)),
94            RemoteLogMetadataRecord::SegmentMetadataUpdate(r) => {
95                Ok(Self::UpdateSegment(from_proto_update(r)?))
96            }
97            RemoteLogMetadataRecord::PartitionDelete(r) => {
98                Ok(Self::PartitionDelete(from_proto_partition_delete(r)?))
99            }
100            RemoteLogMetadataRecord::SegmentMetadataSnapshot(_) => Err(CodecError::Protocol(
101                "apiKey 3 (SegmentMetadataSnapshot) must not appear on __remote_log_metadata"
102                    .into(),
103            )),
104            RemoteLogMetadataRecord::Unknown { api_key, .. } => Err(CodecError::Protocol(format!(
105                "unknown __remote_log_metadata apiKey {api_key}"
106            ))),
107        }
108    }
109}
krabka_remote_storage_topic/serde.rs

krabka_remote_storage_topic/
serde.rs
