Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -248,7 +248,7 @@ extension ReconnectRouteSelectionTests {
let first = try #require(fixture.box.get())
await first.close()

let recoveredWithoutServer = try await pollUntil(attempts: 100) {
let recoveredWithoutServer = try await pollUntil {
guard let replacement = fixture.store.remoteClient else { return false }
return replacement !== firstClient && fixture.store.connectionState == .connected
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1090,7 +1090,7 @@ extension ReconnectRouteSelectionTests {
clock.advance(by: 61)
store.resumeForegroundRefresh()

let recovered = try await pollUntil(attempts: 100) {
let recovered = try await pollUntil {
guard let current = box.get() else { return false }
return current !== firstTransport
&& store.connectionState == .connected
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
import Foundation
import Testing

extension LivenessHostRouter {
@discardableResult
func waitForCount(
of method: String,
atLeast expectedCount: Int,
timeoutNanoseconds: UInt64 = MobileShellWallClockWaitPolicy.defaultWaitTimeoutNanoseconds,
recordIssueOnTimeout: Bool = true
) async -> Bool {
let effectiveTimeoutNanoseconds = MobileShellWallClockWaitPolicy
.timeoutNanoseconds(for: timeoutNanoseconds)
let operation: @Sendable () async throws -> Bool = {
try Task.checkCancellation()
return await withTaskGroup(of: Bool.self) { group in
group.addTask {
await self.waitUntilCountReached(of: method, atLeast: expectedCount)
return true
}
group.addTask {
// Test assertion deadline only; request arrival is signaled by record().
try? await Task.sleep(nanoseconds: effectiveTimeoutNanoseconds)
return false
}
let reached = await group.next() ?? false
group.cancelAll()
return reached
}
}
let reached: Bool
if MobileShellWallClockWaitPolicy.shouldSerializeWait(
timeoutNanoseconds: timeoutNanoseconds
) {
reached = (try? await MobileShellWallClockWaitGate.processWide.withLock(operation)) ?? false
} else {
reached = (try? await operation()) ?? false
}
if !reached, recordIssueOnTimeout {
Issue.record("timed out waiting for \(method) count >= \(expectedCount)")
}
return reached
}

/// Waits for the transport's real replay-request admission signal. This
/// is used by tests that need to distinguish an already-started request
/// from one that must wait for an output acknowledgement.
@discardableResult
func waitForReplayRequestStart(
after existingCount: Int,
timeoutNanoseconds: UInt64 = 250_000_000
) async -> Bool {
await waitForCount(
of: "mobile.terminal.replay",
atLeast: existingCount + 1,
timeoutNanoseconds: timeoutNanoseconds,
recordIssueOnTimeout: false
)
}

func waitUntilCountReached(of method: String, atLeast expectedCount: Int) async {
guard count(of: method) < expectedCount else { return }
let waiterID = UUID()
await withTaskCancellationHandler {
await withCheckedContinuation { continuation in
countWaiters.append((
id: waiterID,
method: method,
expectedCount: expectedCount,
continuation: continuation
))
resumeSatisfiedCountWaiters()
}
} onCancel: {
Task { await self.cancelCountWaiter(id: waiterID) }
}
}

func resumeSatisfiedCountWaiters() {
var remaining: [(
id: UUID,
method: String,
expectedCount: Int,
continuation: CheckedContinuation<Void, Never>
)] = []
var satisfied: [CheckedContinuation<Void, Never>] = []
for waiter in countWaiters {
if count(of: waiter.method) >= waiter.expectedCount {
satisfied.append(waiter.continuation)
} else {
remaining.append(waiter)
}
}
countWaiters = remaining
for continuation in satisfied {
continuation.resume()
}
}

func cancelCountWaiter(id: UUID) {
guard let index = countWaiters.firstIndex(where: { $0.id == id }) else {
return
}
let waiter = countWaiters.remove(at: index)
waiter.continuation.resume()
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
/// Poll until `condition` is true, bounded at `attempts` x 10ms. The default
/// budget is intentionally generous for whole-suite scheduling; callers that
/// assert bounded absence can keep an explicit shorter attempt count. Returns
/// the final value so tests can assert both presence and (bounded) absence.
@MainActor
func pollUntil(
attempts: Int = MobileShellWallClockWaitPolicy.defaultPollAttempts,
_ condition: @escaping @MainActor @Sendable () async -> Bool
) async throws -> Bool {
let operation: @Sendable () async throws -> Bool = {
try Task.checkCancellation()
for _ in 0..<attempts {
if await condition() {
return true
}
try await Task.sleep(nanoseconds: 10_000_000)
}
return await condition()
}
if MobileShellWallClockWaitPolicy.shouldSerializePoll(attempts: attempts) {
return try await MobileShellWallClockWaitGate.processWide.withLock(operation)
}
return try await operation()
}
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ actor LivenessHostRouter {

private var recorded: [RecordedRequest] = []
private var attachTicketFailuresRemaining = 0
private var countWaiters: [(
var countWaiters: [(
id: UUID,
method: String,
expectedCount: Int,
Expand Down Expand Up @@ -171,96 +171,6 @@ actor LivenessHostRouter {
replayResponseCount
}

@discardableResult
func waitForCount(
of method: String,
atLeast expectedCount: Int,
timeoutNanoseconds: UInt64 = 3_000_000_000,
recordIssueOnTimeout: Bool = true
) async -> Bool {
let reached = await withTaskGroup(of: Bool.self) { group in
group.addTask {
await self.waitUntilCountReached(of: method, atLeast: expectedCount)
return true
}
group.addTask {
// Test assertion deadline only; request arrival is signaled by record().
try? await Task.sleep(nanoseconds: timeoutNanoseconds)
return false
}
let reached = await group.next() ?? false
group.cancelAll()
return reached
}
if !reached, recordIssueOnTimeout {
Issue.record("timed out waiting for \(method) count >= \(expectedCount)")
}
return reached
}

/// Waits for the transport's real replay-request admission signal. This
/// is used by tests that need to distinguish an already-started request
/// from one that must wait for an output acknowledgement.
@discardableResult
func waitForReplayRequestStart(
after existingCount: Int,
timeoutNanoseconds: UInt64 = 250_000_000
) async -> Bool {
await waitForCount(
of: "mobile.terminal.replay",
atLeast: existingCount + 1,
timeoutNanoseconds: timeoutNanoseconds,
recordIssueOnTimeout: false
)
}

private func waitUntilCountReached(of method: String, atLeast expectedCount: Int) async {
guard count(of: method) < expectedCount else { return }
let waiterID = UUID()
await withTaskCancellationHandler {
await withCheckedContinuation { continuation in
countWaiters.append((
id: waiterID,
method: method,
expectedCount: expectedCount,
continuation: continuation
))
resumeSatisfiedCountWaiters()
}
} onCancel: {
Task { await self.cancelCountWaiter(id: waiterID) }
}
}

private func resumeSatisfiedCountWaiters() {
var remaining: [(
id: UUID,
method: String,
expectedCount: Int,
continuation: CheckedContinuation<Void, Never>
)] = []
var satisfied: [CheckedContinuation<Void, Never>] = []
for waiter in countWaiters {
if count(of: waiter.method) >= waiter.expectedCount {
satisfied.append(waiter.continuation)
} else {
remaining.append(waiter)
}
}
countWaiters = remaining
for continuation in satisfied {
continuation.resume()
}
}

private func cancelCountWaiter(id: UUID) {
guard let index = countWaiters.firstIndex(where: { $0.id == id }) else {
return
}
let waiter = countWaiters.remove(at: index)
waiter.continuation.resume()
}

func topics(for method: String) -> [[String]] {
recorded.compactMap { request in
guard request.method == method else { return nil }
Expand Down Expand Up @@ -973,22 +883,6 @@ func attachURL(for ticket: CmxAttachTicket) throws -> String {
return "cmux-ios://attach?v=\(ticket.version)&payload=\(payload)"
}

/// Poll until `condition` is true, bounded at `attempts` x 10ms. Returns the
/// final value so tests can assert both presence and (bounded) absence.
@MainActor
func pollUntil(
attempts: Int = 300,
_ condition: @MainActor () async -> Bool
) async throws -> Bool {
for _ in 0..<attempts {
if await condition() {
return true
}
try await Task.sleep(nanoseconds: 10_000_000)
}
return await condition()
}

@MainActor
func waitForReplayResponsesServed(
_ expectedCount: Int,
Expand Down
Loading