commy_wire/
lib.rs

1//! The Commy wire contract: the binary range frame and the error-code vocabulary.
2//!
3//! The specification is in `spec/`, the language-neutral golden vectors are in
4//! `vectors/`, and every SDK runs the same vectors in its own CI. This crate is
5//! the Rust side of that contract and is tested against those exact files.
6
7use serde::{Deserialize, Serialize};
8
9/// First byte of a binary range frame.
10pub const KIND_RANGE: u8 = 1;
11
12/// Bytes before the payload: kind (1) + file_id (4) + seq (8) + offset (8) + len (4).
13pub const HEADER_LEN: usize = 25;
14
15/// Largest payload one frame may carry. A bigger change is split into several frames.
16pub const MAX_PAYLOAD: usize = 1024 * 1024;
17
18/// Why a binary frame was refused.
19#[derive(Clone, Copy, Debug, PartialEq, Eq)]
20#[non_exhaustive]
21pub enum FrameError {
22    /// Fewer bytes than the header needs.
23    Truncated,
24    /// The first byte is not a known frame kind.
25    BadKind,
26    /// The `len` field disagrees with the number of payload bytes.
27    LengthMismatch,
28    /// A range of zero bytes carries nothing and is refused.
29    EmptyRange,
30    /// `offset + len` does not fit in a u64.
31    OffsetOverflow,
32    /// `len` exceeds [`MAX_PAYLOAD`].
33    TooLarge,
34}
35
36/// A run of bytes that changed in one service file. All integers are little-endian.
37#[derive(Clone, Debug, PartialEq, Eq)]
38pub struct RangeFrame {
39    pub file_id: u32,
40    pub seq: u64,
41    pub offset: u64,
42    pub data: Vec<u8>,
43}
44
45impl RangeFrame {
46    /// Split one changed run into frames of at most [`MAX_PAYLOAD`] bytes. All share `seq`.
47    pub fn split(file_id: u32, seq: u64, offset: u64, data: &[u8]) -> Vec<RangeFrame> {
48        data.chunks(MAX_PAYLOAD)
49            .enumerate()
50            .map(|(i, chunk)| RangeFrame {
51                file_id,
52                seq,
53                offset: offset + (i * MAX_PAYLOAD) as u64,
54                data: chunk.to_vec(),
55            })
56            .collect()
57    }
58
59    /// Encode to wire bytes. `data` must be 1 to [`MAX_PAYLOAD`] bytes; use [`RangeFrame::split`].
60    pub fn encode(&self) -> Vec<u8> {
61        let mut out = Vec::with_capacity(HEADER_LEN + self.data.len());
62        out.push(KIND_RANGE);
63        out.extend_from_slice(&self.file_id.to_le_bytes());
64        out.extend_from_slice(&self.seq.to_le_bytes());
65        out.extend_from_slice(&self.offset.to_le_bytes());
66        out.extend_from_slice(&(self.data.len() as u32).to_le_bytes());
67        out.extend_from_slice(&self.data);
68        out
69    }
70
71    pub fn decode(bytes: &[u8]) -> Result<RangeFrame, FrameError> {
72        if bytes.len() < HEADER_LEN {
73            return Err(FrameError::Truncated);
74        }
75        if bytes[0] != KIND_RANGE {
76            return Err(FrameError::BadKind);
77        }
78        let file_id = u32::from_le_bytes(bytes[1..5].try_into().unwrap());
79        let seq = u64::from_le_bytes(bytes[5..13].try_into().unwrap());
80        let offset = u64::from_le_bytes(bytes[13..21].try_into().unwrap());
81        let len = u32::from_le_bytes(bytes[21..25].try_into().unwrap());
82        if len as usize > MAX_PAYLOAD {
83            return Err(FrameError::TooLarge);
84        }
85        let data = &bytes[HEADER_LEN..];
86        if data.len() as u64 != len as u64 {
87            return Err(FrameError::LengthMismatch);
88        }
89        if len == 0 {
90            return Err(FrameError::EmptyRange);
91        }
92        if offset.checked_add(len as u64).is_none() {
93            return Err(FrameError::OffsetOverflow);
94        }
95        Ok(RangeFrame {
96            file_id,
97            seq,
98            offset,
99            data: data.to_vec(),
100        })
101    }
102}
103
104/// The only spellings an error code may have on the wire, from server and SDKs alike.
105#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
106#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
107#[non_exhaustive]
108pub enum ErrorCode {
109    InvalidRequest,
110    Unauthenticated,
111    PermissionDenied,
112    NotFound,
113    RateLimited,
114    UnsupportedVersion,
115    InternalError,
116}
117
118impl ErrorCode {
119    /// Every code, in the order `spec/errors.md` lists them.
120    pub const ALL: [ErrorCode; 7] = [
121        ErrorCode::InvalidRequest,
122        ErrorCode::Unauthenticated,
123        ErrorCode::PermissionDenied,
124        ErrorCode::NotFound,
125        ErrorCode::RateLimited,
126        ErrorCode::UnsupportedVersion,
127        ErrorCode::InternalError,
128    ];
129
130    pub fn as_str(&self) -> &'static str {
131        match self {
132            ErrorCode::InvalidRequest => "INVALID_REQUEST",
133            ErrorCode::Unauthenticated => "UNAUTHENTICATED",
134            ErrorCode::PermissionDenied => "PERMISSION_DENIED",
135            ErrorCode::NotFound => "NOT_FOUND",
136            ErrorCode::RateLimited => "RATE_LIMITED",
137            ErrorCode::UnsupportedVersion => "UNSUPPORTED_VERSION",
138            ErrorCode::InternalError => "INTERNAL_ERROR",
139        }
140    }
141}
142
143/// Client to server. Carries the protocol version.
144#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
145#[serde(deny_unknown_fields)]
146pub struct Authenticate {
147    pub request_id: String,
148    pub protocol_version: u32,
149    pub tenant_id: String,
150    pub client_id: String,
151    pub auth_method: String,
152    pub credentials: String,
153}
154
155/// Server to client: the flush identified by `request_id` is applied; `seq` is the new sequence.
156#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
157#[serde(deny_unknown_fields)]
158pub struct FlushAck {
159    pub request_id: String,
160    pub file_id: u32,
161    pub seq: u64,
162}
163
164/// Client to server: send the whole file again (no replay log in v1).
165#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
166#[serde(deny_unknown_fields)]
167pub struct Resync {
168    pub request_id: String,
169    pub file_id: u32,
170}
171
172/// Server to client: the last range frame of the snapshot has been sent.
173#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
174#[serde(deny_unknown_fields)]
175pub struct ResyncDone {
176    pub request_id: String,
177    pub file_id: u32,
178    pub seq: u64,
179    pub size: u64,
180}
181
182/// Server to client: a request failed.
183#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
184#[serde(deny_unknown_fields)]
185pub struct ErrorReply {
186    pub code: ErrorCode,
187    pub message: String,
188    #[serde(default, skip_serializing_if = "Option::is_none")]
189    pub request_id: Option<String>,
190}
191
192/// Client to server: may this service file be mapped locally?
193#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
194#[serde(deny_unknown_fields)]
195pub struct GetServiceFilePath {
196    pub request_id: String,
197    pub service_id: String,
198}
199
200/// Server to client: the `file_id` that range frames use for this service. `path` and `nonce_path`
201/// are both present (local mapping offered) or both absent (stay on WSS ranges). They are absent
202/// unless the service is open to local mapping and the caller holds a write grant on it.
203#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
204#[serde(deny_unknown_fields)]
205pub struct ServiceFilePath {
206    pub request_id: String,
207    pub service_id: String,
208    pub file_id: u32,
209    #[serde(default, skip_serializing_if = "Option::is_none")]
210    pub path: Option<String>,
211    #[serde(default, skip_serializing_if = "Option::is_none")]
212    pub nonce_path: Option<String>,
213}
214
215/// Client to server: the nonce read from `nonce_path`, proving the same filesystem view.
216#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
217#[serde(deny_unknown_fields)]
218pub struct ConfirmLocal {
219    pub request_id: String,
220    pub file_id: u32,
221    pub nonce: String,
222}
223
224impl std::fmt::Debug for ConfirmLocal {
225    // The nonce is a secret for the life of the connection; never print it.
226    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
227        f.debug_struct("ConfirmLocal")
228            .field("request_id", &self.request_id)
229            .field("file_id", &self.file_id)
230            .field("nonce", &"<redacted>")
231            .finish()
232    }
233}
234
235/// Server to client: the nonce matched; the server deletes the nonce file. A mismatch is an
236/// `Error` with `PERMISSION_DENIED`.
237#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
238#[serde(deny_unknown_fields)]
239pub struct ConfirmLocalAck {
240    pub request_id: String,
241    pub file_id: u32,
242}
243
244/// Why a service file is being replaced.
245#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
246#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
247#[non_exhaustive]
248pub enum RotateReason {
249    /// The file changes size. A drain barrier runs first; old-offset writes are never replayed.
250    Resize,
251    /// A client is removed. Same size and shape; old bytes are kept; no drain barrier.
252    Revocation,
253}
254
255/// Client to server: replace a service's file. `new_size` is required for `RESIZE` and forbidden
256/// for `REVOCATION`; `revoke_client_id` is the reverse. See [`RotateRequest::validate`].
257#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
258#[serde(deny_unknown_fields)]
259pub struct RotateRequest {
260    pub request_id: String,
261    pub service_id: String,
262    pub reason: RotateReason,
263    #[serde(default, skip_serializing_if = "Option::is_none")]
264    pub new_size: Option<u64>,
265    #[serde(default, skip_serializing_if = "Option::is_none")]
266    pub revoke_client_id: Option<String>,
267}
268
269impl RotateRequest {
270    /// Check the fields agree with `reason`. A request that fails is answered `INVALID_REQUEST`.
271    pub fn validate(&self) -> Result<(), &'static str> {
272        match self.reason {
273            RotateReason::Resize => match (self.new_size, &self.revoke_client_id) {
274                (Some(n), None) if n > 0 => Ok(()),
275                (Some(0), _) => Err("new_size must be at least 1"),
276                (None, _) => Err("RESIZE needs new_size"),
277                (_, Some(_)) => Err("RESIZE must not name a client"),
278                _ => Err("invalid RESIZE request"),
279            },
280            RotateReason::Revocation => match (self.new_size, &self.revoke_client_id) {
281                (None, Some(c)) if !c.is_empty() => Ok(()),
282                (_, None) => Err("REVOCATION needs revoke_client_id"),
283                (Some(_), _) => Err("REVOCATION must not set new_size"),
284                _ => Err("invalid REVOCATION request"),
285            },
286        }
287    }
288}
289
290/// Server to client: the request was accepted. The rotation ends with [`RotationCommit`].
291#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
292#[serde(deny_unknown_fields)]
293pub struct RotationAccepted {
294    pub request_id: String,
295    pub rotation_id: String,
296}
297
298/// Server to every SDK on the service (resize only): the drain barrier is open. Stop issuing write
299/// guards, wait for outstanding ones, flush into the old file, then send [`RotationDrained`] within
300/// `deadline_ms`. An SDK that does not is dropped with a `ROTATION_STRAGGLER` [`Loss`].
301#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
302#[serde(deny_unknown_fields)]
303pub struct RotationPending {
304    pub rotation_id: String,
305    pub service_id: String,
306    pub file_id: u32,
307    pub deadline_ms: u64,
308}
309
310/// Client to server: everything written so far is flushed; `seq` is the last sequence flushed.
311#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
312#[serde(deny_unknown_fields)]
313pub struct RotationDrained {
314    pub request_id: String,
315    pub rotation_id: String,
316    pub file_id: u32,
317    pub seq: u64,
318}
319
320/// Server to every SDK on the service: the new file is live. `new_file_id` differs from
321/// `old_file_id` and is the handle for the new mapping; fetch it with `GetServiceFilePath` (local)
322/// or `Resync` (remote). After a revocation, each remaining SDK writes its unflushed ranges into
323/// the new file; after a resize, old-offset writes are never replayed.
324#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
325#[serde(deny_unknown_fields)]
326pub struct RotationCommit {
327    pub rotation_id: String,
328    pub service_id: String,
329    pub reason: RotateReason,
330    pub old_file_id: u32,
331    pub new_file_id: u32,
332    pub generation: u64,
333    pub size: u64,
334}
335
336/// What a program lost.
337#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
338#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
339#[non_exhaustive]
340pub enum LossKind {
341    /// It did not drain in time (or kept writing to the old mapping); those writes are not replayed.
342    RotationStraggler,
343    /// A resync overwrote local changes; `ranges` names them.
344    ResyncOverwrote,
345    /// Its access was revoked; unflushed writes are dropped.
346    Revoked,
347    /// A kind this version does not know. Treat it as a loss of the whole file.
348    #[serde(other)]
349    Unknown,
350}
351
352/// A byte range inside a file.
353#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
354#[serde(deny_unknown_fields)]
355pub struct LossRange {
356    pub offset: u64,
357    pub len: u64,
358}
359
360/// Server to a client: writes of its own were lost, and why. `ranges` are in old-file coordinates;
361/// empty means the whole file. A receiver treats an unknown `kind` as a loss of the whole file.
362#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
363#[serde(deny_unknown_fields)]
364pub struct Loss {
365    pub service_id: String,
366    pub file_id: u32,
367    pub kind: LossKind,
368    #[serde(default, skip_serializing_if = "Vec::is_empty")]
369    pub ranges: Vec<LossRange>,
370}
371
372/// A run of bytes inside a file, named without its data.
373#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
374#[serde(deny_unknown_fields)]
375pub struct ByteRange {
376    pub offset: u64,
377    pub len: u64,
378}
379
380/// Most ranges one `RangesChanged` or `RangesUpdated` may carry. A flush with more is split into
381/// several messages, each getting its own `seq`.
382pub const MAX_RANGES_PER_MESSAGE: usize = 1024;
383
384/// Check a range list: 1 to [`MAX_RANGES_PER_MESSAGE`] ranges, none empty, none whose end overflows a
385/// u64, sorted by offset and not overlapping (adjacent is fine). A message that fails is answered
386/// `INVALID_REQUEST` (or dropped, when it came from the server).
387pub fn validate_ranges(ranges: &[ByteRange]) -> Result<(), &'static str> {
388    if ranges.is_empty() {
389        return Err("ranges must not be empty");
390    }
391    if ranges.len() > MAX_RANGES_PER_MESSAGE {
392        return Err("too many ranges in one message");
393    }
394    let mut prev_end = 0u64;
395    for r in ranges {
396        if r.len == 0 {
397            return Err("a range must cover at least one byte");
398        }
399        let Some(end) = r.offset.checked_add(r.len) else {
400            return Err("offset plus len overflows");
401        };
402        if r.offset < prev_end {
403            return Err("ranges must be sorted and must not overlap");
404        }
405        prev_end = end;
406    }
407    Ok(())
408}
409
410/// Client to server, from a **local** sender (it wrote into the shared mapping): these ranges
411/// changed. No data travels; the server reads the latest bytes from the file when it forwards.
412/// Answered with `FlushAck`.
413#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
414#[serde(deny_unknown_fields)]
415pub struct RangesChanged {
416    pub request_id: String,
417    pub file_id: u32,
418    pub ranges: Vec<ByteRange>,
419}
420
421/// Server to a **local** receiver: these ranges of the file changed, at `seq`. No data; the receiver
422/// already sees the bytes through its own mapping.
423#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
424#[serde(deny_unknown_fields)]
425pub struct RangesUpdated {
426    pub file_id: u32,
427    pub seq: u64,
428    pub ranges: Vec<ByteRange>,
429}
430
431/// Client to server, from a **remote** sender: the range frames for one flush are complete. The
432/// server assigns the flush one new `seq`, forwards it, and answers `FlushAck`.
433#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
434#[serde(deny_unknown_fields)]
435pub struct FlushEnd {
436    pub request_id: String,
437    pub file_id: u32,
438}
439
440/// JSON control messages, sent as WebSocket text frames. Adjacently tagged
441/// (`{"type": ..., "data": {...}}`); unknown fields and unknown types are refused.
442#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
443#[serde(tag = "type", content = "data")]
444#[non_exhaustive]
445pub enum ControlMessage {
446    Authenticate(Authenticate),
447    FlushAck(FlushAck),
448    Resync(Resync),
449    ResyncDone(ResyncDone),
450    Error(ErrorReply),
451    GetServiceFilePath(GetServiceFilePath),
452    ServiceFilePath(ServiceFilePath),
453    ConfirmLocal(ConfirmLocal),
454    ConfirmLocalAck(ConfirmLocalAck),
455    RotateRequest(RotateRequest),
456    RotationAccepted(RotationAccepted),
457    RotationPending(RotationPending),
458    RotationDrained(RotationDrained),
459    RotationCommit(RotationCommit),
460    Loss(Loss),
461    RangesChanged(RangesChanged),
462    RangesUpdated(RangesUpdated),
463    FlushEnd(FlushEnd),
464}
commy_wire/lib.rs

commy_wire/
lib.rs
