perf(transport): raise the per-path transmit queue quota to 1 MiB

crates/mq-integration/src/shard_pair.rs

@@ -540,8 +540,14 @@ pub fn spawn_server(cfg: ServerConfig, max_conns: u32) -> Side<Server> {
 
 /// The real `Client` at `client_addr()` with SOCKS5 and transparent listeners.
 pub fn spawn_client(cfg: ClientConfig) -> Side<Client> {
+    spawn_client_cc(cfg, CongestionControl::Bbr)
+}
+
+/// `spawn_client` with another congestion control.
+pub fn spawn_client_cc(cfg: ClientConfig, cc: CongestionControl) -> Side<Client> {
     let mut tc = transport_cfg(Role::Client, 0);
     tc.scheduler = cfg.scheduler;
+    tc.cc = cc;
     Side::spawn(tc, client_addr(), vec![SOCKS5, TRANSPARENT], move || {
         Client::new(cfg)
     })

crates/mq-integration/tests/pair_relay.rs

@@ -8,7 +8,7 @@ use mq_proxy::config::ClientConfig;
 use mq_proxy::server::Server;
 use mq_runtime::testing::Op;
 use mq_transport_api::fabric::Rule;
-use mq_transport_api::{ConnId, PathId, Time, TransportOps};
+use mq_transport_api::{CongestionControl, ConnId, PathId, Time, TransportOps};
 
 /// spec §8.1: authentication, then a SOCKS5 request relayed to an echo origin.
 #[test]
@@ -36,12 +36,12 @@ fn pair_auth_and_relay() {
 const ACTIVE: u32 = 2; // XQC_PATH_STATE_ACTIVE
 
 /// A client with a second `--path`, authenticated, both paths active.
-fn two_paths() -> (Pair<Server, Client>, ConnId) {
+fn two_paths(cc: CongestionControl) -> (Pair<Server, Client>, ConnId) {
     let cfg = ClientConfig {
         paths: vec![client_addr().ip(), CLIENT_IP2],
         ..client_cfg()
     };
-    let mut p = Pair::new(spawn_server(server_cfg(), 0), spawn_client(cfg));
+    let mut p = Pair::new(spawn_server(server_cfg(), 0), spawn_client_cc(cfg, cc));
     assert!(p.run_until(5 * SEC, |p| p.auth_attempts() == 1));
     let c = p.client_conns()[0];
     let up = move |p: &mut Pair<Server, Client>| {
@@ -59,7 +59,7 @@ fn two_paths() -> (Pair<Server, Client>, ConnId) {
 /// socket and both paths carry the relayed bytes.
 #[test]
 fn pair_two_paths_carry_traffic() {
-    let (mut p, c) = two_paths();
+    let (mut p, c) = two_paths(CongestionControl::Bbr);
     const N: usize = 1 << 20;
     p.origin(OriginMode::Send(bulk(N)));
     let s = p.socks_open(origin_addr(), b"");
@@ -83,10 +83,11 @@ fn pair_two_paths_carry_traffic() {
 /// window has a bandwidth-delay product to grow into.
 #[test]
 fn pair_blocked_path_quota_and_resume() {
-    const QUOTA: usize = 256 * 1024; // spec §4.4
+    const QUOTA: usize = 1024 * 1024; // spec §4.4
     const WARM: usize = 8 << 20;
     const N: usize = 8 << 20;
-    let (mut p, c) = two_paths();
+    // Cubic: in the fabric BBR's window on path 1 settles below one queue quota.
+    let (mut p, c) = two_paths(CongestionControl::Cubic);
     p.fabric.add_rule(Rule::DelayRange {
         min: 5 * MS,
         max: 5 * MS,
@@ -175,7 +176,7 @@ fn pair_blocked_path_quota_and_resume() {
 /// the primary) and both paths are active again.
 fn blackhole_one_path_then_recover(ip: std::net::IpAddr, dead: u64) {
     const CLOSED: u32 = 4; // XQC_PATH_STATE_CLOSED
-    let (mut p, c) = two_paths();
+    let (mut p, c) = two_paths(CongestionControl::Bbr);
     let paths = move |p: &Pair<Server, Client>| {
         let st = p.with_client(move |n| n.transport().conn_stats(c)).unwrap();
         st.paths.iter().map(|x| (x.id, x.state)).collect::<Vec<_>>()

crates/mq-transport/src/txq.rs

@@ -3,10 +3,12 @@ use std::net::SocketAddr;
 
 use mq_transport_api::{ConnId, Transmit, TxKey};
 
-// spec §4.4
-pub const QUEUE_QUOTA: usize = 256 * 1024;
-pub const QUEUE_LOW: usize = 128 * 1024;
-pub const TOTAL_HIGH: usize = 2 * 1024 * 1024;
+// spec §4.4. One drive can hand xquic's output for a path over 256 KiB at ~1 Gbps; each
+// refusal below makes xquic retry on every wakeup, so the quota sits well above that.
+// Queues grow on use, so an idle (conn, path) costs nothing.
+pub const QUEUE_QUOTA: usize = 1024 * 1024;
+pub const QUEUE_LOW: usize = 512 * 1024;
+pub const TOTAL_HIGH: usize = 4 * 1024 * 1024;
 
 #[derive(Clone, Copy, PartialEq, Eq, Debug)]
 pub enum Refusal {
@@ -37,7 +39,7 @@ struct Queue {
 impl Queue {
     fn new() -> Self {
         Self {
-            buf: Vec::with_capacity(QUEUE_QUOTA + 1500),
+            buf: Vec::new(),
             head: 0,
             tail: 0,
             pkts: VecDeque::new(),
@@ -47,7 +49,7 @@ impl Queue {
 
     fn append(&mut self, dst: SocketAddr, pkt: &[u8]) {
         if self.tail + pkt.len() > self.buf.capacity() {
-            self.compact(); // memmove only here
+            self.compact(); // memmove only here; the extend below grows the buffer if still short
         }
         self.buf.truncate(self.tail);
         self.buf.extend_from_slice(pkt);
@@ -233,6 +235,11 @@ mod tests {
     fn addr(n: u8) -> SocketAddr {
         SocketAddr::from(([10, 0, 0, n], 4433))
     }
+    /// Keys that fill to exactly TOTAL_HIGH; key(N) is the next one.
+    const N: u32 = (TOTAL_HIGH / QUEUE_QUOTA) as u32;
+    /// 1024-byte packets to drain a full queue to exactly QUEUE_LOW.
+    const TO_LOW: usize = (QUEUE_QUOTA - QUEUE_LOW) / 1024;
+
     /// Fill `k` to exactly QUEUE_QUOTA with 1024-byte packets.
     fn fill(t: &mut TxQueues, k: TxKey) {
         for _ in 0..QUEUE_QUOTA / 1024 {
@@ -301,9 +308,18 @@ mod tests {
         assert_eq!((q.head, q.tail, q.bytes, t.total), (0, 0, 0, 0));
     }
 
+    #[test]
+    fn fresh_queue_reserves_nothing_and_grows_with_use() {
+        let mut q = Queue::new();
+        assert_eq!(q.buf.capacity(), 0);
+        q.append(addr(1), &[0; 1200]);
+        assert!((1200..QUEUE_QUOTA).contains(&q.buf.capacity()));
+    }
+
     #[test]
     fn compaction_only_when_capacity_reached() {
         let mut q = Queue::new();
+        q.buf.reserve_exact(64 * 1024);
         let cap = q.buf.capacity();
         let p = [9u8; 1000];
         while q.tail + 1000 * 2 <= cap {
@@ -335,7 +351,7 @@ mod tests {
         let k = key(0);
         fill(&mut t, k);
         t.record_blocked(conn(0), k, Refusal::Quota);
-        t.done(k, 128); // exactly at QUEUE_LOW: not below
+        t.done(k, TO_LOW); // exactly at QUEUE_LOW: not below
         assert!(!t.resume_pending());
         t.done(k, 1);
         assert!(t.resume_pending());
@@ -346,24 +362,24 @@ mod tests {
     #[test]
     fn total_refuses_at_high_water() {
         let mut t = TxQueues::default();
-        for i in 0..8 {
+        for i in 0..N {
             fill(&mut t, key(i));
         }
         assert_eq!(t.total, TOTAL_HIGH);
-        assert_eq!(t.push(key(8), addr(1), &[0; 1]), Err(Refusal::Total));
+        assert_eq!(t.push(key(N), addr(1), &[0; 1]), Err(Refusal::Total));
     }
 
     #[test]
     fn total_resumes_on_any_commit_below_high_even_with_two_keys_at_quota() {
         let mut t = TxQueues::default();
-        for i in 0..8 {
+        for i in 0..N {
             fill(&mut t, key(i));
         }
         t.record_blocked(conn(0), key(0), Refusal::Quota);
         t.record_blocked(conn(1), key(1), Refusal::Quota);
-        t.record_blocked(conn(8), key(8), Refusal::Total);
+        t.record_blocked(conn(N), key(N), Refusal::Total);
         t.done(key(3), 1); // unrelated key; total now below high
-        assert_eq!(t.take_resumable(), vec![conn(8)]);
+        assert_eq!(t.take_resumable(), vec![conn(N)]);
         assert!(t.blocked.contains_key(&conn(0)) && t.blocked.contains_key(&conn(1)));
     }
 
@@ -375,7 +391,7 @@ mod tests {
         // blocked on two quota keys: draining only A resumes it
         t.record_blocked(conn(0), key(0), Refusal::Quota);
         t.record_blocked(conn(0), key(1), Refusal::Quota);
-        t.done(key(0), 129);
+        t.done(key(0), TO_LOW + 1);
         assert_eq!(t.take_resumable(), vec![conn(0)]);
         assert!(!t.blocked.contains_key(&conn(0)));
         // quota + total: a commit below high-water clears via total while the quota key stays full
@@ -399,17 +415,17 @@ mod tests {
     #[test]
     fn drop_conn_counts_as_commit_and_forgets_block() {
         let mut t = TxQueues::default();
-        for i in 0..8 {
+        for i in 0..N {
             fill(&mut t, key(i));
         }
         t.push_or_drop((Some(conn(0)), PathId(1)), addr(1), &[0; 1]);
         t.record_blocked(conn(0), key(0), Refusal::Quota);
-        t.record_blocked(conn(8), key(8), Refusal::Total);
+        t.record_blocked(conn(N), key(N), Refusal::Total);
         t.drop_conn(conn(0));
-        assert_eq!(t.total, 7 * QUEUE_QUOTA);
+        assert_eq!(t.total, (N as usize - 1) * QUEUE_QUOTA);
         assert!(t.queues.keys().all(|k| k.0 != Some(conn(0))));
         assert!(!t.blocked.contains_key(&conn(0)));
-        assert_eq!(t.take_resumable(), vec![conn(8)]);
+        assert_eq!(t.take_resumable(), vec![conn(N)]);
     }
 
     #[test]

crates/mq-transport/tests/fabric_two_blocked_paths.rs

@@ -8,7 +8,7 @@ use common::xfer::Xfer;
 use mq_transport_api::{CongestionControl, Event, PathId, Role, TransportOps, TxKey};
 
 /// spec §4.4 per-queue quota.
-const QUEUE_QUOTA: usize = 256 * 1024;
+const QUEUE_QUOTA: usize = 1024 * 1024;
 
 #[test]
 fn two_blocked_paths_one_drains_resumes_at_once() {

crates/mq-transport/tests/fabric_txq.rs

@@ -1,18 +1,21 @@
 //! spec §4.4, §8.4 "Transmit queue full, then drained".
 mod common;
 
-use common::pair::{MS, Pair};
+use common::pair::{MS, Opts, Pair};
 use common::xfer::Xfer;
-use mq_transport_api::{PathId, TransportOps};
+use mq_transport_api::{CongestionControl, PathId, TransportOps};
 
 /// spec §4.4 per-queue quota.
-const QUEUE_QUOTA: usize = 256 * 1024;
+const QUEUE_QUOTA: usize = 1024 * 1024;
 
 #[test]
 fn txq_full_then_drained_resumes() {
-    let mut p = Pair::new();
+    // Cubic: in the fabric BBR's window settles below one queue quota; cubic's keeps growing.
+    let mut o = Opts::default();
+    o.client.cc = CongestionControl::Cubic;
+    let mut p = Pair::with(o);
     let key = (Some(p.conn), PathId(0));
-    let mut x = Xfer::start(&p, 16 * 1024 * 1024);
+    let mut x = Xfer::start(&p, 64 * 1024 * 1024);
     // Grow the congestion window past the queue quota first, or the window, not the queue,
     // stops the sender.
     assert!(p.pump_until(MS, 20_000, |p| {