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

crafty_proto/
lib.rs
