From 9f7c367b4aea136e1d6844049d548c454f76faec Mon Sep 17 00:00:00 2001 From: supfeer Date: Thu, 23 Jul 2026 19:16:51 +0300 Subject: [PATCH] Fix audio route hotkey stalls --- Sources/Fluid/Services/ASRService.swift | 80 ++++++++++++++----- .../Services/AudioEngineRetirementDrain.swift | 49 ++++++++++-- .../AudioEngineRetirementDrainTests.swift | 42 +++++++++- 3 files changed, 141 insertions(+), 30 deletions(-) diff --git a/Sources/Fluid/Services/ASRService.swift b/Sources/Fluid/Services/ASRService.swift index 8c3855c1..6ae0e2c9 100644 --- a/Sources/Fluid/Services/ASRService.swift +++ b/Sources/Fluid/Services/ASRService.swift @@ -665,14 +665,50 @@ final class ASRService: ObservableObject { self.audioEngineRetirementDrain.schedule(token) } - /// Route recovery and engine retry paths use this completion barrier so the - /// old AVAudioEngine and its AVAudioIOUnit are fully deallocated before a - /// replacement can touch Core Audio. - private func retireAudioEngineAndWait(reason: String) async { + private struct AudioEngineRetirementTimeout: LocalizedError { + let reason: String + + var errorDescription: String? { + "The previous audio engine did not stop in time (\(self.reason))." + } + } + + private static let audioEngineRetirementTimeout: TimeInterval = 3 + + /// Route recovery and engine retry paths use this bounded completion barrier + /// so a wedged AVAudioIOUnit cannot block every future recording attempt. + private func retireAudioEngineAndWait(reason: String) async throws { + let completed: Bool if let token = self.detachAudioEngineForRetirement(reason: reason) { - await self.audioEngineRetirementDrain.releaseAndWait(token) + completed = await self.audioEngineRetirementDrain.releaseAndWait( + token, + timeout: Self.audioEngineRetirementTimeout + ) } else { - await self.audioEngineRetirementDrain.waitForScheduledReleases() + completed = await self.audioEngineRetirementDrain.waitForScheduledReleases( + timeout: Self.audioEngineRetirementTimeout + ) + } + + guard completed else { + DebugLogger.shared.error( + "Audio engine retirement timed out (\(reason))", + source: "ASRService" + ) + throw AudioEngineRetirementTimeout(reason: reason) + } + } + + private func waitForAudioEngineRetirement(reason: String) async throws { + let completed = await self.audioEngineRetirementDrain.waitForScheduledReleases( + timeout: Self.audioEngineRetirementTimeout + ) + guard completed else { + DebugLogger.shared.error( + "Audio engine retirement barrier timed out (\(reason))", + source: "ASRService" + ) + throw AudioEngineRetirementTimeout(reason: reason) } } @@ -809,11 +845,6 @@ final class ASRService: ObservableObject { } private func startPreferredAudioCapture() async throws { - // A non-route path may have scheduled a fire-and-forget retirement. Do - // not let capture startup overlap any final AVAudioEngine release that is - // already queued. - await self.audioEngineRetirementDrain.waitForScheduledReleases() - let directCaptureEnabled = SettingsStore.shared.experimentalDirectAudioCaptureEnabled if directCaptureEnabled, self.prepareDirectAudioInputIfPossible(reason: "recording_start"), @@ -852,7 +883,7 @@ final class ASRService: ObservableObject { } private func startCompatibilityAudioCapture(reason: String) async throws { - await self.audioEngineRetirementDrain.waitForScheduledReleases() + try await self.waitForAudioEngineRetirement(reason: "compatibility_start:\(reason)") self.benchmarkLog("audio_backend kind=av_audio_engine_fallback reason=\(reason)") try self.configureSession() try await self.startEngine() @@ -1424,7 +1455,7 @@ final class ASRService: ObservableObject { self.audioCapturePipeline.setRecordingEnabled(false) self.isRunning = false self.stopActiveAudioCapture() - await self.retireAudioEngineAndWait(reason: "start_failed") + try? await self.retireAudioEngineAndWait(reason: "start_failed") DebugLogger.shared.error("Failed to start ASR session: \(error)", source: "ASRService") // Resume media if we paused it before the failure @@ -1892,7 +1923,7 @@ final class ASRService: ObservableObject { DebugLogger.shared.debug("Audio capture stopped", source: "ASRService") // Cancel/no-transcription paths stay conservative and retire the engine. - await self.retireAudioEngineAndWait(reason: "stop_without_transcription") + try? await self.retireAudioEngineAndWait(reason: "stop_without_transcription") // CRITICAL FIX: Await completion of streaming task AND any pending transcriptions // This prevents use-after-free crashes (EXC_BAD_ACCESS) when clearing buffer @@ -2278,7 +2309,7 @@ final class ASRService: ObservableObject { // If this isn't the last attempt, recreate engine and reconfigure if attempts < 3 { DebugLogger.shared.debug("⚠️ Start failed, recreating engine for retry...", source: "ASRService") - await self.retireAudioEngineAndWait(reason: "start_retry") + try await self.retireAudioEngineAndWait(reason: "start_retry") // Need to reconfigure the new engine try? self.configureSession() DebugLogger.shared.debug("✅ Engine recreated and reconfigured, will retry", source: "ASRService") @@ -2442,7 +2473,7 @@ final class ASRService: ObservableObject { _ = await task?.result self.audioRouteRecoveryTask = nil self.isRecoveringAudioRoute = false - await self.audioEngineRetirementDrain.waitForScheduledReleases() + try? await self.waitForAudioEngineRetirement(reason: "cancel_route_recovery") } private func performAudioRouteRecovery(_ request: AudioRouteRecoveryRequest) async { @@ -2488,7 +2519,15 @@ final class ASRService: ObservableObject { reason: request.reason, requiresIdlePrewarm: true ) - await self.retireAudioEngineAndWait(reason: "idle_route_change:\(request.reason)") + do { + try await self.retireAudioEngineAndWait(reason: "idle_route_change:\(request.reason)") + } catch { + DebugLogger.shared.error( + "Idle audio route recovery stopped: \(error.localizedDescription)", + source: "ASRService" + ) + return + } guard request.generation == self.audioRouteRecoveryGeneration, Task.isCancelled == false else { return } self.prewarmAudioEngineIfPossible( @@ -2507,11 +2546,10 @@ final class ASRService: ObservableObject { self.audioCapturePipeline.setRecordingEnabled(false) self.stopMonitoringDevice() self.stopActiveAudioCapture() - await self.retireAudioEngineAndWait(reason: "audio_route_recovery") - - guard request.generation == self.audioRouteRecoveryGeneration, Task.isCancelled == false else { return } - do { + try await self.retireAudioEngineAndWait(reason: "audio_route_recovery") + guard request.generation == self.audioRouteRecoveryGeneration, Task.isCancelled == false else { return } + self.audioCapturePipeline.setRecordingEnabled( true, sessionID: self.benchmarkSessionID, diff --git a/Sources/Fluid/Services/AudioEngineRetirementDrain.swift b/Sources/Fluid/Services/AudioEngineRetirementDrain.swift index c29af6c0..c2cb780e 100644 --- a/Sources/Fluid/Services/AudioEngineRetirementDrain.swift +++ b/Sources/Fluid/Services/AudioEngineRetirementDrain.swift @@ -1,5 +1,25 @@ import Foundation +private final nonisolated class AudioEngineRetirementWait: @unchecked Sendable { + private let lock = NSLock() + private var continuation: CheckedContinuation? + + init(_ continuation: CheckedContinuation) { + self.continuation = continuation + } + + func finish(completed: Bool) { + self.lock.lock() + guard let continuation = self.continuation else { + self.lock.unlock() + return + } + self.continuation = nil + self.lock.unlock() + continuation.resume(returning: completed) + } +} + /// Owns the last strong reference to an audio engine while it is waiting to be /// released on the dedicated retirement queue. /// @@ -36,8 +56,12 @@ final nonisolated class AudioEngineRetirementDrain: @unchecked Sendable { } } - func releaseAndWait(_ token: AudioEngineRetirementToken) async { - await self.enqueueAndWait { + @discardableResult + func releaseAndWait( + _ token: AudioEngineRetirementToken, + timeout: TimeInterval + ) async -> Bool { + await self.enqueueAndWait(timeout: timeout) { token.releaseEngine() } } @@ -45,15 +69,26 @@ final nonisolated class AudioEngineRetirementDrain: @unchecked Sendable { /// Waits for every release already submitted to this drain. This is used at /// capture start so a fire-and-forget retirement from a non-route path cannot /// overlap construction of the next AVAudioEngine. - func waitForScheduledReleases() async { - await self.enqueueAndWait {} + @discardableResult + func waitForScheduledReleases(timeout: TimeInterval) async -> Bool { + await self.enqueueAndWait(timeout: timeout) {} } - private func enqueueAndWait(_ operation: @escaping @Sendable () -> Void) async { - await withCheckedContinuation { (continuation: CheckedContinuation) in + private func enqueueAndWait( + timeout: TimeInterval, + operation: @escaping @Sendable () -> Void + ) async -> Bool { + precondition(timeout > 0, "Audio engine retirement timeout must be positive") + + return await withCheckedContinuation { (continuation: CheckedContinuation) in + let wait = AudioEngineRetirementWait(continuation) self.queue.async { operation() - continuation.resume() + wait.finish(completed: true) + } + + DispatchQueue.global(qos: .utility).asyncAfter(deadline: .now() + timeout) { + wait.finish(completed: false) } } } diff --git a/Tests/FluidDictationIntegrationTests/AudioEngineRetirementDrainTests.swift b/Tests/FluidDictationIntegrationTests/AudioEngineRetirementDrainTests.swift index 57ff1596..cea62f12 100644 --- a/Tests/FluidDictationIntegrationTests/AudioEngineRetirementDrainTests.swift +++ b/Tests/FluidDictationIntegrationTests/AudioEngineRetirementDrainTests.swift @@ -10,7 +10,7 @@ final class AudioEngineRetirementDrainTests: XCTestCase { let token = AudioEngineRetirementToken(try XCTUnwrap(probe)) probe = nil - await drain.releaseAndWait(token) + await drain.releaseAndWait(token, timeout: 1) XCTAssertEqual( recorder.events, @@ -29,7 +29,7 @@ final class AudioEngineRetirementDrainTests: XCTestCase { second = nil drain.schedule(firstToken) - await drain.releaseAndWait(secondToken) + await drain.releaseAndWait(secondToken, timeout: 1) XCTAssertEqual( recorder.events, @@ -39,6 +39,44 @@ final class AudioEngineRetirementDrainTests: XCTestCase { ] ) } + + func testWaitForScheduledReleasesTimesOutWhenEngineDeinitIsBlocked() async throws { + let drain = AudioEngineRetirementDrain(label: "test.audio-engine-retirement.timeout") + let releaseStarted = DispatchSemaphore(value: 0) + let allowRelease = DispatchSemaphore(value: 0) + var probe: BlockingDeinitProbe? = BlockingDeinitProbe( + releaseStarted: releaseStarted, + allowRelease: allowRelease + ) + let token = AudioEngineRetirementToken(try XCTUnwrap(probe)) + probe = nil + + drain.schedule(token) + XCTAssertEqual(releaseStarted.wait(timeout: .now() + 1), .success) + + let completedBeforeTimeout = await drain.waitForScheduledReleases(timeout: 0.01) + XCTAssertFalse(completedBeforeTimeout) + + allowRelease.signal() + let completedAfterRelease = await drain.waitForScheduledReleases(timeout: 1) + XCTAssertTrue(completedAfterRelease) + } + +} + +private final class BlockingDeinitProbe { + private let releaseStarted: DispatchSemaphore + private let allowRelease: DispatchSemaphore + + init(releaseStarted: DispatchSemaphore, allowRelease: DispatchSemaphore) { + self.releaseStarted = releaseStarted + self.allowRelease = allowRelease + } + + deinit { + self.releaseStarted.signal() + self.allowRelease.wait() + } } private final class DeinitProbe {