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) + } +}