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

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