diff --git a/Stream/ScreenCaptureController.swift b/Stream/ScreenCaptureController.swift index da21a6f..bb94520 100644 --- a/Stream/ScreenCaptureController.swift +++ b/Stream/ScreenCaptureController.swift @@ -42,6 +42,15 @@ final class ScreenCaptureController: NSObject { /// ~1s telemetry poll, live only. Cancelled on every teardown path (they all /// funnel through `stopCapture`). @ObservationIgnored private var statsTask: Task? + /// Shared frame-drop / achieved-fps counters (issue #23 / M8), created per + /// broadcast and handed to BOTH the capture output and the publisher so the four + /// shed sites and the encoded rate land in one place. Read once per stats poll to + /// fold measured fps + congestion drops into `liveStats`. + @ObservationIgnored private var frameTelemetry: FrameTelemetry? + /// The previous telemetry snapshot + its capture time, so each poll derives + /// achieved fps from the delta over the real elapsed window. + @ObservationIgnored private var lastTelemetrySnapshot: FrameTelemetrySnapshot? + @ObservationIgnored private var lastTelemetryAt: UInt64 = 0 override init() { super.init() @@ -121,24 +130,56 @@ final class ScreenCaptureController: NSObject { // Write through INCLUDING nil: the publishers return nil during a // reconnect, which must clear the card to its "—" placeholders rather // than freezing on stale last-good telemetry while the stream is down. - self?.updateLiveStats(snapshot) + self?.applyPolledStats(snapshot) try? await Task.sleep(for: .seconds(1)) } } } + /// One stats tick: fold the measured frame telemetry (achieved fps + cumulative + /// congestion drops) into the publisher's target-based snapshot, then store it. + /// Advances the achieved-fps delta baseline as a side effect, so it must run + /// exactly once per poll. + private func applyPolledStats(_ base: LiveStats?) { + updateLiveStats(enrichWithTelemetry(base)) + } + + /// Merges the shared `FrameTelemetry` into the publisher's snapshot: the measured + /// achieved fps (from the encoded-frame delta over the real elapsed window) and + /// the cumulative congestion drops. Returns `base` untouched when telemetry isn't + /// available (pre-live / torn down). Mutates the delta baseline. + private func enrichWithTelemetry(_ base: LiveStats?) -> LiveStats? { + guard let telemetry = frameTelemetry else { return base } + let current = telemetry.snapshot() + let now = DispatchTime.now().uptimeNanoseconds + var achieved = 0 + if let previous = lastTelemetrySnapshot, lastTelemetryAt > 0, now > lastTelemetryAt { + let elapsed = Double(now &- lastTelemetryAt) / 1_000_000_000 + achieved = FrameTelemetry.rate(from: previous, to: current, elapsed: elapsed).achievedFrameRate + } + // Advance the baseline every tick — even when `base` is nil (reconnecting) — + // so the next window measures against a fresh, ~1s-old sample. + lastTelemetrySnapshot = current + lastTelemetryAt = now + guard var stats = base else { return nil } + stats.achievedFrameRate = achieved + stats.droppedFrames = current.congestionDrops + return stats + } + /// Assigns the latest telemetry, skipping a redundant write (and the 1 Hz leaf /// re-render it would otherwise trigger) when the snapshot is unchanged. private func updateLiveStats(_ stats: LiveStats?) { if liveStats != stats { liveStats = stats } } - private func makePublisher(for transport: StreamCore.StreamProtocol) -> any Publisher { + private func makePublisher(for transport: StreamCore.StreamProtocol, + telemetry: FrameTelemetry) -> any Publisher { switch transport { case .rtmp, .rtmps: - RTMPPublisher() + RTMPPublisher(telemetry: telemetry) case .srt, .whip: - SessionPublisher(protocol: transport) + SessionPublisher(protocol: transport, telemetry: telemetry) } } @@ -152,8 +193,15 @@ final class ScreenCaptureController: NSObject { return } - let publisher = makePublisher(for: settings.selectedProtocol) - let output = ScreenCaptureOutput(publisher: publisher, settings: settings) { [weak self] error in + // One telemetry instance per broadcast, shared by the capture output (the + // four shed sites) and the publisher (admission + encoded rate). + let telemetry = FrameTelemetry() + frameTelemetry = telemetry + lastTelemetrySnapshot = nil + lastTelemetryAt = 0 + let publisher = makePublisher(for: settings.selectedProtocol, telemetry: telemetry) + let output = ScreenCaptureOutput(publisher: publisher, settings: settings, + telemetry: telemetry) { [weak self] error in Task { @MainActor [weak self] in await self?.captureDidStop(error: error) } } @@ -228,6 +276,9 @@ final class ScreenCaptureController: NSObject { heartbeatTask = nil statsTask?.cancel() statsTask = nil + frameTelemetry = nil + lastTelemetrySnapshot = nil + lastTelemetryAt = 0 isLive = false thermalNotice = nil liveStats = nil @@ -408,6 +459,10 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg /// the capture pacing so a 1080p60 pick never asks a device for more than it /// can sustain. The thermal governor tightens this further at runtime. private let capability = StreamCapability.current + /// Shared frame telemetry (issue #23 / M8): this output records the capture, + /// pacing (PTS-deadline), backpressure (`bufferingNewest(1)`), and compositor + /// (pool-exhaustion) sites on the serial sample queue. + private let telemetry: FrameTelemetry private let onStopped: @Sendable (Error) -> Void private let micLevelMeter: ScreenCaptureMicrophoneMeter private let micLevelChannel = MicrophoneLevelChannel() @@ -427,9 +482,11 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg init(publisher: any Publisher, settings: StreamSettings, + telemetry: FrameTelemetry, onStopped: @escaping @Sendable (Error) -> Void) { self.publisher = publisher self.settings = settings + self.telemetry = telemetry self.onStopped = onStopped micLevelMeter = ScreenCaptureMicrophoneMeter(gain: settings.micVolume) targetFrameInterval = CMTime(value: 1, @@ -454,22 +511,28 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg await publisher.setOutputSize(target, nativeShortEdge: min(width, height)) targetSize = target } - if settings.pipEnabled, let targetSize, - let camera = self.facecam.latest.take(), - let composited = self.compositor.composite( - screen: image, - camera: camera, - targetSize: targetSize, - orientation: .up, - corner: settings.pipCorner, - scale: settings.pipScale, - cameraPosition: settings.cameraPosition - ), - let output = self.compositor.makeSampleBuffer( - from: composited, - timingSource: sampleBuffer - ) { - await publisher.appendVideo(output) + if settings.pipEnabled, let targetSize, let camera = self.facecam.latest.take() { + if let composited = self.compositor.composite( + screen: image, + camera: camera, + targetSize: targetSize, + orientation: .up, + corner: settings.pipCorner, + scale: settings.pipScale, + cameraPosition: settings.cameraPosition + ), + let output = self.compositor.makeSampleBuffer( + from: composited, + timingSource: sampleBuffer + ) { + await publisher.appendVideo(output) + } else { + // Pool exhausted (a slow encoder holding surfaces) or the wrap + // failed: skip the overlay for this frame and send the raw + // screen. The frame itself is NOT lost — the overlay is. + self.telemetry.recordDrop(.compositor) + await publisher.appendVideo(sampleBuffer) + } } else { await publisher.appendVideo(sampleBuffer) } @@ -491,12 +554,16 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg guard sampleBuffer.isValid, sampleBuffer.dataReadiness == .ready else { return } switch type { case .screen: + telemetry.recordCaptured() // Downsample the native-refresh feed to the target frame rate by PTS // deadline: emit the first frame at/after each deadline, then advance // by one interval; re-anchor after a stall so a gap doesn't burst. let pts = sampleBuffer.presentationTimeStamp if pts.isValid { if nextVideoDeadline.isValid, CMTimeCompare(pts, nextVideoDeadline) < 0 { + // Intentional pacing shed (e.g. 120 Hz → 30 fps), not a fault — + // counted separately from the congestion drops the HUD surfaces. + telemetry.recordDrop(.pacing) return // arrived before the next target-fps slot — drop it } let advanced = CMTimeAdd(nextVideoDeadline.isValid ? nextVideoDeadline : pts, @@ -505,7 +572,12 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg ? advanced : CMTimeAdd(pts, targetFrameInterval) } - videoContinuation.yield(sampleBuffer) + // `bufferingNewest(1)`: if the consumer (compositor/append) hasn't drained + // the previous frame, this yield evicts it — a backpressure shed the + // pipeline couldn't keep up with. `.dropped` carries that evicted frame. + if case .dropped = videoContinuation.yield(sampleBuffer) { + telemetry.recordDrop(.backpressure) + } case .audio: if settings.includeAppAudio { publisher.enqueueApp(sampleBuffer) } case .microphone: diff --git a/Stream/SettingsView.swift b/Stream/SettingsView.swift index 398b52c..bfb212f 100644 --- a/Stream/SettingsView.swift +++ b/Stream/SettingsView.swift @@ -1022,7 +1022,8 @@ private final class MicrophoneLevelMonitor { /// A leaf view so its ~1s telemetry ticks invalidate only this subtree (not the /// whole launcher List), and every value is monospaced + single-line so the card's /// height is invariant per tick — the content-sized drawer detent never springs on -/// a number change (only an occasional thermal-notice appearance resizes it). +/// a number change (only an occasional thermal- or dropped-frames notice appearing +/// resizes it). private struct LiveStatsCard: View { var capture: ScreenCaptureController @@ -1037,6 +1038,7 @@ private struct LiveStatsCard: View { .lineLimit(1) .minimumScaleFactor(0.8) } + droppedNotice } .padding(14) .frame(maxWidth: .infinity, alignment: .leading) @@ -1062,18 +1064,32 @@ private struct LiveStatsCard: View { } } - /// Bitrate (ABR target) · effective fps · uplink health. Values are the - /// encoder's applied targets, not measured throughput (measured fps needs M8). + /// Bitrate (ABR target) · achieved fps · uplink health. Bitrate is the encoder's + /// applied target; FPS is the MEASURED achieved rate (M8), falling back to the + /// target for the first second before a rate is available. private var metricsRow: some View { HStack(spacing: 0) { metric("Bitrate", value: capture.liveStats?.bitRateLabel ?? "—") divider - metric("FPS", value: capture.liveStats.map { "\($0.frameRate)" } ?? "—") + metric("FPS", value: capture.liveStats.map { "\($0.displayFrameRate)" } ?? "—") divider metric("Uplink", value: capture.liveStats?.linkHealth.label ?? "—", tint: uplinkTint) } } + /// A subtle count of the frames congestion has shed this session (backpressure + + /// admission). Hidden until the first drop, so a healthy stream shows nothing; + /// the coloured Uplink metric already carries the live severity signal. + @ViewBuilder private var droppedNotice: some View { + if let stats = capture.liveStats, stats.droppedFrames > 0 { + Label("\(stats.droppedLabel) frames dropped", systemImage: "square.stack.3d.up.slash") + .font(.caption2) + .foregroundStyle(.secondary) + .lineLimit(1) + .minimumScaleFactor(0.8) + } + } + private var divider: some View { Divider().frame(height: 26) } private var uplinkTint: Color { diff --git a/StreamBroadcast/RTMPPublisher.swift b/StreamBroadcast/RTMPPublisher.swift index b17d21a..81ad4f9 100644 --- a/StreamBroadcast/RTMPPublisher.swift +++ b/StreamBroadcast/RTMPPublisher.swift @@ -59,6 +59,12 @@ actor RTMPPublisher: Publisher { /// Sheds input video frames when the outbound queue is deep, to bound latency /// (HaishinKit's send queue is otherwise unbounded). Refreshed by checkNetworkHealth. private let videoAdmission = VideoFrameAdmission() + /// Shared frame-drop / achieved-fps counters (issue #23 / M8). This actor records + /// the admission shed + every encoded append; the capture side records the + /// capture/pacing/backpressure/compositor sites into the same instance, and the + /// controller reads one snapshot per telemetry poll. Reachable here so the + /// adaptive logic (M9) can consume the congestion-drop signal. + private let telemetry: FrameTelemetry // Ordered, lossless audio ingress: ScreenCaptureKit yields synchronously, a // single consumer per track awaits each append — preserving PTS order so @@ -80,7 +86,8 @@ actor RTMPPublisher: Publisher { /// Shared outbound-queue stall watchdog state (stalledTicks + lastQueueBytes). private var watchdog = WatchdogState() - init() { + init(telemetry: FrameTelemetry = FrameTelemetry()) { + self.telemetry = telemetry (micStream, micCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded) (appStream, appCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded) } @@ -770,10 +777,14 @@ actor RTMPPublisher: Publisher { lastVideoBuffer = sb // Under outbound congestion, drop a proportion of video frames (audio is // never dropped) so latency stays bounded instead of the queue growing. - guard videoAdmission.admit() else { return } + guard videoAdmission.admit() else { + telemetry.recordDrop(.admission) + return + } let duration = CMTime(value: 1, timescale: CMTimeScale(encodeFrameRate)) let normalized = timeline.normalize(sb, kind: .video, fallbackDuration: duration) + telemetry.recordEncoded() await mixer.append(enforceMonotonicVideo(normalized, minStep: duration)) } diff --git a/StreamBroadcast/SessionPublisher.swift b/StreamBroadcast/SessionPublisher.swift index 353db3a..ab58f83 100644 --- a/StreamBroadcast/SessionPublisher.swift +++ b/StreamBroadcast/SessionPublisher.swift @@ -69,6 +69,10 @@ actor SessionPublisher: Publisher { private var timeline = MediaTimelineNormalizer() private let videoAdmission = VideoFrameAdmission() + /// Shared frame-drop / achieved-fps counters (issue #23 / M8). See RTMPPublisher: + /// records the admission shed + every encoded append into the same instance the + /// capture side and controller share. + private let telemetry: FrameTelemetry // Ordered, lossless audio ingress: ScreenCaptureKit yields synchronously, a // single consumer per track awaits each append — preserving PTS order. @@ -123,8 +127,10 @@ actor SessionPublisher: Publisher { /// The reconnect loop's currently-sleeping backoff; cancelling = "retry now". private var backoffSleepTask: Task? - init(protocol streamProtocol: StreamCore.StreamProtocol) { + init(protocol streamProtocol: StreamCore.StreamProtocol, + telemetry: FrameTelemetry = FrameTelemetry()) { self.transport = streamProtocol + self.telemetry = telemetry (micStream, micCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded) (appStream, appCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded) } @@ -644,9 +650,13 @@ actor SessionPublisher: Publisher { lastMediaAt = now lastVideoAppendAt = now lastVideoBuffer = sb - guard videoAdmission.admit() else { return } + guard videoAdmission.admit() else { + telemetry.recordDrop(.admission) + return + } let duration = CMTime(value: 1, timescale: CMTimeScale(encodeFrameRate)) let normalized = timeline.normalize(sb, kind: .video, fallbackDuration: duration) + telemetry.recordEncoded() await mixer.append(enforceMonotonicVideo(normalized, minStep: duration)) } diff --git a/StreamCore/FrameTelemetry.swift b/StreamCore/FrameTelemetry.swift new file mode 100644 index 0000000..790396c --- /dev/null +++ b/StreamCore/FrameTelemetry.swift @@ -0,0 +1,158 @@ +import Foundation +import os + +/// The four silent frame-shedding sites in the capture→encode pipeline, each of +/// which historically dropped a frame with no counter (issue #23 / M8). Kept +/// separate because they mean different things: +/// +/// - `pacing`: the native-refresh screen feed downsampled to the target fps by the +/// PTS-deadline throttle. INTENTIONAL — it is how a 120 Hz ProMotion feed becomes +/// a 30/60 fps stream, so it dwarfs the others and is never a fault signal. +/// - `backpressure`: the `bufferingNewest(1)` capture→consumer hand-off evicting a +/// frame the consumer hadn't drained yet — the compositor/append stage can't keep +/// up with capture. +/// - `compositor`: the facecam compositor's pixel-buffer pool was exhausted (a slow +/// encoder holding surfaces), so the overlay was skipped and the raw screen frame +/// was sent instead. The frame still reaches the encoder; only the overlay dropped. +/// - `admission`: `VideoFrameAdmission` shed the frame under outbound congestion so +/// uplink latency stays bounded. The frame never reaches the encoder. +/// +/// `backpressure` + `admission` are the "congestion" drops: a frame the pipeline +/// WANTED to encode but couldn't. Those are the fault signal surfaced in the HUD +/// and available to the adaptive logic; `pacing` and `compositor` are tracked for +/// completeness but excluded from the congestion total. +public enum FrameDropReason: Sendable, Equatable, CaseIterable { + case pacing + case backpressure + case compositor + case admission +} + +/// A cumulative, immutable readout of the frame pipeline for one broadcast. All +/// counts are monotonic session totals; rates are derived from the delta between +/// two snapshots (see `FrameTelemetry.rate`). +public struct FrameTelemetrySnapshot: Equatable, Sendable { + /// Screen frames delivered by ScreenCaptureKit, before any pacing/shedding. + public var captured: Int + /// Frames actually appended to the encoder (real frames + frame-repeats that + /// hold a steady rate on a static screen), post-admission. The basis for + /// achieved fps. + public var encoded: Int + public var pacingDrops: Int + public var backpressureDrops: Int + public var compositorDrops: Int + public var admissionDrops: Int + + public init(captured: Int = 0, + encoded: Int = 0, + pacingDrops: Int = 0, + backpressureDrops: Int = 0, + compositorDrops: Int = 0, + admissionDrops: Int = 0) { + self.captured = captured + self.encoded = encoded + self.pacingDrops = pacingDrops + self.backpressureDrops = backpressureDrops + self.compositorDrops = compositorDrops + self.admissionDrops = admissionDrops + } + + /// Frames the pipeline wanted to encode but shed because it couldn't keep up + /// (backpressure) or the uplink was congested (admission). The HUD + adaptive + /// signal — pacing (intentional downsample) and compositor (overlay-only, frame + /// still sent) are deliberately excluded. + public var congestionDrops: Int { backpressureDrops + admissionDrops } + + /// Every shed frame across all four sites, for diagnostics. + public var totalDrops: Int { + pacingDrops + backpressureDrops + compositorDrops + admissionDrops + } + + public func count(of reason: FrameDropReason) -> Int { + switch reason { + case .pacing: return pacingDrops + case .backpressure: return backpressureDrops + case .compositor: return compositorDrops + case .admission: return admissionDrops + } + } +} + +/// Derived per-second rates between two `FrameTelemetrySnapshot`s. +public struct FrameTelemetryRates: Equatable, Sendable { + /// Measured encode rate (frames/sec reaching the encoder) — the real fps the + /// stream is achieving, distinct from the ABR's target fps. + public var achievedFrameRate: Int + /// Congestion frames (backpressure + admission) shed per second — a fast + /// keep-up signal for the adaptive logic (M9). + public var congestionDropsPerSecond: Int + + public init(achievedFrameRate: Int, congestionDropsPerSecond: Int) { + self.achievedFrameRate = achievedFrameRate + self.congestionDropsPerSecond = congestionDropsPerSecond + } +} + +/// Thread-safe cumulative counters for the capture→encode frame pipeline, shared +/// between the (serial sample-queue) capture output and the publisher actor so the +/// controller can read one snapshot per ~1 Hz telemetry poll. Mirrors +/// `VideoFrameAdmission`'s `OSAllocatedUnfairLock` pattern: a lock acquisition per +/// recorded frame is negligible next to per-frame CoreImage/VideoToolbox work. +/// +/// Lives in StreamCore (transport-agnostic, no HaishinKit/ScreenCaptureKit +/// dependency) so the counting and the rate math are unit-tested on CI, where the +/// iOS 27 app itself can't build. +public final class FrameTelemetry: @unchecked Sendable { + private let state = OSAllocatedUnfairLock(initialState: FrameTelemetrySnapshot()) + + public init() {} + + /// A screen frame arrived from ScreenCaptureKit (counted before pacing/shedding). + public func recordCaptured() { + state.withLock { $0.captured &+= 1 } + } + + /// A frame was appended to the encoder (real or frame-repeat), post-admission. + public func recordEncoded() { + state.withLock { $0.encoded &+= 1 } + } + + public func recordDrop(_ reason: FrameDropReason) { + state.withLock { snapshot in + switch reason { + case .pacing: snapshot.pacingDrops &+= 1 + case .backpressure: snapshot.backpressureDrops &+= 1 + case .compositor: snapshot.compositorDrops &+= 1 + case .admission: snapshot.admissionDrops &+= 1 + } + } + } + + public func snapshot() -> FrameTelemetrySnapshot { + state.withLock { $0 } + } + + public func reset() { + state.withLock { $0 = FrameTelemetrySnapshot() } + } + + /// Pure rate math (static so it is trivially unit-tested): the per-second + /// achieved fps and congestion-drop rate between an earlier and a later + /// snapshot over `elapsed` seconds. A non-positive or too-small `elapsed` + /// (clock hiccup, first poll) yields zero rather than a divide-by-zero spike; + /// counter deltas are floored at zero so a `reset()` between samples can't go + /// negative. + public static func rate(from previous: FrameTelemetrySnapshot, + to current: FrameTelemetrySnapshot, + elapsed: TimeInterval) -> FrameTelemetryRates { + guard elapsed >= 0.2 else { + return FrameTelemetryRates(achievedFrameRate: 0, congestionDropsPerSecond: 0) + } + let encodedDelta = max(0, current.encoded - previous.encoded) + let congestionDelta = max(0, current.congestionDrops - previous.congestionDrops) + let fps = Int((Double(encodedDelta) / elapsed).rounded()) + let drops = Int((Double(congestionDelta) / elapsed).rounded()) + return FrameTelemetryRates(achievedFrameRate: max(0, fps), + congestionDropsPerSecond: max(0, drops)) + } +} diff --git a/StreamCore/LiveStats.swift b/StreamCore/LiveStats.swift index d293d61..7b8eb01 100644 --- a/StreamCore/LiveStats.swift +++ b/StreamCore/LiveStats.swift @@ -5,9 +5,10 @@ import Foundation /// `BroadcastState`, which is the coarse cross-process *liveness* heartbeat — this /// is the fine-grained *metrics* the stats HUD renders while a stream is running. /// -/// The numbers are the encoder's current TARGETS after adaptive/thermal/path -/// clamping (the real, applied rate), not independently measured throughput — -/// measured fps/dropped-frame telemetry depends on M8, which is not yet built. +/// `bitRate`/`frameRate` are the encoder's current TARGETS after adaptive/thermal/ +/// path clamping (the real, applied rate); `achievedFrameRate`/`droppedFrames` are +/// the independently MEASURED pipeline telemetry (M8, issue #23) — the achieved +/// output rate and the frames congestion actually shed. /// /// Pure value type with no iOS dependencies, so the health classification and the /// display formatting are unit-tested in StreamCore on CI. @@ -22,12 +23,27 @@ public struct LiveStats: Equatable, Sendable { /// Consecutive ~1s ticks the socket sent zero bytes while data was queued — the /// stall signal the watchdog uses. public var zeroOutputSeconds: Int + /// Measured frames/sec actually reaching the encoder over the last poll window + /// (`FrameTelemetry`), distinct from the `frameRate` target. 0 before the first + /// rate is available (initial connect); the HUD falls back to `frameRate` then. + public var achievedFrameRate: Int + /// Cumulative frames congestion shed this session (backpressure + admission) — + /// the frames the pipeline wanted to encode but couldn't. Intentional pacing + /// downsampling is excluded, so this stays 0 on a healthy uplink. + public var droppedFrames: Int - public init(bitRate: Int, frameRate: Int, queueBytes: Int, zeroOutputSeconds: Int) { + public init(bitRate: Int, + frameRate: Int, + queueBytes: Int, + zeroOutputSeconds: Int, + achievedFrameRate: Int = 0, + droppedFrames: Int = 0) { self.bitRate = bitRate self.frameRate = frameRate self.queueBytes = queueBytes self.zeroOutputSeconds = zeroOutputSeconds + self.achievedFrameRate = achievedFrameRate + self.droppedFrames = droppedFrames } /// Uplink health, classified from the outbound backlog + stall signal. The @@ -77,6 +93,23 @@ public struct LiveStats: Equatable, Sendable { return "\(bytes / 1024) KB" } + /// The fps to show in the HUD: the measured `achievedFrameRate` once it is + /// available, falling back to the `frameRate` target for the first second (and + /// any window with no encoded frames) so the card never flashes a bare "0". + public var displayFrameRate: Int { + achievedFrameRate > 0 ? achievedFrameRate : frameRate + } + + /// A compact dropped-frame count for the card's congestion notice: "42", or + /// "1.2k" once it passes a thousand. Only shown when non-zero. + public var droppedLabel: String { + let dropped = max(0, droppedFrames) + if dropped >= 1000 { + return String(format: "%.1fk", Double(dropped) / 1000) + } + return "\(dropped)" + } + /// Formats an elapsed broadcast duration: "MM:SS" under an hour, "H:MM:SS" past /// it (e.g. 65 → "01:05", 3665 → "1:01:05"). Negative inputs clamp to zero. public static func uptimeLabel(seconds: Int) -> String { diff --git a/StreamCoreTests/FrameTelemetryTests.swift b/StreamCoreTests/FrameTelemetryTests.swift new file mode 100644 index 0000000..91f31a0 --- /dev/null +++ b/StreamCoreTests/FrameTelemetryTests.swift @@ -0,0 +1,123 @@ +import Testing +import StreamCore + +/// Unit tests for the frame-drop / achieved-fps telemetry (issue #23 / M8): the +/// cumulative counters, the congestion-vs-total accounting, and the pure per-second +/// rate math the stats HUD and the adaptive logic read. +@Suite struct FrameTelemetryCounterTests { + + @Test("A fresh telemetry is all zeros") + func freshIsZero() { + let snapshot = FrameTelemetry().snapshot() + #expect(snapshot == FrameTelemetrySnapshot()) + #expect(snapshot.captured == 0) + #expect(snapshot.encoded == 0) + #expect(snapshot.totalDrops == 0) + #expect(snapshot.congestionDrops == 0) + } + + @Test("Captured and encoded frames accumulate independently") + func capturedAndEncoded() { + let telemetry = FrameTelemetry() + for _ in 0..<5 { telemetry.recordCaptured() } + for _ in 0..<3 { telemetry.recordEncoded() } + let snapshot = telemetry.snapshot() + #expect(snapshot.captured == 5) + #expect(snapshot.encoded == 3) + } + + @Test("Each drop reason increments only its own counter") + func dropsByReason() { + let telemetry = FrameTelemetry() + telemetry.recordDrop(.pacing) + telemetry.recordDrop(.pacing) + telemetry.recordDrop(.backpressure) + telemetry.recordDrop(.compositor) + telemetry.recordDrop(.compositor) + telemetry.recordDrop(.compositor) + telemetry.recordDrop(.admission) + let snapshot = telemetry.snapshot() + #expect(snapshot.pacingDrops == 2) + #expect(snapshot.backpressureDrops == 1) + #expect(snapshot.compositorDrops == 3) + #expect(snapshot.admissionDrops == 1) + for reason in FrameDropReason.allCases { + #expect(snapshot.count(of: reason) == { + switch reason { + case .pacing: return 2 + case .backpressure: return 1 + case .compositor: return 3 + case .admission: return 1 + } + }()) + } + } + + @Test("Congestion drops are backpressure + admission only; total is all four") + func congestionExcludesPacingAndCompositor() { + let telemetry = FrameTelemetry() + telemetry.recordDrop(.pacing) // intentional downsample — excluded + telemetry.recordDrop(.compositor) // overlay only, frame still sent — excluded + telemetry.recordDrop(.backpressure) + telemetry.recordDrop(.admission) + telemetry.recordDrop(.admission) + let snapshot = telemetry.snapshot() + #expect(snapshot.congestionDrops == 3) // 1 backpressure + 2 admission + #expect(snapshot.totalDrops == 5) // + 1 pacing + 1 compositor + } + + @Test("reset() returns every counter to zero") + func resetZeroes() { + let telemetry = FrameTelemetry() + telemetry.recordCaptured() + telemetry.recordEncoded() + telemetry.recordDrop(.admission) + telemetry.reset() + #expect(telemetry.snapshot() == FrameTelemetrySnapshot()) + } +} + +@Suite struct FrameTelemetryRateTests { + + @Test("Achieved fps is the encoded delta over the elapsed window") + func achievedFrameRate() { + let previous = FrameTelemetrySnapshot(encoded: 100) + let current = FrameTelemetrySnapshot(encoded: 130) + #expect(FrameTelemetry.rate(from: previous, to: current, elapsed: 1.0).achievedFrameRate == 30) + // Over a 1.5s window the same 30 frames read as 20 fps (rounded). + #expect(FrameTelemetry.rate(from: previous, to: current, elapsed: 1.5).achievedFrameRate == 20) + // A short 0.5s window doubles the rate. + #expect(FrameTelemetry.rate(from: FrameTelemetrySnapshot(encoded: 0), + to: FrameTelemetrySnapshot(encoded: 30), + elapsed: 0.5).achievedFrameRate == 60) + } + + @Test("Congestion drops per second use backpressure + admission deltas") + func congestionDropsPerSecond() { + let previous = FrameTelemetrySnapshot(backpressureDrops: 1, admissionDrops: 1) + let current = FrameTelemetrySnapshot(pacingDrops: 999, // ignored + backpressureDrops: 3, + compositorDrops: 999, // ignored + admissionDrops: 2) + let rate = FrameTelemetry.rate(from: previous, to: current, elapsed: 1.0) + #expect(rate.congestionDropsPerSecond == 3) // (3-1) backpressure + (2-1) admission + } + + @Test("A window shorter than 0.2s yields zero instead of a divide spike") + func tooShortWindowIsZero() { + let previous = FrameTelemetrySnapshot(encoded: 0) + let current = FrameTelemetrySnapshot(encoded: 5) + let rate = FrameTelemetry.rate(from: previous, to: current, elapsed: 0.05) + #expect(rate == FrameTelemetryRates(achievedFrameRate: 0, congestionDropsPerSecond: 0)) + #expect(FrameTelemetry.rate(from: previous, to: current, elapsed: 0).achievedFrameRate == 0) + } + + @Test("A reset between samples floors deltas at zero (never negative)") + func resetBetweenSamplesFloorsAtZero() { + let previous = FrameTelemetrySnapshot(encoded: 100, backpressureDrops: 50, admissionDrops: 10) + let current = FrameTelemetrySnapshot(encoded: 5, backpressureDrops: 1) // counters reset then ran + let rate = FrameTelemetry.rate(from: previous, to: current, elapsed: 1.0) + #expect(rate.achievedFrameRate == 0) + #expect(rate.congestionDropsPerSecond == 0) + } +} diff --git a/StreamCoreTests/LiveStatsTests.swift b/StreamCoreTests/LiveStatsTests.swift index 3308fe8..db0e643 100644 --- a/StreamCoreTests/LiveStatsTests.swift +++ b/StreamCoreTests/LiveStatsTests.swift @@ -77,3 +77,46 @@ import StreamCore #expect(LiveStats.uptimeLabel(seconds: -5) == "00:00") } } + +/// The M8 telemetry fields folded into the HUD snapshot: the measured achieved fps +/// (with its target fallback) and the compact dropped-frame count. +@Suite struct LiveStatsTelemetryTests { + private func stats(frameRate: Int = 30, achieved: Int = 0, dropped: Int = 0) -> LiveStats { + LiveStats(bitRate: 3_000_000, frameRate: frameRate, queueBytes: 0, zeroOutputSeconds: 0, + achievedFrameRate: achieved, droppedFrames: dropped) + } + + @Test("displayFrameRate shows the measured rate, falling back to the target at zero") + func displayFrameRateFallback() { + // No measured rate yet (first second / an idle window) → show the target. + #expect(stats(frameRate: 60, achieved: 0).displayFrameRate == 60) + // A measured rate takes over once available — even when it differs from target. + #expect(stats(frameRate: 60, achieved: 52).displayFrameRate == 52) + #expect(stats(frameRate: 30, achieved: 30).displayFrameRate == 30) + } + + @Test("droppedLabel is a bare count under 1000 and abbreviates above it") + func droppedLabelFormatting() { + #expect(stats(dropped: 0).droppedLabel == "0") + #expect(stats(dropped: 42).droppedLabel == "42") + #expect(stats(dropped: 999).droppedLabel == "999") + #expect(stats(dropped: 1000).droppedLabel == "1.0k") + #expect(stats(dropped: 1500).droppedLabel == "1.5k") + #expect(stats(dropped: -5).droppedLabel == "0") // clamps + } + + @Test("The new telemetry fields participate in equality") + func equalityIncludesTelemetry() { + #expect(stats(achieved: 30, dropped: 0) != stats(achieved: 29, dropped: 0)) + #expect(stats(achieved: 30, dropped: 0) != stats(achieved: 30, dropped: 4)) + #expect(stats(achieved: 30, dropped: 4) == stats(achieved: 30, dropped: 4)) + } + + @Test("The telemetry fields default to zero for existing call sites") + func defaultsAreZero() { + let base = LiveStats(bitRate: 1_000_000, frameRate: 30, queueBytes: 0, zeroOutputSeconds: 0) + #expect(base.achievedFrameRate == 0) + #expect(base.droppedFrames == 0) + #expect(base.displayFrameRate == 30) // falls back to the target + } +}