From 4ccf25fe294ee6941a9b4cf15a424479401db87b Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Mon, 20 Apr 2026 00:09:06 +0200 Subject: [PATCH 1/3] Do not allow delayed disconnects to prevent reconnections --- Sources/Phoenix/PhoenixSocket.swift | 9 ++++- Tests/PhoenixTests/PhoenixSocketTests.swift | 39 +++++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/Sources/Phoenix/PhoenixSocket.swift b/Sources/Phoenix/PhoenixSocket.swift index 1bb89a9..005c293 100644 --- a/Sources/Phoenix/PhoenixSocket.swift +++ b/Sources/Phoenix/PhoenixSocket.swift @@ -107,7 +107,10 @@ final actor PhoenixSocket { case .waitingToReconnect, .preparingToReconnect: tasks.cancel(forKey: "reconnect") _connectionState.value = .closed(connectionAttempts: 0) - case .connecting, .open, .closing: + case .closing: + guard shouldReconnect == false else { return } + _connectionState.value = .closed(connectionAttempts: 0) + case .connecting, .open: return } shouldReconnect = true @@ -473,6 +476,10 @@ extension PhoenixSocket { _connectionState.value = .closing(ws) try? await ws.close(timeout: timeout) + guard case let .closing(ws) = _connectionState.value, + ws.id == id, + shouldReconnect == false + else { return } _connectionState.value = .closed(connectionAttempts: 0) } diff --git a/Tests/PhoenixTests/PhoenixSocketTests.swift b/Tests/PhoenixTests/PhoenixSocketTests.swift index 4db4065..bcee991 100644 --- a/Tests/PhoenixTests/PhoenixSocketTests.swift +++ b/Tests/PhoenixTests/PhoenixSocketTests.swift @@ -564,6 +564,45 @@ final class PhoenixSocketTests: XCTestCase { XCTAssertEqual(1, closes.access { $0 }) } + func testConnectDuringClientClosingReconnects() async throws { + let allowCloseToFinish = future(timeout: 2) + let openCount = Locked(0) + + let socket = PhoenixSocket( + url: url, + makeWebSocket: { [weak self] _, _, _, onOpen, onClose in + guard let self else { throw CancellationError() } + return fake( + onOpen: { + openCount.access { $0 += 1 } + onOpen() + }, + onClose: onClose, + close: { code, _ in + try await allowCloseToFinish.value + onClose(WebSocketClose(code, nil)) + } + ) + } + ) + + await socket.connect() + await AssertTrue(socket.connectionState.isOpen) + + let disconnectTask = Task { await socket.disconnect() } + await AssertTrueEventually(socket.connectionState.isClosing) + + await socket.connect() + await AssertTrueEventually(socket.connectionState.isOpen) + XCTAssertGreaterThanOrEqual(openCount.access { $0 }, 2) + + allowCloseToFinish.resolve() + await disconnectTask.value + + await AssertTrueEventually(socket.connectionState.isOpen) + await socket.disconnect(timeout: 0.000001) + } + func testReconnectsIfClosedRemotely() async throws { let opens = Locked(0) let closes = Locked(0) From 6316b4ce5213a3af4ab8eba3256e83b8751da549 Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Mon, 20 Apr 2026 01:44:48 +0200 Subject: [PATCH 2/3] Ensure Phoenix reconnects when messages ends --- Sources/Phoenix/PhoenixSocket.swift | 11 ++++- Sources/Phoenix/Socket.swift | 12 ++++-- Tests/PhoenixTests/PhoenixSocketTests.swift | 47 +++++++++++++++++++++ 3 files changed, 65 insertions(+), 5 deletions(-) diff --git a/Sources/Phoenix/PhoenixSocket.swift b/Sources/Phoenix/PhoenixSocket.swift index 005c293..8482255 100644 --- a/Sources/Phoenix/PhoenixSocket.swift +++ b/Sources/Phoenix/PhoenixSocket.swift @@ -291,6 +291,13 @@ extension PhoenixSocket { if Task.isCancelled { break } } + + guard !Task.isCancelled else { return } + + await self?.doCloseFromServer( + id: ws.id, + error: WebSocketError.closeCodeAndReason(.normalClosure, nil) + ) } tasks.insert(task, forKey: "listen") @@ -476,8 +483,8 @@ extension PhoenixSocket { _connectionState.value = .closing(ws) try? await ws.close(timeout: timeout) - guard case let .closing(ws) = _connectionState.value, - ws.id == id, + guard case let .closing(_ws) = _connectionState.value, + _ws.id == id, shouldReconnect == false else { return } _connectionState.value = .closed(connectionAttempts: 0) diff --git a/Sources/Phoenix/Socket.swift b/Sources/Phoenix/Socket.swift index b86388c..e1ffcff 100644 --- a/Sources/Phoenix/Socket.swift +++ b/Sources/Phoenix/Socket.swift @@ -48,11 +48,17 @@ public extension Socket { encoder: @escaping PushEncoder = Push.encode, maxMessageSize: Int = 5 * 1024 * 1024 ) -> Socket { - let makeWebSocket: MakeWebSocket = { _, url, _, _, _ in - try await WebSocket.system( + let makeWebSocket: MakeWebSocket = { id, url, options, onOpen, onClose in + var options = options + options.maximumMessageSize = maxMessageSize + var webSocket = try await WebSocket.system( url: url, - options: WebSocketOptions(maximumMessageSize: maxMessageSize) + options: options, + onOpen: onOpen, + onClose: onClose ) + webSocket.id = id + return webSocket } let phoenix = PhoenixSocket( diff --git a/Tests/PhoenixTests/PhoenixSocketTests.swift b/Tests/PhoenixTests/PhoenixSocketTests.swift index bcee991..fb8be28 100644 --- a/Tests/PhoenixTests/PhoenixSocketTests.swift +++ b/Tests/PhoenixTests/PhoenixSocketTests.swift @@ -625,6 +625,53 @@ final class PhoenixSocketTests: XCTestCase { } } + func testReconnectsWhenMessageStreamFinishesWithoutCloseCallback() async throws { + let openCount = Locked(0) + let creationCount = Locked(0) + let messages = PassthroughSubject() + + let socket = PhoenixSocket( + url: url, + heartbeatInterval: 10, + makeWebSocket: { id, _, _, onOpen, _ in + let attempt = creationCount.access { count in + defer { count += 1 } + return count + } + + return WebSocket( + id: id, + open: { + openCount.access { $0 += 1 } + onOpen() + }, + close: { _, _ in }, + send: { _ in }, + messagesPublisher: { + if attempt == 0 { + messages.eraseToAnyPublisher() + } else { + Empty( + completeImmediately: false + ).eraseToAnyPublisher() + } + } + ) + } + ) + + await socket.connect() + await AssertTrue(socket.connectionState.isOpen) + await wait() + + messages.send(completion: .finished) + + await AssertTrueEventually(openCount.access({ $0 >= 2 })) + await AssertTrueEventually(socket.connectionState.isOpen) + + await socket.disconnect(timeout: 0.000001) + } + func testDisconnectStopsReconnectionLoop() async throws { let openCount = Locked(0) let shouldFailOpen = Locked(false) From a53f08264a2203923acf82aeede269e90f09676e Mon Sep 17 00:00:00 2001 From: Anthony Drendel Date: Mon, 20 Apr 2026 03:54:09 +0200 Subject: [PATCH 3/3] Fix channel bugs - Rejoin channel immediately after join fails - Throw error when pushing unjoined channel - Leave channel on join timeout --- Package.resolved | 4 +- Package.swift | 2 +- Sources/Phoenix/Message.swift | 4 + Sources/Phoenix/PhoenixChannel.swift | 134 ++++++++--- Sources/Phoenix/PhoenixError.swift | 1 + Sources/Phoenix/PhoenixSocket.swift | 22 +- Tests/PhoenixTests/MessageTests.swift | 12 + Tests/PhoenixTests/PhoenixChannelTests.swift | 229 ++++++++++++++++--- Tests/PhoenixTests/PhoenixSocketTests.swift | 2 +- 9 files changed, 334 insertions(+), 76 deletions(-) create mode 100644 Tests/PhoenixTests/MessageTests.swift diff --git a/Package.resolved b/Package.resolved index bbe3297..fa59e4e 100644 --- a/Package.resolved +++ b/Package.resolved @@ -50,8 +50,8 @@ "kind" : "remoteSourceControl", "location" : "https://github.com/shareup/websocket-apple.git", "state" : { - "revision" : "a176fc4f7b9f7ad4f0a1fa245a2bbb024658a30e", - "version" : "4.1.0" + "revision" : "3a934c8d91e85b732605d58424acf75e96114282", + "version" : "4.1.1" } } ], diff --git a/Package.swift b/Package.swift index e2ba405..4430ee1 100644 --- a/Package.swift +++ b/Package.swift @@ -32,7 +32,7 @@ let package = Package( ), .package( url: "https://github.com/shareup/websocket-apple.git", - from: "4.1.0" + from: "4.1.1" ), ], targets: [ diff --git a/Sources/Phoenix/Message.swift b/Sources/Phoenix/Message.swift index 20ce5e2..1afaa0a 100644 --- a/Sources/Phoenix/Message.swift +++ b/Sources/Phoenix/Message.swift @@ -37,6 +37,10 @@ public struct Message: Hashable, Sendable, CustomStringConvertible { throw DecodingError.invalidType(String(describing: jsonArray)) } + guard arr.count >= 5 else { + throw DecodingError.missingValue("expected 5 message elements") + } + let joinRef: Ref? = _ref(arr[0]) let ref: Ref? = _ref(arr[1]) diff --git a/Sources/Phoenix/PhoenixChannel.swift b/Sources/Phoenix/PhoenixChannel.swift index 6d56b0b..89c1f90 100644 --- a/Sources/Phoenix/PhoenixChannel.swift +++ b/Sources/Phoenix/PhoenixChannel.swift @@ -22,6 +22,7 @@ final class PhoenixChannel: @unchecked Sendable { var isJoining: Bool { state.access { $0.isJoining } } var isJoined: Bool { state.access { $0.isJoined } } var isUnjoined: Bool { state.access { $0.isUnjoined } } + private var isReadyToJoin: Bool { state.access { $0.isReadyToJoin } } private let socket: PhoenixSocket private let state: Locked @@ -95,6 +96,10 @@ final class PhoenixChannel: @unchecked Sendable { payload: JSON = [:], timeout: TimeInterval? = nil ) async throws { + guard isReadyToJoin else { + throw PhoenixError.channelNotJoined(topic) + } + let timeout = timeout ?? TimeInterval(nanoseconds: socket.timeout) let push = Push( topic: topic, @@ -111,6 +116,10 @@ final class PhoenixChannel: @unchecked Sendable { payload: JSON = [:], timeout: TimeInterval? = nil ) async throws -> JSON { + guard isReadyToJoin else { + throw PhoenixError.channelNotJoined(topic) + } + let timeout = timeout ?? TimeInterval(nanoseconds: socket.timeout) let push = Push( topic: topic, @@ -136,13 +145,15 @@ final class PhoenixChannel: @unchecked Sendable { func prepareToSend(_ push: Push) async -> Bool { precondition(push.topic == topic) + if push.event == .join { + let ref = await socket.makeRef() + push.prepareToSend(ref: ref) + state.access { $0.didPrepareToSendJoin(ref: ref) } + return true + } + guard let joinRef = state.access({ $0.joinRef }) else { - if push.event == .join { - push.prepareToSend(ref: await socket.makeRef()) - return true - } else { - return false - } + return false } push.prepareToSend(ref: await socket.makeRef(), joinRef: joinRef) @@ -214,6 +225,25 @@ private extension PhoenixChannel { future?.resolve((ref, reply)) return reply + } catch let error as JoinTimeOutError { + state.access { $0.didFailJoin(clearJoinRef: true) }?.fail(TimeoutError()) + tasks.cancel(forKey: "rejoin") + + os_log( + "join: channel=%{public}s error=%{public}@", + log: .phoenix, + type: .error, + topic, + String(describing: TimeoutError()) + ) + + await sendLeaveAfterJoinTimeout( + joinRef: error.joinRef, + timeout: timeout + ) + scheduleRejoinIfPossible(timeout: timeout) + + throw TimeoutError() } catch let error as NotReadyToJoinError { throw error } catch PhoenixError.leavingChannel { @@ -238,6 +268,23 @@ private extension PhoenixChannel { } } + func sendLeaveAfterJoinTimeout( + joinRef: Ref?, + timeout: TimeInterval? + ) async { + guard let joinRef else { return } + + let timeout = timeout ?? TimeInterval(nanoseconds: socket.timeout) + let push = Push( + topic: topic, + event: .leave, + timeout: Date(timeIntervalSinceNow: timeout) + ) + + push.prepareToSend(ref: await socket.makeRef(), joinRef: joinRef) + try? await socket.send(push) + } + func scheduleRejoinIfPossible(timeout: TimeInterval? = nil) { tasks.storedNewTask(key: "rejoin") { [weak self] in try Task.checkCancellation() @@ -248,21 +295,25 @@ private extension PhoenixChannel { private struct State: @unchecked Sendable { private(set) var connection: Connection - private var isReadyToJoin: Bool + private(set) var isReadyToJoin: Bool private var lastJoinTimeout: TimeInterval? + private var currentJoinRef: Ref? private var rejoinAttempts: Int private var rejoinDelay: [TimeInterval] var joinRef: Ref? { switch connection { - case .unjoined, .errored, .joining, .left: + case .unjoined, .left: nil + case .errored, .joining: + currentJoinRef + case let .joined(ref, _): ref case let .leaving(ref, _): - ref + ref ?? currentJoinRef } } @@ -322,6 +373,10 @@ private struct State: @unchecked Sendable { isReadyToJoin = true } + mutating func didPrepareToSendJoin(ref: Ref) { + currentJoinRef = ref + } + mutating func rejoin( topic: Topic, payload: JSON, @@ -366,7 +421,13 @@ private struct State: @unchecked Sendable { timeout: Date(timeIntervalSinceNow: timeout) ) - let message: Message = try await socket.request(push) + let message: Message + do { + message = try await socket.request(push) + } catch is TimeoutError { + throw JoinTimeOutError(joinRef: push.ref) + } + let (ref, isOk, payload) = try message.refAndReply guard isOk else { @@ -398,17 +459,21 @@ private struct State: @unchecked Sendable { case let .joining(future): rejoinAttempts = 0 + currentJoinRef = ref connection = .joined(ref, reply: reply) return future } } - mutating func didFailJoin() -> JoinFuture? { + mutating func didFailJoin(clearJoinRef: Bool = false) -> JoinFuture? { switch connection { case .unjoined, .errored, .joined, .leaving, .left: return nil case let .joining(future): + if clearJoinRef { + currentJoinRef = nil + } connection = .errored return future } @@ -459,18 +524,22 @@ private struct State: @unchecked Sendable { mutating func leaveImmediately() -> () -> Void { switch connection { case .errored, .left, .unjoined: + currentJoinRef = nil connection = .left return {} case let .joining(join): + currentJoinRef = nil connection = .left return { join.fail(CancellationError()) } case let .leaving(_, leave): + currentJoinRef = nil connection = .left return { leave.resolve() } case .joined: + currentJoinRef = nil connection = .left return {} } @@ -493,29 +562,28 @@ private struct State: @unchecked Sendable { socket: PhoenixSocket, sendMessage: @escaping (Message) -> Void ) -> () async -> Bool { - func doSendMessage() -> () -> Void { - switch connection { - case let .joined(joinRef, _): - if message.joinRef == joinRef || message.joinRef == nil { - { sendMessage(message) } - } else { - { - os_log( - "outdated message: channel=%{public}s joinRef=%d message=%d", - log: .phoenix, - type: .debug, - message.topic, - Int(joinRef.rawValue), - Int(message.joinRef?.rawValue ?? 0) - ) - } - } + let currentJoinRef = joinRef - case .unjoined, .errored, .joining, .leaving, .left: - { sendMessage(message) } + if let messageJoinRef = message.joinRef, + messageJoinRef != currentJoinRef + { + return { + os_log( + "outdated message: channel=%{public}s joinRef=%d message=%d", + log: .phoenix, + type: .debug, + message.topic, + Int(currentJoinRef?.rawValue ?? 0), + Int(messageJoinRef.rawValue) + ) + return false } } + func doSendMessage() -> () -> Void { + { sendMessage(message) } + } + switch message.event { case .close: let doSend = doSendMessage() @@ -603,10 +671,12 @@ private struct State: @unchecked Sendable { return {} case let .joining(future): + currentJoinRef = nil connection = .errored return { future.fail(TimeoutError()) } case let .leaving(_, future): + currentJoinRef = nil connection = .left return { future.resolve() } @@ -625,3 +695,7 @@ private struct State: @unchecked Sendable { } private struct NotReadyToJoinError: Error {} + +private struct JoinTimeOutError: Error { + let joinRef: Ref? +} diff --git a/Sources/Phoenix/PhoenixError.swift b/Sources/Phoenix/PhoenixError.swift index af61a47..5d75b2a 100644 --- a/Sources/Phoenix/PhoenixError.swift +++ b/Sources/Phoenix/PhoenixError.swift @@ -4,6 +4,7 @@ import JSON public enum PhoenixError: Error, Hashable, Sendable { case channelError case channelErrorWithResponse(String, String, JSON) + case channelNotJoined(String) case couldNotDecodeMessage case couldNotEncodePush case invalidReply diff --git a/Sources/Phoenix/PhoenixSocket.swift b/Sources/Phoenix/PhoenixSocket.swift index 8482255..f0d0056 100644 --- a/Sources/Phoenix/PhoenixSocket.swift +++ b/Sources/Phoenix/PhoenixSocket.swift @@ -214,14 +214,16 @@ extension PhoenixSocket { else { return } do { - if let channel = await channels[push.topic] { - guard await channel.prepareToSend(push) else { - pushes.putBack(push) - await Task.yield() - continue + if push.ref == nil { + if let channel = await channels[push.topic] { + guard await channel.prepareToSend(push) else { + pushes.putBack(push) + await Task.yield() + continue + } + } else { + push.prepareToSend(ref: await makeRef()) } - } else { - push.prepareToSend(ref: await makeRef()) } try await ws.send(encoder(push)) @@ -293,7 +295,7 @@ extension PhoenixSocket { } guard !Task.isCancelled else { return } - + await self?.doCloseFromServer( id: ws.id, error: WebSocketError.closeCodeAndReason(.normalClosure, nil) @@ -559,8 +561,8 @@ private extension PhoenixSocket { // Serves the same purpose as `reconnectTimer` in PhoenixJS static func reconnectDelay(attempts: Int) -> TimeInterval? { guard attempts > 0 else { return nil } - guard attempts < 9 else { return 5 } - return [0.01, 0.05, 0.1, 0.15, 0.2, 0.25, 0.5, 1, 2][attempts] + guard attempts <= 9 else { return 5 } + return [0.01, 0.05, 0.1, 0.15, 0.2, 0.25, 0.5, 1, 2][attempts - 1] } } diff --git a/Tests/PhoenixTests/MessageTests.swift b/Tests/PhoenixTests/MessageTests.swift new file mode 100644 index 0000000..3de48cb --- /dev/null +++ b/Tests/PhoenixTests/MessageTests.swift @@ -0,0 +1,12 @@ +import WebSocket +import XCTest + +@testable import Phoenix + +final class MessageTests: XCTestCase { + func testDecodeShortMessageThrows() throws { + XCTAssertThrowsError( + try Message.decode(.text(#"[null,1,"topic"]"#)) + ) + } +} diff --git a/Tests/PhoenixTests/PhoenixChannelTests.swift b/Tests/PhoenixTests/PhoenixChannelTests.swift index 47e6a4c..8000ea8 100644 --- a/Tests/PhoenixTests/PhoenixChannelTests.swift +++ b/Tests/PhoenixTests/PhoenixChannelTests.swift @@ -408,38 +408,47 @@ final class PhoenixChannelTests: XCTestCase { await socket.connect() let sentMessages = Locked<[Message]>([]) + let didSendJoin = AsyncThrowingFuture() + let canReplyToJoin = AsyncThrowingFuture() - try await withThrowingTaskGroup(of: Void.self) { group in - group.addTask { - for await msg in self.outgoingMessages { - let message = try Message.decode(msg) - try self.sendReply(for: message) - let count = sentMessages.access { sentMessages in - sentMessages.append(message) - return sentMessages.count - } - if count == 4 { break } + let listener = Task { + for await msg in self.outgoingMessages { + let message = try Message.decode(msg) + + if message.event == .join { + didSendJoin.resolve() + try await canReplyToJoin.value + } + + try self.sendReply(for: message) + + let count = sentMessages.access { sentMessages in + sentMessages.append(message) + return sentMessages.count } + + if count == 4 { break } } + } - let channel = await self.makeChannel(socket) + let channel = await self.makeChannel(socket) + let joinTask = Task { try await channel.join() } + try await didSendJoin.value + + try await withThrowingTaskGroup(of: Void.self) { group in for event in ["one", "two", "three"] { - group.addTask { - do { - try await channel.send(event) - } catch { - XCTFail("Push should have succeeded instead of \(error)") - } - } + group.addTask { try await channel.send(event) } } try await Task.sleep(nanoseconds: NSEC_PER_MSEC * 10) - try await channel.join() - + canReplyToJoin.resolve() try await group.waitForAll() } + _ = try await joinTask.value + _ = try await listener.value + XCTAssertEqual(4, sentMessages.access { $0.count }) XCTAssertEqual(.join, sentMessages.access { $0[0].event }) } @@ -728,6 +737,8 @@ final class PhoenixChannelTests: XCTestCase { let channel = await self.makeChannel(socket) let isJoined = Locked(false) + let didSendJoin = AsyncThrowingFuture() + let canReplyToJoin = AsyncThrowingFuture() try await withThrowingTaskGroup(of: Void.self) { group in group.addTask { @@ -736,6 +747,8 @@ final class PhoenixChannelTests: XCTestCase { for await msg in self.outgoingMessages { let message = try Message.decode(msg) if message.event == .join { + didSendJoin.resolve() + try await canReplyToJoin.value isJoined.access { $0 = true } didJoin = true } else { @@ -750,6 +763,7 @@ final class PhoenixChannelTests: XCTestCase { } group.addTask { + try await didSendJoin.value do { let resp = try await channel.request("test") XCTAssertTrue(isJoined.access { $0 }) @@ -764,35 +778,186 @@ final class PhoenixChannelTests: XCTestCase { try await channel.join() } + group.addTask { + try await didSendJoin.value + try await Task.sleep(nanoseconds: NSEC_PER_MSEC * 10) + canReplyToJoin.resolve() + } + try await group.waitForAll() } } } - func testDoesNotPushIfNotJoined() async throws { + func testSendThrowsIfJoinNeverCalled() async throws { try await withSocket { socket in await socket.connect() let channel = await self.makeChannel(socket) - try await withThrowingTaskGroup(of: Void.self) { group in - let didSendPush = Locked(false) + do { + try await channel.send("test") + XCTFail("Should have thrown before join() was called") + } catch let error as PhoenixError { + guard case let .channelNotJoined(topic) = error else { + return XCTFail("Unexpected error: \(error)") + } + XCTAssertEqual("topic", topic) + } + } + } - group.addTask { - try await channel.send("test") - didSendPush.access { $0 = true } + func testRequestThrowsIfJoinNeverCalled() async throws { + try await withSocket { socket in + await socket.connect() + let channel = await self.makeChannel(socket) + + do { + _ = try await channel.request("test") + XCTFail("Should have thrown before join() was called") + } catch let error as PhoenixError { + guard case let .channelNotJoined(topic) = error else { + return XCTFail("Unexpected error: \(error)") } + XCTAssertEqual("topic", topic) + } + } + } - group.addTask { - await self.wait() - try await channel.join() + func testSendsLeaveBeforeRejoiningAfterJoinTimeout() async throws { + try await withSocket { socket in + await socket.connect() + let channel = await self.makeChannel( + rejoinDelay: [0], + socket + ) + + let didObserveRejoin = AsyncThrowingFuture() + + let listener = Task { + var joinCount = 0 + var sawLeave = false + + for await msg in self.outgoingMessages { + let message = try Message.decode(msg) + + switch message.event { + case .join: + joinCount += 1 + + if joinCount == 2 { + XCTAssertTrue( + sawLeave, + "Expected phx_leave before retrying phx_join" + ) + didObserveRejoin.resolve() + break + } + + case .leave: + sawLeave = true + + case .custom, .reply, .close, .error, .heartbeat: + break + } } + } - try await Task.sleep(nanoseconds: NSEC_PER_MSEC * 50) + do { + try await channel.join(timeout: 0.01) + XCTFail("Join should have timed out") + } catch is TimeoutError {} - XCTAssertFalse(didSendPush.access { $0 }) + try await didObserveRejoin.value + listener.cancel() + } + } - group.cancelAll() + func testIgnoresOutdatedCloseWhileRejoining() async throws { + try await withSocket { socket in + await socket.connect() + let channel = await self.makeChannel( + rejoinDelay: [0], + socket + ) + + let outgoingMessages = self.outgoingMessages + let channelMessages = channel.messages + + let firstJoinRef = AsyncThrowingFuture() + let didReceiveClose = Locked(false) + let didReceiveError = AsyncThrowingFuture() + + let listener = Task { + var joinCount = 0 + + for await msg in outgoingMessages { + let message = try Message.decode(msg) + + guard message.event == .join else { continue } + joinCount += 1 + + switch joinCount { + case 1: + firstJoinRef.resolve(try XCTUnwrap(message.ref)) + try self.sendReply(for: message) + + case 2: + let staleJoinRef = try await firstJoinRef.value + self.receiveSubject.send( + .text( + """ + [\(staleJoinRef.rawValue),null,"topic","phx_close",{}] + """ + ) + ) + try self.sendReply(for: message, payload: ["rejoined": true]) + return + + default: + XCTFail("Expected only two join attempts") + } + } } + + let messageTask = Task { + for await message in channelMessages { + if message.event == .error { + didReceiveError.resolve() + } else if message.event == .close { + didReceiveClose.access { $0 = true } + break + } + } + } + + await self.wait() + + _ = try await channel.join() + + let initialJoinRef = try await firstJoinRef.value + self.receiveSubject.send( + .text( + """ + [\(initialJoinRef.rawValue),null,"topic","phx_error",{}] + """ + ) + ) + + try await didReceiveError.value + let payload = try await channel.join() + XCTAssertEqual(["rejoined": true], payload) + XCTAssertTrue(channel.isJoined) + + let currentChannel = await socket.channels["topic"] + XCTAssertNotNil(currentChannel) + XCTAssertEqual( + ObjectIdentifier(channel), + ObjectIdentifier(try XCTUnwrap(currentChannel)) + ) + XCTAssertFalse(didReceiveClose.access { $0 }) + + listener.cancel() + messageTask.cancel() } } diff --git a/Tests/PhoenixTests/PhoenixSocketTests.swift b/Tests/PhoenixTests/PhoenixSocketTests.swift index fb8be28..be5a00c 100644 --- a/Tests/PhoenixTests/PhoenixSocketTests.swift +++ b/Tests/PhoenixTests/PhoenixSocketTests.swift @@ -666,7 +666,7 @@ final class PhoenixSocketTests: XCTestCase { messages.send(completion: .finished) - await AssertTrueEventually(openCount.access({ $0 >= 2 })) + await AssertTrueEventually(openCount.access { $0 >= 2 }) await AssertTrueEventually(socket.connectionState.isOpen) await socket.disconnect(timeout: 0.000001)