test(io): aetherctl --throttle-kbps slow-CDN simulation

Reproducing the #92 open-GOP glitch needs AVPlayer to rebuffer and do a
fresh decode at a mid-stream non-independent segment, which does not
happen on a fast local network. This adds a source-IO bandwidth cap so
the producer can be starved below real-time on demand.

- SourceThrottle: pure virtual-clock leaky-bucket rate math (testable,
  no sleeping), 7 unit tests.
- AVIOReader: captures the throttle once at init from a static test hook
  and holds delivered bytes to the target rate in read() (the single
  chokepoint for persistent/streaming/seekable), lock-guarded.
- AetherEngine.setSourceThrottleKbpsForTesting: test-only static hook,
  mirrors setForceSoftwarePathForTesting.
- aetherctl: --throttle-kbps N on serve / seektest / audio / extract.

Verified end-to-end: a 0.06s unthrottled extract becomes 10.8s at
800 kbit/s over a local HTTP fixture. Full suite green (XCTest 227,
Swift Testing 268), strict-concurrency build clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XZTEfmztPE8hAdjHdBr9BH

Author: superuser404notfound

Sources/AetherEngine/AetherEngine.swift

@@ -322,6 +322,16 @@ public final class AetherEngine: ObservableObject {
         forceSoftwarePathForTesting = on
     }
 
+    /// TEST-ONLY: throttle source IO to simulate a slow CDN/origin (kbit/s; 0 = unlimited). Read once by
+    /// each `AVIOReader` at init, so set it before `load`/`start`. Used by `aetherctl --throttle-kbps` to
+    /// starve the producer below real-time and provoke AVPlayer rebuffers (e.g. the #92 open-GOP repro).
+    nonisolated(unsafe) static var sourceThrottleKbpsForTesting = 0
+
+    /// TEST-ONLY. Set the source-IO throttle for the `aetherctl --throttle-kbps` harness; not for app use.
+    public nonisolated static func setSourceThrottleKbpsForTesting(_ kbps: Int) {
+        sourceThrottleKbpsForTesting = max(0, kbps)
+    }
+
     /// Reads `AVPlayer.eligibleForHDRPlayback` and `AVPlayer.availableHDRModes` at call time.
     /// macOS reports the built-in display only and may under-report external displays.
     public static var displayCapabilities: DisplayCapabilities {

Sources/AetherEngine/Demuxer/AVIOReader.swift

@@ -264,6 +264,11 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
     private let chunkRequestTimeout: TimeInterval
     private let chunkMaxRetries: Int
 
+    /// TEST-ONLY slow-CDN throttle (kbit/s, 0 = unlimited), captured once from the static hook at init.
+    private let throttleKbps: Int
+    private var throttleVClockNs: UInt64 = 0
+    private let throttleLock = NSLock()
+
     init(url: URL, extraHeaders: [String: String] = [:], chunkSize: Int = 4 * 1024 * 1024, prefetchEnabled: Bool = true, isLive: Bool = false, chunkRequestTimeout: TimeInterval = 35, chunkMaxRetries: Int = 3) {
         self.url = url
         self.extraHeaders = extraHeaders
@@ -272,6 +277,23 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
         self.isLive = isLive
         self.chunkRequestTimeout = chunkRequestTimeout
         self.chunkMaxRetries = max(1, chunkMaxRetries)
+        self.throttleKbps = AetherEngine.sourceThrottleKbpsForTesting
+    }
+
+    /// Slow-CDN simulation: hold delivered bytes to `throttleKbps` by sleeping the demux thread before the
+    /// bytes reach the demuxer. No-op unless the test hook is set. Lock-guarded: prefetch and demux paths
+    /// can both deliver. Sleeping here is consistent with the existing reconnect backoff on this thread.
+    private func applyThrottle(deliveredBytes: Int) {
+        guard throttleKbps > 0, deliveredBytes > 0 else { return }
+        throttleLock.lock()
+        let sleepNs = SourceThrottle.advance(
+            vclockNs: &throttleVClockNs,
+            nowNs: DispatchTime.now().uptimeNanoseconds,
+            deliveredBytes: deliveredBytes,
+            kbps: throttleKbps
+        )
+        throttleLock.unlock()
+        if sleepNs > 0 { Thread.sleep(forTimeInterval: Double(sleepNs) / 1_000_000_000) }
     }
 
     private func applyExtraHeaders(_ request: inout URLRequest) {
@@ -524,9 +546,12 @@ final class AVIOReader: AVIOProvider, @unchecked Sendable {
         if isPastReadDeadline { readDeadlineFired = true; return -1 }
         // Check usePersistentReader before isStreaming: live feeds without
         // Content-Length must use the reconnect-capable persistent path.
-        if usePersistentReader { return readPersistent(into: buf, size: size) }
-        if isStreaming { return readStreaming(into: buf, size: size) }
-        return readSeekable(into: buf, size: size)
+        let n: Int32
+        if usePersistentReader { n = readPersistent(into: buf, size: size) }
+        else if isStreaming { n = readStreaming(into: buf, size: size) }
+        else { n = readSeekable(into: buf, size: size) }
+        if n > 0 { applyThrottle(deliveredBytes: Int(n)) }
+        return n
     }
 
     // MARK: - Seekable Read (Range-based)

Sources/AetherEngine/Demuxer/SourceThrottle.swift

@@ -0,0 +1,26 @@
+import Foundation
+
+/// Virtual-clock leaky-bucket rate limiter for the source IO read path (slow-CDN simulation).
+/// Pure value logic so the pacing can be unit-tested without sleeping. `AVIOReader` keeps one
+/// `vclockNs` per reader and calls `advance` after each delivered chunk; the returned nanoseconds
+/// are how long to `Thread.sleep` before returning the bytes to the demuxer.
+enum SourceThrottle {
+
+    /// Wall-clock nanoseconds it should take to deliver `bytes` at `kbps` kilobits/s. 0 when disabled.
+    static func costNs(bytes: Int, kbps: Int) -> UInt64 {
+        guard kbps > 0, bytes > 0 else { return 0 }
+        let bytesPerSec = Double(kbps) * 125.0   // kbit/s -> byte/s (1000 bits / 8)
+        return UInt64((Double(bytes) / bytesPerSec) * 1_000_000_000)
+    }
+
+    /// Advance the virtual delivery clock by one chunk and return the sleep (ns) needed to hold the rate.
+    /// `vclockNs` is the real time by which all bytes so far should have been delivered. An idle gap
+    /// (now past the clock) resets the base to `now`, so a pause never banks burst credit.
+    static func advance(vclockNs: inout UInt64, nowNs: UInt64, deliveredBytes: Int, kbps: Int) -> UInt64 {
+        guard kbps > 0, deliveredBytes > 0 else { return 0 }
+        let base = max(vclockNs, nowNs)
+        let deadline = base &+ costNs(bytes: deliveredBytes, kbps: kbps)
+        vclockNs = deadline
+        return deadline > nowNs ? deadline - nowNs : 0
+    }
+}

Sources/aetherctl/main.swift

@@ -89,6 +89,13 @@ func printUsage() {
                      non-DV TV / on macOS (where displayCapabilities
                      reports supportsDolbyVision=false anyway).
 
+    Flags (serve / seektest / audio / extract):
+      --throttle-kbps N
+                     TEST-ONLY slow-CDN simulation: cap source-IO
+                     delivery to N kbit/s. Set below the stream bitrate
+                     to starve the producer below real-time and provoke
+                     AVPlayer rebuffers (e.g. the #92 open-GOP repro).
+
     Flags (swdecode only):
       --frames N     Max packets to read / frames to wait for.
                      Default 100.
@@ -231,12 +238,17 @@ if first == "seektest" {
     let seeks   = takeIntFlag("--seeks", from: &rest) ?? 40
     let gapMs   = takeIntFlag("--gap-ms", from: &rest) ?? 60
     let settle  = takeDoubleFlag("--settle", from: &rest) ?? 5.0
+    let throttleKbps = takeIntFlag("--throttle-kbps", from: &rest)
     guard let urlArg = rest.first(where: { !$0.hasPrefix("--") }) else {
         print("ERROR: seektest requires a <url> argument")
         exit(64)
     }
     rest.removeAll { $0 == urlArg }
     rejectStrayFlags(rest, subcommand: "seektest")
+    if let throttleKbps {
+        AetherEngine.setSourceThrottleKbpsForTesting(throttleKbps)
+        print("[aetherctl] source throttle: \(throttleKbps) kbit/s (slow-CDN simulation)")
+    }
     exit(runSeekTest(url: parseSourceURL(urlArg), seeks: seeks, gapMs: gapMs, settleSeconds: settle))
 }
 
@@ -384,6 +396,8 @@ if ["probe", "serve", "validate", "swdecode", "extract", "audio", "customio"].co
     let audioSeconds = takeDoubleFlag("--seconds", from: &rest) ?? 10
     // --native-subs: diagnostics affordance for mov_text subtitle track (#55); serve only.
     let nativeSubsIndex = takeIntFlag("--native-subs", from: &rest)
+    // --throttle-kbps: slow-CDN simulation; starves the producer below real-time to provoke rebuffers.
+    let throttleKbps = takeIntFlag("--throttle-kbps", from: &rest)
     rejectStrayFlags(rest, subcommand: first)
     guard let urlArg = rest.first else {
         print("ERROR: \(first) requires a <url> argument")
@@ -393,6 +407,10 @@ if ["probe", "serve", "validate", "swdecode", "extract", "audio", "customio"].co
     }
     let url = parseSourceURL(urlArg)
     let dvModeAvailable = !noDV
+    if let throttleKbps {
+        AetherEngine.setSourceThrottleKbpsForTesting(throttleKbps)
+        print("[aetherctl] source throttle: \(throttleKbps) kbit/s (slow-CDN simulation)")
+    }
     switch first {
     case "probe":
         exit(runProbe(url: url))

Tests/AetherEngineTests/SourceThrottleTests.swift

@@ -0,0 +1,62 @@
+import XCTest
+@testable import AetherEngine
+
+final class SourceThrottleTests: XCTestCase {
+
+    func testCostNsDisabledOrEmpty() {
+        XCTAssertEqual(SourceThrottle.costNs(bytes: 125_000, kbps: 0), 0)
+        XCTAssertEqual(SourceThrottle.costNs(bytes: 0, kbps: 1000), 0)
+    }
+
+    func testCostNs1000kbpsDelivers125kBPerSecond() {
+        // 1000 kbit/s = 125000 byte/s, so 125000 bytes should cost exactly 1 second.
+        XCTAssertEqual(SourceThrottle.costNs(bytes: 125_000, kbps: 1000), 1_000_000_000)
+        // Half the bytes -> half the time.
+        XCTAssertEqual(SourceThrottle.costNs(bytes: 62_500, kbps: 1000), 500_000_000)
+    }
+
+    func testCostNsScalesInverselyWithRate() {
+        // Same payload at 2x the rate costs half the time.
+        let slow = SourceThrottle.costNs(bytes: 1_000_000, kbps: 1000)
+        let fast = SourceThrottle.costNs(bytes: 1_000_000, kbps: 2000)
+        XCTAssertEqual(fast, slow / 2)
+    }
+
+    func testAdvanceSteadyStateSleepsOneChunkCost() {
+        // Caught up (vclock == now): a 125 kB chunk at 1000 kbps must sleep ~1s.
+        var vclock: UInt64 = 1_000_000_000
+        let sleep = SourceThrottle.advance(
+            vclockNs: &vclock, nowNs: 1_000_000_000, deliveredBytes: 125_000, kbps: 1000)
+        XCTAssertEqual(sleep, 1_000_000_000)
+        XCTAssertEqual(vclock, 2_000_000_000)   // virtual clock advanced by the chunk cost
+    }
+
+    func testAdvanceBacklogAccumulates() {
+        // Two back-to-back chunks at the same `now` (producer reading fast): the second must wait
+        // behind the first's virtual deadline, not deliver immediately.
+        var vclock: UInt64 = 0
+        let now: UInt64 = 0
+        let s1 = SourceThrottle.advance(vclockNs: &vclock, nowNs: now, deliveredBytes: 125_000, kbps: 1000)
+        let s2 = SourceThrottle.advance(vclockNs: &vclock, nowNs: now, deliveredBytes: 125_000, kbps: 1000)
+        XCTAssertEqual(s1, 1_000_000_000)
+        XCTAssertEqual(s2, 2_000_000_000)       // queued behind chunk 1
+        XCTAssertEqual(vclock, 2_000_000_000)
+    }
+
+    func testAdvanceIdleGapDoesNotBankCredit() {
+        // After delivering one chunk (vclock at 1s), a long real gap (now jumps to 10s) must not let
+        // the next chunk deliver "for free" or bank negative sleep; base resets to now.
+        var vclock: UInt64 = 1_000_000_000
+        let sleep = SourceThrottle.advance(
+            vclockNs: &vclock, nowNs: 10_000_000_000, deliveredBytes: 125_000, kbps: 1000)
+        XCTAssertEqual(sleep, 1_000_000_000)            // just one chunk cost, no burst credit
+        XCTAssertEqual(vclock, 11_000_000_000)          // rebased to now + cost
+    }
+
+    func testAdvanceDisabledReturnsZero() {
+        var vclock: UInt64 = 0
+        XCTAssertEqual(
+            SourceThrottle.advance(vclockNs: &vclock, nowNs: 0, deliveredBytes: 125_000, kbps: 0), 0)
+        XCTAssertEqual(vclock, 0)
+    }
+}