From 803eb1064230e8ac566861cf56d2084418434de6 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 18:48:37 -0400 Subject: [PATCH 01/14] Spec the adaptive loss-feedback loop + reactive ARQ Closes the data-plane control loop: an authenticated receiver->sender Control packet carrying delivered-count + missing object counters drives both (1) the class-aware controller (bulk/default repair ratio earns zero on clean flows, activating the dormant FEC bypass + halving datagrams; realtime keeps a floor) and (2) deadline/class-aware reactive ARQ (sender retransmits fresh RaptorQ repair symbols for eligible NACKed objects). Receiver reports raw gaps; sender owns all class attribution and retransmit eligibility. Plan sequences feedback first (throughput checkpoint), then ARQ. Co-Authored-By: Claude Opus 4.8 --- ...26-06-30-data-plane-feedback-arq-design.md | 217 ++++++++++++++++++ 1 file changed, 217 insertions(+) create mode 100644 docs/superpowers/specs/2026-06-30-data-plane-feedback-arq-design.md diff --git a/docs/superpowers/specs/2026-06-30-data-plane-feedback-arq-design.md b/docs/superpowers/specs/2026-06-30-data-plane-feedback-arq-design.md new file mode 100644 index 0000000..e4c6cdb --- /dev/null +++ b/docs/superpowers/specs/2026-06-30-data-plane-feedback-arq-design.md @@ -0,0 +1,217 @@ +# Adaptive loss-feedback loop + reactive ARQ — design + +**Status:** approved (brainstorming, 2026-06-30) +**Scope:** sub-project #1 (core data plane), adaptive-control milestone +**Predecessors:** M1–M6 + M5.5, the benchmark harness (PR #1), and the throughput +pass (PR #2, `c492b5d`) which shipped a *dormant* zero-repair FEC bypass. + +## Goal + +Close the data-plane control loop so the receiver tells the sender what it +actually lost, and the sender adapts. This (a) **activates the throughput win** +the throughput pass set up — clean bulk flows drive the repair ratio to zero, +firing the merged FEC-encode bypass and halving per-packet datagram count — and +(b) adds **reactive ARQ** so the rare loss that escapes proactive FEC is +retransmitted for flows that can still use it. + +## Success criteria + +- On a clean link, a bulk/default flow's repair ratio reaches **zero** (verified: + the FEC bypass fires, ~1 symbol/packet instead of ~2), and **clean-link + throughput measurably rises** vs the post-throughput-pass baseline + (~220–285 Mbit/s single-stream). +- Under `tc netem` loss, a bulk flow's **post-FEC residual loss stays low** and + objects FEC cannot recover are **retransmitted and delivered** (ARQ works). +- A **realtime** flow keeps a small proactive repair floor (never zero) and is + **never** retransmitted (deadline-aware) — its latency profile is unchanged. +- Forged feedback/NACK packets are **rejected** (control packets authenticated). +- All new state is **bounded** (retransmit buffer, pending-loss set, NACK list). +- All existing netns ping/tunnel tests stay green; the data-plane *symbol* wire + frame is unchanged (only an additive new control packet type is introduced). + +## Background — current state + +- `yip_transport::AdaptiveController::observe_loss(loss)` exists and is sensible + (AIMD: jumps ratio to `loss+0.05` on loss, decays 10%/observation when clean) + but **the daemon never calls it** — so the ratio is static at + `initial_repair_ratio`. `repair_count` floors at `max(1)` and `min_ratio = + initial_repair_ratio`, so repair never reaches zero. Hence the throughput + pass's zero-repair bypass is dormant. +- The wire symbol payload carries a monotonic per-object `counter` (one + `Session::seal` per TUN packet → +1 per object, contiguous). The object's + `FlowClass` rides in the authenticated `Frame.flags`. +- `PacketType` in `bin/yipd/src/handshake.rs`: `HandshakeInit=0`, + `HandshakeResp=1`, `Data=2`. The 1-byte prefix is temporary anti-DPI debt + (folds into keyed header-protection in sub-project #3); a new control type + rides the same scheme. +- `Transport::decode` already keys reassembly per object and tolerates + reorder/loss; `FlowParams` carries a per-class `deadline` (currently only + partially honored). + +## Design + +### 1. Shared authenticated control packet (`PacketType::Control = 3`) + +A single receiver→sender packet, **sealed with the session AEAD** (same channel +as data — a forged report otherwise lets an attacker force max redundancy or +strip FEC). It is the only new wire object. Payload (pre-seal): + +- `delivered_count: u32` — objects delivered in the reporting window (for rate). +- `high_counter: u64` — highest object counter the receiver has observed. +- `missing: Vec` — object counters seen-as-gaps and declared lost + (bounded, ≤ `MAX_NACK` per packet; if more are pending, send the oldest and + set a `truncated` flag so the sender treats it as high-loss). + +One packet feeds **both** consumers (controller tuning + ARQ); the receiver does +**no** per-class attribution (it cannot know the class of a fully-lost object). +Sent on a periodic heartbeat (`FEEDBACK_INTERVAL_MS`) **and** opportunistically +the moment new losses are detected. + +Indicative constants (pinned in the plan, tuned against the bench): `FEEDBACK_ +INTERVAL_MS ≈ 20–50`, `GRACE_MS ≈ 5` (≈2×RTT on the target paths), `MAX_NACK ≈ +64` counters/packet, `realtime_floor ≈ 0.05`, retransmit buffer ≈ 1024 objects / +2 s TTL. These are starting points, not load-bearing invariants. + +### 2. Receiver-side loss detection (gap-based) + +Per direction the receiver keeps a bounded structure: + +- `delivered`: advances as objects decode; tracks the contiguous-delivered + watermark and a bounded set of counters seen-but-not-complete (with first-seen + time). +- When object `C` decodes → mark delivered; drop from pending; advance watermark. +- A counter `C < high_counter` that has not completed within `GRACE_MS` (a few ms + / ~2×RTT) is **declared lost**: added to `missing` and counted toward + `delivered_count`'s denominator. +- Gap discovery: when symbols for counter `N` arrive and `N > high_counter+1`, + the skipped counters `(high_counter, N)` are *candidates* — confirmed lost only + after `GRACE_MS` without completion (handles reorder). With the zero-repair + bypass a lost single-symbol object appears **only** as such a gap, so gap + detection (not partial-object state) is the load-bearing signal. + +The receiver clears a `missing` entry once it has been reported (tracking +reported counters to avoid re-NACKing), and re-reports only if still unfilled +after a retransmit window (bounded retries). + +### 3. Sender-side attribution + class-aware controller + +On receiving a (decrypted, authenticated) control packet, the sender — which +holds each sent object's class in its retransmit buffer (§4) — does **all** +attribution: + +- For each `missing` counter, look up its `FlowClass` in the sent buffer and + tally per-class residual loss. Per-class loss fraction = + `class_missing / class_sent` over the window → `Transport::observe_loss(class, + frac)` for each active class (zero for classes with no reported loss → decay). +- **Class-aware zero floor.** Relax the controller so the ratio can decay to + **zero for `Bulk`/`Default`** (they have the ARQ backstop) but only to a small + **`realtime_floor` (> 0) for `Realtime`** (no ARQ; needs latency-free masking). + Concretely: make `min_ratio` class-derived (0 for bulk/default, a small floor + for realtime) and drop `repair_count`'s unconditional `max(1)` — it returns 0 + only when the (class-aware) ratio is 0, else ≥1. `observe_loss` already snaps + the ratio up to `loss+0.05` on any loss, so zero-repair is *earned* by clean + feedback and *abandoned instantly* on loss. This is what fires the dormant + bypass and halves clean bulk datagrams. + +### 4. Reactive ARQ — deadline/class-aware retransmit + +- **Sender retransmit buffer:** bounded (LRU + TTL), keyed by object `counter`, + holding `{ sealed_ciphertext, class, sent_at }`. Populated as the egress loop + sends each object. +- **On NACK (a `missing` counter):** retransmit **iff** the object is still + buffered **and** its class is retransmit-eligible (`Bulk`/`Default`, not + `Realtime`) **and** it is within its deadline (`sent_at + FlowParams.deadline > + now`). Otherwise ignore the NACK. The realtime/deadline filtering lives here, + on the sender (which knows the class), not on the receiver. +- **Retransmit mechanism:** generate **fresh RaptorQ repair symbols** for the + buffered object (rateless — idiomatic; not a blind resend) and send them under + the original `counter`. They top up the receiver's existing decoder for that + object. If the receiver already evicted that object's decode state, the + retransmit must carry enough symbols to complete from scratch (send source + + repair on retransmit). +- **Receiver dedup:** a retransmit that arrives after the object already + completed (counter ≤ watermark / already delivered) is dropped — the existing + `decode` "late symbol returns None after completion" path already covers this; + verify it holds for retransmitted symbols. + +### 5. Security & anti-DPI + +Control packets are sealed/authenticated exactly like data (a forged or replayed +report is rejected by the AEAD + replay window). Every new structure is bounded: +retransmit buffer (LRU+TTL), receiver pending-loss set, per-packet `missing` cap. +The new `Control=3` rides the existing temporary `PacketType` prefix; removing +that prefix (keyed header-protection) remains sub-project #3 and is unaffected. + +### 6. Testing + +- **Controller (unit):** decays to 0 for bulk/default under sustained clean + feedback; decays only to the floor for realtime; snaps to `loss+0.05` on a loss + report; `repair_count` returns 0 iff the class-aware ratio is 0. +- **Loss detector (unit):** from a scripted gappy counter stream (with reorder + inside `GRACE_MS`), computes the correct `missing` set and `delivered_count`; + reorder within grace is **not** falsely reported; bounded pending set. +- **Retransmit buffer (unit):** bounded LRU+TTL; lookup-by-counter; eviction. +- **ARQ round-trip (unit/integration):** a NACK for a buffered bulk object → + fresh repair symbols → receiver completes; a NACK for a realtime object or an + expired-deadline object → no retransmit; duplicate/late retransmit deduped. +- **End-to-end (netns + netem):** (a) clean link → bulk flow ratio reaches 0, + bypass fires (assert ~1 symbol/packet), **clean-link iperf throughput rises** + vs the throughput-pass baseline; (b) under 5–10% netem loss → bulk residual + stays low, ARQ-recovered objects delivered, realtime flow keeps its floor and + is not retransmitted. Wire-compat: existing `tunnel_netns` ping stays green. + +## Components touched + +- `crates/yip-transport`: `control.rs` (class-aware floor, `repair_count` 0-path); + `lib.rs` (`observe_loss` plumbing, retransmit-symbol generation API); a new + loss-detector module (receiver) and retransmit-buffer module (sender), or + folded into existing files if small. +- `crates/yip-wire` or `bin/yipd`: the `Control` packet (de)serialization of + `{ delivered_count, high_counter, missing[] }`. +- `bin/yipd/src/`: `handshake.rs` (`PacketType::Control = 3`); `tunnel.rs` + (egress populates the retransmit buffer; ingress runs the loss detector + emits + control packets; a control-packet handler that decrypts, attributes loss to + `observe_loss`, and retransmits eligible NACKs); a periodic feedback timer. +- `crates/yip-bench`: an end-to-end test asserting clean-link ratio→0 + throughput + rise, and ARQ recovery under loss. + +## Wire compatibility + +Additive: the data-plane *symbol* frame is unchanged; only a new `Control=3` +packet type is added. yip has no deployed third-party peers (pre-release, both +ends are this codebase), so the protocol extension is safe; a peer that did not +understand `Control` would simply drop it (graceful degradation of feedback/ARQ, +data still flows). The anti-DPI prefix story is unchanged (sub-project #3). + +## Implementation sequencing + +Though one cohesive system, the plan should build it in two ordered phases with a +working checkpoint between them: **(A) control channel + loss feedback + +class-aware zero-repair** — this alone activates the throughput win and is +independently testable; then **(B) reactive ARQ** on the same control packet +(retransmit buffer + NACK handling + dedup). The control packet is designed once +(carrying `missing[]` from the start) so phase B adds no wire change. + +## Out of scope (deferred) + +- The unified io_uring busy-poll rewrite (separate; throughput-pass deferred it). +- Removing the `PacketType` prefix / keyed header-protection (sub-project #3). +- Full deadline-based FEC *eviction* beyond what ARQ eligibility needs. +- Multipath / FEC across objects / congestion control. +- L2/TAP path; rekey; PQ handshake. + +## Risks + +- **Re-arm window exposure.** A bulk flow at zero repair loses the ~1 RTT of + packets sent before feedback re-arms FEC. Mitigated by ARQ (those losses are + retransmitted) and the instant `loss+0.05` snap-up. Realtime never goes to zero, + so it is never exposed. +- **Gap mis-detection under reorder.** Mitigated by the `GRACE_MS` confirmation + window before declaring a counter lost. +- **Retransmit storms / amplification.** Bounded `missing` per packet, bounded + retransmit buffer, bounded re-NACK retries; a NACK for an evicted/expired object + is ignored. +- **Forged reports.** Mitigated by sealing control packets (AEAD + replay window). +- **Counter attribution after buffer eviction.** A `missing` counter no longer in + the sender's buffer is treated as loss-for-stats but not retransmitted (it is + too old to matter) — bounded behavior, no error. From e7c64804b4d08dde37ee06187cefa2fa345aebca Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 18:59:43 -0400 Subject: [PATCH 02/14] Plan the loss-feedback loop + reactive ARQ 8 tasks in two phases. Phase A (control channel + feedback + class-aware zero-repair): LossReport (de)serialization, gap-based loss detector, controller that lets ARQ classes earn ratio 0, daemon control-channel wiring, and the Phase-A throughput verdict. Phase B (reactive ARQ): bounded retransmit buffer, retransmit-on-NACK with fresh repair symbols reusing the original object_id, plus the final measurement. Counter is one unified per-direction sequence; sender owns class attribution; ARQ + zero-repair gated on FlowParams.arq. Co-Authored-By: Claude Opus 4.8 --- .../2026-06-30-data-plane-feedback-arq.md | 354 ++++++++++++++++++ 1 file changed, 354 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-30-data-plane-feedback-arq.md diff --git a/docs/superpowers/plans/2026-06-30-data-plane-feedback-arq.md b/docs/superpowers/plans/2026-06-30-data-plane-feedback-arq.md new file mode 100644 index 0000000..964e698 --- /dev/null +++ b/docs/superpowers/plans/2026-06-30-data-plane-feedback-arq.md @@ -0,0 +1,354 @@ +# Adaptive loss-feedback loop + reactive ARQ Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Close the data-plane control loop — the receiver reports loss, the sender adapts the repair ratio (down to zero on clean ARQ-backed flows, firing the dormant FEC bypass) and retransmits NACKed objects for eligible flows. + +**Architecture:** A single authenticated receiver→sender `Control` packet carries `{ delivered_count, high_counter, missing[] }`. The receiver detects loss as gaps in the monotonic per-direction sealed counter. The sender owns all class attribution (it knows each counter's class from its sent log): it feeds per-class residual loss to the controller and retransmits eligible NACKed objects with fresh RaptorQ repair symbols. Built in two phases: (A) control channel + feedback + class-aware zero-repair (the throughput win), then (B) reactive ARQ. + +**Tech Stack:** Rust, `raptorq` 2.0, `snow`/ChaCha20-Poly1305 (existing `Session`), the existing `yip-transport` controller/FEC, `bin/yipd` tunnel loops. + +## Global Constraints + +- **Counter is one unified monotonic per-direction sequence over ALL sealed packets** (data objects AND control packets). Data objects stay uniquely identified; control packets consume sequence numbers too. The receiver reports raw gaps; the **sender** disambiguates every gap via its sent log (data-object-of-class-X vs control) — so a lost control packet yields a NACK the sender safely ignores, and per-class loss attribution is exact. +- **ARQ-eligibility AND zero-repair-eligibility are both gated on `FlowParams.arq`** (currently `Bulk=true`, `Realtime=false`, `Default=false`). ARQ-classes can earn ratio 0; non-ARQ classes keep a `> 0` floor. (To later let `Default` reach zero, flip its `arq` flag — one line; out of scope here.) +- **Control packets are authenticated** via the existing `Session` AEAD (seal/open) — a forged/replayed report is rejected by the AEAD + replay window. No new key material. +- **Everything bounded:** retransmit buffer (LRU + TTL), receiver pending-loss set, `missing` per packet (`MAX_NACK`). +- **Additive wire change only:** the data-plane *symbol* frame is unchanged; only `PacketType::Control = 3` is added. The `tunnel_netns` ping test is the regression gate. +- Mullvad lints, `-D warnings`; no `as` numeric casts except the existing `PacketType::* as u8` idiom; `#![forbid(unsafe_code)]` on every crate except `yip-io`; exact dependency pins; ≥90 % coverage on `yip-transport`; `CHANGELOG.md` per Keep a Changelog. +- Indicative constants (tune against the bench): `FEEDBACK_INTERVAL_MS = 30`, `GRACE_MS = 5`, `MAX_NACK = 64`, realtime/default floor = each class's `initial_repair_ratio`, retransmit buffer = 1024 objects / 2 s TTL. + +--- + +## Phase A — control channel + feedback + class-aware zero-repair + +### Task 1: `Control` report type + (de)serialization + +A plain serializable report, independent of crypto/wire — pure data + bytes. + +**Files:** +- Create: `crates/yip-transport/src/feedback.rs` +- Modify: `crates/yip-transport/src/lib.rs` (add `pub mod feedback;` and re-export) +- Test: in `feedback.rs` `#[cfg(test)]` + +**Interfaces:** +- Produces: `pub struct LossReport { pub delivered_count: u32, pub high_counter: u64, pub missing: Vec }` with `pub fn encode(&self) -> Vec` and `pub fn decode(bytes: &[u8]) -> Option`. Wire layout: `[delivered_count:4 BE][high_counter:8 BE][n_missing:2 BE][missing: n×8 BE]`. `decode` returns `None` on short/inconsistent input (untrusted bytes). `encode` caps `missing` at `MAX_NACK` (defined `pub const MAX_NACK: usize = 64;`). + +- [ ] **Step 1: Write the round-trip + bounds tests** + +```rust +#[test] +fn loss_report_roundtrips() { + let r = LossReport { delivered_count: 1000, high_counter: 5_000, missing: vec![10, 42, 4_999] }; + let bytes = r.encode(); + let got = LossReport::decode(&bytes).expect("decodes"); + assert_eq!(got.delivered_count, 1000); + assert_eq!(got.high_counter, 5_000); + assert_eq!(got.missing, vec![10, 42, 4_999]); +} + +#[test] +fn loss_report_decode_rejects_short_input() { + assert!(LossReport::decode(&[]).is_none()); + assert!(LossReport::decode(&[0u8; 5]).is_none()); // shorter than the 14-byte header +} + +#[test] +fn loss_report_encode_caps_missing_at_max_nack() { + let r = LossReport { delivered_count: 0, high_counter: 0, missing: (0..1000).collect() }; + let got = LossReport::decode(&r.encode()).expect("decodes"); + assert_eq!(got.missing.len(), MAX_NACK); +} +``` + +- [ ] **Step 2: Run to verify it fails** + +Run: `cargo test -p yip-transport loss_report -v` +Expected: FAIL — `LossReport` not defined. + +- [ ] **Step 3: Implement `LossReport`, `encode`, `decode`, `MAX_NACK`** + +Big-endian throughout; `decode` validates `bytes.len() >= 14` and `bytes.len() == 14 + n_missing*8`. No `as` casts (use `u32::from_be_bytes`, `usize::from`, `u16::try_from`). + +- [ ] **Step 4: Run tests** — `cargo test -p yip-transport loss_report -v` → PASS. + +- [ ] **Step 5: Commit** — `git commit -m "Add LossReport control-packet (de)serialization"` + +--- + +### Task 2: Receiver-side gap loss detector + +Consumes the stream of delivered/seen counters; emits a `LossReport`. Pure logic, no I/O. + +**Files:** +- Create: `crates/yip-transport/src/lossdetect.rs` +- Modify: `crates/yip-transport/src/lib.rs` (`pub mod lossdetect;`) +- Test: in `lossdetect.rs` `#[cfg(test)]` + +**Interfaces:** +- Consumes: `LossReport`, `MAX_NACK` (Task 1). +- Produces: `pub struct LossDetector { .. }` with: + - `pub fn new(grace_ms: u64, window: usize) -> Self` + - `pub fn on_seen(&mut self, counter: u64, now_ms: u64)` — call for every received sealed packet's counter (data or control), before knowing if its object decodes. + - `pub fn on_delivered(&mut self, counter: u64)` — call when an object for `counter` fully decodes (so it is never reported missing). + - `pub fn report(&mut self, now_ms: u64) -> LossReport` — declares as missing any counter `< high_counter` first-seen-or-implied older than `grace_ms` and not delivered; advances internal watermark; returns the report and clears reported counters. `delivered_count` = objects delivered since the last `report`. Bounds the pending set to `window` (drops oldest). + +Gap model: track `high_counter` (max seen). `on_seen(c)` records `c` and, if `c > high_counter + 1`, marks the skipped range `(prev_high, c)` as *implied-pending* with timestamp `now_ms`. `on_delivered(c)` removes `c` from pending and counts a delivery. `report` promotes implied-pending entries older than `grace_ms` (still not delivered) to `missing`. + +- [ ] **Step 1: Write the detector tests** + +```rust +#[test] +fn contiguous_no_loss() { + let mut d = LossDetector::new(5, 1024); + for c in 0..100u64 { d.on_seen(c, 0); d.on_delivered(c); } + let r = d.report(100); + assert!(r.missing.is_empty()); + assert_eq!(r.delivered_count, 100); +} + +#[test] +fn gap_reported_after_grace() { + let mut d = LossDetector::new(5, 1024); + // see 0,1,3 (2 is a gap); deliver the ones we saw + for c in [0u64, 1, 3] { d.on_seen(c, 0); d.on_delivered(c); } + // before grace elapses, 2 is not yet declared + assert!(d.report(3).missing.is_empty()); + // after grace, 2 is missing + let r = d.report(10); + assert_eq!(r.missing, vec![2]); +} + +#[test] +fn reorder_within_grace_not_reported() { + let mut d = LossDetector::new(5, 1024); + d.on_seen(0, 0); d.on_delivered(0); + d.on_seen(2, 0); d.on_delivered(2); // 1 appears skipped... + d.on_seen(1, 2); d.on_delivered(1); // ...but arrives within grace + let r = d.report(10); + assert!(r.missing.is_empty(), "in-grace reorder must not be reported lost"); +} + +#[test] +fn missing_set_is_bounded() { + let mut d = LossDetector::new(0, 8); // grace 0 → immediate; window 8 + d.on_seen(0, 0); + d.on_seen(1000, 1); // implies a huge gap + let r = d.report(5); + assert!(r.missing.len() <= 8 + 64); // bounded by window and MAX_NACK at encode +} +``` + +- [ ] **Step 2: Run to verify it fails** — `cargo test -p yip-transport -- lossdetect` → FAIL (undefined). + +- [ ] **Step 3: Implement `LossDetector`** per the gap model. Use a `BTreeMap` (counter → first-implied ms) for pending, a `delivered: u32` counter reset each `report`, and bound the pending map to `window` (evict smallest keys). No `as` casts. + +- [ ] **Step 4: Run tests** — all PASS. + +- [ ] **Step 5: Commit** — `git commit -m "Add gap-based receiver loss detector"` + +--- + +### Task 3: Class-aware controller — let ARQ classes earn ratio 0 + +**Files:** +- Modify: `crates/yip-transport/src/control.rs` (`AdaptiveController`) +- Modify: `crates/yip-transport/src/lib.rs` (`Transport::observe_loss` already exists; confirm it forwards per-class) +- Test: `control.rs` `#[cfg(test)]` + +**Interfaces:** +- Changes `AdaptiveController::new` to take the class's `arq` flag (or a `floor: f32`): for ARQ classes `min_ratio = 0.0`; for non-ARQ classes `min_ratio = initial_repair_ratio` (unchanged). `repair_count` returns `0` when the ratio rounds to 0 (drop the unconditional `.max(1)`; instead `max(1)` only when `ratio > 0.0`). Keep `observe_loss`'s snap-up (`loss+0.05`) and 10 % clean decay. +- Produces: `repair_count(source) == 0` iff the class is ARQ-eligible and has decayed clean to ratio 0; `>= 1` otherwise. + +- [ ] **Step 1: Write the tests** + +```rust +#[test] +fn arq_class_decays_to_zero_when_clean() { + let mut c = AdaptiveController::new_for(FlowClass::Bulk.params()); // arq=true + for _ in 0..200 { c.observe_loss(0.0); } + assert_eq!(c.ratio(), 0.0, "bulk earns zero repair on a clean link"); + assert_eq!(c.repair_count(1), 0, "zero ratio -> zero repair -> bypass fires"); + assert_eq!(c.repair_count(10), 0); +} + +#[test] +fn non_arq_class_keeps_floor() { + let mut c = AdaptiveController::new_for(FlowClass::Realtime.params()); // arq=false + for _ in 0..200 { c.observe_loss(0.0); } + assert!(c.ratio() >= FlowClass::Realtime.params().initial_repair_ratio - 1e-6); + assert!(c.repair_count(1) >= 1, "realtime keeps proactive repair"); +} + +#[test] +fn snaps_up_on_loss_even_from_zero() { + let mut c = AdaptiveController::new_for(FlowClass::Bulk.params()); + for _ in 0..200 { c.observe_loss(0.0); } + assert_eq!(c.ratio(), 0.0); + c.observe_loss(0.10); + assert!(c.ratio() >= 0.10, "any loss re-arms FEC immediately"); + assert!(c.repair_count(10) >= 1); +} +``` + +- [ ] **Step 2: Run to verify it fails** — FAIL (`new_for` undefined / floor not applied). + +- [ ] **Step 3: Implement.** Add `pub fn new_for(params: FlowParams) -> Self` setting `min_ratio = if params.arq { 0.0 } else { params.initial_repair_ratio }`. Update `repair_count`: compute `n`; `if self.ratio <= f32::EPSILON { 0 } else { n.max(1) }`. Keep the existing `new` (or route it through `new_for`). Update `Transport::new` to build controllers via `new_for`. + +- [ ] **Step 4: Run tests** — `cargo test -p yip-transport` all PASS (existing controller tests included; adjust any that assumed `max(1)` for a clean bulk flow). + +- [ ] **Step 5: Commit** — `git commit -m "Let ARQ-class repair ratio decay to zero; keep floor for non-ARQ"` + +--- + +### Task 4: Daemon control channel + feedback wiring (Phase A) + +Wire the pieces into yipd: send/receive `Control` packets, run the detector, feed `observe_loss`. Establishes the sent log (counter→class) reused by ARQ. + +**Files:** +- Modify: `bin/yipd/src/handshake.rs` (`PacketType::Control = 3`) +- Modify: `bin/yipd/src/tunnel.rs` (egress sent-log; ingress detector + control emit; control handler) +- Reference: egress seals per object (`session.seal` → `sealed.counter`), then `transport.encode`; ingress `recv_batch` → `frame_to_symbol` (yields counter+class) → `transport.decode` → on `Some` the object delivered. + +**Interfaces:** +- Consumes: `LossReport`/`encode`/`decode` (T1), `LossDetector` (T2), `Transport::observe_loss` (T3), `Session::seal/open`. +- Produces: yipd sends a sealed `Control` packet (`[PacketType::Control][counter:8][ciphertext]`) every `FEEDBACK_INTERVAL_MS` and on new loss; on receiving one, decrypts, and for each `missing` counter looks up its class in the **sent log** (`HashMap` or the retransmit buffer) to call `observe_loss(class, per_class_fraction)`. A `missing` counter not in the sent log (or a control counter) is ignored for attribution. + +- [ ] **Step 1: Baseline gate** — `sudo -E cargo test -p yipd --test tunnel_netns ping_across_yipd_tunnel -- --nocapture --test-threads=1` → PASS. + +- [ ] **Step 2: Add `PacketType::Control = 3`** in handshake.rs (mirror the existing enum + `as u8` usage). + +- [ ] **Step 3: Implement the wiring.** Egress: record `sent_log.insert(counter, class)` per object (bounded — evict with the retransmit buffer in Phase B; for now a bounded `HashMap`/ring). A periodic timer (or a counter of elapsed `now_ms`) builds a `LossReport` from the ingress detector, seals it, sends it with the `Control` prefix. Ingress: call `detector.on_seen(counter)` for every datagram and `detector.on_delivered(counter)` when `decode` returns `Some`; branch on `PacketType::Control` to decrypt + attribute loss. Keep the two-thread model; the detector and sent-log are shared (`Arc`), like Session/Transport. Per-class fraction = (class missing in this report) / (class sent in the window) — compute from the sent log. + +- [ ] **Step 4: Netns regression + activation check.** Run the netns ping test → PASS (wire unchanged for data). Then a new netns check (extend `tunnel_netns` or add a test): on a CLEAN link, after a few seconds of bulk traffic, assert the bulk controller reached ratio 0 (expose via a log line or a test hook) and the FEC bypass is firing (≈1 symbol/packet). If a full assertion is impractical in the test harness, log the ratio and verify manually + leave a smoke assertion that the tunnel still pings. + +- [ ] **Step 5: Commit** — `git commit -m "Wire the control channel: feedback packets drive observe_loss"` + +--- + +### Task 5: Phase-A end-to-end throughput check + +Confirm the throughput win actually lands now that bulk can reach zero repair. + +**Files:** +- Modify: `crates/yip-bench/README.md` (Phase-A throughput delta) +- Run: `crates/yip-bench/tests/run-iperf-compare.sh` + +- [ ] **Step 1: Build release + measure** + +```bash +cargo build --release -p yipd +sudo -E bash crates/yip-bench/tests/run-iperf-compare.sh target/release/yipd +``` +Expected: yip clean-link (0 % loss) single-stream TCP **rises** vs the throughput-pass baseline (~220–285 Mbit/s) once the bulk flow drives repair to 0 (the bypass fires + datagram count halves). Record the before/after. + +- [ ] **Step 2: Record** in the bench README (a "Feedback loop — Phase A" delta). If throughput did NOT rise, STOP and investigate whether the bulk flow actually reached ratio 0 (the feedback round-trip + classification) before proceeding to ARQ. + +- [ ] **Step 3: Commit** — `git commit -m "Measure the Phase-A clean-link throughput win"` + +--- + +## Phase B — reactive ARQ + +### Task 6: Bounded sender retransmit buffer + +**Files:** +- Create: `crates/yip-transport/src/retxbuf.rs` +- Modify: `crates/yip-transport/src/lib.rs` (`pub mod retxbuf;`) +- Test: `retxbuf.rs` `#[cfg(test)]` + +**Interfaces:** +- Produces: `pub struct RetxBuffer { .. }` with `pub fn new(max: usize, ttl_ms: u64) -> Self`, `pub fn put(&mut self, counter: u64, ciphertext: Vec, class: FlowClass, object_id: u16, now_ms: u64)`, `pub fn get(&self, counter: u64, now_ms: u64) -> Option<(&[u8], FlowClass, u16)>` (returns `None` if absent or older than `ttl`; the `u16` is the original `object_id` — retransmits MUST reuse it so the receiver's existing decoder for that object is topped up rather than a new one started), and internal LRU+TTL eviction keeping `<= max` entries. Mirrors the existing bounded `FlowTable` pattern in `flow.rs` (order `VecDeque` + map, evict oldest past cap/ttl). + +- [ ] **Step 1: Write tests** (put/get round-trip; eviction past `max`; `get` returns `None` past `ttl`; bounded size under churn — mirror `flow.rs::flow_table_is_bounded`). + +```rust +#[test] +fn retx_put_get_roundtrip() { + let mut b = RetxBuffer::new(1024, 2000); + b.put(7, vec![1,2,3], FlowClass::Bulk, 99, 0); + let (ct, class, oid) = b.get(7, 100).expect("present"); + assert_eq!(ct, &[1,2,3]); assert_eq!(class, FlowClass::Bulk); assert_eq!(oid, 99); +} +#[test] +fn retx_evicts_past_ttl() { + let mut b = RetxBuffer::new(1024, 2000); + b.put(7, vec![1], FlowClass::Bulk, 0, 0); + assert!(b.get(7, 3000).is_none(), "expired past ttl"); +} +#[test] +fn retx_is_bounded_under_churn() { + let mut b = RetxBuffer::new(16, 1_000_000); + for c in 0..10_000u64 { b.put(c, vec![0u8; 4], FlowClass::Bulk, 0, c); } + assert!(b.len() <= 16); +} +``` + +- [ ] **Step 2: Run to verify fail.** **Step 3: Implement** (add `len()` for the test). **Step 4: Run PASS.** **Step 5: Commit** — `git commit -m "Add bounded sender retransmit buffer"` + +--- + +### Task 7: Retransmit on NACK (fresh repair symbols, class/deadline-aware) + dedup + +**Files:** +- Modify: `crates/yip-transport/src/lib.rs` (a `Transport::repair_object` API) +- Modify: `bin/yipd/src/tunnel.rs` (egress fills `RetxBuffer`; control handler retransmits eligible NACKs) +- Test: `lib.rs` `#[cfg(test)]` (repair round-trip) + the netns ARQ test + +**Interfaces:** +- Consumes: `RetxBuffer` (T6), `FecEncoder` (existing), `FlowParams.arq`/`deadline`. +- Produces: `pub fn repair_object(&mut self, ciphertext: &[u8], class: FlowClass, object_id: u16, extra_repair: u32) -> Vec` on `Transport` — generates fresh RaptorQ repair symbols (ESI beyond the source range) for the object, **carrying the original `object_id`** so the receiver tops up its existing decoder for that object. (Implementation: a `FecEncoder::repair_with_id(ciphertext, params, object_id, extra_repair)` that builds the encoder and emits the repair-only `EncodingPacket`s through `split_packet(object_id, …)`; verify identity + completion via the decode round-trip below.) + +- [ ] **Step 1: Repair round-trip unit test** + +```rust +#[test] +fn retransmitted_repair_completes_a_missing_object() { + let mut tx = Transport::new(vec![]); + let ct = vec![0x33u8; 2400]; // 2 source symbols + let (cls, syms) = tx.encode(&ct, &ct, false, 0); + let oid = syms[0].object_id; // the original object's identity + // Drop one source symbol; decode stalls. + let mut rx = Transport::new(vec![]); + let mut out = None; + for s in syms.iter().skip(1) { out = out.or(rx.decode(s, cls)); } + assert!(out.is_none(), "one symbol short -> not yet decoded"); + // Retransmit: fresh repair symbols carrying the SAME object_id top up the decoder. + let repair = tx.repair_object(&ct, cls, oid, 2); + assert!(repair.iter().all(|s| s.object_id == oid), "repair reuses object identity"); + for s in &repair { out = out.or(rx.decode(s, cls)); } + assert_eq!(out.as_deref(), Some(ct.as_slice())); +} +``` + +- [ ] **Step 2: Run fail → Step 3: implement `repair_object`** (and have egress capture `object_id = syms[0].object_id` from its `encode` result and `retx.put(counter, sealed.ciphertext.clone(), class, object_id, now)`). The control handler, per `missing` counter: `if let Some((ct, class, oid)) = retx.get(counter, now)` (which already drops entries past the buffer TTL) and `class.params().arq` (excludes realtime — its objects are never retransmitted) → `let syms = transport.repair_object(ct, class, oid, K)` → frame each under that `counter` + class (reuse `wire_glue::symbol_to_frame`) → `send_batch`. Else ignore. Dedup is automatic: a retransmit for an already-delivered object returns `None` from `decode` (the existing "late symbol after completion" path — covered by `decode_late_symbol_returns_none_after_completion`). + +- [ ] **Step 4: Netns ARQ test under loss.** Add/extend a root-gated test: two yipd over netem with ~5–10 % loss carrying a **bulk** flow; assert payload integrity is maintained (ARQ recovers what FEC misses). A **realtime**-classed flow under loss: assert no retransmit traffic for it (e.g., via a counter/log). At minimum, assert the bulk transfer completes intact under loss where it would otherwise drop. + +- [ ] **Step 5: Commit** — `git commit -m "Reactive ARQ: retransmit fresh repair symbols for eligible NACKs"` + +--- + +### Task 8: End-to-end measurement + docs (the verdict) + +**Files:** +- Modify: `crates/yip-bench/README.md`, `CHANGELOG.md` +- Run: `run-iperf-compare.sh`, `run-fec-compare.sh`, the netns ARQ test + +- [ ] **Step 1: Measure** clean-link throughput (should hold the Phase-A gain) and loss-recovery (FEC + ARQ): `run-fec-compare.sh` should still show yip ~full delivery; add an ARQ-under-loss data point. + +- [ ] **Step 2: Record** before/after + the ARQ result in the bench README; add a `CHANGELOG.md` entry. Revert any sub-change that showed no value. + +- [ ] **Step 3: Full gate** — `cargo fmt --all -- --check && cargo clippy --workspace --all-targets -- -D warnings && cargo test --workspace` → green; plus the root-gated netns tests under sudo. + +- [ ] **Step 4: Commit** — `git commit -m "Measure and record the feedback loop + ARQ"` + +--- + +## Self-review notes + +- **Spec coverage:** Control packet → T1; gap detector → T2; class-aware zero-repair → T3; control wiring/attribution → T4; Phase-A throughput verdict → T5; retransmit buffer → T6; ARQ retransmit + dedup + deadline/class gating → T7; final measurement → T8. Security (sealed control) is in T4; bounded buffers in T2/T6. Out-of-scope items untasked. +- **Phase boundary:** T1–T5 deliver a working, independently-valuable system (the throughput win) before any ARQ code — the spec's required sequencing. +- **Type consistency:** `LossReport{delivered_count,high_counter,missing}`, `MAX_NACK`, `LossDetector::{new,on_seen,on_delivered,report}`, `AdaptiveController::new_for`, `RetxBuffer::{new,put,get,len}`, `Transport::repair_object` are used consistently across tasks. +- **Wire-compat gate:** T4 and T7 are guarded by the `tunnel_netns` ping test; data-symbol frame unchanged. +- **Known soft spot:** T4's "assert the controller reached ratio 0" may need a small test hook (a log line or a debug accessor) since it's daemon-internal — the plan allows logging + a ping smoke assertion if a hard assertion is impractical; the real proof is T5's throughput rise. From b3a975fb8d1ed99f11c68e58ce5389ec1cb4f81e Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 19:01:10 -0400 Subject: [PATCH 03/14] Add LossReport control-packet (de)serialization Implement the feedback module for the receiver-to-sender loss report. Includes serialization to big-endian wire format with 14-byte header (delivered_count:u32, high_counter:u64, n_missing:u16) followed by missing sequence numbers (u64 each). Decode validates input length and rejects malformed bytes. Encode caps missing at MAX_NACK (64). Co-Authored-By: Claude Opus 4.8 --- crates/yip-transport/src/feedback.rs | 123 +++++++++++++++++++++++++++ crates/yip-transport/src/lib.rs | 3 + 2 files changed, 126 insertions(+) create mode 100644 crates/yip-transport/src/feedback.rs diff --git a/crates/yip-transport/src/feedback.rs b/crates/yip-transport/src/feedback.rs new file mode 100644 index 0000000..f381306 --- /dev/null +++ b/crates/yip-transport/src/feedback.rs @@ -0,0 +1,123 @@ +//! Loss feedback: receiver → sender control message. +//! +//! Encodes delivered and missing packet ranges as a compact byte stream. +#![forbid(unsafe_code)] + +/// Maximum number of missing packets to report in a single LossReport. +pub const MAX_NACK: usize = 64; + +/// A loss report from receiver to sender. +/// +/// Reports which packets were successfully delivered and which are missing. +/// The wire format is big-endian: 4-byte delivered count, 8-byte high counter, +/// 2-byte missing count, then n × 8-byte missing sequence numbers. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LossReport { + /// Total number of packets delivered so far. + pub delivered_count: u32, + /// The highest sequence number received (even if not delivered). + pub high_counter: u64, + /// Sequence numbers of packets that were not delivered. + pub missing: Vec, +} + +impl LossReport { + /// Encode the report to bytes (big-endian). + /// + /// Caps `missing` at `MAX_NACK` entries. + pub fn encode(&self) -> Vec { + let n_missing = self.missing.len().min(MAX_NACK); + let mut bytes = Vec::with_capacity(14 + n_missing * 8); + bytes.extend_from_slice(&self.delivered_count.to_be_bytes()); + bytes.extend_from_slice(&self.high_counter.to_be_bytes()); + bytes.extend_from_slice(&u16::try_from(n_missing).unwrap().to_be_bytes()); + for i in 0..n_missing { + bytes.extend_from_slice(&self.missing[i].to_be_bytes()); + } + bytes + } + + /// Decode a report from untrusted bytes. + /// + /// Returns `None` if the input is malformed (too short, or inconsistent + /// missing count vs. payload length). + pub fn decode(bytes: &[u8]) -> Option { + // Check minimum header length + if bytes.len() < 14 { + return None; + } + + // Parse header + let delivered_count = u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]); + let high_counter = u64::from_be_bytes([ + bytes[4], bytes[5], bytes[6], bytes[7], bytes[8], bytes[9], bytes[10], bytes[11], + ]); + let n_missing = u16::from_be_bytes([bytes[12], bytes[13]]); + let n_missing_usize = usize::from(n_missing); + + // Validate length matches expected payload + let expected_len = 14 + n_missing_usize * 8; + if bytes.len() != expected_len { + return None; + } + + // Parse missing sequence numbers + let mut missing = Vec::with_capacity(n_missing_usize); + for i in 0..n_missing_usize { + let offset = 14 + i * 8; + let num = u64::from_be_bytes([ + bytes[offset], + bytes[offset + 1], + bytes[offset + 2], + bytes[offset + 3], + bytes[offset + 4], + bytes[offset + 5], + bytes[offset + 6], + bytes[offset + 7], + ]); + missing.push(num); + } + + Some(LossReport { + delivered_count, + high_counter, + missing, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn loss_report_roundtrips() { + let r = LossReport { + delivered_count: 1000, + high_counter: 5_000, + missing: vec![10, 42, 4_999], + }; + let bytes = r.encode(); + let got = LossReport::decode(&bytes).expect("decodes"); + assert_eq!(got.delivered_count, 1000); + assert_eq!(got.high_counter, 5_000); + assert_eq!(got.missing, vec![10, 42, 4_999]); + } + + #[test] + fn loss_report_decode_rejects_short_input() { + assert!(LossReport::decode(&[]).is_none()); + assert!(LossReport::decode(&[0u8; 5]).is_none()); // shorter than the 14-byte header + } + + #[test] + fn loss_report_encode_caps_missing_at_max_nack() { + let r = LossReport { + delivered_count: 0, + high_counter: 0, + missing: (0..1000).collect(), + }; + let got = LossReport::decode(&r.encode()).expect("decodes"); + assert_eq!(got.missing.len(), MAX_NACK); + } +} diff --git a/crates/yip-transport/src/lib.rs b/crates/yip-transport/src/lib.rs index 54d9c40..6583dca 100644 --- a/crates/yip-transport/src/lib.rs +++ b/crates/yip-transport/src/lib.rs @@ -15,6 +15,9 @@ pub use control::AdaptiveController; pub mod fec; pub use fec::{FecEncoder, FecReassembler, Symbol}; +pub mod feedback; +pub use feedback::{LossReport, MAX_NACK}; + use std::collections::HashMap; use std::time::Duration; From 6aa3b778850cf9a7b9d4fb15196bd1569af4f49d Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 19:06:41 -0400 Subject: [PATCH 04/14] Add gap-based receiver loss detector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements LossDetector in crates/yip-transport/src/lossdetect.rs: BTreeMap pending set (counter → first-implied-ms), grace window before declaring missing, window-bounded eviction of smallest keys. Four tests green; full yip-transport suite 35/35; clippy clean. Co-Authored-By: Claude Sonnet 4.6 --- crates/yip-transport/src/lib.rs | 3 + crates/yip-transport/src/lossdetect.rs | 209 +++++++++++++++++++++++++ 2 files changed, 212 insertions(+) create mode 100644 crates/yip-transport/src/lossdetect.rs diff --git a/crates/yip-transport/src/lib.rs b/crates/yip-transport/src/lib.rs index 6583dca..44922ad 100644 --- a/crates/yip-transport/src/lib.rs +++ b/crates/yip-transport/src/lib.rs @@ -18,6 +18,9 @@ pub use fec::{FecEncoder, FecReassembler, Symbol}; pub mod feedback; pub use feedback::{LossReport, MAX_NACK}; +pub mod lossdetect; +pub use lossdetect::LossDetector; + use std::collections::HashMap; use std::time::Duration; diff --git a/crates/yip-transport/src/lossdetect.rs b/crates/yip-transport/src/lossdetect.rs new file mode 100644 index 0000000..c571d6d --- /dev/null +++ b/crates/yip-transport/src/lossdetect.rs @@ -0,0 +1,209 @@ +//! Receiver-side gap-based loss detector. +//! +//! Turns a stream of seen/delivered packet counters into a [`LossReport`]. +//! No I/O; pure logic. +#![forbid(unsafe_code)] + +use std::collections::BTreeMap; + +use crate::{LossReport, MAX_NACK}; + +/// Receiver-side loss detector. +/// +/// Call [`on_seen`](Self::on_seen) for every received sealed packet, and +/// [`on_delivered`](Self::on_delivered) when an object fully decodes. +/// Call [`report`](Self::report) periodically to obtain a [`LossReport`]. +/// +/// # Gap model +/// +/// The detector tracks `high_counter` (the maximum counter ever seen). +/// When `on_seen(c)` is called and `c > high_counter + 1`, every counter +/// in the half-open range `(prev_high, c)` is recorded as *implied-pending* +/// with timestamp `now_ms`. When `on_delivered(c)` is called, `c` is +/// removed from the pending set. +/// +/// `report(now_ms)` promotes all implied-pending entries whose timestamp is +/// older than `grace_ms` — and that have still not been delivered — into the +/// `missing` list of the returned [`LossReport`]. Counters that arrive +/// (via `on_seen` + `on_delivered`) within the grace window are silently +/// discarded, so transient reordering is not falsely reported. +/// +/// The pending set is bounded to `window` entries; when it would exceed that +/// size the smallest (oldest-sequence) entries are evicted to prevent a huge +/// gap from exhausting memory. +pub struct LossDetector { + /// Grace period in milliseconds before an implied-pending entry is + /// promoted to missing. + grace_ms: u64, + /// Maximum number of entries to keep in the pending set. + window: usize, + /// Highest counter ever seen. + high_counter: u64, + /// Whether we have seen at least one packet (so `high_counter` is valid). + seen_any: bool, + /// Implied-pending counters: counter → timestamp when first implied. + pending: BTreeMap, + /// Objects delivered since the last `report` call. + delivered: u32, +} + +impl LossDetector { + /// Create a new detector. + /// + /// - `grace_ms`: reordering window; implied-pending entries younger than + /// this are not yet declared missing. + /// - `window`: maximum number of entries in the pending set. + pub fn new(grace_ms: u64, window: usize) -> Self { + Self { + grace_ms, + window, + high_counter: 0, + seen_any: false, + pending: BTreeMap::new(), + delivered: 0, + } + } + + /// Record that a sealed packet with the given `counter` was received at + /// wall-clock time `now_ms`. + /// + /// If `counter` is greater than `high_counter + 1`, every counter in the + /// range `(high_counter, counter)` — i.e. the skipped ones — is inserted + /// into the pending set with timestamp `now_ms` (if not already present). + pub fn on_seen(&mut self, counter: u64, now_ms: u64) { + if !self.seen_any { + self.high_counter = counter; + self.seen_any = true; + return; + } + + if counter > self.high_counter { + // Mark every counter in the gap as implied-pending. + let gap_start = self.high_counter + 1; + for c in gap_start..counter { + self.pending.entry(c).or_insert(now_ms); + } + self.high_counter = counter; + // Enforce the window bound: evict smallest keys when over capacity. + while self.pending.len() > self.window { + // BTreeMap::pop_first is stable since Rust 1.66. + self.pending.pop_first(); + } + } else { + // Counter is <= high_counter: it was either already recorded as + // pending (gap filled / reorder) or already seen. Remove it from + // pending so it won't be declared missing. + self.pending.remove(&counter); + } + } + + /// Record that the object for `counter` fully decoded. + /// + /// Removes `counter` from the pending set (so it is never reported missing) + /// and increments the per-window delivered count. + pub fn on_delivered(&mut self, counter: u64) { + self.pending.remove(&counter); + self.delivered = self.delivered.saturating_add(1); + } + + /// Emit a [`LossReport`] for the current window. + /// + /// Counters that have been implied-pending for longer than `grace_ms` and + /// have still not been delivered are promoted to `missing`. Those + /// counters are then removed from the pending set. The `delivered_count` + /// field reflects deliveries since the last call to `report`; it is reset + /// afterwards. + pub fn report(&mut self, now_ms: u64) -> LossReport { + let mut missing: Vec = Vec::new(); + + // Collect counters whose grace period has expired. + let promoted: Vec = self + .pending + .iter() + .filter_map(|(&c, &first_ms)| { + // now_ms - first_ms >= grace_ms (saturating to avoid underflow) + let age = now_ms.saturating_sub(first_ms); + if age >= self.grace_ms { + Some(c) + } else { + None + } + }) + .collect(); + + for c in promoted { + self.pending.remove(&c); + missing.push(c); + } + + // Cap at MAX_NACK (the wire-format limit). + missing.truncate(MAX_NACK); + + let high_counter = if self.seen_any { self.high_counter } else { 0 }; + let delivered_count = self.delivered; + self.delivered = 0; + + LossReport { + delivered_count, + high_counter, + missing, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn contiguous_no_loss() { + let mut d = LossDetector::new(5, 1024); + for c in 0..100u64 { + d.on_seen(c, 0); + d.on_delivered(c); + } + let r = d.report(100); + assert!(r.missing.is_empty()); + assert_eq!(r.delivered_count, 100); + } + + #[test] + fn gap_reported_after_grace() { + let mut d = LossDetector::new(5, 1024); + // see 0,1,3 (2 is a gap); deliver the ones we saw + for c in [0u64, 1, 3] { + d.on_seen(c, 0); + d.on_delivered(c); + } + // before grace elapses, 2 is not yet declared + assert!(d.report(3).missing.is_empty()); + // after grace, 2 is missing + let r = d.report(10); + assert_eq!(r.missing, vec![2]); + } + + #[test] + fn reorder_within_grace_not_reported() { + let mut d = LossDetector::new(5, 1024); + d.on_seen(0, 0); + d.on_delivered(0); + d.on_seen(2, 0); + d.on_delivered(2); // 1 appears skipped... + d.on_seen(1, 2); + d.on_delivered(1); // ...but arrives within grace + let r = d.report(10); + assert!( + r.missing.is_empty(), + "in-grace reorder must not be reported lost" + ); + } + + #[test] + fn missing_set_is_bounded() { + let mut d = LossDetector::new(0, 8); // grace 0 → immediate; window 8 + d.on_seen(0, 0); + d.on_seen(1000, 1); // implies a huge gap + let r = d.report(5); + assert!(r.missing.len() <= 8 + 64); // bounded by window and MAX_NACK at encode + } +} From 3266883e2424c0ecb4557c0fa2cf22dc31984653 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 19:10:28 -0400 Subject: [PATCH 05/14] Document loss-detector gap-model limitation; tighten bound test Co-Authored-By: Claude Opus 4.8 --- crates/yip-transport/src/lossdetect.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/crates/yip-transport/src/lossdetect.rs b/crates/yip-transport/src/lossdetect.rs index c571d6d..5c5ba02 100644 --- a/crates/yip-transport/src/lossdetect.rs +++ b/crates/yip-transport/src/lossdetect.rs @@ -31,6 +31,15 @@ use crate::{LossReport, MAX_NACK}; /// The pending set is bounded to `window` entries; when it would exceed that /// size the smallest (oldest-sequence) entries are evicted to prevent a huge /// gap from exhausting memory. +/// +/// # Limitation: Incomplete objects are not reported as missing +/// +/// A counter that is *seen* (via `on_seen`) but never *delivered* (never +/// calls `on_delivered`) — such as a multi-symbol FEC object that received +/// some but insufficient symbols to decode — is **not** reported as missing. +/// Only fully-absent counters (true gaps in the sequence, where `on_seen` is +/// never called) produce `missing` entries. This is exact for single-symbol +/// (zero-repair) objects, where a loss is always a full gap. pub struct LossDetector { /// Grace period in milliseconds before an implied-pending entry is /// promoted to missing. @@ -204,6 +213,6 @@ mod tests { d.on_seen(0, 0); d.on_seen(1000, 1); // implies a huge gap let r = d.report(5); - assert!(r.missing.len() <= 8 + 64); // bounded by window and MAX_NACK at encode + assert!(r.missing.len() <= 8); // bounded by window } } From 6d2ef4fb55025ec6249b1cdddc10216da5e156fc Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 19:13:08 -0400 Subject: [PATCH 06/14] Let ARQ-class repair ratio decay to zero; keep floor for non-ARQ --- crates/yip-transport/src/control.rs | 74 +++++++++++++++++++++++++++-- crates/yip-transport/src/lib.rs | 6 +-- 2 files changed, 72 insertions(+), 8 deletions(-) diff --git a/crates/yip-transport/src/control.rs b/crates/yip-transport/src/control.rs index 0ed98b3..209a1e6 100644 --- a/crates/yip-transport/src/control.rs +++ b/crates/yip-transport/src/control.rs @@ -12,15 +12,29 @@ pub struct AdaptiveController { } impl AdaptiveController { - /// Start from a class's initial repair ratio. - pub fn new(params: FlowParams) -> Self { + /// Start from a class's initial repair ratio, using the class's `arq` flag + /// to determine the floor: ARQ-eligible classes may decay to zero (bypass FEC + /// encode entirely on a clean link); non-ARQ classes keep a proactive floor. + pub fn new_for(params: FlowParams) -> Self { Self { ratio: params.initial_repair_ratio, - min_ratio: params.initial_repair_ratio, + min_ratio: if params.arq { + 0.0 + } else { + params.initial_repair_ratio + }, target_residual: 0.001, // aim for <0.1% post-FEC loss } } + /// Start from a class's initial repair ratio. + /// + /// Equivalent to `new_for(params)` — kept for callers that don't need + /// class-aware zero-ratio behaviour. + pub fn new(params: FlowParams) -> Self { + Self::new_for(params) + } + /// The current repair ratio (repair symbols per source symbol). pub fn ratio(&self) -> f32 { self.ratio @@ -37,13 +51,26 @@ impl AdaptiveController { // losing more than we can repair: add headroom above the loss rate self.ratio = (loss + 0.05).min(1.0); } else if loss <= self.target_residual { - // clean: decay 10% toward the floor - self.ratio = (self.ratio * 0.9).max(self.min_ratio); + // clean: decay 10% toward the floor; snap to the floor once the + // value is negligibly small so ARQ-class controllers reach exactly 0.0 + let decayed = (self.ratio * 0.9).max(self.min_ratio); + self.ratio = if decayed <= f32::EPSILON { + self.min_ratio + } else { + decayed + }; } } /// How many repair symbols to emit for an object with `source_symbols` source symbols. + /// + /// Returns `0` when `ratio` has decayed to zero (ARQ-eligible classes on a + /// clean link) — the caller should skip FEC-encode entirely in that case. + /// Returns at least `1` whenever `ratio > 0`. pub fn repair_count(&self, source_symbols: u32) -> u32 { + if self.ratio <= f32::EPSILON { + return 0; + } let raw = (f64::from(source_symbols) * f64::from(self.ratio)).ceil(); // f64::ceil of a non-negative product; the i64 cast is the one unavoidable float->int conversion, then checked-narrowed to u32 let n = u32::try_from(raw as i64).unwrap_or(u32::MAX); @@ -81,4 +108,41 @@ mod tests { assert!(c.repair_count(100) >= 1); assert!(c.repair_count(0) == 0 || c.repair_count(0) >= 1); // never panics on zero } + + #[test] + fn arq_class_decays_to_zero_when_clean() { + let mut c = AdaptiveController::new_for(FlowClass::Bulk.params()); // arq=true + for _ in 0..200 { + c.observe_loss(0.0); + } + assert_eq!(c.ratio(), 0.0, "bulk earns zero repair on a clean link"); + assert_eq!( + c.repair_count(1), + 0, + "zero ratio -> zero repair -> bypass fires" + ); + assert_eq!(c.repair_count(10), 0); + } + + #[test] + fn non_arq_class_keeps_floor() { + let mut c = AdaptiveController::new_for(FlowClass::Realtime.params()); // arq=false + for _ in 0..200 { + c.observe_loss(0.0); + } + assert!(c.ratio() >= FlowClass::Realtime.params().initial_repair_ratio - 1e-6); + assert!(c.repair_count(1) >= 1, "realtime keeps proactive repair"); + } + + #[test] + fn snaps_up_on_loss_even_from_zero() { + let mut c = AdaptiveController::new_for(FlowClass::Bulk.params()); + for _ in 0..200 { + c.observe_loss(0.0); + } + assert_eq!(c.ratio(), 0.0); + c.observe_loss(0.10); + assert!(c.ratio() >= 0.10, "any loss re-arms FEC immediately"); + assert!(c.repair_count(10) >= 1); + } } diff --git a/crates/yip-transport/src/lib.rs b/crates/yip-transport/src/lib.rs index 44922ad..db187ef 100644 --- a/crates/yip-transport/src/lib.rs +++ b/crates/yip-transport/src/lib.rs @@ -100,9 +100,9 @@ impl Transport { classifier: Classifier::new(rules), encoder: FecEncoder::new(), controllers: [ - AdaptiveController::new(FlowClass::Realtime.params()), - AdaptiveController::new(FlowClass::Bulk.params()), - AdaptiveController::new(FlowClass::Default.params()), + AdaptiveController::new_for(FlowClass::Realtime.params()), + AdaptiveController::new_for(FlowClass::Bulk.params()), + AdaptiveController::new_for(FlowClass::Default.params()), ], reassemblers: HashMap::new(), } From bd6d6d237ab61fbf384acaae8c7a1b0c7e5df8f9 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 19:22:33 -0400 Subject: [PATCH 07/14] Wire the control channel: feedback packets drive observe_loss MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both peers now send a sealed Control packet every 30 ms containing a LossReport (delivered_count, high_counter, missing counters). The receiver decrypts the report, looks up each missing counter in a bounded sent-log (counter → FlowClass, capacity 4096), computes per-class loss fractions, and feeds them to Transport::observe_loss, closing the feedback loop into the adaptive FEC controller. Changes: - handshake.rs: add PacketType::Control = 3 - tunnel.rs: SentLog ring (capacity 4096), LossDetector Arc, egress records counter→class in sent_log, ingress calls on_seen for every datagram (data and control) and on_delivered when FEC decodes, emits feedback every FEEDBACK_INTERVAL_MS=30, handles incoming Control packets to call observe_loss per class; periodic log of the bulk controller repair ratio every 5 s. - yip-transport/lib.rs: add Transport::bulk_repair_ratio() for diagnostics. Co-Authored-By: Claude Opus 4.8 --- bin/yipd/src/handshake.rs | 2 + bin/yipd/src/tunnel.rs | 357 ++++++++++++++++++++++++++++---- crates/yip-transport/src/lib.rs | 8 + 3 files changed, 322 insertions(+), 45 deletions(-) diff --git a/bin/yipd/src/handshake.rs b/bin/yipd/src/handshake.rs index 9b5e1b3..5543063 100644 --- a/bin/yipd/src/handshake.rs +++ b/bin/yipd/src/handshake.rs @@ -28,6 +28,8 @@ pub enum PacketType { HandshakeResp = 1, /// Data-plane packet (used by later tunnel tasks). Data = 2, + /// Loss-feedback control packet (receiver → sender). + Control = 3, } // ── established session ─────────────────────────────────────────────────────── diff --git a/bin/yipd/src/tunnel.rs b/bin/yipd/src/tunnel.rs index f4a5670..50ccf22 100644 --- a/bin/yipd/src/tunnel.rs +++ b/bin/yipd/src/tunnel.rs @@ -9,7 +9,18 @@ //! bytes of the Noise channel binding (which both peers compute identically //! after the handshake, so both will use the same tag). This is a placeholder: //! M7+ will rotate the tag every epoch for unlinkability. - +//! +//! # Feedback loop +//! +//! Both peers run identical code: the ingress side tracks gaps via +//! `LossDetector` and periodically seals a `LossReport` into a `Control` +//! packet (`[PacketType::Control][counter:8be][ciphertext]`) sent to the peer. +//! The peer's ingress thread decrypts the report, looks up each missing counter +//! in a bounded `sent_log` (`counter → FlowClass`), computes per-class loss +//! fractions, and feeds them to `Transport::observe_loss`, which adjusts the +//! FEC repair ratios. + +use std::collections::{HashMap, VecDeque}; use std::io; use std::net::UdpSocket; use std::sync::{Arc, Mutex}; @@ -17,7 +28,7 @@ use std::time::Instant; use yip_device::{DeviceKind, TunTap}; use yip_io::{set_socket_buffers, DataPlaneIo, PlainIo, MAX_DATAGRAM_BATCH, MAX_WIRE_DATAGRAM}; -use yip_transport::Transport; +use yip_transport::{FlowClass, LossDetector, LossReport, Transport}; use yip_wire::{Codec, WireCodec}; use crate::config::Config; @@ -27,6 +38,54 @@ use crate::wire_glue; // Maximum UDP datagram we ever allocate for recv. const MAX_DATAGRAM: usize = 65_535; +// How often (in milliseconds of elapsed tunnel time) the ingress thread emits +// a loss-feedback Control packet to the peer. +const FEEDBACK_INTERVAL_MS: u64 = 30; + +// Maximum number of entries in the sent-log (counter → FlowClass). Once full, +// the oldest entry is evicted before each new insertion, so memory is O(1). +const SENT_LOG_CAPACITY: usize = 4096; + +/// A bounded ring-log that maps sealed-packet counters to their `FlowClass`. +/// +/// Entries are inserted in arrival order (monotone counter sequence from the +/// AEAD) and evicted oldest-first once `capacity` is reached. Lookup is +/// O(1) via the `HashMap`; eviction is O(1) via the `VecDeque`. +struct SentLog { + capacity: usize, + map: HashMap, + order: VecDeque, +} + +impl SentLog { + fn new(capacity: usize) -> Self { + Self { + capacity, + map: HashMap::with_capacity(capacity), + order: VecDeque::with_capacity(capacity), + } + } + + fn insert(&mut self, counter: u64, class: FlowClass) { + if self.map.len() >= self.capacity { + if let Some(oldest) = self.order.pop_front() { + self.map.remove(&oldest); + } + } + self.map.insert(counter, class); + self.order.push_back(counter); + } + + fn get(&self, counter: u64) -> Option { + self.map.get(&counter).copied() + } + + /// Number of entries in the log whose `FlowClass` matches `class`. + fn count_class(&self, class: FlowClass) -> u32 { + u32::try_from(self.map.values().filter(|&&c| c == class).count()).unwrap_or(u32::MAX) + } +} + // ── conn_tag derivation ─────────────────────────────────────────────────────── /// Derive the per-session `conn_tag` from the first 8 bytes of the Noise @@ -93,6 +152,13 @@ pub fn run(config: Config) -> io::Result<()> { let session = Arc::new(Mutex::new(established.session)); let transport = Arc::new(Mutex::new(Transport::new(vec![]))); + // Sent-log: egress records counter→class; ingress control-handler reads it. + let sent_log: Arc> = Arc::new(Mutex::new(SentLog::new(SENT_LOG_CAPACITY))); + + // Loss detector: ingress updates it from every received datagram; the + // ingress thread also reads it periodically to emit feedback. + let detector: Arc> = Arc::new(Mutex::new(LossDetector::new(5, 1024))); + // ── create + split the TUN device ──────────────────────────────────────── let tun = TunTap::create(&config.device, DeviceKind::Tun).map_err(io::Error::other)?; let (mut tun_reader, mut tun_writer) = tun.split().map_err(io::Error::other)?; @@ -109,6 +175,7 @@ pub fn run(config: Config) -> io::Result<()> { // ── egress thread: TUN → UDP ────────────────────────────────────────────── let session_tx = Arc::clone(&session); let transport_tx = Arc::clone(&transport); + let sent_log_tx = Arc::clone(&sent_log); let egress = std::thread::Builder::new() .name("yipd-egress".into()) @@ -145,6 +212,13 @@ pub fn run(config: Config) -> io::Result<()> { .expect("transport lock poisoned") .encode(&sealed.ciphertext, inner, false, now_ms); + // Record counter → class in the sent-log so the ingress thread + // can attribute received loss reports to the right flow class. + sent_log_tx + .lock() + .expect("sent_log lock poisoned") + .insert(sealed.counter, class); + // Frame each symbol into a reused arena slot, then emit all // datagrams in one batched syscall (chunked by MAX_DATAGRAM_BATCH). let n_syms = symbols.len(); @@ -191,6 +265,8 @@ pub fn run(config: Config) -> io::Result<()> { // ── ingress thread: UDP → TUN ───────────────────────────────────────────── let session_rx = Arc::clone(&session); let transport_rx = Arc::clone(&transport); + let sent_log_rx = Arc::clone(&sent_log); + let detector_rx = Arc::clone(&detector); let ingress = std::thread::Builder::new() .name("yipd-ingress".into()) @@ -204,69 +280,244 @@ pub fn run(config: Config) -> io::Result<()> { let mut bufs = vec![[0u8; MAX_WIRE_DATAGRAM]; MAX_DATAGRAM_BATCH]; let mut lens = vec![0usize; MAX_DATAGRAM_BATCH]; + // Track when we last emitted a feedback packet (in tunnel-uptime ms). + let start = Instant::now(); + let mut last_feedback_ms: u64 = 0; + + // Periodic log interval for controller ratio (every ~5 s). + let mut last_log_ms: u64 = 0; + const LOG_INTERVAL_MS: u64 = 5_000; + loop { // Block until ≥1 datagram arrives (MSG_WAITFORONE), then drain // however many are immediately available (up to MAX_DATAGRAM_BATCH). let n = io.recv_batch(&mut bufs, &mut lens)?; + let now_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX); + for i in 0..n { let dg = &bufs[i][..lens[i]]; - // Validate and strip the 1-byte packet type prefix. - if dg.is_empty() || dg[0] != PacketType::Data as u8 { + if dg.is_empty() { continue; } - let wire = &dg[1..]; - // Deframe (auth + header-deprotect). On failure, drop the - // packet without killing the loop. - let frame = match codec_rx.deframe(wire) { - Ok(f) => f, - Err(e) => { - eprintln!("yipd ingress: deframe error: {e}"); - continue; + match dg[0] { + b if b == PacketType::Data as u8 => { + let wire = &dg[1..]; + + // Deframe (auth + header-deprotect). On failure, drop the + // packet without killing the loop. + let frame = match codec_rx.deframe(wire) { + Ok(f) => f, + Err(e) => { + eprintln!("yipd ingress: deframe error: {e}"); + continue; + } + }; + + // Parse the FEC symbol + counter out of the frame. + let (sym, counter, class) = match wire_glue::frame_to_symbol(&frame) { + Some(t) => t, + None => { + eprintln!( + "yipd ingress: frame_to_symbol returned None (short payload)" + ); + continue; + } + }; + + // Notify the loss detector that we saw this counter. + detector_rx + .lock() + .expect("detector lock poisoned") + .on_seen(counter, now_ms); + + // Feed the symbol to the FEC reassembler; continue until an + // object decodes. + let ciphertext = match transport_rx + .lock() + .expect("transport lock poisoned") + .decode(&sym, class) + { + Some(ct) => ct, + None => continue, + }; + + // The object decoded: tell the detector it was delivered. + detector_rx + .lock() + .expect("detector lock poisoned") + .on_delivered(counter); + + // Open the AEAD ciphertext. Replay / AEAD failures are logged + // and the packet is dropped. + let inner = match session_rx + .lock() + .expect("session lock poisoned") + .open(counter, &ciphertext) + { + Ok(p) => p, + Err(e) => { + eprintln!("yipd ingress: open error: {e}"); + continue; + } + }; + + // Inject the plaintext inner frame into the TUN device. + // An I/O error here is fatal (device gone). + tun_writer.write_frame(&inner)?; } - }; - // Parse the FEC symbol + counter out of the frame. - let (sym, counter, class) = match wire_glue::frame_to_symbol(&frame) { - Some(t) => t, - None => { - eprintln!( - "yipd ingress: frame_to_symbol returned None (short payload)" + b if b == PacketType::Control as u8 => { + // Control packet layout: + // [1-byte type][8-byte counter BE][ciphertext...] + if dg.len() < 9 { + eprintln!("yipd ingress: control packet too short"); + continue; + } + let counter = u64::from_be_bytes( + dg[1..9].try_into().expect("exactly 8 bytes"), ); - continue; + let ct = &dg[9..]; + + // Notify the detector that we saw this control counter too, + // so the unified counter sequence stays consistent. + detector_rx + .lock() + .expect("detector lock poisoned") + .on_seen(counter, now_ms); + + // Decrypt the control payload. + let plaintext = match session_rx + .lock() + .expect("session lock poisoned") + .open(counter, ct) + { + Ok(p) => p, + Err(e) => { + eprintln!("yipd ingress: control open error: {e}"); + continue; + } + }; + + // Decode the LossReport. + let report = match LossReport::decode(&plaintext) { + Some(r) => r, + None => { + eprintln!("yipd ingress: malformed LossReport"); + continue; + } + }; + + // Attribute missing counters to flow classes via the sent-log. + // + // Per-class fraction estimate: + // fraction = class_missing / max(1, class_sent_in_log) + // + // `class_sent_in_log` is the count of entries in the bounded + // sent-log for that class — a conservative approximation of + // how many packets of that class were recently in-flight. + // It underestimates the true window (the log only holds the + // last SENT_LOG_CAPACITY entries) but is always ≥ class_missing, + // so the fraction stays ∈ [0, 1]. + let log = sent_log_rx.lock().expect("sent_log lock poisoned"); + + let mut missing_rt: u32 = 0; + let mut missing_bulk: u32 = 0; + let mut missing_default: u32 = 0; + + for &c in &report.missing { + match log.get(c) { + Some(FlowClass::Realtime) => { + missing_rt = + missing_rt.saturating_add(1); + } + Some(FlowClass::Bulk) => { + missing_bulk = + missing_bulk.saturating_add(1); + } + Some(FlowClass::Default) => { + missing_default = + missing_default.saturating_add(1); + } + None => { + // Counter not in log (too old or was a + // control packet) — ignore for attribution. + } + } + } + + let sent_rt = log.count_class(FlowClass::Realtime).max(1); + let sent_bulk = log.count_class(FlowClass::Bulk).max(1); + let sent_default = log.count_class(FlowClass::Default).max(1); + + drop(log); // release before locking transport + + // Compute loss fractions as f64 (lossless from u32), + // then narrow to f32 via saturating conversion. + let frac_rt = fraction_f32(missing_rt, sent_rt); + let frac_bulk = fraction_f32(missing_bulk, sent_bulk); + let frac_default = fraction_f32(missing_default, sent_default); + + let mut t = transport_rx.lock().expect("transport lock poisoned"); + t.observe_loss(FlowClass::Realtime, frac_rt); + t.observe_loss(FlowClass::Bulk, frac_bulk); + t.observe_loss(FlowClass::Default, frac_default); } - }; - // Feed the symbol to the FEC reassembler; continue until an - // object decodes. - let ciphertext = match transport_rx + _ => { + // Unknown packet type — drop silently. + } + } + } + + // ── periodic feedback emission ──────────────────────────────── + // After processing this batch, check if it is time to send a + // loss-feedback Control packet to the peer. + if now_ms.saturating_sub(last_feedback_ms) >= FEEDBACK_INTERVAL_MS { + last_feedback_ms = now_ms; + + // Build the report from the current detector state. + let report = detector_rx .lock() - .expect("transport lock poisoned") - .decode(&sym, class) - { - Some(ct) => ct, - None => continue, - }; - - // Open the AEAD ciphertext. Replay / AEAD failures are logged - // and the packet is dropped. - let inner = match session_rx + .expect("detector lock poisoned") + .report(now_ms); + + let report_bytes = report.encode(); + + // Seal the report. The counter comes from the unified session + // sequence; the peer's detector will call on_seen for it. + let sealed = session_rx .lock() .expect("session lock poisoned") - .open(counter, &ciphertext) - { - Ok(p) => p, - Err(e) => { - eprintln!("yipd ingress: open error: {e}"); - continue; - } - }; + .seal(&report_bytes) + .map_err(io::Error::other)?; + + // Build and send: [type:1][counter:8be][ciphertext] + let mut pkt = Vec::with_capacity(9 + sealed.ciphertext.len()); + pkt.push(PacketType::Control as u8); + pkt.extend_from_slice(&sealed.counter.to_be_bytes()); + pkt.extend_from_slice(&sealed.ciphertext); + + // A transient send error is logged but not fatal. + if let Err(e) = io.send_batch(&[pkt.as_slice()]) { + eprintln!("yipd ingress: control send error: {e}"); + } + } - // Inject the plaintext inner frame into the TUN device. - // An I/O error here is fatal (device gone). - tun_writer.write_frame(&inner)?; + // ── periodic controller ratio log ───────────────────────────── + if now_ms.saturating_sub(last_log_ms) >= LOG_INTERVAL_MS { + last_log_ms = now_ms; + // Use a try_lock to avoid blocking the ingress hot path if + // transport is momentarily held by egress. + if let Ok(t) = transport_rx.try_lock() { + eprintln!( + "yipd [{}ms] bulk controller repair ratio: {:.4}", + now_ms, + t.bulk_repair_ratio(), + ); + } } } })?; @@ -283,3 +534,19 @@ pub fn run(config: Config) -> io::Result<()> { // Report the first error, if any. egress_result.and(ingress_result) } + +/// Compute `numerator / denominator` as an `f32` loss fraction ∈ [0.0, 1.0]. +/// +/// Both operands are narrowed to `u16` (saturating) so that `f32::from` +/// can accept them without any numeric `as` cast. For the small counts +/// that arise from the bounded sent-log (capacity 4096) and MAX_NACK (64), +/// u16 is always large enough. +#[inline] +fn fraction_f32(numerator: u32, denominator: u32) -> f32 { + if denominator == 0 { + return 0.0_f32; + } + let n = f32::from(u16::try_from(numerator).unwrap_or(u16::MAX)); + let d = f32::from(u16::try_from(denominator).unwrap_or(u16::MAX)); + (n / d).clamp(0.0_f32, 1.0_f32) +} diff --git a/crates/yip-transport/src/lib.rs b/crates/yip-transport/src/lib.rs index db187ef..24dcb1c 100644 --- a/crates/yip-transport/src/lib.rs +++ b/crates/yip-transport/src/lib.rs @@ -142,6 +142,14 @@ impl Transport { pub fn observe_loss(&mut self, class: FlowClass, loss: f32) { self.controllers[class_index(class)].observe_loss(loss); } + + /// Current repair ratio for the `Bulk` class controller. + /// + /// Useful for diagnostics: on a clean link this decays to 0.0 (ARQ bypass + /// fires, no proactive FEC symbols sent for bulk traffic). + pub fn bulk_repair_ratio(&self) -> f32 { + self.controllers[class_index(FlowClass::Bulk)].ratio() + } } #[cfg(test)] From 33b642d5f2e13ed5930298d0620a960eb956fb62 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 20:19:57 -0400 Subject: [PATCH 08/14] Measure the Phase-A clean-link throughput win Feedback loop converges the bulk controller to repair ratio 0.0000 (confirmed via yipd diagnostic log), firing the FEC-encode bypass and halving per-packet datagrams. Clean-link single-stream TCP rises from ~273-285 to ~457 Mbit/s (~60%), consistent with removing the ~24us encode (80% of egress CPU) + 2->1 datagrams. Co-Authored-By: Claude Opus 4.8 --- crates/yip-bench/README.md | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/crates/yip-bench/README.md b/crates/yip-bench/README.md index 4d1a096..b1cc83e 100644 --- a/crates/yip-bench/README.md +++ b/crates/yip-bench/README.md @@ -341,3 +341,29 @@ forfeit yip's loss-recovery thesis. **The clean-link throughput win (skip the en datagram count) is therefore unlocked by the adaptive loss-feedback loop, not by this pass alone.** This pass delivered the plumbing (batched I/O, buffers, a ready-and-tested bypass fast-path); activating the win is the next milestone. + +--- + +## Feedback loop — Phase A (the clean-link throughput unlock) + +The throughput pass shipped a *dormant* zero-repair FEC bypass. The adaptive +loss-feedback loop (this milestone) activates it: the receiver reports post-FEC +residual loss in an authenticated `Control` packet, the sender feeds it to the +per-class controller, and an ARQ-eligible (`Bulk`) flow on a clean link decays +its repair ratio to **zero** — firing the bypass (skip the ~24 µs encode) and +halving the per-packet datagram count (1 source symbol, no repair). + +Measured (release, kernel 6.18, Ryzen 5 7640U; clean-link single-stream TCP over +the netns tunnel): + +| build | bulk repair ratio | clean-link TCP | +|-------|-------------------|----------------| +| throughput pass (bypass dormant) | 0.05 floor (≥1 repair, encode always runs) | ~273–285 Mbit/s | +| feedback loop (Phase A) | **0.0000** (converged; bypass fires) | **~457 Mbit/s** | + +The bulk controller's repair ratio was confirmed at `0.0000` throughout a 40 s +run via yipd's diagnostic log — direct proof the loop converged, not merely +inferred from the throughput rise. The ~60 % jump is consistent with removing the +FEC encode (which the per-stage profile showed is ~80 % of egress CPU) and sending +one datagram per packet instead of two. Under loss the controller snaps repair +back up immediately (Phase B's ARQ then backstops the residual). From 28d4577722d6bc90cb4073187d0a8b9a2b89c95b Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 20:21:40 -0400 Subject: [PATCH 09/14] Add bounded sender retransmit buffer Introduces `RetxBuffer` in `crates/yip-transport/src/retxbuf.rs`: a `HashMap` + `VecDeque` order list that holds at most `max` ciphertext objects keyed by send-counter, evicting the oldest entry (by insertion order) on cap overflow and skipping expired entries (age > ttl_ms) on `get`. Mirrors the `FlowTable` pattern in `flow.rs`. The stored `object_id: u16` is preserved so ARQ retransmits reuse the receiver's existing FEC decoder for that object rather than starting a new one. Co-Authored-By: Claude Sonnet 4.6 --- crates/yip-transport/src/lib.rs | 3 + crates/yip-transport/src/retxbuf.rs | 143 ++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+) create mode 100644 crates/yip-transport/src/retxbuf.rs diff --git a/crates/yip-transport/src/lib.rs b/crates/yip-transport/src/lib.rs index 24dcb1c..b9fb589 100644 --- a/crates/yip-transport/src/lib.rs +++ b/crates/yip-transport/src/lib.rs @@ -21,6 +21,9 @@ pub use feedback::{LossReport, MAX_NACK}; pub mod lossdetect; pub use lossdetect::LossDetector; +pub mod retxbuf; +pub use retxbuf::RetxBuffer; + use std::collections::HashMap; use std::time::Duration; diff --git a/crates/yip-transport/src/retxbuf.rs b/crates/yip-transport/src/retxbuf.rs new file mode 100644 index 0000000..dcfa22a --- /dev/null +++ b/crates/yip-transport/src/retxbuf.rs @@ -0,0 +1,143 @@ +//! Bounded sender retransmit buffer. +//! +//! Holds at most `max` ciphertext objects keyed by send-counter, evicting the +//! oldest entry (by insertion order) when the cap is reached. Entries older +//! than `ttl_ms` are considered expired and are not returned by `get`. + +use crate::FlowClass; +use std::collections::{HashMap, VecDeque}; + +struct Entry { + ciphertext: Vec, + class: FlowClass, + object_id: u16, + inserted_ms: u64, +} + +/// Bounded LRU+TTL buffer of sent ciphertext objects, keyed by send-counter. +/// +/// Used by the ARQ sender: after [`put`]ting an object, a later NACK can +/// retrieve it via [`get`] and retransmit it with the *same* `object_id` so +/// the receiver's existing FEC decoder is topped up rather than a new one +/// being started. +pub struct RetxBuffer { + map: HashMap, + order: VecDeque, + max: usize, + ttl_ms: u64, +} + +impl RetxBuffer { + /// Create a buffer holding at most `max` entries, expiring any entry whose + /// age exceeds `ttl_ms` milliseconds. + pub fn new(max: usize, ttl_ms: u64) -> Self { + Self { + map: HashMap::new(), + order: VecDeque::new(), + max: max.max(1), + ttl_ms, + } + } + + /// Number of entries currently in the buffer. + pub fn len(&self) -> usize { + self.map.len() + } + + /// Whether the buffer is empty. + pub fn is_empty(&self) -> bool { + self.map.is_empty() + } + + /// Store a sent object. If the buffer is already at capacity the oldest + /// entry is evicted to make room. + pub fn put( + &mut self, + counter: u64, + ciphertext: Vec, + class: FlowClass, + object_id: u16, + now_ms: u64, + ) { + // Evict any entries that have passed their TTL before checking capacity. + self.evict_expired(now_ms); + + if self.map.len() >= self.max { + if let Some(old) = self.order.pop_front() { + self.map.remove(&old); + } + } + + self.map.insert( + counter, + Entry { + ciphertext, + class, + object_id, + inserted_ms: now_ms, + }, + ); + self.order.push_back(counter); + } + + /// Retrieve a stored object by send-counter. + /// + /// Returns `None` if: + /// - the entry does not exist, or + /// - the entry is older than `ttl_ms` (measured from `now_ms`). + pub fn get(&self, counter: u64, now_ms: u64) -> Option<(&[u8], FlowClass, u16)> { + let entry = self.map.get(&counter)?; + if now_ms.saturating_sub(entry.inserted_ms) > self.ttl_ms { + return None; + } + Some((&entry.ciphertext, entry.class, entry.object_id)) + } + + /// Evict all entries whose age exceeds `ttl_ms`. + fn evict_expired(&mut self, now_ms: u64) { + while let Some(front) = self.order.front() { + let expired = self + .map + .get(front) + .is_none_or(|e| now_ms.saturating_sub(e.inserted_ms) > self.ttl_ms); + if expired { + let k = self.order.pop_front().expect("front exists"); + self.map.remove(&k); + } else { + break; + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::FlowClass; + + #[test] + fn retx_put_get_roundtrip() { + let mut b = RetxBuffer::new(1024, 2000); + b.put(7, vec![1, 2, 3], FlowClass::Bulk, 99, 0); + let (ct, class, oid) = b.get(7, 100).expect("present"); + assert_eq!(ct, &[1, 2, 3]); + assert_eq!(class, FlowClass::Bulk); + assert_eq!(oid, 99); + } + + #[test] + fn retx_evicts_past_ttl() { + let mut b = RetxBuffer::new(1024, 2000); + b.put(7, vec![1], FlowClass::Bulk, 0, 0); + assert!(b.get(7, 3000).is_none(), "expired past ttl"); + } + + #[test] + fn retx_is_bounded_under_churn() { + let mut b = RetxBuffer::new(16, 1_000_000); + for c in 0..10_000u64 { + b.put(c, vec![0u8; 4], FlowClass::Bulk, 0, c); + } + assert!(b.len() <= 16); + } +} From 187dd7f64042cbccbb60749cae7888485dde027d Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 20:33:57 -0400 Subject: [PATCH 10/14] Reactive ARQ: retransmit fresh repair symbols for eligible NACKs - FecEncoder::repair_with_id: encode with an explicit object_id so the receiver's existing FEC decoder is topped up (not a new decode object). - Transport::repair_object: thin wrapper used by the daemon. - Unit test: retransmitted_repair_completes_a_missing_object (stall then top-up with extra_repair=4 symbols carrying the original object_id). - tunnel.rs egress: after encode(), put (counter, ciphertext, class, oid) into the shared RetxBuffer (TTL 2s, cap 1024). - tunnel.rs ingress control handler: after observe_loss, iterate report.missing; for each Bulk-class counter in the retx buffer, call repair_object and send the repair datagrams. Lock discipline enforced: retx lock is never held across transport or io calls. - run-netns-tunnel-loss.sh + ping_across_yipd_tunnel_under_loss: netem 10% loss on both veth legs; 10 pings must all succeed. Co-Authored-By: Claude Sonnet 4.6 --- bin/yipd/src/tunnel.rs | 96 +++++++++++- bin/yipd/tests/run-netns-tunnel-loss.sh | 196 ++++++++++++++++++++++++ bin/yipd/tests/tunnel_netns.rs | 22 +++ crates/yip-transport/src/fec.rs | 22 +++ crates/yip-transport/src/lib.rs | 43 ++++++ 5 files changed, 374 insertions(+), 5 deletions(-) create mode 100755 bin/yipd/tests/run-netns-tunnel-loss.sh diff --git a/bin/yipd/src/tunnel.rs b/bin/yipd/src/tunnel.rs index 50ccf22..ab9a2f0 100644 --- a/bin/yipd/src/tunnel.rs +++ b/bin/yipd/src/tunnel.rs @@ -28,7 +28,7 @@ use std::time::Instant; use yip_device::{DeviceKind, TunTap}; use yip_io::{set_socket_buffers, DataPlaneIo, PlainIo, MAX_DATAGRAM_BATCH, MAX_WIRE_DATAGRAM}; -use yip_transport::{FlowClass, LossDetector, LossReport, Transport}; +use yip_transport::{FlowClass, LossDetector, LossReport, RetxBuffer, Transport}; use yip_wire::{Codec, WireCodec}; use crate::config::Config; @@ -46,6 +46,13 @@ const FEEDBACK_INTERVAL_MS: u64 = 30; // the oldest entry is evicted before each new insertion, so memory is O(1). const SENT_LOG_CAPACITY: usize = 4096; +// ARQ retransmit buffer: maximum number of sent ciphertext objects to hold. +const RETX_BUFFER_MAX: usize = 1024; +// How long (ms) to keep a sent object available for retransmission. +const RETX_BUFFER_TTL_MS: u64 = 2000; +// Number of extra repair symbols to emit per ARQ retransmit. +const RETX_EXTRA_REPAIR: u32 = 4; + /// A bounded ring-log that maps sealed-packet counters to their `FlowClass`. /// /// Entries are inserted in arrival order (monotone counter sequence from the @@ -159,6 +166,13 @@ pub fn run(config: Config) -> io::Result<()> { // ingress thread also reads it periodically to emit feedback. let detector: Arc> = Arc::new(Mutex::new(LossDetector::new(5, 1024))); + // ARQ retransmit buffer: egress puts sent ciphertext objects here; ingress + // retrieves them by counter when a NACK arrives. + let retx: Arc> = Arc::new(Mutex::new(RetxBuffer::new( + RETX_BUFFER_MAX, + RETX_BUFFER_TTL_MS, + ))); + // ── create + split the TUN device ──────────────────────────────────────── let tun = TunTap::create(&config.device, DeviceKind::Tun).map_err(io::Error::other)?; let (mut tun_reader, mut tun_writer) = tun.split().map_err(io::Error::other)?; @@ -176,6 +190,7 @@ pub fn run(config: Config) -> io::Result<()> { let session_tx = Arc::clone(&session); let transport_tx = Arc::clone(&transport); let sent_log_tx = Arc::clone(&sent_log); + let retx_tx = Arc::clone(&retx); let egress = std::thread::Builder::new() .name("yipd-egress".into()) @@ -219,6 +234,18 @@ pub fn run(config: Config) -> io::Result<()> { .expect("sent_log lock poisoned") .insert(sealed.counter, class); + // Record in the ARQ retransmit buffer so NACKs can top up the + // receiver's decoder with fresh repair symbols. + if let Some(oid) = symbols.first().map(|s| s.object_id) { + retx_tx.lock().expect("retx lock poisoned").put( + sealed.counter, + sealed.ciphertext.clone(), + class, + oid, + now_ms, + ); + } + // Frame each symbol into a reused arena slot, then emit all // datagrams in one batched syscall (chunked by MAX_DATAGRAM_BATCH). let n_syms = symbols.len(); @@ -267,6 +294,7 @@ pub fn run(config: Config) -> io::Result<()> { let transport_rx = Arc::clone(&transport); let sent_log_rx = Arc::clone(&sent_log); let detector_rx = Arc::clone(&detector); + let retx_rx = Arc::clone(&retx); let ingress = std::thread::Builder::new() .name("yipd-ingress".into()) @@ -460,10 +488,68 @@ pub fn run(config: Config) -> io::Result<()> { let frac_bulk = fraction_f32(missing_bulk, sent_bulk); let frac_default = fraction_f32(missing_default, sent_default); - let mut t = transport_rx.lock().expect("transport lock poisoned"); - t.observe_loss(FlowClass::Realtime, frac_rt); - t.observe_loss(FlowClass::Bulk, frac_bulk); - t.observe_loss(FlowClass::Default, frac_default); + { + let mut t = + transport_rx.lock().expect("transport lock poisoned"); + t.observe_loss(FlowClass::Realtime, frac_rt); + t.observe_loss(FlowClass::Bulk, frac_bulk); + t.observe_loss(FlowClass::Default, frac_default); + } // transport lock released here + + // ARQ: for each missing counter reported by the peer, + // retransmit fresh repair symbols carrying the original + // object_id so the receiver's in-progress decoder is + // topped up rather than starting over. + // + // Lock discipline: never hold retx lock across transport + // or io operations — copy data out first. + for &counter in &report.missing { + let retx_entry = { + let buf = + retx_rx.lock().expect("retx lock poisoned"); + buf.get(counter, now_ms) + .map(|(ct, cls, oid)| (ct.to_vec(), cls, oid)) + }; + + let Some((ct, cls, oid)) = retx_entry else { + continue; + }; + if !cls.params().arq { + continue; + } + + // Generate fresh repair symbols with the original object_id. + let repair_syms = transport_rx + .lock() + .expect("transport lock poisoned") + .repair_object(&ct, cls, oid, RETX_EXTRA_REPAIR); + + // Frame and collect all repair datagrams, then send. + let mut repair_frames: Vec> = + Vec::with_capacity(repair_syms.len()); + for sym in &repair_syms { + let frame = wire_glue::symbol_to_frame( + conn_tag, sym, counter, cls, + ); + let dg = codec_rx.frame(&frame); + let mut pkt = + Vec::with_capacity(1 + dg.len()); + pkt.push(PacketType::Data as u8); + pkt.extend_from_slice(&dg); + repair_frames.push(pkt); + } + let slices: Vec<&[u8]> = repair_frames + .iter() + .map(Vec::as_slice) + .collect(); + for chunk in slices.chunks(MAX_DATAGRAM_BATCH) { + if let Err(e) = io.send_batch(chunk) { + eprintln!( + "yipd ingress: retransmit send error: {e}" + ); + } + } + } } _ => { diff --git a/bin/yipd/tests/run-netns-tunnel-loss.sh b/bin/yipd/tests/run-netns-tunnel-loss.sh new file mode 100755 index 0000000..e925bac --- /dev/null +++ b/bin/yipd/tests/run-netns-tunnel-loss.sh @@ -0,0 +1,196 @@ +#!/usr/bin/env bash +# End-to-end netns tunnel test for yipd under packet loss. +# Usage: run-netns-tunnel-loss.sh +# +# Creates two network namespaces (yipA, yipB) joined by a veth pair, +# applies 10% random packet loss via tc-netem on both veth interfaces, +# starts a yipd daemon in each, brings up TUN devices with tunnel IPs, +# then pings across the encrypted FEC+ARQ tunnel with a generous count. +# +# The test passes if all 10 pings succeed despite the underlying 10% loss, +# demonstrating that FEC + reactive ARQ together recover the missing packets. +set -euo pipefail + +YIPD="${1:?Usage: $0 }" +TMPDIR_TEST="$(mktemp -d /tmp/yipd-netns-loss-test.XXXXXX)" + +NS_A="yipA" +NS_B="yipB" +VETH_A="vethA" +VETH_B="vethB" +VETH_A_IP="10.0.0.1" +VETH_B_IP="10.0.0.2" +TUN_A_IP="10.9.0.1" +TUN_B_IP="10.9.0.2" +TUN_PREFIX="24" +VETH_PREFIX="24" +PORT_A="51820" +PORT_B="51821" +TUN_DEV="yip0" + +PID_A="" +PID_B="" + +cleanup() { + echo "[cleanup] killing daemons and removing namespaces" + [ -n "$PID_A" ] && kill "$PID_A" 2>/dev/null || true + [ -n "$PID_B" ] && kill "$PID_B" 2>/dev/null || true + # Give them a moment to die + sleep 0.2 + [ -n "$PID_A" ] && kill -9 "$PID_A" 2>/dev/null || true + [ -n "$PID_B" ] && kill -9 "$PID_B" 2>/dev/null || true + ip netns del "$NS_A" 2>/dev/null || true + ip netns del "$NS_B" 2>/dev/null || true + rm -rf "$TMPDIR_TEST" +} +trap cleanup EXIT + +# ── 1. generate keypairs ────────────────────────────────────────────────────── +echo "[setup] generating keypairs" +GENKEY_A="$("$YIPD" --genkey)" +GENKEY_B="$("$YIPD" --genkey)" + +PRIV_A="$(echo "$GENKEY_A" | grep '^private=' | cut -d= -f2)" +PUB_A="$(echo "$GENKEY_A" | grep '^public=' | cut -d= -f2)" +PRIV_B="$(echo "$GENKEY_B" | grep '^private=' | cut -d= -f2)" +PUB_B="$(echo "$GENKEY_B" | grep '^public=' | cut -d= -f2)" + +# ── 2. write config files ───────────────────────────────────────────────────── +CFG_A="$TMPDIR_TEST/yipA.conf" +CFG_B="$TMPDIR_TEST/yipB.conf" + +cat > "$CFG_A" < "$CFG_B" <"$LOG_A" 2>&1 & +PID_A=$! + +echo "[start] starting yipB (initiator)" +ip netns exec "$NS_B" "$YIPD" "$CFG_B" >"$LOG_B" 2>&1 & +PID_B=$! + +# ── 6. wait for handshake and TUN device creation ──────────────────────────── +# The daemons create the TUN device after a successful handshake. +# Poll for the TUN device to appear in each namespace (up to 30s under loss). +TUN_WAIT=30 +INTERVAL=0.25 + +echo "[wait] waiting for TUN devices to appear (up to ${TUN_WAIT}s)" +elapsed=0 +while true; do + A_UP=0 + B_UP=0 + ip netns exec "$NS_A" ip link show "$TUN_DEV" >/dev/null 2>&1 && A_UP=1 || true + ip netns exec "$NS_B" ip link show "$TUN_DEV" >/dev/null 2>&1 && B_UP=1 || true + + if [ "$A_UP" -eq 1 ] && [ "$B_UP" -eq 1 ]; then + echo "[wait] both TUN devices are up" + break + fi + + # Check if either daemon died unexpectedly + if ! kill -0 "$PID_A" 2>/dev/null; then + echo "[error] yipA daemon died unexpectedly" + echo "=== yipA log ===" + cat "$LOG_A" || true + exit 1 + fi + if ! kill -0 "$PID_B" 2>/dev/null; then + echo "[error] yipB daemon died unexpectedly" + echo "=== yipB log ===" + cat "$LOG_B" || true + exit 1 + fi + + elapsed=$(awk "BEGIN {print $elapsed + $INTERVAL}") + if awk "BEGIN {exit ($elapsed >= $TUN_WAIT) ? 0 : 1}"; then + echo "[error] timed out waiting for TUN devices" + echo "=== yipA log ===" + cat "$LOG_A" || true + echo "=== yipB log ===" + cat "$LOG_B" || true + exit 1 + fi + sleep "$INTERVAL" +done + +# ── 7. assign tunnel IPs ────────────────────────────────────────────────────── +echo "[setup] assigning tunnel IPs" +ip netns exec "$NS_A" ip addr add "${TUN_A_IP}/${TUN_PREFIX}" dev "$TUN_DEV" +ip netns exec "$NS_B" ip addr add "${TUN_B_IP}/${TUN_PREFIX}" dev "$TUN_DEV" + +# The TUN device is already brought up by the daemon (bring_up in TunTap::create). +# Verify the link is up; if not, bring it up explicitly. +ip netns exec "$NS_A" ip link show "$TUN_DEV" | grep -q "UP" || \ + ip netns exec "$NS_A" ip link set "$TUN_DEV" up +ip netns exec "$NS_B" ip link show "$TUN_DEV" | grep -q "UP" || \ + ip netns exec "$NS_B" ip link set "$TUN_DEV" up + +echo "[check] interface state in yipA:" +ip netns exec "$NS_A" ip addr show "$TUN_DEV" +echo "[check] interface state in yipB:" +ip netns exec "$NS_B" ip addr show "$TUN_DEV" + +# Brief additional settle time to ensure the data loops are ready +sleep 0.5 + +# ── 8. ping across the tunnel (10 pings, 3s each) ──────────────────────────── +echo "[test] pinging ${TUN_A_IP} from yipB across the tunnel under 10% loss" +if ip netns exec "$NS_B" ping -c 10 -W 3 "$TUN_A_IP"; then + echo "[PASS] ping succeeded under 10% netem loss" +else + PING_EXIT=$? + echo "[FAIL] ping failed under 10% loss (exit $PING_EXIT)" + echo "=== yipA log ===" + cat "$LOG_A" || true + echo "=== yipB log ===" + cat "$LOG_B" || true + exit "$PING_EXIT" +fi diff --git a/bin/yipd/tests/tunnel_netns.rs b/bin/yipd/tests/tunnel_netns.rs index 893c9dc..e6f4a22 100644 --- a/bin/yipd/tests/tunnel_netns.rs +++ b/bin/yipd/tests/tunnel_netns.rs @@ -21,3 +21,25 @@ fn ping_across_yipd_tunnel() { let status = Command::new("bash").arg(script).arg(yipd).status().unwrap(); assert!(status.success(), "netns tunnel ping failed"); } + +#[test] +fn ping_across_yipd_tunnel_under_loss() { + let is_root = Command::new("id") + .arg("-u") + .output() + .ok() + .and_then(|o| String::from_utf8(o.stdout).ok()) + .map(|s| s.trim() == "0") + .unwrap_or(false); + if !is_root { + eprintln!("SKIP ping_across_yipd_tunnel_under_loss: needs root"); + return; + } + let yipd = env!("CARGO_BIN_EXE_yipd"); + let script = concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/run-netns-tunnel-loss.sh" + ); + let status = Command::new("bash").arg(script).arg(yipd).status().unwrap(); + assert!(status.success(), "netns tunnel ping under 10% loss failed"); +} diff --git a/crates/yip-transport/src/fec.rs b/crates/yip-transport/src/fec.rs index e0f08cf..9e8975b 100644 --- a/crates/yip-transport/src/fec.rs +++ b/crates/yip-transport/src/fec.rs @@ -76,6 +76,28 @@ impl FecEncoder { .map(|p| split_packet(object_id, object_size, &p)) .collect() } + + /// Encode a ciphertext with an EXPLICIT `object_id` (used for retransmits), + /// returning source + `extra_repair` repair symbols, all carrying `object_id`. + pub fn repair_with_id( + &self, + ciphertext: &[u8], + params: crate::FlowParams, + object_id: u16, + extra_repair: u32, + ) -> Vec { + let object_size = u32::try_from(ciphertext.len()).expect("frame fits u32"); + let oti = ObjectTransmissionInformation::with_defaults( + u64::from(object_size), + params.symbol_size, + ); + let encoder = Encoder::new(ciphertext, oti); + encoder + .get_encoded_packets(extra_repair) + .into_iter() + .map(|p| split_packet(object_id, object_size, &p)) + .collect() + } } /// Emit the source symbols of `ciphertext` directly without constructing a diff --git a/crates/yip-transport/src/lib.rs b/crates/yip-transport/src/lib.rs index b9fb589..531f0e9 100644 --- a/crates/yip-transport/src/lib.rs +++ b/crates/yip-transport/src/lib.rs @@ -141,6 +141,24 @@ impl Transport { .push(symbol) } + /// Generate fresh RaptorQ repair symbols for a previously-sent object, + /// carrying the ORIGINAL `object_id` so the receiver's existing decoder + /// can be topped up rather than starting a new one. + /// + /// Returns all source + `extra_repair` repair symbols — enough that a + /// receiver that got zero original symbols can still reconstruct. + pub fn repair_object( + &mut self, + ciphertext: &[u8], + class: FlowClass, + object_id: u16, + extra_repair: u32, + ) -> Vec { + let params = class.params(); + self.encoder + .repair_with_id(ciphertext, params, object_id, extra_repair) + } + /// Feed an observed loss fraction into `class`'s controller. pub fn observe_loss(&mut self, class: FlowClass, loss: f32) { self.controllers[class_index(class)].observe_loss(loss); @@ -233,6 +251,31 @@ mod tests { assert!(late.is_none(), "late symbol returns None after completion"); } + #[test] + fn retransmitted_repair_completes_a_missing_object() { + let mut tx = Transport::new(vec![]); + let ct = vec![0x33u8; 2400]; // 2 source symbols + let (cls, syms) = tx.encode(&ct, &ct, false, 0); + let oid = syms[0].object_id; // the original object's identity + // Feed only the very first symbol; with 2 source symbols, 1 symbol is never enough. + let mut rx = Transport::new(vec![]); + let mut out = None; + for s in syms.iter().take(1) { + out = out.or(rx.decode(s, cls)); + } + assert!(out.is_none(), "one of N symbols -> not yet decoded"); + // Retransmit: fresh repair symbols carrying the SAME object_id top up the decoder. + let repair = tx.repair_object(&ct, cls, oid, 4); + assert!( + repair.iter().all(|s| s.object_id == oid), + "repair reuses object identity" + ); + for s in &repair { + out = out.or(rx.decode(s, cls)); + } + assert_eq!(out.as_deref(), Some(ct.as_slice())); + } + #[test] fn classify_ipv6_ef_maps_to_realtime() { let mut tx = Transport::new(vec![]); From cf27a4f4f29ed861c2d32603ef1996ea36206c47 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 20:37:36 -0400 Subject: [PATCH 11/14] Clarify ARQ repair round-trip test: feed one symbol, document proactive-repair rationale The Default class emits a proactive repair symbol, so a 2-source object encodes to 3 symbols; skipping one still decodes. Deliver exactly one symbol to guarantee the object is genuinely incomplete before the retransmit tops it up. Co-Authored-By: Claude Opus 4.8 --- crates/yip-transport/src/lib.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/crates/yip-transport/src/lib.rs b/crates/yip-transport/src/lib.rs index 531f0e9..c093480 100644 --- a/crates/yip-transport/src/lib.rs +++ b/crates/yip-transport/src/lib.rs @@ -257,15 +257,18 @@ mod tests { let ct = vec![0x33u8; 2400]; // 2 source symbols let (cls, syms) = tx.encode(&ct, &ct, false, 0); let oid = syms[0].object_id; // the original object's identity - // Feed only the very first symbol; with 2 source symbols, 1 symbol is never enough. + // Deliver only ONE symbol of a 2-source-symbol object; decode stalls. + // (The class emits 3 symbols — 2 source + 1 proactive repair — so + // skipping a single symbol would still leave enough to decode; feed + // exactly one to guarantee the object is genuinely incomplete.) let mut rx = Transport::new(vec![]); let mut out = None; for s in syms.iter().take(1) { out = out.or(rx.decode(s, cls)); } - assert!(out.is_none(), "one of N symbols -> not yet decoded"); + assert!(out.is_none(), "one symbol short -> not yet decoded"); // Retransmit: fresh repair symbols carrying the SAME object_id top up the decoder. - let repair = tx.repair_object(&ct, cls, oid, 4); + let repair = tx.repair_object(&ct, cls, oid, 2); assert!( repair.iter().all(|s| s.object_id == oid), "repair reuses object identity" From 8c90c1574848039cdcec8fb165c982f33b9735d0 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 20:41:48 -0400 Subject: [PATCH 12/14] Cap LossReport::decode missing at MAX_NACK (defense-in-depth) Defense-in-depth hardening: LossReport::decode now caps the decoded missing vec at MAX_NACK entries regardless of peer-supplied count, bounding retransmit work per report. Added decode_caps_missing_at_max_nack test that constructs raw bytes for a 100-entry report and verifies successful decode with truncation to MAX_NACK (64). Co-Authored-By: Claude Opus 4.8 --- crates/yip-transport/src/feedback.rs | 37 ++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/crates/yip-transport/src/feedback.rs b/crates/yip-transport/src/feedback.rs index f381306..da88bf0 100644 --- a/crates/yip-transport/src/feedback.rs +++ b/crates/yip-transport/src/feedback.rs @@ -41,6 +41,7 @@ impl LossReport { /// /// Returns `None` if the input is malformed (too short, or inconsistent /// missing count vs. payload length). + /// Caps the decoded `missing` vec at `MAX_NACK` entries (defense-in-depth). pub fn decode(bytes: &[u8]) -> Option { // Check minimum header length if bytes.len() < 14 { @@ -78,6 +79,9 @@ impl LossReport { missing.push(num); } + // Defense-in-depth: cap missing at MAX_NACK regardless of what peer sent. + missing.truncate(MAX_NACK); + Some(LossReport { delivered_count, high_counter, @@ -120,4 +124,37 @@ mod tests { let got = LossReport::decode(&r.encode()).expect("decodes"); assert_eq!(got.missing.len(), MAX_NACK); } + + #[test] + fn decode_caps_missing_at_max_nack() { + // Construct raw bytes for a report claiming n_missing = 100 + // Layout: [delivered:4][high:8][n_missing=100:2 BE][100 × 8-byte counters] + let delivered_count = 42u32; + let high_counter = 9999u64; + let n_missing = 100u16; + + let mut bytes = Vec::new(); + bytes.extend_from_slice(&delivered_count.to_be_bytes()); + bytes.extend_from_slice(&high_counter.to_be_bytes()); + bytes.extend_from_slice(&n_missing.to_be_bytes()); + + // Add 100 × 8-byte missing sequence numbers (just incrementing values) + for i in 0..100 { + bytes.extend_from_slice(&u64::try_from(i).unwrap().to_be_bytes()); + } + + // Decode should succeed and cap missing at MAX_NACK + let report = LossReport::decode(&bytes).expect("decode succeeds"); + assert_eq!(report.delivered_count, 42); + assert_eq!(report.high_counter, 9999); + assert_eq!( + report.missing.len(), + MAX_NACK, + "missing should be capped at MAX_NACK" + ); + // Verify the first MAX_NACK values are correctly decoded + for i in 0..MAX_NACK { + assert_eq!(report.missing[i], u64::try_from(i).unwrap()); + } + } } From dd5de9e964f0102baac01fb0435fe4c83ccbd626 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 20:48:45 -0400 Subject: [PATCH 13/14] Measure and record the feedback loop + ARQ milestone Phase A throughput win holds (bulk ratio 0.0000, clean-link ~457 Mbit/s). Loss recovery is unchanged with the feedback loop live: yip delivers 99.7%/99.0% UDP at 5%/10% netem loss (controller re-arms FEC on loss). Reactive ARQ retransmits Bulk NACKs with fresh repair symbols reusing the original object id; codec is unit-proven, wiring reviewed, tunnel survives 10% loss. A dedicated end-to-end Bulk-ARQ integrity harness is a tracked follow-up. Recorded in the bench README and CHANGELOG. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 13 +++++++++++++ crates/yip-bench/README.md | 22 ++++++++++++++++++++++ 2 files changed, 35 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea7d2e0..d5b35d2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,19 @@ All notable changes to this project are documented here, following ## [Unreleased] +### Added +- Adaptive loss-feedback loop + reactive ARQ. The receiver detects post-FEC + residual loss as gaps in the object counter and reports it (with NACKs) in an + authenticated `Control` packet; the sender attributes loss per class and drives + the repair controller. ARQ-eligible (`Bulk`) flows on a clean link now decay + their repair ratio to **zero**, activating the FEC-encode bypass — clean-link + single-stream TCP rises from ~273–285 to ~457 Mbit/s. On loss the controller + re-arms FEC instantly and NACKed `Bulk` objects are retransmitted with fresh + RaptorQ repair symbols (reusing the original object id); `Realtime`/`Default` + flows keep a proactive floor and are not retransmitted. New `yip-transport` + modules: `feedback` (`LossReport`), `lossdetect` (`LossDetector`), `retxbuf` + (`RetxBuffer`), plus `Transport::repair_object`. + ### Changed - Data-plane throughput pass: yipd now batches egress sends (`sendmmsg`) and ingress reads (`recvmmsg`) through yip-io's `PlainIo`, reuses framing buffers diff --git a/crates/yip-bench/README.md b/crates/yip-bench/README.md index b1cc83e..052240a 100644 --- a/crates/yip-bench/README.md +++ b/crates/yip-bench/README.md @@ -367,3 +367,25 @@ inferred from the throughput rise. The ~60 % jump is consistent with removing th FEC encode (which the per-stage profile showed is ~80 % of egress CPU) and sending one datagram per packet instead of two. Under loss the controller snaps repair back up immediately (Phase B's ARQ then backstops the residual). + +### Phase B — reactive ARQ + loss-recovery under the feedback loop + +With the feedback loop active, loss recovery is unchanged — the controller +re-arms FEC the instant loss is reported. Measured UDP delivery under `tc netem` +(`run-fec-compare.sh`), feedback loop live: + +| loss% | bare_recv% | yip_recv% | +|-------|------------|-----------| +| 0 | 100.0 | 100.0 | +| 5 | 94.7 | 99.7 | +| 10 | 89.9 | 99.0 | + +Reactive ARQ retransmits `Bulk` objects the receiver NACKs, using fresh RaptorQ +repair symbols that carry the original object id (so the receiver tops up its +existing decoder). The retransmit codec is unit-proven (`repair_object` completes +an object delivered only one symbol short), the daemon wiring was reviewed for +object-id preservation and lock discipline, and the tunnel stays alive under 10 % +netem loss (netns ping 10/10). A dedicated end-to-end harness that forces +FEC-insufficiency and asserts ARQ-specific `Bulk` recovery (establishing the +tunnel *before* applying loss, with a retransmit buffer sized to the flow rate) +is a tracked follow-up. From cfea8f8325125a78c493b1d88b228ce5728273db Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Tue, 30 Jun 2026 20:58:01 -0400 Subject: [PATCH 14/14] Authenticate control packets before touching the loss detector; raise yip-transport coverage to >=90% Fix 1 (security): in tunnel.rs ingress, the control-packet path called detector.on_seen(counter, now_ms) BEFORE session.open(counter, ct), letting a forged packet poison the loss detector's high_counter even when authentication fails. Move on_seen to run only inside the Ok branch of session.open. Data-path on_seen calls are unchanged. Fix 2 (coverage): yip-transport line coverage was 89.53%; add four focused unit tests in fec.rs covering repair_with_id, the late/duplicate symbol guard (state.done), the SBN out-of-range guard for already-tracked objects, and the full-Encoder encode path. Line coverage: 90.05%. Co-Authored-By: Claude Opus 4.8 --- bin/yipd/src/tunnel.rs | 19 +++--- crates/yip-transport/src/fec.rs | 112 ++++++++++++++++++++++++++++++++ 2 files changed, 123 insertions(+), 8 deletions(-) diff --git a/bin/yipd/src/tunnel.rs b/bin/yipd/src/tunnel.rs index ab9a2f0..3d2b1e6 100644 --- a/bin/yipd/src/tunnel.rs +++ b/bin/yipd/src/tunnel.rs @@ -409,14 +409,9 @@ pub fn run(config: Config) -> io::Result<()> { ); let ct = &dg[9..]; - // Notify the detector that we saw this control counter too, - // so the unified counter sequence stays consistent. - detector_rx - .lock() - .expect("detector lock poisoned") - .on_seen(counter, now_ms); - - // Decrypt the control payload. + // Decrypt the control payload FIRST — authenticate before + // any side-effect. A forged packet with a bogus counter + // must not poison the loss detector's high_counter. let plaintext = match session_rx .lock() .expect("session lock poisoned") @@ -429,6 +424,14 @@ pub fn run(config: Config) -> io::Result<()> { } }; + // Authentication succeeded: notify the detector that we + // saw this control counter, keeping the unified counter + // sequence consistent. + detector_rx + .lock() + .expect("detector lock poisoned") + .on_seen(counter, now_ms); + // Decode the LossReport. let report = match LossReport::decode(&plaintext) { Some(r) => r, diff --git a/crates/yip-transport/src/fec.rs b/crates/yip-transport/src/fec.rs index 9e8975b..dfa16c4 100644 --- a/crates/yip-transport/src/fec.rs +++ b/crates/yip-transport/src/fec.rs @@ -524,6 +524,118 @@ mod tests { // --- End malformed-input tests ----------------------------------------------- + /// C2 (existing object): A second symbol for an already-tracked object whose + /// SBN is out of range must be rejected with None, not passed to raptorq. + #[test] + fn push_out_of_range_sbn_for_existing_object_returns_none() { + let params = FlowClass::Default.params(); + let ct = vec![0x42u8; 1200]; + let mut enc = FecEncoder::new(); + // Encode with repair symbols so we have multiple symbols for the same object. + let syms = enc.encode(&ct, params, 4); + let mut re = FecReassembler::new(params.symbol_size, 64); + + // Feed the first (valid) symbol to register the object. + re.push(&syms[0]); + assert_eq!(re.in_flight(), 1, "object should be registered"); + + // Now craft a symbol with the same object_id but SBN out of range. + let mut bad = syms[0].clone(); + bad.payload_id[0] = 255; // SBN 255 is always out of range for a single-block object + assert_eq!( + re.push(&bad), + None, + "out-of-range SBN on tracked object must be rejected" + ); + } + + /// Late/duplicate symbol after the object has already decoded must return None. + #[test] + fn push_late_symbol_after_decode_returns_none() { + let params = FlowClass::Default.params(); + let ct = vec![0xABu8; 1200]; + let mut enc = FecEncoder::new(); + let syms = enc.encode(&ct, params, 4); + let mut re = FecReassembler::new(params.symbol_size, 64); + + // Feed all symbols until the object decodes. + let mut decoded = false; + for s in &syms { + if re.push(s).is_some() { + decoded = true; + break; + } + } + assert!(decoded, "object must decode"); + + // Now push another symbol for the same object_id — must get None. + let result = re.push(&syms[0]); + assert_eq!(result, None, "late symbol after decode must return None"); + } + + /// `repair_with_id` encodes with an explicit object_id and the symbols + /// carry that id; the reassembler can decode the result. + #[test] + fn repair_with_id_produces_decodable_symbols() { + let params = FlowClass::Default.params(); + let ct: Vec = (0..3000u32) + .map(|i| u8::try_from(i % 251).unwrap()) + .collect(); + let enc = FecEncoder::new(); + // Use explicit object_id 42. + let repair_syms = enc.repair_with_id(&ct, params, 42, 8); + assert!(!repair_syms.is_empty(), "must produce symbols"); + assert!( + repair_syms.iter().all(|s| s.object_id == 42), + "all symbols must carry object_id 42" + ); + assert!( + repair_syms.iter().all(|s| s.object_size == 3000), + "all symbols must carry correct object_size" + ); + + // Verify the symbols decode correctly. + let mut re = FecReassembler::new(params.symbol_size, 64); + let mut out = None; + for s in &repair_syms { + if let Some(o) = re.push(s) { + out = Some(o); + break; + } + } + assert_eq!( + out.as_deref(), + Some(ct.as_slice()), + "repair_with_id symbols must decode to original" + ); + } + + /// Exercises the `encode` full-encoder fallthrough path (repair > 0 when + /// sub_blocks == 1 means the fast bypass is skipped and Encoder::new is used). + #[test] + fn encode_with_repair_uses_full_encoder_and_decodes() { + let params = FlowClass::Realtime.params(); + let ct = vec![0x55u8; 600]; + let mut enc = FecEncoder::new(); + // repair > 0 takes the full Encoder path even for small objects. + let syms = enc.encode(&ct, params, 2); + assert!(!syms.is_empty()); + + let mut re = FecReassembler::new(params.symbol_size, 64); + let mut out = None; + for s in &syms { + if let Some(o) = re.push(s) { + out = Some(o); + break; + } + } + assert_eq!( + out.as_deref(), + Some(ct.as_slice()), + "full-encoder path must decode" + ); + } + #[test] fn pipelines_two_objects_and_evicts_when_full() { let mut enc = FecEncoder::new();