Fix connect() resolving early with Rust client

Author: simolus3

.changeset/new-readers-dance.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': patch
+---
+
+Rust sync client: Fix `connect()` resolving before a connection is made.

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -928,11 +928,16 @@ The next upload iteration will be delayed.`);
         }
 
         const info = instruction.UpdateSyncStatus.status;
+        const lastStatus = syncImplementation.syncStatus;
         const coreCompleteSync = info.priority_status.find((s) => s.priority == FULL_SYNC_PRIORITY);
         const completeSync = coreCompleteSync != null ? coreStatusToJs(coreCompleteSync) : null;
 
         syncImplementation.updateSyncStatus({
-          connected: info.connected,
+          // The first update to the connected field should only happen when it has really changed - there's a
+          // statusUpdated listener in connect() that will only make the promise complete once connected has changed.
+          // We only want to apply that workaround while we're connecting because it's only relevant in the initial
+          // connect call. Afterwards, we want to forward the sync status unchanged.
+          connected: lastStatus.connected == info.connected && info.connecting ? undefined : info.connected,
           connecting: info.connecting,
           dataFlow: {
             downloading: info.downloading != null,

packages/node/tests/sync.test.ts

@@ -112,6 +112,27 @@ function defineSyncTests(impl: SyncClientImplementation) {
     connectionMethod: SyncStreamConnectionMethod.HTTP
   };
 
+  mockSyncServiceTest('connect() waits for connection', async ({ syncService }) => {
+    const database = await syncService.createDatabase();
+    let connectCompleted = false;
+    database.connect(new TestConnector(), options).then(() => {
+      connectCompleted = true;
+    });
+    expect(connectCompleted).toBeFalsy();
+
+    await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
+    // Opening the socket is not enough, we set connected: true when the first line is received.
+    expect(connectCompleted).toBeFalsy();
+
+    syncService.pushLine({
+      checkpoint: {
+        last_op_id: '10',
+        buckets: [bucket('a', 10)]
+      }
+    });
+    await vi.waitFor(() => connectCompleted);
+  });
+
   describe('reports progress', () => {
     let lastOpId = 0;