aven_protocol/wire/bootstrap/
batch.rs

1//! Binary framing of one staging batch:
2//! `b"AVBU" | u16 version | u32 header_len | JSON header | record bytes`.
3//!
4//! Integers are big-endian. The header names each record's slot and length;
5//! record bytes follow back to back in header order with nothing after them.
6//! Decoding checks every bound and length before handing out any record, so a
7//! server can refuse a malformed batch without touching storage.
8use anyhow::{Result, ensure};
9use serde::{Deserialize, Deserializer, Serialize, Serializer};
10
11use super::{Component, MAX_REQUEST_BYTES};
12
13pub const CONTENT_TYPE: &str = "application/vnd.aven.bootstrap-batch";
14pub const MAGIC: &[u8; 4] = b"AVBU";
15pub const VERSION: u16 = 1;
16/// Upper bound on the sum of record lengths in one batch.
17pub const MAX_PAYLOAD: usize = 4 * 1_048_576;
18pub const MAX_RECORDS: usize = 64;
19pub const MAX_HEADER: usize = 4096;
20const PREFIX: usize = MAGIC.len() + 2 + 4;
21/// Upper bound on one encoded batch. A single record of at most
22/// [`MAX_REQUEST_BYTES`] always fits.
23pub const MAX_BYTES: usize = PREFIX + MAX_HEADER + MAX_PAYLOAD;
24
25#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
26#[serde(deny_unknown_fields)]
27pub struct Header {
28    #[serde(with = "hex32")]
29    pub vault: [u8; 32],
30    #[serde(with = "hex32")]
31    pub genesis: [u8; 32],
32    #[serde(with = "hex32")]
33    pub bootstrap: [u8; 32],
34    #[serde(with = "hex32")]
35    pub commitment: [u8; 32],
36    pub records: Vec<Slot>,
37}
38
39#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
40#[serde(deny_unknown_fields)]
41pub struct Slot {
42    #[serde(serialize_with = "component_out", deserialize_with = "component_in")]
43    pub component: Component,
44    pub index: u64,
45    pub len: u64,
46}
47
48/// A decoded batch whose records borrow the request body.
49pub struct Batch<'a> {
50    pub header: Header,
51    pub records: Vec<&'a [u8]>,
52}
53
54/// Checks the header's shape: record count, per-record length, the payload
55/// total, unique slots, and that catalog slices never share a batch with
56/// the records a catalog describes.
57fn check(header: &Header) -> Result<usize> {
58    let records = &header.records;
59    ensure!(
60        !records.is_empty() && records.len() <= MAX_RECORDS,
61        "error bootstrap-batch-shape"
62    );
63    let mut payload = 0_usize;
64    for slot in records {
65        let len = usize::try_from(slot.len)
66            .ok()
67            .filter(|len| *len > 0 && *len <= MAX_REQUEST_BYTES);
68        let len = len.ok_or_else(|| anyhow::anyhow!("error bootstrap-batch-shape"))?;
69        payload = payload
70            .checked_add(len)
71            .filter(|total| *total <= MAX_PAYLOAD)
72            .ok_or_else(|| anyhow::anyhow!("error bootstrap-batch-shape"))?;
73    }
74    let catalogs = records
75        .iter()
76        .filter(|slot| slot.component.catalog().is_some())
77        .count();
78    ensure!(
79        catalogs == 0 || catalogs == records.len(),
80        "error bootstrap-batch-mixed"
81    );
82    for (position, slot) in records.iter().enumerate() {
83        ensure!(
84            records[..position]
85                .iter()
86                .all(|other| (other.component, other.index) != (slot.component, slot.index)),
87            "error bootstrap-batch-shape"
88        );
89    }
90    Ok(payload)
91}
92
93/// The encoded header length, or `None` if it exceeds [`MAX_HEADER`].
94pub fn header_len(header: &Header) -> Option<usize> {
95    serde_json::to_vec(header)
96        .ok()
97        .map(|bytes| bytes.len())
98        .filter(|len| *len <= MAX_HEADER)
99}
100
101pub fn encode(header: &Header, records: &[&[u8]]) -> Result<Vec<u8>> {
102    let payload = check(header)?;
103    ensure!(
104        records.len() == header.records.len()
105            && records
106                .iter()
107                .zip(&header.records)
108                .all(|(bytes, slot)| bytes.len() as u64 == slot.len),
109        "error bootstrap-batch-shape"
110    );
111    let json = serde_json::to_vec(header)?;
112    ensure!(json.len() <= MAX_HEADER, "error bootstrap-batch-shape");
113    let mut out = Vec::with_capacity(PREFIX + json.len() + payload);
114    out.extend_from_slice(MAGIC);
115    out.extend_from_slice(&VERSION.to_be_bytes());
116    out.extend_from_slice(&(json.len() as u32).to_be_bytes());
117    out.extend_from_slice(&json);
118    for record in records {
119        out.extend_from_slice(record);
120    }
121    Ok(out)
122}
123
124pub fn decode(bytes: &[u8]) -> Result<Batch<'_>> {
125    ensure!(
126        bytes.len() >= PREFIX && bytes.len() <= MAX_BYTES,
127        "error bootstrap-batch-framing"
128    );
129    let (magic, rest) = bytes.split_at(MAGIC.len());
130    ensure!(magic == MAGIC, "error bootstrap-batch-framing");
131    let (version, rest) = rest.split_at(2);
132    ensure!(
133        u16::from_be_bytes([version[0], version[1]]) == VERSION,
134        "error bootstrap-batch-version"
135    );
136    let (len, rest) = rest.split_at(4);
137    let len = u32::from_be_bytes([len[0], len[1], len[2], len[3]]) as usize;
138    ensure!(
139        len <= MAX_HEADER && len <= rest.len(),
140        "error bootstrap-batch-framing"
141    );
142    let (json, mut body) = rest.split_at(len);
143    let header: Header = serde_json::from_slice(json)
144        .map_err(|_| anyhow::anyhow!("error bootstrap-batch-framing"))?;
145    let payload = check(&header)?;
146    ensure!(body.len() == payload, "error bootstrap-batch-framing");
147    let mut records = Vec::with_capacity(header.records.len());
148    for slot in &header.records {
149        let (record, rest) = body.split_at(slot.len as usize);
150        records.push(record);
151        body = rest;
152    }
153    Ok(Batch { header, records })
154}
155
156fn component_out<S: Serializer>(component: &Component, s: S) -> Result<S::Ok, S::Error> {
157    s.serialize_str(&hex::encode(component.key()))
158}
159
160fn component_in<'de, D: Deserializer<'de>>(d: D) -> Result<Component, D::Error> {
161    let text = String::deserialize(d)?;
162    hex::decode(&text)
163        .ok()
164        .and_then(|key| Component::from_key(&key))
165        .ok_or_else(|| serde::de::Error::custom("invalid component"))
166}
167
168mod hex32 {
169    use serde::{Deserialize, Deserializer, Serializer};
170
171    pub(super) fn serialize<S: Serializer>(value: &[u8; 32], s: S) -> Result<S::Ok, S::Error> {
172        s.serialize_str(&hex::encode(value))
173    }
174
175    pub(super) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 32], D::Error> {
176        let text = String::deserialize(d)?;
177        let mut out = [0; 32];
178        hex::decode_to_slice(&text, &mut out)
179            .map_err(|_| serde::de::Error::custom("invalid 32-byte hex"))?;
180        Ok(out)
181    }
182}
183
184#[cfg(test)]
185mod tests;
aven_protocol/wire/bootstrap/batch.rs

aven_protocol/wire/bootstrap/
batch.rs
