1pub const PKT_CONTROL: u8 = 4;
32
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35#[repr(u8)]
36pub enum FrameType {
37 Ack = 0x01,
39 Nak = 0x02,
41 Loss = 0x03,
43 Timing = 0x04,
45 Ring = 0x05,
47 Path = 0x06,
49 Link = 0x07,
51 LossAcct = 0x08,
54 Pmtu = 0x09,
56 BwProbe = 0x0A,
58 Trace = 0x0B,
60 AvailBw = 0x0C,
63 Forecast = 0x0D,
66 Periodicity = 0x0E,
69}
70
71impl FrameType {
72 fn from_u8(v: u8) -> Option<Self> {
75 Some(match v {
76 0x01 => Self::Ack,
77 0x02 => Self::Nak,
78 0x03 => Self::Loss,
79 0x04 => Self::Timing,
80 0x05 => Self::Ring,
81 0x06 => Self::Path,
82 0x07 => Self::Link,
83 0x08 => Self::LossAcct,
84 0x09 => Self::Pmtu,
85 0x0A => Self::BwProbe,
86 0x0B => Self::Trace,
87 0x0C => Self::AvailBw,
88 0x0D => Self::Forecast,
89 0x0E => Self::Periodicity,
90 _ => return None,
91 })
92 }
93}
94
95#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
97pub struct AckFrame {
98 pub ack_through: u32,
99}
100
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
103pub struct NakFrame {
104 pub block: u32,
105 pub mask: u32,
106}
107
108#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
110pub struct LossFrame {
111 pub loss_x255: u8,
112 pub burstiness_x255: u8,
113 pub owd_trend_class: u8,
114 pub loss_class: u8,
118}
119
120#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
123pub struct TimingFrame {
124 pub send_ts: u64,
125 pub echo_ts: u64,
126}
127
128#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
130pub struct RingFrame {
131 pub fill_pct: u8,
132 pub ring_kind: u8,
133 pub producers: u8,
134 pub consumers: u8,
135 pub trend: u8,
136 pub flags: u8,
137}
138
139#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
148pub struct PathFrame {
149 pub ttl: u8,
150 pub ecn: u8,
151 pub hop_count: u8,
152 pub ce_count: u64,
153 pub ect_count: u64,
154}
155
156#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
159pub struct LinkFrame {
160 pub class: u8,
161 pub quality: u8,
162}
163
164pub mod link_class {
166 pub const UNKNOWN: u8 = 0;
167 pub const LOOPBACK: u8 = 1;
168 pub const WIRED: u8 = 2;
169 pub const WIFI: u8 = 3;
170 pub const CELLULAR: u8 = 4;
171}
172
173#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
180pub struct LossAcctFrame {
181 pub seq: u32,
182 pub last_recv_seq: u32,
183}
184
185#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
188pub struct PmtuFrame {
189 pub pmtu: u16,
190}
191
192#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
195pub struct BwProbeFrame {
196 pub probe_id: u8,
197 pub idx: u8,
198 pub send_ts: u64,
199}
200
201#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
205pub struct TraceFrame {
206 pub hop_ttl: u8,
207 pub probe_id: u8,
208}
209
210#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
214pub struct AvailBwFrame {
215 pub avail_kbps: u64,
217 pub capacity_kbps: u64,
220}
221
222#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
226pub struct ForecastFrame {
227 pub forecast_kbps: u64,
229}
230
231#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
236pub struct PeriodicityFrame {
237 pub period_ds: u64,
238 pub secs_to_spike_ds: u64,
239 pub confidence_x255: u8,
240}
241
242#[derive(Debug, Clone, Default, PartialEq, Eq)]
246pub struct ControlPacket {
247 pub ack: Option<AckFrame>,
248 pub nak: Option<NakFrame>,
249 pub loss: Option<LossFrame>,
250 pub timing: Option<TimingFrame>,
251 pub ring: Option<RingFrame>,
252 pub path: Option<PathFrame>,
253 pub link: Option<LinkFrame>,
254 pub loss_acct: Option<LossAcctFrame>,
255 pub pmtu: Option<PmtuFrame>,
256 pub bw_probe: Vec<BwProbeFrame>,
257 pub trace: Vec<TraceFrame>,
258 pub avail_bw: Option<AvailBwFrame>,
259 pub forecast: Option<ForecastFrame>,
260 pub periodicity: Option<PeriodicityFrame>,
261}
262
263impl ControlPacket {
264 pub fn new() -> Self {
266 Self::default()
267 }
268
269 pub fn is_empty(&self) -> bool {
271 self.ack.is_none()
272 && self.nak.is_none()
273 && self.loss.is_none()
274 && self.timing.is_none()
275 && self.ring.is_none()
276 && self.path.is_none()
277 && self.link.is_none()
278 && self.loss_acct.is_none()
279 && self.pmtu.is_none()
280 && self.bw_probe.is_empty()
281 && self.trace.is_empty()
282 }
283}
284
285pub fn is_control(buf: &[u8]) -> bool {
287 !buf.is_empty() && buf[0] == PKT_CONTROL
288}
289
290fn put_varint(out: &mut Vec<u8>, v: u64) {
296 const MAX62: u64 = (1 << 62) - 1;
297 let v = v.min(MAX62);
298 if v < (1 << 6) {
299 out.push(v as u8);
300 } else if v < (1 << 14) {
301 out.push(0x40 | (v >> 8) as u8);
302 out.push(v as u8);
303 } else if v < (1 << 30) {
304 out.push(0x80 | (v >> 24) as u8);
305 out.extend_from_slice(&(v as u32).to_be_bytes()[1..]);
306 } else {
307 out.push(0xC0 | (v >> 56) as u8);
308 out.extend_from_slice(&v.to_be_bytes()[1..]);
309 }
310}
311
312fn get_varint(buf: &[u8], pos: usize) -> Option<(u64, usize)> {
315 let first = *buf.get(pos)?;
316 let len = 1usize << (first >> 6);
317 if pos + len > buf.len() {
318 return None;
319 }
320 let mut v = (first & 0x3F) as u64;
321 for &b in &buf[pos + 1..pos + len] {
322 v = (v << 8) | b as u64;
323 }
324 Some((v, pos + len))
325}
326
327fn put_frame(out: &mut Vec<u8>, ty: FrameType, body: &[u8]) {
331 out.push(ty as u8);
332 put_varint(out, body.len() as u64);
333 out.extend_from_slice(body);
334}
335
336pub fn pad_control_to(buf: &mut Vec<u8>, target_len: usize) {
345 const PAD_TYPE: u8 = 0x7F;
346 const HEADER: usize = 3; if target_len < buf.len() + HEADER + 64 {
348 return;
349 }
350 let body_len = target_len - buf.len() - HEADER;
351 buf.push(PAD_TYPE);
352 put_varint(buf, body_len as u64);
353 buf.resize(buf.len() + body_len, 0);
354}
355
356pub fn encode_control(p: &ControlPacket) -> Vec<u8> {
358 let mut out = Vec::with_capacity(64);
359 out.push(PKT_CONTROL);
360 let mut body = Vec::with_capacity(16);
361
362 if let Some(f) = p.ack {
363 body.clear();
364 put_varint(&mut body, f.ack_through as u64);
365 put_frame(&mut out, FrameType::Ack, &body);
366 }
367 if let Some(f) = p.nak {
368 body.clear();
369 put_varint(&mut body, f.block as u64);
370 put_varint(&mut body, f.mask as u64);
371 put_frame(&mut out, FrameType::Nak, &body);
372 }
373 if let Some(f) = p.loss {
374 put_frame(
375 &mut out,
376 FrameType::Loss,
377 &[f.loss_x255, f.burstiness_x255, f.owd_trend_class, f.loss_class],
378 );
379 }
380 if let Some(f) = p.timing {
381 body.clear();
382 put_varint(&mut body, f.send_ts);
383 put_varint(&mut body, f.echo_ts);
384 put_frame(&mut out, FrameType::Timing, &body);
385 }
386 if let Some(f) = p.ring {
387 put_frame(
388 &mut out,
389 FrameType::Ring,
390 &[
391 f.fill_pct,
392 f.ring_kind,
393 f.producers,
394 f.consumers,
395 f.trend,
396 f.flags,
397 ],
398 );
399 }
400 if let Some(f) = p.path {
401 body.clear();
402 body.extend_from_slice(&[f.ttl, f.ecn, f.hop_count]);
403 put_varint(&mut body, f.ce_count);
404 put_varint(&mut body, f.ect_count);
405 put_frame(&mut out, FrameType::Path, &body);
406 }
407 if let Some(f) = p.link {
408 put_frame(&mut out, FrameType::Link, &[f.class, f.quality]);
409 }
410 if let Some(f) = p.loss_acct {
411 body.clear();
412 put_varint(&mut body, f.seq as u64);
413 put_varint(&mut body, f.last_recv_seq as u64);
414 put_frame(&mut out, FrameType::LossAcct, &body);
415 }
416 if let Some(f) = p.pmtu {
417 body.clear();
418 put_varint(&mut body, f.pmtu as u64);
419 put_frame(&mut out, FrameType::Pmtu, &body);
420 }
421 for f in &p.bw_probe {
422 body.clear();
423 body.push(f.probe_id);
424 body.push(f.idx);
425 put_varint(&mut body, f.send_ts);
426 put_frame(&mut out, FrameType::BwProbe, &body);
427 }
428 for f in &p.trace {
429 put_frame(&mut out, FrameType::Trace, &[f.hop_ttl, f.probe_id]);
430 }
431 if let Some(f) = p.avail_bw {
432 body.clear();
433 put_varint(&mut body, f.avail_kbps);
434 put_varint(&mut body, f.capacity_kbps);
435 put_frame(&mut out, FrameType::AvailBw, &body);
436 }
437 if let Some(f) = p.forecast {
438 body.clear();
439 put_varint(&mut body, f.forecast_kbps);
440 put_frame(&mut out, FrameType::Forecast, &body);
441 }
442 if let Some(f) = p.periodicity {
443 body.clear();
444 put_varint(&mut body, f.period_ds);
445 put_varint(&mut body, f.secs_to_spike_ds);
446 body.push(f.confidence_x255);
447 put_frame(&mut out, FrameType::Periodicity, &body);
448 }
449 out
450}
451
452pub fn decode_control(buf: &[u8]) -> Option<ControlPacket> {
457 if !is_control(buf) {
458 return None;
459 }
460 let mut p = ControlPacket::new();
461 let mut pos = 1usize;
462 while pos < buf.len() {
463 let ty = buf[pos];
464 pos += 1;
465 let (len, next) = match get_varint(buf, pos) {
466 Some(v) => v,
467 None => break,
468 };
469 pos = next;
470 let end = pos + len as usize;
471 if end > buf.len() {
472 break;
473 }
474 let body = &buf[pos..end];
475 match FrameType::from_u8(ty) {
476 Some(FrameType::Ack) => {
477 if let Some((v, _)) = get_varint(body, 0) {
478 p.ack = Some(AckFrame {
479 ack_through: v as u32,
480 });
481 }
482 }
483 Some(FrameType::Nak) => {
484 if let Some((block, q)) = get_varint(body, 0)
485 && let Some((mask, _)) = get_varint(body, q)
486 {
487 p.nak = Some(NakFrame {
488 block: block as u32,
489 mask: mask as u32,
490 });
491 }
492 }
493 Some(FrameType::Loss) if body.len() >= 3 => {
494 p.loss = Some(LossFrame {
495 loss_x255: body[0],
496 burstiness_x255: body[1],
497 owd_trend_class: body[2],
498 loss_class: body.get(3).copied().unwrap_or(0),
501 });
502 }
503 Some(FrameType::Timing) => {
504 if let Some((send_ts, q)) = get_varint(body, 0)
505 && let Some((echo_ts, _)) = get_varint(body, q)
506 {
507 p.timing = Some(TimingFrame { send_ts, echo_ts });
508 }
509 }
510 Some(FrameType::Ring) if body.len() >= 6 => {
511 p.ring = Some(RingFrame {
512 fill_pct: body[0],
513 ring_kind: body[1],
514 producers: body[2],
515 consumers: body[3],
516 trend: body[4],
517 flags: body[5],
518 });
519 }
520 Some(FrameType::Path) if body.len() >= 3 => {
521 let (ce_count, n1) = get_varint(body, 3).unwrap_or((0, 3));
524 let (ect_count, _) = get_varint(body, n1).unwrap_or((0, n1));
525 p.path = Some(PathFrame {
526 ttl: body[0],
527 ecn: body[1],
528 hop_count: body[2],
529 ce_count,
530 ect_count,
531 });
532 }
533 Some(FrameType::Link) if body.len() >= 2 => {
534 p.link = Some(LinkFrame {
535 class: body[0],
536 quality: body[1],
537 });
538 }
539 Some(FrameType::LossAcct) => {
540 if let Some((seq, n)) = get_varint(body, 0)
541 && let Some((lrs, _)) = get_varint(body, n)
542 {
543 p.loss_acct = Some(LossAcctFrame {
544 seq: seq as u32,
545 last_recv_seq: lrs as u32,
546 });
547 }
548 }
549 Some(FrameType::Pmtu) => {
550 if let Some((v, _)) = get_varint(body, 0) {
551 p.pmtu = Some(PmtuFrame { pmtu: v as u16 });
552 }
553 }
554 Some(FrameType::BwProbe) if body.len() >= 2 => {
555 if let Some((send_ts, _)) = get_varint(body, 2) {
556 p.bw_probe.push(BwProbeFrame {
557 probe_id: body[0],
558 idx: body[1],
559 send_ts,
560 });
561 }
562 }
563 Some(FrameType::Trace) if body.len() >= 2 => {
564 p.trace.push(TraceFrame {
565 hop_ttl: body[0],
566 probe_id: body[1],
567 });
568 }
569 Some(FrameType::AvailBw) => {
570 if let Some((avail, n)) = get_varint(body, 0)
571 && let Some((cap, _)) = get_varint(body, n)
572 {
573 p.avail_bw = Some(AvailBwFrame {
574 avail_kbps: avail,
575 capacity_kbps: cap,
576 });
577 }
578 }
579 Some(FrameType::Forecast) => {
580 if let Some((fc, _)) = get_varint(body, 0) {
581 p.forecast = Some(ForecastFrame { forecast_kbps: fc });
582 }
583 }
584 Some(FrameType::Periodicity) => {
585 if let Some((period, n1)) = get_varint(body, 0)
586 && let Some((to_spike, n2)) = get_varint(body, n1)
587 && n2 < body.len()
588 {
589 p.periodicity = Some(PeriodicityFrame {
590 period_ds: period,
591 secs_to_spike_ds: to_spike,
592 confidence_x255: body[n2],
593 });
594 }
595 }
596 _ => {}
598 }
599 pos = end;
600 }
601 Some(p)
602}
603
604#[cfg(test)]
605mod tests {
606 use super::*;
607
608 #[test]
609 fn varint_round_trips_each_length_class() {
610 for v in [0u64, 1, 63, 64, 16383, 16384, (1 << 30) - 1, 1 << 30, (1u64 << 62) - 1] {
611 let mut b = Vec::new();
612 put_varint(&mut b, v);
613 let (got, end) = get_varint(&b, 0).expect("decode");
614 assert_eq!(got, v, "value {v} round-trip");
615 assert_eq!(end, b.len(), "consumed all bytes for {v}");
616 }
617 }
618
619 #[test]
620 fn varint_uses_minimal_encoding() {
621 let mut b = Vec::new();
622 put_varint(&mut b, 63);
623 assert_eq!(b.len(), 1, "6-bit value is one byte");
624 b.clear();
625 put_varint(&mut b, 64);
626 assert_eq!(b.len(), 2, "14-bit value is two bytes");
627 }
628
629 #[test]
630 fn full_packet_round_trips_every_frame() {
631 let p = ControlPacket {
632 ack: Some(AckFrame { ack_through: 70_000 }),
633 nak: Some(NakFrame {
634 block: 12,
635 mask: 0b1011,
636 }),
637 loss: Some(LossFrame {
638 loss_x255: 40,
639 burstiness_x255: 200,
640 owd_trend_class: 2,
641 loss_class: 2,
642 }),
643 timing: Some(TimingFrame {
644 send_ts: 1_234_567,
645 echo_ts: 1_234_000,
646 }),
647 ring: Some(RingFrame {
648 fill_pct: 30,
649 ring_kind: 1,
650 producers: 2,
651 consumers: 3,
652 trend: 1,
653 flags: 1,
654 }),
655 path: Some(PathFrame {
656 ttl: 53,
657 ecn: 0b11,
658 hop_count: 11,
659 ce_count: 4242,
660 ect_count: 99999,
661 }),
662 link: Some(LinkFrame {
663 class: link_class::WIFI,
664 quality: 180,
665 }),
666 loss_acct: Some(LossAcctFrame {
667 seq: 6000,
668 last_recv_seq: 5000,
669 }),
670 pmtu: Some(PmtuFrame { pmtu: 1280 }),
671 bw_probe: vec![
672 BwProbeFrame {
673 probe_id: 7,
674 idx: 0,
675 send_ts: 999,
676 },
677 BwProbeFrame {
678 probe_id: 7,
679 idx: 1,
680 send_ts: 1099,
681 },
682 ],
683 trace: vec![TraceFrame {
684 hop_ttl: 5,
685 probe_id: 7,
686 }],
687 avail_bw: Some(AvailBwFrame {
688 avail_kbps: 45_000,
689 capacity_kbps: 100_000,
690 }),
691 forecast: Some(ForecastFrame {
692 forecast_kbps: 38_500,
693 }),
694 periodicity: Some(PeriodicityFrame {
695 period_ds: 150,
696 secs_to_spike_ds: 42,
697 confidence_x255: 200,
698 }),
699 };
700 let wire = encode_control(&p);
701 assert_eq!(wire[0], PKT_CONTROL);
702 let got = decode_control(&wire).expect("decode");
703 assert_eq!(got, p, "full packet round-trips");
704 }
705
706 #[test]
707 fn padding_reaches_exact_size_and_still_decodes() {
708 let mut p = ControlPacket::new();
709 p.bw_probe.push(BwProbeFrame {
710 probe_id: 3,
711 idx: 1,
712 send_ts: 42,
713 });
714 let mut wire = encode_control(&p);
715 pad_control_to(&mut wire, 1400);
716 assert_eq!(wire.len(), 1400, "padded to the exact target size");
717 let got = decode_control(&wire).expect("decode");
718 assert_eq!(got.bw_probe, p.bw_probe, "the probe survives the padding");
719 assert!(got.avail_bw.is_none(), "the pad frame is skipped, not misread");
720 }
721
722 #[test]
723 fn empty_packet_is_just_the_tag() {
724 let p = ControlPacket::new();
725 assert!(p.is_empty());
726 let wire = encode_control(&p);
727 assert_eq!(wire, vec![PKT_CONTROL]);
728 assert_eq!(decode_control(&wire).unwrap(), p);
729 }
730
731 #[test]
732 fn unknown_frame_is_skipped_not_fatal() {
733 let mut wire = vec![PKT_CONTROL];
737 wire.push(FrameType::Ack as u8);
738 put_varint(&mut wire, 1);
739 wire.push(9); wire.push(0x7F); put_varint(&mut wire, 4);
742 wire.extend_from_slice(&[0xDE, 0xAD, 0xBE, 0xEF]);
743 wire.push(FrameType::Link as u8);
744 put_varint(&mut wire, 2);
745 wire.extend_from_slice(&[link_class::CELLULAR, 99]);
746
747 let p = decode_control(&wire).expect("decode");
748 assert_eq!(p.ack, Some(AckFrame { ack_through: 9 }));
749 assert_eq!(
750 p.link,
751 Some(LinkFrame {
752 class: link_class::CELLULAR,
753 quality: 99
754 })
755 );
756 }
757
758 #[test]
759 fn truncated_frame_length_aborts_cleanly() {
760 let mut wire = vec![PKT_CONTROL];
763 wire.push(FrameType::Ack as u8);
764 put_varint(&mut wire, 1);
765 wire.push(5);
766 wire.push(FrameType::Pmtu as u8);
767 put_varint(&mut wire, 10); wire.extend_from_slice(&[0x01, 0x02]);
769 let p = decode_control(&wire).expect("decode");
770 assert_eq!(p.ack, Some(AckFrame { ack_through: 5 }));
771 assert_eq!(p.pmtu, None, "truncated frame dropped");
772 }
773
774 #[test]
775 fn non_control_datagram_returns_none() {
776 assert!(decode_control(&[1, 2, 3]).is_none());
777 assert!(decode_control(&[]).is_none());
778 assert!(!is_control(&[2]));
779 assert!(is_control(&[PKT_CONTROL]));
780 }
781}