zero-swift: list a group's running clients in activeClients

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