Mark connected before first line

Author: simolus3

packages/powersync_core/lib/src/sync/stream_utils.dart

@@ -115,7 +115,7 @@ Future<void> cancelAll(List<StreamSubscription<void>> subscriptions) async {
 /// await the original future and report errors.
 ///
 /// When using the regular [Stream.fromFuture], cancelling the subscription
-/// before the future completes with an error could cause an handled error to
+/// before the future completes with an error could cause an unhandled error to
 /// be reported.
 /// Further, it could cause concurrency issues in the stream client because it
 /// was possible for us to:

packages/powersync_core/lib/src/sync/streaming_sync.dart

@@ -470,6 +470,7 @@ class StreamingSyncImplementation implements StreamingSync {
           await handleLine(line as StreamingSyncLine);
         case UploadCompleted():
         case HandleChangedSubscriptions():
+        case ConnectionEvent():
           // Only relevant for the Rust sync implementation.
           break;
         case AbortCurrentIteration():
@@ -647,20 +648,24 @@ final class _ActiveRustStreamingIteration {
     }
   }
 
-  Stream<ReceivedLine> _receiveLines(Object? data,
+  Stream<SyncEvent> _receiveLines(Object? data,
       {required Future<void> onAbort}) {
     return streamFromFutureAwaitInCancellation(
             sync._postStreamRequest(data, true, onAbort: onAbort))
-        .asyncExpand<Object /* Uint8List | String */ >((response) {
+        .asyncExpand<SyncEvent>((response) async* {
       if (response == null) {
-        return null;
+        return;
       } else {
+        yield ConnectionEvent.established;
+
         final contentType = response.headers['content-type'];
         final isBson = contentType == 'application/vnd.powersync.bson-stream';
 
-        return isBson ? response.stream.bsonDocuments : response.stream.lines;
+        yield* (isBson ? response.stream.bsonDocuments : response.stream.lines)
+            .map(ReceivedLine.new);
+        yield ConnectionEvent.end;
       }
-    }).map(ReceivedLine.new);
+    });
   }
 
   Future<RustSyncIterationResult> _handleLines(
@@ -696,6 +701,8 @@ final class _ActiveRustStreamingIteration {
         }
 
         switch (event) {
+          case ConnectionEvent():
+            await _control('connection', event.name);
           case ReceivedLine(line: final Uint8List line):
             _triggerCrudUploadOnFirstLine();
             await _control('line_binary', line);
@@ -802,6 +809,11 @@ typedef RustSyncIterationResult = ({bool immediateRestart});
 
 sealed class SyncEvent {}
 
+enum ConnectionEvent implements SyncEvent {
+  established,
+  end,
+}
+
 final class ReceivedLine implements SyncEvent {
   final Object /* String|Uint8List|StreamingSyncLine */ line;
 

packages/powersync_core/test/connected_test.dart

@@ -98,23 +98,9 @@ void main() {
       // Create a new completer to await the next upload
       uploadTriggeredCompleter = Completer();
 
-      // Connect the PowerSync instance
-      final connectedCompleter = Completer<void>();
-      // The first connection attempt will fail
-      final connectedErroredCompleter = Completer<void>();
-
-      db.statusStream.listen((status) {
-        if (status.connected && !connectedCompleter.isCompleted) {
-          connectedCompleter.complete();
-        }
-        if (status.downloadError != null &&
-            !connectedErroredCompleter.isCompleted) {
-          connectedErroredCompleter.complete();
-        }
-      });
-
       // The first command will not be valid, this simulates a failed connection
       testServer.addEvent('asdf\n');
+      // Connect the PowerSync instance. The first connection attempt will fail
       await db.connect(connector: connector);
 
       // The connect operation should have triggered an upload (even though it fails to connect)
@@ -124,14 +110,20 @@ void main() {
       uploadTriggeredCompleter = Completer();
 
       // Connection attempt should initially fail
-      await connectedErroredCompleter.future;
-      expect(db.currentStatus.anyError, isNotNull);
+      await expectLater(
+        db.statusStream,
+        emitsThrough(isA<SyncStatus>()
+            .having((e) => e.downloadError, 'downloadError', isNotNull)),
+      );
 
       // Now send a valid command. Which will result in successful connection
       await testServer.clearEvents();
       testServer.addEvent('{"token_expires_in": 3600}\n');
-      await connectedCompleter.future;
-      expect(db.connected, isTrue);
+      await expectLater(
+        db.statusStream,
+        emitsThrough(
+            isA<SyncStatus>().having((e) => e.connected, 'connected', isTrue)),
+      );
 
       await uploadTriggeredCompleter.future;
       expect(uploadCounter, equals(2));

packages/powersync_core/test/sync/in_memory_sync_test.dart

@@ -94,7 +94,7 @@ void _declareTests(String name, SyncOptions options, bool bson) {
     });
 
     Future<StreamQueue<SyncStatus>> waitForConnection(
-        {bool expectNoWarnings = true}) async {
+        {bool expectNoWarnings = true, bool addKeepLive = true}) async {
       if (expectNoWarnings) {
         logger.onRecord.listen((e) {
           if (e.level >= Level.WARNING) {
@@ -311,6 +311,11 @@ void _declareTests(String name, SyncOptions options, bool bson) {
 
         await expectLater(query, emits(isEmpty));
       });
+
+      test('marks as connected even without sync line', () async {
+        await waitForConnection(addKeepLive: false);
+        expect(database.currentStatus.connected, isTrue);
+      });
     }
 
     group('partial sync', () {