barnabas_core/lib.rs
1//! A Kafka client core with no IO, no runtime, and no clock.
2//!
3//! Everything here is a state machine over bytes. Nothing opens a socket, waits
4//! on a timer, or spawns a task — a caller feeds it received bytes and drains
5//! the bytes it wants sent. Binding crates (`barnabas-glommio`, and later
6//! `barnabas-tokio`) supply the sockets.
7//!
8//! # Why
9//!
10//! An abstraction spanning async runtimes is lowest-common-denominator, and LCD
11//! requires `Send` — which forbids exactly the per-core, `!Send` design a
12//! thread-per-core runtime exists for. A sans-io core sidesteps the argument by
13//! naming no runtime at all: the *binding* decides `Send`-ness, so one state
14//! machine serves both a per-core `Rc` handle and a work-stealing one.
15//!
16//! The second reason is testing, and in a Kafka client it is the bigger one.
17//! Exactly-once bugs are silent — a wrong retry duplicates records and every
18//! status code stays green — so they have to be caught by driving the state
19//! machine adversarially rather than by watching a broker behave. A core with
20//! no IO can be driven that way in a unit test, deterministically, with no
21//! broker and no executor. Every test in this crate is one.
22//!
23//! # Layout
24//!
25//! - [`frame`] — Kafka's length-prefixed framing, the one place partial reads
26//! are handled.
27//! - [`conn`] — request/response correlation over a single broker connection.
28//! - [`consumer`] — assign-only fetch positions and READ_COMMITTED filtering.
29//! - [`metadata`] — the cluster map, and knowing when it is stale.
30//! - [`partitioner`] — which partition a keyed record lands on, and why the
31//! answer differs between Kafka clients.
32//! - [`producer`] — idempotent sequencing and the transaction state machine.
33
34pub mod conn;
35pub mod consumer;
36pub mod frame;
37pub mod group;
38pub mod member;
39pub mod metadata;
40pub mod partitioner;
41pub mod producer;
42pub mod records;
43
44pub use conn::{Connection, PendingResponse};
45pub use consumer::{FetchPosition, IsolationLevel};
46pub use group::{
47 Assignment, Assignor, CooperativeStickyAssignor, RangeAssignor, RoundRobinAssignor,
48 StickyAssignor, Subscription, TopicPartition,
49};
50pub use member::{GroupMember, MemberState, RebalanceProtocol, Step};
51pub use metadata::{BrokerAddr, Metadata};
52pub use partitioner::Partitioner;
53pub use producer::{ProducerIdentity, ProducerState, SequenceRange, TxnState};
54
55/// Everything that can go wrong in the core.
56///
57/// Deliberately small and deliberately *not* an alias for the protocol crate's
58/// error: a caller distinguishes "the peer sent something impossible" (fatal,
59/// reconnect) from "the broker answered with an error code" (which is
60/// [`ErrorCode`]'s business and often retriable).
61#[derive(Debug, thiserror::Error)]
62pub enum Error {
63 #[error("protocol encode/decode: {0}")]
64 Codec(String),
65
66 /// A frame arrived that this connection did not ask for. Fatal: the stream
67 /// is no longer interpretable, so the connection must be dropped rather
68 /// than resynchronised.
69 #[error("unexpected correlation id {got}, expected {expected}")]
70 Correlation { got: i32, expected: i32 },
71
72 /// A response arrived with no request outstanding.
73 #[error("response with no request in flight")]
74 Unsolicited,
75
76 /// A frame longer than the caller's limit. Guards against a hostile or
77 /// corrupt peer steering us into an enormous allocation.
78 #[error("frame of {len} bytes exceeds the {limit} byte limit")]
79 FrameTooLarge { len: usize, limit: usize },
80}
81
82pub type Result<T> = std::result::Result<T, Error>;
83
84/// A broker error code, classified.
85///
86/// **This classification is the client.** P0 found that a cold cluster answers
87/// `FindCoordinator` with 15, then 14, then 16 — all of them states of a
88/// healthy cluster rather than failures — so a client that treats a non-zero
89/// code as an error cannot start a transaction at all. The taxonomy is
90/// therefore load-bearing from the first request, not late hardening.
91#[derive(Debug, Clone, Copy, PartialEq, Eq)]
92pub struct ErrorCode(pub i16);
93
94/// What a caller should *do* about an error code.
95#[derive(Debug, Clone, Copy, PartialEq, Eq)]
96pub enum Disposition {
97 /// No error.
98 Ok,
99 /// Retry the same request against the same broker after a backoff.
100 Retry,
101 /// Refresh metadata — the partition leader moved — then retry.
102 RefreshMetadata,
103 /// Re-run `FindCoordinator` before retrying.
104 ///
105 /// Distinct from [`Self::Retry`] because the coordinator genuinely moves:
106 /// P0 saw `NOT_COORDINATOR` *after* a successful discovery, and a client
107 /// that retries in place spins against a broker that will never answer.
108 FindCoordinator,
109 /// Unrecoverable for this producer or consumer. Fencing, authorization,
110 /// and the sequence errors that mean the stream is already wrong.
111 Fatal,
112}
113
114impl ErrorCode {
115 pub const NONE: Self = Self(0);
116 pub const OFFSET_OUT_OF_RANGE: Self = Self(1);
117 /// The topic or partition is not (yet) known to this broker. Transient
118 /// while a topic is being auto-created, and part of Kafka's
119 /// invalid-metadata family — so it refreshes rather than failing.
120 pub const UNKNOWN_TOPIC_OR_PARTITION: Self = Self(3);
121 pub const LEADER_NOT_AVAILABLE: Self = Self(5);
122 pub const NOT_LEADER_OR_FOLLOWER: Self = Self(6);
123 pub const REQUEST_TIMED_OUT: Self = Self(7);
124 pub const COORDINATOR_LOAD_IN_PROGRESS: Self = Self(14);
125 pub const COORDINATOR_NOT_AVAILABLE: Self = Self(15);
126 pub const NOT_COORDINATOR: Self = Self(16);
127 pub const OUT_OF_ORDER_SEQUENCE_NUMBER: Self = Self(45);
128 pub const DUPLICATE_SEQUENCE_NUMBER: Self = Self(46);
129 pub const INVALID_PRODUCER_EPOCH: Self = Self(47);
130 /// The previous transaction's markers are still being written. Starting a
131 /// second transaction immediately after ending the first hits this every
132 /// time — found while writing P1's broker tests.
133 pub const CONCURRENT_TRANSACTIONS: Self = Self(51);
134 pub const PRODUCER_FENCED: Self = Self(90);
135
136 #[must_use]
137 pub fn is_ok(self) -> bool {
138 self == Self::NONE
139 }
140
141 /// How to react.
142 ///
143 /// Unknown codes are [`Disposition::Fatal`] on purpose. Guessing "probably
144 /// retriable" for a code we have never seen is how a client retries an
145 /// operation that already partially succeeded — which, for a producer, is
146 /// how records get duplicated.
147 #[must_use]
148 pub fn disposition(self) -> Disposition {
149 match self {
150 Self::NONE => Disposition::Ok,
151 Self::LEADER_NOT_AVAILABLE
152 | Self::NOT_LEADER_OR_FOLLOWER
153 | Self::UNKNOWN_TOPIC_OR_PARTITION => Disposition::RefreshMetadata,
154 Self::COORDINATOR_LOAD_IN_PROGRESS | Self::COORDINATOR_NOT_AVAILABLE => {
155 Disposition::Retry
156 }
157 Self::NOT_COORDINATOR => Disposition::FindCoordinator,
158 Self::REQUEST_TIMED_OUT | Self::CONCURRENT_TRANSACTIONS => Disposition::Retry,
159 _ => Disposition::Fatal,
160 }
161 }
162}
163
164#[cfg(test)]
165mod tests {
166 use super::*;
167
168 /// The three codes P0 met on a cold cluster, and the distinction that
169 /// matters: 16 is not a plain retry.
170 #[test]
171 fn coordinator_warmup_codes_are_not_fatal() {
172 assert_eq!(
173 ErrorCode::COORDINATOR_NOT_AVAILABLE.disposition(),
174 Disposition::Retry
175 );
176 assert_eq!(
177 ErrorCode::COORDINATOR_LOAD_IN_PROGRESS.disposition(),
178 Disposition::Retry
179 );
180 assert_eq!(
181 ErrorCode::NOT_COORDINATOR.disposition(),
182 Disposition::FindCoordinator,
183 "NOT_COORDINATOR must re-discover: the coordinator moves, and \
184 retrying in place spins against a broker that will never answer"
185 );
186 }
187
188 /// A sequence error means the stream is already wrong. Retrying it is how
189 /// duplicates get written, so it must never be classified as retriable.
190 #[test]
191 fn sequence_and_fencing_errors_are_fatal() {
192 for code in [
193 ErrorCode::OUT_OF_ORDER_SEQUENCE_NUMBER,
194 ErrorCode::DUPLICATE_SEQUENCE_NUMBER,
195 ErrorCode::INVALID_PRODUCER_EPOCH,
196 ErrorCode::PRODUCER_FENCED,
197 ] {
198 assert_eq!(code.disposition(), Disposition::Fatal, "code {}", code.0);
199 }
200 }
201
202 /// Starting a transaction right after ending one is a normal thing to do,
203 /// and it must not be a failure.
204 #[test]
205 fn concurrent_transactions_is_retriable() {
206 assert_eq!(
207 ErrorCode::CONCURRENT_TRANSACTIONS.disposition(),
208 Disposition::Retry
209 );
210 }
211
212 /// A partition that has just been created reports this even after metadata
213 /// named a leader, so it must refresh rather than fail.
214 #[test]
215 fn not_leader_refreshes_metadata() {
216 assert_eq!(
217 ErrorCode::NOT_LEADER_OR_FOLLOWER.disposition(),
218 Disposition::RefreshMetadata
219 );
220 }
221
222 /// A topic being auto-created answers 3 for a moment. Treating it as fatal
223 /// makes the first write to a new topic fail, which is what happened when
224 /// the sink moved onto this client.
225 #[test]
226 fn an_unknown_topic_refreshes_metadata() {
227 assert_eq!(
228 ErrorCode::UNKNOWN_TOPIC_OR_PARTITION.disposition(),
229 Disposition::RefreshMetadata
230 );
231 }
232
233 /// An unknown code is fatal rather than optimistically retried.
234 #[test]
235 fn unknown_codes_are_fatal() {
236 assert_eq!(ErrorCode(9999).disposition(), Disposition::Fatal);
237 }
238}