Web: Fix reconnect

Author: simolus3

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

@@ -528,7 +528,8 @@ class StreamingSyncImplementation implements StreamingSync {
   }
 
   Future<http.StreamedResponse?> _postStreamRequest(
-      Object? data, bool acceptBson) async {
+      Object? data, bool acceptBson,
+      {Future<void>? onAbort}) async {
     const ndJson = 'application/x-ndjson';
     const bson = 'application/vnd.powersync.bson-stream';
 
@@ -538,8 +539,8 @@ class StreamingSyncImplementation implements StreamingSync {
     }
     final uri = credentials.endpointUri('sync/stream');
 
-    final request =
-        http.AbortableRequest('POST', uri, abortTrigger: _abort!.onAbort);
+    final request = http.AbortableRequest('POST', uri,
+        abortTrigger: onAbort ?? _abort!.onAbort);
     request.headers['Content-Type'] = 'application/json';
     request.headers['Authorization'] = "Token ${credentials.token}";
     request.headers['Accept'] =
@@ -645,9 +646,10 @@ final class _ActiveRustStreamingIteration {
     }
   }
 
-  Stream<ReceivedLine> _receiveLines(Object? data) {
+  Stream<ReceivedLine> _receiveLines(Object? data,
+      {required Future<void> onAbort}) {
     return streamFromFutureAwaitInCancellation(
-            sync._postStreamRequest(data, true))
+            sync._postStreamRequest(data, true, onAbort: onAbort))
         .asyncExpand<Object /* Uint8List | String */ >((response) {
       if (response == null) {
         return null;
@@ -662,33 +664,66 @@ final class _ActiveRustStreamingIteration {
 
   Future<RustSyncIterationResult> _handleLines(
       EstablishSyncStream request) async {
+    // This is a workaround for https://github.com/dart-lang/http/issues/1820:
+    // When cancelling the stream subscription of an HTTP response with the
+    // fetch-based client implementation, cancelling the subscription is delayed
+    // until the next chunk (typically a token_expires_in message in our case).
+    // So, before cancelling, we complete an abort controller for the request to
+    // speed things up. This is not an issue in most cases because the abort
+    // controller on this stream would be completed when disconnecting. But
+    // when switching sync streams, that's not the case and we need a second
+    // abort controller for the inner iteration.
+    final innerAbort = Completer<void>.sync();
     final events = addBroadcast(
-        _receiveLines(request.request), sync._nonLineSyncEvents.stream);
+      _receiveLines(
+        request.request,
+        onAbort: Future.any([
+          sync._abort!.onAbort,
+          innerAbort.future,
+        ]),
+      ),
+      sync._nonLineSyncEvents.stream,
+    );
 
     var needsImmediateRestart = false;
     loop:
-    await for (final event in events) {
-      if (!_isActive || sync.aborted) {
-        break;
-      }
+    try {
+      await for (final event in events) {
+        if (!_isActive || sync.aborted) {
+          innerAbort.complete();
+          break;
+        }
 
-      switch (event) {
-        case ReceivedLine(line: final Uint8List line):
-          _triggerCrudUploadOnFirstLine();
-          await _control('line_binary', line);
-        case ReceivedLine(line: final line as String):
-          _triggerCrudUploadOnFirstLine();
-          await _control('line_text', line);
-        case UploadCompleted():
-          await _control('completed_upload');
-        case AbortCurrentIteration(:final hideDisconnectState):
-          needsImmediateRestart = hideDisconnectState;
-          break loop;
-        case TokenRefreshComplete():
-          await _control('refreshed_token');
-        case HandleChangedSubscriptions(:final currentSubscriptions):
-          await _control('update_subscriptions',
-              convert.json.encode(_encodeSubscriptions(currentSubscriptions)));
+        switch (event) {
+          case ReceivedLine(line: final Uint8List line):
+            _triggerCrudUploadOnFirstLine();
+            await _control('line_binary', line);
+          case ReceivedLine(line: final line as String):
+            _triggerCrudUploadOnFirstLine();
+            await _control('line_text', line);
+          case UploadCompleted():
+            await _control('completed_upload');
+          case AbortCurrentIteration(:final hideDisconnectState):
+            innerAbort.complete();
+            needsImmediateRestart = hideDisconnectState;
+            break loop;
+          case TokenRefreshComplete():
+            await _control('refreshed_token');
+          case HandleChangedSubscriptions(:final currentSubscriptions):
+            await _control(
+                'update_subscriptions',
+                convert.json
+                    .encode(_encodeSubscriptions(currentSubscriptions)));
+        }
+      }
+    } on http.RequestAbortedException {
+      // Unlike a regular cancellation, cancelling via the abort controller
+      // emits an error. We did mean to just cancel the stream, so we can
+      // safely ignore that.
+      if (innerAbort.isCompleted) {
+        // ignore
+      } else {
+        rethrow;
       }
     }