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/// JSON control messages, sent as WebSocket text frames. Adjacently tagged
373/// (`{"type": ..., "data": {...}}`); unknown fields and unknown types are refused.
374#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
375#[serde(tag = "type", content = "data")]
376#[non_exhaustive]
377pub enum ControlMessage {
378    Authenticate(Authenticate),
379    FlushAck(FlushAck),
380    Resync(Resync),
381    ResyncDone(ResyncDone),
382    Error(ErrorReply),
383    GetServiceFilePath(GetServiceFilePath),
384    ServiceFilePath(ServiceFilePath),
385    ConfirmLocal(ConfirmLocal),
386    ConfirmLocalAck(ConfirmLocalAck),
387    RotateRequest(RotateRequest),
388    RotationAccepted(RotationAccepted),
389    RotationPending(RotationPending),
390    RotationDrained(RotationDrained),
391    RotationCommit(RotationCommit),
392    Loss(Loss),
393}
commy_wire/lib.rs

commy_wire/
lib.rs
