crafty_proto/
lib.rs

1//! `crafty-proto` — wire types and [`postcard`] codec shared across all crafty crates.
2//!
3//! Defines the on-the-wire representation for Raft peer RPCs, the client API,
4//! cluster join handshakes, and actor messaging. All bodies are encoded with
5//! `postcard` (wire-transport, serialization). Nothing here performs I/O.
6
7use serde::de::DeserializeOwned;
8use serde::{Deserialize, Serialize};
9
10pub mod actor;
11pub mod catalog;
12pub mod client;
13pub mod group;
14pub mod group_migrate;
15pub mod join;
16pub mod leave;
17pub mod queue;
18pub mod queue_autoscale;
19pub mod raft;
20pub mod saga_journal;
21pub mod two_phase;
22pub mod two_phase_journal;
23
24pub use actor::{
25    ActorEnvelope, ActorId, ActorRef, ActorRegistration, ActorTypeId, DeliverAck, DirectoryUpdate,
26    MigrateReply, MigrateRequest, RegisterAck, ScaleReply, ScaleRequest, SpawnReply, SpawnRequest,
27    StopReply, StopRequest,
28};
29pub use catalog::{CatalogAddRequest, CatalogAddResponse, CatalogCommand, CatalogRejection};
30pub use client::{ClientRequest, ClientResponse};
31pub use group::GroupPeerEnvelope;
32pub use group_migrate::{
33    GroupMigrateReply, GroupMigrateRequest, GroupMigrationBundle, GroupMigrationHardState,
34    GroupMigrationSnapshot, GroupMigrationSnapshotMeta,
35};
36pub use join::{JoinRejection, JoinRequest, JoinResponse, PeerBook, PeerEntry};
37pub use leave::{LeaveRejection, LeaveRequest, LeaveResponse};
38pub use queue::{
39    QueueAckReply, QueueAckRequest, QueueEnqueueReply, QueueEnqueueRequest, QueueLeaseReply,
40    QueueLeaseRequest, QueueLeasedJobWire, QueueMetricsReply, QueueMetricsRequest, QueueNackReply,
41    QueueNackRequest, QueueReplicateOp, QueueReplicateReply, QueueReplicateRequest,
42};
43pub use queue_autoscale::{
44    AutoscalePolicyWire, MembershipAutoscalePolicyWire, QueueAutoscalePolicyCommand,
45};
46pub use raft::{
47    AppendEntries, AppendEntriesReply, EntryPayload, InstallSnapshot, InstallSnapshotReply,
48    LogEntry, Membership, RaftRpc, RaftRpcReply, RequestVote, RequestVoteReply,
49};
50pub use saga_journal::SagaJournalCommand;
51pub use two_phase::{TwoPhaseAbortCommand, TwoPhasePrepareCommand};
52pub use two_phase_journal::TwoPhaseJournalCommand;
53
54/// Wire/protocol version negotiated on join (join-version-skew: hard reject on mismatch).
55pub const PROTOCOL_VERSION: u32 = 1;
56
57/// Oldest wire protocol this release accepts during rolling upgrades (N/N−1).
58pub const MIN_COMPATIBLE_PROTOCOL_VERSION: u32 = 1;
59
60/// Whether `got` is in the supported compatibility band `[MIN..=PROTOCOL]`.
61#[must_use]
62pub fn protocol_version_compatible(got: u32) -> bool {
63    got >= MIN_COMPATIBLE_PROTOCOL_VERSION && got <= PROTOCOL_VERSION
64}
65
66/// Stable identifier for a cluster node.
67#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
68pub struct NodeId(pub u64);
69
70/// Raft term (monotonic election epoch).
71#[derive(
72    Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Default, Serialize, Deserialize,
73)]
74pub struct Term(pub u64);
75
76/// 1-based index into the Raft log.
77#[derive(
78    Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Default, Serialize, Deserialize,
79)]
80pub struct LogIndex(pub u64);
81
82/// A leader replication/heartbeat round, used to confirm leadership for
83/// linearizable `ReadIndex` reads (read-consistency). Monotonic per leader term.
84#[derive(
85    Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Default, Serialize, Deserialize,
86)]
87pub struct Round(pub u64);
88
89/// A position in the Raft log: the `(term, index)` pair that uniquely
90/// identifies an entry. Ordering is lexicographic on `(term, index)`, which is
91/// exactly Raft's log "up-to-date" comparison (§5.4.1), so `LogId` values can
92/// be compared directly instead of juggling two primitives.
93#[derive(
94    Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Default, Serialize, Deserialize,
95)]
96pub struct LogId {
97    /// Term of the entry at `index`.
98    pub term: Term,
99    /// Log index.
100    pub index: LogIndex,
101}
102
103impl Term {
104    /// Term zero (before any election).
105    pub const ZERO: Term = Term(0);
106
107    /// The next term.
108    #[must_use]
109    pub fn next(self) -> Term {
110        Term(self.0 + 1)
111    }
112}
113
114impl LogIndex {
115    /// Index zero (empty log sentinel).
116    pub const ZERO: LogIndex = LogIndex(0);
117
118    /// The next index.
119    #[must_use]
120    pub fn next(self) -> LogIndex {
121        LogIndex(self.0 + 1)
122    }
123}
124
125impl Round {
126    /// The zeroth round (no heartbeat sent yet).
127    pub const ZERO: Round = Round(0);
128
129    /// The next round.
130    #[must_use]
131    pub fn next(self) -> Round {
132        Round(self.0 + 1)
133    }
134}
135
136impl LogId {
137    /// The empty-log sentinel `(term 0, index 0)`.
138    pub const ZERO: LogId = LogId {
139        term: Term::ZERO,
140        index: LogIndex::ZERO,
141    };
142
143    /// Construct a [`LogId`] from a term and index.
144    #[must_use]
145    pub fn new(term: Term, index: LogIndex) -> Self {
146        Self { term, index }
147    }
148}
149
150/// The wire codec in effect for this build: `"postcard"` by default, or
151/// `"json"` when the dev-only `json-wire` feature is enabled (future-work-and-risks item 4).
152/// Surfaced so a node can log/advertise its wire format at startup.
153pub const WIRE_CODEC: &str = if cfg!(feature = "json-wire") {
154    "json"
155} else {
156    "postcard"
157};
158
159/// Errors from encoding/decoding wire bodies.
160#[derive(Debug, thiserror::Error)]
161pub enum CodecError {
162    /// Serialization failed.
163    #[error("wire encode failed: {0}")]
164    Encode(String),
165    /// Deserialization failed.
166    #[error("wire decode failed: {0}")]
167    Decode(String),
168}
169
170/// Encode a value to a wire byte vector.
171///
172/// Uses the compact `postcard` binary format (wire-transport, serialization) unless the dev-only
173/// `json-wire` feature is enabled, in which case bodies are human-readable JSON.
174///
175/// # Errors
176/// Returns [`CodecError::Encode`] if serialization fails.
177#[cfg(not(feature = "json-wire"))]
178pub fn encode<T: Serialize>(value: &T) -> Result<Vec<u8>, CodecError> {
179    postcard::to_stdvec(value).map_err(|e| CodecError::Encode(e.to_string()))
180}
181
182/// Decode a value from a wire byte slice. See [`encode`] for the format.
183///
184/// # Errors
185/// Returns [`CodecError::Decode`] if deserialization fails.
186#[cfg(not(feature = "json-wire"))]
187pub fn decode<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, CodecError> {
188    postcard::from_bytes(bytes).map_err(|e| CodecError::Decode(e.to_string()))
189}
190
191/// Encode a value as JSON (dev-only `json-wire` build). See [`WIRE_CODEC`].
192///
193/// # Errors
194/// Returns [`CodecError::Encode`] if serialization fails.
195#[cfg(feature = "json-wire")]
196pub fn encode<T: Serialize>(value: &T) -> Result<Vec<u8>, CodecError> {
197    serde_json::to_vec(value).map_err(|e| CodecError::Encode(e.to_string()))
198}
199
200/// Decode a value from JSON (dev-only `json-wire` build). See [`WIRE_CODEC`].
201///
202/// # Errors
203/// Returns [`CodecError::Decode`] if deserialization fails.
204#[cfg(feature = "json-wire")]
205pub fn decode<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, CodecError> {
206    serde_json::from_slice(bytes).map_err(|e| CodecError::Decode(e.to_string()))
207}
208
209#[cfg(test)]
210mod tests {
211    use super::*;
212
213    #[test]
214    fn term_and_index_advance() {
215        assert_eq!(Term::ZERO.next(), Term(1));
216        assert_eq!(LogIndex::ZERO.next(), LogIndex(1));
217    }
218
219    #[test]
220    fn roundtrip_log_entry() {
221        let entry = LogEntry {
222            term: Term(3),
223            index: LogIndex(7),
224            payload: EntryPayload::Command(vec![1, 2, 3]),
225        };
226        let bytes = encode(&entry).expect("encode");
227        let back: LogEntry = decode(&bytes).expect("decode");
228        assert_eq!(entry, back);
229    }
230
231    #[test]
232    fn roundtrip_saga_journal_entry() {
233        let entry = LogEntry {
234            term: Term(2),
235            index: LogIndex(4),
236            payload: EntryPayload::SagaJournal(SagaJournalCommand {
237                saga_id: b"saga-1".to_vec(),
238                record: vec![1, 2, 3],
239            }),
240        };
241        let bytes = encode(&entry).expect("encode");
242        let back: LogEntry = decode(&bytes).expect("decode");
243        assert_eq!(entry, back);
244    }
245
246    #[test]
247    fn roundtrip_two_phase_prepare_entry() {
248        let entry = LogEntry {
249            term: Term(2),
250            index: LogIndex(5),
251            payload: EntryPayload::TwoPhasePrepare(TwoPhasePrepareCommand {
252                tx_id: b"tx".to_vec(),
253                route_key: b"key".to_vec(),
254                command: vec![1, 2],
255                prepared_at_ms: 0,
256            }),
257        };
258        let bytes = encode(&entry).expect("encode");
259        let back: LogEntry = decode(&bytes).expect("decode");
260        assert_eq!(entry, back);
261    }
262
263    #[test]
264    fn roundtrip_raft_rpc() {
265        let rpc = RaftRpc::AppendEntries(AppendEntries {
266            term: Term(2),
267            leader_id: NodeId(1),
268            prev_log: LogId::new(Term(1), LogIndex(4)),
269            entries: vec![LogEntry {
270                term: Term(2),
271                index: LogIndex(5),
272                payload: EntryPayload::Noop,
273            }],
274            leader_commit: LogIndex(4),
275            round: Round(7),
276        });
277        let bytes = encode(&rpc).expect("encode");
278        let back: RaftRpc = decode(&bytes).expect("decode");
279        assert_eq!(rpc, back);
280    }
281
282    #[test]
283    fn log_id_orders_by_term_then_index() {
284        // Up-to-dateness: higher term wins regardless of index; same term
285        // compares by index (Raft §5.4.1).
286        assert!(LogId::new(Term(2), LogIndex(1)) > LogId::new(Term(1), LogIndex(9)));
287        assert!(LogId::new(Term(1), LogIndex(5)) > LogId::new(Term(1), LogIndex(4)));
288        assert_eq!(
289            LogEntry {
290                term: Term(3),
291                index: LogIndex(7),
292                payload: EntryPayload::Noop
293            }
294            .id(),
295            LogId::new(Term(3), LogIndex(7))
296        );
297    }
298
299    #[test]
300    fn decode_rejects_garbage() {
301        let err = decode::<LogEntry>(&[0xff, 0xff, 0xff, 0xff]);
302        assert!(err.is_err());
303    }
304
305    #[test]
306    fn protocol_version_compatible_accepts_current_and_min() {
307        assert!(protocol_version_compatible(PROTOCOL_VERSION));
308        assert!(protocol_version_compatible(MIN_COMPATIBLE_PROTOCOL_VERSION));
309        assert!(!protocol_version_compatible(0));
310        assert!(!protocol_version_compatible(PROTOCOL_VERSION + 1));
311    }
312
313    #[test]
314    fn wire_codec_matches_build_feature() {
315        // The default build is postcard; `--features json-wire` flips it.
316        if cfg!(feature = "json-wire") {
317            assert_eq!(WIRE_CODEC, "json");
318            // JSON is human-readable: the encoded RPC contains field names.
319            let bytes = encode(&RequestVote {
320                term: Term(1),
321                candidate_id: NodeId(2),
322                last_log: LogId::ZERO,
323                pre_vote: true,
324            })
325            .unwrap();
326            let text = String::from_utf8(bytes).unwrap();
327            assert!(text.contains("candidate_id"), "json body: {text}");
328        } else {
329            assert_eq!(WIRE_CODEC, "postcard");
330        }
331    }
332}
crafty_proto/lib.rs

crafty_proto/
lib.rs
