pub const PKT_CONTROL: u8 = 4;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum FrameType {
Ack = 0x01,
Nak = 0x02,
Loss = 0x03,
Timing = 0x04,
Ring = 0x05,
Path = 0x06,
Link = 0x07,
LossAcct = 0x08,
Pmtu = 0x09,
BwProbe = 0x0A,
Trace = 0x0B,
AvailBw = 0x0C,
Forecast = 0x0D,
Periodicity = 0x0E,
SessionChallenge = 0x0F,
SessionResponse = 0x10,
SessionAnnounce = 0x11,
}
impl FrameType {
fn from_u8(v: u8) -> Option<Self> {
Some(match v {
0x01 => Self::Ack,
0x02 => Self::Nak,
0x03 => Self::Loss,
0x04 => Self::Timing,
0x05 => Self::Ring,
0x06 => Self::Path,
0x07 => Self::Link,
0x08 => Self::LossAcct,
0x09 => Self::Pmtu,
0x0A => Self::BwProbe,
0x0B => Self::Trace,
0x0C => Self::AvailBw,
0x0D => Self::Forecast,
0x0E => Self::Periodicity,
0x0F => Self::SessionChallenge,
0x10 => Self::SessionResponse,
0x11 => Self::SessionAnnounce,
_ => return None,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct SessionFrame {
pub epoch: u32,
pub nonce: u64,
}
pub const NONCE_MASK: u64 = (1u64 << 62) - 1;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct AckFrame {
pub ack_through: u32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct NakFrame {
pub block: u32,
pub mask: u32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct LossFrame {
pub loss_x255: u8,
pub burstiness_x255: u8,
pub owd_trend_class: u8,
pub loss_class: u8,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct TimingFrame {
pub send_ts: u64,
pub echo_ts: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct RingFrame {
pub fill_pct: u8,
pub ring_kind: u8,
pub producers: u8,
pub consumers: u8,
pub trend: u8,
pub flags: u8,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct PathFrame {
pub ttl: u8,
pub ecn: u8,
pub hop_count: u8,
pub ce_count: u64,
pub ect_count: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct LinkFrame {
pub class: u8,
pub quality: u8,
}
pub mod link_class {
pub const UNKNOWN: u8 = 0;
pub const LOOPBACK: u8 = 1;
pub const WIRED: u8 = 2;
pub const WIFI: u8 = 3;
pub const CELLULAR: u8 = 4;
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct LossAcctFrame {
pub seq: u32,
pub last_recv_seq: u32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct PmtuFrame {
pub pmtu: u16,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct BwProbeFrame {
pub probe_id: u8,
pub idx: u8,
pub send_ts: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct TraceFrame {
pub hop_ttl: u8,
pub probe_id: u8,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct AvailBwFrame {
pub avail_kbps: u64,
pub capacity_kbps: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct ForecastFrame {
pub forecast_kbps: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct PeriodicityFrame {
pub period_ds: u64,
pub secs_to_spike_ds: u64,
pub confidence_x255: u8,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ControlPacket {
pub ack: Option<AckFrame>,
pub nak: Option<NakFrame>,
pub loss: Option<LossFrame>,
pub timing: Option<TimingFrame>,
pub ring: Option<RingFrame>,
pub path: Option<PathFrame>,
pub link: Option<LinkFrame>,
pub loss_acct: Option<LossAcctFrame>,
pub pmtu: Option<PmtuFrame>,
pub bw_probe: Vec<BwProbeFrame>,
pub trace: Vec<TraceFrame>,
pub avail_bw: Option<AvailBwFrame>,
pub forecast: Option<ForecastFrame>,
pub periodicity: Option<PeriodicityFrame>,
pub session_challenge: Option<SessionFrame>,
pub session_response: Option<SessionFrame>,
pub session_announce: Option<u32>,
}
impl ControlPacket {
pub fn new() -> Self {
Self::default()
}
pub fn is_empty(&self) -> bool {
self.ack.is_none()
&& self.nak.is_none()
&& self.loss.is_none()
&& self.timing.is_none()
&& self.ring.is_none()
&& self.path.is_none()
&& self.link.is_none()
&& self.loss_acct.is_none()
&& self.pmtu.is_none()
&& self.bw_probe.is_empty()
&& self.trace.is_empty()
}
}
pub fn is_control(buf: &[u8]) -> bool {
!buf.is_empty() && buf[0] == PKT_CONTROL
}
fn put_varint(out: &mut Vec<u8>, v: u64) {
const MAX62: u64 = (1 << 62) - 1;
let v = v.min(MAX62);
if v < (1 << 6) {
out.push(v as u8);
} else if v < (1 << 14) {
out.push(0x40 | (v >> 8) as u8);
out.push(v as u8);
} else if v < (1 << 30) {
out.push(0x80 | (v >> 24) as u8);
out.extend_from_slice(&(v as u32).to_be_bytes()[1..]);
} else {
out.push(0xC0 | (v >> 56) as u8);
out.extend_from_slice(&v.to_be_bytes()[1..]);
}
}
fn get_varint(buf: &[u8], pos: usize) -> Option<(u64, usize)> {
let first = *buf.get(pos)?;
let len = 1usize << (first >> 6);
if pos + len > buf.len() {
return None;
}
let mut v = (first & 0x3F) as u64;
for &b in &buf[pos + 1..pos + len] {
v = (v << 8) | b as u64;
}
Some((v, pos + len))
}
fn put_frame(out: &mut Vec<u8>, ty: FrameType, body: &[u8]) {
out.push(ty as u8);
put_varint(out, body.len() as u64);
out.extend_from_slice(body);
}
pub fn pad_control_to(buf: &mut Vec<u8>, target_len: usize) {
const PAD_TYPE: u8 = 0x7F;
const HEADER: usize = 3; if target_len < buf.len() + HEADER + 64 {
return;
}
let body_len = target_len - buf.len() - HEADER;
buf.push(PAD_TYPE);
put_varint(buf, body_len as u64);
buf.resize(buf.len() + body_len, 0);
}
pub fn encode_control(p: &ControlPacket) -> Vec<u8> {
let mut out = Vec::with_capacity(64);
out.push(PKT_CONTROL);
let mut body = Vec::with_capacity(16);
if let Some(f) = p.ack {
body.clear();
put_varint(&mut body, f.ack_through as u64);
put_frame(&mut out, FrameType::Ack, &body);
}
if let Some(f) = p.nak {
body.clear();
put_varint(&mut body, f.block as u64);
put_varint(&mut body, f.mask as u64);
put_frame(&mut out, FrameType::Nak, &body);
}
if let Some(f) = p.loss {
put_frame(
&mut out,
FrameType::Loss,
&[f.loss_x255, f.burstiness_x255, f.owd_trend_class, f.loss_class],
);
}
if let Some(f) = p.timing {
body.clear();
put_varint(&mut body, f.send_ts);
put_varint(&mut body, f.echo_ts);
put_frame(&mut out, FrameType::Timing, &body);
}
if let Some(f) = p.ring {
put_frame(
&mut out,
FrameType::Ring,
&[
f.fill_pct,
f.ring_kind,
f.producers,
f.consumers,
f.trend,
f.flags,
],
);
}
if let Some(f) = p.path {
body.clear();
body.extend_from_slice(&[f.ttl, f.ecn, f.hop_count]);
put_varint(&mut body, f.ce_count);
put_varint(&mut body, f.ect_count);
put_frame(&mut out, FrameType::Path, &body);
}
if let Some(f) = p.link {
put_frame(&mut out, FrameType::Link, &[f.class, f.quality]);
}
if let Some(f) = p.loss_acct {
body.clear();
put_varint(&mut body, f.seq as u64);
put_varint(&mut body, f.last_recv_seq as u64);
put_frame(&mut out, FrameType::LossAcct, &body);
}
if let Some(f) = p.pmtu {
body.clear();
put_varint(&mut body, f.pmtu as u64);
put_frame(&mut out, FrameType::Pmtu, &body);
}
for f in &p.bw_probe {
body.clear();
body.push(f.probe_id);
body.push(f.idx);
put_varint(&mut body, f.send_ts);
put_frame(&mut out, FrameType::BwProbe, &body);
}
for f in &p.trace {
put_frame(&mut out, FrameType::Trace, &[f.hop_ttl, f.probe_id]);
}
if let Some(f) = p.avail_bw {
body.clear();
put_varint(&mut body, f.avail_kbps);
put_varint(&mut body, f.capacity_kbps);
put_frame(&mut out, FrameType::AvailBw, &body);
}
if let Some(f) = p.forecast {
body.clear();
put_varint(&mut body, f.forecast_kbps);
put_frame(&mut out, FrameType::Forecast, &body);
}
if let Some(f) = p.periodicity {
body.clear();
put_varint(&mut body, f.period_ds);
put_varint(&mut body, f.secs_to_spike_ds);
body.push(f.confidence_x255);
put_frame(&mut out, FrameType::Periodicity, &body);
}
if let Some(f) = p.session_challenge {
body.clear();
put_varint(&mut body, u64::from(f.epoch));
put_varint(&mut body, f.nonce);
put_frame(&mut out, FrameType::SessionChallenge, &body);
}
if let Some(f) = p.session_response {
body.clear();
put_varint(&mut body, u64::from(f.epoch));
put_varint(&mut body, f.nonce);
put_frame(&mut out, FrameType::SessionResponse, &body);
}
if let Some(epoch) = p.session_announce {
body.clear();
put_varint(&mut body, u64::from(epoch));
put_frame(&mut out, FrameType::SessionAnnounce, &body);
}
out
}
pub fn decode_control(buf: &[u8]) -> Option<ControlPacket> {
if !is_control(buf) {
return None;
}
let mut p = ControlPacket::new();
let mut pos = 1usize;
while pos < buf.len() {
let ty = buf[pos];
pos += 1;
let (len, next) = match get_varint(buf, pos) {
Some(v) => v,
None => break,
};
pos = next;
let end = pos + len as usize;
if end > buf.len() {
break;
}
let body = &buf[pos..end];
match FrameType::from_u8(ty) {
Some(FrameType::Ack) => {
if let Some((v, _)) = get_varint(body, 0) {
p.ack = Some(AckFrame {
ack_through: v as u32,
});
}
}
Some(FrameType::Nak) => {
if let Some((block, q)) = get_varint(body, 0)
&& let Some((mask, _)) = get_varint(body, q)
{
p.nak = Some(NakFrame {
block: block as u32,
mask: mask as u32,
});
}
}
Some(FrameType::Loss) if body.len() >= 3 => {
p.loss = Some(LossFrame {
loss_x255: body[0],
burstiness_x255: body[1],
owd_trend_class: body[2],
loss_class: body.get(3).copied().unwrap_or(0),
});
}
Some(FrameType::Timing) => {
if let Some((send_ts, q)) = get_varint(body, 0)
&& let Some((echo_ts, _)) = get_varint(body, q)
{
p.timing = Some(TimingFrame { send_ts, echo_ts });
}
}
Some(FrameType::Ring) if body.len() >= 6 => {
p.ring = Some(RingFrame {
fill_pct: body[0],
ring_kind: body[1],
producers: body[2],
consumers: body[3],
trend: body[4],
flags: body[5],
});
}
Some(FrameType::Path) if body.len() >= 3 => {
let (ce_count, n1) = get_varint(body, 3).unwrap_or((0, 3));
let (ect_count, _) = get_varint(body, n1).unwrap_or((0, n1));
p.path = Some(PathFrame {
ttl: body[0],
ecn: body[1],
hop_count: body[2],
ce_count,
ect_count,
});
}
Some(FrameType::Link) if body.len() >= 2 => {
p.link = Some(LinkFrame {
class: body[0],
quality: body[1],
});
}
Some(FrameType::LossAcct) => {
if let Some((seq, n)) = get_varint(body, 0)
&& let Some((lrs, _)) = get_varint(body, n)
{
p.loss_acct = Some(LossAcctFrame {
seq: seq as u32,
last_recv_seq: lrs as u32,
});
}
}
Some(FrameType::Pmtu) => {
if let Some((v, _)) = get_varint(body, 0) {
p.pmtu = Some(PmtuFrame { pmtu: v as u16 });
}
}
Some(FrameType::BwProbe) if body.len() >= 2 => {
if let Some((send_ts, _)) = get_varint(body, 2) {
p.bw_probe.push(BwProbeFrame {
probe_id: body[0],
idx: body[1],
send_ts,
});
}
}
Some(FrameType::Trace) if body.len() >= 2 => {
p.trace.push(TraceFrame {
hop_ttl: body[0],
probe_id: body[1],
});
}
Some(FrameType::AvailBw) => {
if let Some((avail, n)) = get_varint(body, 0)
&& let Some((cap, _)) = get_varint(body, n)
{
p.avail_bw = Some(AvailBwFrame {
avail_kbps: avail,
capacity_kbps: cap,
});
}
}
Some(FrameType::SessionAnnounce) => {
if let Some((epoch, _)) = get_varint(body, 0) {
p.session_announce = Some(epoch as u32);
}
}
Some(FrameType::SessionChallenge) => {
if let Some((epoch, n)) = get_varint(body, 0)
&& let Some((nonce, _)) = get_varint(body, n)
{
p.session_challenge = Some(SessionFrame { epoch: epoch as u32, nonce });
}
}
Some(FrameType::SessionResponse) => {
if let Some((epoch, n)) = get_varint(body, 0)
&& let Some((nonce, _)) = get_varint(body, n)
{
p.session_response = Some(SessionFrame { epoch: epoch as u32, nonce });
}
}
Some(FrameType::Forecast) => {
if let Some((fc, _)) = get_varint(body, 0) {
p.forecast = Some(ForecastFrame { forecast_kbps: fc });
}
}
Some(FrameType::Periodicity) => {
if let Some((period, n1)) = get_varint(body, 0)
&& let Some((to_spike, n2)) = get_varint(body, n1)
&& n2 < body.len()
{
p.periodicity = Some(PeriodicityFrame {
period_ds: period,
secs_to_spike_ds: to_spike,
confidence_x255: body[n2],
});
}
}
_ => {}
}
pos = end;
}
Some(p)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn varint_round_trips_each_length_class() {
for v in [0u64, 1, 63, 64, 16383, 16384, (1 << 30) - 1, 1 << 30, (1u64 << 62) - 1] {
let mut b = Vec::new();
put_varint(&mut b, v);
let (got, end) = get_varint(&b, 0).expect("decode");
assert_eq!(got, v, "value {v} round-trip");
assert_eq!(end, b.len(), "consumed all bytes for {v}");
}
}
#[test]
fn varint_uses_minimal_encoding() {
let mut b = Vec::new();
put_varint(&mut b, 63);
assert_eq!(b.len(), 1, "6-bit value is one byte");
b.clear();
put_varint(&mut b, 64);
assert_eq!(b.len(), 2, "14-bit value is two bytes");
}
#[test]
fn full_packet_round_trips_every_frame() {
let p = ControlPacket {
ack: Some(AckFrame { ack_through: 70_000 }),
nak: Some(NakFrame {
block: 12,
mask: 0b1011,
}),
loss: Some(LossFrame {
loss_x255: 40,
burstiness_x255: 200,
owd_trend_class: 2,
loss_class: 2,
}),
timing: Some(TimingFrame {
send_ts: 1_234_567,
echo_ts: 1_234_000,
}),
ring: Some(RingFrame {
fill_pct: 30,
ring_kind: 1,
producers: 2,
consumers: 3,
trend: 1,
flags: 1,
}),
path: Some(PathFrame {
ttl: 53,
ecn: 0b11,
hop_count: 11,
ce_count: 4242,
ect_count: 99999,
}),
link: Some(LinkFrame {
class: link_class::WIFI,
quality: 180,
}),
loss_acct: Some(LossAcctFrame {
seq: 6000,
last_recv_seq: 5000,
}),
pmtu: Some(PmtuFrame { pmtu: 1280 }),
bw_probe: vec![
BwProbeFrame {
probe_id: 7,
idx: 0,
send_ts: 999,
},
BwProbeFrame {
probe_id: 7,
idx: 1,
send_ts: 1099,
},
],
trace: vec![TraceFrame {
hop_ttl: 5,
probe_id: 7,
}],
avail_bw: Some(AvailBwFrame {
avail_kbps: 45_000,
capacity_kbps: 100_000,
}),
forecast: Some(ForecastFrame {
forecast_kbps: 38_500,
}),
periodicity: Some(PeriodicityFrame {
period_ds: 150,
secs_to_spike_ds: 42,
confidence_x255: 200,
}),
session_challenge: Some(SessionFrame {
epoch: 0xDEAD_BEEF,
nonce: 0x0123_4567_89AB_CDEF,
}),
session_response: Some(SessionFrame {
epoch: 0xFEED_FACE,
nonce: 0xFEDC_BA98_7654_3210 & NONCE_MASK,
}),
session_announce: Some(0x1234_5678),
};
let wire = encode_control(&p);
assert_eq!(wire[0], PKT_CONTROL);
let got = decode_control(&wire).expect("decode");
assert_eq!(got, p, "full packet round-trips");
}
#[test]
fn padding_reaches_exact_size_and_still_decodes() {
let mut p = ControlPacket::new();
p.bw_probe.push(BwProbeFrame {
probe_id: 3,
idx: 1,
send_ts: 42,
});
let mut wire = encode_control(&p);
pad_control_to(&mut wire, 1400);
assert_eq!(wire.len(), 1400, "padded to the exact target size");
let got = decode_control(&wire).expect("decode");
assert_eq!(got.bw_probe, p.bw_probe, "the probe survives the padding");
assert!(got.avail_bw.is_none(), "the pad frame is skipped, not misread");
}
#[test]
fn empty_packet_is_just_the_tag() {
let p = ControlPacket::new();
assert!(p.is_empty());
let wire = encode_control(&p);
assert_eq!(wire, vec![PKT_CONTROL]);
assert_eq!(decode_control(&wire).unwrap(), p);
}
#[test]
fn unknown_frame_is_skipped_not_fatal() {
let mut wire = vec![PKT_CONTROL];
wire.push(FrameType::Ack as u8);
put_varint(&mut wire, 1);
wire.push(9); wire.push(0x7F); put_varint(&mut wire, 4);
wire.extend_from_slice(&[0xDE, 0xAD, 0xBE, 0xEF]);
wire.push(FrameType::Link as u8);
put_varint(&mut wire, 2);
wire.extend_from_slice(&[link_class::CELLULAR, 99]);
let p = decode_control(&wire).expect("decode");
assert_eq!(p.ack, Some(AckFrame { ack_through: 9 }));
assert_eq!(
p.link,
Some(LinkFrame {
class: link_class::CELLULAR,
quality: 99
})
);
}
#[test]
fn truncated_frame_length_aborts_cleanly() {
let mut wire = vec![PKT_CONTROL];
wire.push(FrameType::Ack as u8);
put_varint(&mut wire, 1);
wire.push(5);
wire.push(FrameType::Pmtu as u8);
put_varint(&mut wire, 10); wire.extend_from_slice(&[0x01, 0x02]);
let p = decode_control(&wire).expect("decode");
assert_eq!(p.ack, Some(AckFrame { ack_through: 5 }));
assert_eq!(p.pmtu, None, "truncated frame dropped");
}
#[test]
fn non_control_datagram_returns_none() {
assert!(decode_control(&[1, 2, 3]).is_none());
assert!(decode_control(&[]).is_none());
assert!(!is_control(&[2]));
assert!(is_control(&[PKT_CONTROL]));
}
}