krabka_raft/
types.rs

1//! Shared controller types: the `NodeId` alias, the KIP-853 voter `Node`
2//! identity, and the `AppData`/`AppDataResponse` records carried through the
3//! controller. These are the plain Krabka types the engine and reconfig
4//! coordinator use.
5
6pub use krabka_ids::NodeId;
7use {
8    krabka_metadata::{DelegationTokenRecord, MetadataRecord, VoterEndpoint},
9    serde::{Deserialize, Serialize},
10};
11
12/// KIP-853 voter node identity used by controller membership.
13#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
14pub struct Node {
15    pub directory_id: uuid::Uuid,
16    pub endpoints: Vec<krabka_metadata::VoterEndpoint>,
17    pub kraft_version: krabka_metadata::KRaftVersionRange,
18}
19
20impl Node {
21    /// The controller RPC endpoint that peers dial, as a `"<host>:<port>"`
22    /// string.
23    ///
24    /// By convention this is the first endpoint named "CONTROLLER". If there is
25    /// no such endpoint, this method falls back to the first endpoint.
26    ///
27    /// This method returns the host VERBATIM, as a DNS name, and never
28    /// pre-resolves it to a `SocketAddr`. Voter endpoints carry per-pod
29    /// `StatefulSet` FQDNs, and the dialer re-resolves the host on each connect
30    /// through `TcpStream::connect`, so a peer that restarts on a new pod IP
31    /// stays reachable. A parse to a `SocketAddr` here gives `None` for any DNS
32    /// hostname, and that silently breaks `submit_change` leader-forwarding
33    /// through `ControllerHandle::voter_addr`. That function is now
34    /// `controller_endpoint_addr` in `controller.rs`, and it mirrors
35    /// `controller_addr` in `network.rs`.
36    #[must_use]
37    pub fn controller_addr(&self) -> Option<String> { controller_endpoint_addr(&self.endpoints) }
38}
39
40pub fn controller_endpoint_addr(endpoints: &[VoterEndpoint]) -> Option<String> {
41    let endpoint = endpoints
42        .iter()
43        .find(|e| e.name == "CONTROLLER")
44        .or_else(|| endpoints.first())?;
45    Some(format!("{}:{}", endpoint.host, endpoint.port))
46}
47
48/// What the controller asks Raft to replicate.
49///
50/// This is a batch of `MetadataRecord`s, so `submit_change` can group related
51/// records, such as one Topic and N Partitions, in a single committed entry.
52#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
53pub struct AppData {
54    pub records: Vec<MetadataRecord>,
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
58pub struct AppDataResponse {
59    /// The state machine fills this in on apply. It carries the new log index,
60    /// so callers can correlate.
61    pub applied_index: u64,
62    /// Records that failed `MetadataImage::validate` at apply time and were
63    /// skipped. This field carries the validation error message in order of
64    /// rejection. `submit_change` translates a non-empty list into
65    /// `RaftError::Metadata`, so a concurrent `CreateTopics` race ends with one
66    /// winner and one `TopicExists` for each loser, and not with a silent
67    /// commit of every duplicate.
68    pub rejected: Vec<String>,
69}
70
71#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
72pub struct SubmitChangeResult {
73    pub offset_reservations: Vec<OffsetReservation>,
74}
75
76/// A token mutation bound to the exact committed record observed by its caller.
77///
78/// The controller compares `expected` and applies the change in one engine
79/// turn, so a snapshot/write race cannot replace a newer generation.
80#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
81pub enum DelegationTokenMutation {
82    Renew {
83        expected: DelegationTokenRecord,
84        replacement: DelegationTokenRecord,
85    },
86    Expire {
87        expected: DelegationTokenRecord,
88        replacement: DelegationTokenRecord,
89    },
90    Delete {
91        expected: DelegationTokenRecord,
92    },
93}
94
95#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
96pub struct OffsetReservation {
97    pub topic: String,
98    pub partition: i32,
99    pub base_offset: i64,
100    pub count: i64,
101    /// `KRaft` leader epoch that appended and committed this reservation.
102    pub leader_epoch: u64,
103}
104
105#[cfg(test)]
106mod node_tests {
107
108    use super::*;
109    #[test]
110    fn node_controller_addr_prefers_controller_listener() {
111        let n = Node {
112            directory_id: uuid::Uuid::nil(),
113            endpoints: vec![
114                krabka_metadata::VoterEndpoint {
115                    name: "PLAINTEXT".into(),
116                    host: "127.0.0.1".into(),
117                    port: 9092,
118                },
119                krabka_metadata::VoterEndpoint {
120                    name: "CONTROLLER".into(),
121                    host: "127.0.0.1".into(),
122                    port: 9093,
123                },
124            ],
125            kraft_version: krabka_metadata::KRaftVersionRange::default(),
126        };
127        assert2::assert!(n.controller_addr() == Some("127.0.0.1:9093".to_owned()));
128    }
129
130    #[test]
131    fn node_controller_addr_keeps_dns_hostname_not_parsed_socketaddr() {
132        // A per-pod FQDN must come back verbatim as "<host>:<port>", NOT parsed
133        // to a SocketAddr (which is None for a hostname). The dialer re-resolves
134        // it per connect, so a peer on a new pod IP stays reachable.
135        let host = "demo-broker-0-0.demo-broker-headless.default.svc.cluster.local";
136        let n = Node {
137            directory_id: uuid::Uuid::nil(),
138            endpoints: vec![krabka_metadata::VoterEndpoint {
139                name: "CONTROLLER".into(),
140                host: host.into(),
141                port: 9093,
142            }],
143            kraft_version: krabka_metadata::KRaftVersionRange::default(),
144        };
145        assert2::assert!(n.controller_addr() == Some(format!("{host}:9093")));
146    }
147}
krabka_raft/types.rs

krabka_raft/
types.rs
