fix(zero-swift): send WebSocket frames off the main actor (#107)

* 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 <tmm@tmm.dev>

* 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 <tmm@tmm.dev>

* 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 <tmm@tmm.dev>

---------

Co-authored-by: Amp <amp@ampcode.com>

Author: tmm

.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.

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
   }

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):

swift/Tests/AppleTests/URLSessionTransportTests.swift

@@ -66,33 +66,161 @@
       }
     }
 
+    /// 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<FrameTimes>
+
+      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<FrameTimes>
+
+      init(base: any Socket, times: Locked<FrameTimes>) {
+        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)
       for i in 0..<200 { connection.send("\(i)") }
-      guard case .connected(let peer, _)? = await events.next() else {
+
+      // 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 watchdog = Task {
+        try await Task.sleep(for: .seconds(30))
+        Issue.record("After 30 s: \(times.withLock { $0.report })")
+      }
+      defer {
+        probe.cancel()
+        watchdog.cancel()
+      }
+      guard let read = await reading.value else {
         Issue.record("not connected")
         return
       }
-      var texts: [String] = []
-      while texts.count < 200, case .message(_, .text(let text))? = await events.next() {
-        texts.append(text)
-      }
-      #expect(texts == (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))
     }
 
+    @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()