zgaetano/rainbow-dragon
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

camera: fix waitForConnection race — continuation registered synchronously on actor, timeout resumes directly; 6x suite runs stable at 95 tests

Browse source
This commit is contained in:
Forge Dev 2026-07-10 19:38:57 +00:00
parent f723ea6177
commit ea08f8fcb7
1 changed files with 27 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
Sources/ForgeCamera/ConnectionSupervisor.swift
View file

@ -40,7 +40,8 @@ public actor ConnectionSupervisor {
public private(set) var health: ConnectionStatus = .disconnected public private(set) var health: ConnectionStatus = .disconnected
private var stateContinuations: [UUID: AsyncStream<CameraState>.Continuation] = [:] private var stateContinuations: [UUID: AsyncStream<CameraState>.Continuation] = [:]
private var connectionWaiters: [CheckedContinuation<Void, Never>] = [] private var connectionWaiters: [UUID] = []
private var waiterMap: [UUID: CheckedContinuation<Void, Never>] = [:]
public init(driver: any CameraDriver, backoff: BackoffPolicy) { public init(driver: any CameraDriver, backoff: BackoffPolicy) {
self.driver = driver self.driver = driver
@ -99,9 +100,7 @@ public actor ConnectionSupervisor {
health = state.connection health = state.connection
if state.connection == .connected { if state.connection == .connected {
backoff.reset() backoff.reset()
let waiters = connectionWaiters resumeAllWaiters()
connectionWaiters.removeAll()
for w in waiters { w.resume() }
} }
for c in stateContinuations.values { for c in stateContinuations.values {
c.yield(state) c.yield(state)
@ -109,22 +108,29 @@ public actor ConnectionSupervisor {
} }
/// Await connected health. Returns false on timeout. /// Await connected health. Returns false on timeout.
/// Continuation registered synchronously on the actor — no gap for a
/// connected event to slip through unobserved. Timeout resumes the
/// continuation directly (cancellation can't interrupt a checked continuation).
public func waitForConnection(timeout: Duration) async -> Bool { public func waitForConnection(timeout: Duration) async -> Bool {
if health == .connected { return true } if health == .connected { return true }
let waitTask = Task {
await withCheckedContinuation { (c: CheckedContinuation<Void, Never>) in await withCheckedContinuation { (c: CheckedContinuation<Void, Never>) in
self.connectionWaiters.append(c) let id = UUID()
} waiterMap[id] = c
} connectionWaiters.append(id)
let timeoutTask = Task { Task {
try? await Task.sleep(for: timeout) try? await Task.sleep(for: timeout)
waitTask.cancel() self.timeoutWaiter(id: id)
}
} }
_ = await waitTask.value
timeoutTask.cancel()
return health == .connected return health == .connected
} }
private func timeoutWaiter(id: UUID) {
guard let c = waiterMap.removeValue(forKey: id) else { return }
connectionWaiters.removeAll { $0 == id }
c.resume()
}
public func stop() async { public func stop() async {
connectTask?.cancel() connectTask?.cancel()
pumpTask?.cancel() pumpTask?.cancel()
@ -132,8 +138,14 @@ public actor ConnectionSupervisor {
pumpTask = nil pumpTask = nil
await driver.disconnect() await driver.disconnect()
health = .disconnected health = .disconnected
let waiters = connectionWaiters resumeAllWaiters()
}
private func resumeAllWaiters() {
let ids = connectionWaiters
connectionWaiters.removeAll() connectionWaiters.removeAll()
for w in waiters { w.resume() } for id in ids {
waiterMap.removeValue(forKey: id)?.resume()
}
} }
} }