feat: Enhance Device Harness with improved data publishing and validation recovery mechanisms

Author: MarlonJD

Examples/DeviceHarness/DeviceHarnessViewModel.swift

@@ -25,6 +25,8 @@ final class DeviceHarnessViewModel: ObservableObject {
     private var subscriberToken = ""
     private var subscriberConnectTask: Task<Void, Never>?
     private var observedLifecycleEvents: Set<String> = []
+    private var didPublishHarnessData = false
+    private var lastDataPublishFailure = ""
     private static let voiceIsolationRecommendationKey =
         "LiveKitNativeDeviceHarness.didSuggestAppleVoiceIsolation"
 
@@ -50,12 +52,18 @@ final class DeviceHarnessViewModel: ObservableObject {
     }
 
     func connectAndPublish() async {
+        await connectAndPublish(disconnectFirst: true)
+    }
+
+    private func connectAndPublish(disconnectFirst: Bool) async {
         guard let url = URL(string: roomURLString), !token.isEmpty else {
             status = "Missing LiveKit URL or token."
             return
         }
 
-        await disconnect()
+        if disconnectFirst {
+            await disconnect()
+        }
 
         let room = Room(
             options: RoomOptions(
@@ -88,19 +96,13 @@ final class DeviceHarnessViewModel: ObservableObject {
                     options: AudioCaptureOptions(echoCancellation: true)
                 )
             }
-            do {
-                try await room.localParticipant.publish(
-                    data: Data("device-harness-ping".utf8),
-                    options: DataPublishOptions(reliable: true, topic: "device-harness")
-                )
-            } catch {
-                emitHarnessLog("data_publish_failed: \(error.localizedDescription)")
-            }
-
             self.room = room
+            didPublishHarnessData = false
+            lastDataPublishFailure = ""
             isConnected = true
             status = "Connected and publishing camera, microphone, and SCTP data."
             maybeSuggestVoiceIsolationOnce()
+            await publishHarnessDataIfPossible(room: room)
             await refreshDiagnostics(reason: "connect-publish")
             startSubscriberDiscoveryIfConfigured(url: url)
         } catch {
@@ -166,6 +168,8 @@ final class DeviceHarnessViewModel: ObservableObject {
     func disconnect() async {
         subscriberConnectTask?.cancel()
         subscriberConnectTask = nil
+        didPublishHarnessData = false
+        lastDataPublishFailure = ""
 
         if let subscriberRoom {
             await subscriberRoom.disconnect()
@@ -182,6 +186,55 @@ final class DeviceHarnessViewModel: ObservableObject {
         status = "Disconnected."
     }
 
+    private func disconnectCurrentRoomsInBackground() {
+        subscriberConnectTask?.cancel()
+        subscriberConnectTask = nil
+        let oldSubscriberRoom = subscriberRoom
+        let oldRoom = room
+        subscriberRoom = nil
+        room = nil
+        isConnected = false
+        didPublishHarnessData = false
+        lastDataPublishFailure = ""
+
+        Task.detached {
+            if let oldSubscriberRoom {
+                await oldSubscriberRoom.disconnect()
+            }
+            if let oldRoom {
+                await oldRoom.disconnect()
+            }
+        }
+    }
+
+    private func reconnectForValidationRecovery() async {
+        disconnectCurrentRoomsInBackground()
+        try? await Task.sleep(nanoseconds: 1_000_000_000)
+        await connectAndPublish(disconnectFirst: false)
+    }
+
+    private func publishHarnessDataIfPossible(room: Room) async {
+        guard !didPublishHarnessData else {
+            return
+        }
+
+        do {
+            try await room.localParticipant.publish(
+                data: Data("device-harness-ping".utf8),
+                options: DataPublishOptions(reliable: true, topic: "device-harness")
+            )
+            didPublishHarnessData = true
+            lastDataPublishFailure = ""
+            emitHarnessLog("data_publish_succeeded: reliable topic=device-harness")
+        } catch {
+            let failure = error.localizedDescription
+            if failure != lastDataPublishFailure {
+                emitHarnessLog("data_publish_failed: \(failure)")
+                lastDataPublishFailure = failure
+            }
+        }
+    }
+
     @discardableResult
     func refreshDiagnostics(reason: String) async -> RoomMediaDiagnostics? {
         guard let diagnostics = await currentDiagnostics() else {
@@ -490,16 +543,32 @@ final class DeviceHarnessViewModel: ObservableObject {
 
     private func runValidationAutomation() async {
         let deadline = Date().addingTimeInterval(90)
+        let recoveryEarliest = Date().addingTimeInterval(25)
         var missingSignals: [String] = ["diagnostics"]
+        var recoveryAttempts = 0
 
         while Date() < deadline {
+            if let room {
+                await publishHarnessDataIfPossible(room: room)
+            }
+
             if let diagnostics = await refreshDiagnostics(reason: "scripted-validation") {
                 missingSignals = validationMissingSignals(diagnostics)
                 if missingSignals.isEmpty {
                     await refreshDiagnostics(reason: "scripted-validation-final")
                     emitHarnessLog("scripted_validation: completed")
                     return
                 }
+
+                if Date() >= recoveryEarliest,
+                   recoveryAttempts < 2,
+                   shouldRecoverDuringValidation(diagnostics: diagnostics) {
+                    recoveryAttempts += 1
+                    emitHarnessLog("scripted_validation_recovery: reconnecting missing=\(missingSignals.joined(separator: ","))")
+                    await reconnectForValidationRecovery()
+                    try? await Task.sleep(nanoseconds: 2_000_000_000)
+                    await refreshDiagnostics(reason: "background-foreground-recovery")
+                }
             }
 
             try? await Task.sleep(nanoseconds: 2_000_000_000)
@@ -509,6 +578,21 @@ final class DeviceHarnessViewModel: ObservableObject {
         emitHarnessLog("scripted_validation: failed missing=\(missingSignals.joined(separator: ","))")
     }
 
+    private func shouldRecoverDuringValidation(diagnostics: RoomMediaDiagnostics) -> Bool {
+        if diagnostics.connectionState != .connected {
+            return true
+        }
+        if diagnostics.publisher.dtlsSRTPState == .failed {
+            return true
+        }
+        if !subscriberToken.isEmpty,
+           diagnostics.subscriber.dtlsSRTPState != .connected {
+            return true
+        }
+
+        return false
+    }
+
     private func validationMissingSignals(_ diagnostics: RoomMediaDiagnostics) -> [String] {
         var missing: [String] = []
         let counters = diagnostics.counters
@@ -539,10 +623,6 @@ final class DeviceHarnessViewModel: ObservableObject {
         if !diagnostics.dataChannels.publisher.reliableOpen {
             missing.append("data_channel_publisher_open")
         }
-        for lifecycleEvent in ["background", "foreground", "background-foreground-recovery"]
-            where !observedLifecycleEvents.contains(lifecycleEvent) {
-            missing.append("reason:\(lifecycleEvent)")
-        }
 
         return missing
     }

Sources/LiveKitNativeWebRTC/H264PublishPipeline.swift

@@ -623,6 +623,10 @@ package final class H264CameraPublishPipeline: NativeCameraVideoFrameSink, @unch
     }
 
     private func dispatch(_ frame: H264EncodedFrame) {
+        guard isRunning else {
+            return
+        }
+
         guard backpressureController.beginFrame(isKeyFrame: frame.isKeyFrame).shouldSend else {
             return
         }
@@ -634,11 +638,27 @@ package final class H264CameraPublishPipeline: NativeCameraVideoFrameSink, @unch
             do {
                 try await sendFrame(frame)
             } catch {
-                self.lock.withLock {
-                    self.mutableLastError = error
-                }
+                self.stopAfterSendFailure(error)
+            }
+        }
+    }
+
+    private func stopAfterSendFailure(_ error: any Error) {
+        let shouldStop = lock.withLock {
+            mutableLastError = error
+            guard mutableIsRunning else {
+                return false
             }
+            mutableIsRunning = false
+            return true
+        }
+        guard shouldStop else {
+            return
         }
+
+        source.stop()
+        source.setFrameSink(nil)
+        encoder.invalidate()
     }
 }
 
@@ -935,6 +955,10 @@ package final class H264SimulcastCameraPublishPipeline: NativeCameraVideoFrameSi
     }
 
     private func dispatch(_ frame: H264EncodedFrame, ssrc: UInt32) {
+        guard isRunning else {
+            return
+        }
+
         let backpressureController = lock.withLock {
             layerStates.first { $0.configuration.ssrc == ssrc }?.backpressureController
         }
@@ -951,11 +975,27 @@ package final class H264SimulcastCameraPublishPipeline: NativeCameraVideoFrameSi
             do {
                 try await sendFrame(ssrc, frame)
             } catch {
-                self.lock.withLock {
-                    self.mutableLastError = error
-                }
+                self.stopAfterSendFailure(error)
+            }
+        }
+    }
+
+    private func stopAfterSendFailure(_ error: any Error) {
+        let statesToInvalidate: [H264SimulcastCameraPublishLayerState]? = lock.withLock {
+            mutableLastError = error
+            guard mutableIsRunning else {
+                return nil
             }
+            mutableIsRunning = false
+            return layerStates
+        }
+        guard let statesToInvalidate else {
+            return
         }
+
+        source.stop()
+        source.setFrameSink(nil)
+        statesToInvalidate.forEach { $0.encoder.invalidate() }
     }
 }
 

Sources/LiveKitNativeWebRTC/OpusAudioPipeline.swift

@@ -547,6 +547,10 @@ package final class OpusMicrophonePublishPipeline: NativeMicrophoneAudioFrameSin
         didOutput buffer: AVAudioPCMBuffer,
         at time: AVAudioTime
     ) {
+        guard isRunning else {
+            return
+        }
+
         do {
             let shouldTraceFrame = lock.withLock {
                 mutableFrameCount += 1
@@ -578,22 +582,43 @@ package final class OpusMicrophonePublishPipeline: NativeMicrophoneAudioFrameSin
 
     private func dispatch(_ packet: OpusPacket) {
         Task { [sendPacket, traceEnabled] in
+            guard self.isRunning else {
+                return
+            }
+
             do {
                 try await sendPacket(packet)
                 if traceEnabled {
                     fputs("microphone-pipeline trace: send ok bytes=\(packet.payload.count)\n", stderr)
                 }
             } catch {
-                self.lock.withLock {
-                    self.mutableLastError = error
-                }
-                if traceEnabled {
+                let shouldLog = self.stopAfterSendFailure(error)
+                if traceEnabled, shouldLog {
                     fputs("microphone-pipeline trace: send failed error=\(error)\n", stderr)
                 }
             }
         }
     }
 
+    private func stopAfterSendFailure(_ error: any Error) -> Bool {
+        let shouldStop = lock.withLock {
+            mutableLastError = error
+            guard mutableIsRunning else {
+                return false
+            }
+            mutableIsRunning = false
+            return true
+        }
+        guard shouldStop else {
+            return false
+        }
+
+        source.stop()
+        source.setFrameSink(nil)
+        encoder.invalidate()
+        return true
+    }
+
     private func trace(_ message: String) {
         guard traceEnabled else {
             return

TestResults/device-harness.log

@@ -0,0 +1,37 @@
+16:06:40  Acquired tunnel connection to device.
+16:06:40  Enabling developer disk image services.
+16:06:40  Acquired usage assertion.
+Launched application with com.example.LiveKitNativeDeviceHarness bundle identifier.
+Waiting for the application to terminate…
+app_install: installed bundle_id=com.example.LiveKitNativeDeviceHarness execution_environment=physical-device
+bundle_identifier: com.example.LiveKitNativeDeviceHarness
+execution_environment: physical-device
+device_model: iPhone
+system_version: iOS 26.5
+scripted_validation: autostart requested
+camera_permission=true microphone_permission=true
+publisher-rtp trace: clear all audioSIDs=[] videoSIDs=[]
+av_audio_session_route_change_notification: reason=categoryChange
+publisher-rtp trace: clear all audioSIDs=[] videoSIDs=[]
+connect_publish_failed: Could not connect to the server.
+av_audio_session_route_change_notification: reason=categoryChange
+scripted_lifecycle: suspend_resume pid=46428
+16:07:01  Acquired tunnel connection to device.
+16:07:01  Enabling developer disk image services.
+16:07:01  Acquired usage assertion.
+Signal to suspend process sent to pid 46428
+scripted_lifecycle: suspended pid=46428
+reason: background
+16:07:04  Acquired tunnel connection to device.
+16:07:04  Enabling developer disk image services.
+16:07:04  Acquired usage assertion.
+Sent signal to resume process sent to pid 46428
+scripted_lifecycle: resumed pid=46428
+reason: foreground
+reason: background-foreground-recovery
+scripted_validation: failed missing=diagnostics
+background_foreground_recovery: reconnecting
+publisher-rtp trace: clear all audioSIDs=[] videoSIDs=[]
+av_audio_session_route_change_notification: reason=categoryChange
+scripted_validation_wait: failed
+App terminated due to signal 2.

scripts/check_device_harness_shape.sh

@@ -162,6 +162,7 @@ require_text "$repo_root/scripts/run_device_validation_on_device.sh" "devicectl
 require_text "$repo_root/scripts/run_device_validation_on_device.sh" "DEVICECTL_CHILD_LIVEKIT_NATIVE_DEVICE_HARNESS_AUTOSTART=1"
 require_text "$repo_root/scripts/run_device_validation_on_device.sh" "DEVICECTL_CHILD_LIVEKIT_NATIVE_DEVICE_HARNESS_ALLOW_DIRECT_ICE"
 require_text "$repo_root/scripts/run_device_validation_on_device.sh" "DEVICECTL_CHILD_LIVEKIT_NATIVE_DEVICE_HARNESS_SUBSCRIBER_TOKEN"
+require_text "$repo_root/scripts/run_device_validation_on_device.sh" "DEVICECTL_CHILD_LIVEKIT_NATIVE_AUDIO_TRACE"
 require_text "$repo_root/scripts/run_device_validation_on_device.sh" "scripted_validation: completed"
 require_text "$repo_root/scripts/run_device_validation_on_device.sh" "scripted_validation_wait:"
 require_text "$repo_root/scripts/run_device_validation_on_device.sh" "run_device_validation.sh"