From 705e16729a1499692bfe09c4bd2ed2941df40c74 Mon Sep 17 00:00:00 2001 From: Joe Blau Date: Sat, 4 Jul 2026 19:59:09 -0400 Subject: [PATCH] refactor: make the real-time policy layer testable (extract ABR + timeline normalizer) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the StreamBroadcast-policy extraction for issue #21. The prior #20 work already moved the reconnect backoff, VideoFrameAdmission ratios, per-path bitrate ceiling, and the watchdog >-vs->= frozen-queue rule into StreamCore with tests; this lands the remaining two pieces so the whole real-time policy layer is @testable-import-able (CI-run) instead of resting on manual device testing. - StreamCore/AdaptiveBitRateState.swift (NEW): the adaptive-bitrate decision math — minimum floor, effectiveMaximum (path + thermal), insufficient-BW reduction (65% headroom / halve-on-zero), status stall halving + healthy accounting, the 30-tick upward probe (step + cap), per-path interface seeding (lastGoodTarget / 0.6x), thermal-ceiling clamp, frameInterval, and currentFrameRate — as a pure Sendable value type with mutating handlers returning an ABRDecision. Copied character-for-character (truncation + floor-division preserved); frameInterval10/30 redefined bit-identically to HaishinKit's. BroadcastAdaptiveBitRateController becomes a thin actor that unwraps the HaishinKit event, delegates, and applies once (−241 net lines). - StreamCore/MediaTimelineNormalizer.swift (NEW): MediaTimelineNormalizer + MediaTimelineKind moved out of the app target (pure CoreMedia, made public), with a pure static rebasedOffset() so the gap-accumulation math is testable with plain CMTime. +25 StreamCore tests (19 ABR reduction/probe/floor/seed/thermal/frameInterval cases + 6 normalizer rebase cases) — 124 total, all passing. Adversarially reviewed for byte-for-byte behavior preservation: no encoder-behavior change (a stall-vs-recovered log-line divergence the review caught is fixed here). Full app builds against the iOS 27 SDK. fixes #21 Co-Authored-By: Claude Opus 4.8 (1M context) --- StreamBroadcast/RTMPPublisher.swift | 346 +++--------------- StreamCore/AdaptiveBitRateState.swift | 221 +++++++++++ StreamCore/MediaTimelineNormalizer.swift | 133 +++++++ .../AdaptiveBitRateStateTests.swift | 238 ++++++++++++ .../MediaTimelineNormalizerTests.swift | 85 +++++ 5 files changed, 732 insertions(+), 291 deletions(-) create mode 100644 StreamCore/AdaptiveBitRateState.swift create mode 100644 StreamCore/MediaTimelineNormalizer.swift create mode 100644 StreamCoreTests/AdaptiveBitRateStateTests.swift create mode 100644 StreamCoreTests/MediaTimelineNormalizerTests.swift diff --git a/StreamBroadcast/RTMPPublisher.swift b/StreamBroadcast/RTMPPublisher.swift index cef57ad..b17d21a 100644 --- a/StreamBroadcast/RTMPPublisher.swift +++ b/StreamBroadcast/RTMPPublisher.swift @@ -926,51 +926,20 @@ struct BroadcastNetworkHealth: Sendable { /// congestion reduction but does not apply it. This implementation applies every /// reduction immediately and exposes the socket telemetry needed by the watchdog. actor BroadcastAdaptiveBitRateController: StreamBitRateStrategy { + // `StreamBitRateStrategy` requires a synchronous get-only `mamimumVideoBitRate`, + // which on an actor must stay a nonisolated `let` (sic misspelling is HaishinKit's). let mamimumVideoBitRate: Int let mamimumAudioBitRate = 0 - private let minimumVideoBitRate: Int - private let configuredFrameRate: Int - private let preferredFrameInterval: Double - private var targetBitRate: Int - private var healthySeconds = 0 - private var zeroOutputSeconds = 0 - private var queueBytes = 0 - private var congestionActive = false - private var capturePaused = false - private var lastEventAt = DispatchTime.now().uptimeNanoseconds - - /// Per-network-path ceiling and seeding. The protocol requires a - /// synchronous get-only `mamimumVideoBitRate`, which on an actor must stay - /// a nonisolated `let`; the dynamic per-path ceiling therefore lives here. - private var pathCeiling: Int - private var currentInterface: NetworkPathSnapshot.Interface? - private var lastGoodTarget: [NetworkPathSnapshot.Interface: Int] = [:] - - /// Thermal / Low-Power ceiling from `ThermalPowerGovernor`, composed with the - /// network path ceiling. `min()`'d into `effectiveMaximum` so the upward probe - /// never climbs past it, and folded into every frame-interval decision. - private var thermalBitRateScale: Double = 1.0 - private var thermalFrameRateCap: Int = .max - - private var effectiveMaximum: Int { - let thermalCeiling = Int(Double(mamimumVideoBitRate) * thermalBitRateScale) - return max(minimumVideoBitRate, - min(min(mamimumVideoBitRate, pathCeiling), thermalCeiling)) - } + /// All the numeric decision math now lives in the pure, unit-tested StreamCore + /// `AdaptiveBitRateState` (issue #21); this actor is a thin wrapper that reads/ + /// writes HaishinKit's encoder around it. + private var abr: AdaptiveBitRateState + private var lastEventAt = DispatchTime.now().uptimeNanoseconds // telemetry only init(maximumBitRate: Int, frameRate: Int) { mamimumVideoBitRate = maximumBitRate - minimumVideoBitRate = max(300_000, maximumBitRate / 10) - targetBitRate = maximumBitRate - pathCeiling = maximumBitRate - // Clamp only to a sane hardware maximum, NOT 30. The caller already passes - // a capability-resolved rate (≤60 on capable devices); keeping the real - // value here is what makes the `configuredFrameRate > 30` congestion tier - // in `frameInterval` reachable — it was dead code while this pinned to ≤30. - let clampedFrameRate = min(max(frameRate, 1), 120) - configuredFrameRate = clampedFrameRate - preferredFrameInterval = max(0, (1.0 / Double(clampedFrameRate)) - 0.001) + abr = AdaptiveBitRateState(maximumBitRate: maximumBitRate, frameRate: frameRate) } func adjustBitrate(_ event: NetworkMonitorEvent, @@ -978,86 +947,50 @@ actor BroadcastAdaptiveBitRateController: StreamBitRateStrategy { lastEventAt = DispatchTime.now().uptimeNanoseconds switch event { case .reset: - // Keep the bitrate learned from the previous socket. Returning to the - // configured maximum here caused a congestion/reconnect feedback loop - // on constrained Wi-Fi links. - healthySeconds = 0 - zeroOutputSeconds = 0 - queueBytes = 0 - await applyTarget(to: stream) + let decision = abr.onReset() + if decision.shouldApply { await applyDecision(decision, to: stream) } case .publishInsufficientBWOccured(let report): - queueBytes = report.currentQueueBytesOut - guard !capturePaused else { - zeroOutputSeconds = 0 - healthySeconds = 0 - return - } - healthySeconds = 0 - congestionActive = true - let audio = await stream.audioSettings - if report.currentBytesOutPerSecond > 0 { - zeroOutputSeconds = 0 - let available = max(0, report.currentBytesOutPerSecond * 8 - audio.bitRate) - // Leave substantial headroom for Wi-Fi variance, RTMP overhead, - // and keyframes instead of targeting the measured ceiling. - let reduced = Int(Double(available) * 0.65) - targetBitRate = max(minimumVideoBitRate, - min(targetBitRate, reduced)) - } else { - zeroOutputSeconds += 1 - targetBitRate = max(minimumVideoBitRate, targetBitRate / 2) + // Read the LIVE audio bitrate only when not paused — the original + // returned before touching stream.audioSettings on the paused path, so + // this adds no `await` it lacked. + let audioBitRate = abr.capturePaused ? 0 : await stream.audioSettings.bitRate + let decision = abr.onInsufficientBandwidth( + bytesOutPerSecond: report.currentBytesOutPerSecond, + queueBytesOut: report.currentQueueBytesOut, + audioBitRate: audioBitRate) + if decision.shouldApply { + await applyDecision(decision, to: stream) + streamLog.warning("ABR reduced video to \(self.abr.targetBitRate) bps; queue=\(self.abr.queueBytes) bytes") } - await applyTarget(to: stream, - severe: report.currentBytesOutPerSecond == 0) - streamLog.warning("ABR reduced video to \(self.targetBitRate) bps; queue=\(self.queueBytes) bytes") case .status(let report): - queueBytes = report.currentQueueBytesOut - guard !capturePaused else { - zeroOutputSeconds = 0 - healthySeconds = 0 - return - } - // Zero output with an empty queue is normal for a paused/static - // capture. It is a stall only when bytes are waiting to be sent. - if report.currentBytesOutPerSecond == 0, queueBytes > 0 { - zeroOutputSeconds += 1 - if zeroOutputSeconds >= 2 { - congestionActive = true - healthySeconds = 0 - targetBitRate = max(minimumVideoBitRate, targetBitRate / 2) - await applyTarget(to: stream, severe: true) + let decision = abr.onStatus(bytesOutPerSecond: report.currentBytesOutPerSecond, + queueBytesOut: report.currentQueueBytesOut) + if decision.shouldApply { + await applyDecision(decision, to: stream) + // Match the original: only the upward-PROBE apply logs "recovered". + // The severe stall-halving apply (decision.severe) logged nothing — + // logging "recovered" there would misread as recovery during a cut. + if !decision.severe { + streamLog.info("ABR recovered video to \(self.abr.targetBitRate) bps") } - } else { - zeroOutputSeconds = 0 - } - if queueBytes <= 32 * 1_024, report.currentBytesOutPerSecond > 0 { - healthySeconds += 1 - } else { - healthySeconds = 0 } - // Probe upward only after a sustained clean queue and in small steps. - // A fast 10%-every-10s ramp repeatedly overshot variable Wi-Fi uplinks. - guard healthySeconds >= 30 else { return } - healthySeconds = 0 - if targetBitRate < effectiveMaximum { - targetBitRate = min(effectiveMaximum, - targetBitRate + max(75_000, effectiveMaximum / 20)) - } else { - congestionActive = false - } - await applyTarget(to: stream) - streamLog.info("ABR recovered video to \(self.targetBitRate) bps") } } - func setCapturePaused(_ paused: Bool) { - capturePaused = paused - healthySeconds = 0 - zeroOutputSeconds = 0 + /// Applies the pure decision's target bitrate + frame interval to the encoder. + /// `abr.frameInterval` uses constants bit-identical to VideoCodecSettings's, so + /// the encoder value is byte-identical to the pre-extraction controller. + private func applyDecision(_ decision: ABRDecision, to stream: some StreamConvertible) async { + var video = await stream.videoSettings + video.bitRate = abr.targetBitRate + video.frameInterval = abr.frameInterval(severe: decision.severe) + try? await stream.setVideoSettings(video) } + func setCapturePaused(_ paused: Bool) { abr.onCapturePaused(paused) } + /// Called from RTMPPublisher.handlePathUpdate on every path emission. /// Remembers the last achieved target per interface so Wi-Fi -> 5G does not /// inherit Wi-Fi's degraded learned rate (which `.reset` deliberately @@ -1068,90 +1001,30 @@ actor BroadcastAdaptiveBitRateController: StreamBitRateStrategy { interface: NetworkPathSnapshot.Interface, isBaseline: Bool, applyingTo stream: some StreamConvertible) async { - let clamped = max(minimumVideoBitRate, min(ceiling, mamimumVideoBitRate)) - if isBaseline || currentInterface == nil { - currentInterface = interface - } else if let current = currentInterface, interface != current { - lastGoodTarget[current] = targetBitRate - currentInterface = interface - let seed = lastGoodTarget[interface] ?? Int(Double(clamped) * 0.6) - targetBitRate = max(minimumVideoBitRate, min(seed, clamped)) - healthySeconds = 0 - zeroOutputSeconds = 0 - } - let raised = clamped > pathCeiling - pathCeiling = clamped - // Clamp to effectiveMaximum (which folds in the thermal ceiling), not just - // pathCeiling: otherwise an interface change re-seeds targetBitRate from a - // full-rate last-good value and silently defeats an active thermal cap. - targetBitRate = min(targetBitRate, effectiveMaximum) - if raised { healthySeconds = 0 } // restart the upward probe cleanly - await applyTarget(to: stream) // apply NOW, not at the next 1 Hz event + let decision = abr.onPathProfile(ceiling: ceiling, interface: interface, isBaseline: isBaseline) + if decision.shouldApply { await applyDecision(decision, to: stream) } } - func currentTargetBitRate() -> Int { targetBitRate } - - /// Applies the learned bitrate and an adaptive frame-rate ceiling together. - /// On a requested 60 fps stream, using 30 fps while congested gives each frame - /// enough bits to remain legible and reduces encoder pressure. The configured - /// frame rate returns only after the uplink has proven stable. - private func applyTarget(to stream: some StreamConvertible, - severe: Bool = false) async { - var video = await stream.videoSettings - video.bitRate = targetBitRate - video.frameInterval = frameInterval(severe: severe) - try? await stream.setVideoSettings(video) - } - - /// The frame interval for the current congestion + thermal state. A larger - /// interval is a lower frame rate, so taking `max()` of the congestion and - /// thermal intervals always yields the more restrictive of the two. - private func frameInterval(severe: Bool) -> Double { - let congestionInterval: Double - if severe { - congestionInterval = VideoCodecSettings.frameInterval10 - } else if congestionActive, configuredFrameRate > 30 { - congestionInterval = VideoCodecSettings.frameInterval30 - } else { - congestionInterval = preferredFrameInterval - } - let cappedFrameRate = max(1, min(configuredFrameRate, thermalFrameRateCap)) - let thermalInterval = max(0, (1.0 / Double(cappedFrameRate)) - 0.001) - return max(congestionInterval, thermalInterval) - } + func currentTargetBitRate() -> Int { abr.targetBitRate } /// Stores a thermal/Low-Power ceiling and clamps the current target to it. /// Used when there is no stream to apply to yet (pre-connect); the value then /// takes effect on the first adaptive event after connect. func storeThermalCeiling(bitRateScale: Double, frameRateCap: Int) { - thermalBitRateScale = min(max(bitRateScale, 0.1), 1.0) - thermalFrameRateCap = max(1, frameRateCap) - // Only lower the target to fit a tightened ceiling. Leave healthySeconds - // alone: resetting it on every change would let brief thermal flapping - // across a boundary keep restarting the up-probe and starve recovery. - targetBitRate = min(targetBitRate, effectiveMaximum) + abr.onThermalCeiling(bitRateScale: bitRateScale, frameRateCap: frameRateCap) } - func currentFrameInterval() -> Double { frameInterval(severe: false) } - - /// The effective encode frame rate the ABR is currently targeting (fps), folding - /// in the same congestion + thermal caps as `frameInterval()` (most-restrictive - /// wins). Used by the live stats HUD; not a measured output rate (measured fps - /// needs M8). Mirrors `frameInterval`'s steady-state branches directly rather - /// than lossily inverting the interval Double. - func currentFrameRate() -> Int { - let congestionFps = (congestionActive && configuredFrameRate > 30) ? 30 : configuredFrameRate - let thermalFps = max(1, min(configuredFrameRate, thermalFrameRateCap)) - return min(congestionFps, thermalFps) - } + func currentFrameInterval() -> Double { abr.frameInterval(severe: false) } + + func currentFrameRate() -> Int { abr.currentFrameRate() } /// Applies the current (possibly thermally-clamped) target + frame interval to /// a freshly-connected stream. SRT/WHIP emit no `.reset` event, so the /// pre-connect store path relies on this to land the ceiling on the encoder. func applyCurrentTarget(to stream: any StreamConvertible) async { var video = await stream.videoSettings - video.bitRate = targetBitRate - video.frameInterval = frameInterval(severe: false) + video.bitRate = abr.targetBitRate + video.frameInterval = abr.frameInterval(severe: false) try? await stream.setVideoSettings(video) } @@ -1162,129 +1035,20 @@ actor BroadcastAdaptiveBitRateController: StreamBitRateStrategy { func setThermalCeiling(bitRateScale: Double, frameRateCap: Int, applyingTo stream: any StreamConvertible) async { - storeThermalCeiling(bitRateScale: bitRateScale, frameRateCap: frameRateCap) + abr.onThermalCeiling(bitRateScale: bitRateScale, frameRateCap: frameRateCap) var video = await stream.videoSettings - video.bitRate = targetBitRate - video.frameInterval = frameInterval(severe: false) + video.bitRate = abr.targetBitRate + video.frameInterval = abr.frameInterval(severe: false) try? await stream.setVideoSettings(video) } func healthSnapshot() -> BroadcastNetworkHealth { let now = DispatchTime.now().uptimeNanoseconds return BroadcastNetworkHealth( - queueBytes: queueBytes, - zeroOutputSeconds: zeroOutputSeconds, + queueBytes: abr.queueBytes, + zeroOutputSeconds: abr.zeroOutputSeconds, eventAgeSeconds: Int((now &- lastEventAt) / 1_000_000_000), - targetBitRate: targetBitRate - ) - } -} - -enum MediaTimelineKind: Hashable { - case video - case mic - case app -} - -/// Removes capture/reconnect wall-clock gaps before samples reach HaishinKit. -/// Without this, its audio ring buffer materializes a long pause as thousands of -/// silence buffers and its RTMP timestamp accumulator sends a large first delta. -struct MediaTimelineNormalizer { - private var accumulatedOffset = CMTime.zero - private var lastPresentationTime: [MediaTimelineKind: CMTime] = [:] - private var needsRebase = false - - mutating func markDiscontinuity() { - needsRebase = true - } - - mutating func normalize(_ sampleBuffer: CMSampleBuffer, - kind: MediaTimelineKind, - fallbackDuration: CMTime) -> CMSampleBuffer { - let sourcePTS = sampleBuffer.presentationTimeStamp - guard sourcePTS.isValid, sourcePTS.isNumeric else { return sampleBuffer } - - let sampleDuration = sampleBuffer.duration.isValid && sampleBuffer.duration.isNumeric && sampleBuffer.duration > .zero - ? sampleBuffer.duration - : fallbackDuration - - if needsRebase { - let prior = lastPresentationTime[kind] - ?? lastPresentationTime.values.max(by: { CMTimeCompare($0, $1) < 0 }) - if let prior { - let prospective = CMTimeSubtract(sourcePTS, accumulatedOffset) - let desired = CMTimeAdd(prior, sampleDuration) - let gap = CMTimeSubtract(prospective, desired) - if CMTimeCompare(gap, .zero) > 0 { - accumulatedOffset = CMTimeAdd(accumulatedOffset, gap) - } - } - needsRebase = false - } - - // Steady-state fast path: with no accumulated offset the copy below would - // produce a bit-identical buffer (PTS − 0 == PTS), so skip the per-buffer - // heap alloc + CMSampleBuffer copy entirely. This is the state for the - // whole broadcast until the first pause/reconnect sets an offset — at - // ~150 buffers/s it was pure waste on the gated real-time append path. - if accumulatedOffset == .zero { - lastPresentationTime[kind] = sourcePTS - return sampleBuffer - } - - var entryCount: CMItemCount = 0 - guard CMSampleBufferGetSampleTimingInfoArray( - sampleBuffer, - entryCount: 0, - arrayToFill: nil, - entriesNeededOut: &entryCount - ) == noErr, entryCount > 0 else { - return sampleBuffer - } - - var timings = [CMSampleTimingInfo]( - repeating: CMSampleTimingInfo(duration: .invalid, - presentationTimeStamp: .invalid, - decodeTimeStamp: .invalid), - count: entryCount + targetBitRate: abr.targetBitRate ) - let status = timings.withUnsafeMutableBufferPointer { buffer in - CMSampleBufferGetSampleTimingInfoArray( - sampleBuffer, - entryCount: entryCount, - arrayToFill: buffer.baseAddress, - entriesNeededOut: nil - ) - } - guard status == noErr else { return sampleBuffer } - - for index in timings.indices { - if timings[index].presentationTimeStamp.isValid { - timings[index].presentationTimeStamp = CMTimeSubtract( - timings[index].presentationTimeStamp, - accumulatedOffset - ) - } - if timings[index].decodeTimeStamp.isValid { - timings[index].decodeTimeStamp = CMTimeSubtract( - timings[index].decodeTimeStamp, - accumulatedOffset - ) - } - } - - var adjusted: CMSampleBuffer? - let copyStatus = timings.withUnsafeMutableBufferPointer { buffer in - CMSampleBufferCreateCopyWithNewTiming( - allocator: kCFAllocatorDefault, - sampleBuffer: sampleBuffer, - sampleTimingEntryCount: entryCount, - sampleTimingArray: buffer.baseAddress!, - sampleBufferOut: &adjusted - ) - } - guard copyStatus == noErr, let adjusted else { return sampleBuffer } - lastPresentationTime[kind] = adjusted.presentationTimeStamp - return adjusted } } diff --git a/StreamCore/AdaptiveBitRateState.swift b/StreamCore/AdaptiveBitRateState.swift new file mode 100644 index 0000000..1779d2e --- /dev/null +++ b/StreamCore/AdaptiveBitRateState.swift @@ -0,0 +1,221 @@ +/// The apply signal returned by every mutating ABR handler. The pure policy +/// decides WHETHER to touch the encoder and how severe the frame-rate cut is; the +/// actor maps `severe` onto the real HaishinKit VideoCodecSettings constant. +/// `shouldApply == false` models the original `guard !capturePaused { return }` +/// early-returns (counter resets still happen; no setVideoSettings call). +public struct ABRDecision: Sendable, Equatable { + public var shouldApply: Bool + public var severe: Bool + public init(shouldApply: Bool, severe: Bool) { + self.shouldApply = shouldApply + self.severe = severe + } +} + +/// The pure adaptive-bitrate decision math extracted from +/// `BroadcastAdaptiveBitRateController` so its reduction / probe / floor rules are +/// unit-tested on CI instead of resting on manual device testing (issue #21). The +/// actor owns HaishinKit application (reading `stream.audioSettings` + calling +/// `stream.setVideoSettings`) and delegates every numeric decision here. Every +/// formula is copied character-for-character from the original controller to +/// preserve the `Double→Int` truncation and `Int` floor-division exactly. +public struct AdaptiveBitRateState: Sendable, Equatable { + /// Frame-interval constants, bit-identical to HaishinKit's + /// `VideoCodecSettings.frameInterval30/10` (`(1/30)-0.001` / `(1/10)-0.001`, + /// which Swift evaluates as Doubles — verified equal `bitPattern`). Kept local + /// so the pure type needs no HaishinKit import; the ACTOR keeps calling + /// `abr.frameInterval` so the encoder value stays byte-identical. + static let frameInterval30 = (1.0 / 30.0) - 0.001 // 0.03233333333333333 + static let frameInterval10 = (1.0 / 10.0) - 0.001 // 0.099 + + public let maximumBitRate: Int + public let minimumBitRate: Int + public let configuredFrameRate: Int + public let preferredFrameInterval: Double + + public private(set) var targetBitRate: Int + public private(set) var healthySeconds = 0 + public private(set) var zeroOutputSeconds = 0 + public private(set) var queueBytes = 0 + public private(set) var congestionActive = false + public private(set) var capturePaused = false + public private(set) var pathCeiling: 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 + + public init(maximumBitRate: Int, frameRate: Int) { + self.maximumBitRate = maximumBitRate + minimumBitRate = max(300_000, maximumBitRate / 10) + targetBitRate = maximumBitRate + pathCeiling = 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. + let clampedFrameRate = min(max(frameRate, 1), 120) + configuredFrameRate = clampedFrameRate + preferredFrameInterval = max(0, (1.0 / Double(clampedFrameRate)) - 0.001) + } + + /// Folds the per-path ceiling + thermal scale live. NEVER cache this across a + /// mutation of `pathCeiling`/`thermalBitRateScale`; callers read it fresh after + /// mutating so an interface change can't silently defeat an active thermal cap. + public var effectiveMaximum: Int { + let thermalCeiling = Int(Double(maximumBitRate) * thermalBitRateScale) + return max(minimumBitRate, + min(min(maximumBitRate, pathCeiling), thermalCeiling)) + } + + // MARK: - Event handlers (each mutates state, returns an apply signal) + + /// `.reset`: zero the counters, KEEP the learned target (returning to the + /// configured maximum caused a congestion/reconnect feedback loop on + /// constrained Wi-Fi). Always applies. No capturePaused guard (matches source). + public mutating func onReset() -> ABRDecision { + healthySeconds = 0 + zeroOutputSeconds = 0 + queueBytes = 0 + return ABRDecision(shouldApply: true, severe: false) + } + + /// `.publishInsufficientBWOccured`. `audioBitRate` is the LIVE + /// `stream.audioSettings.bitRate`, passed per-call (never snapshotted). + public mutating func onInsufficientBandwidth(bytesOutPerSecond: Int, + queueBytesOut: Int, + audioBitRate: Int) -> ABRDecision { + queueBytes = queueBytesOut + guard !capturePaused else { + zeroOutputSeconds = 0 + healthySeconds = 0 + return ABRDecision(shouldApply: false, severe: false) + } + healthySeconds = 0 + congestionActive = true + if bytesOutPerSecond > 0 { + zeroOutputSeconds = 0 + let available = max(0, bytesOutPerSecond * 8 - audioBitRate) + // Leave substantial headroom for Wi-Fi variance, RTMP overhead, and + // keyframes instead of targeting the measured ceiling. + let reduced = Int(Double(available) * 0.65) + targetBitRate = max(minimumBitRate, min(targetBitRate, reduced)) + } else { + zeroOutputSeconds += 1 + targetBitRate = max(minimumBitRate, targetBitRate / 2) + } + return ABRDecision(shouldApply: true, severe: bytesOutPerSecond == 0) + } + + /// `.status`. The stall-apply (severe:true) and probe-apply (severe:false) sites + /// are MUTUALLY EXCLUSIVE within one tick: the stall path requires + /// bytesOut == 0, which forces `healthySeconds = 0` in the healthy-accounting + /// `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 { + queueBytes = queueBytesOut + guard !capturePaused else { + zeroOutputSeconds = 0 + healthySeconds = 0 + return ABRDecision(shouldApply: false, severe: false) + } + var apply = false + var severe = false + // Zero output with an empty queue is normal for a paused/static capture; it + // is a stall only when bytes are waiting to be sent. + if bytesOutPerSecond == 0, queueBytes > 0 { + zeroOutputSeconds += 1 + if zeroOutputSeconds >= 2 { + congestionActive = true + healthySeconds = 0 + targetBitRate = max(minimumBitRate, targetBitRate / 2) + apply = true + severe = true + } + } else { + zeroOutputSeconds = 0 + } + if queueBytes <= 32 * 1_024, bytesOutPerSecond > 0 { + healthySeconds += 1 + } else { + healthySeconds = 0 + } + // Probe upward only after a sustained clean queue and in small steps. + guard healthySeconds >= 30 else { + return ABRDecision(shouldApply: apply, severe: severe) + } + healthySeconds = 0 + if targetBitRate < effectiveMaximum { + targetBitRate = min(effectiveMaximum, + targetBitRate + max(75_000, effectiveMaximum / 20)) + } else { + congestionActive = false + } + return ABRDecision(shouldApply: true, severe: false) + } + + public mutating func onCapturePaused(_ paused: Bool) { + capturePaused = paused + healthySeconds = 0 + 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. + public mutating func onPathProfile(ceiling: Int, + interface: NetworkPathSnapshot.Interface, + isBaseline: Bool) -> ABRDecision { + let clamped = max(minimumBitRate, min(ceiling, maximumBitRate)) + if isBaseline || currentInterface == nil { + currentInterface = interface + } 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)) + healthySeconds = 0 + zeroOutputSeconds = 0 + } + let raised = clamped > pathCeiling + pathCeiling = clamped + targetBitRate = min(targetBitRate, effectiveMaximum) + if raised { healthySeconds = 0 } // restart the upward probe cleanly + return ABRDecision(shouldApply: true, severe: false) + } + + /// Store a thermal/Low-Power ceiling and clamp the current target to it. MUST + /// NOT touch `healthySeconds` — resetting it on every thermal flap across a + /// boundary would keep restarting the up-probe and starve recovery. + public mutating func onThermalCeiling(bitRateScale: Double, frameRateCap: Int) { + thermalBitRateScale = min(max(bitRateScale, 0.1), 1.0) + thermalFrameRateCap = max(1, frameRateCap) + targetBitRate = min(targetBitRate, effectiveMaximum) + } + + // MARK: - Pure readouts + + /// The frame interval for the current congestion + thermal state; `max()` yields + /// the more restrictive (lower fps) of the two. + public func frameInterval(severe: Bool) -> Double { + let congestionInterval: Double + if severe { + congestionInterval = Self.frameInterval10 + } else if congestionActive, configuredFrameRate > 30 { + congestionInterval = Self.frameInterval30 + } else { + congestionInterval = preferredFrameInterval + } + let cappedFrameRate = max(1, min(configuredFrameRate, thermalFrameRateCap)) + let thermalInterval = max(0, (1.0 / Double(cappedFrameRate)) - 0.001) + return max(congestionInterval, thermalInterval) + } + + /// The effective encode frame rate (fps), folding the same congestion + thermal + /// caps as `frameInterval` (most-restrictive wins). + public func currentFrameRate() -> Int { + let congestionFps = (congestionActive && configuredFrameRate > 30) ? 30 : configuredFrameRate + let thermalFps = max(1, min(configuredFrameRate, thermalFrameRateCap)) + return min(congestionFps, thermalFps) + } +} diff --git a/StreamCore/MediaTimelineNormalizer.swift b/StreamCore/MediaTimelineNormalizer.swift new file mode 100644 index 0000000..b12fade --- /dev/null +++ b/StreamCore/MediaTimelineNormalizer.swift @@ -0,0 +1,133 @@ +import CoreMedia + +/// Which capture track a sample belongs to. Each track keeps its own last-seen +/// presentation time so a rebase measures the gap against the right timeline. +public enum MediaTimelineKind: Hashable, Sendable { + case video + case mic + case app +} + +/// Removes capture/reconnect wall-clock gaps before samples reach the encoder. +/// Without this, HaishinKit's audio ring buffer materializes a long pause as +/// thousands of silence buffers and its RTMP timestamp accumulator sends a large +/// first delta. Pure CoreMedia (no HaishinKit), so it lives in StreamCore and its +/// gap math is unit-tested on CI. +public struct MediaTimelineNormalizer { + private var accumulatedOffset = CMTime.zero + private var lastPresentationTime: [MediaTimelineKind: CMTime] = [:] + private var needsRebase = false + + public init() {} + + public mutating func markDiscontinuity() { + needsRebase = true + } + + /// The pure rebase decision: given a new sample's source PTS, the current + /// accumulated offset, the prior last-seen PTS, and the sample duration, return + /// the (possibly grown) accumulated offset. The offset only ever GROWS to swallow + /// a forward gap (`prospective > desired`); a zero/negative gap leaves it + /// unchanged, so timestamps never move backwards. Exposed so the gap accounting + /// is testable with plain `CMTime` (no `CMSampleBuffer` needed). + public static func rebasedOffset(sourcePTS: CMTime, + accumulatedOffset: CMTime, + prior: CMTime, + sampleDuration: CMTime) -> CMTime { + let prospective = CMTimeSubtract(sourcePTS, accumulatedOffset) + let desired = CMTimeAdd(prior, sampleDuration) + let gap = CMTimeSubtract(prospective, desired) + return CMTimeCompare(gap, .zero) > 0 ? CMTimeAdd(accumulatedOffset, gap) : accumulatedOffset + } + + /// Normalizes a sample buffer's timing by the accumulated offset, growing the + /// offset to swallow a discontinuity gap on the first sample after a + /// `markDiscontinuity()`. Returns the input buffer unchanged while the offset is + /// zero (the steady state for a whole broadcast until the first pause/reconnect), + /// avoiding a per-buffer heap alloc + copy on the ~150 buffers/s hot path. + public mutating func normalize(_ sampleBuffer: CMSampleBuffer, + kind: MediaTimelineKind, + fallbackDuration: CMTime) -> CMSampleBuffer { + let sourcePTS = sampleBuffer.presentationTimeStamp + guard sourcePTS.isValid, sourcePTS.isNumeric else { return sampleBuffer } + + let sampleDuration = sampleBuffer.duration.isValid && sampleBuffer.duration.isNumeric && sampleBuffer.duration > .zero + ? sampleBuffer.duration + : fallbackDuration + + if needsRebase { + let prior = lastPresentationTime[kind] + ?? lastPresentationTime.values.max(by: { CMTimeCompare($0, $1) < 0 }) + if let prior { + accumulatedOffset = Self.rebasedOffset(sourcePTS: sourcePTS, + accumulatedOffset: accumulatedOffset, + prior: prior, + sampleDuration: sampleDuration) + } + needsRebase = false + } + + // Steady-state fast path: with no accumulated offset the copy below would + // produce a bit-identical buffer (PTS − 0 == PTS), so skip the per-buffer + // heap alloc + CMSampleBuffer copy entirely. + if accumulatedOffset == .zero { + lastPresentationTime[kind] = sourcePTS + return sampleBuffer + } + + var entryCount: CMItemCount = 0 + guard CMSampleBufferGetSampleTimingInfoArray( + sampleBuffer, + entryCount: 0, + arrayToFill: nil, + entriesNeededOut: &entryCount + ) == noErr, entryCount > 0 else { + return sampleBuffer + } + + var timings = [CMSampleTimingInfo]( + repeating: CMSampleTimingInfo(duration: .invalid, + presentationTimeStamp: .invalid, + decodeTimeStamp: .invalid), + count: entryCount + ) + let status = timings.withUnsafeMutableBufferPointer { buffer in + CMSampleBufferGetSampleTimingInfoArray( + sampleBuffer, + entryCount: entryCount, + arrayToFill: buffer.baseAddress, + entriesNeededOut: nil + ) + } + guard status == noErr else { return sampleBuffer } + + for index in timings.indices { + if timings[index].presentationTimeStamp.isValid { + timings[index].presentationTimeStamp = CMTimeSubtract( + timings[index].presentationTimeStamp, + accumulatedOffset + ) + } + if timings[index].decodeTimeStamp.isValid { + timings[index].decodeTimeStamp = CMTimeSubtract( + timings[index].decodeTimeStamp, + accumulatedOffset + ) + } + } + + var adjusted: CMSampleBuffer? + let copyStatus = timings.withUnsafeMutableBufferPointer { buffer in + CMSampleBufferCreateCopyWithNewTiming( + allocator: kCFAllocatorDefault, + sampleBuffer: sampleBuffer, + sampleTimingEntryCount: entryCount, + sampleTimingArray: buffer.baseAddress!, + sampleBufferOut: &adjusted + ) + } + guard copyStatus == noErr, let adjusted else { return sampleBuffer } + lastPresentationTime[kind] = adjusted.presentationTimeStamp + return adjusted + } +} diff --git a/StreamCoreTests/AdaptiveBitRateStateTests.swift b/StreamCoreTests/AdaptiveBitRateStateTests.swift new file mode 100644 index 0000000..4a3b0c8 --- /dev/null +++ b/StreamCoreTests/AdaptiveBitRateStateTests.swift @@ -0,0 +1,238 @@ +import Testing +import StreamCore + +/// Unit tests for the adaptive-bitrate reduction / probe / floor policy — the +/// live-bitrate control that previously rested on manual device testing (issue #21). +/// These pin the exact arithmetic so the extraction from the publisher's actor, and +/// any future tuning, can't silently drift it. +@Suite struct AdaptiveBitRateStateTests { + + // MARK: - Floor + ceilings + + @Test("Minimum floor is the larger of 300 kbps and max/10") + func minimumFloor() { + #expect(AdaptiveBitRateState(maximumBitRate: 1_000_000, frameRate: 30).minimumBitRate == 300_000) + #expect(AdaptiveBitRateState(maximumBitRate: 10_000_000, frameRate: 30).minimumBitRate == 1_000_000) + } + + @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.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) + #expect(s.effectiveMaximum == 2_000_000) + } + + // MARK: - Insufficient-bandwidth reduction + + @Test("Reduction with throughput targets 65% of the available headroom") + func reductionWithThroughput() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + // available = 500_001*8 - 128_000 = 3_872_008; reduced = Int(3_872_008*0.65) = 2_516_805. + let d = s.onInsufficientBandwidth(bytesOutPerSecond: 500_001, queueBytesOut: 0, audioBitRate: 128_000) + #expect(s.targetBitRate == 2_516_805) + #expect(d == ABRDecision(shouldApply: true, severe: false)) + #expect(s.zeroOutputSeconds == 0) + } + + @Test("Zero throughput halves the target and flags severe") + func reductionZeroThroughput() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + let d = s.onInsufficientBandwidth(bytesOutPerSecond: 0, queueBytesOut: 500_000, audioBitRate: 128_000) + #expect(s.targetBitRate == 3_000_000) // max(600k, 6M/2) + #expect(d.severe) + #expect(s.zeroOutputSeconds == 1) + } + + @Test("Reduction never drops below the minimum floor") + func reductionClampsToFloor() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + // bytesOut*8 <= audio → available 0 → reduced 0 → clamps to floor. + _ = s.onInsufficientBandwidth(bytesOutPerSecond: 10_000, queueBytesOut: 0, audioBitRate: 128_000) + #expect(s.targetBitRate == s.minimumBitRate) // 600_000 + } + + // MARK: - Status stall + probe + + @Test("A sustained zero-output stall halves the target on the 2nd tick") + func statusStallHalving() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + let start = s.targetBitRate + let first = s.onStatus(bytesOutPerSecond: 0, queueBytesOut: 50_000) + #expect(s.zeroOutputSeconds == 1) + #expect(first.shouldApply == false) + #expect(s.targetBitRate == start) // not yet + let second = s.onStatus(bytesOutPerSecond: 0, queueBytesOut: 50_000) + #expect(s.zeroOutputSeconds == 2) + #expect(s.targetBitRate == start / 2) + #expect(second == ABRDecision(shouldApply: true, severe: true)) + } + + @Test("The upward probe fires only after 30 clean ticks, one step at a time") + func statusProbeSteps() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + // Reduce below effMax first: available = 400_000*8 - 128_000 = 3_072_000; reduced = 1_996_800. + _ = s.onInsufficientBandwidth(bytesOutPerSecond: 400_000, queueBytesOut: 1_000_000, audioBitRate: 128_000) + let reduced = s.targetBitRate + #expect(reduced == 1_996_800) + // 29 clean ticks: no apply, just accrue health. + for _ in 0..<29 { + #expect(s.onStatus(bytesOutPerSecond: 100, queueBytesOut: 1_000).shouldApply == false) + } + #expect(s.healthySeconds == 29) + // 30th: one probe step of max(75k, effMax/20 = 300k). + let d = s.onStatus(bytesOutPerSecond: 100, queueBytesOut: 1_000) + #expect(d == ABRDecision(shouldApply: true, severe: false)) + #expect(s.targetBitRate == reduced + 300_000) + #expect(s.healthySeconds == 0) + } + + @Test("The probe caps at effectiveMaximum and then clears congestion") + func statusProbeCapsAndClears() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = s.onInsufficientBandwidth(bytesOutPerSecond: 400_000, queueBytesOut: 1_000_000, audioBitRate: 128_000) + // Drive many probe cycles; target must never exceed effMax. + for _ in 0..<40 { + for _ in 0..<30 { _ = s.onStatus(bytesOutPerSecond: 100, queueBytesOut: 1_000) } + #expect(s.targetBitRate <= s.effectiveMaximum) + } + #expect(s.targetBitRate == s.effectiveMaximum) // 6_000_000 + #expect(s.congestionActive == false) // cleared once pinned at the ceiling + } + + @Test("A single status tick never both halves AND probes") + func statusStallProbeMutuallyExclusive() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + // Force a stall on a tick — it must reset health so the probe can't fire same-tick. + _ = s.onStatus(bytesOutPerSecond: 0, queueBytesOut: 50_000) // zeroOut=1 + let d = s.onStatus(bytesOutPerSecond: 0, queueBytesOut: 50_000) // zeroOut=2 -> halve + #expect(d.severe) // halving apply + #expect(s.healthySeconds == 0) // probe cannot have run + } + + // MARK: - Path profile seeding + + @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) + // 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) + #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.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) + #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.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 + #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.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) + #expect(s.targetBitRate <= s.effectiveMaximum) // 3M cap holds + } + + // MARK: - Thermal + capture-paused + reset + + @Test("Thermal ceiling clamps the target and never touches healthySeconds") + func thermalClampKeepsHealth() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = s.onInsufficientBandwidth(bytesOutPerSecond: 400_000, queueBytesOut: 0, audioBitRate: 128_000) + for _ in 0..<5 { _ = s.onStatus(bytesOutPerSecond: 100, queueBytesOut: 1_000) } + let health = s.healthySeconds + s.onThermalCeiling(bitRateScale: 0.4, frameRateCap: 24) + s.onThermalCeiling(bitRateScale: 0.4, frameRateCap: 24) + #expect(s.healthySeconds == health) // untouched across thermal calls + #expect(s.targetBitRate <= s.effectiveMaximum) + // Scale + cap clamp into range. + var s2 = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + s2.onThermalCeiling(bitRateScale: 5.0, frameRateCap: -3) + #expect(s2.thermalBitRateScale == 1.0) + #expect(s2.thermalFrameRateCap == 1) + } + + @Test("Capture-paused events reset counters and never apply") + func capturePausedNoApply() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + let start = s.targetBitRate + s.onCapturePaused(true) + let a = s.onInsufficientBandwidth(bytesOutPerSecond: 0, queueBytesOut: 500_000, audioBitRate: 128_000) + let b = s.onStatus(bytesOutPerSecond: 0, queueBytesOut: 500_000) + #expect(a == ABRDecision(shouldApply: false, severe: false)) + #expect(b == ABRDecision(shouldApply: false, severe: false)) + #expect(s.targetBitRate == start) // unchanged while paused + #expect(s.zeroOutputSeconds == 0) + #expect(s.healthySeconds == 0) + } + + @Test("Reset keeps the learned target and zeroes the counters") + func resetKeepsLearnedTarget() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = s.onInsufficientBandwidth(bytesOutPerSecond: 400_000, queueBytesOut: 500_000, audioBitRate: 128_000) + let reduced = s.targetBitRate + let d = s.onReset() + #expect(s.targetBitRate == reduced) // learned rate retained + #expect(s.healthySeconds == 0) + #expect(s.zeroOutputSeconds == 0) + #expect(s.queueBytes == 0) + #expect(d == ABRDecision(shouldApply: true, severe: false)) + } + + // MARK: - Frame interval + rate + + @Test("Severe congestion drops to the 10 fps interval") + func frameIntervalSevere() { + let s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + #expect(s.frameInterval(severe: true) == (1.0 / 10.0) - 0.001) + } + + @Test("Congestion drops a 60 fps stream to the 30 fps interval, but leaves 30 fps alone") + func frameIntervalCongestionTier() { + var hi = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 60) + _ = hi.onInsufficientBandwidth(bytesOutPerSecond: 400_000, queueBytesOut: 0, audioBitRate: 128_000) // congestion on + #expect(hi.frameInterval(severe: false) == (1.0 / 30.0) - 0.001) + + var lo = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 30) + _ = lo.onInsufficientBandwidth(bytesOutPerSecond: 400_000, queueBytesOut: 0, audioBitRate: 128_000) + #expect(lo.frameInterval(severe: false) == lo.preferredFrameInterval) // 30fps: no >30 drop + } + + @Test("A thermal fps cap dominates when it is more restrictive") + func frameIntervalThermalDominates() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 60) + s.onThermalCeiling(bitRateScale: 1.0, frameRateCap: 5) + #expect(s.frameInterval(severe: false) == (1.0 / 5.0) - 0.001) + } + + @Test("currentFrameRate folds congestion + thermal caps (most restrictive wins)") + func currentFrameRateFolds() { + var s = AdaptiveBitRateState(maximumBitRate: 6_000_000, frameRate: 60) + #expect(s.currentFrameRate() == 60) // clean + _ = s.onInsufficientBandwidth(bytesOutPerSecond: 400_000, queueBytesOut: 0, audioBitRate: 128_000) + #expect(s.currentFrameRate() == 30) // congestion drops 60 -> 30 + s.onThermalCeiling(bitRateScale: 1.0, frameRateCap: 24) + #expect(s.currentFrameRate() == 24) // thermal cap wins + } +} diff --git a/StreamCoreTests/MediaTimelineNormalizerTests.swift b/StreamCoreTests/MediaTimelineNormalizerTests.swift new file mode 100644 index 0000000..c427636 --- /dev/null +++ b/StreamCoreTests/MediaTimelineNormalizerTests.swift @@ -0,0 +1,85 @@ +import Testing +import CoreMedia +import StreamCore + +/// Tests the timeline-rebasing math that removes capture/reconnect wall-clock gaps +/// before samples reach the encoder — previously untestable inside the app target. +@Suite struct MediaTimelineRebaseTests { + // Milliseconds timescale keeps the arithmetic exact and readable. + private func ms(_ v: Int64) -> CMTime { CMTime(value: v, timescale: 1000) } + private func same(_ a: CMTime, _ b: CMTime) -> Bool { CMTimeCompare(a, b) == 0 } + + @Test("A forward gap grows the offset to swallow it") + func forwardGapSwallowed() { + // source jumped to 10s; last sample ended at 1s + 33ms → gap ≈ 8.967s. + let offset = MediaTimelineNormalizer.rebasedOffset( + sourcePTS: ms(10_000), accumulatedOffset: .zero, prior: ms(1_000), sampleDuration: ms(33)) + #expect(same(offset, ms(8_967))) + } + + @Test("A perfectly continuous sample leaves the offset unchanged") + func continuousNoOp() { + // source == prior + duration → gap 0 → no change. + let offset = MediaTimelineNormalizer.rebasedOffset( + sourcePTS: ms(1_033), accumulatedOffset: .zero, prior: ms(1_000), sampleDuration: ms(33)) + #expect(same(offset, .zero)) + } + + @Test("A backward jump never moves the offset (timestamps can't go back)") + func backwardJumpIgnored() { + let offset = MediaTimelineNormalizer.rebasedOffset( + sourcePTS: ms(500), accumulatedOffset: .zero, prior: ms(1_000), sampleDuration: ms(33)) + #expect(same(offset, .zero)) + } + + @Test("A second gap accumulates on top of the existing offset") + func gapAccumulates() { + // offset already 8.967s; source now 20s, prior 1.033s (post-offset), dur 33ms. + let offset = MediaTimelineNormalizer.rebasedOffset( + sourcePTS: ms(20_000), accumulatedOffset: ms(8_967), prior: ms(1_033), sampleDuration: ms(33)) + // prospective 20000-8967=11033; desired 1033+33=1066; gap 9967 → 8967+9967=18934. + #expect(same(offset, ms(18_934))) + } +} + +/// Full-path tests through the CMSampleBuffer machinery (steady-state passthrough +/// and a real rebase after a discontinuity). +@Suite struct MediaTimelineNormalizerBufferTests { + private func ms(_ v: Int64) -> CMTime { CMTime(value: v, timescale: 1000) } + + /// A timing-only CMSampleBuffer (no data/format) — enough for the timing path. + private func buffer(pts: CMTime, duration: CMTime) -> CMSampleBuffer? { + var timing = CMSampleTimingInfo(duration: duration, + presentationTimeStamp: pts, + decodeTimeStamp: .invalid) + var out: CMSampleBuffer? + let status = CMSampleBufferCreate( + allocator: kCFAllocatorDefault, dataBuffer: nil, dataReady: true, + makeDataReadyCallback: nil, refcon: nil, formatDescription: nil, + sampleCount: 1, sampleTimingEntryCount: 1, sampleTimingArray: &timing, + sampleSizeEntryCount: 0, sampleSizeArray: nil, sampleBufferOut: &out) + return status == noErr ? out : nil + } + + @Test("Steady state returns the buffer unchanged (no discontinuity)") + func steadyStatePassthrough() throws { + var n = MediaTimelineNormalizer() + let sb = try #require(buffer(pts: ms(1_000), duration: ms(33))) + let out = n.normalize(sb, kind: .video, fallbackDuration: ms(33)) + #expect(CMTimeCompare(out.presentationTimeStamp, ms(1_000)) == 0) + } + + @Test("After a discontinuity the gap is removed from the sample's PTS") + func rebaseRemovesGap() throws { + var n = MediaTimelineNormalizer() + // Prime the last-PTS at 1s. + let first = try #require(buffer(pts: ms(1_000), duration: ms(33))) + _ = n.normalize(first, kind: .video, fallbackDuration: ms(33)) + // Discontinuity, then a sample that jumped to 10s — should be pulled back so + // it continues ~1.033s (gap of ~8.967s removed). + n.markDiscontinuity() + let jumped = try #require(buffer(pts: ms(10_000), duration: ms(33))) + let out = n.normalize(jumped, kind: .video, fallbackDuration: ms(33)) + #expect(CMTimeCompare(out.presentationTimeStamp, ms(1_033)) == 0) + } +}