diff --git a/.changes/websocket-network-framework b/.changes/websocket-network-framework new file mode 100644 index 000000000..68587c887 --- /dev/null +++ b/.changes/websocket-network-framework @@ -0,0 +1,2 @@ +patch type="fixed" "Stop leaking about 25 KB per signalling connection: the WebSocket runs over Network.framework, as Foundation keeps every URLSession that ran a WebSocket task alive after it is closed; a rejected upgrade still reports its HTTP status" +patch type="changed" "The signalling WebSocket fails on an HTTP redirect and does not answer HTTP authentication challenges. livekit-server sends neither on the signal upgrade" diff --git a/Sources/LiveKit/Support/Network/WebSocket.swift b/Sources/LiveKit/Support/Network/WebSocket.swift index 644a06165..949805450 100644 --- a/Sources/LiveKit/Support/Network/WebSocket.swift +++ b/Sources/LiveKit/Support/Network/WebSocket.swift @@ -23,165 +23,324 @@ struct WebSocketUpgradeFailure: Error, Sendable { let statusCode: Int } +/// The signalling WebSocket, over Network.framework. +/// +/// Not `URLSessionWebSocketTask`: Foundation keeps every `URLSession` that ran a WebSocket task, and the task, +/// alive after the task is cancelled and the session invalidated, whichever way it is closed (about 25 KB a +/// connection, measured on macOS 27 and the iOS 26.5 simulator). Every full or quick reconnect left one behind +/// for the life of the process. An `NWConnection` is freed once cancelled. actor WebSocket: Loggable, AsyncSequence { typealias Element = URLSessionWebSocketTask.Message + typealias EventHandler = @Sendable (Event) -> Void - private let delegate: Delegate - private let urlSession: URLSession - private let task: URLSessionWebSocketTask + let connection: NWConnection + enum Event: Sendable { + case receiveRegistered + case closeStarted + case teardown + case closeDeadline + } + + fileprivate enum Lifecycle: Sendable { case open, closing, closed } - private static func makeSessionConfiguration() -> URLSessionConfiguration { - let config = URLSessionConfiguration.default - config.timeoutIntervalForRequest = 60 - config.timeoutIntervalForResource = 604_800 - config.shouldUseExtendedBackgroundIdleMode = true - config.networkServiceType = .callSignaling + private let lifecycle = StateSync(Lifecycle.open) + private let onEvent: EventHandler? + /// Network.framework delivers on a queue it is given; a shared one, not one of our own: every callback here only + /// resumes an `AsyncCompleter` or a continuation. + private static var queue: DispatchQueue { .global(qos: .default) } + + static func makeParameters(secure: Bool, timeoutInterval: TimeInterval) -> NWParameters { + let tcp = NWProtocolTCP.Options() + tcp.connectionTimeout = Swift.max(1, Int(timeoutInterval.rounded(.up))) + // URLSession turned Nagle off; signal messages are small. + tcp.noDelay = true + let parameters = NWParameters(tls: secure ? NWProtocolTLS.Options() : nil, tcp: tcp) + // As `.callSignaling` on the URLSession configuration this replaces. + parameters.serviceClass = .signaling #if os(iOS) || os(visionOS) // https://developer.apple.com/documentation/foundation/urlsessionconfiguration/improving_network_reliability_using_multipath_tcp - config.multipathServiceType = .handover + parameters.multipathServiceType = .handover #endif - return config - } - - /// Every header entry is added, not set, so a repeated name keeps all of its values. - static func makeRequest(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval) -> URLRequest { - var request = URLRequest(url: url, - cachePolicy: .useProtocolCachePolicy, - timeoutInterval: timeoutInterval) - for (name, value) in headers { - request.addValue(value, forHTTPHeaderField: name) - } - return request + return parameters } - init(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval) async throws { - let request = Self.makeRequest(url: url, headers: headers, timeoutInterval: timeoutInterval) - + init(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval, + onEvent: EventHandler? = nil) async throws + { + self.onEvent = onEvent #if targetEnvironment(simulator) if #available(iOS 26.0, *) { nw_tls_create_options() } #endif - delegate = Delegate() - urlSession = URLSession(configuration: Self.makeSessionConfiguration(), - delegate: delegate, delegateQueue: nil) - task = urlSession.webSocketTask(with: request) + let headers = Self.upgradeHeaders(url: url, headers: headers) + let webSocket = NWProtocolWebSocket.Options() + webSocket.autoReplyPing = true + // URLSessionWebSocketTask's limit on a received message. + webSocket.maximumMessageSize = 1 << 20 + webSocket.setAdditionalHeaders(headers) + let parameters = Self.makeParameters(secure: url.scheme == "wss", timeoutInterval: timeoutInterval) + parameters.defaultProtocolStack.applicationProtocols.insert(webSocket, at: 0) + let connection = NWConnection(to: .url(url), using: parameters) + self.connection = connection - try await withTaskCancellationHandler { - try await withCheckedThrowingContinuation { continuation in - delegate.setConnectContinuation(continuation) - task.resume() + let opening = AsyncCompleter(label: "WebSocket.open", defaultTimeout: timeoutInterval) + connection.stateUpdateHandler = { state in + switch state { + case .ready: + opening.resume(returning: ()) + case let .failed(error), let .waiting(error): + // `.waiting` too: a connection that cannot be made now fails, as URLSession's did + // (waitsForConnectivity off); the signal client owns the retries. + opening.resume(with: .failure(error)) + case .cancelled: + opening.resume(with: .failure(LiveKitError(.cancelled))) + default: + break } - } onCancel: { - self.close() + } + connection.start(queue: Self.queue) + do { + try await opening.wait() + } catch let error as NWError { + Self.finish(connection, lifecycle, onEvent) + // The server answered the upgrade with an HTTP status: Network.framework aborts the connection + // without it, so the same upgrade, sent once more by hand, reads it. + if case .posix(.ECONNABORTED) = error, + let status = await Self.upgradeStatus(url: url, headers: headers, timeoutInterval: timeoutInterval) + { + throw LiveKitError(.network, internalError: WebSocketUpgradeFailure(statusCode: status)) + } + throw LiveKitError(.network, internalError: Self.bridged(error)) + } catch { + Self.finish(connection, lifecycle, onEvent) + throw error is LiveKitError ? error : LiveKitError.from(error: error) ?? LiveKitError(.unknown) } } deinit { - close() + Self.close(connection, lifecycle, onEvent) } + /// Sends a close frame with code 1000, as `URLSessionWebSocketTask.cancel(with: .normalClosure)` did, and cancels + /// once the server answers or after 250 ms (the JS SDK's wait), whichever comes first. Does not block. New reads + /// end immediately; a pending receive ends on whatever arrives next, at the latest on the server's reply or the + /// deadline. Only the first call does anything. nonisolated func close() { - task.cancel(with: .normalClosure, reason: nil) - urlSession.finishTasksAndInvalidate() - delegate.cancelConnection() + Self.close(connection, lifecycle, onEvent) + } + + fileprivate static func close(_ connection: NWConnection, _ lifecycle: StateSync, _ onEvent: EventHandler?) { + guard lifecycle.mutate({ + guard $0 == .open else { return false } + $0 = .closing + return true + }) else { return } + onEvent?(.closeStarted) + let metadata = NWProtocolWebSocket.Metadata(opcode: .close) + metadata.closeCode = .protocolCode(.normalClosure) + let context = NWConnection.ContentContext(identifier: "close", metadata: [metadata]) + connection.send(content: nil, contentContext: context, isComplete: true, completion: .contentProcessed { _ in + drain(connection, lifecycle, onEvent) + }) + queue.asyncAfter(deadline: .now() + .milliseconds(250)) { + finish(connection, lifecycle, onEvent) + onEvent?(.closeDeadline) + } + } + + /// Drops the messages still in flight and cancels once the close handshake is over. + private static func drain(_ connection: NWConnection, _ lifecycle: StateSync, _ onEvent: EventHandler?) { + guard lifecycle.copy() == .closing else { return } + connection.receiveMessage { _, context, _, error in + endsClose(context, error) ? finish(connection, lifecycle, onEvent) : drain(connection, lifecycle, onEvent) + } + } + + fileprivate static func finish(_ connection: NWConnection, _ lifecycle: StateSync, _ onEvent: EventHandler?) { + guard lifecycle.mutate({ + guard $0 != .closed else { return false } + $0 = .closed + return true + }) else { return } + onEvent?(.teardown) + cancel(connection) + } + + /// Whether a receive after our close frame ends the handshake: the server's close frame, an error, or the end of the + /// stream (no WebSocket metadata). + fileprivate static func endsClose(_ context: NWConnection.ContentContext?, _ error: NWError?) -> Bool { + let metadata = context?.protocolMetadata(definition: NWProtocolWebSocket.definition) as? NWProtocolWebSocket.Metadata + return error != nil || metadata.map { $0.opcode == .close } ?? true + } + + /// Below iOS 16 and macOS 13 a cancelled connection keeps its state handler, and whatever that captures. + fileprivate static func cancel(_ connection: NWConnection) { + connection.cancel() + connection.stateUpdateHandler = nil + } + + /// The upgrade headers as URLSession sent them. A repeated name is joined with commas, as `URLRequest.addValue` + /// did (`setAdditionalHeaders` keeps only the last value). The cookies stored for the URL go along, matched on + /// its `https` form so that secure ones count. `Host` carries the port and IPv6 brackets, which Network.framework's + /// own drops; it comes last, after the join, so it replaces the generated one and any the caller passed. + static func upgradeHeaders(url: URL, headers: [(name: String, value: String)]) -> [(name: String, value: String)] { + var components = URLComponents(url: url, resolvingAgainstBaseURL: false) + components?.scheme = url.scheme == "wss" ? "https" : "http" + let cookies = components?.url.flatMap { HTTPCookieStorage.shared.cookies(for: $0) } ?? [] + let stored = HTTPCookie.requestHeaderFields(with: cookies).map { (name: $0.key, value: $0.value) } + + var joined: [(name: String, value: String)] = [] + for (name, value) in headers + stored where name.caseInsensitiveCompare("Host") != .orderedSame { + if let index = joined.firstIndex(where: { $0.name.caseInsensitiveCompare(name) == .orderedSame }) { + joined[index].value += (name.caseInsensitiveCompare("Cookie") == .orderedSame ? "; " : ",") + value + } else { + joined.append((name, value)) + } + } + guard let host = url.host else { return joined } + let port = url.port.map { ":\($0)" } ?? "" + return joined + [(name: "Host", value: (host.contains(":") ? "[\(host)]" : host) + port)] + } + + /// The HTTP status the server answers the upgrade with, read off the status line of the same request sent over + /// a plain TCP (and TLS) connection; nil when it gives none or switches protocols after all. `headers` are the + /// ``upgradeHeaders(url:headers:)``, `Host` included. + static func upgradeStatus(url: URL, headers: [(name: String, value: String)], timeoutInterval: TimeInterval) async -> Int? { + guard let host = url.host else { return nil } + let secure = url.scheme == "wss" + let port = url.port ?? (secure ? 443 : 80) + // The request target as the upgrade sent it: escaped, as `URL.path` would not give it. + guard let components = URLComponents(url: url, resolvingAgainstBaseURL: false) else { return nil } + var target = components.percentEncodedPath.isEmpty ? "/" : components.percentEncodedPath + if let query = components.percentEncodedQuery { target += "?\(query)" } + let key = Data((0 ..< 16).map { _ in UInt8.random(in: 0 ... 255) }).base64EncodedString() + var request = "GET \(target) HTTP/1.1\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" + + "Sec-WebSocket-Key: \(key)\r\nSec-WebSocket-Version: 13\r\n" + for (name, value) in headers { + request += "\(name): \(value)\r\n" + } + request += "\r\n" + let bytes = Data(request.utf8) + + guard let nwPort = NWEndpoint.Port(rawValue: UInt16(clamping: port)) else { return nil } + let connection = NWConnection(host: NWEndpoint.Host(host), port: nwPort, + using: makeParameters(secure: secure, timeoutInterval: timeoutInterval)) + defer { cancel(connection) } + let answer = AsyncCompleter(label: "WebSocket.upgradeStatus", defaultTimeout: timeoutInterval) + connection.stateUpdateHandler = { state in + switch state { + case .ready: + connection.send(content: bytes, completion: .contentProcessed { _ in }) + connection.receive(minimumIncompleteLength: 12, maximumLength: 512) { content, _, _, _ in + // "HTTP/1.1 401 Unauthorized" + let line = content.flatMap { String(data: $0, encoding: .utf8) }?.split(separator: "\r\n").first + let status = line.flatMap { $0.split(separator: " ").dropFirst().first }.flatMap { Int($0) } + answer.resume(returning: status == 101 ? nil : status) + } + case .failed, .waiting, .cancelled: + answer.resume(returning: nil) + default: + break + } + } + connection.start(queue: queue) + return try? await answer.wait() + } + + /// Network.framework's errors in the shapes the SDK reads elsewhere (`isNetworkError`, `asTransportError`): + /// a timeout as `URLError(.timedOut)`, any other POSIX code as an `NSPOSIXErrorDomain` error. + static func bridged(_ error: NWError) -> Error { + switch error { + case .posix(.ETIMEDOUT): + URLError(.timedOut) + case let .posix(code): + NSError(domain: NSPOSIXErrorDomain, code: Int(code.rawValue)) + default: + error + } } // MARK: - AsyncSequence struct AsyncIterator: AsyncIteratorProtocol { - fileprivate let task: URLSessionWebSocketTask + fileprivate let connection: NWConnection + fileprivate let lifecycle: StateSync + fileprivate let onEvent: EventHandler? func next() async throws -> URLSessionWebSocketTask.Message? { - guard task.closeCode == .invalid else { return nil } - return try await withTaskCancellationHandler { - do { - // Use the callback API instead of the async overlay to avoid - // a TSan-visible data race inside Foundation's continuation bridge. - return try await withCheckedThrowingContinuation { continuation in - task.receive { result in - continuation.resume(with: result) + while lifecycle.copy() == .open { + guard let frame = try await receive() else { return nil } + if let message = frame { return message } + // A control frame (ping, pong): not a message; read on. + } + return nil + } + + /// nil at the end of the stream, `.some(nil)` for a frame that is not a message. + private func receive() async throws -> URLSessionWebSocketTask.Message?? { + try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in + connection.receiveMessage { content, context, _, error in + // Our own close ends the sequence, as a cancelled task's did, whatever arrives after it. This + // receive can be the one that takes the server's answer, ahead of the close path's own. + if lifecycle.copy() != .open { + if WebSocket.endsClose(context, error) { WebSocket.finish(connection, lifecycle, onEvent) } + return continuation.resume(returning: nil) + } + if let error { + WebSocket.finish(connection, lifecycle, onEvent) + return continuation.resume(throwing: LiveKitError(.network, internalError: WebSocket.bridged(error))) + } + let metadata = context?.protocolMetadata(definition: NWProtocolWebSocket.definition) as? NWProtocolWebSocket.Metadata + switch metadata?.opcode { + case .binary?: + continuation.resume(returning: .some(.data(content ?? Data()))) + case .text?: + if let text = String(bytes: content ?? Data(), encoding: .utf8) { + continuation.resume(returning: .some(.string(text))) + } else { + continuation.resume(throwing: LiveKitError(.network, message: "Text frame is not valid UTF-8")) + } + case .close?: + WebSocket.finish(connection, lifecycle, onEvent) + continuation.resume(returning: nil) + case nil where content?.isEmpty ?? true: + // A TCP FIN with no close frame: the end of the stream, which fails the sequence. + WebSocket.finish(connection, lifecycle, onEvent) + continuation.resume(throwing: LiveKitError(.network, internalError: URLError(.networkConnectionLost))) + default: + continuation.resume(returning: .some(nil)) } } - } catch { - // On clean shutdown, task.receive() throws URLError(.cancelled) - // rather than CancellationError. Return nil (end-of-sequence) - // instead of propagating, so `subscribe` doesn't call onFailure. - if task.closeCode != .invalid || Task.isCancelled { return nil } - throw LiveKitError.from(error: error) ?? error + onEvent?(.receiveRegistered) } } onCancel: { - task.cancel(with: .normalClosure, reason: nil) + // As the URLSession task's `cancel(with: .normalClosure)`: `SignalClient.cleanUp` cancels the + // read before it calls `close()`. + WebSocket.close(connection, lifecycle, onEvent) } } } nonisolated func makeAsyncIterator() -> AsyncIterator { - AsyncIterator(task: task) + AsyncIterator(connection: connection, lifecycle: lifecycle, onEvent: onEvent) } // MARK: - Send nonisolated func send(data: Data) async throws { - // Callback API rather than the async overlay, for the same reason as `next()` below: the - // overlay's continuation bridge carries a TSan-visible data race between the thread that - // creates the continuation and the one that completes it. CI caught it on the TSan leg, - // where it aborts the whole test bundle. + let context = NWConnection.ContentContext(identifier: "binary", + metadata: [NWProtocolWebSocket.Metadata(opcode: .binary)]) try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in - task.send(.data(data)) { error in + connection.send(content: data, contentContext: context, isComplete: true, completion: .contentProcessed { error in if let error { - continuation.resume(throwing: error) + continuation.resume(throwing: LiveKitError(.network, internalError: WebSocket.bridged(error))) } else { continuation.resume() } - } - } - } - - // MARK: - Delegate - - private final class Delegate: NSObject, Loggable, URLSessionWebSocketDelegate { - private let _continuation = StateSync?>(nil) - - func setConnectContinuation(_ continuation: CheckedContinuation) { - _continuation.mutate { $0 = continuation } - } - - func cancelConnection() { - _continuation.mutate { - $0?.resume(throwing: LiveKitError(.cancelled)) - $0 = nil - } - } - - func urlSession(_: URLSession, webSocketTask _: URLSessionWebSocketTask, didOpenWithProtocol _: String?) { - _continuation.mutate { - $0?.resume() - $0 = nil - } - } - - func urlSession(_: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) { - log("didCompleteWithError: \(String(describing: error))", error != nil ? .error : .debug) - - // A rejected upgrade (401 on a bad token, 404 on a wrong path) arrives as a - // plain URLError with the HTTP response still attached. Keep the status: it is - // the difference between failing fast and reconnecting forever. - let upgradeStatus = (task.response as? HTTPURLResponse) - .map(\.statusCode) - .flatMap { $0 == 101 ? nil : WebSocketUpgradeFailure(statusCode: $0) } - - _continuation.mutate { - if let upgradeStatus { - $0?.resume(throwing: LiveKitError(.network, internalError: upgradeStatus)) - } else if let error { - $0?.resume(throwing: LiveKitError.from(error: error) ?? LiveKitError(.unknown)) - } else { - $0?.resume() - } - $0 = nil - } + }) } } } diff --git a/Tests/LiveKitCoreTests/RustTransportTests.swift b/Tests/LiveKitCoreTests/RustTransportTests.swift index fe4d6b52b..9daf9ddc5 100644 --- a/Tests/LiveKitCoreTests/RustTransportTests.swift +++ b/Tests/LiveKitCoreTests/RustTransportTests.swift @@ -16,11 +16,11 @@ import Foundation @testable import LiveKit +import LiveKitFFI import Testing #if canImport(LiveKitTestSupport) import LiveKitTestSupport #endif -import LiveKitFFI /// Drives the registered transport from the Rust side, through the FFI, using /// `livekit-net`'s own `self_test_*` exports — the same entry points the Rust @@ -262,18 +262,3 @@ struct RustHTTPRequestTests { #expect(request.httpBody == Data([0x01, 0x02])) } } - -/// `livekit-net` hands the Swift side an ordered header list for the WebSocket upgrade. -@Suite(.tags(.networking)) -struct RustWebSocketRequestTests { - @Test("Repeated header names keep every value") - func repeatedHeadersKeepEveryValue() throws { - let request = try WebSocket.makeRequest(url: #require(URL(string: "wss://example.test/rtc")), - headers: [(name: "Sec-WebSocket-Protocol", value: "a"), - (name: "Sec-WebSocket-Protocol", value: "b"), - (name: "sec-websocket-protocol", value: "c")], - timeoutInterval: 10) - - #expect(request.value(forHTTPHeaderField: "Sec-WebSocket-Protocol") == "a,b,c") - } -} diff --git a/Tests/LiveKitCoreTests/WebSocket/FakeWebSocketServer.swift b/Tests/LiveKitCoreTests/WebSocket/FakeWebSocketServer.swift new file mode 100644 index 000000000..afbec63a3 --- /dev/null +++ b/Tests/LiveKitCoreTests/WebSocket/FakeWebSocketServer.swift @@ -0,0 +1,246 @@ +/* + * Copyright 2026 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import CryptoKit +import Foundation +@testable import LiveKit +import Network +import Testing + +/// A WebSocket server on a loopback port that answers the upgrade by hand and then misbehaves on purpose. +/// Raw TCP rather than Network.framework's own WebSocket listener, so it can send what that never would: +/// a bare FIN, a reset, an HTTP status, or nothing at all. +final class FakeWebSocketServer: Sendable { + enum Behavior: Sendable { + /// Upgrade, then stay quiet until the client sends a close frame, and answer it with one at once. + case idle + /// As `idle`, but answer the client's close frame with a binary message first, and the close frame 50 ms later. + case messageBeforeCloseAnswer + /// Answer the client's close frame at the client's 250 ms deadline. + case closeAnswerAtDeadline + /// As `idle`, but never answer the client's close frame. + case unansweredClose + /// Upgrade, then a close frame with code 1000, then drop the socket once the client answers it. + case closeFrame + /// Upgrade, then one binary message of this many bytes. + case message(size: Int) + /// Upgrade, then a TCP FIN with no close frame. + case fin + /// Upgrade, then a TCP reset. + case reset + /// Answer the upgrade with an HTTP status instead of 101. + case reject(status: Int) + /// Accept TCP and never answer the upgrade. + case blackhole + } + + let behavior: Behavior + /// Resumed once the listener has accepted a TCP connection, before any upgrade is read. + let accepted = AsyncCompleter(label: "FakeWebSocketServer.accepted", defaultTimeout: 5) + /// Resumed once the server has finished with a connection: sent its last bytes or dropped the socket. + let served = AsyncCompleter(label: "FakeWebSocketServer.served", defaultTimeout: 5) + /// The header lines of the last upgrade request, the request line excluded. + let requestHeaders = StateSync<[String]>([]) + /// The client's frames, unmasked, in the order they arrived. + let frames = StateSync<[(opcode: UInt8, payload: Data)]>([]) + /// When the server answered the client's close frame, in `DispatchTime` uptime nanoseconds. + let closeAnswered = StateSync(nil) + /// Resumed with the time, in `DispatchTime` uptime nanoseconds, the client dropped TCP after its close frame. + let clientGone = AsyncCompleter(label: "FakeWebSocketServer.clientGone", defaultTimeout: 5) + + private let listener: NWListener + /// Network.framework delivers on a queue it is given; a shared one, as the transport under test uses. + private static var queue: DispatchQueue { .global(qos: .default) } + private let ready = AsyncCompleter(label: "FakeWebSocketServer.ready", defaultTimeout: 5) + private let stopped = AsyncCompleter(label: "FakeWebSocketServer.stopped", defaultTimeout: 5) + private let connections = StateSync<[NWConnection]>([]) + + init(_ behavior: Behavior) throws { + self.behavior = behavior + listener = try NWListener(using: .tcp, on: .any) + listener.stateUpdateHandler = { [ready, stopped, listener] state in + switch state { + case .ready: ready.resume(returning: listener.port?.rawValue ?? 0) + case let .failed(error): ready.resume(throwing: error) + case .cancelled: + // Below iOS 16 and macOS 13 a cancelled listener keeps its handlers; this one captures the listener. + listener.stateUpdateHandler = nil + stopped.resume(returning: ()) + default: break + } + } + listener.newConnectionHandler = { [self] connection in + connections.mutate { $0.append(connection) } + connection.start(queue: Self.queue) + accepted.resume(returning: ()) + if case .blackhole = behavior { return } + readUpgrade(connection, buffered: Data()) + } + listener.start(queue: Self.queue) + } + + /// `host` as it goes in a URL: an IPv6 address in brackets. + func url(host: String = "127.0.0.1") async throws -> URL { + let port = try await ready.wait() + return try #require(URL(string: "ws://\(host):\(port)/rtc")) + } + + /// Drops every connection and the listener; safe to call more than once. + func cancel() { + connections.mutate { $0.forEach { $0.cancel() }; $0.removeAll() } + listener.cancel() + // It captures the server, which holds the listener. + listener.newConnectionHandler = nil + } + + /// Returns once the port refuses connections again. + func stop() async throws { + cancel() + try await stopped.wait() + } + + private func readUpgrade(_ connection: NWConnection, buffered: Data) { + connection.receive(minimumIncompleteLength: 1, maximumLength: 4096) { [self] content, _, _, error in + guard error == nil, let content else { return connection.cancel() } + let request = buffered + content + guard let headerEnd = request.range(of: Data("\r\n\r\n".utf8)) else { + return readUpgrade(connection, buffered: request) + } + let header = String(bytes: request[.. Void)? = nil, + completion: @escaping @Sendable (_ opcode: UInt8, _ rest: Data) -> Void) + { + let bytes = [UInt8](buffered) + if bytes.count >= 2, bytes[1] & 0x7F == 127 { return connection.cancel() } + let lengthEnd = bytes.count >= 2 && bytes[1] & 0x7F == 126 ? 4 : 2 + if bytes.count >= lengthEnd { + let length = lengthEnd == 4 ? Int(bytes[2]) << 8 | Int(bytes[3]) : Int(bytes[1] & 0x7F) + let maskEnd = lengthEnd + (bytes[1] & 0x80 != 0 ? 4 : 0) + if bytes.count >= maskEnd + length { + let mask = maskEnd > lengthEnd ? Array(bytes[lengthEnd ..< maskEnd]) : [0, 0, 0, 0] + let payload = Data(bytes[maskEnd ..< maskEnd + length].enumerated().map { $1 ^ mask[$0 % 4] }) + frames.mutate { $0.append((opcode: bytes[0] & 0x0F, payload: payload)) } + return completion(bytes[0] & 0x0F, Data(bytes[(maskEnd + length)...])) + } + } + connection.receive(minimumIncompleteLength: 1, maximumLength: 4096) { [self] content, _, _, error in + guard error == nil, let content, !content.isEmpty else { + if let onEnd { onEnd() } else { connection.cancel() } + return + } + readFrame(connection, buffered: buffered + content, onEnd: onEnd, completion: completion) + } + } + + /// Answers the client's close frame as `behavior` says and keeps the socket open: the client must drop it. + private func answerClose(_ connection: NWConnection, buffered: Data) { + readFrame(connection, buffered: buffered) { [self] opcode, rest in + guard opcode == 0x8 else { return answerClose(connection, buffered: rest) } + awaitTeardown(connection, buffered: rest) + switch behavior { + case .messageBeforeCloseAnswer: + connection.send(content: Data([0x82, 0x01, 0x00]), completion: .contentProcessed { _ in }) + Self.queue.asyncAfter(deadline: .now() + .milliseconds(50)) { [self] in sendCloseAnswer(connection) } + case .closeAnswerAtDeadline: + Self.queue.asyncAfter(deadline: .now() + .milliseconds(250)) { [self] in sendCloseAnswer(connection) } + case .unansweredClose: + break + default: + sendCloseAnswer(connection) + } + } + } + + private func sendCloseAnswer(_ connection: NWConnection) { + closeAnswered.mutate { $0 = DispatchTime.now().uptimeNanoseconds } + connection.send(content: Data([0x88, 0x02, 0x03, 0xE8]), completion: .contentProcessed { [served] _ in + served.resume(returning: ()) + }) + } + + /// Reads on until the client's FIN or reset. + private func awaitTeardown(_ connection: NWConnection, buffered: Data) { + readFrame(connection, buffered: buffered, onEnd: { [clientGone] in + clientGone.resume(returning: DispatchTime.now().uptimeNanoseconds) + }) { [self] _, rest in + awaitTeardown(connection, buffered: rest) + } + } + + private func goAway(_ connection: NWConnection) { + switch behavior { + case .idle, .messageBeforeCloseAnswer, .closeAnswerAtDeadline, .unansweredClose, .blackhole, .reject: + break + case .closeFrame: + connection.send(content: Data([0x88, 0x02, 0x03, 0xE8]), completion: .contentProcessed { [self] _ in + // The client's whole close reply, then the socket goes: the client sees a state change after its stream ended. + readFrame(connection, buffered: Data()) { [served] _, _ in + connection.cancel() + served.resume(returning: ()) + } + }) + case let .message(size): + // A server frame is not masked; 127 takes a 64-bit length. + let header: [UInt8] = [0x82, 127] + (0 ..< 8).reversed().map { UInt8(truncatingIfNeeded: UInt64(size) >> ($0 * 8)) } + connection.send(content: Data(header) + Data(count: size), completion: .contentProcessed { [served] _ in + served.resume(returning: ()) + }) + case .fin: + connection.send(content: nil, contentContext: .finalMessage, isComplete: true, completion: .contentProcessed { [served] _ in + served.resume(returning: ()) + }) + case .reset: + connection.forceCancel() + served.resume(returning: ()) + } + } +} diff --git a/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift new file mode 100644 index 000000000..3ec90f930 --- /dev/null +++ b/Tests/LiveKitCoreTests/WebSocket/WebSocketLifecycleTests.swift @@ -0,0 +1,390 @@ +/* + * Copyright 2026 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import Foundation +@testable import LiveKit +import Network +import Testing + +/// How the socket is closed: `close()`, a cancelled read (how `SignalClient.cleanUp` ends it first), or the last +/// reference going away (`deinit`). +enum ClosePath: Sendable, CaseIterable { + case close + case cancelledRead + case release +} + +/// How iteration over the socket ends. +enum Ending: Sendable, CustomTestStringConvertible { + /// A close frame delivered to a pending receive ends the stream. One that lands while no receive is pending is + /// consumed by Network.framework's close handshake, and the next receive fails with ENOTCONN instead. + case closed + case networkError(posix: Int32?) + + var testDescription: String { + switch self { + case .closed: "closed" + case let .networkError(posix): "network error\(posix.map { " POSIX \($0)" } ?? "")" + } + } +} + +private enum Limits { + /// As `TestLimits.e2e`, resolved at runtime for the iOS 13 floor; a minute is ample for loopback. + static var unit: any SuiteTrait { + if #available(iOS 16, macOS 13, tvOS 16, visionOS 1, *) { + let limit: TimeLimitTrait = .timeLimit(.minutes(1)) + return limit + } + let none: Tag.List = .tags() + return none + } +} + +/// The signalling socket's lifecycle against a server that is not a LiveKit server: how it opens, how it fails to, +/// and how its message sequence ends for each way a peer can go away. +@Suite(.tags(.networking), Limits.unit) +struct WebSocketLifecycleTests { + private static let headers = [(name: "Authorization", value: "Bearer test")] + /// How long `close()` waits for the server's answer, in nanoseconds. + private static let closeLimit: UInt64 = 250_000_000 + + // MARK: - Opening + + @Test func ownCloseEndsTheSequence() async throws { + let server = try FakeWebSocketServer(.idle) + defer { server.cancel() } + let events = SocketEvents() + let socket = try await Self.open(server, events: events) + let reader = Task { try await Self.messageCount(of: socket) } + try await events.receiveRegistered.wait() + socket.close() + #expect(try await reader.value == 0) + } + + /// As URLSession's `cancel(with: .normalClosure)`: the server sees a normal disconnect, and then the client drops the + /// connection, whichever way the socket closes. + @Test(arguments: ClosePath.allCases) + func closeSendsNormalClosure(path: ClosePath) async throws { + let server = try FakeWebSocketServer(.unansweredClose) + defer { server.cancel() } + let events = SocketEvents() + var socket: LiveKit.WebSocket? = try await Self.open(server, events: events) + switch path { + case .close: + socket?.close() + socket?.close() + case .cancelledRead: + let reading = try #require(socket) + let reader = Task { try await Self.messageCount(of: reading) } + try await events.receiveRegistered.wait() + reader.cancel() + #expect(try await reader.value == 0) + case .release: + socket = nil + } + _ = try await server.clientGone.wait() + try await events.closeDeadline.wait() + #expect(events.teardowns.copy() == 1) + #expect(server.frames.read { $0.filter { $0.opcode == 0x8 }.count } == 1) + let close = server.frames.read { $0.last } + #expect(close?.opcode == 0x8) + #expect(close?.payload == Data([0x03, 0xE8])) + } + + @Test(arguments: [FakeWebSocketServer.Behavior.idle, .messageBeforeCloseAnswer, .closeAnswerAtDeadline, .unansweredClose]) + func closeWithPendingReadTearsDownOnce(behavior: FakeWebSocketServer.Behavior) async throws { + let server = try FakeWebSocketServer(behavior) + defer { server.cancel() } + let events = SocketEvents() + let socket = try await Self.open(server, events: events) + let reader = Task { try await Self.messageCount(of: socket) } + try await events.receiveRegistered.wait() + + socket.close() + socket.close() + #expect(try await reader.value == 0) + _ = try await server.clientGone.wait() + try await events.closeDeadline.wait() + #expect(events.teardowns.copy() == 1) + #expect(server.frames.read { $0.filter { $0.opcode == 0x8 }.count } == 1) + #expect(await socket.connection.stateUpdateHandler == nil) + } + + @Test func promptCloseAnswerEndsBeforeTheLimit() async throws { + let server = try FakeWebSocketServer(.idle) + defer { server.cancel() } + let socket = try await Self.open(server) + let start = DispatchTime.now().uptimeNanoseconds + socket.close() + let gone = try await server.clientGone.wait() + let answered = try #require(server.closeAnswered.copy()) + #expect(gone >= answered) + #expect(gone < start + Self.closeLimit) + } + + /// A message that the server sends before it answers our close frame does not end the wait: the client drops the + /// connection after the answer, or at the 250 ms limit if the answer comes later than that. + @Test func closeWaitsPastAMessageForTheAnswer() async throws { + let server = try FakeWebSocketServer(.messageBeforeCloseAnswer) + defer { server.cancel() } + let socket = try await Self.open(server) + let start = DispatchTime.now().uptimeNanoseconds + socket.close() + let gone = try await server.clientGone.wait() + let answered = try #require(server.closeAnswered.copy()) + #expect(gone >= answered) + #expect(gone < start + Self.closeLimit) + } + + /// With no answer the client drops the connection at the 250 ms limit and within one second. + @Test func unansweredCloseEndsAtTheLimit() async throws { + let server = try FakeWebSocketServer(.unansweredClose) + defer { server.cancel() } + let socket = try await Self.open(server) + let start = DispatchTime.now().uptimeNanoseconds + socket.close() + let gone = try await server.clientGone.wait() + #expect(gone >= start + Self.closeLimit) + #expect(gone < start + 1_000_000_000) + } + + /// Network.framework's own `Host` has no port and no IPv6 brackets, and `setAdditionalHeaders` keeps only the last + /// value of a repeated name. + @Test(arguments: ["127.0.0.1", "[::1]"]) + func upgradeHeaders(host: String) async throws { + let server = try FakeWebSocketServer(.idle) + defer { server.cancel() } + let url = try await server.url(host: host) + let headers = Self.headers + [(name: "X-Repeated", value: "a"), (name: "x-repeated", value: "b"), (name: "Host", value: "caller")] + _ = try await LiveKit.WebSocket(url: url, headers: headers, timeoutInterval: 10) + + let port = try #require(url.port) + let lines = server.requestHeaders.copy() + #expect(lines.filter { $0.lowercased().hasPrefix("host:") } == ["Host: \(host):\(port)"]) + #expect(lines.contains("X-Repeated: a,b")) + } + + /// URLSessionWebSocketTask's limit on a received message. + @Test(arguments: [1 << 20, (1 << 20) + 1]) + func maximumMessageSize(size: Int) async throws { + let server = try FakeWebSocketServer(.message(size: size)) + defer { server.cancel() } + let socket = try await Self.open(server) + try await socket.send(data: Data([0x01])) + + let iterator = socket.makeAsyncIterator() + if size <= 1 << 20 { + guard case let .data(message) = try await iterator.next() else { return #expect(Bool(false)) } + #expect(message.count == size) + } else { + let error = await #expect(throws: LiveKitError.self) { try await iterator.next() } + #expect(error?.type == .network) + } + } + + /// The cookies stored for the URL go with the upgrade, as URLSession sent them; secure ones on a `wss` URL too. + @Test func upgradeSendsStoredCookies() async throws { + let server = try FakeWebSocketServer(.idle) + defer { server.cancel() } + let url = try await server.url() + let name = "lk-test-\(UUID().uuidString.prefix(8))" + let cookies = try [ + #require(HTTPCookie(properties: [.domain: "127.0.0.1", .path: "/", .name: name, .value: "plain"])), + #require(HTTPCookie(properties: [.domain: "cookie.test", .path: "/", .name: name, .value: "secure", .secure: "TRUE"])), + ] + cookies.forEach(HTTPCookieStorage.shared.setCookie) + defer { cookies.forEach(HTTPCookieStorage.shared.deleteCookie) } + + let headers = Self.headers + [(name: "Cookie", value: "caller=one"), (name: "cookie", value: "other=two")] + let socket = try await LiveKit.WebSocket(url: url, headers: headers, timeoutInterval: 10) + defer { socket.close() } + let lines = server.requestHeaders.copy().filter { $0.lowercased().hasPrefix("cookie:") } + #expect(lines.count == 1) + let cookie = try #require(lines.first) + #expect(cookie.hasPrefix("Cookie: caller=one; other=two; ")) + #expect(cookie.contains("\(name)=plain")) + + let secure = try LiveKit.WebSocket.upgradeHeaders(url: #require(URL(string: "wss://cookie.test/rtc")), headers: []) + #expect(secure.contains { $0.name == "Cookie" && $0.value == "\(name)=secure" }) + } + + @Test func refusedConnectionFails() async throws { + let server = try FakeWebSocketServer(.idle) + defer { server.cancel() } + let url = try await server.url() + try await server.stop() + + let events = SocketEvents() + let error = await #expect(throws: LiveKitError.self) { + try await LiveKit.WebSocket(url: url, headers: Self.headers, timeoutInterval: 10, onEvent: events.record) + } + #expect(events.teardowns.copy() == 1) + #expect(events.closeStarts.copy() == 0) + #expect(error?.type == .network) + #expect((error?.internalError as? NSError)?.code == Int(ECONNREFUSED)) + } + + @Test func deadlineFailsAnUnansweredUpgrade() async throws { + let server = try FakeWebSocketServer(.blackhole) + defer { server.cancel() } + let started = Date() + let events = SocketEvents() + let error = await #expect(throws: LiveKitError.self) { + try await Self.open(server, timeoutInterval: 1, events: events) + } + #expect(events.teardowns.copy() == 1) + #expect(events.closeStarts.copy() == 0) + #expect(Date().timeIntervalSince(started) < 5) + #expect(error?.type == .timedOut || error?.type == .network) + #expect(error?.isRetryableForRegionFailover == true) + } + + @Test func cancellationDuringConnect() async throws { + let server = try FakeWebSocketServer(.blackhole) + defer { server.cancel() } + let url = try await server.url() + let events = SocketEvents() + let connect = Task { try await LiveKit.WebSocket(url: url, headers: Self.headers, timeoutInterval: 10, onEvent: events.record) } + try await server.accepted.wait() + let started = Date() + connect.cancel() + let error = await #expect(throws: LiveKitError.self) { try await connect.value } + #expect(events.teardowns.copy() == 1) + #expect(events.closeStarts.copy() == 0) + #expect(error?.type == .cancelled) + #expect(Date().timeIntervalSince(started) < 5) + } + + @Test func rejectedUpgradeCarriesTheStatus() async throws { + let server = try FakeWebSocketServer(.reject(status: 401)) + defer { server.cancel() } + let events = SocketEvents() + let error = await #expect(throws: LiveKitError.self) { try await Self.open(server, events: events) } + #expect(events.teardowns.copy() == 1) + #expect(events.closeStarts.copy() == 0) + #expect(error?.type == .network) + #expect((error?.internalError as? WebSocketUpgradeFailure)?.statusCode == 401) + } + + // MARK: - Release + + /// What the Network.framework transport exists for: the URLSession one kept its session and task alive after + /// `close()`, about 25 KB a connection, for the life of the process. Here the only transport object is the + /// `NWConnection`, and it goes with the socket. + @Test(.bug("https://github.com/livekit/client-sdk-swift/pull/1173", "URLSession keeps every WebSocket session after close"), + .bug("https://developer.apple.com/forums/thread/650733"), + .bug("https://developer.apple.com/forums/thread/725039")) + func closedSocketsAreReleased() async throws { + let server = try FakeWebSocketServer(.idle) + defer { server.cancel() } + var sockets: [LiveKit.WebSocket] = [] + var transports: [WeakTransport] = [] + for _ in 0 ..< 20 { + let socket = try await Self.open(server) + sockets.append(socket) + await transports.append(WeakTransport(socket: socket, connection: socket.connection)) + } + for socket in sockets { + socket.close() + } + sockets.removeAll() + + let deadline = Date().addingTimeInterval(10) + while transports.contains(where: { !$0.isReleased }), Date() < deadline { + await Task.yield() + } + for (index, transport) in transports.enumerated() { + #expect(transport.socket == nil, "Socket \(index) is still retained") + #expect(transport.connection == nil, "Connection \(index) is still retained") + } + } + + // MARK: - Peer closure + + /// A close frame is the only end of stream; a FIN or a reset fails the sequence so the signal client reconnects + /// at once instead of waiting for its ping timeout. + @Test(arguments: zip( + [FakeWebSocketServer.Behavior.closeFrame, .fin, .reset], + [Ending.closed, .networkError(posix: nil), .networkError(posix: ECONNRESET)], + )) + func peerClosure(behavior: FakeWebSocketServer.Behavior, ending: Ending) async throws { + let server = try FakeWebSocketServer(behavior) + defer { server.cancel() } + let events = SocketEvents() + let socket = try await Self.open(server, events: events) + try await socket.send(data: Data([0x01])) + + switch ending { + case .closed: + do { + #expect(try await Self.messageCount(of: socket) == 0) + } catch let error as LiveKitError { + #expect(error.type == .network) + #expect((error.internalError as? NSError)?.code == Int(ENOTCONN)) + } + case let .networkError(posix): + let error = await #expect(throws: LiveKitError.self) { try await Self.messageCount(of: socket) } + #expect(error?.type == .network) + if let posix { + #expect((error?.internalError as? NSError)?.code == Int(posix)) + } + } + + // The server drops its side after the sequence ended: the late state change lands on a socket whose + // opening is long settled, and must neither crash nor resurface. + try await server.served.wait() + #expect(events.teardowns.copy() == 1) + #expect(await socket.connection.stateUpdateHandler == nil) + withExtendedLifetime(socket) {} + } +} + +private extension WebSocketLifecycleTests { + static func open(_ server: FakeWebSocketServer, timeoutInterval: TimeInterval = 10, + events: SocketEvents? = nil) async throws -> LiveKit.WebSocket + { + try await LiveKit.WebSocket(url: server.url(), headers: headers, timeoutInterval: timeoutInterval, + onEvent: events?.record) + } + + static func messageCount(of socket: LiveKit.WebSocket) async throws -> Int { + try await socket.reduce(0) { count, _ in count + 1 } + } +} + +private struct WeakTransport { + weak var socket: LiveKit.WebSocket? + weak var connection: NWConnection? + + var isReleased: Bool { socket == nil && connection == nil } +} + +private struct SocketEvents: Sendable { + let receiveRegistered = AsyncCompleter(label: "SocketEvents.receiveRegistered", defaultTimeout: 5) + let closeDeadline = AsyncCompleter(label: "SocketEvents.closeDeadline", defaultTimeout: 5) + let teardowns = StateSync(0) + let closeStarts = StateSync(0) + + func record(_ event: LiveKit.WebSocket.Event) { + switch event { + case .receiveRegistered: receiveRegistered.resume(returning: ()) + case .closeDeadline: closeDeadline.resume(returning: ()) + case .teardown: teardowns.mutate { $0 += 1 } + case .closeStarted: closeStarts.mutate { $0 += 1 } + } + } +}