From 7c2151e8fd950ba886d7679ab04c41458cdfad00 Mon Sep 17 00:00:00 2001 From: Joe Blau Date: Mon, 6 Jul 2026 12:01:12 -0400 Subject: [PATCH] feat: EWMA throughput estimator + measured cellular ceiling (fixes #24) The ABR was reactive-only and hard-capped any cellular path at 2.5 Mbps regardless of the real 5G uplink. This turns that hard cap into a soft SEED the upward probe can exceed toward the user max once measured throughput and a sustained clean queue justify it, and adds an EWMA throughput estimator. StreamCore/ThroughputEstimator.swift (new): - Pure, Sendable/Equatable EWMA (0.4 factor) fed the 1 Hz currentBytesOutPerSecond the network monitor already surfaces. First positive sample is adopted verbatim; non-positive samples (stalls) are ignored; reset() drops the estimate on handoff. NetworkPathSnapshot: - videoBitRateCeiling no longer hard-caps cellular/expensive at 2.5 Mbps (Low Data Mode / ultra-constrained / minimal-link-quality remain hard caps). - New videoBitRateSeed returns the conservative opening bid: 2.5 Mbps on cellular/expensive, the hard ceiling otherwise. AdaptiveBitRateState: - pathSeed + throughput fields. onPathProfile(ceiling:seed:...) opens cellular at the seed on baseline, seeds a handoff from lastGoodTarget (measurement-derived) or the conservative cold seed, and resets the estimator on a genuine handoff. - Upward probe climbs freely below the seed (recovery) but only ABOVE it when the audio-netted EWMA is within 90% of the target AND the queue has been clean 30 s. - A sustained clean queue clears congestion (restoring full frame rate) even when the target is held at the seed below effectiveMaximum, and signals an apply on the clearing tick so the encoder actually restores fps. Publishers: both RTMP and the shared SRT/WHIP controller pass the seed alongside the ceiling, and feed live audio bitrate into the status path so the probe gate compares video-vs-video. Reduction stays on the raw congestion-tick egress (true capacity during sustained queue growth); the EWMA is used only for the probe gate, because egress conflates encoder demand (healthy ticks) with link capacity (congestion) and smoothing the cut against it regresses congestion response either way. Validated by two adversarial review rounds and CI-run Swift Testing suites (AdaptiveBitRateStateTests, ThroughputEstimatorTests, SharedSupervisionTests). Co-Authored-By: Claude Opus 4.8 (1M context) --- StreamBroadcast/RTMPPublisher.swift | 12 +- StreamBroadcast/SessionPublisher.swift | 2 + StreamCore/AdaptiveBitRateState.swift | 89 +++++++++++-- StreamCore/NetworkPathSnapshot.swift | 24 +++- StreamCore/ThroughputEstimator.swift | 50 +++++++ .../AdaptiveBitRateStateTests.swift | 125 ++++++++++++++++-- StreamCoreTests/SharedSupervisionTests.swift | 15 ++- .../ThroughputEstimatorTests.swift | 50 +++++++ 8 files changed, 332 insertions(+), 35 deletions(-) create mode 100644 StreamCore/ThroughputEstimator.swift create mode 100644 StreamCoreTests/ThroughputEstimatorTests.swift diff --git a/StreamBroadcast/RTMPPublisher.swift b/StreamBroadcast/RTMPPublisher.swift index 00b4093..44cf9f4 100644 --- a/StreamBroadcast/RTMPPublisher.swift +++ b/StreamBroadcast/RTMPPublisher.swift @@ -528,6 +528,7 @@ actor RTMPPublisher: Publisher { if previous == nil || snapshot != previous { await networkController.setPathProfile( ceiling: snapshot.videoBitRateCeiling(configuredMaximum: settings.videoBitrate), + seed: snapshot.videoBitRateSeed(configuredMaximum: settings.videoBitrate), interface: snapshot.interface, isBaseline: previous == nil, applyingTo: stream) @@ -1013,8 +1014,13 @@ actor BroadcastAdaptiveBitRateController: StreamBitRateStrategy { } case .status(let report): + // Pass live audio bitrate so the upward-probe's throughput gate compares + // video-vs-video (the estimate carries total socket egress). Paused → 0, + // mirroring the insufficient-bandwidth path's guard. + let audioBitRate = abr.capturePaused ? 0 : await stream.audioSettings.bitRate let decision = abr.onStatus(bytesOutPerSecond: report.currentBytesOutPerSecond, - queueBytesOut: report.currentQueueBytesOut) + queueBytesOut: report.currentQueueBytesOut, + audioBitRate: audioBitRate) if decision.shouldApply { await applyDecision(decision, to: stream) // Match the original: only the upward-PROBE apply logs "recovered". @@ -1046,10 +1052,12 @@ actor BroadcastAdaptiveBitRateController: StreamBitRateStrategy { /// seed. The baseline (first) emission only adopts the interface and the /// ceiling, so a broadcast still STARTS at the configured full bitrate. func setPathProfile(ceiling: Int, + seed: Int, interface: NetworkPathSnapshot.Interface, isBaseline: Bool, applyingTo stream: some StreamConvertible) async { - let decision = abr.onPathProfile(ceiling: ceiling, interface: interface, isBaseline: isBaseline) + let decision = abr.onPathProfile(ceiling: ceiling, seed: seed, + interface: interface, isBaseline: isBaseline) if decision.shouldApply { await applyDecision(decision, to: stream) } } diff --git a/StreamBroadcast/SessionPublisher.swift b/StreamBroadcast/SessionPublisher.swift index d7a045f..1c283c8 100644 --- a/StreamBroadcast/SessionPublisher.swift +++ b/StreamBroadcast/SessionPublisher.swift @@ -284,6 +284,7 @@ actor SessionPublisher: Publisher { if let path = currentPath { await networkController.setPathProfile( ceiling: path.videoBitRateCeiling(configuredMaximum: settings.videoBitrate), + seed: path.videoBitRateSeed(configuredMaximum: settings.videoBitrate), interface: path.interface, isBaseline: false, applyingTo: stream) @@ -468,6 +469,7 @@ actor SessionPublisher: Publisher { if result.shouldUpdateCeiling, let stream { await networkController.setPathProfile( ceiling: snapshot.videoBitRateCeiling(configuredMaximum: settings.videoBitrate), + seed: snapshot.videoBitRateSeed(configuredMaximum: settings.videoBitrate), interface: snapshot.interface, isBaseline: result.isBaseline, applyingTo: stream) diff --git a/StreamCore/AdaptiveBitRateState.swift b/StreamCore/AdaptiveBitRateState.swift index 1779d2e..609e07b 100644 --- a/StreamCore/AdaptiveBitRateState.swift +++ b/StreamCore/AdaptiveBitRateState.swift @@ -40,16 +40,30 @@ public struct AdaptiveBitRateState: Sendable, Equatable { public private(set) var congestionActive = false public private(set) var capturePaused = false public private(set) var pathCeiling: Int + /// The conservative opening bid for the current path (2.5 Mbps on cellular, the + /// hard ceiling otherwise). The upward probe climbs freely up to this seed to + /// RECOVER, but only exceeds it with measured-throughput evidence (issue #24). + public private(set) var pathSeed: Int public private(set) var currentInterface: NetworkPathSnapshot.Interface? public private(set) var lastGoodTarget: [NetworkPathSnapshot.Interface: Int] = [:] public private(set) var thermalBitRateScale: Double = 1.0 public private(set) var thermalFrameRateCap: Int = .max + /// Smoothed measured egress throughput, fed the 1 Hz `bytesOutPerSecond` on + /// every status/insufficient event. Drives the smoothed reduction basis and the + /// probe's "is the link really carrying this?" gate. + public private(set) var throughput = ThroughputEstimator() + + /// Above the seed, the probe requires measured throughput to be within this + /// fraction of the current target before it climbs further — evidence the link + /// is actually delivering the rate, not the encoder idling on low-motion frames. + static let probeThroughputFraction = 0.9 public init(maximumBitRate: Int, frameRate: Int) { self.maximumBitRate = maximumBitRate minimumBitRate = max(300_000, maximumBitRate / 10) targetBitRate = maximumBitRate pathCeiling = maximumBitRate + pathSeed = maximumBitRate // Clamp only to a sane hardware maximum, NOT 30 — a 60 fps stream must keep // configuredFrameRate = 60 so the `> 30` congestion tier in frameInterval is // reachable. @@ -94,6 +108,16 @@ public struct AdaptiveBitRateState: Sendable, Equatable { congestionActive = true if bytesOutPerSecond > 0 { zeroOutputSeconds = 0 + throughput.record(bytesOutPerSecond: bytesOutPerSecond) + // Reduce against the RAW current egress. This event only fires after + // HaishinKit sees ~3 s of sustained queue growth, so the sample already + // reflects true link capacity (the queue is backed up, we're draining as + // fast as the link allows). The EWMA is deliberately NOT used here: it is + // also fed healthy status ticks where egress = encoder DEMAND, not + // capacity, so smoothing the cut against it either under-reduces (after a + // high-demand period) or over-reduces (after a static-screen low-demand + // period) — both failure modes were caught in review. The estimator's + // sound job is the upward-probe climb-above-seed gate below (issue #24). let available = max(0, bytesOutPerSecond * 8 - audioBitRate) // Leave substantial headroom for Wi-Fi variance, RTMP overhead, and // keyframes instead of targeting the measured ceiling. @@ -112,13 +136,15 @@ public struct AdaptiveBitRateState: Sendable, Equatable { /// `else`, so the `healthySeconds >= 30` probe guard cannot pass the same tick. /// A single `ABRDecision` is therefore byte-exact — do NOT collapse the branches. public mutating func onStatus(bytesOutPerSecond: Int, - queueBytesOut: Int) -> ABRDecision { + queueBytesOut: Int, + audioBitRate: Int = 0) -> ABRDecision { queueBytes = queueBytesOut guard !capturePaused else { zeroOutputSeconds = 0 healthySeconds = 0 return ABRDecision(shouldApply: false, severe: false) } + throughput.record(bytesOutPerSecond: bytesOutPerSecond) var apply = false var severe = false // Zero output with an empty queue is normal for a paused/static capture; it @@ -145,12 +171,38 @@ public struct AdaptiveBitRateState: Sendable, Equatable { return ABRDecision(shouldApply: apply, severe: severe) } healthySeconds = 0 - if targetBitRate < effectiveMaximum { - targetBitRate = min(effectiveMaximum, - targetBitRate + max(75_000, effectiveMaximum / 20)) - } else { - congestionActive = false + // A sustained clean queue means we are no longer congested — restore full + // frame rate NOW, even when the target is deliberately held at the seed and + // never reaches `effectiveMaximum`. (Clearing this only at the ceiling would + // pin a >30 fps cellular stream at 30 fps forever once the seed gate holds.) + // `wasCongested` forces an apply on the CLEARING tick so the actor actually + // re-applies the restored frame interval to the encoder — clearing the pure + // flag without an apply signal would leave the encoder stuck at 30 fps. + let wasCongested = congestionActive + congestionActive = false + guard targetBitRate < effectiveMaximum else { + return ABRDecision(shouldApply: true, severe: false) + } + // Below the conservative per-path seed the probe climbs freely — this is + // recovery back toward a rate the link already sustained. To climb ABOVE the + // seed (e.g. past 2.5 Mbps on a strong 5G uplink), a clean queue alone isn't + // enough: require measured throughput within `probeThroughputFraction` of the + // current target, so a low-motion encoder idling below the seed doesn't push + // the target onto capacity the link never demonstrated (issue #24). The + // estimate carries total socket egress, so compare video-vs-video by netting + // out audio, mirroring the reduction path. + if targetBitRate >= pathSeed { + let measuredVideo = throughput.bitsPerSecond - audioBitRate + let justifiesClimb = throughput.hasSample + && measuredVideo >= Int(Double(targetBitRate) * Self.probeThroughputFraction) + guard justifiesClimb else { + // Hold the target at the seed, but still apply if this tick cleared + // congestion so the encoder's frame rate is restored. + return ABRDecision(shouldApply: wasCongested, severe: false) + } } + targetBitRate = min(effectiveMaximum, + targetBitRate + max(75_000, effectiveMaximum / 20)) return ABRDecision(shouldApply: true, severe: false) } @@ -160,20 +212,35 @@ public struct AdaptiveBitRateState: Sendable, Equatable { zeroOutputSeconds = 0 } - /// Per-path ceiling + interface seeding. ORDER IS LOAD-BEARING: `raised` is - /// computed against the OLD `pathCeiling`; `pathCeiling` is assigned BEFORE the - /// `effectiveMaximum` clamp so the clamp folds in the NEW ceiling + thermal cap. + /// Per-path ceiling + interface seeding. `ceiling` is the HARD cap; `seed` is the + /// conservative opening bid (2.5 Mbps on cellular). ORDER IS LOAD-BEARING: + /// `raised` is computed against the OLD `pathCeiling`; `pathCeiling` is assigned + /// BEFORE the `effectiveMaximum` clamp so the clamp folds in the NEW ceiling + + /// thermal cap. The conservative-seed clamp is applied ONLY on the baseline + /// emission and on a genuine interface handoff — never on a same-interface + /// ceiling re-emission, or every path churn would yank a probed-up target back + /// down to the seed. public mutating func onPathProfile(ceiling: Int, + seed: Int, interface: NetworkPathSnapshot.Interface, isBaseline: Bool) -> ABRDecision { let clamped = max(minimumBitRate, min(ceiling, maximumBitRate)) + let clampedSeed = max(minimumBitRate, min(seed, clamped)) + pathSeed = clampedSeed if isBaseline || currentInterface == nil { currentInterface = interface + // Open at the seed on cellular (start conservative), full otherwise. + targetBitRate = min(targetBitRate, clampedSeed) } else if let current = currentInterface, interface != current { lastGoodTarget[current] = targetBitRate currentInterface = interface - let seed = lastGoodTarget[interface] ?? Int(Double(clamped) * 0.6) - targetBitRate = max(minimumBitRate, min(seed, clamped)) + // The old link's throughput average says nothing about this one. + throughput.reset() + // Resume a remembered rate for a known interface; else open at the + // conservative seed but never above the historical 0.6x cold-start. + let cold = min(clampedSeed, Int(Double(clamped) * 0.6)) + let seedTarget = lastGoodTarget[interface] ?? cold + targetBitRate = max(minimumBitRate, min(seedTarget, clamped)) healthySeconds = 0 zeroOutputSeconds = 0 } diff --git a/StreamCore/NetworkPathSnapshot.swift b/StreamCore/NetworkPathSnapshot.swift index bfc1296..b76434e 100644 --- a/StreamCore/NetworkPathSnapshot.swift +++ b/StreamCore/NetworkPathSnapshot.swift @@ -85,17 +85,31 @@ public struct NetworkPathSnapshot: Sendable, Equatable { /// What "a different link" means for reconnect purposes. public var linkIdentity: String { "\(interface.rawValue)/\(interfaceName)" } - /// Per-path video bitrate ceiling policy. Cellular caps the max so a 3 Mbps - /// Wi-Fi target is never re-attempted verbatim on a weak cell; Low Data - /// Mode and iOS 26 link-quality signals cap harder. Numbers are policy, - /// tunable on device. + /// Per-path video bitrate ceiling — the HARD cap the upward probe may never + /// exceed. Low Data Mode and the iOS 26 link-quality signals are genuine + /// hardware/policy limits and stay hard here. Cellular is deliberately NOT a + /// hard cap any more: a real 5G uplink often exceeds the old 2.5 Mbps clamp, so + /// that value moved to `videoBitRateSeed` as a conservative starting bid the + /// probe can climb above with measured evidence (issue #24). Numbers are + /// policy, tunable on device. public func videoBitRateCeiling(configuredMaximum maximum: Int) -> Int { guard isSatisfied else { return maximum } var ceiling = maximum - if interface == .cellular || isExpensive { ceiling = min(ceiling, 2_500_000) } if isConstrained { ceiling = min(ceiling, 1_200_000) } if isUltraConstrained { ceiling = min(ceiling, 800_000) } if linkQualityIsMinimal { ceiling = Swift.max(300_000, ceiling / 2) } return ceiling } + + /// The conservative INITIAL bitrate for this path — where a fresh connection or + /// a post-handoff reseed opens before the EWMA probe measures real capacity. + /// Cellular/expensive links seed at 2.5 Mbps (a safe 5G/LTE opening bid); every + /// other link seeds at its hard ceiling (i.e. starts full). Never above the + /// hard ceiling, so Low Data Mode still wins. + public func videoBitRateSeed(configuredMaximum maximum: Int) -> Int { + let ceiling = videoBitRateCeiling(configuredMaximum: maximum) + guard isSatisfied else { return ceiling } + if interface == .cellular || isExpensive { return min(ceiling, 2_500_000) } + return ceiling + } } diff --git a/StreamCore/ThroughputEstimator.swift b/StreamCore/ThroughputEstimator.swift new file mode 100644 index 0000000..e4d4576 --- /dev/null +++ b/StreamCore/ThroughputEstimator.swift @@ -0,0 +1,50 @@ +/// A pure exponentially-weighted moving average of measured egress throughput, +/// fed the 1 Hz `currentBytesOutPerSecond` deltas HaishinKit's `NetworkMonitor` +/// already surfaces. It smooths the single-sample capacity reading the ABR used +/// to react to, so a lone keyframe burst or a momentary radio stall no longer +/// drives an over-deep cut, and it gives the upward probe measured evidence that +/// a link is actually carrying the current rate before it climbs above the +/// conservative per-path seed (issue #24). +/// +/// Lives in StreamCore, `Sendable` + `Equatable`, so the arithmetic is unit-tested +/// on CI alongside `AdaptiveBitRateState` rather than resting on device testing. +public struct ThroughputEstimator: Sendable, Equatable { + /// EWMA weight on the newest sample. 0.4 tracks a genuine capacity change + /// (e.g. a 5G→LTE handoff) within a few ticks while still filtering the + /// per-second jitter of RTMP framing and keyframes. + public static let smoothingFactor = 0.4 + + /// The smoothed estimate in BITS per second (the sample is bytes/s × 8), or 0 + /// before the first positive sample. + public private(set) var bitsPerSecond = 0 + + /// False until at least one positive sample has seeded the average, so the + /// first sample is adopted verbatim instead of being blended with 0. + public private(set) var hasSample = false + + public init() {} + + /// Fold one 1 Hz egress sample (bytes/s, as reported by the network monitor) + /// into the average. Non-positive samples are ignored: a zero-output tick is a + /// stall, not evidence of low capacity, and blending it in would poison the + /// estimate right when the ABR most needs a stable capacity read. + public mutating func record(bytesOutPerSecond: Int) { + guard bytesOutPerSecond > 0 else { return } + let sample = bytesOutPerSecond * 8 + if hasSample { + bitsPerSecond = Int(Self.smoothingFactor * Double(sample) + + (1 - Self.smoothingFactor) * Double(bitsPerSecond)) + } else { + bitsPerSecond = sample + hasSample = true + } + } + + /// Drop the learned estimate. Called on a handoff to a DIFFERENT physical link + /// whose capacity is unknown, so a stale Wi-Fi average can never justify + /// probing a fresh, weaker cellular link above its seed. + public mutating func reset() { + bitsPerSecond = 0 + hasSample = false + } +} diff --git a/StreamCoreTests/AdaptiveBitRateStateTests.swift b/StreamCoreTests/AdaptiveBitRateStateTests.swift index 4a3b0c8..db573d4 100644 --- a/StreamCoreTests/AdaptiveBitRateStateTests.swift +++ b/StreamCoreTests/AdaptiveBitRateStateTests.swift @@ -18,11 +18,11 @@ import StreamCore @Test("effectiveMaximum folds the path ceiling and the thermal scale live") func effectiveMaximumFolds() { var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) - _ = s.onPathProfile(ceiling: 4_000_000, interface: .wifi, isBaseline: true) + _ = s.onPathProfile(ceiling: 4_000_000, seed: 4_000_000, interface: .wifi, isBaseline: true) s.onThermalCeiling(bitRateScale: 0.5, frameRateCap: .max) // thermal ceiling = 3.0M #expect(s.effectiveMaximum == 3_000_000) // min(min(6M,4M),3M) // Lower the path below the thermal ceiling → path wins. - _ = s.onPathProfile(ceiling: 2_000_000, interface: .wifi, isBaseline: false) + _ = s.onPathProfile(ceiling: 2_000_000, seed: 2_000_000, interface: .wifi, isBaseline: false) #expect(s.effectiveMaximum == 2_000_000) } @@ -118,39 +118,39 @@ import StreamCore @Test("An interface switch remembers the old target and seeds the new one") func pathProfileInterfaceSwitch() { var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) - _ = s.onPathProfile(ceiling: 6_000_000, interface: .wifi, isBaseline: true) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 6_000_000, interface: .wifi, isBaseline: true) // Learn a Wi-Fi target via a reduction. _ = s.onInsufficientBandwidth(bytesOutPerSecond: 400_000, queueBytesOut: 0, audioBitRate: 128_000) let wifiTarget = s.targetBitRate - // Switch to cellular: seed = 0.6 * clamped (no prior cellular target). - _ = s.onPathProfile(ceiling: 3_000_000, interface: .cellular, isBaseline: false) + // Switch to cellular: cold seed = min(2.5M seed, 0.6 * clamped) = min(2.5M, 1.8M) = 1.8M. + _ = s.onPathProfile(ceiling: 3_000_000, seed: 2_500_000, interface: .cellular, isBaseline: false) #expect(s.lastGoodTarget[.wifi] == wifiTarget) #expect(s.currentInterface == .cellular) - #expect(s.targetBitRate == max(s.minimumBitRate, min(Int(Double(3_000_000) * 0.6), 3_000_000))) // 1_800_000 + #expect(s.targetBitRate == min(min(2_500_000, Int(Double(3_000_000) * 0.6)), 3_000_000)) // 1_800_000 #expect(s.healthySeconds == 0) - // Switch back to Wi-Fi: seed from the remembered target, not 0.6*clamped. - _ = s.onPathProfile(ceiling: 6_000_000, interface: .wifi, isBaseline: false) + // Switch back to Wi-Fi: seed from the remembered target, not the cold seed. + _ = s.onPathProfile(ceiling: 6_000_000, seed: 6_000_000, interface: .wifi, isBaseline: false) #expect(s.targetBitRate == min(wifiTarget, s.effectiveMaximum)) } @Test("A raised path ceiling restarts the probe (raised uses the OLD ceiling)") func pathProfileRaisedRestartsProbe() { var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) - _ = s.onPathProfile(ceiling: 3_000_000, interface: .wifi, isBaseline: true) + _ = s.onPathProfile(ceiling: 3_000_000, seed: 3_000_000, interface: .wifi, isBaseline: true) _ = s.onInsufficientBandwidth(bytesOutPerSecond: 300_000, queueBytesOut: 0, audioBitRate: 128_000) for _ in 0..<10 { _ = s.onStatus(bytesOutPerSecond: 100, queueBytesOut: 1_000) } #expect(s.healthySeconds > 0) - _ = s.onPathProfile(ceiling: 5_000_000, interface: .wifi, isBaseline: false) // raised + _ = s.onPathProfile(ceiling: 5_000_000, seed: 5_000_000, interface: .wifi, isBaseline: false) // raised #expect(s.healthySeconds == 0) } @Test("An interface switch cannot defeat an active thermal cap") func pathProfileRespectsThermalCap() { var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) - _ = s.onPathProfile(ceiling: 6_000_000, interface: .wifi, isBaseline: true) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 6_000_000, interface: .wifi, isBaseline: true) s.onThermalCeiling(bitRateScale: 0.5, frameRateCap: .max) // effMax = 3M - // Pretend the cellular last-good was full rate, then switch to it. - _ = s.onPathProfile(ceiling: 6_000_000, interface: .cellular, isBaseline: false) + // Switch to a cellular link; even its conservative seed is clamped by thermal. + _ = s.onPathProfile(ceiling: 6_000_000, seed: 2_500_000, interface: .cellular, isBaseline: false) #expect(s.targetBitRate <= s.effectiveMaximum) // 3M cap holds } @@ -235,4 +235,103 @@ import StreamCore s.onThermalCeiling(bitRateScale: 1.0, frameRateCap: 24) #expect(s.currentFrameRate() == 24) // thermal cap wins } + + // MARK: - EWMA throughput estimator (issue #24) + + @Test("Reduction targets raw current capacity, unaffected by a demand-inflated EWMA") + func reductionUsesRawCapacityNotEWMA() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 6_000_000, interface: .wifi, isBaseline: true) + // 20 healthy ticks at ~6 Mbps (750_000 B/s * 8) drive the EWMA to the target — + // healthy egress reflects encoder DEMAND, not link capacity. + for _ in 0..<20 { _ = s.onStatus(bytesOutPerSecond: 750_000, queueBytesOut: 1_000) } + #expect(s.throughput.bitsPerSecond == 6_000_000) + // The link then collapses to 2 Mbps (250_000 B/s). The cut targets the RAW current + // drain (true capacity during congestion), NOT the demand-inflated EWMA: + // available = 2_000_000 - 128_000 = 1_872_000; reduced = Int(*0.65) = 1_216_800. + // Using the EWMA here would have under-reduced to ~2.78 Mbps — the exact multi-second + // lag issue #24 set out to eliminate (caught in adversarial review). + let d = s.onInsufficientBandwidth(bytesOutPerSecond: 250_000, queueBytesOut: 500_000, audioBitRate: 128_000) + #expect(s.targetBitRate == 1_216_800) + #expect(d == ABRDecision(shouldApply: true, severe: false)) + } + + @Test("Cellular opens at the 2.5 Mbps seed with the ceiling left at the user max") + func cellularSeedsButCeilingIsUserMax() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 2_500_000, interface: .cellular, isBaseline: true) + #expect(s.targetBitRate == 2_500_000) // conservative opening bid + #expect(s.pathSeed == 2_500_000) + #expect(s.effectiveMaximum == 6_000_000) // the probe MAY climb toward the user max + } + + @Test("At the seed, the probe HOLDS without measured-throughput evidence") + func probeHoldsAtSeedWithoutThroughput() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 2_500_000, interface: .cellular, isBaseline: true) + // 30 clean ticks but a low-motion encoder sends far below the target (1 KB/s). + for _ in 0..<30 { _ = s.onStatus(bytesOutPerSecond: 1_000, queueBytesOut: 1_000) } + #expect(s.targetBitRate == 2_500_000) // no blind climb above the cellular seed + } + + @Test("A strong 5G uplink lets the probe climb ABOVE the 2.5 Mbps cellular seed") + func probeExceedsSeedWithThroughput() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 2_500_000, interface: .cellular, isBaseline: true) + // Sustain a clean queue AND real throughput at ~2.5 Mbps (312_500 B/s * 8). + for _ in 0..<30 { _ = s.onStatus(bytesOutPerSecond: 312_500, queueBytesOut: 1_000) } + #expect(s.throughput.bitsPerSecond == 2_500_000) + // ewma (2.5M) >= 0.9*target → one step above the seed: max(75k, 6M/20 = 300k) = 300k. + #expect(s.targetBitRate == 2_800_000) // exceeds the old 2.5 Mbps hard cap + } + + @Test("The above-seed probe gate nets out audio (video-vs-video)") + func probeGateNetsAudio() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 2_500_000, interface: .cellular, isBaseline: true) + // Total egress 2.5 Mbps but 500 kbps is audio → video ≈ 2.0 Mbps < 0.9*2.5M = 2.25M. + // Netting audio out, the gate HOLDS (with audio ignored it would wrongly climb). + for _ in 0..<30 { _ = s.onStatus(bytesOutPerSecond: 312_500, queueBytesOut: 1_000, audioBitRate: 500_000) } + #expect(s.throughput.bitsPerSecond == 2_500_000) + #expect(s.targetBitRate == 2_500_000) // held: video-only throughput below the bar + } + + @Test("A 60 fps cellular stream held at the seed still restores full fps once the queue is clean") + func cellularHeldAtSeedRestoresFrameRate() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 60) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 2_500_000, interface: .cellular, isBaseline: true) + // Congestion at high egress leaves the target pinned at the 2.5M seed but flags + // congestion → 60 fps drops to 30. (reduced = 0.65*(4M-128k) = 2_516_800 ≥ seed.) + _ = s.onInsufficientBandwidth(bytesOutPerSecond: 500_000, queueBytesOut: 500_000, audioBitRate: 128_000) + #expect(s.targetBitRate == 2_500_000) + #expect(s.congestionActive) + #expect(s.currentFrameRate() == 30) + // 30 clean low-motion ticks: the probe HOLDS at the seed (target < effMax forever), + // but congestion must clear so fps returns to 60 — the regression the review caught. + var clearing = ABRDecision(shouldApply: false, severe: false) + for _ in 0..<30 { clearing = s.onStatus(bytesOutPerSecond: 1_000, queueBytesOut: 1_000) } + #expect(s.targetBitRate == 2_500_000) // still held at the seed + #expect(s.targetBitRate < s.effectiveMaximum) // below the ceiling + #expect(s.congestionActive == false) // cleared despite not reaching effMax + #expect(s.currentFrameRate() == 60) // full frame rate restored (pure state) + // The clearing tick MUST signal an apply, or the actor never re-applies the 60 fps + // interval and the encoder stays at 30 fps while state/HUD say 60 (review Defect B). + #expect(clearing == ABRDecision(shouldApply: true, severe: false)) + // A subsequent held cycle should NOT keep re-applying (congestion already clear). + var next = ABRDecision(shouldApply: false, severe: false) + for _ in 0..<30 { next = s.onStatus(bytesOutPerSecond: 1_000, queueBytesOut: 1_000) } + #expect(next.shouldApply == false) + } + + @Test("A handoff to a new link drops the stale throughput estimate") + func handoffResetsThroughput() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 6_000_000, interface: .wifi, isBaseline: true) + for _ in 0..<3 { _ = s.onStatus(bytesOutPerSecond: 500_000, queueBytesOut: 1_000) } + #expect(s.throughput.bitsPerSecond > 0) + _ = s.onPathProfile(ceiling: 6_000_000, seed: 2_500_000, interface: .cellular, isBaseline: false) + #expect(s.throughput.bitsPerSecond == 0) // wifi estimate can't justify probing the cell + #expect(s.throughput.hasSample == false) + #expect(s.pathSeed == 2_500_000) + } } diff --git a/StreamCoreTests/SharedSupervisionTests.swift b/StreamCoreTests/SharedSupervisionTests.swift index a771610..1c68389 100644 --- a/StreamCoreTests/SharedSupervisionTests.swift +++ b/StreamCoreTests/SharedSupervisionTests.swift @@ -299,10 +299,17 @@ import StreamCore #expect(snap(satisfied: false).videoBitRateCeiling(configuredMaximum: 5_000_000) == 5_000_000) } - @Test("Cellular and expensive links cap at 2.5 Mbps") - func cellularCap() { - #expect(snap(.cellular, name: "pdp_ip0").videoBitRateCeiling(configuredMaximum: 5_000_000) == 2_500_000) - #expect(snap(expensive: true).videoBitRateCeiling(configuredMaximum: 5_000_000) == 2_500_000) + @Test("Cellular and expensive links SEED at 2.5 Mbps but no longer hard-cap the ceiling") + func cellularSeed() { + // The hard ceiling is now the configured max — the probe may climb there with + // measured throughput + a clean queue (issue #24). + #expect(snap(.cellular, name: "pdp_ip0").videoBitRateCeiling(configuredMaximum: 5_000_000) == 5_000_000) + #expect(snap(expensive: true).videoBitRateCeiling(configuredMaximum: 5_000_000) == 5_000_000) + // The conservative OPENING bid stays 2.5 Mbps. + #expect(snap(.cellular, name: "pdp_ip0").videoBitRateSeed(configuredMaximum: 5_000_000) == 2_500_000) + #expect(snap(expensive: true).videoBitRateSeed(configuredMaximum: 5_000_000) == 2_500_000) + // Wi-Fi seeds at its full ceiling (starts full, no conservative bid). + #expect(snap(.wifi).videoBitRateSeed(configuredMaximum: 5_000_000) == 5_000_000) } @Test("Low Data Mode and iOS 26 link signals cap harder") diff --git a/StreamCoreTests/ThroughputEstimatorTests.swift b/StreamCoreTests/ThroughputEstimatorTests.swift new file mode 100644 index 0000000..7b085bc --- /dev/null +++ b/StreamCoreTests/ThroughputEstimatorTests.swift @@ -0,0 +1,50 @@ +import Testing +import StreamCore + +/// Unit tests for the EWMA throughput estimator (issue #24). These pin the exact +/// smoothing arithmetic so the ABR's smoothed reduction and probe gate can't +/// silently drift. +@Suite struct ThroughputEstimatorTests { + + @Test("The first positive sample is adopted verbatim (bytes → bits)") + func firstSampleVerbatim() { + var e = ThroughputEstimator() + #expect(e.hasSample == false) + #expect(e.bitsPerSecond == 0) + e.record(bytesOutPerSecond: 500_000) + #expect(e.hasSample) + #expect(e.bitsPerSecond == 4_000_000) // 500_000 * 8, no blending with 0 + } + + @Test("Subsequent samples blend at the 0.4 smoothing factor") + func blendsAtSmoothingFactor() { + var e = ThroughputEstimator() + e.record(bytesOutPerSecond: 500_000) // 4_000_000 + e.record(bytesOutPerSecond: 100_000) // 0.4*800_000 + 0.6*4_000_000 = 2_720_000 + #expect(e.bitsPerSecond == 2_720_000) + e.record(bytesOutPerSecond: 100_000) // 0.4*800_000 + 0.6*2_720_000 = 1_952_000 + #expect(e.bitsPerSecond == 1_952_000) + } + + @Test("Non-positive samples are ignored, not blended toward zero") + func ignoresNonPositiveSamples() { + var e = ThroughputEstimator() + e.record(bytesOutPerSecond: 500_000) + e.record(bytesOutPerSecond: 0) // a stall is not evidence of low capacity + e.record(bytesOutPerSecond: -10) + #expect(e.bitsPerSecond == 4_000_000) // unchanged + #expect(e.hasSample) + } + + @Test("reset() clears the estimate so a new link starts blind") + func resetClears() { + var e = ThroughputEstimator() + e.record(bytesOutPerSecond: 500_000) + e.reset() + #expect(e.bitsPerSecond == 0) + #expect(e.hasSample == false) + // After reset the next sample is again adopted verbatim. + e.record(bytesOutPerSecond: 250_000) + #expect(e.bitsPerSecond == 2_000_000) + } +}