From 8f4e34d95618d1ab7e986fbefa42871b2f30d7cf Mon Sep 17 00:00:00 2001
From: Mikkeli <mikkeli@novoyuuparosk.org>
Date: Sun, 11 Oct 2026 16:33:55 +0900
Subject: [PATCH] Link test: download series (bulk sampled, warm bursts, paced)
 and burst-vs-idle; car results

Orin -> tablet over NCM: paced streams up to 4 MB/s are clean; unpaced
bursts and bulk hit multi-second holds that inflate the RTO to 10-16 s.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz
---
 .../kotlin/dev/yukart/display/MainActivity.kt |   2 +-
 docs/impl/reports/LINK-TEST-2026-10-11.md     |  58 +++++
 lib/linktest.dart                             | 213 ++++++++++++++++++
 lib/main.dart                                 |  12 +-
 tool/linktest_server.py                       |  41 +++-
 tool/linktest_summary.py                      |  16 +-
 6 files changed, 330 insertions(+), 12 deletions(-)

diff --git a/android/app/src/main/kotlin/dev/yukart/display/MainActivity.kt b/android/app/src/main/kotlin/dev/yukart/display/MainActivity.kt
index 3cd6986..83229b4 100644
--- a/android/app/src/main/kotlin/dev/yukart/display/MainActivity.kt
+++ b/android/app/src/main/kotlin/dev/yukart/display/MainActivity.kt
@@ -22,7 +22,7 @@ import io.flutter.plugin.common.MethodChannel
  *   When asked, the whole process is bound to the Ethernet (NCM) network while it
  *   exists, and unbound when it goes.
  *
- * Also a development hook: `am start -n dev.yukart.display/.MainActivity --es linktest <label>[@<host>]`
+ * Also a development hook: `am start -n dev.yukart.display/.MainActivity --es linktest [dl:|bi:]<label>[@<host>]`
  * runs the link test (lib/linktest.dart) without touching the screen.
  */
 class MainActivity : FlutterActivity() {
diff --git a/docs/impl/reports/LINK-TEST-2026-10-11.md b/docs/impl/reports/LINK-TEST-2026-10-11.md
index 274a746..4e01c50 100644
--- a/docs/impl/reports/LINK-TEST-2026-10-11.md
+++ b/docs/impl/reports/LINK-TEST-2026-10-11.md
@@ -61,3 +61,61 @@ adb -s localhost:15555 shell am start -n dev.yukart.display/.MainActivity --es l
 
 The host defaults to the gateway's. A run takes about 75 s and reports each
 result to the Orin as it goes.
+
+## Download in the car setup (PD hub + charger)
+
+Two identical download series (`dl:`), then a burst-vs-idle series (`bi:`). Raw
+records under the labels `car-dl1`, `car-dl2`, `car-bi1`.
+
+**Paced streams are clean.** 30 frames/s at 0.5, 1, 2 and 4 MB/s, 10 s each, in
+both series: every frame arrived, the full rate was delivered, and queueing never
+exceeded 44 ms. A camera-like stream up to 4 MB/s (32 Mbit/s) fits.
+
+**Unpaced traffic is not:**
+
+| | Series 1 | Series 2 |
+|---|---|---|
+| Bulk, 5 × 10 s: rate while flowing | 5.6 MB/s | 5.6 MB/s |
+| Bulk: longest stall per run | 1.8, 14.0, 3.6, 0.3, 0.8 s | 0.2, 3.7, 0.8, 0.8, 14.4 s |
+| Bulk: Orin retransmits | 0 | 0 |
+| Burst 64 KB after 1 s idle, p50 | 17 ms | 18 ms |
+| Burst 256 KB, p50 | 12.1 s | 14.1 s |
+| Burst 1 MB, p50 | 14.8 s | 13.8 s |
+| Burst 4 MB, p50 | 4.5 s | 4.6 s |
+| Burst connection: Orin retransmits, smoothed RTT, RTO | 50, 2.35 s, 10.8 s | 49, 2.39 s, 11.0 s |
+
+**Not idle-related.** 256 KB bursts on fresh connections, 8 each:
+
+| Idle before each burst | p50 | Bursts over 1 s | Orin retransmits, smoothed RTT |
+|---|---|---|---|
+| 0 | 12.6 s | 7 of 8 | 15, 1.6 s |
+| 50 ms | 61 ms | 1 of 8 | 8, 1.8 s |
+| 200 ms | 14.3 s | 6 of 8 | 13, 1.8 s |
+| 1 s | 56 ms | 0 of 8 | 0, 16 ms |
+| 1 s, 50/s UDP keepalive | 56 ms | 0 of 8 (3 near 0.5 s) | 5, 21 ms |
+| 0 (repeat) | 13.6 s | 8 of 8 | 14, 1.5 s |
+
+**Reading.** A fresh connection's first burst is fast. Then a few segments go
+missing, and the Orin measures round trips of 1.5–2.4 s on a 3 ms link: packets
+are being *held* for seconds, not only dropped. The smoothed RTT and the RTO
+(10–16 s) inflate, and every later burst on that connection waits out the RTO.
+Bulk stalls of up to 14 s with zero retransmits are the same holding, seen
+without loss. Paced traffic at ≤ 4 MB/s never triggers it.
+
+Nothing on the Orin's side records it: no kernel messages, `tx_dropped` and
+`tx_errors` 0 on `usb1`. The packets vanish or wait inside the USB layer, in the
+Orin's NCM gadget or the tablet's NCM host driver; the tablet's sysfs is not
+readable without root.
+
+**Consequences:**
+
+- Anything the Orin sends to the tablet should be **paced**, not dumped at line
+  rate: video at a steady frame rate is fine; a TTS utterance should stream in
+  chunks rather than be written in one go (the protocol already plans chunks).
+- A connection that has hit this is poisoned for many seconds (inflated RTO).
+  The heartbeat timeout will see it as a dead link and reconnect, which resets it.
+- `adb push`/`install` is exactly an unpaced bulk transfer, which explains its
+  0.14 MB/s.
+
+Open: whether RNDIS instead of NCM avoids it, and whether pacing all egress on
+the Orin (`tc` `fq` with `maxrate`) fixes unpaced senders too.
diff --git a/lib/linktest.dart b/lib/linktest.dart
index 63ed465..b708040 100644
--- a/lib/linktest.dart
+++ b/lib/linktest.dart
@@ -145,6 +145,219 @@ class LinkTest extends ChangeNotifier {
     return {'bytes': bytes, 'secs': (t * 1000).round() / 1000, 'mb_s': (bytes / t / 1e4).round() / 100};
   }
 
+  /// Orin → tablet only: sustained bulk sampled every 250 ms, TTS-like bursts on a
+  /// warm connection, and camera-like paced streams.
+  Future<void> runDownload(String host, String label, {int repeats = 5, int bulkSecs = 10, int pacedSecs = 10}) async {
+    if (running) return;
+    running = true;
+    lines.clear();
+    final id = '$label-${DateTime.now().toUtc().toIso8601String().substring(0, 19)}';
+    _line('download series $id → $host');
+    try {
+      await usb.bindFor(Uri(scheme: 'ws', host: host));
+      await _report(host, {'id': id, 'kind': 'start', 'label': label, 'series': 'download'});
+      for (var i = 0; i < repeats; i++) {
+        _status('bulk ${i + 1}/$repeats, $bulkSecs s');
+        final r = await tcpBulkSampled(host, id, bulkSecs);
+        _line('bulk ${i + 1}: ${r['mb_s']} MB/s, 250 ms windows min ${r['win_min_mb_s']} p10 ${r['win_p10_mb_s']} max ${r['win_max_mb_s']}, longest stall ${r['max_gap_ms']} ms');
+        await _report(host, {'id': id, 'kind': 'dl_bulk', 'n': i, ...r});
+      }
+      _status('bursts');
+      final b = await tcpBursts(host, id, sizes: const [65536, 262144, 1048576, 4194304], repeats: repeats);
+      for (final e in (b['sizes'] as List)) {
+        _line('burst ${e['bytes'] ~/ 1024} KB: p50 ${e['p50_ms']} ms, max ${e['max_ms']} ms (${e['mb_s_p50']} MB/s at p50)');
+      }
+      await _report(host, {'id': id, 'kind': 'dl_bursts', ...b});
+      for (final rate in const [500000, 1000000, 2000000, 4000000]) {
+        _status('paced ${rate / 1e6} MB/s, $pacedSecs s');
+        final r = await tcpPaced(host, id, rate, pacedSecs);
+        _line('paced ${rate / 1e6} MB/s: got ${r['mb_s']} MB/s, ${r['frames']}/${r['expected']} frames, queueing p50 ${r['q_p50_ms']} p99 ${r['q_p99_ms']} max ${r['q_max_ms']} ms, >100 ms ${r['late_100']}, >500 ms ${r['late_500']}');
+        await _report(host, {'id': id, 'kind': 'dl_paced', ...r});
+      }
+      await _report(host, {'id': id, 'kind': 'end'});
+      _status('done');
+    } catch (e) {
+      _line('error: $e');
+      _status('failed');
+      try {
+        await _report(host, {'id': id, 'kind': 'error', 'error': '$e'});
+      } catch (_) {}
+    } finally {
+      running = false;
+      notifyListeners();
+    }
+  }
+
+  /// Burst latency against the idle gap before each burst, with and without keepalive traffic.
+  Future<void> runBurstIdle(String host, String label) async {
+    if (running) return;
+    running = true;
+    lines.clear();
+    final id = '$label-${DateTime.now().toUtc().toIso8601String().substring(0, 19)}';
+    _line('burst/idle series $id → $host');
+    try {
+      await usb.bindFor(Uri(scheme: 'ws', host: host));
+      await _report(host, {'id': id, 'kind': 'start', 'label': label, 'series': 'burst-idle'});
+      for (final (idleMs, ka) in const [(0, 0), (50, 0), (200, 0), (1000, 0), (1000, 50), (0, 0)]) {
+        _status('256 KB bursts, idle $idleMs ms, keepalive $ka/s');
+        final r = await tcpBursts(host, id, sizes: const [262144], repeats: 8, idle: Duration(milliseconds: idleMs), keepalivePps: ka);
+        final e = (r['sizes'] as List).first;
+        _line('idle $idleMs ms ka $ka/s: p50 ${e['p50_ms']} max ${e['max_ms']} ms ${e['ms']}');
+        await _report(host, {'id': id, 'kind': 'dl_bursts', ...r});
+      }
+      await _report(host, {'id': id, 'kind': 'end'});
+      _status('done');
+    } catch (e) {
+      _line('error: $e');
+      _status('failed');
+      try {
+        await _report(host, {'id': id, 'kind': 'error', 'error': '$e'});
+      } catch (_) {}
+    } finally {
+      running = false;
+      notifyListeners();
+    }
+  }
+
+  static Future<Map<String, dynamic>> tcpBulkSampled(String host, String id, int secs) async {
+    final s = await Socket.connect(host, _down, timeout: const Duration(seconds: 5));
+    s.write('$id $secs\n');
+    final clock = Stopwatch()..start();
+    final windows = <int, int>{};
+    var bytes = 0, maxGap = 0;
+    int? first, last;
+    await s.timeout(Duration(seconds: secs + 60)).forEach((b) {
+      final now = clock.elapsedMicroseconds;
+      first ??= now;
+      if (last != null) maxGap = max(maxGap, now - last!);
+      last = now;
+      bytes += b.length;
+      windows.update((now - first!) ~/ 250000, (v) => v + b.length, ifAbsent: () => b.length);
+    });
+    s.destroy();
+    final t = ((last ?? 0) - (first ?? 0)) / 1e6;
+    final n = windows.isEmpty ? 0 : windows.keys.reduce(max); // drop the last, partial window
+    final rates = [for (var i = 0; i < n; i++) (windows[i] ?? 0) * 4 / 1e6]..sort();
+    double r2(num v) => (v * 100).round() / 100;
+    return {
+      'bytes': bytes,
+      'secs': r2(t),
+      'mb_s': t > 0 ? r2(bytes / t / 1e6) : 0,
+      'win_min_mb_s': rates.isEmpty ? null : r2(rates.first),
+      'win_p10_mb_s': rates.isEmpty ? null : r2(rates[rates.length ~/ 10]),
+      'win_p50_mb_s': rates.isEmpty ? null : r2(rates[rates.length ~/ 2]),
+      'win_max_mb_s': rates.isEmpty ? null : r2(rates.last),
+      'zero_windows': rates.where((r) => r == 0).length,
+      'max_gap_ms': (maxGap / 100).round() / 10,
+    };
+  }
+
+  /// Each burst is requested on the same connection after 1 s idle, as utterances would be.
+  static Future<Map<String, dynamic>> tcpBursts(String host, String id,
+      {required List<int> sizes, required int repeats, Duration idle = const Duration(seconds: 1), int keepalivePps = 0}) async {
+    // Optional light UDP traffic during the idle gaps, to tell link power saving from TCP behaviour.
+    RawDatagramSocket? ka;
+    Timer? kaTimer;
+    if (keepalivePps > 0) {
+      ka = await RawDatagramSocket.bind(InternetAddress.anyIPv4, 0);
+      final addr = InternetAddress(host);
+      kaTimer = Timer.periodic(Duration(microseconds: 1000000 ~/ keepalivePps), (_) => ka!.send(Uint8List(16), addr, _echo));
+    }
+    final s = await Socket.connect(host, _down, timeout: const Duration(seconds: 5));
+    s.write('$id burst\n');
+    var got = 0;
+    Completer<void>? want;
+    var target = 0;
+    final sub = s.listen((b) {
+      got += b.length;
+      final w = want;
+      if (w != null && !w.isCompleted && got >= target) w.complete();
+    });
+    final out = <Map<String, dynamic>>[];
+    for (final size in sizes) {
+      final ms = <double>[];
+      for (var i = 0; i < repeats; i++) {
+        await Future<void>.delayed(idle);
+        got = 0;
+        target = size;
+        final w = want = Completer<void>();
+        final clock = Stopwatch()..start();
+        s.write('$size\n');
+        await w.future.timeout(const Duration(seconds: 60));
+        want = null;
+        ms.add(clock.elapsedMicroseconds / 1000);
+      }
+      ms.sort();
+      final p50 = ms[ms.length ~/ 2];
+      out.add({
+        'bytes': size,
+        'ms': [for (final m in ms) (m * 10).round() / 10],
+        'p50_ms': (p50 * 10).round() / 10,
+        'max_ms': (ms.last * 10).round() / 10,
+        'mb_s_p50': (size / p50 / 10).round() / 100,
+      });
+    }
+    await sub.cancel();
+    s.destroy();
+    kaTimer?.cancel();
+    ka?.close();
+    return {'idle_ms': idle.inMilliseconds, 'keepalive_pps': keepalivePps, 'sizes': out};
+  }
+
+  /// Frames carry the Orin's send time. The clocks are not compared absolutely: queueing
+  /// delay is each frame's one-way delay minus the run's smallest.
+  static Future<Map<String, dynamic>> tcpPaced(String host, String id, int rate, int secs) async {
+    final s = await Socket.connect(host, _down, timeout: const Duration(seconds: 5));
+    s.write('$id paced $rate $secs\n');
+    final header = BytesBuilder();
+    var skip = 0, bytes = 0;
+    final owd = <double>[];
+    double? firstRx, lastRx;
+    await s.timeout(Duration(seconds: secs + 60)).forEach((chunk) {
+      bytes += chunk.length;
+      var i = 0;
+      while (i < chunk.length) {
+        if (skip > 0) {
+          final n = min(skip, chunk.length - i);
+          skip -= n;
+          i += n;
+          continue;
+        }
+        final n = min(12 - header.length, chunk.length - i);
+        header.add(Uint8List.sublistView(chunk, i, i + n));
+        i += n;
+        if (header.length == 12) {
+          final h = ByteData.sublistView(header.takeBytes());
+          final size = h.getUint32(0);
+          final sentAt = h.getFloat64(4);
+          skip = size - 12;
+          // Arrival of the header stands for the frame; with skip pending, the rest follows.
+          final now = DateTime.now().microsecondsSinceEpoch / 1e6;
+          firstRx ??= now;
+          lastRx = now;
+          owd.add(now - sentAt);
+        }
+      }
+    });
+    s.destroy();
+    final base = owd.isEmpty ? 0.0 : owd.reduce(min);
+    final q = [for (final d in owd) (d - base) * 1000]..sort();
+    double r1(num v) => (v * 10).round() / 10;
+    final t = (lastRx ?? 0) - (firstRx ?? 0);
+    return {
+      'rate': rate,
+      'secs': secs,
+      'expected': secs * 30,
+      'frames': owd.length,
+      'mb_s': t > 0 ? (bytes / t / 1e4).round() / 100 : 0,
+      'q_p50_ms': q.isEmpty ? null : r1(q[q.length ~/ 2]),
+      'q_p99_ms': q.isEmpty ? null : r1(q[min(q.length - 1, (q.length * 0.99).floor())]),
+      'q_max_ms': q.isEmpty ? null : r1(q.last),
+      'late_100': q.where((v) => v > 100).length,
+      'late_500': q.where((v) => v > 500).length,
+    };
+  }
+
   static Future<Map<String, dynamic>> tcpUp(String host, String id, int secs) async {
     final s = await Socket.connect(host, _up, timeout: const Duration(seconds: 5));
     final reply = utf8.decoder.bind(s).join();
diff --git a/lib/main.dart b/lib/main.dart
index 528d7b1..4629c69 100644
--- a/lib/main.dart
+++ b/lib/main.dart
@@ -18,14 +18,20 @@ Future<void> main() async {
   await usb.bindFor(Uri.parse(defaultUrl));
   final boundary = Boundary(defaultUrl)..start();
   final linkTest = LinkTest(usb);
-  // Development hook: `am start ... --es linktest <label>[@<host>]` runs the test,
-  // on the gateway's host unless one is given.
+  // Development hook: `am start ... --es linktest [dl:|bi:]<label>[@<host>]` runs the test
+  // (`dl:` the download series, `bi:` burst latency against idle gaps), on the gateway's host unless one is given.
   Future<void> runLinkTest(String request) async {
+    final mode = request.startsWith('dl:') || request.startsWith('bi:') ? request.substring(0, 2) : '';
+    if (mode.isNotEmpty) request = request.substring(3);
     final at = request.indexOf('@');
     final label = at < 0 ? request : request.substring(0, at);
     final host = at < 0 ? Uri.parse(boundary.url).host : request.substring(at + 1);
     navigatorKey.currentState?.push(MaterialPageRoute<void>(builder: (_) => linkTestPage(boundary, linkTest, usb)));
-    await linkTest.run(host, label);
+    await switch (mode) {
+      'dl' => linkTest.runDownload(host, label),
+      'bi' => linkTest.runBurstIdle(host, label),
+      _ => linkTest.run(host, label),
+    };
     await usb.bindFor(Uri.parse(boundary.url));
   }
 
diff --git a/tool/linktest_server.py b/tool/linktest_server.py
index 7f0e6f3..7d52bf0 100644
--- a/tool/linktest_server.py
+++ b/tool/linktest_server.py
@@ -4,7 +4,12 @@
 The tablet's own sockets are the only ones bound to the USB link (`adb shell`
 traffic goes over WiFi), so the app drives the test and this end only serves:
 
-  7791/tcp  down: client sends "<id> <secs>\\n", gets zeros for <secs>
+  7791/tcp  down: client sends one line, then reads until the server closes:
+              "<id> <secs>"               zeros for <secs>
+              "<id> burst"                then per line "<bytes>": exactly <bytes> zeros,
+                                          on the same connection until the client closes
+              "<id> paced <B/s> <secs>"   30 frames/s, each [u32 length][f64 send time][zeros],
+                                          length counting the whole frame
   7792/tcp  up:   client sends "<id>\\n" then data until EOF; gets "<bytes>\\n" back
   7793/udp  echo
   7794/tcp  results: one JSON line per connection, appended to the results file
@@ -51,15 +56,37 @@ def readline(conn):
 
 
 def down(conn, peer):
-    test_id, secs = readline(conn).split()
-    end = time.monotonic() + float(secs)
+    test_id, *args = readline(conn).split()
+    mode = args[0] if args[0] in ("burst", "paced") else "bulk"
     sent = 0
     try:
-        while time.monotonic() < end:
-            sent += conn.send(CHUNK)
+        if mode == "burst":
+            # One warm connection, as an utterance would arrive on the open WebSocket.
+            while line := readline(conn):
+                left = int(line)
+                while left > 0:
+                    n = conn.send(CHUNK[:min(left, len(CHUNK))])
+                    left -= n
+                    sent += n
+        elif mode == "paced":
+            rate, secs = float(args[1]), float(args[2])
+            size = max(12, int(rate / 30))
+            t0 = time.monotonic()
+            for i in range(int(secs * 30)):
+                # Pace on the schedule, not on the last send: a stall must not lower the offered rate.
+                delay = t0 + i / 30 - time.monotonic()
+                if delay > 0:
+                    time.sleep(delay)
+                conn.sendall(struct.pack("!Id", size, time.time()) + bytes(size - 12))
+                sent += size
+        else:
+            end = time.monotonic() + float(args[0])
+            while time.monotonic() < end:
+                sent += conn.send(CHUNK)
     except OSError as e:
-        log({"id": test_id, "kind": "tcp_down_orin", "error": str(e)})
-    log({"id": test_id, "kind": "tcp_down_orin", "peer": peer, "bytes": sent, **tcp_info(conn)})
+        log({"id": test_id, "kind": "tcp_down_orin", "mode": mode, "error": str(e)})
+    log({"id": test_id, "kind": "tcp_down_orin", "mode": mode, "args": args, "peer": peer, "bytes": sent,
+         **tcp_info(conn)})
 
 
 def up(conn, peer):
diff --git a/tool/linktest_summary.py b/tool/linktest_summary.py
index acebb08..03d3cbc 100644
--- a/tool/linktest_summary.py
+++ b/tool/linktest_summary.py
@@ -25,7 +25,21 @@ for run_id, recs in runs.items():
         elif k in ("tcp_down", "tcp_up"):
             print(f"  {k:8} {r['mb_s']:>6} MB/s over {r['secs']} s")
         elif k in ("tcp_down_orin", "tcp_up_orin"):
-            print(f"  {k:13} retrans {r.get('retrans')}  rtt {r.get('rtt_ms')} ms  rto {r.get('rto_ms')} ms {r.get('error', '')}")
+            mode = r.get("mode", "")
+            print(f"  {k:13} {mode:6} retrans {r.get('retrans')}  rtt {r.get('rtt_ms')} ms  rto {r.get('rto_ms')} ms {r.get('error', '')}")
+        elif k == "dl_bulk":
+            print(f"  bulk  #{r['n']}  {r['mb_s']:>5} MB/s over {r['secs']} s   250 ms windows: min {r['win_min_mb_s']} "
+                  f"p10 {r['win_p10_mb_s']} p50 {r['win_p50_mb_s']} max {r['win_max_mb_s']}  zero {r['zero_windows']}  "
+                  f"longest stall {r['max_gap_ms']} ms")
+        elif k == "dl_bursts":
+            print(f"  bursts after {r.get('idle_ms', 1000)} ms idle, keepalive {r.get('keepalive_pps', 0)}/s:")
+            for e in r["sizes"]:
+                print(f"  burst {e['bytes'] // 1024:>5} KB  p50 {e['p50_ms']:>7} ms  max {e['max_ms']:>7} ms  "
+                      f"({e['mb_s_p50']} MB/s at p50)  all {e['ms']}")
+        elif k == "dl_paced":
+            print(f"  paced {r['rate'] / 1e6:>3} MB/s  got {r['mb_s']:>5} MB/s  frames {r['frames']}/{r['expected']}  "
+                  f"queueing p50 {r['q_p50_ms']} p99 {r['q_p99_ms']} max {r['q_max_ms']} ms  "
+                  f">100 ms {r['late_100']}  >500 ms {r['late_500']}")
         elif k == "error":
             print(f"  ERROR {r['error']}")
         elif k == "end":