use raptorq::{Decoder, Encoder, EncodingPacket, ObjectTransmissionInformation, PayloadId};
use super::RpcCodecError;
use crate::aux_mux::AUX_MAX_PAYLOAD;
use crate::node_status::MAX_DEVICE_ID;
pub const RPC_RESPONSE_OVERHEAD_BASE: usize = 17;
pub const RESPONSE_HEADERS_FLAG: u16 = 0x8000;
pub const MAX_RESPONSE_HEADER_BLOCK: usize = 1024;
pub const RPC_REPAIR_SYMBOLS: u32 = 4;
pub const MAX_RESPONSE_FRAGMENT: usize =
AUX_MAX_PAYLOAD - RPC_RESPONSE_OVERHEAD_BASE - MAX_DEVICE_ID;
pub const MAX_RESPONSE_FRAGMENTS: usize = 64;
pub const MAX_RESPONSE_BODY: usize =
(MAX_RESPONSE_FRAGMENTS - RPC_REPAIR_SYMBOLS as usize) * MAX_RESPONSE_FRAGMENT;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RpcResponse<'a> {
pub id: u32,
pub sender: &'a [u8],
pub status: u16,
pub index: u16,
pub total: u16,
pub oti: u32,
pub body: &'a [u8],
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResponseSymbols {
pub oti: u32,
pub symbols: Vec<Vec<u8>>,
}
fn response_shape(len: usize) -> (usize, usize) {
let k = len.div_ceil(MAX_RESPONSE_FRAGMENT);
(k, len.div_ceil(k))
}
fn response_config(len: u32, symbol_size: usize) -> ObjectTransmissionInformation {
ObjectTransmissionInformation::new(len as u64, symbol_size as u16, 1, 1, 1)
}
pub fn split_response(body: &[u8]) -> Option<ResponseSymbols> {
if body.is_empty() {
return Some(ResponseSymbols {
oti: 0,
symbols: vec![Vec::new()],
});
}
if body.len() > MAX_RESPONSE_BODY {
return None;
}
let (_, symbol_size) = response_shape(body.len());
let encoder = Encoder::new(body, response_config(body.len() as u32, symbol_size));
let oti = encoder.get_config().transfer_length() as u32;
Some(ResponseSymbols {
oti,
symbols: encoder
.get_encoded_packets(RPC_REPAIR_SYMBOLS)
.into_iter()
.map(|packet| packet.split().1)
.collect(),
})
}
pub type ResponseHeader = (String, String);
pub fn pack_response(status: u16, body: &[u8], headers: &[ResponseHeader]) -> (u16, Vec<u8>) {
let mut block: Vec<u8> = Vec::new();
for (name, value) in headers {
if name.is_empty() || name.len() > u8::MAX as usize || value.len() > u16::MAX as usize {
continue;
}
let entry = 1 + name.len() + 2 + value.len();
if block.len() + entry > MAX_RESPONSE_HEADER_BLOCK {
break;
}
block.push(name.len() as u8);
block.extend_from_slice(name.as_bytes());
block.extend_from_slice(&(value.len() as u16).to_be_bytes());
block.extend_from_slice(value.as_bytes());
}
if block.is_empty() {
return (status, body.to_vec());
}
let mut object = Vec::with_capacity(body.len() + block.len() + 4);
object.extend_from_slice(body);
let block_len = block.len() as u32;
object.extend_from_slice(&block);
object.extend_from_slice(&block_len.to_be_bytes());
(status | RESPONSE_HEADERS_FLAG, object)
}
pub fn unpack_response(status: u16, object: Vec<u8>) -> (u16, Vec<u8>, Vec<ResponseHeader>) {
if status & RESPONSE_HEADERS_FLAG == 0 {
return (status, object, Vec::new());
}
let status = status & !RESPONSE_HEADERS_FLAG;
let Some(len_at) = object.len().checked_sub(4) else {
return (status, object, Vec::new());
};
let block_len = u32::from_be_bytes([
object[len_at],
object[len_at + 1],
object[len_at + 2],
object[len_at + 3],
]) as usize;
let Some(block_at) = len_at.checked_sub(block_len) else {
return (status, object, Vec::new());
};
let Some(headers) = parse_header_block(&object[block_at..len_at]) else {
return (status, object, Vec::new());
};
let mut body = object;
body.truncate(block_at);
(status, body, headers)
}
fn parse_header_block(mut block: &[u8]) -> Option<Vec<ResponseHeader>> {
let mut out = Vec::new();
while !block.is_empty() {
let name_len = *block.first()? as usize;
if name_len == 0 {
return None;
}
let rest = block.get(1..)?;
let name = std::str::from_utf8(rest.get(..name_len)?).ok()?;
let rest = rest.get(name_len..)?;
let value_len = u16::from_be_bytes([*rest.first()?, *rest.get(1)?]) as usize;
let rest = rest.get(2..)?;
let value = std::str::from_utf8(rest.get(..value_len)?).ok()?;
out.push((name.to_string(), value.to_string()));
block = rest.get(value_len..)?;
}
Some(out)
}
pub fn encode_response_fragment(
sender: &[u8],
id: u32,
status: u16,
index: u16,
total: u16,
oti: u32,
symbol: &[u8],
) -> Option<Vec<u8>> {
if sender.len() > MAX_DEVICE_ID {
return None;
}
if total == 0 || index >= total {
return None;
}
if symbol.len() > u16::MAX as usize {
return None;
}
let encoded_len = RPC_RESPONSE_OVERHEAD_BASE + sender.len() + symbol.len();
if encoded_len > AUX_MAX_PAYLOAD {
return None;
}
let mut out = Vec::with_capacity(encoded_len);
out.push(sender.len() as u8);
out.extend_from_slice(sender);
out.extend_from_slice(&id.to_be_bytes());
out.extend_from_slice(&status.to_be_bytes());
out.extend_from_slice(&index.to_be_bytes());
out.extend_from_slice(&total.to_be_bytes());
out.extend_from_slice(&oti.to_be_bytes());
out.extend_from_slice(&(symbol.len() as u16).to_be_bytes());
out.extend_from_slice(symbol);
Some(out)
}
pub fn decode_response(payload: &[u8]) -> Result<RpcResponse<'_>, RpcCodecError> {
let &sender_len = payload.first().ok_or(RpcCodecError::TooShort)?;
if sender_len as usize > MAX_DEVICE_ID {
return Err(RpcCodecError::BadSenderLen(sender_len));
}
let sender_len = sender_len as usize;
let header = RPC_RESPONSE_OVERHEAD_BASE + sender_len;
if payload.len() < header {
return Err(RpcCodecError::TooShort);
}
let sender = &payload[1..1 + sender_len];
let fixed = &payload[1 + sender_len..];
let id = u32::from_be_bytes([fixed[0], fixed[1], fixed[2], fixed[3]]);
let status = u16::from_be_bytes([fixed[4], fixed[5]]);
let index = u16::from_be_bytes([fixed[6], fixed[7]]);
let total = u16::from_be_bytes([fixed[8], fixed[9]]);
let oti = u32::from_be_bytes([fixed[10], fixed[11], fixed[12], fixed[13]]);
let frag_len = u16::from_be_bytes([fixed[14], fixed[15]]) as usize;
if header + frag_len != payload.len() {
return Err(RpcCodecError::LengthMismatch {
declared: header + frag_len,
actual: payload.len(),
});
}
if total == 0 || index >= total {
return Err(RpcCodecError::BadFragmentIndex { index, total });
}
Ok(RpcResponse {
id,
sender,
status,
index,
total,
oti,
body: &payload[header..],
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FragmentOutcome {
Pending,
Complete(Vec<u8>),
BadSymbol,
}
#[derive(Debug)]
pub struct ResponseDecoder {
inner: Option<Decoder>,
symbol_size: usize,
symbol_id_limit: u32,
}
impl ResponseDecoder {
pub fn new(oti: u32) -> Option<Self> {
let len = oti as usize;
if len > MAX_RESPONSE_BODY {
return None;
}
if len == 0 {
return Some(Self {
inner: None,
symbol_size: 0,
symbol_id_limit: RPC_REPAIR_SYMBOLS,
});
}
let (k, symbol_size) = response_shape(len);
Some(Self {
inner: Some(Decoder::new(response_config(oti, symbol_size))),
symbol_size,
symbol_id_limit: k as u32 + RPC_REPAIR_SYMBOLS,
})
}
pub fn symbol_size(&self) -> usize {
self.symbol_size
}
pub fn push(&mut self, index: u16, symbol: &[u8]) -> FragmentOutcome {
if symbol.len() != self.symbol_size || index as u32 >= self.symbol_id_limit {
return FragmentOutcome::BadSymbol;
}
let Some(decoder) = self.inner.as_mut() else {
return FragmentOutcome::Complete(Vec::new());
};
let packet = EncodingPacket::new(PayloadId::new(0, index as u32), symbol.to_vec());
match decoder.decode(packet) {
Some(body) => FragmentOutcome::Complete(body),
None => FragmentOutcome::Pending,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
const SENDER: &[u8] = b"ados-abc123";
fn hdr(name: &str, value: &str) -> ResponseHeader {
(name.to_string(), value.to_string())
}
#[test]
fn a_headerless_response_is_byte_identical_to_a_build_without_header_support() {
let body = br#"{"ok":true}"#;
let (status, object) = pack_response(200, body, &[]);
assert_eq!(status, 200, "no flag when there is nothing to carry");
assert_eq!(object, body);
}
#[test]
fn a_content_type_survives_the_round_trip() {
let body = b"id,lat,lon\n1,0,0\n";
let headers = vec![
hdr("content-type", "text/csv; charset=utf-8"),
hdr("content-disposition", "attachment; filename=\"log.csv\""),
];
let (wire_status, object) = pack_response(200, body, &headers);
assert_eq!(wire_status & RESPONSE_HEADERS_FLAG, RESPONSE_HEADERS_FLAG);
let (status, got_body, got_headers) = unpack_response(wire_status, object);
assert_eq!(
status, 200,
"the flag is stripped, not leaked to the caller"
);
assert_eq!(got_body, body);
assert_eq!(got_headers, headers);
}
#[test]
fn headers_survive_the_whole_fragment_round_trip_with_a_lost_fragment() {
let body = vec![0xABu8; 4000];
let headers = vec![hdr("content-type", "application/octet-stream")];
let (wire_status, object) = pack_response(200, &body, &headers);
let split = split_response(&object).unwrap();
let total = split.symbols.len() as u16;
let mut decoder = ResponseDecoder::new(split.oti).unwrap();
let mut assembled = None;
for (i, symbol) in split.symbols.iter().enumerate() {
if i == 1 {
continue; }
let payload = encode_response_fragment(
SENDER,
9,
wire_status,
i as u16,
total,
split.oti,
symbol,
)
.unwrap();
let frag = decode_response(&payload).unwrap();
assert_eq!(frag.status, wire_status);
if let FragmentOutcome::Complete(out) = decoder.push(frag.index, frag.body) {
assembled = Some(out);
break;
}
}
let (status, got_body, got_headers) = unpack_response(
wire_status,
assembled.expect("repair symbols must close the block"),
);
assert_eq!(status, 200);
assert_eq!(got_body, body);
assert_eq!(got_headers, headers);
}
#[test]
fn a_damaged_trailer_yields_the_object_verbatim_rather_than_a_truncated_body() {
let (wire_status, mut object) = pack_response(200, b"payload", &[hdr("a", "b")]);
let last = object.len() - 1;
object[last] = 0xFF; let (status, body, headers) = unpack_response(wire_status, object.clone());
assert_eq!(status, 200);
assert_eq!(body, object);
assert!(headers.is_empty());
}
#[test]
fn a_header_block_past_the_ceiling_is_truncated_not_allowed_to_eat_the_body() {
let huge = "x".repeat(MAX_RESPONSE_HEADER_BLOCK);
let headers = vec![hdr("content-type", "text/plain"), hdr("bloat", &huge)];
let (wire_status, object) = pack_response(200, b"body", &headers);
let (_, body, got) = unpack_response(wire_status, object);
assert_eq!(body, b"body");
assert_eq!(got, vec![hdr("content-type", "text/plain")]);
}
fn round_trip(body: &[u8], order: &[usize], drop: &[usize]) -> Option<Vec<u8>> {
let split = split_response(body).unwrap();
let total = split.symbols.len() as u16;
let mut decoder: Option<ResponseDecoder> = None;
for &i in order {
if drop.contains(&i) {
continue;
}
let payload = encode_response_fragment(
SENDER,
7,
200,
i as u16,
total,
split.oti,
&split.symbols[i],
)
.unwrap();
let frag = decode_response(&payload).unwrap();
assert_eq!(frag.sender, SENDER);
assert_eq!(frag.oti, split.oti);
assert_eq!(frag.total, total);
let d = decoder.get_or_insert_with(|| ResponseDecoder::new(frag.oti).unwrap());
if let FragmentOutcome::Complete(out) = d.push(frag.index, frag.body) {
return Some(out);
}
}
None
}
#[test]
fn round_trips_a_single_fragment_response() {
let body = br#"{"device_id":"abc","name":"Skynode A7S"}"#;
let split = split_response(body).unwrap();
assert_eq!(split.oti, body.len() as u32);
let enc = encode_response_fragment(
SENDER,
42,
200,
0,
split.symbols.len() as u16,
split.oti,
&split.symbols[0],
)
.unwrap();
let dec = decode_response(&enc).unwrap();
assert_eq!(dec.id, 42);
assert_eq!(dec.sender, SENDER);
assert_eq!(dec.status, 200);
assert_eq!(dec.index, 0);
assert_eq!(round_trip(body, &[0], &[]).as_deref(), Some(&body[..]));
}
#[test]
fn an_empty_body_still_produces_one_fragment() {
let split = split_response(&[]).unwrap();
assert_eq!(split.oti, 0);
assert_eq!(split.symbols.len(), 1);
assert!(split.symbols[0].is_empty());
let enc = encode_response_fragment(SENDER, 7, 204, 0, 1, 0, &split.symbols[0]).unwrap();
let dec = decode_response(&enc).unwrap();
assert_eq!(dec.status, 204);
assert_eq!(dec.total, 1);
assert!(dec.body.is_empty());
let mut decoder = ResponseDecoder::new(dec.oti).unwrap();
assert_eq!(
decoder.push(dec.index, dec.body),
FragmentOutcome::Complete(Vec::new())
);
}
#[test]
fn a_response_carries_k_systematic_plus_the_repair_symbols() {
let body: Vec<u8> = (0..3000u32).map(|i| (i % 251) as u8).collect();
let split = split_response(&body).unwrap();
let k = body.len().div_ceil(MAX_RESPONSE_FRAGMENT);
assert_eq!(k, 3);
assert_eq!(split.symbols.len(), k + RPC_REPAIR_SYMBOLS as usize);
let size = split.symbols[0].len();
assert!(split.symbols.iter().all(|s| s.len() == size));
assert!(size <= MAX_RESPONSE_FRAGMENT);
}
#[test]
fn a_thirty_kilobyte_body_survives_losing_any_four_fragments() {
let body: Vec<u8> = (0..29_339u32).map(|i| (i % 251) as u8).collect();
let split = split_response(&body).unwrap();
assert_eq!(split.symbols.len(), 30, "26 systematic + 4 repair");
let order: Vec<usize> = (0..30).collect();
for drop in [
vec![0usize, 1, 2, 3],
vec![26usize, 27, 28, 29],
vec![0usize, 9, 17, 29],
vec![5usize, 6, 7, 25],
] {
assert_eq!(
round_trip(&body, &order, &drop).as_deref(),
Some(&body[..]),
"dropping {drop:?}"
);
}
}
#[test]
fn fragments_rejoin_out_of_order() {
let body: Vec<u8> = (0..8000u32).map(|i| (i % 251) as u8).collect();
let split = split_response(&body).unwrap();
let mut order: Vec<usize> = (0..split.symbols.len()).collect();
order.reverse();
assert_eq!(round_trip(&body, &order, &[]).as_deref(), Some(&body[..]));
}
#[test]
fn a_body_past_the_reassembly_ceiling_does_not_split() {
assert!(split_response(&vec![0u8; MAX_RESPONSE_BODY + 1]).is_none());
let at_ceiling = vec![0u8; MAX_RESPONSE_BODY];
assert_eq!(
split_response(&at_ceiling).unwrap().symbols.len(),
MAX_RESPONSE_FRAGMENTS
);
}
#[test]
fn rejects_a_fragment_larger_than_one_aux_frame() {
let big = vec![0u8; MAX_RESPONSE_FRAGMENT + 1];
let sender = vec![b'a'; MAX_DEVICE_ID];
assert!(encode_response_fragment(&sender, 1, 200, 0, 1, 1, &big).is_none());
let fits = vec![0u8; MAX_RESPONSE_FRAGMENT];
let enc = encode_response_fragment(&sender, 1, 200, 0, 1, 1, &fits).unwrap();
assert_eq!(enc.len(), AUX_MAX_PAYLOAD, "the budget is exact");
assert_eq!(
decode_response(&enc).unwrap().body.len(),
MAX_RESPONSE_FRAGMENT
);
}
#[test]
fn rejects_a_sender_longer_than_a_device_id() {
let too_long = vec![b'a'; MAX_DEVICE_ID + 1];
assert!(encode_response_fragment(&too_long, 1, 200, 0, 1, 1, b"x").is_none());
let mut bad = vec![(MAX_DEVICE_ID + 1) as u8];
bad.extend_from_slice(&too_long);
bad.extend_from_slice(&[0u8; 16]);
assert_eq!(
decode_response(&bad).unwrap_err(),
RpcCodecError::BadSenderLen((MAX_DEVICE_ID + 1) as u8)
);
}
#[test]
fn rejects_a_fragment_whose_index_is_outside_its_total() {
assert!(encode_response_fragment(SENDER, 1, 200, 3, 3, 1, b"x").is_none());
assert!(encode_response_fragment(SENDER, 1, 200, 0, 0, 1, b"x").is_none());
let mut bad = vec![0u8];
bad.extend_from_slice(&1u32.to_be_bytes());
bad.extend_from_slice(&200u16.to_be_bytes());
bad.extend_from_slice(&4u16.to_be_bytes());
bad.extend_from_slice(&2u16.to_be_bytes());
bad.extend_from_slice(&1u32.to_be_bytes());
bad.extend_from_slice(&1u16.to_be_bytes());
bad.push(b'x');
assert_eq!(
decode_response(&bad).unwrap_err(),
RpcCodecError::BadFragmentIndex { index: 4, total: 2 }
);
}
#[test]
fn rejects_a_truncated_response() {
let full = encode_response_fragment(SENDER, 9, 200, 0, 1, 4, b"body").unwrap();
for cut in [0usize, 1, 4, RPC_RESPONSE_OVERHEAD_BASE + SENDER.len() - 1] {
let mut truncated = full.clone();
truncated.truncate(cut);
assert_eq!(
decode_response(&truncated).unwrap_err(),
RpcCodecError::TooShort,
"cut to {cut}"
);
}
}
#[test]
fn catches_a_length_mismatch_in_a_response() {
let mut bad = vec![0u8];
bad.extend_from_slice(&1u32.to_be_bytes());
bad.extend_from_slice(&200u16.to_be_bytes());
bad.extend_from_slice(&0u16.to_be_bytes());
bad.extend_from_slice(&1u16.to_be_bytes());
bad.extend_from_slice(&50u32.to_be_bytes());
bad.extend_from_slice(&50u16.to_be_bytes());
assert_eq!(
decode_response(&bad).unwrap_err(),
RpcCodecError::LengthMismatch {
declared: RPC_RESPONSE_OVERHEAD_BASE + 50,
actual: RPC_RESPONSE_OVERHEAD_BASE
}
);
}
#[test]
fn an_impossible_index_total_pair_is_its_own_fault_not_a_length_fault() {
let mut bad = vec![0u8];
bad.extend_from_slice(&1u32.to_be_bytes());
bad.extend_from_slice(&200u16.to_be_bytes());
bad.extend_from_slice(&3u16.to_be_bytes());
bad.extend_from_slice(&2u16.to_be_bytes());
bad.extend_from_slice(&0u32.to_be_bytes());
bad.extend_from_slice(&0u16.to_be_bytes());
assert_eq!(
decode_response(&bad).unwrap_err(),
RpcCodecError::BadFragmentIndex { index: 3, total: 2 }
);
}
#[test]
fn a_decoder_refuses_an_unbounded_transfer_length() {
assert!(ResponseDecoder::new(u32::MAX).is_none());
assert!(ResponseDecoder::new(MAX_RESPONSE_BODY as u32 + 1).is_none());
assert!(ResponseDecoder::new(MAX_RESPONSE_BODY as u32).is_some());
}
#[test]
fn a_decoder_drops_a_symbol_of_the_wrong_size_or_id() {
let body: Vec<u8> = (0..5000u32).map(|i| (i % 251) as u8).collect();
let split = split_response(&body).unwrap();
let mut decoder = ResponseDecoder::new(split.oti).unwrap();
let size = decoder.symbol_size();
assert_eq!(size, split.symbols[0].len());
assert_eq!(
decoder.push(0, &vec![0u8; size - 1]),
FragmentOutcome::BadSymbol
);
let k = body.len().div_ceil(MAX_RESPONSE_FRAGMENT);
assert_eq!(
decoder.push((k + RPC_REPAIR_SYMBOLS as usize) as u16, &vec![0u8; size]),
FragmentOutcome::BadSymbol
);
for (i, symbol) in split.symbols.iter().enumerate() {
if let FragmentOutcome::Complete(out) = decoder.push(i as u16, symbol) {
assert_eq!(out, body);
return;
}
}
panic!("the full symbol set must decode");
}
#[test]
fn every_body_size_across_the_fragment_range_round_trips() {
for k in [1usize, 2, 26, 60] {
for len in [
(k - 1) * MAX_RESPONSE_FRAGMENT + 1,
k * MAX_RESPONSE_FRAGMENT,
] {
let body: Vec<u8> = (0..len).map(|i| ((i * 7 + 13) % 251) as u8).collect();
let split = split_response(&body).unwrap();
assert_eq!(
split.symbols.len(),
k + RPC_REPAIR_SYMBOLS as usize,
"len {len}"
);
let order: Vec<usize> = (0..split.symbols.len()).collect();
assert_eq!(
round_trip(&body, &order, &[0, 1]).as_deref(),
Some(&body[..]),
"len {len}"
);
}
}
}
#[test]
fn frame_budgets_account_for_fixed_overhead() {
assert_eq!(RPC_RESPONSE_OVERHEAD_BASE, 17);
assert_eq!(MAX_RESPONSE_FRAGMENT, 1200 - 17 - 32);
assert_eq!(MAX_RESPONSE_FRAGMENT, 1151);
assert_eq!(MAX_RESPONSE_BODY, 60 * 1151);
const { assert!(MAX_RESPONSE_BODY > 29_339) };
assert_eq!(29_339usize.div_ceil(MAX_RESPONSE_FRAGMENT), 26);
}
}