Apply subscription changes made while the sync stream is being established

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 1.16.2
+
+* Fix sync stream subscriptions changed while the connection was still being established only
+  taking effect on the next keep-alive line from the service, typically 20 seconds later.
+
 ## 1.16.1
 
 * Fix a re-query storm on databases opened at an absolute path (App Group container). The

Sources/PowerSync/CurrentVersion.swift

@@ -1,2 +1,2 @@
 // The current version of the PowerSync Swift SDK. This should be updated to the latest version in `CHANGELOG.md` when a new version is released.
-let libraryVersion = "1.16.1"
+let libraryVersion = "1.16.2"

Sources/PowerSync/Implementation/sync/StreamingSyncClient.swift

@@ -514,6 +514,8 @@ private struct ActiveSyncIteration: Sendable {
             signals.markPendingCheckpointRequestsRequiringAffirmation()
         }
 
+        let pendingLocalEvents = localEvents.subscribe()
+
         // Notify the core extension for changed Sync Stream subscriptions, as we might have to reconnect.
         let (currentStreams, streamChanges) = syncClient.db.group.syncCoordinator.streams.observeActiveStreams()
         async let _ = watchSyncStreams(changes: streamChanges)
@@ -557,7 +559,7 @@ private struct ActiveSyncIteration: Sendable {
                     controlArgs = AsyncAlgorithms.merge(
                         serviceEvents,
                         checkpointRequestStateValidationEvents(task: checkpointRequestStateSeed),
-                        localEvents.subscribe()
+                        pendingLocalEvents
                     )
                 } catch {
                     let streamError = error

Tests/PowerSyncTests/SyncTests.swift

@@ -2192,6 +2192,55 @@ class InMemorySyncIntegrationTests {
         try await db.close()
     }
 
+    @Test func subscriptionAddedWhileConnectingReconnects() async throws {
+        // Regression test: a subscription added after `connect()` returned but before the first
+        // `/sync/stream` response arrived must trigger a reconnect on its own. Nothing here sends a
+        // keep-alive, so the second request can only come from the subscription change.
+        let firstRequestStarted = Signal()
+        let releaseFirstResponse = Signal()
+        let requests = AsyncMutex<[JsonParam]>([])
+        let db = openDatabase(MockHttpSession { request in
+            let body = try StreamingSyncClient.jsonDecoder.decode(JsonParam.self, from: try #require(request.httpBody))
+            let count = await requests.withMutex { requests in
+                requests.append(body)
+                return requests.count
+            }
+            if count == 1 {
+                await firstRequestStarted.complete()
+                await releaseFirstResponse.await()
+            }
+            return AsyncThrowingChannel<Data, any Error>()
+        })
+
+        try await db.connect(connector: TestConnector(), options: ConnectOptions())
+        await firstRequestStarted.await()
+
+        // The first request is in flight and its response has not arrived yet.
+        let subscription = try await db.syncStream(name: "a", params: nil).subscribe()
+        await releaseFirstResponse.complete()
+
+        try await waitUntilAsync { await requests.inner.count == 2 }
+        let allRequests = await requests.inner
+        if case let .object(streams) = allRequests[0]["streams"] {
+            try #require(streams["subscriptions"] == .array([]))
+        } else {
+            Issue.record("Should have streams key in first body")
+        }
+        if case let .object(streams) = allRequests[1]["streams"] {
+            try #require(streams["subscriptions"] == .array([
+                .object([
+                    "stream": .string("a"),
+                    "parameters": .null,
+                    "override_priority": .null,
+                ])
+            ]))
+        } else {
+            Issue.record("Should have streams key in second body")
+        }
+        let _ = consume subscription
+        try await db.close()
+    }
+
     @Test func subscriptionsUpdateWhileOffline() async throws {
         let db = openDatabase(MockHttpSession {
             request in throw PowerSyncError.operationFailed(message: "Unexpected connection", underlyingError: nil)