krabka_client_coordination/
lib.rs

1//! Leader election, leases, and fencing tokens for Apache Kafka clients.
2//!
3//! This crate gives an active or standby process a way to elect one leader per
4//! role and to prove that leadership on every guarded write. It builds the
5//! primitives on Kafka itself, so a cluster needs no second consensus service.
6//! Per-role state lives in the compacted internal topic
7//! [`COORDINATION_STATE_TOPIC`].
8//!
9//! The leadership epoch is the producer epoch. Kafka's transaction coordinator
10//! mints that epoch when a member calls `InitProducerId` for the
11//! `transactional.id` of the role. The quorum picks the value, so the epoch is
12//! monotonic, and every broker enforces it. A write that carries an older epoch
13//! fails with a fenced error. The lease adds no safety of its own. It is an
14//! anti-flap device: a standby waits for the deadline of the holder to pass
15//! before it challenges, so a short pause does not move the role.
16//!
17//! # Key Types
18//!
19//! - [`Role`] and [`MemberId`] — the validated names of a role and of one
20//!   member that competes for it.
21//! - [`FencingToken`] — the quorum-minted proof that a member holds a role.
22//! - [`CoordinationKey`] and [`CoordinationRecord`] — the decoded key and value
23//!   of one record of [`COORDINATION_STATE_TOPIC`].
24//! - [`encode_key`], [`decode_key`], [`encode_value`], and [`decode_value`] —
25//!   the codec for the frozen record layouts.
26//!
27//! ## Quick start
28//!
29//! The record codec is the part of the crate that exists now. It encodes and
30//! decodes the frozen layouts that `krabka-streams-java` and
31//! `krabka-streams-go` also implement.
32//!
33//! ```
34//! use krabka_client_coordination::{
35//!     CoordinationKey, CoordinationRecord, FencingToken, Lease, MemberId, Role, decode_key,
36//!     decode_value, encode_key, encode_value,
37//! };
38//!
39//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
40//! let key = CoordinationKey::lease(Role::new("controller")?);
41//! let record = CoordinationRecord::Lease(Lease {
42//!     member: MemberId::new("node-1")?,
43//!     token: FencingToken::new(4242, 7)?,
44//!     granted_at: 1_700_000_000_000,
45//!     deadline: 1_700_000_030_000,
46//! });
47//!
48//! // Produce these two buffers to `__coordination_state`. A `None` value is a
49//! // tombstone, and it clears the lease.
50//! let key_bytes = encode_key(&key);
51//! let value_bytes = encode_value(&record);
52//!
53//! // Read them back on the consumer side.
54//! let decoded_key = decode_key(&key_bytes)?;
55//! let decoded = decode_value(decoded_key.kind, value_bytes.as_deref())?;
56//! let CoordinationRecord::Lease(lease) = decoded else {
57//!     unreachable!("the record is a lease")
58//! };
59//! println!(
60//!     "{} holds {} with token {} until {}",
61//!     lease.member, decoded_key.role, lease.token, lease.deadline
62//! );
63//! # Ok(())
64//! # }
65//! ```
66//!
67//! ## Scope and boundaries
68//!
69//! This crate owns the coordination record format and the client API built on
70//! it. It does not change the broker, and it needs no broker-side feature. A
71//! caller supplies its own producer and consumer, and this crate supplies the
72//! succession rules, the lease clock, and the codec below.
73#![doc(html_root_url = "https://docs.rs/krabka-client-coordination/0.7.1")]
74
75mod broker;
76mod error;
77mod lease;
78mod record;
79mod succession;
80mod transport;
81
82pub use broker::{
83    BrokerTransport, DEFAULT_COORDINATION_CLIENT_ID, DEFAULT_COORDINATION_FETCH_MAX_WAIT,
84    DEFAULT_COORDINATION_FETCH_PARTITION_MAX, DEFAULT_COORDINATION_PARTITIONS,
85    DEFAULT_COORDINATION_REPLICATION, DEFAULT_COORDINATION_REQUEST_TIMEOUT, role_partition,
86};
87pub use error::CoordinationError;
88pub use lease::{
89    Clock, DEFAULT_CHALLENGE_STAGGER, DEFAULT_LEASE_DURATION, DEFAULT_RENEW_INTERVAL, LeaseConfig,
90    LeaseTiming, SystemClock, TestClock,
91};
92pub use record::{
93    COORDINATION_STATE_TOPIC, CoordinationKey, CoordinationRecord, FencingToken, Lease,
94    MAX_MEMBER_ID_LEN, MAX_ROLE_LEN, MemberId, RecordError, RecordKind, Registration, Role,
95    decode_key, decode_value, encode_key, encode_value,
96};
97pub use succession::{Decision, RoleState, RoleStateBuilder, RosterEntry, evaluate};
98pub use transport::{CoordinationTransport, RoleRecords};
krabka_client_coordination/lib.rs

krabka_client_coordination/
lib.rs
