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

crafty_proto/
lib.rs
