From 0d44909d4d38cc89abfc21b3bb3fc9504ac677c6 Mon Sep 17 00:00:00 2001 From: tmm Date: Thu, 5 Nov 2026 23:25:52 +0000 Subject: [PATCH 1/3] test(zero-swift): report where the URLSession frame order test stalls After 30 s the test records how many frames the server got and what the client's writer logged, so a stall shows whether a client send or the server's receive stopped. Amp-Thread-ID: https://ampcode.com/threads/T-c513a037-a257-45e0-b9a9-15c15edfe859 Co-authored-by: tmm --- .../AppleTests/URLSessionTransportTests.swift | 24 +++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/swift/Tests/AppleTests/URLSessionTransportTests.swift b/swift/Tests/AppleTests/URLSessionTransportTests.swift index beac08c426..613a677e49 100644 --- a/swift/Tests/AppleTests/URLSessionTransportTests.swift +++ b/swift/Tests/AppleTests/URLSessionTransportTests.swift @@ -76,16 +76,32 @@ let factory = TransportSocketFactory(URLSessionTransport()) let connection = factory.open( url: "ws://127.0.0.1:\(port)/", secProtocol: nil, events: recorder) + // The writer logs a send still waiting after a second, so a stall + // shows whether the client's send or the server's receive stopped. + let sink = RecordingLogSink() + (connection as? TransportConnection)?.log = Log(level: .debug, sink: sink) for i in 0..<200 { connection.send("\(i)") } guard case .connected(let peer, _)? = await events.next() else { Issue.record("not connected") return } - var texts: [String] = [] - while texts.count < 200, case .message(_, .text(let text))? = await events.next() { - texts.append(text) + let texts = Locked<[String]>([]) + let watchdog = Task { + try await Task.sleep(for: .seconds(30)) + let received = texts.withLock { $0 } + Issue.record( + """ + The server got \(received.count) of 200 frames in 30 s, the last \ + \(received.last ?? "none"); the client logged \(sink.entries.map(\.message)) + """) } - #expect(texts == (0..<200).map(String.init)) + defer { watchdog.cancel() } + while texts.withLock({ $0.count }) < 200, + case .message(_, .text(let text))? = await events.next() + { + texts.withLock { $0.append(text) } + } + #expect(texts.withLock { $0 } == (0..<200).map(String.init)) #expect(recorder.opened) peer.close(code: 4003, reason: "done") From 199ce11e2dd3ea046560de70ee527a342a217da8 Mon Sep 17 00:00:00 2001 From: Amp Date: Thu, 5 Nov 2026 23:36:48 +0000 Subject: [PATCH 2/3] test(zero-swift): time the frame order test's sends off the main actor The first report showed the frames arriving late, not lost, but the writer's log runs on the main actor and couldn't tell a slow send from a busy main actor. The test now times each send in a wrapping socket, counts the server's frames off the main actor, and measures the main actor's longest wait. Amp-Thread-ID: https://ampcode.com/threads/T-c513a037-a257-45e0-b9a9-15c15edfe859 Co-authored-by: tmm --- .../AppleTests/URLSessionTransportTests.swift | 132 +++++++++++++++--- 1 file changed, 109 insertions(+), 23 deletions(-) diff --git a/swift/Tests/AppleTests/URLSessionTransportTests.swift b/swift/Tests/AppleTests/URLSessionTransportTests.swift index 613a677e49..a797e009e4 100644 --- a/swift/Tests/AppleTests/URLSessionTransportTests.swift +++ b/swift/Tests/AppleTests/URLSessionTransportTests.swift @@ -66,45 +66,131 @@ } } + /// When the client's sends ran and finished and the server got the + /// frames, timed off the main actor, which the client's writer runs on. + struct FrameTimes: Sendable { + let start = ContinuousClock.now + var started = 0 + var sent = 0 + var failures: [String] = [] + var lastSent: Duration = .zero + var slowestSend: Duration = .zero + var received = 0 + var lastReceived: Duration = .zero + var mainActorWait: Duration = .zero + var mainActorWaitAt: Duration = .zero + + var report: String { + func ms(_ d: Duration) -> Int { Int(d / .milliseconds(1)) } + return """ + The server got \(received) of 200 frames, the last at \(ms(lastReceived)) ms. \ + The client started \(started) sends and finished \(sent), the last at \ + \(ms(lastSent)) ms and the slowest in \(ms(slowestSend)) ms; failed: \(failures). \ + The main actor's longest wait was \(ms(mainActorWait)) ms, at \(ms(mainActorWaitAt)) ms. + """ + } + } + + struct TimedTransport: Transport { + let base: any Transport + let times: Locked + + func connect(url: String, secProtocol: String?) async throws -> any Socket { + TimedSocket(base: try await base.connect(url: url, secProtocol: secProtocol), times: times) + } + } + + final class TimedSocket: Socket { + let base: any Socket + let times: Locked + + init(base: any Socket, times: Locked) { + self.base = base + self.times = times + } + + func send(_ message: SocketMessage) async throws { + let began = ContinuousClock.now + times.withLock { $0.started += 1 } + do { + try await base.send(message) + } catch { + times.withLock { $0.failures.append("\(error)") } + throw error + } + let now = ContinuousClock.now + times.withLock { t in + t.sent += 1 + t.lastSent = now - t.start + t.slowestSend = max(t.slowestSend, now - began) + } + } + + func receive() async throws -> SocketMessage? { try await base.receive() } + + func close(code: Int, reason: String) async { await base.close(code: code, reason: reason) } + } + @Test @MainActor func keepsTheOrderOfFramesSentBeforeAndAfterTheOpen() async throws { let server = try WebSocketServer() let port = try await server.start() defer { server.stop() } - var events = server.events.makeAsyncIterator() + let times = Locked(FrameTimes()) let recorder = Recorder() - let factory = TransportSocketFactory(URLSessionTransport()) + let factory = TransportSocketFactory( + TimedTransport(base: URLSessionTransport(), times: times)) let connection = factory.open( url: "ws://127.0.0.1:\(port)/", secProtocol: nil, events: recorder) - // The writer logs a send still waiting after a second, so a stall - // shows whether the client's send or the server's receive stopped. - let sink = RecordingLogSink() - (connection as? TransportConnection)?.log = Log(level: .debug, sink: sink) for i in 0..<200 { connection.send("\(i)") } - guard case .connected(let peer, _)? = await events.next() else { - Issue.record("not connected") - return + + // Off the main actor, so that a busy main actor doesn't delay when the + // frames are counted. + let events = server.events + let reading = Task.detached { () -> (peer: WebSocketServer.Peer, texts: [String])? in + var events = events.makeAsyncIterator() + guard case .connected(let peer, _)? = await events.next() else { return nil } + var texts: [String] = [] + while texts.count < 200, case .message(_, .text(let text))? = await events.next() { + texts.append(text) + let now = ContinuousClock.now + times.withLock { t in + t.received += 1 + t.lastReceived = now - t.start + } + } + return (peer, texts) + } + let probe = Task.detached { + while !Task.isCancelled { + let asked = ContinuousClock.now + await MainActor.run {} + let waited = ContinuousClock.now - asked + times.withLock { t in + if waited > t.mainActorWait { + t.mainActorWait = waited + t.mainActorWaitAt = asked - t.start + } + } + try? await Task.sleep(for: .milliseconds(50)) + } } - let texts = Locked<[String]>([]) let watchdog = Task { try await Task.sleep(for: .seconds(30)) - let received = texts.withLock { $0 } - Issue.record( - """ - The server got \(received.count) of 200 frames in 30 s, the last \ - \(received.last ?? "none"); the client logged \(sink.entries.map(\.message)) - """) + Issue.record("After 30 s: \(times.withLock { $0.report })") } - defer { watchdog.cancel() } - while texts.withLock({ $0.count }) < 200, - case .message(_, .text(let text))? = await events.next() - { - texts.withLock { $0.append(text) } + defer { + probe.cancel() + watchdog.cancel() + } + guard let read = await reading.value else { + Issue.record("not connected") + return } - #expect(texts.withLock { $0 } == (0..<200).map(String.init)) + #expect(read.texts == (0..<200).map(String.init)) #expect(recorder.opened) - peer.close(code: 4003, reason: "done") + read.peer.close(code: 4003, reason: "done") while recorder.closed == nil { try await Task.sleep(for: .milliseconds(10)) } #expect(recorder.closed == SocketClosed(code: 4003, reason: "done", wasClean: true)) } From b04c6b2cfec8f5244175e1ae6b37a89f5032f8f8 Mon Sep 17 00:00:00 2001 From: Amp Date: Thu, 5 Nov 2026 23:56:44 +0000 Subject: [PATCH 3/3] fix(zero-swift): send WebSocket frames off the main actor The frame order test's report showed each send finishing in at most 7 ms but only 107 of 200 starting in 30 s, with the main actor waiting up to 6.6 s: the writer ran on the main actor and waited for it after every frame. It now runs detached. A new test blocks the main actor for a second after queuing 200 frames; without the fix the server gets none. Amp-Thread-ID: https://ampcode.com/threads/T-c513a037-a257-45e0-b9a9-15c15edfe859 Co-authored-by: tmm --- .changeset/swift-writer-off-main.md | 5 ++++ swift/Sources/Zero/Client/Scheduler.swift | 2 +- swift/Sources/Zero/Client/Socket.swift | 6 +++-- .../AppleTests/URLSessionTransportTests.swift | 26 +++++++++++++++++++ 4 files changed, 36 insertions(+), 3 deletions(-) create mode 100644 .changeset/swift-writer-off-main.md diff --git a/.changeset/swift-writer-off-main.md b/.changeset/swift-writer-off-main.md new file mode 100644 index 0000000000..7b114ffb54 --- /dev/null +++ b/.changeset/swift-writer-off-main.md @@ -0,0 +1,5 @@ +--- +'zero-swift': patch +--- + +The client's WebSocket frames, such as mutations and query changes, no longer wait for a busy main thread. The writer that sends them one at a time ran on the main actor and went back to it after every frame, so while the app kept the main thread busy each frame waited its turn: on CI, 107 frames took 30 seconds to go out. The writer now runs off the main actor and still sends the frames, and the close after them, in order. diff --git a/swift/Sources/Zero/Client/Scheduler.swift b/swift/Sources/Zero/Client/Scheduler.swift index 2027e12138..387c171c08 100644 --- a/swift/Sources/Zero/Client/Scheduler.swift +++ b/swift/Sources/Zero/Client/Scheduler.swift @@ -76,7 +76,7 @@ public final class SystemScheduler: Scheduler { return timer } - static func milliseconds(_ d: Duration) -> Double { + nonisolated static func milliseconds(_ d: Duration) -> Double { let (s, attoseconds) = d.components return Double(s) * 1000 + Double(attoseconds) / 1e15 } diff --git a/swift/Sources/Zero/Client/Socket.swift b/swift/Sources/Zero/Client/Socket.swift index 56e58b33fc..a0c719399b 100644 --- a/swift/Sources/Zero/Client/Socket.swift +++ b/swift/Sources/Zero/Client/Socket.swift @@ -89,9 +89,11 @@ final class TransportConnection: SocketConnection { return } // One writer keeps the frames in order; each send in a task of its - // own could overtake the one before. + // own could overtake the one before. It runs off the main actor: on + // it, every frame would wait for the main actor to come back from its + // send, so a busy main thread would hold up the frames. let log = self.log.flatMap { $0.level <= .debug ? $0 : nil } - Task { + Task.detached { for await frame in frames { switch frame { case .text(let text): diff --git a/swift/Tests/AppleTests/URLSessionTransportTests.swift b/swift/Tests/AppleTests/URLSessionTransportTests.swift index a797e009e4..de927980f1 100644 --- a/swift/Tests/AppleTests/URLSessionTransportTests.swift +++ b/swift/Tests/AppleTests/URLSessionTransportTests.swift @@ -195,6 +195,32 @@ #expect(recorder.closed == SocketClosed(code: 4003, reason: "done", wasClean: true)) } + @Test @MainActor func sendsWhileTheMainActorIsBusy() async throws { + let server = try WebSocketServer() + let port = try await server.start() + defer { server.stop() } + let events = server.events + let received = Locked(0) + let reading = Task.detached { + for await event in events { + guard case .message = event else { continue } + received.withLock { $0 += 1 } + } + } + defer { reading.cancel() } + + let recorder = Recorder() + let connection = TransportSocketFactory(URLSessionTransport()).open( + url: "ws://127.0.0.1:\(port)/", secProtocol: nil, events: recorder) + defer { connection.close(code: nil) } + while !recorder.opened { try await Task.sleep(for: .milliseconds(10)) } + + for i in 0..<200 { connection.send("\(i)") } + // The main actor stays busy; the frames still go out. + usleep(1_000_000) + #expect(received.withLock { $0 } == 200) + } + @Test @MainActor func connectsAClientThroughTheDefaultFactory() async throws { let server = try WebSocketServer() let port = try await server.start()