From f3aeb3b5464bf7708e1a57a1c91cf4f68f59c938 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Ku=C5=82aczkowski?= Date: Thu, 11 Jun 2026 17:46:34 +0200 Subject: [PATCH] Fix WHEP/WHIP: track retention, vanilla-ICE offer, RTP depacketization, playout pacing Fixes a chain of issues that prevented HTTP (WHEP) playback from ever delivering a frame, verified end-to-end against MediaMTX (0 frames -> ~30fps): - HTTPSession: retain playback tracks. RTCTrack.deinit calls rtcDeleteTrack, so discarding the addTrack(...) result deleted the native tracks before negotiation ("No DataChannel or Track to negotiate" -> connect always threw). - HTTPSession: libdatachannel's rtcCreateOffer requires disableAutoNegotiation and returns RTC_ERR_FAILURE under the default config. Use the canonical flow: setLocalDescription, wait for ICE gathering to complete, then read rtcGetLocalDescription - the offer then also carries the gathered candidates, which a non-trickle client needs for the server to reach it. - HTTPSession: throw on non-2xx WHEP/WHIP responses instead of feeding error bodies into setRemoteDescription ("Remote description has no ICE user fragment"). - HTTPSession: retry playback with a video-only offer when the server rejects the audio m-line (e.g. MediaMTX "codecs not supported by client" for any stream without Opus audio). Opus streams keep audio+video. - HTTPSession: convert URL userinfo into an HTTP Basic Authorization header (URLSession does not transmit userinfo; MediaMTX authenticates WHIP this way). - RTCPeerConnection: offer multiple H264 profile variants (42e01f/42c01f/ 42001f/4d001f/64001f) like browsers do; a single hardcoded profile is rejected for streams whose profile bytes differ. Adds currentLocalDescription() wrapping rtcGetLocalDescription. - RTPJitterBuffer: prime the expected sequence from the first packet (RTP sequence numbers start at a random value per RFC 3550), drop late packets wrap-aware, and jump over gaps once the reorder window fills. The previous advance-by-one stale handling let expectedSequence run past the live sequence after the first loss, permanently stalling delivery. - RTPH264Packetizer: implement STAP-A (RFC 6184 5.7.1) - SPS/PPS commonly arrive aggregated; without it no frame ever decodes. Accumulate FU-A NAL units into full access units and emit on the RTP marker (multi-slice frames were emitted per-slice -> kVTVideoDecoderBadDataErr). Build the AVCC buffer forward from parsed units; the in-place start-code rewrite corrupted access units whose NAL lengths fall in 256-511 (a written length 00 00 01 xx re-matches as a start code). - MediaLink: use the audio clock for playout pacing only while it is advancing; an attached but silent AudioPlayerNode reports currentTime == 0 forever, pinning video at the first frame for video-only streams. - VTDecompressionSession: log decode failures (throttled) instead of silently dropping them. - DisplayLinkChoreographer (macOS): fall back to the display refresh period when frameInterval is 0 so elapsed-time consumers advance. --- .../VTDecompressionSession+Extension.swift | 15 +++ .../Screen/DisplayLinkChoreographer.swift | 14 ++- HaishinKit/Sources/Stream/MediaLink.swift | 9 +- RTCHaishinKit/Sources/HTTP/HTTPSession.swift | 101 +++++++++++++++--- .../Sources/RTC/RTCPeerConnection.swift | 39 ++++++- .../Sources/RTP/RTPH264Packetizer.swift | 84 +++++++++++---- .../Sources/RTP/RTPJitterBuffer.swift | 40 +++++-- 7 files changed, 258 insertions(+), 44 deletions(-) diff --git a/HaishinKit/Sources/Extension/VTDecompressionSession+Extension.swift b/HaishinKit/Sources/Extension/VTDecompressionSession+Extension.swift index 6f7b6cc50..a43c28998 100644 --- a/HaishinKit/Sources/Extension/VTDecompressionSession+Extension.swift +++ b/HaishinKit/Sources/Extension/VTDecompressionSession+Extension.swift @@ -1,6 +1,20 @@ import Foundation import VideoToolbox +/// Per-frame decode failures were previously swallowed (`guard let imageBuffer +/// else { return }` ignored the status), which made bitstream-level bugs +/// (e.g. partial access units) invisible. Throttled so a continuously-failing +/// stream cannot flood the log. +enum DecodeFailureLog { + nonisolated(unsafe) static var count = 0 + static func log(_ status: OSStatus) { + count += 1 + if count <= 5 || count % 300 == 0 { + logger.warn("video decode failed #\(count) status=\(status)") + } + } +} + extension VTDecompressionSession: VTSessionConvertible { static let defaultDecodeFlags: VTDecodeFrameFlags = [ ._EnableAsynchronousDecompression, @@ -18,6 +32,7 @@ extension VTDecompressionSession: VTSessionConvertible { infoFlagsOut: &flagsOut, outputHandler: { status, _, imageBuffer, presentationTimeStamp, duration in guard let imageBuffer else { + DecodeFailureLog.log(status) return } var status = noErr diff --git a/HaishinKit/Sources/Screen/DisplayLinkChoreographer.swift b/HaishinKit/Sources/Screen/DisplayLinkChoreographer.swift index d3fe32ba4..73880c43e 100644 --- a/HaishinKit/Sources/Screen/DisplayLinkChoreographer.swift +++ b/HaishinKit/Sources/Screen/DisplayLinkChoreographer.swift @@ -54,7 +54,19 @@ final class DisplayLink: NSObject, @unchecked Sendable { } if frameInterval == 0 || frameInterval <= inNow.pointee.timestamp - self.timestamp { self.timestamp = Double(inNow.pointee.timestamp) - self.targetTimestamp = self.timestamp + frameInterval + // With the default preferredFramesPerSecond (0), frameInterval is 0 and + // targetTimestamp == timestamp — consumers measuring elapsed time per tick + // (MediaLink playout pacing) accumulate zero and never advance. Fall back + // to the display's actual refresh period. + let interval: Double + if 0 < frameInterval { + interval = frameInterval + } else if 0 < inNow.pointee.videoTimeScale && 0 < inNow.pointee.videoRefreshPeriod { + interval = Double(inNow.pointee.videoRefreshPeriod) / Double(inNow.pointee.videoTimeScale) + } else { + interval = 1.0 / 60.0 + } + self.targetTimestamp = self.timestamp + interval _ = self.delegate?.perform(self.selector, with: self) } return kCVReturnSuccess diff --git a/HaishinKit/Sources/Stream/MediaLink.swift b/HaishinKit/Sources/Stream/MediaLink.swift index 4d83ace89..971124560 100644 --- a/HaishinKit/Sources/Stream/MediaLink.swift +++ b/HaishinKit/Sources/Stream/MediaLink.swift @@ -52,7 +52,14 @@ final actor MediaLink { defer { duration += timestamp } - return await audioPlayer?.currentTime ?? duration + // Use the audio clock only when it is actually advancing. An attached + // AudioPlayerNode that is not playing (video-only stream, or audio not + // yet flowing) reports currentTime == 0 forever, which pinned the video + // playout position at the first frame. + if let audioTime = await audioPlayer?.currentTime, 0 < audioTime { + return audioTime + } + return duration } } diff --git a/RTCHaishinKit/Sources/HTTP/HTTPSession.swift b/RTCHaishinKit/Sources/HTTP/HTTPSession.swift index ae5cbca1f..700c6e461 100644 --- a/RTCHaishinKit/Sources/HTTP/HTTPSession.swift +++ b/RTCHaishinKit/Sources/HTTP/HTTPSession.swift @@ -22,6 +22,10 @@ actor HTTPSession: StreamSession { private var mode: StreamSessionMode private var configuration: HTTPSessionConfiguration? private var peerConnection: RTCPeerConnection? + // Playback tracks MUST be retained: RTCTrack.deinit calls rtcDeleteTrack, so + // discarding the addTrack(...) return value deletes the native track before + // negotiation → "No DataChannel or Track to negotiate" → connect always fails. + private var tracks: [RTCTrack] = [] init(uri: URL, mode: StreamSessionMode, configuration: (any StreamSessionConfiguration)?) { logger.level = .debug @@ -41,6 +45,38 @@ actor HTTPSession: StreamSession { return } _readyState.value = .connecting + do { + do { + try await attemptConnect(videoOnly: false) + } catch let error as URLError where mode == .playback && error.localizedDescription.contains("codecs not supported") { + // MediaMTX hard-rejects ("codecs not supported by client", HTTP 400) + // any offer whose audio m-line it cannot satisfy — i.e. whenever the + // published stream has no WebRTC-compatible (Opus) audio track, which + // is every AAC/no-audio stream. Browsers slip through; this client's + // single-opus audio section does not. Retry once with a video-only + // offer: video plays for AAC/no-audio streams, and Opus streams keep + // full audio+video via the first attempt. + logger.warn("WHEP offer rejected for audio codec; retrying video-only") + teardownAttempt() + try await attemptConnect(videoOnly: true) + } + _readyState.value = .open + } catch { + logger.warn(error) + await _stream.close() + teardownAttempt() + _readyState.value = .closed + throw error + } + } + + private func teardownAttempt() { + tracks.removeAll() + peerConnection?.close() + peerConnection = nil + } + + private func attemptConnect(videoOnly: Bool) async throws { let peerConnection = try makePeerConnection() switch mode { case .publish: @@ -50,22 +86,28 @@ actor HTTPSession: StreamSession { try peerConnection.addTrack(VideoStreamTrack(videoSettings), stream: _stream) case .playback: await _stream.setDirection(.recvonly) - try peerConnection.addTrack(.audio, stream: _stream) - try peerConnection.addTrack(.video, stream: _stream) + // Retain the returned tracks: RTCTrack.deinit calls rtcDeleteTrack, so + // discarding them deletes the native tracks before negotiation + // ("No DataChannel or Track to negotiate"). + if !videoOnly { + tracks.append(try peerConnection.addTrack(.audio, stream: _stream)) + } + tracks.append(try peerConnection.addTrack(.video, stream: _stream)) } - do { - self.peerConnection = peerConnection - try peerConnection.setLocalDesciption(.offer) - let answer = try await requestOffer(uri, offer: peerConnection.createOffer()) - try peerConnection.setRemoteDesciption(answer, type: .answer) - _readyState.value = .open - } catch { - logger.warn(error) - await _stream.close() - peerConnection.close() - _readyState.value = .closed - throw error + self.peerConnection = peerConnection + try peerConnection.setLocalDesciption(.offer) + // Canonical libdatachannel (vanilla-ICE) WHIP/WHEP flow: createOffer() + // requires disableAutoNegotiation and returns RTC_ERR_FAILURE under the + // default auto-negotiation config. Instead, wait for ICE gathering to + // complete, then read the local description back — it now contains the + // gathered candidates, which this non-trickle client (no PATCH) needs + // in the offer for the server to reach it. + for _ in 0..<60 where peerConnection.iceGatheringState != .complete { + try await Task.sleep(nanoseconds: 50_000_000) } + let offer = try peerConnection.currentLocalDescription() + let answer = try await requestOffer(uri, offer: offer) + try peerConnection.setRemoteDesciption(answer, type: .answer) } func close() async throws { @@ -73,23 +115,50 @@ actor HTTPSession: StreamSession { return } _readyState.value = .closing - var request = URLRequest(url: location) + var request = makeRequest(location) request.httpMethod = "DELETE" request.addValue("application/sdp", forHTTPHeaderField: "Content-Type") _ = try await URLSession.shared.data(for: request) await _stream.close() + tracks.removeAll() peerConnection?.close() self.location = nil _readyState.value = .closed } + /// Builds a request for the WHIP/WHEP endpoint, converting URL userinfo + /// (https://user:pass@host/…) into an HTTP Basic Authorization header. + /// URLSession does not transmit userinfo, and MediaMTX authenticates WHIP + /// publishing via Basic auth (query parameters are not accepted there, + /// unlike its RTMP/HLS endpoints). + private func makeRequest(_ url: URL) -> URLRequest { + var components = URLComponents(url: url, resolvingAgainstBaseURL: false) + let user = components?.user + let password = components?.password + components?.user = nil + components?.password = nil + var request = URLRequest(url: components?.url ?? url) + if let user, !user.isEmpty { + let credentials = Data("\(user):\(password ?? "")".utf8).base64EncodedString() + request.setValue("Basic \(credentials)", forHTTPHeaderField: "Authorization") + } + return request + } + private func requestOffer(_ url: URL, offer: String) async throws -> String { logger.debug(offer) - var request = URLRequest(url: url) + var request = makeRequest(url) request.httpMethod = "POST" request.addValue("application/sdp", forHTTPHeaderField: "Content-Type") request.httpBody = offer.data(using: .utf8) let (data, response) = try await URLSession.shared.data(for: request) + // Surface server-side rejections: upstream ignored the HTTP status and fed + // error bodies (JSON) into setRemoteDescription, masking the real reason. + if let http = response as? HTTPURLResponse, !(200...299).contains(http.statusCode) { + let body = String(data: data, encoding: .utf8) ?? "" + logger.warn("WHEP/WHIP endpoint rejected offer: HTTP \(http.statusCode) — \(body)") + throw URLError(.badServerResponse, userInfo: [NSLocalizedDescriptionKey: "HTTP \(http.statusCode): \(body)"]) + } if let response = response as? HTTPURLResponse { if let location = response.allHeaderFields["Location"] as? String { if location.hasSuffix("http") { diff --git a/RTCHaishinKit/Sources/RTC/RTCPeerConnection.swift b/RTCHaishinKit/Sources/RTC/RTCPeerConnection.swift index 817dd2020..9f972cdee 100644 --- a/RTCHaishinKit/Sources/RTC/RTCPeerConnection.swift +++ b/RTCHaishinKit/Sources/RTC/RTCPeerConnection.swift @@ -77,8 +77,15 @@ a=rtpmap:111 opus/48000/2 a=fmtp:111 minptime=10;useinbandfec=1;stereo=1;sprop-stereo=1 """ + // Offer MULTIPLE H264 profile variants like a browser does. MediaMTX/Pion + // matches profile-level-id's profile bytes strictly; a single hardcoded + // 42e01f offer is rejected ("codecs not supported by client") whenever the + // published stream is plain/constrained baseline (42001f/42c01f), main + // (4d001f) or high (64001f) — i.e. most drones and third-party encoders. + // The server's answer picks exactly one payload; the depacketizer follows + // the negotiated description, so extra variants are harmless. static let videoMediaDescription = """ -m=video 9 UDP/TLS/RTP/SAVPF 98 +m=video 9 UDP/TLS/RTP/SAVPF 98 99 100 101 102 a=mid:1 a=recvonly a=rtpmap:98 H264/90000 @@ -86,6 +93,26 @@ a=rtcp-fb:98 goog-remb a=rtcp-fb:98 nack a=rtcp-fb:98 nack pli a=fmtp:98 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f +a=rtpmap:99 H264/90000 +a=rtcp-fb:99 goog-remb +a=rtcp-fb:99 nack +a=rtcp-fb:99 nack pli +a=fmtp:99 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42c01f +a=rtpmap:100 H264/90000 +a=rtcp-fb:100 goog-remb +a=rtcp-fb:100 nack +a=rtcp-fb:100 nack pli +a=fmtp:100 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42001f +a=rtpmap:101 H264/90000 +a=rtcp-fb:101 goog-remb +a=rtcp-fb:101 nack +a=rtcp-fb:101 nack pli +a=fmtp:101 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=4d001f +a=rtpmap:102 H264/90000 +a=rtcp-fb:102 goog-remb +a=rtcp-fb:102 nack +a=rtcp-fb:102 nack pli +a=fmtp:102 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=64001f """ static let bufferSize: Int = 1024 * 16 @@ -263,6 +290,16 @@ a=fmtp:98 level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f } } + /// Returns the CURRENT local description via rtcGetLocalDescription — unlike + /// `createOffer()` (which requires disableAutoNegotiation and returns + /// RTC_ERR_FAILURE otherwise), this works with auto-negotiation and, when read + /// after ICE gathering completes, includes the gathered candidates (vanilla ICE). + public func currentLocalDescription() throws -> String { + return try CUtil.getString { buffer, size in + rtcGetLocalDescription(connection, buffer, size) + } + } + public func createAnswer() throws -> String { return try CUtil.getString { buffer, size in rtcCreateAnswer(connection, buffer, size) diff --git a/RTCHaishinKit/Sources/RTP/RTPH264Packetizer.swift b/RTCHaishinKit/Sources/RTP/RTPH264Packetizer.swift index 28de28115..317de422a 100644 --- a/RTCHaishinKit/Sources/RTP/RTPH264Packetizer.swift +++ b/RTCHaishinKit/Sources/RTP/RTPH264Packetizer.swift @@ -129,6 +129,8 @@ final class RTPH264Packetizer: RTPPacketizer { switch nalUnitType { case 1...23: decodeSingleNALUnit(packet) + case 24: + decodeSingleTimeAggregation(packet) case 28: decodeFragmentUnitA(packet) default: @@ -137,32 +139,60 @@ final class RTPH264Packetizer: RTPPacketizer { } private func decodeSingleNALUnit(_ packet: RTPPacket) { - let nalUnitType = packet.payload[0] & 0x1F + appendNALUnit(packet.payload) + flushIfMarked(packet) + } + + /// STAP-A (RFC 6184 §5.7.1): one RTP payload aggregating several NAL units, + /// each prefixed by a 2-byte big-endian size. MediaMTX/Pion delivers SPS+PPS + /// (and other small NALs) this way — without STAP-A support the decoder never + /// receives parameter sets, so no frame is ever emitted. + private func decodeSingleTimeAggregation(_ packet: RTPPacket) { + let payload = Data(packet.payload) + var offset = 1 + while offset + 2 <= payload.count { + let size = Int(payload[offset]) << 8 | Int(payload[offset + 1]) + offset += 2 + guard 0 < size, offset + size <= payload.count else { + break + } + appendNALUnit(payload.subdata(in: offset..: RTPPacketizer { } if end && fragmentedStarted { - if let buffer = makeSampleBuffer(&fragmentedBuffer, timestamp: fragmentedTimestamp) { - delegate?.packetizer(self, didOutput: buffer) - } - // flush buffers + // A completed FU-A is ONE NAL unit (one slice), not a whole access + // unit — multi-slice frames (x264 zerolatency uses several slices per + // frame) span several FU-As. Emitting per-fragment hands VideoToolbox + // a partial frame → kVTVideoDecoderBadDataErr on every frame. Append + // to the AU accumulator and emit only at the RTP marker (end of AU), + // same as the single-NAL and STAP-A paths. + buffer.append(fragmentedBuffer) fragmentedBuffer.removeAll(keepingCapacity: false) fragmentedStarted = false + flushIfMarked(packet) } } @@ -207,7 +241,21 @@ final class RTPH264Packetizer: RTPPacketizer { let presentationTimeStamp: CMTime = self.timestamp.convert(timestamp) let units = nalUnitReader.read(&buffer, type: H264NALUnit.self) var blockBuffer: CMBlockBuffer? - ISOTypeBufferUtil.toNALFileFormat(&buffer) + // Build the AVCC (length-prefixed) buffer FORWARD from the parsed units. + // The previous in-place ISOTypeBufferUtil.toNALFileFormat conversion is + // unsafe here: a written 4-byte length of 256–511 is 00 00 01 xx, which + // its continuing reverse scan re-matches as a start code and "converts" + // again — corrupting the access unit (VT kVTVideoDecoderBadDataErr -12909 + // on every frame). NALUnitReader.read returns units last-first, so append + // them reversed to preserve decode order. + var avcc = Data(capacity: buffer.count) + for unit in units.reversed() { + let unitData = unit.data + var length = UInt32(unitData.count).bigEndian + withUnsafeBytes(of: &length) { avcc.append(contentsOf: $0) } + avcc.append(unitData) + } + buffer = avcc blockBuffer = buffer.makeBlockBuffer() var sampleSizes: [Int] = [] var sampleBuffer: CMSampleBuffer? diff --git a/RTCHaishinKit/Sources/RTP/RTPJitterBuffer.swift b/RTCHaishinKit/Sources/RTP/RTPJitterBuffer.swift index aee2d471b..326d7ad17 100644 --- a/RTCHaishinKit/Sources/RTP/RTPJitterBuffer.swift +++ b/RTCHaishinKit/Sources/RTP/RTPJitterBuffer.swift @@ -9,18 +9,44 @@ final class RTPJitterBuffer { private var buffer: [UInt16: RTPPacket] = [:] private var expectedSequence: UInt16 = 0 - private let stalePacketCounts: Int = 4 + private var primed = false + /// Reorder window before declaring the gap lost and skipping ahead. A + /// fragmented IDR can span 20+ packets, so this must exceed one frame. + private let maxBuffered = 32 func append(_ packet: RTPPacket) { + // RTP sequence numbers start at a RANDOM value (RFC 3550 §5.1); waiting + // for sequence 0 stalls delivery for up to ~65k packets (minutes of + // received-but-undecoded media). Prime from the first packet seen. + if !primed { + expectedSequence = packet.sequenceNumber + primed = true + } + // Drop packets older than the playout point (wrap-aware signed distance); + // they were already delivered or skipped. Without this, a late packet + // sits in the dictionary forever and permanently inflates the count. + if Int16(bitPattern: packet.sequenceNumber &- expectedSequence) < 0 { + return + } buffer[packet.sequenceNumber] = packet - - while let packet = buffer[expectedSequence] { - delegate?.jitterBuffer(self, sequenced: packet) - buffer.removeValue(forKey: expectedSequence) - expectedSequence &+= 1 + drain() + // Gap recovery: a lost packet would otherwise block delivery forever + // (the old "advance by 1 per append" crawl let expectedSequence run PAST + // the live sequence, wedging the stream permanently). Once the reorder + // window fills, jump straight to the oldest buffered packet and resume. + if maxBuffered <= buffer.count { + if let oldest = buffer.keys.min(by: { + Int16(bitPattern: $0 &- expectedSequence) < Int16(bitPattern: $1 &- expectedSequence) + }) { + expectedSequence = oldest + drain() + } } + } - if stalePacketCounts <= buffer.count { + private func drain() { + while let packet = buffer.removeValue(forKey: expectedSequence) { + delegate?.jitterBuffer(self, sequenced: packet) expectedSequence &+= 1 } }