Fix WHEP playback: track retention, vanilla-ICE offer, RTP depacketization, playout pacing

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

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

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

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,46 +86,79 @@ 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 {
         guard let location, _readyState.value == .open else {
             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") {

RTCHaishinKit/Sources/RTC/RTCPeerConnection.swift

@@ -77,15 +77,42 @@ 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
 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)

RTCHaishinKit/Sources/RTP/RTPH264Packetizer.swift

@@ -129,6 +129,8 @@ final class RTPH264Packetizer<T: RTPPacketizerDelegate>: RTPPacketizer {
         switch nalUnitType {
         case 1...23:
             decodeSingleNALUnit(packet)
+        case 24:
+            decodeSingleTimeAggregation(packet)
         case 28:
             decodeFragmentUnitA(packet)
         default:
@@ -137,32 +139,60 @@ final class RTPH264Packetizer<T: RTPPacketizerDelegate>: 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..<offset + size))
+            offset += size
+        }
+        flushIfMarked(packet)
+    }
+
+    private func appendNALUnit(_ nal: Data) {
+        guard !nal.isEmpty else {
+            return
+        }
+        let nalUnitType = nal[nal.startIndex] & 0x1F
         switch nalUnitType {
-        case 5: // idr
-            buffer.append(RTPH264Packetizer_startCode)
-            buffer.append(packet.payload)
         case 7: // sps
             if sequenceParameterSets == nil {
-                sequenceParameterSets = packet.payload
+                sequenceParameterSets = nal
             }
         case 8: // pps
             if pictureParameterSets == nil {
-                pictureParameterSets = packet.payload
+                pictureParameterSets = nal
             }
-        default:
+        default: // idr/non-idr slices and everything else
             buffer.append(RTPH264Packetizer_startCode)
-            buffer.append(packet.payload)
+            buffer.append(nal)
         }
         if formatDescription == nil && sequenceParameterSets != nil && pictureParameterSets != nil {
             formatDescription = makeFormatDescription()
         }
-        if packet.marker {
-            if let sampleBuffer = makeSampleBuffer(&buffer, timestamp: packet.timestamp) {
-                delegate?.packetizer(self, didOutput: sampleBuffer)
-            }
-            buffer.removeAll(keepingCapacity: false)
+    }
+
+    private func flushIfMarked(_ packet: RTPPacket) {
+        guard packet.marker else {
+            return
+        }
+        if let sampleBuffer = makeSampleBuffer(&buffer, timestamp: packet.timestamp) {
+            delegate?.packetizer(self, didOutput: sampleBuffer)
         }
+        buffer.removeAll(keepingCapacity: false)
     }
 
     private func decodeFragmentUnitA(_ packet: RTPPacket) {
@@ -191,12 +221,16 @@ final class RTPH264Packetizer<T: RTPPacketizerDelegate>: 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<T: RTPPacketizerDelegate>: 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?