Skip to content

Fix bugs #51

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
atdrendel merged 3 commits into from
Apr 20, 2026
Merged

Fix bugs #51

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4ccf25f
Do not allow delayed disconnects to prevent reconnections
atdrendel Apr 19, 2026
6316b4c
Ensure Phoenix reconnects when messages ends
atdrendel Apr 19, 2026
a53f082
Fix channel bugs
atdrendel Apr 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions Package.resolved
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Package.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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: [
Expand Down
4 changes: 4 additions & 0 deletions Sources/Phoenix/Message.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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])

Expand Down
134 changes: 104 additions & 30 deletions Sources/Phoenix/PhoenixChannel.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<State>
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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)
Expand Down Expand Up @@ -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 {
Expand All @@ -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)
Comment on lines +278 to +285

Copilot AI Apr 20, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sendLeaveAfterJoinTimeout awaits socket.send(push), which waits until the leave push is actually sent (and can itself take up to the socket/join timeout). This can extend the overall join() call beyond the intended timeout in cases where sending is blocked. Consider firing the leave push in a detached task (best-effort) or adding a non-blocking send path so the original join timeout returns promptly while still attempting to notify the server.

Suggested change
let push = Push(
topic: topic,
event: .leave,
timeout: Date(timeIntervalSinceNow: timeout)
)
push.prepareToSend(ref: await socket.makeRef(), joinRef: joinRef)
try? await socket.send(push)
Task.detached { [socket, topic] in
let push = Push(
topic: topic,
event: .leave,
timeout: Date(timeIntervalSinceNow: timeout)
)
push.prepareToSend(ref: await socket.makeRef(), joinRef: joinRef)
try? await socket.send(push)
}

Copilot uses AI. Check for mistakes.
}

func scheduleRejoinIfPossible(timeout: TimeInterval? = nil) {
tasks.storedNewTask(key: "rejoin") { [weak self] in
try Task.checkCancellation()
Expand All @@ -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
}
}

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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 {}
}
Expand All @@ -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()
Expand Down Expand Up @@ -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() }

Expand All @@ -625,3 +695,7 @@ private struct State: @unchecked Sendable {
}

private struct NotReadyToJoinError: Error {}

private struct JoinTimeOutError: Error {

Copilot AI Apr 20, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JoinTimeOutError uses inconsistent capitalization compared to TimeoutError ("TimeOut" vs "Timeout"). Renaming to JoinTimeoutError would better match Swift naming conventions and the existing type name.

Suggested change
private struct JoinTimeOutError: Error {
private struct JoinTimeoutError: Error {

Copilot uses AI. Check for mistakes.
let joinRef: Ref?
}
1 change: 1 addition & 0 deletions Sources/Phoenix/PhoenixError.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
36 changes: 26 additions & 10 deletions Sources/Phoenix/PhoenixSocket.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -211,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))
Expand Down Expand Up @@ -288,6 +293,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")
Expand Down Expand Up @@ -473,6 +485,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)
}

Expand Down Expand Up @@ -545,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]
}
}

Expand Down
Loading
Loading