krabka_raft/
lib.rs

1//! Metadata Raft quorum for Krabka.
2//!
3//! `krabka-raft` runs a hand-rolled KIP-595 `KRaft` consensus engine, the
4//! [`kraft::KraftController`], over Krabka's storage ([`krabka_log`]) and
5//! transport ([`krabka_client_core`]). The public entry point is
6//! [`Controller::start`]. It spawns the engine and opens a TCP listener. That
7//! listener serves the real KIP-595 RPCs (Fetch=1, Vote=52,
8//! BeginQuorumEpoch=53, EndQuorumEpoch=54) and the Krabka-private observer and
9//! forward RPCs. [`Controller::start`] returns a [`ControllerHandle`], which
10//! submits metadata changes and reads the current
11//! [`krabka_metadata::MetadataImage`].
12//!
13//! ## Quick start
14//!
15//! ```no_run
16//! use {
17//!     krabka_metadata::{MetadataRecord, TopicRecord},
18//!     krabka_raft::{Controller, ControllerConfig, NodeId},
19//!     std::time::Duration,
20//!     uuid::Uuid,
21//! };
22//!
23//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
24//! let dir = tempfile::tempdir()?;
25//! let cfg = ControllerConfig::for_tests(NodeId(1), dir.path().to_path_buf());
26//! let controller = Controller::start(cfg).await?;
27//!
28//! controller
29//!     .submit_change(vec![MetadataRecord::V1Topic(TopicRecord {
30//!         name: "my-topic".into(),
31//!         topic_id: Uuid::new_v4(),
32//!         partitions: 3,
33//!         replication_factor: 1,
34//!     })])
35//!     .await?;
36//!
37//! assert2::assert!(controller.current_image().topic("my-topic").is_some());
38//! controller.shutdown().await;
39//! # Ok(())
40//! # }
41//! ```
42//!
43//! ## Capabilities and boundaries
44//!
45//! The controller persists and recovers `KRaft` metadata records, and it serves
46//! and installs KIP-630 snapshots through `FetchSnapshot`. It also publishes
47//! the current metadata image to broker tasks and exposes Krabka-private submit
48//! and fetch RPCs for broker and observer integration.
49//!
50//! KIP-853-style observer bootstrap and auto-join are wired through the broker
51//! and controller configuration. [`ControllerHandle::add_learner`] stages a
52//! node identity; [`ControllerHandle::change_membership`] reconciles a
53//! one-voter delta into `add_voter` or `remove_voter`.
54//! Mixed JVM and Krabka controller quorums are outside this crate's
55//! compatibility target.
56
57#![doc(html_root_url = "https://docs.rs/krabka-raft/1.1.0")]
58
59mod config;
60mod connection_limiter;
61mod controller;
62mod error;
63pub mod event_queue;
64pub mod handshake;
65pub mod kraft;
66mod network;
67mod observed_stream;
68pub mod reconfig;
69pub mod voter_requests;
70mod voter_wire;
71/// The deterministic `KRaft` failure-scenario simulator with trace recording,
72/// re-exported from the leaf [`krabka_kraft_core::sim`] module. `krabka-docgen`
73/// runs [`scenarios::scenarios`] in-process to render the failure-scenario
74/// slideshow.
75#[cfg(feature = "scenarios")]
76pub use krabka_kraft_core::sim as scenarios;
77mod server;
78mod snapshot;
79#[cfg(test)]
80mod test_support;
81mod types;
82mod wire;
83
84pub use {
85    config::{
86        BootstrapMode, ControllerAdminRequest, ControllerAdminResponse, ControllerAdminRouteFuture,
87        ControllerAdminRouter, ControllerApiVersion, ControllerConfig, ControllerFetchMissLimit,
88        DEFAULT_METADATA_LOG_SEGMENT_ROLL_INTERVAL, DEFAULT_METADATA_LOG_SEGMENT_SIZE,
89        DEFAULT_METADATA_MAX_IDLE_INTERVAL, DEFAULT_METADATA_MAX_RETENTION,
90        DEFAULT_METADATA_MAX_RETENTION_SIZE, KAFKA_4_3_1_APIS, LATEST_PRODUCTION_METADATA_VERSION,
91        ListenerLimits, METADATA_PARTITION_DIR, MIN_METADATA_LOG_SEGMENT_SIZE, MetadataLogConfig,
92        MetadataRaftCommandQueueCapacity, MetadataRaftFetchMax, RaftShardRouter,
93        RegistrationAdmission, RegistrationAdmissionFuture, ReleasedApi, ShardRouteFuture,
94        UnstableApiVersions, UnstableFeatureVersions, kafka_4_3_1_api, kafka_4_3_1_max,
95        metadata_partition_dir, supported_feature_range, supported_feature_ranges,
96    },
97    connection_limiter::{
98        AcceptingListener, ConnectionGuard, ConnectionLimit, ConnectionLimiter, ConnectionLimits,
99        ConnectionQuotaSettings, ConnectionQuotas, IpOverrides, MAX_CONNECTION_CREATION_RATE,
100        MAX_CONNECTIONS, MAX_CONNECTIONS_PER_IP, MAX_CONNECTIONS_PER_IP_OVERRIDES,
101    },
102    controller::{
103        Controller, ControllerHandle, QuorumState, QuorumStateSnapshot, SnapshotRange,
104        SnapshotSlice, metadata_log_nonempty,
105    },
106    error::{MetadataReplayError, PersistedFormatError, RaftError},
107    event_queue::{EventQueueRecorder, NoEventQueueRecorder},
108    handshake::{
109        AllowAllGrants, ClusterGrants, ClusterOperation, ControllerApiVersions, HandshakeStream,
110        RaftConnection, RaftHandshakeError, RaftListenerHandshake,
111    },
112    kraft::{
113        MetadataFetchSlice,
114        controller::{control_batch_image_records, decode_committed_value, is_kip835_noop},
115    },
116    network::{OutboundDialer, PlaintextDialer},
117    observed_stream::{ObservedStream, StreamObserver},
118    reconfig::{AddVoter, ReconfigOutcome, RemoveVoter, UpdateVoter},
119    server::{
120        api_versions_max_version, describe_quorum::describe_quorum, fetch_error_response,
121        finalized_feature_keys, is_valid_api_versions_request, is_valid_client_info,
122        supported_feature_key, supported_feature_keys, unsupported_version_response,
123    },
124    types::{
125        AppData, AppDataResponse, DelegationTokenMutation, Node, NodeId, OffsetReservation,
126        SubmitChangeResult,
127    },
128};
129
130/// Serialize a Kafka metadata snapshot, including KIP-853 control state.
131///
132/// # Errors
133/// Returns an error if a metadata or control record cannot be encoded.
134pub fn serialize_metadata_snapshot(
135    image: &krabka_metadata::MetadataImage,
136    last_contained_log_timestamp: i64,
137) -> Result<bytes::Bytes, RaftError> {
138    snapshot::SnapshotWriter::serialize(image, last_contained_log_timestamp)
139}
140
141/// Serialize the bootstrap checkpoint of a dynamic format, as Kafka's
142/// `Formatter.writeBoostrapSnapshot` writes it: the KIP-853 `kraft.version`
143/// and voter set, then the bootstrap records in their own order.
144///
145/// A controller does not apply these records to its image. The active
146/// controller writes them to an empty metadata log, and every replica applies
147/// them from there.
148///
149/// # Errors
150/// Returns an error if `kraft_version` exceeds `int16`, or if a voter or a
151/// bootstrap record cannot be encoded.
152pub fn serialize_bootstrap_snapshot(
153    kraft_version: u16,
154    voters: &krabka_metadata::VoterSet,
155    records: &[krabka_metadata::MetadataRecord],
156    last_contained_log_timestamp: i64,
157) -> Result<bytes::Bytes, RaftError> {
158    snapshot::SnapshotWriter::serialize_bootstrap(
159        &snapshot::SnapshotControlState {
160            kraft_version,
161            voters: voters.clone(),
162        },
163        records,
164        last_contained_log_timestamp,
165    )
166}
167
168/// Decode the KIP-630 metadata records from a Kafka metadata snapshot.
169///
170/// KIP-853 quorum controls are intentionally omitted: a restore formats a new
171/// quorum, while the returned records describe the cluster state it recovers.
172///
173/// # Errors
174/// Returns an error when the snapshot framing, ordering, or a metadata record
175/// is invalid.
176pub fn deserialize_metadata_snapshot(
177    bytes: &[u8],
178) -> Result<Vec<krabka_metadata::MetadataRecord>, RaftError> {
179    Ok(snapshot::SnapshotReader::read(bytes)?.metadata_records)
180}
181
182/// Decode the whole image a Kafka metadata snapshot holds: its KIP-853 quorum
183/// controls, as a `V1KRaftVersion` and a `V1Voters` record, ahead of its
184/// KIP-630 metadata records.
185///
186/// A broker-only observer installs a snapshot through this, so its image names
187/// the voters, and the endpoints that reach them, as a controller's image does.
188///
189/// # Errors
190/// Returns an error when the snapshot framing, ordering, a control record, or a
191/// metadata record is invalid.
192pub fn deserialize_metadata_snapshot_image(
193    bytes: &[u8],
194) -> Result<Vec<krabka_metadata::MetadataRecord>, RaftError> {
195    let snapshot = snapshot::SnapshotReader::read(bytes)?;
196    let controls = snapshot.control_state.into_iter().flat_map(|controls| {
197        [
198            krabka_metadata::MetadataRecord::V1KRaftVersion(krabka_metadata::KRaftVersionRecord {
199                kraft_version: controls.kraft_version,
200            }),
201            krabka_metadata::MetadataRecord::V1Voters(krabka_metadata::VotersRecord {
202                voters: controls.voters,
203            }),
204        ]
205    });
206    Ok(controls.chain(snapshot.metadata_records).collect())
207}
208
209/// Read the append timestamp of the last record batch contained in a KIP-630
210/// metadata snapshot.
211///
212/// # Errors
213/// Returns an error when the snapshot header is missing or malformed.
214pub fn metadata_snapshot_last_contained_log_timestamp(bytes: &[u8]) -> Result<i64, RaftError> {
215    snapshot::SnapshotReader::last_contained_log_timestamp(bytes)
216}
217pub use wire::{
218    API_KEY_DELEGATION_TOKEN_MUTATION, API_KEY_METADATA_FETCH, API_KEY_SUBMIT_CHANGE,
219    KrabkaMetadataFetchRequest, KrabkaMetadataFetchResponse, KrabkaSubmitChangeRequest,
220    KrabkaSubmitChangeResponse, PRIVATE_BASELINE_VERSION, PRIVATE_CLUSTER_AUTHORIZATION_FAILED,
221    PRIVATE_UNSUPPORTED_VERSION, SUBMIT_CHANGE_UNCOMMITTED_TAIL, private_api_highest_version,
222    private_request_version,
223};
krabka_raft/lib.rs

krabka_raft/
lib.rs
