//! The AVA1 core messages, structs and constants (SPEC.md §3).
//!
//! Generated by ava1-gen from `schema/ava1.toml`. Do not edit:
//! run `cargo run -p ava1-gen` after changing the schema.
#![allow(clippy::field_reassign_with_default, clippy::single_match, clippy::match_single_binding)]
use crate::wire::{DecodeError, EncodeError, FrameMessage, Message, Reader, SplitMix, Writer};
/// The protocol version this code speaks: the range in `HelloInfo` and the choice in `ServerInfo` (SPEC.md §5).
pub const PROTOCOL_VERSION: u16 = 1;
/// The TCP port a node listens on unless its application picks another (SPEC.md §1).
pub const DEFAULT_PORT: u16 = 9120;
/// Data lanes per session; lane ids are 1..=8, lane 0 is the control connection (SPEC.md §1, §9).
pub const MAX_LANES: u8 = 8;
/// RPC `node.info`: empty request, `NodeInfo` reply (SPEC.md §7.1).
pub const METHOD_NODE_INFO: u16 = 1;
/// RPC `pairing.open`: `PairingOpen` request, empty reply; opens the pairing window (SPEC.md §5, §7.1).
pub const METHOD_PAIRING_OPEN: u16 = 2;
/// RPC `crypto.bench`: `CryptoBench` request, `CryptoBenchResult` reply; a diagnostic (SPEC.md §7.1).
pub const METHOD_CRYPTO_BENCH: u16 = 3;
/// `RpcResponse.status` (and job status) for success (SPEC.md §7).
pub const STATUS_OK: u16 = 0;
/// An RPC or a lane join from a session whose pairing is not accepted (SPEC.md §5, §7.2).
pub const ERR_NOT_PAIRED: u16 = 1;
/// An unknown client while the pairing window is closed, or an unconfirmed session whose window or 60 s ended (SPEC.md §5, §7.2).
pub const ERR_PAIRING_CLOSED: u16 = 2;
/// The two peers' version ranges do not overlap; sent unsealed (SPEC.md §5, §7.2).
pub const ERR_UNSUPPORTED_VERSION: u16 = 3;
/// A frame or body that does not decode or is not allowed where it arrived, or a request the protocol forbids (SPEC.md §7.2).
pub const ERR_PROTOCOL: u16 = 4;
/// A `Join` for an unknown session, a lane id outside 1..=8, a wrong tag or a replayed nonce (SPEC.md §9, §7.2).
pub const ERR_BAD_JOIN: u16 = 5;
/// A method the node does not serve (SPEC.md §7.2, §7.3).
pub const ERR_UNKNOWN_METHOD: u16 = 6;
/// The node failed for a reason that is neither the peer's nor the disk's, such as out of memory (SPEC.md §7.2).
pub const ERR_INTERNAL: u16 = 7;
/// A limit was reached (connections, sessions, in-flight RPCs, jobs, a destination in use, buffer budget); retry later (SPEC.md §7.2, §8).
pub const ERR_BUSY: u16 = 8;
/// `ServerInfo.caps` bit: the node hosts the data plane, jobs and methods 16–19 (SPEC.md §5 step 2).
pub const CAP_DATA_PLANE: u64 = 1;
/// `ServerInfo.caps` bit: the node serves the methods 4 and up other than 16–19, among them `job.run`, `job.list` and the application's (SPEC.md §5 step 2, §7.3).
pub const CAP_MGMT: u64 = 2;
/// `ServerInfo.caps` bit: any paired device may read and cancel a transfer and take over a parked upload (SPEC.md §18.5).
pub const CAP_SHARED_JOBS: u64 = 4;
/// RPC `job.copy`: `JobCopy` request, `Status` reply; a copy or move on the node itself (SPEC.md §7.1, §15.5).
pub const METHOD_JOB_COPY: u16 = 16;
/// RPC `job.status`: `JobRef` request, `Status` reply (SPEC.md §7.1).
pub const METHOD_JOB_STATUS: u16 = 17;
/// RPC `job.cancel`: `JobRef` request, empty reply (SPEC.md §7.1, §11.7).
pub const METHOD_JOB_CANCEL: u16 = 18;
/// RPC `disk.calibrate`: `DiskCalibrate` request, `DiskCalibrateResult` reply (SPEC.md §16.10).
pub const METHOD_DISK_CALIBRATE: u16 = 19;
/// RPC `c2c.allow`, on the receiving node: `C2cAllow` request, `C2cTicket` reply; admits one job from another node (SPEC.md §18.1).
pub const METHOD_C2C_ALLOW: u16 = 22;
/// RPC `c2c.send`, on the sending node: `C2cSend` request, `Status` reply; sends to another node directly (SPEC.md §18.2).
pub const METHOD_C2C_SEND: u16 = 23;
/// A path that breaks SPEC.md §11.2, a root the node's policy refuses, or a source or staging parent that cannot be used (SPEC.md §7.2).
pub const ERR_PATH: u16 = 9;
/// The destination drive is full (SPEC.md §7.2).
pub const ERR_NO_SPACE: u16 = 10;
/// A job the node does not list, or lists for another peer key (SPEC.md §7.2).
pub const ERR_UNKNOWN_JOB: u16 = 11;
/// A disk or filesystem failure on the node's side (SPEC.md §7.2).
pub const ERR_IO: u16 = 12;
/// Bytes that do not match their root and that a `FileRetry` cannot fix (SPEC.md §7.2).
pub const ERR_VERIFY: u16 = 13;
/// A destination that already exists and may not be replaced (SPEC.md §7.2).
pub const ERR_EXISTS: u16 = 14;
/// The job was cancelled (SPEC.md §7.2).
pub const ERR_CANCELLED: u16 = 15;
/// A rename whose two sides are on different devices; never attempted (SPEC.md §7.2).
pub const ERR_CROSS_DEVICE: u16 = 16;
/// A lane frame larger than the credit the receiver granted (SPEC.md §12.4).
pub const ERR_CREDIT: u16 = 17;
/// The receiver ended a job whose sender kept the link alive but sent no file data (SPEC.md §12.8).
pub const ERR_STALLED: u16 = 18;
/// Client side: the node refused the typed pairing code, or could not prove it knows it (SPEC.md §5.5, §7.2).
pub const ERR_PAIRING_CODE: u16 = 19;
/// Job kind: the node that receives the `JobOpen` receives the files (SPEC.md §11.3).
pub const JOB_UPLOAD: u8 = 1;
/// Job kind: the node that receives the `JobOpen` sends the files (SPEC.md §11.3).
pub const JOB_DOWNLOAD: u8 = 2;
/// Job kind: the node copies `src` to the root locally (`job.copy`, SPEC.md §11.3, §15.5).
pub const JOB_COPY: u8 = 3;
/// Job policy: send everything the receiver's map does not list (SPEC.md §11.4).
pub const POLICY_REPLACE: u8 = 0;
/// Job policy: the receiver marks a file done when one of the same size and mtime seconds exists (SPEC.md §11.4).
pub const POLICY_SKIP_EXISTING: u8 = 1;
/// Job policy: the receiver marks a file done when an existing file of the same size hashes to the root in `ManifestEntry.root` (SPEC.md §11.4).
pub const POLICY_VERIFY: u8 = 2;
/// Job flag: the root is a file path and the manifest has one file entry (SPEC.md §11.8).
pub const JF_SINGLE_FILE: u32 = 1;
/// Job flag: the receiver consumes the files in manifest order, as a download into a zip (SPEC.md §11.8).
pub const JF_ORDERED: u32 = 2;
/// Job flag: the sender may read outside the roots its read policy allows, such as system files (SPEC.md §11.8).
pub const JF_UNSAFE_READ: u32 = 4;
/// Job flag, `job.copy` only: delete each source file once its copy is durable (SPEC.md §11.8).
pub const JF_MOVE: u32 = 8;
/// Job flag, `job.copy` only: replace destination files that already exist; unset, an existing destination root is refused with `ERR_EXISTS` (SPEC.md §11.8).
pub const JF_OVERWRITE: u32 = 16;
/// `ManifestEntry.kind`: a file (SPEC.md §11.3).
pub const ENTRY_FILE: u8 = 0;
/// `ManifestEntry.kind`: a directory (SPEC.md §11.3).
pub const ENTRY_DIR: u8 = 1;
/// log2 of the verification group size: files are hashed in groups of 1 MiB (SPEC.md §13.1).
pub const GROUP_SHIFT: u8 = 20;
/// Bytes: a file of at least this size is a large file and travels in `Chunk`s; smaller files
/// travel whole as `BundleRecord`s. A protocol constant both sides use (SPEC.md §12.2).
pub const LARGE_CUTOFF: u32 = 262144;
/// Bottleneck: none reported yet (SPEC.md §16).
pub const BN_NONE: u8 = 0;
/// Bottleneck: the network (SPEC.md §16).
pub const BN_NETWORK: u8 = 1;
/// Bottleneck: the sender's source reads (SPEC.md §16).
pub const BN_SOURCE: u8 = 2;
/// Bottleneck: the receiver's disk; its workers are at their maximum or adding one did not help (SPEC.md §16.9).
pub const BN_DISK: u8 = 3;
/// Bottleneck: the receiver's workers, while it is still adding them (SPEC.md §16.9).
pub const BN_WORKERS: u8 = 4;
/// Bottleneck: the sender waits for credit and the receiver reported no bottleneck of its own (SPEC.md §16).
pub const BN_CREDIT: u8 = 5;
/// `FileRetry.reason`: the file's root did not match (SPEC.md §11.9).
pub const RETRY_VERIFY: u16 = 1;
/// `FileRetry.reason`: the receiver lost the part file or outboard (SPEC.md §11.9).
pub const RETRY_IO: u16 = 2;
/// `FileRetry.reason`: the record's length disagrees with the manifest (SPEC.md §11.9).
pub const RETRY_CHANGED: u16 = 3;
/// RPC `job.run`: `JobRun` request, `Status` reply; starts an application operation as a job (SPEC.md §7.1.1).
pub const METHOD_JOB_RUN: u16 = 20;
/// RPC `job.list`: empty request, `JobListResult` reply (SPEC.md §7.1.1).
pub const METHOD_JOB_LIST: u16 = 21;
/// The largest `MgmtText` text in a reply, in bytes: the 256 KiB reply cap minus 16 (SPEC.md §7.3).
pub const RPC_TEXT_MAX: u32 = 262128;
/// Handshake step 1, client to node, unsealed (SPEC.md §5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Hs1 {
/// Noise XX message 1; its payload is a `HelloInfo`, in plaintext.
pub noise: Vec<u8>,
}
impl Message for Hs1 {
const NAME: &'static str = "Hs1";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.bytes(&self.noise)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.noise = r.bytes()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Hs1 {
const TYPE: u8 = 0x01;
}
/// Handshake step 2, node to client (SPEC.md §5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Hs2 {
/// Noise XX message 2; its payload is a `ServerInfo`, encrypted.
pub noise: Vec<u8>,
}
impl Message for Hs2 {
const NAME: &'static str = "Hs2";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.bytes(&self.noise)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.noise = r.bytes()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Hs2 {
const TYPE: u8 = 0x02;
}
/// Handshake step 3, client to node; every later frame on lane 0 is sealed (SPEC.md §5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Hs3 {
/// Noise XX message 3; its payload is a `ClientInfo`, encrypted.
pub noise: Vec<u8>,
}
impl Message for Hs3 {
const NAME: &'static str = "Hs3";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.bytes(&self.noise)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.noise = r.bytes()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Hs3 {
const TYPE: u8 = 0x03;
}
/// The node's first sealed frame, after `Hs3`: whether it knows the client's key (SPEC.md §5 step 4).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Welcome {
/// Nonzero when the node already trusts the client's key (or a ticket admits it, SPEC.md §18.2); 0 means the client must pair.
pub knows_you: u8,
/// The reveal of `ServerInfo.pair_commit`: BLAKE2b-256 of it must equal that commit (SPEC.md §4.6). Required, so it cannot be stripped.
pub nonce_s: [u8; 16],
/// Proof of the launch token, sent only to the key in this node's trust slot when the slot carries a token (SPEC.md §5.2). Older clients skip it.
pub launch_proof: Option<[u8; 16]>,
}
impl Message for Welcome {
const NAME: &'static str = "Welcome";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u8(self.knows_you);
w.fixed(&self.nonce_s);
let mut ext_n: u16 = 0;
if self.launch_proof.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.launch_proof { w.ext(1, |w| { w.fixed(v); Ok(()) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.knows_you = r.u8()?;
m.nonce_s = r.fixed::<16>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.launch_proof.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.launch_proof = Some(vr.fixed::<16>()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Welcome {
const TYPE: u8 = 0x04;
}
/// Pairing step 1, client to node: the client's CPace public value (SPEC.md §4.6, §5.5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PairPakeClient {
/// `X25519(x, G)` for a fresh scalar x, with G derived from the handshake hash and the code.
pub y: [u8; 32],
}
impl Message for PairPakeClient {
const NAME: &'static str = "PairPakeClient";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.y);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.y = r.fixed::<32>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for PairPakeClient {
const TYPE: u8 = 0x0d;
}
/// Pairing step 2, node to client: the node's CPace public value (SPEC.md §4.6, §5.5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PairPakeServer {
/// `X25519(x, G)` for the node's fresh scalar x.
pub y: [u8; 32],
}
impl Message for PairPakeServer {
const NAME: &'static str = "PairPakeServer";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.y);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.y = r.fixed::<32>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for PairPakeServer {
const TYPE: u8 = 0x0e;
}
/// Pairing step 3, client to node: the client's key confirmation (SPEC.md §4.6, §5.5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PairConfirm {
/// Keyed BLAKE2b-256(key = K, "client" || h). Required.
pub mac: [u8; 32],
}
impl Message for PairConfirm {
const NAME: &'static str = "PairConfirm";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.mac);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.mac = r.fixed::<32>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for PairConfirm {
const TYPE: u8 = 0x05;
}
/// Pairing step 4, node to client: the outcome; the session ends unless accepted (SPEC.md §4.6, §5.5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PairResult {
/// 1 when the node verified the client's `mac` and stored its key, else 0.
pub accepted: u8,
/// Keyed BLAKE2b-256(key = K, "server" || h) when accepted, else zero: the node's key confirmation, which the client checks before it stores the node.
pub mac: [u8; 32],
}
impl Message for PairResult {
const NAME: &'static str = "PairResult";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u8(self.accepted);
w.fixed(&self.mac);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.accepted = r.u8()?;
m.mac = r.fixed::<32>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for PairResult {
const TYPE: u8 = 0x06;
}
/// Opens a data lane: the first frame on a new connection, client to node, unsealed (SPEC.md §9).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Join {
/// The session's id, from `ServerInfo`.
pub session_id: [u8; 16],
/// The lane, 1..=8.
pub lane_id: u16,
/// Fresh random per join; feeds the lane keys (SPEC.md §4.3).
pub client_nonce: [u8; 16],
/// The Join proof (SPEC.md §4.5).
pub tag: [u8; 16],
}
impl Message for Join {
const NAME: &'static str = "Join";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.session_id);
w.u16(self.lane_id);
w.fixed(&self.client_nonce);
w.fixed(&self.tag);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.session_id = r.fixed::<16>()?;
m.lane_id = r.u16()?;
m.client_nonce = r.fixed::<16>()?;
m.tag = r.fixed::<16>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Join {
const TYPE: u8 = 0x07;
}
/// Accepts a `Join`, node to client, unsealed; both sides then seal under the lane's keys (SPEC.md §9).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JoinAck {
/// The lane, as in the `Join`.
pub lane_id: u16,
/// Fresh random per join; feeds the lane keys (SPEC.md §4.3).
pub server_nonce: [u8; 16],
/// The JoinAck proof (SPEC.md §4.5).
pub tag: [u8; 16],
}
impl Message for JoinAck {
const NAME: &'static str = "JoinAck";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u16(self.lane_id);
w.fixed(&self.server_nonce);
w.fixed(&self.tag);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.lane_id = r.u16()?;
m.server_nonce = r.fixed::<16>()?;
m.tag = r.fixed::<16>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for JoinAck {
const TYPE: u8 = 0x08;
}
/// Heartbeat, every 2 s by default on every connection, channel 0; answered by `Pong` (SPEC.md §6).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Ping {
/// The sender's sequence number, echoed in the `Pong`.
pub seq: u32,
/// The sender's monotonic clock in microseconds when the Ping was written; RTT = now - t_us.
pub t_us: u64,
}
impl Message for Ping {
const NAME: &'static str = "Ping";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u32(self.seq);
w.u64(self.t_us);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.seq = r.u32()?;
m.t_us = r.u64()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Ping {
const TYPE: u8 = 0x09;
}
/// Answer to a `Ping`, with its values (SPEC.md §6).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Pong {
/// The Ping's `seq`.
pub seq: u32,
/// The Ping's `t_us`.
pub t_us: u64,
}
impl Message for Pong {
const NAME: &'static str = "Pong";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u32(self.seq);
w.u64(self.t_us);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.seq = r.u32()?;
m.t_us = r.u64()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Pong {
const TYPE: u8 = 0x0a;
}
/// Reports why the connection ends; the sender then closes it (SPEC.md §6, §7.2).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Error {
/// An `ERR_*` code.
pub code: u16,
/// The cause, as text.
pub message: String,
}
impl Message for Error {
const NAME: &'static str = "Error";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u16(self.code);
w.str(&self.message)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.code = r.u16()?;
m.message = r.str()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Error {
const TYPE: u8 = 0x0b;
}
/// Ends the session (SPEC.md §6).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Bye {
/// Not interpreted by version 1; implementations send 0.
pub reason: u8,
}
impl Message for Bye {
const NAME: &'static str = "Bye";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u8(self.reason);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.reason = r.u8()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Bye {
const TYPE: u8 = 0x0c;
}
/// An RPC call on the control connection; the frame's channel is the request id (SPEC.md §7).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RpcRequest {
/// A `METHOD_*` number, or an application's (SPEC.md §7.1, §7.3).
pub method: u16,
/// The method's request body, at most 56 KiB (SPEC.md §7.4).
pub body: Vec<u8>,
}
impl Message for RpcRequest {
const NAME: &'static str = "RpcRequest";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u16(self.method);
w.bytes(&self.body)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.method = r.u16()?;
m.body = r.bytes()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for RpcRequest {
const TYPE: u8 = 0x60;
}
/// The answer to an `RpcRequest`, on the request's channel (SPEC.md §7).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RpcResponse {
/// `STATUS_OK` or an `ERR_*` code (SPEC.md §7.2).
pub status: u16,
/// On `STATUS_OK` the method's reply body; on an error the cause as UTF-8 text. At most 256 KiB (SPEC.md §7.4).
pub body: Vec<u8>,
}
impl Message for RpcResponse {
const NAME: &'static str = "RpcResponse";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u16(self.status);
w.bytes(&self.body)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.status = r.u16()?;
m.body = r.bytes()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for RpcResponse {
const TYPE: u8 = 0x61;
}
/// Opens a job, or reopens a known one to resume it; opener to peer, control connection (SPEC.md §11.3, §11.5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobOpen {
/// The job's id, chosen by the opener (SPEC.md §11.1).
pub job_id: [u8; 16],
/// A `JOB_*` kind, as seen by the node that receives this.
pub kind: u8,
/// A `POLICY_*`.
pub policy: u8,
/// `JF_*` bits (SPEC.md §11.8).
pub flags: u32,
/// Upload: the destination; download: the source.
pub root: String,
/// `JOB_COPY`: the source.
pub src: Option<String>,
/// `JOB_DOWNLOAD`: the credit the opener grants, in bytes (SPEC.md §12.4).
pub credit: Option<u64>,
}
impl Message for JobOpen {
const NAME: &'static str = "JobOpen";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u8(self.kind);
w.u8(self.policy);
w.u32(self.flags);
w.str(&self.root)?;
let mut ext_n: u16 = 0;
if self.src.is_some() { ext_n += 1; }
if self.credit.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.src { w.ext(1, |w| { w.str(v) })?; }
if let Some(v) = &self.credit { w.ext(2, |w| { w.u64(*v); Ok(()) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.kind = r.u8()?;
m.policy = r.u8()?;
m.flags = r.u32()?;
m.root = r.str()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.src.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.src = Some(vr.str()?);
vr.finish()?;
}
2 => {
if m.credit.is_some() { return Err(DecodeError::DupExt(2)); }
let mut vr = Reader::new(v);
m.credit = Some(vr.u64()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for JobOpen {
const TYPE: u8 = 0x20;
}
/// The answer to a `JobOpen` (SPEC.md §11.3).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobOpenAck {
/// The job.
pub job_id: [u8; 16],
/// `STATUS_OK` or an `ERR_*` refusal.
pub status: u16,
/// `JOB_UPLOAD`: lane-frame body bytes the sender may have outstanding (SPEC.md §12.4).
pub credit: u64,
/// 1 when the receiver writes under `<root>.ava-part/` and renames at the end (SPEC.md §11.6).
pub staged: u8,
/// The receiver's current number of write workers.
pub workers: u8,
/// The cause of a refusal, as text.
pub message: Option<String>,
}
impl Message for JobOpenAck {
const NAME: &'static str = "JobOpenAck";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u16(self.status);
w.u64(self.credit);
w.u8(self.staged);
w.u8(self.workers);
let mut ext_n: u16 = 0;
if self.message.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.message { w.ext(1, |w| { w.str(v) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.status = r.u16()?;
m.credit = r.u64()?;
m.staged = r.u8()?;
m.workers = r.u8()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.message.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.message = Some(vr.str()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for JobOpenAck {
const TYPE: u8 = 0x21;
}
/// One page of the manifest, sender to receiver; pages are sized to fit a control frame (SPEC.md §11.3, §11.7).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ManifestPage {
/// The job.
pub job_id: [u8; 16],
/// The next entries, in manifest order.
pub entries: Vec<ManifestEntry>,
}
impl Message for ManifestPage {
const NAME: &'static str = "ManifestPage";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.records(&self.entries)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.entries = r.records::<ManifestEntry>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for ManifestPage {
const TYPE: u8 = 0x22;
}
/// Ends the manifest, sender to receiver (SPEC.md §11.3).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ManifestEnd {
/// The job.
pub job_id: [u8; 16],
/// How many entries are files.
pub files: u32,
/// The files' total size in bytes.
pub bytes: u64,
/// BLAKE3 over every entry, without its ext (SPEC.md §11.3).
pub manifest_hash: [u8; 32],
}
impl Message for ManifestEnd {
const NAME: &'static str = "ManifestEnd";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u32(self.files);
w.u64(self.bytes);
w.fixed(&self.manifest_hash);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.files = r.u32()?;
m.bytes = r.u64()?;
m.manifest_hash = r.fixed::<32>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for ManifestEnd {
const TYPE: u8 = 0x23;
}
/// What the receiver already has, receiver to sender; everything else is needed. May span pages (SPEC.md §11.5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobMap {
/// The job.
pub job_id: [u8; 16],
/// `STATUS_OK`, or an `ERR_*` such as `ERR_UNKNOWN_JOB` for a `Resume` the receiver cannot serve.
pub status: u16,
/// 1 on the final page.
pub last: u8,
/// Files the receiver already has whole.
pub done: Vec<FileRun>,
/// Durable ranges of unfinished files.
pub partial: Vec<FileRange>,
/// The cause of an error status, as text.
pub message: Option<String>,
/// Final page: bytes the drive already holds for this job's unfinished large files (the allocated
/// blocks of their part files, at most each file's size). A sender that checks free space credits it;
/// absent means none is claimed (SPEC.md §11.5).
pub held: Option<u64>,
}
impl Message for JobMap {
const NAME: &'static str = "JobMap";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u16(self.status);
w.u8(self.last);
w.records(&self.done)?;
w.records(&self.partial)?;
let mut ext_n: u16 = 0;
if self.message.is_some() { ext_n += 1; }
if self.held.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.message { w.ext(1, |w| { w.str(v) })?; }
if let Some(v) = &self.held { w.ext(2, |w| { w.u64(*v); Ok(()) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.status = r.u16()?;
m.last = r.u8()?;
m.done = r.records::<FileRun>()?;
m.partial = r.records::<FileRange>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.message.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.message = Some(vr.str()?);
vr.finish()?;
}
2 => {
if m.held.is_some() { return Err(DecodeError::DupExt(2)); }
let mut vr = Reader::new(v);
m.held = Some(vr.u64()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for JobMap {
const TYPE: u8 = 0x24;
}
/// Fast-path resume when the sender still holds the same manifest; answered by `JobMap` (SPEC.md §11.5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Resume {
/// The job.
pub job_id: [u8; 16],
/// The hash of the manifest the sender holds.
pub manifest_hash: [u8; 32],
}
impl Message for Resume {
const NAME: &'static str = "Resume";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.fixed(&self.manifest_hash);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.manifest_hash = r.fixed::<32>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Resume {
const TYPE: u8 = 0x25;
}
/// Part of a large file, on a lane only (SPEC.md §12.2).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Chunk {
/// The job.
pub job_id: [u8; 16],
/// The file.
pub file_id: u32,
/// Where `data` starts in the file: a multiple of 1 MiB.
pub offset: u64,
/// Whole 1 MiB groups, except at the end of the file.
pub data: Vec<u8>,
}
impl Message for Chunk {
const NAME: &'static str = "Chunk";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u32(self.file_id);
w.u64(self.offset);
w.bytes(&self.data)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.file_id = r.u32()?;
m.offset = r.u64()?;
m.data = r.bytes()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Chunk {
const TYPE: u8 = 0x26;
}
/// Small files, whole, on a lane only (SPEC.md §12.2).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Bundle {
/// The job.
pub job_id: [u8; 16],
/// One record per file.
pub records: Vec<BundleRecord>,
}
impl Message for Bundle {
const NAME: &'static str = "Bundle";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.records(&self.records)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.records = r.records::<BundleRecord>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Bundle {
const TYPE: u8 = 0x27;
}
/// A lane frame is in the receiver's memory, before any disk work (SPEC.md §12.3).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Received {
/// The job.
pub job_id: [u8; 16],
/// The lane the frame arrived on.
pub lane: u16,
/// The frame's header channel: the sender's per-job sequence number (SPEC.md §12.1).
pub seq: u32,
}
impl Message for Received {
const NAME: &'static str = "Received";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u16(self.lane);
w.u32(self.seq);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.lane = r.u16()?;
m.seq = r.u32()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Received {
const TYPE: u8 = 0x28;
}
/// Buffer space returned to the sender; adds to its window (SPEC.md §12.4).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Credit {
/// The job.
pub job_id: [u8; 16],
/// Lane-frame body bytes freed.
pub bytes: u64,
}
impl Message for Credit {
const NAME: &'static str = "Credit";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u64(self.bytes);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.bytes = r.u64()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Credit {
const TYPE: u8 = 0x29;
}
/// Data synced and journaled, receiver to sender; never paged (SPEC.md §12.6).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Durable {
/// The job.
pub job_id: [u8; 16],
/// Files now complete and durable.
pub files: Vec<FileRun>,
/// Newly durable ranges of unfinished files.
pub ranges: Vec<FileRange>,
}
impl Message for Durable {
const NAME: &'static str = "Durable";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.records(&self.files)?;
w.records(&self.ranges)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.files = r.records::<FileRun>()?;
m.ranges = r.records::<FileRange>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Durable {
const TYPE: u8 = 0x2a;
}
/// A large file's BLAKE3 root, sender to receiver, once its last group is read (SPEC.md §13.2).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct FileRoot {
/// The job.
pub job_id: [u8; 16],
/// The file.
pub file_id: u32,
/// Its BLAKE3 root (SPEC.md §13.1).
pub root: [u8; 32],
}
impl Message for FileRoot {
const NAME: &'static str = "FileRoot";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u32(self.file_id);
w.fixed(&self.root);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.file_id = r.u32()?;
m.root = r.fixed::<32>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for FileRoot {
const TYPE: u8 = 0x2b;
}
/// The receiver dropped this file; send all of it again (SPEC.md §11.9, §13.3).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct FileRetry {
/// The job.
pub job_id: [u8; 16],
/// The file.
pub file_id: u32,
/// A `RETRY_*` code.
pub reason: u16,
}
impl Message for FileRetry {
const NAME: &'static str = "FileRetry";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u32(self.file_id);
w.u16(self.reason);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.file_id = r.u32()?;
m.reason = r.u16()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for FileRetry {
const TYPE: u8 = 0x2c;
}
/// A job's progress: sent by the receiver every 250 ms, IGNORABLE; also the reply body of `job.status` and the job-starting methods (SPEC.md §7.1, §16.9).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Status {
/// The job.
pub job_id: [u8; 16],
/// Files finished.
pub files_done: u32,
/// Files in the job; 0 while unknown.
pub files_total: u32,
/// Bytes the receiver has received.
pub bytes_received: u64,
/// Bytes durable on the receiver.
pub bytes_durable: u64,
/// Bytes in the job; 0 while unknown.
pub bytes_total: u64,
/// The receiver's own `BN_*` bottleneck.
pub bottleneck: u8,
/// The receiver's write workers.
pub workers: u8,
/// Data lanes in use.
pub lanes: u8,
/// Reserved for the sender's sequential choice (SPEC.md §16); version 1 receivers send 0.
pub sequential: u8,
/// The current step; on a failed job, the cause.
pub current: Option<String>,
/// RPC replies: 0 running, 1 finished OK, 2 failed (SPEC.md §7.1).
pub state: Option<u8>,
/// `job.run`: the finished operation's reply body, at most 128 KiB (SPEC.md §7.1.1).
pub result: Option<Vec<u8>>,
/// A failed job's `ERR_*` code.
pub code: Option<u16>,
/// Durable-by-log: done files not yet durable in place; absent means 0 (SPEC.md §15.7).
pub unswept: Option<u32>,
}
impl Message for Status {
const NAME: &'static str = "Status";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u32(self.files_done);
w.u32(self.files_total);
w.u64(self.bytes_received);
w.u64(self.bytes_durable);
w.u64(self.bytes_total);
w.u8(self.bottleneck);
w.u8(self.workers);
w.u8(self.lanes);
w.u8(self.sequential);
let mut ext_n: u16 = 0;
if self.current.is_some() { ext_n += 1; }
if self.state.is_some() { ext_n += 1; }
if self.result.is_some() { ext_n += 1; }
if self.code.is_some() { ext_n += 1; }
if self.unswept.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.current { w.ext(1, |w| { w.str(v) })?; }
if let Some(v) = &self.state { w.ext(2, |w| { w.u8(*v); Ok(()) })?; }
if let Some(v) = &self.result { w.ext(3, |w| { w.bytes(v) })?; }
if let Some(v) = &self.code { w.ext(4, |w| { w.u16(*v); Ok(()) })?; }
if let Some(v) = &self.unswept { w.ext(5, |w| { w.u32(*v); Ok(()) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.files_done = r.u32()?;
m.files_total = r.u32()?;
m.bytes_received = r.u64()?;
m.bytes_durable = r.u64()?;
m.bytes_total = r.u64()?;
m.bottleneck = r.u8()?;
m.workers = r.u8()?;
m.lanes = r.u8()?;
m.sequential = r.u8()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.current.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.current = Some(vr.str()?);
vr.finish()?;
}
2 => {
if m.state.is_some() { return Err(DecodeError::DupExt(2)); }
let mut vr = Reader::new(v);
m.state = Some(vr.u8()?);
vr.finish()?;
}
3 => {
if m.result.is_some() { return Err(DecodeError::DupExt(3)); }
let mut vr = Reader::new(v);
m.result = Some(vr.bytes()?);
vr.finish()?;
}
4 => {
if m.code.is_some() { return Err(DecodeError::DupExt(4)); }
let mut vr = Reader::new(v);
m.code = Some(vr.u16()?);
vr.finish()?;
}
5 => {
if m.unswept.is_some() { return Err(DecodeError::DupExt(5)); }
let mut vr = Reader::new(v);
m.unswept = Some(vr.u32()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for Status {
const TYPE: u8 = 0x2d;
}
/// The job ended, receiver to sender; sent after the last durable commit (SPEC.md §12.6).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobDone {
/// The job.
pub job_id: [u8; 16],
/// `STATUS_OK` or the `ERR_*` the job ended with.
pub status: u16,
/// Files done.
pub files: u32,
/// Bytes durable.
pub bytes: u64,
/// The cause of a failure, as text.
pub message: Option<String>,
/// 1: the log is durable but files are still being made durable in place (SPEC.md §15.7).
pub settling: Option<u8>,
}
impl Message for JobDone {
const NAME: &'static str = "JobDone";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u16(self.status);
w.u32(self.files);
w.u64(self.bytes);
let mut ext_n: u16 = 0;
if self.message.is_some() { ext_n += 1; }
if self.settling.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.message { w.ext(1, |w| { w.str(v) })?; }
if let Some(v) = &self.settling { w.ext(2, |w| { w.u8(*v); Ok(()) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.status = r.u16()?;
m.files = r.u32()?;
m.bytes = r.u64()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.message.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.message = Some(vr.str()?);
vr.finish()?;
}
2 => {
if m.settling.is_some() { return Err(DecodeError::DupExt(2)); }
let mut vr = Reader::new(v);
m.settling = Some(vr.u8()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for JobDone {
const TYPE: u8 = 0x2e;
}
/// Ends a job; either peer may send it, and the receiver keeps the journal (SPEC.md §11.7).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobCancel {
/// The job.
pub job_id: [u8; 16],
/// The `ERR_*` code the ender wants reported, such as `ERR_CANCELLED` for a user's cancel.
pub reason: u16,
}
impl Message for JobCancel {
const NAME: &'static str = "JobCancel";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u16(self.reason);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.reason = r.u16()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for JobCancel {
const TYPE: u8 = 0x2f;
}
/// Job-scoped liveness (SPEC.md §12.8): a sender reading past data the receiver already holds (a
/// solid archive decoded from its start on resume) sends this at least every 10 s; the receiver
/// counts it as progress for that job.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct SkipProgress {
/// The job.
pub job_id: [u8; 16],
/// Running total of bytes the sender has read past for this job; informational.
pub bytes_skipped: u64,
}
impl Message for SkipProgress {
const NAME: &'static str = "SkipProgress";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u64(self.bytes_skipped);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.bytes_skipped = r.u64()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
impl FrameMessage for SkipProgress {
const TYPE: u8 = 0x30;
}
/// The `node.info` reply (SPEC.md §7.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct NodeInfo {
/// The node software's version.
pub version: String,
/// The platform the node runs on.
pub platform: String,
/// The node's name.
pub name: String,
/// The platform's system software version.
pub firmware: Option<String>,
}
impl Message for NodeInfo {
const NAME: &'static str = "NodeInfo";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.str(&self.version)?;
w.str(&self.platform)?;
w.str(&self.name)?;
let mut ext_n: u16 = 0;
if self.firmware.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.firmware { w.ext(1, |w| { w.str(v) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.version = r.str()?;
m.platform = r.str()?;
m.name = r.str()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.firmware.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.firmware = Some(vr.str()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
/// The payload of `Hs1`: the protocol versions the client speaks (SPEC.md §5 step 1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct HelloInfo {
/// The lowest protocol version the client speaks.
pub version_min: u16,
/// The highest protocol version the client speaks.
pub version_max: u16,
/// Client capabilities; 0 in version 1.
pub caps: u64,
}
impl Message for HelloInfo {
const NAME: &'static str = "HelloInfo";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u16(self.version_min);
w.u16(self.version_max);
w.u64(self.caps);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.version_min = r.u16()?;
m.version_max = r.u16()?;
m.caps = r.u64()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The payload of `Hs2` (SPEC.md §5 step 2).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ServerInfo {
/// The protocol version the node chose from the client's range.
pub version: u16,
/// The node's `CAP_*` bits.
pub caps: u64,
/// Random id of this session; lanes name it in `Join`.
pub session_id: [u8; 16],
/// BLAKE2b-256(nonce_s), revealed in `Welcome` (SPEC.md §4.6). Kept for wire stability.
pub pair_commit: [u8; 32],
/// The node's name, for the client to show.
pub name: Option<String>,
}
impl Message for ServerInfo {
const NAME: &'static str = "ServerInfo";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u16(self.version);
w.u64(self.caps);
w.fixed(&self.session_id);
w.fixed(&self.pair_commit);
let mut ext_n: u16 = 0;
if self.name.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.name { w.ext(1, |w| { w.str(v) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.version = r.u16()?;
m.caps = r.u64()?;
m.session_id = r.fixed::<16>()?;
m.pair_commit = r.fixed::<32>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.name.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.name = Some(vr.str()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
/// The payload of `Hs3` (SPEC.md §5 step 3).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ClientInfo {
/// The client's pairing nonce (SPEC.md §4.6). Required; kept for wire stability.
pub nonce_c: [u8; 16],
/// The client's name, which the node stores with its key when they pair.
pub name: Option<String>,
/// A node sending to another node shows the ticket that node's `c2c.allow` handed out; it admits that one key for that one job (SPEC.md §18.2).
pub token: Option<[u8; 16]>,
}
impl Message for ClientInfo {
const NAME: &'static str = "ClientInfo";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.nonce_c);
let mut ext_n: u16 = 0;
if self.name.is_some() { ext_n += 1; }
if self.token.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.name { w.ext(1, |w| { w.str(v) })?; }
if let Some(v) = &self.token { w.ext(2, |w| { w.fixed(v); Ok(()) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.nonce_c = r.fixed::<16>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.name.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.name = Some(vr.str()?);
vr.finish()?;
}
2 => {
if m.token.is_some() { return Err(DecodeError::DupExt(2)); }
let mut vr = Reader::new(v);
m.token = Some(vr.fixed::<16>()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
/// The `pairing.open` request (SPEC.md §5 step 6, §7.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PairingOpen {
/// How long to keep the pairing window open, in seconds; at most 600.
pub seconds: u16,
}
impl Message for PairingOpen {
const NAME: &'static str = "PairingOpen";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u16(self.seconds);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.seconds = r.u16()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `crypto.bench` request (SPEC.md §7.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct CryptoBench {
/// How much data to seal and open, in MiB.
pub mib: u16,
}
impl Message for CryptoBench {
const NAME: &'static str = "CryptoBench";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u16(self.mib);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.mib = r.u16()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `crypto.bench` reply, a diagnostic (SPEC.md §7.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct CryptoBenchResult {
/// The bytes measured.
pub bytes: u64,
/// Microseconds to seal `bytes` as 1 MiB frames.
pub micros: u64,
/// Microseconds to open them.
pub open_micros: Option<u64>,
/// The ChaCha20 implementation in use: "avx2" or "portable".
pub backend: Option<String>,
}
impl Message for CryptoBenchResult {
const NAME: &'static str = "CryptoBenchResult";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u64(self.bytes);
w.u64(self.micros);
let mut ext_n: u16 = 0;
if self.open_micros.is_some() { ext_n += 1; }
if self.backend.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.open_micros { w.ext(1, |w| { w.u64(*v); Ok(()) })?; }
if let Some(v) = &self.backend { w.ext(2, |w| { w.str(v) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.bytes = r.u64()?;
m.micros = r.u64()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.open_micros.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.open_micros = Some(vr.u64()?);
vr.finish()?;
}
2 => {
if m.backend.is_some() { return Err(DecodeError::DupExt(2)); }
let mut vr = Reader::new(v);
m.backend = Some(vr.str()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
/// One manifest entry, a file or a directory (SPEC.md §11.2, §11.3).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ManifestEntry {
/// The entry's index in manifest order, from 0.
pub file_id: u32,
/// `ENTRY_FILE` or `ENTRY_DIR`.
pub kind: u8,
/// Unix permission bits.
pub mode: u32,
/// Size in bytes.
pub size: u64,
/// Last modification, in seconds since the Unix epoch; 0 when unknown, and always 0 for a directory (SPEC.md §11.4).
pub mtime: u64,
/// Relative to the job root, '/'-separated (SPEC.md §11.2).
pub path: String,
/// The file's BLAKE3 root, under `POLICY_VERIFY` only (SPEC.md §11.4).
pub root: Option<[u8; 32]>,
}
impl Message for ManifestEntry {
const NAME: &'static str = "ManifestEntry";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u32(self.file_id);
w.u8(self.kind);
w.u32(self.mode);
w.u64(self.size);
w.u64(self.mtime);
w.str(&self.path)?;
let mut ext_n: u16 = 0;
if self.root.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.root { w.ext(1, |w| { w.fixed(v); Ok(()) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.file_id = r.u32()?;
m.kind = r.u8()?;
m.mode = r.u32()?;
m.size = r.u64()?;
m.mtime = r.u64()?;
m.path = r.str()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.root.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.root = Some(vr.fixed::<32>()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
/// Files `first..first + count` (records item).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct FileRun {
/// The first file id.
pub first: u32,
/// How many consecutive file ids.
pub count: u32,
}
impl Message for FileRun {
const NAME: &'static str = "FileRun";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u32(self.first);
w.u32(self.count);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.first = r.u32()?;
m.count = r.u32()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// Bytes `offset..offset + len` of one file (records item).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct FileRange {
/// The file.
pub file_id: u32,
/// The first byte.
pub offset: u64,
/// The length in bytes.
pub len: u64,
}
impl Message for FileRange {
const NAME: &'static str = "FileRange";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u32(self.file_id);
w.u64(self.offset);
w.u64(self.len);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.file_id = r.u32()?;
m.offset = r.u64()?;
m.len = r.u64()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// One whole small file, inside a `Bundle` (SPEC.md §12.2, §15.2).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct BundleRecord {
/// The file.
pub file_id: u32,
/// BLAKE3 of `data` (SPEC.md §13.2).
pub root: [u8; 32],
/// The file's whole content.
pub data: Vec<u8>,
}
impl Message for BundleRecord {
const NAME: &'static str = "BundleRecord";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u32(self.file_id);
w.fixed(&self.root);
w.bytes(&self.data)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.file_id = r.u32()?;
m.root = r.fixed::<32>()?;
m.data = r.bytes()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// A file's BLAKE3 root, as the journal records it (records item, SPEC.md §14.2).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RootItem {
/// The file.
pub file_id: u32,
/// Its BLAKE3 root (SPEC.md §13.1).
pub root: [u8; 32],
}
impl Message for RootItem {
const NAME: &'static str = "RootItem";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u32(self.file_id);
w.fixed(&self.root);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.file_id = r.u32()?;
m.root = r.fixed::<32>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `job.copy` request: copy or move on the node itself (SPEC.md §15.5).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobCopy {
/// The job's id, chosen by the caller; a repeat with the same parameters answers the existing job.
pub job_id: [u8; 16],
/// The source path.
pub src: String,
/// The destination root.
pub dest: String,
/// `JF_*` bits, such as `JF_MOVE` and `JF_OVERWRITE` (SPEC.md §11.8).
pub flags: u32,
}
impl Message for JobCopy {
const NAME: &'static str = "JobCopy";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.str(&self.src)?;
w.str(&self.dest)?;
w.u32(self.flags);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.src = r.str()?;
m.dest = r.str()?;
m.flags = r.u32()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `job.status` and `job.cancel` request (SPEC.md §7.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobRef {
/// The job.
pub job_id: [u8; 16],
}
impl Message for JobRef {
const NAME: &'static str = "JobRef";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `disk.calibrate` request (SPEC.md §16.10).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct DiskCalibrate {
/// Where to write the test files; must pass the node's write policy.
pub dir: String,
/// Files to write per measurement, at most 20,000.
pub files: u32,
/// Bytes per file, at most 1 MiB.
pub size: u32,
}
impl Message for DiskCalibrate {
const NAME: &'static str = "DiskCalibrate";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.str(&self.dir)?;
w.u32(self.files);
w.u32(self.size);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.dir = r.str()?;
m.files = r.u32()?;
m.size = r.u32()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// One `disk.calibrate` measurement (SPEC.md §16.10).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct CalPoint {
/// The number of workers writing at once.
pub workers: u8,
/// Files written and synced per second.
pub files_per_s: u32,
/// Mean microseconds to create and write one file.
pub create_us: u32,
/// Mean microseconds to fsync one file.
pub fsync_us: u32,
}
impl Message for CalPoint {
const NAME: &'static str = "CalPoint";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u8(self.workers);
w.u32(self.files_per_s);
w.u32(self.create_us);
w.u32(self.fsync_us);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.workers = r.u8()?;
m.files_per_s = r.u32()?;
m.create_us = r.u32()?;
m.fsync_us = r.u32()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `disk.calibrate` reply: a hint for a client's starting worker count, not a contract (SPEC.md §16.10).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct DiskCalibrateResult {
/// One measurement each at 1, 2, 4, 8 and 16 workers.
pub points: Vec<CalPoint>,
}
impl Message for DiskCalibrateResult {
const NAME: &'static str = "DiskCalibrateResult";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.records(&self.points)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.points = r.records::<CalPoint>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `c2c.allow` request, on the receiving node (SPEC.md §18.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct C2cAllow {
/// The one job the ticket admits.
pub job_id: [u8; 16],
/// The sending node's identity (its static X25519 key).
pub key: [u8; 32],
/// Where the job may write.
pub root: String,
}
impl Message for C2cAllow {
const NAME: &'static str = "C2cAllow";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.fixed(&self.key);
w.str(&self.root)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.key = r.fixed::<32>()?;
m.root = r.str()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `c2c.allow` reply; the sending node shows the token in its `ClientInfo` (SPEC.md §18.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct C2cTicket {
/// 16 random bytes; valid for 10 minutes from its last use.
pub token: [u8; 16],
}
impl Message for C2cTicket {
const NAME: &'static str = "C2cTicket";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.token);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.token = r.fixed::<16>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `c2c.send` request, on the sending node (SPEC.md §18.2).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct C2cSend {
/// The job, as allowed on the receiving node.
pub job_id: [u8; 16],
/// The receiving node's IPv4 address.
pub host: String,
/// Its TCP port.
pub port: u16,
/// Its identity; the handshake must prove it.
pub key: [u8; 32],
/// The ticket from its `c2c.allow`.
pub token: [u8; 16],
/// The source path on the sending node.
pub src: String,
/// The destination root on the receiving node.
pub dest: String,
/// `JF_*` bits (SPEC.md §11.8).
pub flags: u32,
}
impl Message for C2cSend {
const NAME: &'static str = "C2cSend";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.str(&self.host)?;
w.u16(self.port);
w.fixed(&self.key);
w.fixed(&self.token);
w.str(&self.src)?;
w.str(&self.dest)?;
w.u32(self.flags);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.host = r.str()?;
m.port = r.u16()?;
m.key = r.fixed::<32>()?;
m.token = r.fixed::<16>()?;
m.src = r.str()?;
m.dest = r.str()?;
m.flags = r.u32()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// An application method's opaque text body, in practice UTF-8 JSON (SPEC.md §7.3).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct MgmtText {
/// The text, carried unchanged.
pub body: Vec<u8>,
/// 1: call again with the next offset (paged methods); absent or 0 when the reply is complete.
pub more: Option<u8>,
}
impl Message for MgmtText {
const NAME: &'static str = "MgmtText";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.bytes(&self.body)?;
let mut ext_n: u16 = 0;
if self.more.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.more { w.ext(1, |w| { w.u8(*v); Ok(()) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.body = r.bytes()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.more.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.more = Some(vr.u8()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
/// The `job.run` request (SPEC.md §7.1.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobRun {
/// The job's id, chosen by the caller.
pub job_id: [u8; 16],
/// The operation, a number the application registered.
pub op: u8,
/// The operation's request body, opaque to the protocol; at most 60 KiB.
pub args: Vec<u8>,
}
impl Message for JobRun {
const NAME: &'static str = "JobRun";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u8(self.op);
w.bytes(&self.args)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.op = r.u8()?;
m.args = r.bytes()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// One job in a `job.list` reply (SPEC.md §7.1.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobEntry {
/// The job.
pub job_id: [u8; 16],
/// 1 upload, 2 download, 3 copy, 4 operation.
pub kind: u8,
/// 0 running, 1 finished OK, 2 failed.
pub state: u8,
/// Files finished.
pub files_done: u32,
/// Files in the job; 0 while unknown.
pub files_total: u32,
/// Bytes done (durable, for a transfer).
pub bytes_done: u64,
/// Bytes in the job; 0 while unknown.
pub bytes_total: u64,
}
impl Message for JobEntry {
const NAME: &'static str = "JobEntry";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.u8(self.kind);
w.u8(self.state);
w.u32(self.files_done);
w.u32(self.files_total);
w.u64(self.bytes_done);
w.u64(self.bytes_total);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.kind = r.u8()?;
m.state = r.u8()?;
m.files_done = r.u32()?;
m.files_total = r.u32()?;
m.bytes_done = r.u64()?;
m.bytes_total = r.u64()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// The `job.list` reply: the caller's jobs of every kind (SPEC.md §7.1.1).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JobListResult {
/// One entry per job.
pub jobs: Vec<JobEntry>,
}
impl Message for JobListResult {
const NAME: &'static str = "JobListResult";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.records(&self.jobs)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.jobs = r.records::<JobEntry>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// Journal record kind 1: the job as opened (SPEC.md §14.1). Never on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JnlOpen {
/// The job.
pub job_id: [u8; 16],
/// The hash of the stored manifest.
pub manifest_hash: [u8; 32],
/// The job's `JOB_*` kind.
pub kind: u8,
/// The job's `JF_*` flags.
pub flags: u32,
/// Bit 0: the tree is written under `<root>.ava-part`; bit 1: the receiver took `<root>` with `mkdir` (SPEC.md §11.6).
pub staged: u8,
/// The destination root.
pub root: String,
}
impl Message for JnlOpen {
const NAME: &'static str = "JnlOpen";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.fixed(&self.job_id);
w.fixed(&self.manifest_hash);
w.u8(self.kind);
w.u32(self.flags);
w.u8(self.staged);
w.str(&self.root)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.job_id = r.fixed::<16>()?;
m.manifest_hash = r.fixed::<32>()?;
m.kind = r.u8()?;
m.flags = r.u32()?;
m.staged = r.u8()?;
m.root = r.str()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// Journal record kind 2: one sync batch (SPEC.md §14.2). Never on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JnlBatch {
/// Files now done; their ranges are dropped.
pub files: Vec<FileRun>,
/// Durable ranges added.
pub ranges: Vec<FileRange>,
/// Roots recorded.
pub roots: Vec<RootItem>,
/// Durable-by-log: the pack segment `pack.<n>` that holds this batch's `PackFile` records (SPEC.md §15.7).
pub pack_segment: Option<u32>,
/// Where those records start in the segment, in bytes.
pub pack_offset: Option<u64>,
/// Their length in bytes.
pub pack_len: Option<u64>,
}
impl Message for JnlBatch {
const NAME: &'static str = "JnlBatch";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.records(&self.files)?;
w.records(&self.ranges)?;
w.records(&self.roots)?;
let mut ext_n: u16 = 0;
if self.pack_segment.is_some() { ext_n += 1; }
if self.pack_offset.is_some() { ext_n += 1; }
if self.pack_len.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.pack_segment { w.ext(1, |w| { w.u32(*v); Ok(()) })?; }
if let Some(v) = &self.pack_offset { w.ext(2, |w| { w.u64(*v); Ok(()) })?; }
if let Some(v) = &self.pack_len { w.ext(3, |w| { w.u64(*v); Ok(()) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.files = r.records::<FileRun>()?;
m.ranges = r.records::<FileRange>()?;
m.roots = r.records::<RootItem>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.pack_segment.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.pack_segment = Some(vr.u32()?);
vr.finish()?;
}
2 => {
if m.pack_offset.is_some() { return Err(DecodeError::DupExt(2)); }
let mut vr = Reader::new(v);
m.pack_offset = Some(vr.u64()?);
vr.finish()?;
}
3 => {
if m.pack_len.is_some() { return Err(DecodeError::DupExt(3)); }
let mut vr = Reader::new(v);
m.pack_len = Some(vr.u64()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
/// Where a batch's `PackFile` records live; in journal snapshots only (SPEC.md §14.2, §15.7).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PackRef {
/// The pack segment `pack.<n>`.
pub segment: u32,
/// Where the records start, in bytes.
pub offset: u64,
/// Their length in bytes.
pub len: u64,
/// The first file id the records cover.
pub first_file: u32,
/// File ids covered: `first_file..first_file + count`.
pub count: u32,
}
impl Message for PackRef {
const NAME: &'static str = "PackRef";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u32(self.segment);
w.u64(self.offset);
w.u64(self.len);
w.u32(self.first_file);
w.u32(self.count);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.segment = r.u32()?;
m.offset = r.u64()?;
m.len = r.u64()?;
m.first_file = r.u32()?;
m.count = r.u32()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// Journal record kind 6: these files are now durable in place (SPEC.md §14.2, §15.7). Never on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JnlSweep {
/// The files swept.
pub files: Vec<FileRun>,
}
impl Message for JnlSweep {
const NAME: &'static str = "JnlSweep";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.records(&self.files)?;
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.files = r.records::<FileRun>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// Journal record kind 3: forget one file's progress (SPEC.md §14.2). Never on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JnlReset {
/// The file.
pub file_id: u32,
}
impl Message for JnlReset {
const NAME: &'static str = "JnlReset";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u32(self.file_id);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.file_id = r.u32()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// Journal record kind 4: the whole state, written when the journal is compacted (SPEC.md §14.1, §14.2). Never on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JnlSnapshot {
/// Files done.
pub done: Vec<FileRun>,
/// Durable ranges of unfinished files.
pub ranges: Vec<FileRange>,
/// Roots recorded.
pub roots: Vec<RootItem>,
/// The `FileRun` item stream of done files not yet swept (SPEC.md §14.2).
pub unswept: Option<Vec<u8>>,
/// The `PackRef` item stream of the pack ranges whose files are not all swept (SPEC.md §14.2).
pub segments: Option<Vec<u8>>,
}
impl Message for JnlSnapshot {
const NAME: &'static str = "JnlSnapshot";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.records(&self.done)?;
w.records(&self.ranges)?;
w.records(&self.roots)?;
let mut ext_n: u16 = 0;
if self.unswept.is_some() { ext_n += 1; }
if self.segments.is_some() { ext_n += 1; }
w.u16(ext_n);
if let Some(v) = &self.unswept { w.ext(1, |w| { w.bytes(v) })?; }
if let Some(v) = &self.segments { w.ext(2, |w| { w.bytes(v) })?; }
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.done = r.records::<FileRun>()?;
m.ranges = r.records::<FileRange>()?;
m.roots = r.records::<RootItem>()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
match tag {
1 => {
if m.unswept.is_some() { return Err(DecodeError::DupExt(1)); }
let mut vr = Reader::new(v);
m.unswept = Some(vr.bytes()?);
vr.finish()?;
}
2 => {
if m.segments.is_some() { return Err(DecodeError::DupExt(2)); }
let mut vr = Reader::new(v);
m.segments = Some(vr.bytes()?);
vr.finish()?;
}
_ => {}
}
}
r.finish()?;
Ok(m)
}
}
/// Journal record kind 5: the job's final status (SPEC.md §14.2). Never on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct JnlDone {
/// `STATUS_OK` or an `ERR_*` code.
pub status: u16,
}
impl Message for JnlDone {
const NAME: &'static str = "JnlDone";
fn encode_into(&self, w: &mut Writer) -> Result<(), EncodeError> {
w.u16(self.status);
w.u16(0);
Ok(())
}
fn decode(b: &[u8]) -> Result<Self, DecodeError> {
let mut r = Reader::new(b);
let mut m = Self::default();
m.status = r.u16()?;
let ext_n = r.u16()?;
for _ in 0..ext_n {
let tag = r.u16()?;
let len = r.u32()? as usize;
let v = r.take(len)?;
let _ = (tag, v);
}
r.finish()?;
Ok(m)
}
}
/// Every message and struct, by name (conformance tests).
pub const ALL: &[&str] = &["Hs1", "Hs2", "Hs3", "Welcome", "PairPakeClient", "PairPakeServer", "PairConfirm", "PairResult", "Join", "JoinAck", "Ping", "Pong", "Error", "Bye", "RpcRequest", "RpcResponse", "JobOpen", "JobOpenAck", "ManifestPage", "ManifestEnd", "JobMap", "Resume", "Chunk", "Bundle", "Received", "Credit", "Durable", "FileRoot", "FileRetry", "Status", "JobDone", "JobCancel", "SkipProgress", "NodeInfo", "HelloInfo", "ServerInfo", "ClientInfo", "PairingOpen", "CryptoBench", "CryptoBenchResult", "ManifestEntry", "FileRun", "FileRange", "BundleRecord", "RootItem", "JobCopy", "JobRef", "DiskCalibrate", "CalPoint", "DiskCalibrateResult", "C2cAllow", "C2cTicket", "C2cSend", "MgmtText", "JobRun", "JobEntry", "JobListResult", "JnlOpen", "JnlBatch", "PackRef", "JnlSweep", "JnlReset", "JnlSnapshot", "JnlDone", ];
#[doc(hidden)]
pub fn sample(name: &str, rng: &mut SplitMix) -> Option<Vec<u8>> {
match name {
"Hs1" => Hs1 { noise: { let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }, }.to_bytes().ok(),
"Hs2" => Hs2 { noise: { let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }, }.to_bytes().ok(),
"Hs3" => Hs3 { noise: { let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }, }.to_bytes().ok(),
"Welcome" => Welcome { knows_you: rng.next_u64() as u8, nonce_s: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, launch_proof: if rng.below(2) == 1 { Some({ let mut a = [0u8; 16]; rng.fill(&mut a); a }) } else { None }, }.to_bytes().ok(),
"PairPakeClient" => PairPakeClient { y: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"PairPakeServer" => PairPakeServer { y: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"PairConfirm" => PairConfirm { mac: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"PairResult" => PairResult { accepted: rng.next_u64() as u8, mac: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"Join" => Join { session_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, lane_id: rng.next_u64() as u16, client_nonce: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, tag: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"JoinAck" => JoinAck { lane_id: rng.next_u64() as u16, server_nonce: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, tag: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"Ping" => Ping { seq: rng.next_u64() as u32, t_us: rng.next_u64(), }.to_bytes().ok(),
"Pong" => Pong { seq: rng.next_u64() as u32, t_us: rng.next_u64(), }.to_bytes().ok(),
"Error" => Error { code: rng.next_u64() as u16, message: rng.ascii(20), }.to_bytes().ok(),
"Bye" => Bye { reason: rng.next_u64() as u8, }.to_bytes().ok(),
"RpcRequest" => RpcRequest { method: rng.next_u64() as u16, body: { let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }, }.to_bytes().ok(),
"RpcResponse" => RpcResponse { status: rng.next_u64() as u16, body: { let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }, }.to_bytes().ok(),
"JobOpen" => JobOpen { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, kind: rng.next_u64() as u8, policy: rng.next_u64() as u8, flags: rng.next_u64() as u32, root: rng.ascii(20), src: if rng.below(2) == 1 { Some(rng.ascii(20)) } else { None }, credit: if rng.below(2) == 1 { Some(rng.next_u64()) } else { None }, }.to_bytes().ok(),
"JobOpenAck" => JobOpenAck { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, status: rng.next_u64() as u16, credit: rng.next_u64(), staged: rng.next_u64() as u8, workers: rng.next_u64() as u8, message: if rng.below(2) == 1 { Some(rng.ascii(20)) } else { None }, }.to_bytes().ok(),
"ManifestPage" => ManifestPage { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, entries: vec![ManifestEntry::default(); rng.below(3) as usize], }.to_bytes().ok(),
"ManifestEnd" => ManifestEnd { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, files: rng.next_u64() as u32, bytes: rng.next_u64(), manifest_hash: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"JobMap" => JobMap { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, status: rng.next_u64() as u16, last: rng.next_u64() as u8, done: vec![FileRun::default(); rng.below(3) as usize], partial: vec![FileRange::default(); rng.below(3) as usize], message: if rng.below(2) == 1 { Some(rng.ascii(20)) } else { None }, held: if rng.below(2) == 1 { Some(rng.next_u64()) } else { None }, }.to_bytes().ok(),
"Resume" => Resume { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, manifest_hash: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"Chunk" => Chunk { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, file_id: rng.next_u64() as u32, offset: rng.next_u64(), data: { let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }, }.to_bytes().ok(),
"Bundle" => Bundle { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, records: vec![BundleRecord::default(); rng.below(3) as usize], }.to_bytes().ok(),
"Received" => Received { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, lane: rng.next_u64() as u16, seq: rng.next_u64() as u32, }.to_bytes().ok(),
"Credit" => Credit { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, bytes: rng.next_u64(), }.to_bytes().ok(),
"Durable" => Durable { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, files: vec![FileRun::default(); rng.below(3) as usize], ranges: vec![FileRange::default(); rng.below(3) as usize], }.to_bytes().ok(),
"FileRoot" => FileRoot { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, file_id: rng.next_u64() as u32, root: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"FileRetry" => FileRetry { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, file_id: rng.next_u64() as u32, reason: rng.next_u64() as u16, }.to_bytes().ok(),
"Status" => Status { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, files_done: rng.next_u64() as u32, files_total: rng.next_u64() as u32, bytes_received: rng.next_u64(), bytes_durable: rng.next_u64(), bytes_total: rng.next_u64(), bottleneck: rng.next_u64() as u8, workers: rng.next_u64() as u8, lanes: rng.next_u64() as u8, sequential: rng.next_u64() as u8, current: if rng.below(2) == 1 { Some(rng.ascii(20)) } else { None }, state: if rng.below(2) == 1 { Some(rng.next_u64() as u8) } else { None }, result: if rng.below(2) == 1 { Some({ let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }) } else { None }, code: if rng.below(2) == 1 { Some(rng.next_u64() as u16) } else { None }, unswept: if rng.below(2) == 1 { Some(rng.next_u64() as u32) } else { None }, }.to_bytes().ok(),
"JobDone" => JobDone { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, status: rng.next_u64() as u16, files: rng.next_u64() as u32, bytes: rng.next_u64(), message: if rng.below(2) == 1 { Some(rng.ascii(20)) } else { None }, settling: if rng.below(2) == 1 { Some(rng.next_u64() as u8) } else { None }, }.to_bytes().ok(),
"JobCancel" => JobCancel { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, reason: rng.next_u64() as u16, }.to_bytes().ok(),
"SkipProgress" => SkipProgress { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, bytes_skipped: rng.next_u64(), }.to_bytes().ok(),
"NodeInfo" => NodeInfo { version: rng.ascii(20), platform: rng.ascii(20), name: rng.ascii(20), firmware: if rng.below(2) == 1 { Some(rng.ascii(20)) } else { None }, }.to_bytes().ok(),
"HelloInfo" => HelloInfo { version_min: rng.next_u64() as u16, version_max: rng.next_u64() as u16, caps: rng.next_u64(), }.to_bytes().ok(),
"ServerInfo" => ServerInfo { version: rng.next_u64() as u16, caps: rng.next_u64(), session_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, pair_commit: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, name: if rng.below(2) == 1 { Some(rng.ascii(20)) } else { None }, }.to_bytes().ok(),
"ClientInfo" => ClientInfo { nonce_c: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, name: if rng.below(2) == 1 { Some(rng.ascii(20)) } else { None }, token: if rng.below(2) == 1 { Some({ let mut a = [0u8; 16]; rng.fill(&mut a); a }) } else { None }, }.to_bytes().ok(),
"PairingOpen" => PairingOpen { seconds: rng.next_u64() as u16, }.to_bytes().ok(),
"CryptoBench" => CryptoBench { mib: rng.next_u64() as u16, }.to_bytes().ok(),
"CryptoBenchResult" => CryptoBenchResult { bytes: rng.next_u64(), micros: rng.next_u64(), open_micros: if rng.below(2) == 1 { Some(rng.next_u64()) } else { None }, backend: if rng.below(2) == 1 { Some(rng.ascii(20)) } else { None }, }.to_bytes().ok(),
"ManifestEntry" => ManifestEntry { file_id: rng.next_u64() as u32, kind: rng.next_u64() as u8, mode: rng.next_u64() as u32, size: rng.next_u64(), mtime: rng.next_u64(), path: rng.ascii(20), root: if rng.below(2) == 1 { Some({ let mut a = [0u8; 32]; rng.fill(&mut a); a }) } else { None }, }.to_bytes().ok(),
"FileRun" => FileRun { first: rng.next_u64() as u32, count: rng.next_u64() as u32, }.to_bytes().ok(),
"FileRange" => FileRange { file_id: rng.next_u64() as u32, offset: rng.next_u64(), len: rng.next_u64(), }.to_bytes().ok(),
"BundleRecord" => BundleRecord { file_id: rng.next_u64() as u32, root: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, data: { let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }, }.to_bytes().ok(),
"RootItem" => RootItem { file_id: rng.next_u64() as u32, root: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"JobCopy" => JobCopy { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, src: rng.ascii(20), dest: rng.ascii(20), flags: rng.next_u64() as u32, }.to_bytes().ok(),
"JobRef" => JobRef { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"DiskCalibrate" => DiskCalibrate { dir: rng.ascii(20), files: rng.next_u64() as u32, size: rng.next_u64() as u32, }.to_bytes().ok(),
"CalPoint" => CalPoint { workers: rng.next_u64() as u8, files_per_s: rng.next_u64() as u32, create_us: rng.next_u64() as u32, fsync_us: rng.next_u64() as u32, }.to_bytes().ok(),
"DiskCalibrateResult" => DiskCalibrateResult { points: vec![CalPoint::default(); rng.below(3) as usize], }.to_bytes().ok(),
"C2cAllow" => C2cAllow { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, key: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, root: rng.ascii(20), }.to_bytes().ok(),
"C2cTicket" => C2cTicket { token: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, }.to_bytes().ok(),
"C2cSend" => C2cSend { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, host: rng.ascii(20), port: rng.next_u64() as u16, key: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, token: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, src: rng.ascii(20), dest: rng.ascii(20), flags: rng.next_u64() as u32, }.to_bytes().ok(),
"MgmtText" => MgmtText { body: { let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }, more: if rng.below(2) == 1 { Some(rng.next_u64() as u8) } else { None }, }.to_bytes().ok(),
"JobRun" => JobRun { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, op: rng.next_u64() as u8, args: { let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }, }.to_bytes().ok(),
"JobEntry" => JobEntry { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, kind: rng.next_u64() as u8, state: rng.next_u64() as u8, files_done: rng.next_u64() as u32, files_total: rng.next_u64() as u32, bytes_done: rng.next_u64(), bytes_total: rng.next_u64(), }.to_bytes().ok(),
"JobListResult" => JobListResult { jobs: vec![JobEntry::default(); rng.below(3) as usize], }.to_bytes().ok(),
"JnlOpen" => JnlOpen { job_id: { let mut a = [0u8; 16]; rng.fill(&mut a); a }, manifest_hash: { let mut a = [0u8; 32]; rng.fill(&mut a); a }, kind: rng.next_u64() as u8, flags: rng.next_u64() as u32, staged: rng.next_u64() as u8, root: rng.ascii(20), }.to_bytes().ok(),
"JnlBatch" => JnlBatch { files: vec![FileRun::default(); rng.below(3) as usize], ranges: vec![FileRange::default(); rng.below(3) as usize], roots: vec![RootItem::default(); rng.below(3) as usize], pack_segment: if rng.below(2) == 1 { Some(rng.next_u64() as u32) } else { None }, pack_offset: if rng.below(2) == 1 { Some(rng.next_u64()) } else { None }, pack_len: if rng.below(2) == 1 { Some(rng.next_u64()) } else { None }, }.to_bytes().ok(),
"PackRef" => PackRef { segment: rng.next_u64() as u32, offset: rng.next_u64(), len: rng.next_u64(), first_file: rng.next_u64() as u32, count: rng.next_u64() as u32, }.to_bytes().ok(),
"JnlSweep" => JnlSweep { files: vec![FileRun::default(); rng.below(3) as usize], }.to_bytes().ok(),
"JnlReset" => JnlReset { file_id: rng.next_u64() as u32, }.to_bytes().ok(),
"JnlSnapshot" => JnlSnapshot { done: vec![FileRun::default(); rng.below(3) as usize], ranges: vec![FileRange::default(); rng.below(3) as usize], roots: vec![RootItem::default(); rng.below(3) as usize], unswept: if rng.below(2) == 1 { Some({ let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }) } else { None }, segments: if rng.below(2) == 1 { Some({ let n = rng.below(41) as usize; let mut v = vec![0u8; n]; rng.fill(&mut v); v }) } else { None }, }.to_bytes().ok(),
"JnlDone" => JnlDone { status: rng.next_u64() as u16, }.to_bytes().ok(),
_ => None,
}
}
#[doc(hidden)]
pub fn roundtrip(name: &str, bytes: &[u8]) -> Option<Result<Vec<u8>, String>> {
fn rt<M: Message>(b: &[u8]) -> Result<Vec<u8>, String> {
M::decode(b).map_err(|e| e.to_string())?.to_bytes().map_err(|e| e.to_string())
}
Some(match name {
"Hs1" => rt::<Hs1>(bytes),
"Hs2" => rt::<Hs2>(bytes),
"Hs3" => rt::<Hs3>(bytes),
"Welcome" => rt::<Welcome>(bytes),
"PairPakeClient" => rt::<PairPakeClient>(bytes),
"PairPakeServer" => rt::<PairPakeServer>(bytes),
"PairConfirm" => rt::<PairConfirm>(bytes),
"PairResult" => rt::<PairResult>(bytes),
"Join" => rt::<Join>(bytes),
"JoinAck" => rt::<JoinAck>(bytes),
"Ping" => rt::<Ping>(bytes),
"Pong" => rt::<Pong>(bytes),
"Error" => rt::<Error>(bytes),
"Bye" => rt::<Bye>(bytes),
"RpcRequest" => rt::<RpcRequest>(bytes),
"RpcResponse" => rt::<RpcResponse>(bytes),
"JobOpen" => rt::<JobOpen>(bytes),
"JobOpenAck" => rt::<JobOpenAck>(bytes),
"ManifestPage" => rt::<ManifestPage>(bytes),
"ManifestEnd" => rt::<ManifestEnd>(bytes),
"JobMap" => rt::<JobMap>(bytes),
"Resume" => rt::<Resume>(bytes),
"Chunk" => rt::<Chunk>(bytes),
"Bundle" => rt::<Bundle>(bytes),
"Received" => rt::<Received>(bytes),
"Credit" => rt::<Credit>(bytes),
"Durable" => rt::<Durable>(bytes),
"FileRoot" => rt::<FileRoot>(bytes),
"FileRetry" => rt::<FileRetry>(bytes),
"Status" => rt::<Status>(bytes),
"JobDone" => rt::<JobDone>(bytes),
"JobCancel" => rt::<JobCancel>(bytes),
"SkipProgress" => rt::<SkipProgress>(bytes),
"NodeInfo" => rt::<NodeInfo>(bytes),
"HelloInfo" => rt::<HelloInfo>(bytes),
"ServerInfo" => rt::<ServerInfo>(bytes),
"ClientInfo" => rt::<ClientInfo>(bytes),
"PairingOpen" => rt::<PairingOpen>(bytes),
"CryptoBench" => rt::<CryptoBench>(bytes),
"CryptoBenchResult" => rt::<CryptoBenchResult>(bytes),
"ManifestEntry" => rt::<ManifestEntry>(bytes),
"FileRun" => rt::<FileRun>(bytes),
"FileRange" => rt::<FileRange>(bytes),
"BundleRecord" => rt::<BundleRecord>(bytes),
"RootItem" => rt::<RootItem>(bytes),
"JobCopy" => rt::<JobCopy>(bytes),
"JobRef" => rt::<JobRef>(bytes),
"DiskCalibrate" => rt::<DiskCalibrate>(bytes),
"CalPoint" => rt::<CalPoint>(bytes),
"DiskCalibrateResult" => rt::<DiskCalibrateResult>(bytes),
"C2cAllow" => rt::<C2cAllow>(bytes),
"C2cTicket" => rt::<C2cTicket>(bytes),
"C2cSend" => rt::<C2cSend>(bytes),
"MgmtText" => rt::<MgmtText>(bytes),
"JobRun" => rt::<JobRun>(bytes),
"JobEntry" => rt::<JobEntry>(bytes),
"JobListResult" => rt::<JobListResult>(bytes),
"JnlOpen" => rt::<JnlOpen>(bytes),
"JnlBatch" => rt::<JnlBatch>(bytes),
"PackRef" => rt::<PackRef>(bytes),
"JnlSweep" => rt::<JnlSweep>(bytes),
"JnlReset" => rt::<JnlReset>(bytes),
"JnlSnapshot" => rt::<JnlSnapshot>(bytes),
"JnlDone" => rt::<JnlDone>(bytes),
_ => return None,
})
}