fix(zero-swift): list the group's running clients in activeClients

Two Zero instances in one client group, in one process, list each other
in initConnection's activeClients, and the survivor sends deleteClients
for one that closes, as JS's ActiveClientsManager does.

Amp-Thread-ID: https://ampcode.com/threads/T-c513a037-a257-45e0-b9a9-15c15edfe859
Co-authored-by: tmm <tmm@tmm.dev>

Author: ampagent

.changeset/active-clients.md

@@ -0,0 +1,5 @@
+---
+'zero-swift': patch
+---
+
+Two `Zero` instances of one client group in one process (the same user, `storageKey`, and storage, such as two windows) no longer drop each other's queries: `initConnection` lists every running client of the group in `activeClients`, as in JS, so zero-cache doesn't delete the other one. When one closes, the others send `deleteClients` for it once.

swift/Sources/Zero/Client/Engine.swift

@@ -94,6 +94,7 @@ final class Engine {
     for t in lifecycleTimers { t.cancel() }
     if !repClosed {
       database.removeWriteFailureListener(clientID)
+      database.clientStopped(clientID)
       database.release()
     }
   }
@@ -253,6 +254,14 @@ final class Engine {
     database.addWriteFailureListener(clientID) { [weak self] failure in
       self?.storageFailed(failure)
     }
+    // Another client of the group stopped (the `onDelete` of
+    // `makeActiveClientsManager`): tell the server, if connected.
+    database.clientStarted(clientID, clientGroupID: clientGroupID) { [weak self] id in
+      guard let self else { return }
+      self.loop.dispatch {
+        self.sendIfConnected(.deleteClients(DeleteClientsBody(clientIDs: [id])))
+      }
+    }
     if let openFailure {
       log.warn(
         "Local store storage failed (\(openFailure.kind.rawValue)) opening \(name); "
@@ -690,7 +699,10 @@ final class Engine {
       clientSchema: connectCookie.isNull ? clientSchema : nil, deleted: deleted,
       userPushURL: options.mutateURL, userPushHeaders: options.mutateHeaders,
       userQueryURL: options.queryURL, userQueryHeaders: options.queryHeaders,
-      activeClients: [clientID])
+      activeClients: [clientID]
+        + database.activeClients(ofGroup: clientGroupID).filter {
+          $0 != clientID
+        })
     let authToken = auth.flatMap { $0.isEmpty ? nil : $0 }
     var secProtocol = SecProtocols.encode(initConnection: initConnection, authToken: authToken)
     if secProtocol.utf8.count > options.maxHeaderLength {
@@ -1030,6 +1042,7 @@ final class Engine {
     for t in lifecycleTimers { t.cancel() }
     lifecycleTimers = []
     database.removeWriteFailureListener(clientID)
+    database.clientStopped(clientID)
     database.release()
     connectionManager.closed()
   }

swift/Sources/Zero/Storage/ReplicaDatabase.swift

@@ -163,6 +163,12 @@ final class ReplicaDatabase {
   private var writeFailureListeners: [(clientID: String, listener: (StorageFailure) -> Void)] =
     []
 
+  /// The clients running on this database in this process, in the order
+  /// they started (`ActiveClientsManager`, which JS keeps across tabs
+  /// with Web Locks). Each hears when another of its group stops.
+  private var running: [(clientID: String, clientGroupID: String, onDelete: (String) -> Void)] =
+    []
+
   private static var open: [String: ReplicaDatabase] = [:]
   /// The profile ID of memory databases, one per process as in JS.
   private static var memoryProfileID: String?
@@ -238,6 +244,29 @@ final class ReplicaDatabase {
     sqlite.close()
   }
 
+  /// Records that `clientID` runs until `clientStopped`. `onDelete` gets
+  /// the clients of its group that stop before it.
+  func clientStarted(
+    _ clientID: String, clientGroupID: String, onDelete: @escaping (String) -> Void
+  ) {
+    running.append((clientID, clientGroupID, onDelete))
+  }
+
+  /// Tells the other running clients of its group, once, that `clientID`
+  /// stopped (`ActiveClientsManager.onDelete`).
+  func clientStopped(_ clientID: String) {
+    guard let i = running.firstIndex(where: { $0.clientID == clientID }) else { return }
+    let stopped = running.remove(at: i)
+    for other in running where other.clientGroupID == stopped.clientGroupID {
+      other.onDelete(clientID)
+    }
+  }
+
+  /// The running clients of group `id` (`ActiveClientsManager.activeClients`).
+  func activeClients(ofGroup id: String) -> [String] {
+    running.filter { $0.clientGroupID == id }.map(\.clientID)
+  }
+
   /// Waits for the writes queued so far.
   func flush() { sqlite?.flush() }
 

swift/Tests/E2ETests/ClientE2ETests.swift

@@ -103,4 +103,35 @@ struct ClientE2ETests {
     #expect(Set(["i1", "i3", "i4"]).isSubset(of: cached.map(\.id)))
     #expect(!cached.contains { $0.id == id })
   }
+
+  /// Two clients of one client group (two windows) both keep their
+  /// queries: each lists the other in `activeClients`, so zero-cache
+  /// doesn't delete it.
+  @Test func twoClientsOfAGroupBothKeepTheirQueries() async throws {
+    defer { try? FileManager.default.removeItem(atPath: directory) }
+    let key = UUID().uuidString
+    let first = try makeZero(storageKey: key)
+    defer { first.close() }
+    // Without a TTL, a query of a client the server deletes goes at once.
+    let open = try first.liveQuery(E2EApp.open(), ttl: .none)
+    try await eventually("the first client's issues") { open.resultType == .complete }
+
+    let second = try makeZero(storageKey: key)
+    defer { second.close() }
+    #expect(second.clientGroupID == first.clientGroupID)
+    let byID = try second.liveQuery(E2EApp.byID("i1"))
+    try await eventually("the second client's issue") { byID.resultType == .complete }
+
+    // The first client's query still syncs a row only it asks for.
+    let id = "swift-\(UUID().uuidString)"
+    let created = second.mutate(E2EApp.create(.init(id: id, title: "Two windows", ownerId: nil)))
+    #expect(try await created.server.value == .success(data: nil))
+    try await eventually("the first client to see the new issue") {
+      open.data.contains { $0.id == id }
+    }
+    #expect(open.resultType == .complete)
+    #expect(byID.data?.id == "i1")
+    let closed = second.mutate(E2EApp.close(id))
+    #expect(try await closed.server.value == .success(data: nil))
+  }
 }

swift/Tests/ZeroTests/SweepActiveClientsTests.swift

@@ -0,0 +1,81 @@
+import Foundation
+import Testing
+import ZeroTesting
+
+@testable import Zero
+
+/// Two clients of one client group in one process, checked against the JS
+/// client (`packages/zero-client` active clients).
+@Suite(.serialized) @MainActor struct SweepActiveClientsTests {
+  static let create = defineMutator("issue.create") { (tx, id: String) in
+    try tx.mutate.issue.insert([
+      "id": .string(id), "title": "", "description": "", "closed": false, "ownerId": .null,
+      "createdAt": 1,
+    ])
+  }
+
+  func makeZero(
+    storageKey: String, sockets: MockSocketFactory, scheduler: ManualScheduler
+  ) throws -> Zero<Void> {
+    var options = ZeroOptions(
+      cacheURL: "http://zero.test", userID: "u", storageKey: storageKey, storage: .memory)
+    options.logSink = RecordingLogSink()
+    return try Zero(
+      schema: .test, mutators: [Self.create], options: options, scheduler: scheduler,
+      socketFactory: sockets)
+  }
+
+  // MARK: - Two clients in one client group
+
+  /// Opens the newest socket and acknowledges it.
+  func connect(_ sockets: MockSocketFactory, _ scheduler: ManualScheduler) throws -> MockSocket {
+    scheduler.runDueTimers()
+    let socket = try #require(sockets.sockets.last)
+    socket.receive(text: #"["connected",{"wsid":"w"}]"#)
+    scheduler.runDueTimers()
+    return socket
+  }
+
+  /// JS lists every client of the group running in the process
+  /// (`ActiveClientsManager`; without `navigator.locks` its fallback lock
+  /// manager is process-wide). zero-cache deletes the clients of the group
+  /// that are not in `activeClients` (`view-syncer.ts`), so listing only
+  /// itself makes the server delete the other client's queries while it
+  /// runs.
+  @Test func initConnectionListsTheOtherActiveClientOfTheGroup() throws {
+    let scheduler = ManualScheduler(now: 1_700_000_000_000)
+    let sockets = MockSocketFactory()
+    let key = randomID()
+    let first = try makeZero(storageKey: key, sockets: sockets, scheduler: scheduler)
+    _ = try connect(sockets, scheduler)
+    let second = try makeZero(storageKey: key, sockets: sockets, scheduler: scheduler)
+    #expect(second.clientGroupID == first.clientGroupID)
+    let socket = try connect(sockets, scheduler)
+    let initConnection = try SecProtocols.decode(#require(socket.secProtocol))[
+      "initConnectionMessage"]?[1]
+    let active = Set(initConnection?["activeClients"]?.arrayValue?.compactMap(\.stringValue) ?? [])
+    #expect(active == [first.clientID, second.clientID])
+    first.close()
+    second.close()
+  }
+
+  /// JS: "We should send a deleteClient when a Zero instance is closed"
+  /// (`zero.test.ts`): the other client of the group sends `deleteClients`
+  /// for the closed one, once.
+  @Test func closingAClientMakesTheOtherOfTheGroupDeleteIt() throws {
+    let scheduler = ManualScheduler(now: 1_700_000_000_000)
+    let sockets = MockSocketFactory()
+    let key = randomID()
+    let first = try makeZero(storageKey: key, sockets: sockets, scheduler: scheduler)
+    _ = try connect(sockets, scheduler)
+    let second = try makeZero(storageKey: key, sockets: sockets, scheduler: scheduler)
+    let socket = try connect(sockets, scheduler)
+    #expect(second.clientGroupID == first.clientGroupID)
+
+    first.close()
+    scheduler.runDueTimers()
+    let deletes = socket.messages.filter { $0.hasPrefix(#"["deleteClients""#) }
+    #expect(deletes == [#"["deleteClients",{"clientIDs":["\#(first.clientID)"]}]"#])
+    second.close()
+  }
+}