feat(congestion): actually pace sends instead of releasing whole windows

pacing_rate had been computed on every ACK since the controller was written
and never read by anything: admission was decided purely by cwnd. But cwnd
bounds how much may be UNACKNOWLEDGED, not how fast it reaches the wire, so a
full window went out back-to-back. On a bottleneck with a deep buffer that
burst is absorbed rather than dropped, and it lands as standing queue — the
mechanism behind the multi-second RTT this protocol has been showing on
mobile. It is also why BBR could not simply be dropped in: BBR's whole model
is "send at the estimated bottleneck rate", which is meaningless without a
pacer underneath it.

Adds a token bucket to CongestionController, charged in on_send so every byte
that reaches the wire pays exactly once — retransmits included, since those
are precisely what must not bypass the limit and pile into a full queue.

Burst allowance is 10ms-at-rate rather than one packet. Pacing intervals here
are fractions of a millisecond, so strict per-packet release would need a
sub-millisecond timer per packet; sizing the burst to the loop's existing
~10ms wakeup lets the configured rate still be saturated. There is a floor of
4 MTU so a cold or collapsed rate estimate can never wedge sending entirely.

Wired into both directions. The client gates its proxy-event branch on it,
and the server's per-session backpressure snapshot reports zero headroom when
the bucket is empty. The download path matters most here — that is the one
carrying video — and it was also still clamped to the old 16384-packet
ceiling (~20 MB outstanding), now aligned with MAX_CWND_PACKETS.

Tested that the bucket denies once drained and refills over time; the second
is what keeps a stalled bucket from wedging sending permanently.

ostp-client/src/bridge.rs

@@ -309,7 +309,15 @@ impl Bridge {
                     // which on a mobile uplink is minutes of buffered queue rather
                     // than throughput — the app kept handing over data long after
                     // the path had stopped draining it.
-                    s.iter().any(|ses| ses.machine.in_flight_count() < ses.machine.cwnd_packets().clamp(16, 1024))
+                    // Two independent gates. cwnd bounds how much may be in
+                    // flight; pacing bounds how FAST it is released. Without the
+                    // second, a full window goes out back-to-back and lands in
+                    // the bottleneck's buffer as standing queue rather than
+                    // throughput — the thing that produced multi-second RTT.
+                    s.iter().any(|ses| {
+                        ses.machine.in_flight_count() < ses.machine.cwnd_packets().clamp(16, 1024)
+                            && ses.machine.can_pace_packet()
+                    })
                 }).unwrap_or(true) => {
                     self.handle_proxy_event(proxy_ev, &mut sessions_opt, &mut stream_map, &tx, &proxy_tx).await;
                 }

ostp-core/src/congestion.rs

@@ -39,6 +39,9 @@ pub struct CongestionController {
     loss_count: u32,
     /// Pacing rate: bytes per second
     pacing_rate: u64,
+    /// Token-bucket allowance for pacing, in bytes.
+    pacing_tokens: f64,
+    pacing_last_refill: Instant,
     /// MTU estimate (used for cwnd → packet count conversion)
     mtu: u64,
     /// Min RTT expiry: re-probe after 10 seconds
@@ -77,6 +80,8 @@ const MAX_CWND_PACKETS: u64 = 1024;
 const RTT_INFLATION_EXIT_SLOW_START: f64 = 2.0;
 /// SRTT/min_rtt ratio treated as a standing queue that must be actively drained.
 const RTT_INFLATION_BACKOFF: f64 = 4.0;
+/// How much pacing allowance may accumulate, expressed as time-at-rate.
+const PACING_BURST: Duration = Duration::from_millis(10);
 const RTO_MIN: Duration = Duration::from_millis(50);
 /// Maximum RTO
 const RTO_MAX: Duration = Duration::from_secs(16);
@@ -125,9 +130,50 @@ impl CongestionController {
             min_rtt_stamp: now,
             slow_start_losses: 0,
             slow_start_loss_window_start: now,
+            pacing_tokens: (INITIAL_CWND_PACKETS * mtu) as f64,
+            pacing_last_refill: now,
         }
     }
 
+    /// Bytes of pacing allowance available right now, without consuming any.
+    ///
+    /// Read-only so the send path can use it as an admission check before it
+    /// commits to building a datagram.
+    pub fn pacing_available(&self) -> f64 {
+        let elapsed = self.pacing_last_refill.elapsed().as_secs_f64();
+        (self.pacing_tokens + elapsed * self.pacing_rate as f64).min(self.pacing_burst())
+    }
+
+    /// Whether at least one full-size packet may be released right now.
+    pub fn can_pace_packet(&self) -> bool {
+        self.pacing_available() >= self.mtu as f64
+    }
+
+    /// Ceiling on accumulated allowance.
+    ///
+    /// Pacing intervals here are fractions of a millisecond, so releasing
+    /// strictly one packet at a time would need a sub-millisecond timer per
+    /// packet. Instead we allow a short burst — the same trade every real
+    /// pacing implementation makes — sized so the loop's existing ~10ms wakeups
+    /// can still saturate the configured rate, with a small floor so a
+    /// cold/low estimate can never wedge sending entirely.
+    fn pacing_burst(&self) -> f64 {
+        let by_rate = self.pacing_rate as f64 * PACING_BURST.as_secs_f64();
+        by_rate.max((self.mtu * 4) as f64)
+    }
+
+    /// Refill from elapsed time and deduct `bytes`. Called on the real send
+    /// path; allowance is permitted to go negative so an oversized packet still
+    /// pays for itself rather than being released for free.
+    fn consume_pacing(&mut self, bytes: u64) {
+        let now = Instant::now();
+        let elapsed = now.duration_since(self.pacing_last_refill).as_secs_f64();
+        self.pacing_last_refill = now;
+        self.pacing_tokens =
+            (self.pacing_tokens + elapsed * self.pacing_rate as f64).min(self.pacing_burst())
+                - bytes as f64;
+    }
+
     /// Returns the current congestion window in bytes.
     pub fn cwnd(&self) -> u64 {
         self.cwnd
@@ -179,6 +225,11 @@ impl CongestionController {
     /// Record that we sent `bytes` of data.
     pub fn on_send(&mut self, bytes: u64) {
         self.bytes_in_flight = self.bytes_in_flight.saturating_add(bytes);
+        // Charge the pacing bucket here rather than at the admission check, so
+        // every byte that actually reaches the wire is paid for exactly once —
+        // including retransmits, which are precisely what must not be allowed
+        // to bypass the rate limit and pile into an already-full queue.
+        self.consume_pacing(bytes);
     }
 
     /// Record that `bytes` were acknowledged but WITHOUT a usable RTT sample
@@ -425,6 +476,47 @@ mod tests {
         );
     }
 
+    /// Pacing must actually bound the release rate: draining the bucket has to
+    /// deny the next packet. Without this the congestion window alone decides,
+    /// and a whole window leaves back-to-back.
+    #[test]
+    fn test_pacing_bucket_denies_once_drained() {
+        let mut cc = CongestionController::new(1200);
+        assert!(cc.can_pace_packet(), "a fresh controller must allow sending");
+
+        // Spend well beyond one burst allowance.
+        let burst_bytes = cc.pacing_available();
+        let mut spent = 0.0;
+        while spent <= burst_bytes + 1200.0 {
+            cc.on_send(1200);
+            spent += 1200.0;
+        }
+
+        assert!(
+            !cc.can_pace_packet(),
+            "pacing allowed unbounded sending: {} bytes still available after spending {}",
+            cc.pacing_available(),
+            spent
+        );
+    }
+
+    /// The allowance must refill over time, or sending would stall permanently
+    /// once the first burst is spent.
+    #[test]
+    fn test_pacing_bucket_refills_over_time() {
+        let mut cc = CongestionController::new(1200);
+        while cc.can_pace_packet() {
+            cc.on_send(1200);
+        }
+        assert!(!cc.can_pace_packet());
+
+        std::thread::sleep(Duration::from_millis(25));
+        assert!(
+            cc.can_pace_packet(),
+            "pacing bucket never refilled; sending would be stuck forever"
+        );
+    }
+
     /// cwnd must never exceed the absolute ceiling, however long slow start
     /// runs unopposed — above it the window is buffered queue, not throughput.
     #[test]

ostp-core/src/protocol.rs

@@ -188,6 +188,16 @@ impl ProtocolMachine {
         self.cc.cwnd_packets() as usize
     }
 
+    /// Whether the pacing bucket currently allows releasing another packet.
+    ///
+    /// The congestion window bounds how much may be UNACKNOWLEDGED; it says
+    /// nothing about how fast that window is emptied onto the wire. Sending a
+    /// whole window back-to-back is what drives a deep buffer into standing
+    /// queue, so admission is gated on both.
+    pub fn can_pace_packet(&self) -> bool {
+        self.cc.can_pace_packet()
+    }
+
     pub fn on_send(&mut self, bytes: u64) {
         self.cc.on_send(bytes);
     }

ostp-server/src/dispatcher.rs

@@ -263,8 +263,19 @@ impl Dispatcher {
         self.peer_machines
             .iter()
             .map(|(&sid, ps)| {
-                let cwnd = (ps.machine.cwnd_packets() as i64).clamp(16, 16384);
+                // Ceiling matches MAX_CWND_PACKETS in ostp-core. The old 16384
+                // allowed ~20 MB outstanding toward one client — on a mobile
+                // downlink that is standing queue, not throughput, and it is the
+                // download direction that carries video.
+                let cwnd = (ps.machine.cwnd_packets() as i64).clamp(16, 1024);
                 let in_flight = ps.machine.in_flight_count() as i64;
+                // Pacing gates the RATE, cwnd only the outstanding amount. With
+                // the pacing bucket empty, report no headroom so the relay
+                // reader pauses instead of handing over another chunk that would
+                // leave back-to-back.
+                if !ps.machine.can_pace_packet() {
+                    return (sid, 0);
+                }
                 (sid, cwnd - in_flight)
             })
             .collect()