From 3d3a8f46d8a9692ce5acbe7f55cd6997f0eb4b00 Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Mon, 20 Apr 2026 19:07:53 +0200 Subject: [PATCH 1/6] Do not wait for sendLeaveAfterJoinTimeout --- Sources/Phoenix/PhoenixChannel.swift | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/Sources/Phoenix/PhoenixChannel.swift b/Sources/Phoenix/PhoenixChannel.swift index 89c1f90..f0a3e2c 100644 --- a/Sources/Phoenix/PhoenixChannel.swift +++ b/Sources/Phoenix/PhoenixChannel.swift @@ -225,7 +225,7 @@ private extension PhoenixChannel { future?.resolve((ref, reply)) return reply - } catch let error as JoinTimeOutError { + } catch let error as JoinTimeoutError { state.access { $0.didFailJoin(clearJoinRef: true) }?.fail(TimeoutError()) tasks.cancel(forKey: "rejoin") @@ -237,10 +237,12 @@ private extension PhoenixChannel { String(describing: TimeoutError()) ) - await sendLeaveAfterJoinTimeout( - joinRef: error.joinRef, - timeout: timeout - ) + Task { + await sendLeaveAfterJoinTimeout( + joinRef: error.joinRef, + timeout: timeout + ) + } scheduleRejoinIfPossible(timeout: timeout) throw TimeoutError() @@ -425,7 +427,7 @@ private struct State: @unchecked Sendable { do { message = try await socket.request(push) } catch is TimeoutError { - throw JoinTimeOutError(joinRef: push.ref) + throw JoinTimeoutError(joinRef: push.ref) } let (ref, isOk, payload) = try message.refAndReply @@ -696,6 +698,6 @@ private struct State: @unchecked Sendable { private struct NotReadyToJoinError: Error {} -private struct JoinTimeOutError: Error { +private struct JoinTimeoutError: Error { let joinRef: Ref? } From ff84f0b576e5d4693c5f79bcb46a369700ca0eac Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Mon, 20 Apr 2026 21:44:32 +0200 Subject: [PATCH 2/6] Restore reconnect backoff --- Sources/Phoenix/PhoenixSocket.swift | 22 ++++++++++--- Tests/PhoenixTests/PhoenixSocketTests.swift | 35 +++++++++++++++++++++ 2 files changed, 52 insertions(+), 5 deletions(-) diff --git a/Sources/Phoenix/PhoenixSocket.swift b/Sources/Phoenix/PhoenixSocket.swift index f0d0056..725b96f 100644 --- a/Sources/Phoenix/PhoenixSocket.swift +++ b/Sources/Phoenix/PhoenixSocket.swift @@ -432,7 +432,8 @@ extension PhoenixSocket { do { if let timeout = Self.reconnectDelay(attempts: attempts) { - try await Task.sleep(nanoseconds: NSEC_PER_SEC * UInt64(timeout)) + let ns = UInt64(Double(NSEC_PER_SEC) * timeout) + try await Task.sleep(nanoseconds: ns) } guard case .waitingToReconnect = _connectionState.value, @@ -507,8 +508,18 @@ extension PhoenixSocket { } switch _connectionState.value { - case let .connecting(ws) where ws.id == id, - let .open(ws) where ws.id == id: + case let .connecting(ws) where ws.id == id: + os_log( + "close: %@", + log: .phoenix, + type: .error, + String(describing: error) + ) + _connectionState.value = .closing(ws) + cancelAllInputOutput() + try? await ws.close(closeCode(from: error), timeout) + + case let .open(ws) where ws.id == id: os_log( "close: %@", log: .phoenix, @@ -538,7 +549,6 @@ private extension PhoenixSocket { id += 1 return id } - return try await makeWebSocket( id, // id url(), // url @@ -546,7 +556,9 @@ private extension PhoenixSocket { {}, // onOpen { [id] close in Task { [weak self] in - guard let self, !Task.isCancelled else { return } + guard let self, + !Task.isCancelled + else { return } await doCloseFromServer( id: id, error: WebSocketError.closeCodeAndReason( diff --git a/Tests/PhoenixTests/PhoenixSocketTests.swift b/Tests/PhoenixTests/PhoenixSocketTests.swift index be5a00c..9bee979 100644 --- a/Tests/PhoenixTests/PhoenixSocketTests.swift +++ b/Tests/PhoenixTests/PhoenixSocketTests.swift @@ -748,6 +748,41 @@ final class PhoenixSocketTests: XCTestCase { XCTAssertLessThan(elapsed, .milliseconds(500)) } + func testAppliesBackoffWhenConnectionFailsAndOnCloseFires() async throws { + let openCount = Locked(0) + + let socket = PhoenixSocket( + url: url, + heartbeatInterval: 10, + makeWebSocket: { id, _, _, _, onClose in + WebSocket( + id: id, + open: { + openCount.access { $0 += 1 } + onClose(WebSocketClose(.abnormalClosure, nil)) + throw URLError(.notConnectedToInternet) + }, + close: { _, _ in }, + send: { _ in }, + messagesPublisher: { + Empty( + completeImmediately: false + ).eraseToAnyPublisher() + } + ) + } + ) + + _ = Task { await socket.connect() } + + try await Task.sleep(nanoseconds: NSEC_PER_MSEC * 300) + + let count = openCount.access { $0 } + XCTAssertLessThanOrEqual(count, 5) + + await socket.disconnect(timeout: 0.000001) + } + func testTriggersChannelErrorIfJoining() async throws { let didErrorWhileJoining = Locked(false) From e32b22e8880fca79b7a341eef120f1c0c700aaa4 Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Mon, 20 Apr 2026 23:54:43 +0200 Subject: [PATCH 3/6] Ensure Phoenix client always reconnects --- Sources/Phoenix/PhoenixSocket.swift | 11 ++-- Tests/PhoenixTests/PhoenixSocketTests.swift | 68 +++++++++++++++++++++ 2 files changed, 72 insertions(+), 7 deletions(-) diff --git a/Sources/Phoenix/PhoenixSocket.swift b/Sources/Phoenix/PhoenixSocket.swift index 725b96f..20d53aa 100644 --- a/Sources/Phoenix/PhoenixSocket.swift +++ b/Sources/Phoenix/PhoenixSocket.swift @@ -502,11 +502,6 @@ extension PhoenixSocket { ) async { let timeout = TimeInterval(nanoseconds: timeout ?? self.timeout) - func cancelAllInputOutput() { - pushes.pause(error: error) - tasks.cancelAll() - } - switch _connectionState.value { case let .connecting(ws) where ws.id == id: os_log( @@ -516,7 +511,8 @@ extension PhoenixSocket { String(describing: error) ) _connectionState.value = .closing(ws) - cancelAllInputOutput() + pushes.pause(error: error) + tasks.cancelAll(where: { $0 != "reconnect" }) try? await ws.close(closeCode(from: error), timeout) case let .open(ws) where ws.id == id: @@ -528,7 +524,8 @@ extension PhoenixSocket { ) _connectionState.value = .closing(ws) - cancelAllInputOutput() + pushes.pause(error: error) + tasks.cancelAll() try? await ws.close(closeCode(from: error), timeout) _connectionState.value = .closed(connectionAttempts: 0) diff --git a/Tests/PhoenixTests/PhoenixSocketTests.swift b/Tests/PhoenixTests/PhoenixSocketTests.swift index 9bee979..26dba89 100644 --- a/Tests/PhoenixTests/PhoenixSocketTests.swift +++ b/Tests/PhoenixTests/PhoenixSocketTests.swift @@ -783,6 +783,74 @@ final class PhoenixSocketTests: XCTestCase { await socket.disconnect(timeout: 0.000001) } + func testReconnectContinuesWhenReconnectOpenFailsAfterOnClose() async throws { + let creationCount = Locked(0) + let openAttemptCount = Locked(0) + let initialOnClose = Locked(nil) + + let socket = PhoenixSocket( + url: url, + heartbeatInterval: 10, + makeWebSocket: { id, _, _, onOpen, onClose in + let creation = creationCount.access { count in + defer { count += 1 } + return count + } + + guard creation > 0 else { + initialOnClose.access { $0 = onClose } + return WebSocket( + id: id, + open: { + openAttemptCount.access { $0 += 1 } + onOpen() + }, + close: { _, _ in }, + send: { _ in }, + messagesPublisher: { + Empty( + completeImmediately: false + ).eraseToAnyPublisher() + } + ) + } + + return WebSocket( + id: id, + open: { + let attempt = openAttemptCount.access { count in + defer { count += 1 } + return count + } + guard attempt >= 3 else { + onClose(WebSocketClose(.abnormalClosure, nil)) + throw URLError(.notConnectedToInternet) + } + onOpen() + }, + close: { _, _ in }, + send: { _ in }, + messagesPublisher: { + Empty( + completeImmediately: false + ).eraseToAnyPublisher() + } + ) + } + ) + + await socket.connect() + await AssertTrue(socket.connectionState.isOpen) + + let close = try XCTUnwrap(initialOnClose.access { $0 }) + close(WebSocketClose(.abnormalClosure, nil)) + + await AssertTrueEventually(openAttemptCount.access { $0 >= 4 }, 2) + await AssertTrueEventually(socket.connectionState.isOpen, 2) + + await socket.disconnect(timeout: 0.000001) + } + func testTriggersChannelErrorIfJoining() async throws { let didErrorWhileJoining = Locked(false) From 6a0142f18b744ee61f460578bdc70f6c014911a3 Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Tue, 21 Apr 2026 14:02:52 +0200 Subject: [PATCH 4/6] Update PhoenixSocket.swift Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- Sources/Phoenix/PhoenixSocket.swift | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/Sources/Phoenix/PhoenixSocket.swift b/Sources/Phoenix/PhoenixSocket.swift index 20d53aa..17458f7 100644 --- a/Sources/Phoenix/PhoenixSocket.swift +++ b/Sources/Phoenix/PhoenixSocket.swift @@ -432,8 +432,7 @@ extension PhoenixSocket { do { if let timeout = Self.reconnectDelay(attempts: attempts) { - let ns = UInt64(Double(NSEC_PER_SEC) * timeout) - try await Task.sleep(nanoseconds: ns) + try await Task.sleep(nanoseconds: timeout.nanoseconds) } guard case .waitingToReconnect = _connectionState.value, From c3c6bbd476ce6d924fdaa4225a98b12bd90e4361 Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Tue, 21 Apr 2026 14:18:14 +0200 Subject: [PATCH 5/6] Address pull request comment --- Sources/Phoenix/PhoenixChannel.swift | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/Sources/Phoenix/PhoenixChannel.swift b/Sources/Phoenix/PhoenixChannel.swift index f0a3e2c..b38acf3 100644 --- a/Sources/Phoenix/PhoenixChannel.swift +++ b/Sources/Phoenix/PhoenixChannel.swift @@ -237,12 +237,16 @@ private extension PhoenixChannel { String(describing: TimeoutError()) ) - Task { - await sendLeaveAfterJoinTimeout( - joinRef: error.joinRef, - timeout: timeout - ) + if let joinRef = error.joinRef { + tasks.storedNewTask(key: "leave-\(joinRef)") { [weak self] in + guard let self else { return } + await sendLeaveAfterJoinTimeout( + joinRef: joinRef, + timeout: timeout + ) + } } + scheduleRejoinIfPossible(timeout: timeout) throw TimeoutError() @@ -271,10 +275,10 @@ private extension PhoenixChannel { } func sendLeaveAfterJoinTimeout( - joinRef: Ref?, + joinRef: Ref, timeout: TimeInterval? ) async { - guard let joinRef else { return } + defer { tasks.cancel(forKey: "leave-\(joinRef)") } let timeout = timeout ?? TimeInterval(nanoseconds: socket.timeout) let push = Push( From 48e315f4038e4d8874fbbf79fb480216e713a09d Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Tue, 21 Apr 2026 15:24:53 +0200 Subject: [PATCH 6/6] Fix race condition where WebSocket.open() returns after the Phoenix socket has received close --- Sources/Phoenix/PhoenixSocket.swift | 21 ++++-- Tests/PhoenixTests/PhoenixChannelTests.swift | 77 +++++++++++--------- Tests/PhoenixTests/PhoenixSocketTests.swift | 56 ++++++++++++++ 3 files changed, 115 insertions(+), 39 deletions(-) diff --git a/Sources/Phoenix/PhoenixSocket.swift b/Sources/Phoenix/PhoenixSocket.swift index 17458f7..451de89 100644 --- a/Sources/Phoenix/PhoenixSocket.swift +++ b/Sources/Phoenix/PhoenixSocket.swift @@ -451,12 +451,21 @@ extension PhoenixSocket { try Task.checkCancellation() - _connectionState.value = .open(ws) - pushes.resume() - listen() - flush() - scheduleHeartbeat() - + switch _connectionState.value { + case let .connecting(_ws) where _ws.id == ws.id: + _connectionState.value = .open(ws) + pushes.resume() + listen() + flush() + scheduleHeartbeat() + + case let .closing(_ws) where _ws.id == ws.id && shouldReconnect: + _connectionState.value = .closed(connectionAttempts: 0) + await doConnect() + + default: + break + } } catch { guard !Task.isCancelled else { return } _connectionState.value = .closed(connectionAttempts: attempts + 1) diff --git a/Tests/PhoenixTests/PhoenixChannelTests.swift b/Tests/PhoenixTests/PhoenixChannelTests.swift index 8000ea8..411d5ef 100644 --- a/Tests/PhoenixTests/PhoenixChannelTests.swift +++ b/Tests/PhoenixTests/PhoenixChannelTests.swift @@ -187,52 +187,63 @@ final class PhoenixChannelTests: XCTestCase { func testRetriesJoinWithBackoffAfterTimeout() async throws { try await withSocket { socket in - await socket.connect() - let channel = await self.makeChannel( - rejoinDelay: [0, 0.001, 0.1, 100], - socket - ) + try await self.serialized { + await socket.connect() + let channel = await self.makeChannel( + rejoinDelay: [0, 0.001, 0.1, 100], + socket + ) - let start = Date() + let start = Date.now + let joinFuture = AsyncThrowingFuture(timeout: 2) + await self.yield(3) - Task { - var attempt = 0 - for await msg in self.outgoingMessages { - let message = try! Message.decode(msg) + let messagesTask = self.task { + var attempt = 0 + for await msg in self.outgoingMessages { + let message = try! Message.decode(msg) - // Ignore leave messages - if message.event == .leave { continue } - XCTAssert(message.event == .join) + // Ignore leave messages + if message.event == .leave { continue } + XCTAssert(message.event == .join) - defer { attempt += 1 } + defer { attempt += 1 } - switch attempt { - case 0: - break + switch attempt { + case 0: + break - case 1: - break + case 1: + break - case 2: - try self.sendReply(for: message) + case 2: + try self.sendReply(for: message) + joinFuture.resolve() - default: - XCTFail() + default: + XCTFail() + } } } - } + defer { messagesTask.cancel() } - Task { - await self.wait() - try await channel.join(timeout: 0.01) - } + await self.yield(2) - await AssertTrueEventually(channel.isJoined) + let joinTask = self.task { + try await channel.join(timeout: 0.01) + } + defer { joinTask.cancel() } - XCTAssertGreaterThanOrEqual( - Date().timeIntervalSince(start), - 0.11 - ) + try await joinFuture.value + await self.yield() + XCTAssertTrue(channel.isJoined) + + let stop = Date.now + XCTAssertGreaterThanOrEqual( + stop.timeIntervalSince(start), + 0.11 + ) + } } } diff --git a/Tests/PhoenixTests/PhoenixSocketTests.swift b/Tests/PhoenixTests/PhoenixSocketTests.swift index 26dba89..9d7b904 100644 --- a/Tests/PhoenixTests/PhoenixSocketTests.swift +++ b/Tests/PhoenixTests/PhoenixSocketTests.swift @@ -783,6 +783,62 @@ final class PhoenixSocketTests: XCTestCase { await socket.disconnect(timeout: 0.000001) } + func testReconnectsWhenOpenReturnsAfterCloseCallback() async throws { + let creationCount = Locked(0) + let openCount = Locked(0) + + let socket = PhoenixSocket( + url: url, + heartbeatInterval: 10, + makeWebSocket: { id, _, _, onOpen, onClose in + let creation = creationCount.access { count in + defer { count += 1 } + return count + } + + guard creation == 0 else { + return WebSocket( + id: id, + open: { + openCount.access { $0 += 1 } + onOpen() + }, + close: { _, _ in }, + send: { _ in }, + messagesPublisher: { + Empty( + completeImmediately: false + ).eraseToAnyPublisher() + } + ) + } + + return WebSocket( + id: id, + open: { + openCount.access { $0 += 1 } + onClose(WebSocketClose(.abnormalClosure, nil)) + try await Task.sleep(nanoseconds: NSEC_PER_MSEC * 10) + }, + close: { _, _ in }, + send: { _ in }, + messagesPublisher: { + Empty( + completeImmediately: false + ).eraseToAnyPublisher() + } + ) + } + ) + + await socket.connect() + + await AssertTrueEventually(openCount.access { $0 >= 2 }) + await AssertTrueEventually(socket.connectionState.isOpen) + + await socket.disconnect(timeout: 0.000001) + } + func testReconnectContinuesWhenReconnectOpenFailsAfterOnClose() async throws { let creationCount = Locked(0) let openAttemptCount = Locked(0)