diff --git a/Sources/Phoenix/PhoenixChannel.swift b/Sources/Phoenix/PhoenixChannel.swift index 89c1f90..b38acf3 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,16 @@ private extension PhoenixChannel { String(describing: TimeoutError()) ) - 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() @@ -269,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( @@ -425,7 +431,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 +702,6 @@ private struct State: @unchecked Sendable { private struct NotReadyToJoinError: Error {} -private struct JoinTimeOutError: Error { +private struct JoinTimeoutError: Error { let joinRef: Ref? } diff --git a/Sources/Phoenix/PhoenixSocket.swift b/Sources/Phoenix/PhoenixSocket.swift index f0d0056..451de89 100644 --- a/Sources/Phoenix/PhoenixSocket.swift +++ b/Sources/Phoenix/PhoenixSocket.swift @@ -432,7 +432,7 @@ extension PhoenixSocket { do { if let timeout = Self.reconnectDelay(attempts: attempts) { - try await Task.sleep(nanoseconds: NSEC_PER_SEC * UInt64(timeout)) + try await Task.sleep(nanoseconds: timeout.nanoseconds) } guard case .waitingToReconnect = _connectionState.value, @@ -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) @@ -501,14 +510,20 @@ extension PhoenixSocket { ) async { let timeout = TimeInterval(nanoseconds: timeout ?? self.timeout) - func cancelAllInputOutput() { + switch _connectionState.value { + case let .connecting(ws) where ws.id == id: + os_log( + "close: %@", + log: .phoenix, + type: .error, + String(describing: error) + ) + _connectionState.value = .closing(ws) pushes.pause(error: error) - tasks.cancelAll() - } + tasks.cancelAll(where: { $0 != "reconnect" }) + try? await ws.close(closeCode(from: error), timeout) - switch _connectionState.value { - case let .connecting(ws) where ws.id == id, - let .open(ws) where ws.id == id: + case let .open(ws) where ws.id == id: os_log( "close: %@", log: .phoenix, @@ -517,7 +532,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) @@ -538,7 +554,6 @@ private extension PhoenixSocket { id += 1 return id } - return try await makeWebSocket( id, // id url(), // url @@ -546,7 +561,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/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 be5a00c..9d7b904 100644 --- a/Tests/PhoenixTests/PhoenixSocketTests.swift +++ b/Tests/PhoenixTests/PhoenixSocketTests.swift @@ -748,6 +748,165 @@ 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 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) + 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)