From f1c98fed6ff8d48c900c3028a76323f22f1c1ba5 Mon Sep 17 00:00:00 2001 From: Iva Horn Date: Wed, 30 Sep 2026 17:52:00 +0200 Subject: [PATCH] Fixed a crash when URLSession reports a WebSocket pong more than once URLSessionWebSocketTask.sendPing(pongReceiveHandler:) calls its handler more than once, even concurrently, when a connection is torn down between ping and pong (Apple r. 129686846), and sendPing() resumed its CheckedContinuation on every call, which terminated Cirruscope 1.0.0 and 1.1.0 in 21 reported crashes. PendingPing now delivers exactly one outcome and lets the waiting task's cancellation compete for it, since the handler is also known never to be called. PushNotificationsConnection closes its channel as soon as a session ends or its task is cancelled, handshake included, because neither URLSession's receive() nor its pong handler observes cancellation and a stopped stream otherwise kept its socket open. A ping without a pong within ten seconds now closes the connection so the coordinator reconnects. Signed-off-by: Iva Horn Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 5 +- Sources/Rainmaker/Push/PendingPing.swift | 179 +++++++++++++++++ Sources/Rainmaker/Push/PendingPingState.swift | 29 +++ .../Push/PushNotificationsConnection.swift | 118 +++++++---- .../Push/ServerEventCoordinator.swift | 14 +- .../Push/URLSessionWebSocketChannel.swift | 18 +- .../Rainmaker/Requests/WebSocketChannel.swift | 7 +- Tests/RainmakerTests/Eventually.swift | 30 +++ Tests/RainmakerTests/LockedValue.swift | 68 +++++++ Tests/RainmakerTests/MockPingBehavior.swift | 22 ++ .../RainmakerTests/MockWebSocketChannel.swift | 77 ++++++- .../RainmakerTests/MockWebSocketClosure.swift | 76 +++++++ Tests/RainmakerTests/PendingPingTests.swift | 188 ++++++++++++++++++ Tests/RainmakerTests/ServerEventsTests.swift | 92 ++++++++- 14 files changed, 859 insertions(+), 64 deletions(-) create mode 100644 Sources/Rainmaker/Push/PendingPing.swift create mode 100644 Sources/Rainmaker/Push/PendingPingState.swift create mode 100644 Tests/RainmakerTests/Eventually.swift create mode 100644 Tests/RainmakerTests/LockedValue.swift create mode 100644 Tests/RainmakerTests/MockPingBehavior.swift create mode 100644 Tests/RainmakerTests/MockWebSocketClosure.swift create mode 100644 Tests/RainmakerTests/PendingPingTests.swift diff --git a/AGENTS.md b/AGENTS.md index 2301699..665b8c8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -10,7 +10,7 @@ The human readable introduction of this Swift Package is in [README.md](README.m - `Sources/Rainmaker/Extensions/` is for implementations of extensions of first-party or platform types. One source code file per extended type and added feature. - `Sources/Rainmaker/Models` contains Swift source code for data models which are also publicly available types. They do not necessarily mirror the structure and types as returned by the server in responses. They are meant to be as elegant and plausible as possible from a Swift client developer perspective, not necessarily mirroring the server responses exactly. - `Sources/Rainmaker/Responses/Models` contains Swift source code for data models which actually enable the use of Swift's `Decodable` for server response data. They are not meant to be exposed outside the Swift package module but only as an intermediate representation to simplify deserialization. -- `Sources/Rainmaker/Push/` contains the `notify_push` WebSocket transport and the coordinator behind `Server.events(_:)`, which prefers the WebSocket when the server advertises the capability and falls back to polling otherwise. The mockable WebSocket abstraction protocols (`WebSocketConnecting`, `WebSocketChannel`, `WebSocketFrame`) live alongside `Requesting` in `Sources/Rainmaker/Requests/`, mirroring how the HTTP session is abstracted. +- `Sources/Rainmaker/Push/` contains the `notify_push` WebSocket transport and the coordinator behind `Server.events(_:)`, which prefers the WebSocket when the server advertises the capability and falls back to polling otherwise. The mockable WebSocket abstraction protocols (`WebSocketConnecting`, `WebSocketChannel`, `WebSocketFrame`) live alongside `Requesting` in `Sources/Rainmaker/Requests/`, mirroring how the HTTP session is abstracted. `PendingPing` (with its `PendingPingState`) is what `URLSessionWebSocketChannel.sendPing()` waits through: it resumes the waiting task exactly once, because `URLSessionWebSocketTask.sendPing(pongReceiveHandler:)` has been seen calling its handler more than once for one ping, concurrently, when a connection is torn down, and also never calling it at all, which is why the waiting task's cancellation competes for the same outcome. Neither URLSession's `receive()` nor its pong handler observes task cancellation, so `PushNotificationsConnection` ends a session by closing its channel rather than by cancelling its tasks, and bounds every ping with a pong timeout. - `Sources/RainmakerCLI/` contains the Swift source code for the accompanying command line utility which enables the usage of the library in a terminal environment without any additional upstream project. - `Sources/RainmakerCLI/Commands/RecordFixtures.swift` and `Sources/RainmakerCLI/Fixtures/` implement the `record-fixtures` subcommand which automates the creation of test fixtures. It is macOS-only: it deploys ephemeral Nextcloud containers via the `NextcloudContainerManager` package, runs the test suite against them in recording mode to capture real responses into `Tests/RainmakerTests/Responses/`, and verifies the captures replay without a server. Containers are deployed with the apps a suite's fixtures depend on but which are not part of a Nextcloud installation, currently the notes app and the Talk app, which makes a recording run depend on the app store being reachable. All of its Docker-facing code is guarded with `#if os(macOS)` so the simulator builds compile the CLI without it. - `Sources/RainmakerTestServerTags/` is a small internal module holding `ServerVersion`, the single source of truth for the supported Nextcloud versions. Both the test target and the CLI's fixture recorder depend on it, so the version list is declared once. @@ -18,7 +18,8 @@ The human readable introduction of this Swift Package is in [README.md](README.m - `Tests/` contains the automated tests per target. - `Tests/RainmakerTests/Responses/` contains static test fixtures which are the HTTP response bodies of actual server responses. They either are in JSON or XML format. - `Tests/RainmakerTests/URLTestSession.swift` replays those fixtures during normal test runs, while `Tests/RainmakerTests/URLRecordingSession.swift` is its recording counterpart used by the `record-fixtures` subcommand. Both derive fixture paths through the shared `Tests/RainmakerTests/FixtureLocator.swift` so recording and replay can never diverge, and `Tests/RainmakerTests/FixtureCanonicalizer.swift` normalizes recorded responses (canonical host, redacted volatile fields) so fixtures stay stable. `Tests/RainmakerTests/ServerTesting.swift` selects between the two sessions based on the `RAINMAKER_FIXTURE_RECORD` environment variable. -- The `Server.events(_:)` tests do not use fixtures: they drive hand-authored WebSocket and request doubles (`Tests/RainmakerTests/MockWebSocketConnecting.swift`, `MockWebSocketChannel.swift`, `MockRequesting.swift`) so the WebSocket handshake, frame mapping, polling fallback, and reconnection are exercised deterministically without a server. +- The `Server.events(_:)` tests do not use fixtures: they drive hand-authored WebSocket and request doubles (`Tests/RainmakerTests/MockWebSocketConnecting.swift`, `MockWebSocketChannel.swift`, `MockRequesting.swift`) so the WebSocket handshake, frame mapping, polling fallback, reconnection, and teardown are exercised deterministically without a server. `MockPingBehavior` scripts whether a mock channel answers pings, and `MockWebSocketClosure` records that a channel was closed and lets its receiving ignore task cancellation until then, as `URLSessionWebSocketTask` does, so the teardown is tested against the behaviour it actually has to cope with. +- `Tests/RainmakerTests/PendingPingTests.swift` drives `PendingPing` directly, a `URLSessionWebSocketTask` being impossible to substitute: it stores the pong handler a ping hands out and calls it the way the framework has been seen to, later, repeatedly, concurrently, or never. `Eventually.swift` is the polling wait those tests and the teardown tests use instead of `WithTimeout.swift`, which cannot bound an operation that ignores cancellation, and `LockedValue.swift` is the lock-protected box they share state with `@Sendable` closures through. - `Rainmaker.png` is a static artwork file for presentation on the web and can be ignored. ## Code Style diff --git a/Sources/Rainmaker/Push/PendingPing.swift b/Sources/Rainmaker/Push/PendingPing.swift new file mode 100644 index 0000000..73fec7d --- /dev/null +++ b/Sources/Rainmaker/Push/PendingPing.swift @@ -0,0 +1,179 @@ +// SPDX-FileCopyrightText: 2026 Iva Horn +// SPDX-License-Identifier: MIT + +import Foundation +import os + +/// +/// One ping sent by ``URLSessionWebSocketChannel/sendPing()`` which resumes the task waiting for it exactly once. +/// +/// `URLSessionWebSocketTask.sendPing(pongReceiveHandler:)` calls its handler more than once for a single ping, even concurrently from different queues, when the connection is torn down between ping and pong, which Apple acknowledged as a bug of its own. +/// Resuming a `CheckedContinuation` a second time terminates the process, so the handler's outcomes are funnelled through the ``PendingPingState`` held here and only the first one reaches the waiting task. +/// The handler is also known never to be called at all, which is why the waiting task's cancellation competes for the same single outcome: that is what lets ``PushNotificationsConnection`` end a session whose pong never arrives. +/// +/// It is `@unchecked Sendable` because ``state`` is only read and written while ``lock`` is held; `Mutex` and `OSAllocatedUnfairLock` would express that to the compiler but are not available at every deployment target of this package. +/// +final class PendingPing: @unchecked Sendable { + /// + /// Serializes every access to ``state``, which the pong handler, the waiting task and its cancellation handler reach from different threads. + /// + private let lock = NSLock() + + /// + /// Where the ping stands, only ever accessed while ``lock`` is held. + /// + private var state = PendingPingState.idle + + /// + /// The logger recording the outcomes which arrive after the one that resumed the waiting task. + /// + private let logger: Logger + + /// + /// Prepare a ping which is not sent before ``wait(function:sending:)`` is called. + /// + /// - Parameters: + /// - logger: The logger to record dropped outcomes through, which should be the one of the channel sending the ping. + /// + init(logger: Logger) { + self.logger = logger + } + + /// + /// Send the ping and suspend until its first outcome arrives or the calling task is cancelled, whichever happens first. + /// + /// - Parameters: + /// - function: The name a continuation misuse or leak message reports, which defaults to the caller's so such a message names the method sending the ping rather than this one. + /// - send: Sends the ping and hands the given pong handler to the framework; it is not called at all when the calling task is already cancelled. + /// + /// - Throws: The first error the pong handler reports, or `CancellationError` when the calling task is cancelled first. + /// + func wait(function: String = #function, sending send: (@escaping @Sendable ((any Error)?) -> Void) -> Void) async throws { + try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation(function: function) { (continuation: CheckedContinuation) in + guard install(continuation) else { + return + } + + send { error in + self.resolve(error) + } + } + } onCancel: { + self.cancel() + } + } + + /// + /// Deliver one outcome of the pong handler, of which only the first resumes the waiting task. + /// + /// - Parameters: + /// - error: The error the pong handler reported, or `nil` when the pong arrived. + /// + /// - Returns: Whether this outcome resumed the waiting task. + /// + @discardableResult + func resolve(_ error: (any Error)?) -> Bool { + let previous = lock.withLock { + let previous = state + + if case .waiting = previous { + state = .resolved + } + + return previous + } + + switch previous { + case let .waiting(continuation): + if let error { + continuation.resume(throwing: error) + } else { + continuation.resume() + } + + return true + + case .cancelled: + // Expected: closing the channel of a cancelled session makes the framework report the ping it still had outstanding. + logger.debug("Dropped the outcome of a cancelled ping: \(Self.describe(error), privacy: .public)") + return false + + case .resolved: + logger.notice("Dropped a repeated pong handler call for a ping which already resumed its task: \(Self.describe(error), privacy: .public)") + return false + + case .idle: + // Unreachable: the pong handler is only handed out once the continuation is installed. + logger.error("Dropped a pong handler call for a ping which was never sent: \(Self.describe(error), privacy: .public)") + return false + } + } + + /// + /// Install the continuation of the waiting task unless the task was cancelled first. + /// + /// - Parameters: + /// - continuation: The continuation to resume with the first outcome. + /// + /// - Returns: Whether the continuation was installed and the ping should be sent; when `false`, the continuation was already resumed with `CancellationError`. + /// + private func install(_ continuation: CheckedContinuation) -> Bool { + let installed = lock.withLock { + guard case .idle = state else { + return false + } + + state = .waiting(continuation) + return true + } + + if installed == false { + logger.debug("Did not send a ping for a task cancelled before it could") + continuation.resume(throwing: CancellationError()) + } + + return installed + } + + /// + /// Resume the waiting task with `CancellationError` unless an outcome already did, and drop every outcome arriving afterwards. + /// + private func cancel() { + let previous = lock.withLock { + let previous = state + + switch previous { + case .idle, .waiting: + state = .cancelled + + case .cancelled, .resolved: + break + } + + return previous + } + + if case let .waiting(continuation) = previous { + logger.debug("Cancelled a ping still waiting for its pong") + continuation.resume(throwing: CancellationError()) + } + } + + /// + /// Describe an outcome of the pong handler without any user data, so it can be logged publicly. + /// + /// - Parameters: + /// - error: The error the pong handler reported, or `nil` when the pong arrived. + /// + /// - Returns: `"pong received"`, or the error's domain, code and description. + /// + private static func describe(_ error: (any Error)?) -> String { + guard let error else { + return "pong received" + } + + let nsError = error as NSError + return "\(nsError.domain) \(nsError.code): \(nsError.localizedDescription)" + } +} diff --git a/Sources/Rainmaker/Push/PendingPingState.swift b/Sources/Rainmaker/Push/PendingPingState.swift new file mode 100644 index 0000000..6d24711 --- /dev/null +++ b/Sources/Rainmaker/Push/PendingPingState.swift @@ -0,0 +1,29 @@ +// SPDX-FileCopyrightText: 2026 Iva Horn +// SPDX-License-Identifier: MIT + +/// +/// Where one ``PendingPing`` stands between being sent and delivering its single outcome to the task waiting for it. +/// +/// Every case but ``idle`` and ``waiting(_:)`` is final for the ping, which is what lets ``PendingPing`` resume the waiting task exactly once no matter how often and in which order the pong handler and the task's cancellation report. +/// +enum PendingPingState { + /// + /// The waiting task has not installed its continuation yet, so the ping has not been sent either. + /// + case idle + + /// + /// The ping was sent and its continuation is waiting for the first outcome. + /// + case waiting(CheckedContinuation) + + /// + /// The waiting task was cancelled before any outcome arrived, so its continuation is resumed with `CancellationError`, either when the cancellation finds it waiting or when it is installed afterwards. + /// + case cancelled + + /// + /// The pong handler delivered its first outcome to the waiting task, so any later one is dropped. + /// + case resolved +} diff --git a/Sources/Rainmaker/Push/PushNotificationsConnection.swift b/Sources/Rainmaker/Push/PushNotificationsConnection.swift index 52ba146..25d78b5 100644 --- a/Sources/Rainmaker/Push/PushNotificationsConnection.swift +++ b/Sources/Rainmaker/Push/PushNotificationsConnection.swift @@ -7,7 +7,7 @@ import os /// /// One lifetime of a `notify_push` WebSocket connection: it authenticates, then maps incoming frames to ``ServerEvent`` values yielded into a stream until the connection ends. /// -/// It is a value type carrying only immutable, `Sendable` inputs and therefore needs no isolation of its own; ``ServerEventCoordinator`` owns the reconnection state and decides when to open a new connection. +/// It is a value type carrying only `Sendable` inputs it never mutates and therefore needs no isolation of its own; ``ServerEventCoordinator`` owns the reconnection state and decides when to open a new connection. /// struct PushNotificationsConnection { /// @@ -25,11 +25,6 @@ struct PushNotificationsConnection { case disconnected(wasAuthenticated: Bool) } - /// - /// How long to wait between liveness pings, matching the server's own 30 second ping interval. - /// - private static let pingInterval: TimeInterval = 30 - /// /// The connector vending the underlying channel. /// @@ -65,6 +60,20 @@ struct PushNotificationsConnection { /// let logger: Logger + /// + /// How long to wait between liveness pings, matching the server's own 30 second ping interval by default. + /// + /// This is configurable so tests can exercise the ping loop without waiting for it. + /// + var pingInterval: TimeInterval = 30 + + /// + /// How long a liveness ping may wait for its pong before the connection is considered dead and closed, so ``ServerEventCoordinator`` reconnects. + /// + /// The framework is known to sometimes never report a ping's outcome at all, which without this limit would silently end liveness checking for the rest of the connection. + /// + var pongTimeout: TimeInterval = 10 + /// /// Open the connection, authenticate, and pump events into `continuation` until the connection ends. /// @@ -78,31 +87,39 @@ struct PushNotificationsConnection { let channel = webSocket.channel(for: request) channel.resume() - var authenticated = false - - do { - try await channel.send(user) - try await channel.send(password) - try await authenticate(on: channel) - authenticated = true - logger.debug("notify_push connection authenticated") - - if listenFileIDs { - try await channel.send("listen notify_file_id") + let logger = logger + + // Neither URLSession's send() nor its receive() observes task cancellation, so a cancelled subscription has to close the channel itself, which is what ends whichever of them is outstanding, during authentication as much as afterwards. + try await withTaskCancellationHandler { + var authenticated = false + + do { + try await channel.send(user) + try await channel.send(password) + try await authenticate(on: channel) + authenticated = true + logger.debug("notify_push connection authenticated") + + if listenFileIDs { + try await channel.send("listen notify_file_id") + } + + continuation.yield(.connected) + try await pump(channel: channel, into: continuation) + + // The pump only returns by throwing; reaching here means the read loop ended without error, which is still a disconnect. + channel.cancel() + throw Failure.disconnected(wasAuthenticated: true) + } catch let failure as Failure { + channel.cancel() + throw failure + } catch { + channel.cancel() + throw Failure.disconnected(wasAuthenticated: authenticated) } - - continuation.yield(.connected) - try await pump(channel: channel, into: continuation) - - // The pump only returns by throwing; reaching here means the read loop ended without error, which is still a disconnect. - channel.cancel() - throw Failure.disconnected(wasAuthenticated: true) - } catch let failure as Failure { + } onCancel: { + logger.debug("Closing notify_push connection of a cancelled subscription") channel.cancel() - throw failure - } catch { - channel.cancel() - throw Failure.disconnected(wasAuthenticated: authenticated) } } @@ -134,24 +151,30 @@ struct PushNotificationsConnection { } /// - /// Run the read loop and the liveness ping loop concurrently until either ends, which ends the connection. + /// Run the read loop and the liveness ping loop concurrently until either ends, which ends the connection and closes the channel. /// private func pump(channel: any WebSocketChannel, into continuation: AsyncThrowingStream.Continuation) async throws { try await withThrowingTaskGroup(of: Void.self) { group in let subjects = subjects let logger = logger + let pingInterval = pingInterval + let pongTimeout = pongTimeout group.addTask { try await Self.read(from: channel, subjects: subjects, into: continuation, logger: logger) } group.addTask { - try await Self.ping(on: channel) + try await Self.ping(on: channel, interval: pingInterval, timeout: pongTimeout) + } + + // The first task to finish or throw ends the session and its outcome is surfaced. Closing the channel right away, rather than once the group has drained, is what ends the other task, because neither URLSession's receive() nor its pong handler observes task cancellation. Cancelling first lets the ping the channel still has outstanding be told apart from a repeated report of one that already ended. + defer { + group.cancelAll() + channel.cancel() } - // The first task to finish or throw ends the session; cancel the other and surface the outcome. try await group.next() - group.cancelAll() } } @@ -175,12 +198,33 @@ struct PushNotificationsConnection { } /// - /// Send a liveness ping on the given interval, throwing when a pong fails to arrive. + /// Send a liveness ping on the given interval, throwing when the channel reports a pong as failed or none arrives within the given timeout. /// - private static func ping(on channel: any WebSocketChannel) async throws { + private static func ping(on channel: any WebSocketChannel, interval: TimeInterval, timeout: TimeInterval) async throws { while true { - try await Task.sleep(nanoseconds: UInt64(pingInterval * 1_000_000_000)) - try await channel.sendPing() + try await Task.sleep(nanoseconds: UInt64(interval * 1_000_000_000)) + + try await withThrowingTaskGroup(of: Void.self) { group in + group.addTask { + try await channel.sendPing() + } + + group.addTask { + // Cancellation must end this sleep with an error rather than fall through, or a ping answered in time would close the channel below. + try await Task.sleep(nanoseconds: UInt64(timeout * 1_000_000_000)) + + // Closing the channel is what ends a ping which does not observe cancellation, since this group waits for it before it can report the timeout. + channel.cancel() + throw URLError(.timedOut) + } + + // Whichever task ends first decides the ping, and the other one is cancelled. + defer { + group.cancelAll() + } + + try await group.next() + } } } diff --git a/Sources/Rainmaker/Push/ServerEventCoordinator.swift b/Sources/Rainmaker/Push/ServerEventCoordinator.swift index 32f6ba9..5b2ad8b 100644 --- a/Sources/Rainmaker/Push/ServerEventCoordinator.swift +++ b/Sources/Rainmaker/Push/ServerEventCoordinator.swift @@ -52,6 +52,18 @@ struct ServerEventCoordinator { /// var initialBackoff: TimeInterval = 1 + /// + /// How long each WebSocket connection waits between liveness pings, matching the server's own 30 second ping interval by default. + /// + /// This is configurable so tests can exercise the ping loop without waiting for it. + /// + var pingInterval: TimeInterval = 30 + + /// + /// How long a liveness ping may wait for its pong before the connection is considered dead and reconnected. + /// + var pongTimeout: TimeInterval = 10 + /// /// Run the subscription until the consumer stops it or an unrecoverable authentication failure occurs. /// @@ -136,7 +148,7 @@ struct ServerEventCoordinator { var request = URLRequest(url: endpoint) request.setValue(server.userAgent, forHTTPHeaderField: "User-Agent") - let connection = PushNotificationsConnection(webSocket: server.webSocket, request: request, user: user, password: password, subjects: pushedSubjects, listenFileIDs: options.listenFileIDs, logger: logger) + let connection = PushNotificationsConnection(webSocket: server.webSocket, request: request, user: user, password: password, subjects: pushedSubjects, listenFileIDs: options.listenFileIDs, logger: logger, pingInterval: pingInterval, pongTimeout: pongTimeout) let backstopInterval = options.backstopPollInterval let pollInterval = options.pollInterval diff --git a/Sources/Rainmaker/Push/URLSessionWebSocketChannel.swift b/Sources/Rainmaker/Push/URLSessionWebSocketChannel.swift index 1b8027e..f0d5c3f 100644 --- a/Sources/Rainmaker/Push/URLSessionWebSocketChannel.swift +++ b/Sources/Rainmaker/Push/URLSessionWebSocketChannel.swift @@ -2,18 +2,26 @@ // SPDX-License-Identifier: MIT import Foundation +import os /// /// The production ``WebSocketChannel`` backed by a `URLSessionWebSocketTask`. /// /// It is `@unchecked Sendable` because it wraps a `URLSessionWebSocketTask`, whose send, receive, ping and cancel operations are safe to call from concurrent tasks, mirroring the reasoning behind ``Server``'s `nonisolated(unsafe)` file manager. /// +/// The one callback it bridges itself, the pong handler, goes through a ``PendingPing``, because the framework may call that handler more than once for a single ping or not at all. +/// final class URLSessionWebSocketChannel: WebSocketChannel, @unchecked Sendable { /// /// The underlying task performing the connection. /// private let task: URLSessionWebSocketTask + /// + /// The logger recording what ``PendingPing`` drops of the pong handler's outcomes. + /// + private let logger = Logger(category: "URLSessionWebSocketChannel") + /// /// Wrap the given task. /// @@ -46,14 +54,8 @@ final class URLSessionWebSocketChannel: WebSocketChannel, @unchecked Sendable { } func sendPing() async throws { - try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in - task.sendPing { error in - if let error { - continuation.resume(throwing: error) - } else { - continuation.resume() - } - } + try await PendingPing(logger: logger).wait { handler in + task.sendPing(pongReceiveHandler: handler) } } diff --git a/Sources/Rainmaker/Requests/WebSocketChannel.swift b/Sources/Rainmaker/Requests/WebSocketChannel.swift index f9b687d..070ca32 100644 --- a/Sources/Rainmaker/Requests/WebSocketChannel.swift +++ b/Sources/Rainmaker/Requests/WebSocketChannel.swift @@ -34,12 +34,15 @@ public protocol WebSocketChannel: Sendable { /// /// Send a ping and wait for its pong. /// - /// - Throws: When the pong is not received, which the transport treats as a dead connection. + /// - Throws: When the pong is not received, which the transport treats as a dead connection, or `CancellationError` when the calling task is cancelled first. /// func sendPing() async throws /// - /// Close the channel. + /// Close the channel, which ends any outstanding ``receive()``, ``send(_:)`` and ``sendPing()`` with an error. + /// + /// The transport relies on this to end a session and to end a ping whose pong does not arrive in time, because a conformer's receiving, sending and pinging need not observe task cancellation. + /// It is called more than once per connection, possibly concurrently, so it must be safe to repeat. /// func cancel() } diff --git a/Tests/RainmakerTests/Eventually.swift b/Tests/RainmakerTests/Eventually.swift new file mode 100644 index 0000000..2914d02 --- /dev/null +++ b/Tests/RainmakerTests/Eventually.swift @@ -0,0 +1,30 @@ +// SPDX-FileCopyrightText: 2026 Iva Horn +// SPDX-License-Identifier: MIT + +import Foundation + +/// +/// Poll a condition until it holds, giving up once the deadline passes. +/// +/// Unlike ``withTimeout(seconds:_:)`` this also bounds a wait for something that does not observe task cancellation, such as a continuation which is never resumed: a task group waits for all of its children, so racing such an operation against a timer hangs instead of failing, while polling never waits on the operation at all. +/// The deadline is a guard against a hang, not an assertion about speed, and therefore generous on purpose. +/// +/// - Parameters: +/// - seconds: How long to keep polling before giving up. +/// - condition: The condition to wait for, evaluated on the calling task. +/// +/// - Returns: Whether the condition held before the deadline passed. +/// +func eventually(within seconds: Double = 30, _ condition: () async -> Bool) async throws -> Bool { + let deadline = Date().addingTimeInterval(seconds) + + while await condition() == false { + guard Date() < deadline else { + return false + } + + try await Task.sleep(nanoseconds: 2_000_000) + } + + return true +} diff --git a/Tests/RainmakerTests/LockedValue.swift b/Tests/RainmakerTests/LockedValue.swift new file mode 100644 index 0000000..75a917c --- /dev/null +++ b/Tests/RainmakerTests/LockedValue.swift @@ -0,0 +1,68 @@ +// SPDX-FileCopyrightText: 2026 Iva Horn +// SPDX-License-Identifier: MIT + +import Foundation + +/// +/// A value shared between a test and the closures it hands to the code under test, which may call them from any thread. +/// +/// It is `@unchecked Sendable` because ``value`` is only accessed while ``lock`` is held, which is what makes it safe to capture into a `@Sendable` closure the way a mutable local variable cannot be. +/// +final class LockedValue: @unchecked Sendable { + /// + /// Serializes every access to ``value``. + /// + private let lock = NSLock() + + /// + /// The shared value, only ever accessed while ``lock`` is held. + /// + private var value: Value + + /// + /// Share the given initial value. + /// + /// - Parameters: + /// - value: The initial value. + /// + init(_ value: Value) { + self.value = value + } + + /// + /// Read the current value. + /// + /// - Returns: A copy of the current value. + /// + func get() -> Value { + lock.withLock { + value + } + } + + /// + /// Replace the current value. + /// + /// - Parameters: + /// - newValue: The value to store. + /// + func set(_ newValue: Value) { + lock.withLock { + value = newValue + } + } + + /// + /// Read and change the value in one step no other thread can interleave with. + /// + /// - Parameters: + /// - body: Changes the value in place and returns a result. + /// + /// - Returns: What `body` returned. + /// + func withValue(_ body: (inout Value) -> Result) -> Result { + lock.withLock { + body(&value) + } + } +} diff --git a/Tests/RainmakerTests/MockPingBehavior.swift b/Tests/RainmakerTests/MockPingBehavior.swift new file mode 100644 index 0000000..df3cb6a --- /dev/null +++ b/Tests/RainmakerTests/MockPingBehavior.swift @@ -0,0 +1,22 @@ +// SPDX-FileCopyrightText: 2026 Iva Horn +// SPDX-License-Identifier: MIT + +/// +/// How a ``MockWebSocketChannel`` answers ``WebSocketChannel/sendPing()``. +/// +enum MockPingBehavior { + /// + /// Every ping is answered immediately. + /// + case pong + + /// + /// No ping is ever answered: the ping waits until the calling task is cancelled or the channel is closed, which is how the production channel behaves when the framework never reports a ping's outcome. + /// + case never + + /// + /// No ping is ever answered and the ping ignores task cancellation too, ending only when the channel is closed, as a conformer bridging the framework's pong handler without observing cancellation would. + /// + case neverIgnoringCancellation +} diff --git a/Tests/RainmakerTests/MockWebSocketChannel.swift b/Tests/RainmakerTests/MockWebSocketChannel.swift index ed556c1..bf30686 100644 --- a/Tests/RainmakerTests/MockWebSocketChannel.swift +++ b/Tests/RainmakerTests/MockWebSocketChannel.swift @@ -5,9 +5,10 @@ import Foundation @testable import Rainmaker /// -/// A scripted ``WebSocketChannel`` for tests: it replays a fixed sequence of frames, records what was sent, and then either closes or waits until cancelled. +/// A scripted ``WebSocketChannel`` for tests: it replays a fixed sequence of frames, records what was sent, answers pings as scripted, and then either closes or waits until cancelled or closed. /// -/// It is an actor so its mutable state is serialized safely; the synchronous ``resume()`` and ``cancel()`` protocol requirements are satisfied by `nonisolated` no-ops. +/// It is an actor so its mutable state is serialized safely; the synchronous ``resume()`` protocol requirement is satisfied by a `nonisolated` no-op, and ``cancel()`` records the close in ``closure``, which ends any wait for it. +/// By default its waits observe task cancellation, which `URLSessionWebSocketTask` does not; ``ignoresTaskCancellation`` makes ``receive()`` behave like the latter, so the transport's teardown is exercised against the framework it actually runs on. /// actor MockWebSocketChannel: WebSocketChannel { /// @@ -16,10 +17,30 @@ actor MockWebSocketChannel: WebSocketChannel { private let frames: [WebSocketFrame] /// - /// Whether ``receive()`` throws ``MockWebSocketError/closed`` once the frames are exhausted, simulating a drop; when `false` it instead waits until the task is cancelled. + /// Whether ``receive()`` throws ``MockWebSocketError/closed`` once the frames are exhausted, simulating a drop; when `false` it instead waits until the task is cancelled or the channel is closed, or only until the latter when ``ignoresTaskCancellation`` is set. /// private let closesWhenExhausted: Bool + /// + /// Whether ``receive()`` ignores task cancellation once the frames are exhausted and ends only when the channel is closed, as `URLSessionWebSocketTask` does. + /// + private let ignoresTaskCancellation: Bool + + /// + /// How pings are answered. + /// + private let pingBehavior: MockPingBehavior + + /// + /// Records whether ``cancel()`` was called and ends the waits for that, which the transport relies on to end a session. + /// + nonisolated let closure = MockWebSocketClosure() + + /// + /// How many pings were sent so far. + /// + private var pings = 0 + /// /// The index of the next frame to replay. /// @@ -35,16 +56,22 @@ actor MockWebSocketChannel: WebSocketChannel { /// /// - Parameters: /// - frames: The frames to replay in order. - /// - closesWhenExhausted: Whether to report a drop once the frames run out. Defaults to `false`, which keeps the connection open until cancelled. + /// - closesWhenExhausted: Whether to report a drop once the frames run out. Defaults to `false`, which keeps the connection open until cancelled or closed. + /// - ignoresTaskCancellation: Whether keeping the connection open ignores task cancellation and ends only when the channel is closed, as `URLSessionWebSocketTask` does. Defaults to `false`. + /// - pingBehavior: How pings are answered. Defaults to answering every ping. /// - init(frames: [WebSocketFrame], closesWhenExhausted: Bool = false) { + init(frames: [WebSocketFrame], closesWhenExhausted: Bool = false, ignoresTaskCancellation: Bool = false, pingBehavior: MockPingBehavior = .pong) { self.frames = frames self.closesWhenExhausted = closesWhenExhausted + self.ignoresTaskCancellation = ignoresTaskCancellation + self.pingBehavior = pingBehavior } nonisolated func resume() {} - nonisolated func cancel() {} + nonisolated func cancel() { + closure.close() + } func send(_ text: String) async throws { sent.append(text) @@ -61,15 +88,47 @@ actor MockWebSocketChannel: WebSocketChannel { throw MockWebSocketError.closed } - // Keep the connection "open" until the consuming task is cancelled, then report a close. - while Task.isCancelled == false { + if ignoresTaskCancellation { + // Keep the connection "open" until the channel is closed, whether the consuming task is cancelled or not. + await closure.wait() + throw MockWebSocketError.closed + } + + // Keep the connection "open" until the consuming task is cancelled or the channel is closed, then report a close. + while Task.isCancelled == false, closure.isClosed == false { try? await Task.sleep(nanoseconds: 20_000_000) } throw MockWebSocketError.closed } - func sendPing() async throws {} + func sendPing() async throws { + pings += 1 + + switch pingBehavior { + case .pong: + return + + case .never: + // Observes cancellation through the sleep, as the production channel's ping does. + while closure.isClosed == false { + try await Task.sleep(nanoseconds: 5_000_000) + } + + throw MockWebSocketError.closed + + case .neverIgnoringCancellation: + await closure.wait() + throw MockWebSocketError.closed + } + } + + /// + /// How many pings the transport has sent so far. + /// + func pingCount() -> Int { + pings + } /// /// The text frames the transport has sent so far. diff --git a/Tests/RainmakerTests/MockWebSocketClosure.swift b/Tests/RainmakerTests/MockWebSocketClosure.swift new file mode 100644 index 0000000..ed77ec7 --- /dev/null +++ b/Tests/RainmakerTests/MockWebSocketClosure.swift @@ -0,0 +1,76 @@ +// SPDX-FileCopyrightText: 2026 Iva Horn +// SPDX-License-Identifier: MIT + +import Foundation + +/// +/// Whether a ``MockWebSocketChannel`` was closed, and a wait for that which, like `URLSessionWebSocketTask`, ignores task cancellation. +/// +/// It is not an `AsyncStream`, because iterating one does observe cancellation, which is exactly what the production transport cannot rely on. +/// It is `@unchecked Sendable` because ``closed`` and ``waiters`` are only accessed while ``lock`` is held. +/// +final class MockWebSocketClosure: @unchecked Sendable { + /// + /// Serializes every access to ``closed`` and ``waiters``. + /// + private let lock = NSLock() + + /// + /// Whether ``close()`` was called. + /// + private var closed = false + + /// + /// The continuations of the ``wait()`` calls still suspended, resumed by ``close()``. + /// + private var waiters: [CheckedContinuation] = [] + + /// + /// Whether the channel was closed. + /// + var isClosed: Bool { + lock.withLock { + closed + } + } + + /// + /// Record the channel as closed and end every ``wait()``, of which only the first call has any effect. + /// + func close() { + let waiting = lock.withLock { + guard closed == false else { + return [CheckedContinuation]() + } + + closed = true + let waiting = waiters + waiters = [] + return waiting + } + + for waiter in waiting { + waiter.resume() + } + } + + /// + /// Suspend until the channel is closed, regardless of whether the calling task is cancelled. + /// + func wait() async { + await withCheckedContinuation { (continuation: CheckedContinuation) in + let alreadyClosed = lock.withLock { + if closed { + return true + } + + waiters.append(continuation) + return false + } + + if alreadyClosed { + continuation.resume() + } + } + } +} diff --git a/Tests/RainmakerTests/PendingPingTests.swift b/Tests/RainmakerTests/PendingPingTests.swift new file mode 100644 index 0000000..445e3be --- /dev/null +++ b/Tests/RainmakerTests/PendingPingTests.swift @@ -0,0 +1,188 @@ +// SPDX-FileCopyrightText: 2026 Iva Horn +// SPDX-License-Identifier: MIT + +import Foundation +import os +@testable import Rainmaker +import Testing + +/// +/// About how ``PendingPing`` turns the pong handler of `URLSessionWebSocketTask.sendPing(pongReceiveHandler:)` into exactly one resumption of the task waiting for it. +/// +/// A `URLSessionWebSocketTask` cannot be substituted, so these tests stand in for it the way it behaves: the handler the ping hands out is stored, then called later, from other queues, after the waiting task has resumed, more than once, concurrently, or never. +/// Each case waits on a real `CheckedContinuation`, so a regression terminates the test process exactly as it terminated the apps shipping this library, rather than failing an assertion. +/// Every wait polls through ``eventually(within:_:)``, since a regression which ignores cancellation would otherwise hang the suite instead of failing it. +/// +@Suite("Pending Ping") struct PendingPingTests { + /// + /// The logger the pings under test record dropped outcomes through. + /// + private let logger = Logger(subsystem: "RainmakerTests", category: "PendingPing") + + /// + /// The pong handler the ping under test handed out, standing in for the `URLSessionWebSocketTask` which would keep it until reporting. + /// + private typealias PongHandler = @Sendable ((any Error)?) -> Void + + /// + /// Start a task waiting on the given ping, recording the handler it sends and the outcome it ends with. + /// + /// - Parameters: + /// - ping: The ping to wait on. + /// - handler: Receives the pong handler once the ping is sent. + /// - outcome: Receives how the wait ended. + /// - cancelledFirst: Whether the task cancels itself before it starts waiting. + /// + /// - Returns: The waiting task. + /// + private func startWaiting(on ping: PendingPing, handler: LockedValue, outcome: LockedValue?>, cancelledFirst: Bool = false) -> Task { + Task { + if cancelledFirst { + withUnsafeCurrentTask { task in + task?.cancel() + } + } + + do { + try await ping.wait { pongHandler in + handler.set(pongHandler) + } + + outcome.set(.success(())) + } catch { + outcome.set(.failure(error)) + } + } + } + + /// + /// Call the given handler on a queue of its own and wait until it returned. + /// + private func report(_ error: (any Error)?, to handler: @escaping PongHandler, label: String) async { + await withCheckedContinuation { (continuation: CheckedContinuation) in + DispatchQueue(label: label).async { + handler(error) + continuation.resume() + } + } + } + + @Test("Delivers Only The First Of Two Outcomes", arguments: [(nil, nil), (nil, .networkConnectionLost), (.networkConnectionLost, nil), (.networkConnectionLost, .cancelled)] as [(URLError.Code?, URLError.Code?)]) + func deliversOnlyTheFirstOfTwoOutcomes(first: URLError.Code?, second: URLError.Code?) async throws { + let ping = PendingPing(logger: logger) + let handler = LockedValue(nil) + let outcome = LockedValue?>(nil) + let task = startWaiting(on: ping, handler: handler, outcome: outcome) + + try #require(try await eventually { handler.get() != nil }) + let pong = try #require(handler.get()) + + await report(first.map { URLError($0) }, to: pong, label: "first") + try #require(try await eventually { outcome.get() != nil }) + await task.value + + // The framework reports again although the waiting task has long resumed, which is what terminated the process before. + await report(second.map { URLError($0) }, to: pong, label: "second") + + switch outcome.get() { + case .success: + #expect(first == nil) + + case let .failure(error): + #expect((error as? URLError)?.code == first) + + case nil: + Issue.record("The wait never ended.") + } + } + + @Test("Resumes Once When Many Queues Report At Once") + func resumesOnceWhenManyQueuesReportAtOnce() async throws { + let ping = PendingPing(logger: logger) + let handler = LockedValue(nil) + let outcome = LockedValue?>(nil) + let resumptions = LockedValue(0) + let task = startWaiting(on: ping, handler: handler, outcome: outcome) + + try #require(try await eventually { handler.get() != nil }) + + DispatchQueue.concurrentPerform(iterations: 256) { iteration in + if ping.resolve(iteration.isMultiple(of: 2) ? nil : URLError(.networkConnectionLost)) { + resumptions.withValue { $0 += 1 } + } + } + + try #require(try await eventually { outcome.get() != nil }) + await task.value + #expect(resumptions.get() == 1) + } + + @Test("Survives Cancellation Racing Concurrent Reports") + func survivesCancellationRacingConcurrentReports() async throws { + for _ in 0 ..< 300 { + let ping = PendingPing(logger: logger) + let handler = LockedValue(nil) + let outcome = LockedValue?>(nil) + let task = startWaiting(on: ping, handler: handler, outcome: outcome) + + try #require(try await eventually { handler.get() != nil }) + let pong = try #require(handler.get()) + + DispatchQueue.concurrentPerform(iterations: 8) { iteration in + if iteration == 0 { + task.cancel() + } else { + pong(iteration.isMultiple(of: 2) ? nil : URLError(.networkConnectionLost)) + } + } + + try #require(try await eventually { outcome.get() != nil }) + await task.value + } + } + + @Test("Does Not Send A Ping For A Cancelled Task") + func doesNotSendAPingForACancelledTask() async throws { + let ping = PendingPing(logger: logger) + let handler = LockedValue(nil) + let outcome = LockedValue?>(nil) + let task = startWaiting(on: ping, handler: handler, outcome: outcome, cancelledFirst: true) + + try #require(try await eventually { outcome.get() != nil }) + await task.value + #expect(handler.get() == nil) + + guard case let .failure(error) = outcome.get() else { + Issue.record("A cancelled task sent its ping and waited.") + return + } + + #expect(error is CancellationError) + } + + @Test("Cancellation Ends The Wait For A Pong That Never Arrives") + func cancellationEndsTheWaitForAPongThatNeverArrives() async throws { + let ping = PendingPing(logger: logger) + let handler = LockedValue(nil) + let outcome = LockedValue?>(nil) + let task = startWaiting(on: ping, handler: handler, outcome: outcome) + + try #require(try await eventually { handler.get() != nil }) + let pong = try #require(handler.get()) + + task.cancel() + try #require(try await eventually { outcome.get() != nil }) + await task.value + + // Closing the connection makes the framework report the ping it still had outstanding, which must be dropped. + await report(URLError(.cancelled), to: pong, label: "late") + #expect(ping.resolve(nil) == false) + + guard case let .failure(error) = outcome.get() else { + Issue.record("A cancelled ping reported success.") + return + } + + #expect(error is CancellationError) + } +} diff --git a/Tests/RainmakerTests/ServerEventsTests.swift b/Tests/RainmakerTests/ServerEventsTests.swift index 457b4d7..c99fe51 100644 --- a/Tests/RainmakerTests/ServerEventsTests.swift +++ b/Tests/RainmakerTests/ServerEventsTests.swift @@ -42,14 +42,14 @@ import Testing /// /// Build the event stream through a coordinator configured with tiny retry and backoff timing so the tests run quickly. /// - private func makeStream(server: Server, options: ServerEventOptions) -> AsyncThrowingStream { + private func makeStream(server: Server, options: ServerEventOptions, pingInterval: TimeInterval = 30, pongTimeout: TimeInterval = 10) -> AsyncThrowingStream { AsyncThrowingStream { continuation in guard server.user != nil, server.password != nil else { continuation.finish(throwing: RainmakerError.credentialsRequired) return } - let coordinator = ServerEventCoordinator(server: server, options: options, logger: Logger(subsystem: "RainmakerTests", category: "ServerEvents"), maximumAuthenticationAttempts: 2, authenticationRetryInterval: 0.02, rediscoverInterval: 0.3, backoffCeiling: 0.05, initialBackoff: 0.01) + let coordinator = ServerEventCoordinator(server: server, options: options, logger: Logger(subsystem: "RainmakerTests", category: "ServerEvents"), maximumAuthenticationAttempts: 2, authenticationRetryInterval: 0.02, rediscoverInterval: 0.3, backoffCeiling: 0.05, initialBackoff: 0.01, pingInterval: pingInterval, pongTimeout: pongTimeout) let task = Task { await coordinator.run(into: continuation) } @@ -60,9 +60,6 @@ import Testing } } - /// - /// Collect the first `count` events, guarded by a timeout so a stalled stream fails the test instead of hanging. - /// /// /// Collect the given number of events from a stream, failing the test rather than hanging it if they never arrive. /// @@ -129,6 +126,91 @@ import Testing #expect(events == [.connected, .notifications, .connected, .notifications]) } + @Test("Closes The Socket When The Consumer Stops") + func closesTheSocketWhenTheConsumerStops() async throws { + let channel = MockWebSocketChannel(frames: [.text("authenticated")], ignoresTaskCancellation: true) + let server = makeServer(session: MockRequesting(string: capabilities(pushing: ["notifications"])), webSocket: MockWebSocketConnecting(channels: [channel])) + let stream = makeStream(server: server, options: ServerEventOptions(subjects: [.notifications], pollInterval: 100, emitConnectedOnStart: false)) + let connected = LockedValue(false) + + let consumer = Task { + for try await event in stream where event == .connected { + connected.set(true) + } + } + + try #require(try await eventually { connected.get() }) + + // The channel keeps receiving through a stop like URLSession does, so only closing it ends the session. + consumer.cancel() + #expect(try await eventually { channel.closure.isClosed }) + } + + @Test("Closes The Socket When The Consumer Stops During Authentication") + func closesTheSocketWhenTheConsumerStopsDuringAuthentication() async throws { + let channel = MockWebSocketChannel(frames: [], ignoresTaskCancellation: true) + let server = makeServer(session: MockRequesting(string: capabilities(pushing: ["notifications"])), webSocket: MockWebSocketConnecting(channels: [channel])) + let stream = makeStream(server: server, options: ServerEventOptions(subjects: [.notifications], pollInterval: 100, emitConnectedOnStart: false)) + + let consumer = Task { + for try await _ in stream {} + } + + // The server never confirms authentication, so the session waits in its handshake when the consumer stops. + try #require(try await eventually { await channel.sentFrames() == ["admin", "admin"] }) + consumer.cancel() + #expect(try await eventually { channel.closure.isClosed }) + } + + @Test("Reconnects When A Pong Never Arrives") + func reconnectsWhenAPongNeverArrives() async throws { + let silent = MockWebSocketChannel(frames: [.text("authenticated")], ignoresTaskCancellation: true, pingBehavior: .never) + let answering = MockWebSocketChannel(frames: [.text("authenticated")]) + let server = makeServer(session: MockRequesting(string: capabilities(pushing: ["notifications"])), webSocket: MockWebSocketConnecting(channels: [silent, answering])) + let stream = makeStream(server: server, options: ServerEventOptions(subjects: [.notifications], pollInterval: 100, emitConnectedOnStart: false), pingInterval: 0.02, pongTimeout: 0.05) + + // The first socket stays open but never answers a ping, so the timeout closes it and the coordinator reconnects. + let events = try await firstEvents(2, from: stream) + #expect(events == [.connected, .connected]) + #expect(silent.closure.isClosed) + } + + @Test("Reconnects When A Pong Never Arrives Although The Ping Ignores Cancellation") + func reconnectsWhenAPongNeverArrivesAlthoughThePingIgnoresCancellation() async throws { + let silent = MockWebSocketChannel(frames: [.text("authenticated")], ignoresTaskCancellation: true, pingBehavior: .neverIgnoringCancellation) + let answering = MockWebSocketChannel(frames: [.text("authenticated")]) + let server = makeServer(session: MockRequesting(string: capabilities(pushing: ["notifications"])), webSocket: MockWebSocketConnecting(channels: [silent, answering])) + let stream = makeStream(server: server, options: ServerEventOptions(subjects: [.notifications], pollInterval: 100, emitConnectedOnStart: false), pingInterval: 0.02, pongTimeout: 0.05) + + // Only closing the channel ends a ping like this one, so that is what the timeout has to do. + let events = try await firstEvents(2, from: stream) + #expect(events == [.connected, .connected]) + #expect(silent.closure.isClosed) + } + + @Test("Keeps The Socket Open While Pongs Arrive") + func keepsTheSocketOpenWhilePongsArrive() async throws { + let channel = MockWebSocketChannel(frames: [.text("authenticated")]) + let server = makeServer(session: MockRequesting(string: capabilities(pushing: ["notifications"])), webSocket: MockWebSocketConnecting(channels: [channel])) + let stream = makeStream(server: server, options: ServerEventOptions(subjects: [.notifications], pollInterval: 100, emitConnectedOnStart: false), pingInterval: 0.01, pongTimeout: 5) + let events = LockedValue<[ServerEvent]>([]) + + let consumer = Task { + for try await event in stream { + events.withValue { $0.append(event) } + } + } + + defer { + consumer.cancel() + } + + // The timeout is far above any scheduling delay, so only a ping answered in time being treated as lost would end the session. + try #require(try await eventually { await channel.pingCount() >= 5 }) + #expect(channel.closure.isClosed == false) + #expect(events.get() == [.connected]) + } + @Test("Falls Back To Polling After Repeated Auth Rejection") func fallsBackToPollingAfterAuthRejection() async throws { let rejecting = { MockWebSocketChannel(frames: [.text("err: Invalid credentials")], closesWhenExhausted: true) }