exit crud loop on disconnect faster

Author: stevensJourney

packages/powersync/lib/src/streaming_sync.dart

@@ -154,7 +154,7 @@ class StreamingSyncImplementation {
 
           // On error, wait a little before retrying
           // When aborting, don't wait
-          await Future.any([Future.delayed(retryDelay), _abort!.onAbort]);
+          await _delayRetry();
         }
       }
     } finally {
@@ -173,11 +173,9 @@ class StreamingSyncImplementation {
     // This has the potential (in rare cases) to affect the crudThrottleTime,
     // but it should not result in excessive uploads since the
     // sync reconnects are also throttled.
+    // The stream here is closed on abort.
     await for (var _ in mergeStreams(
         [crudUpdateTriggerStream, _internalCrudTriggerController.stream])) {
-      if (_abort?.aborted == true) {
-        break;
-      }
       await uploadAllCrud();
     }
   }
@@ -189,6 +187,13 @@ class StreamingSyncImplementation {
 
       while (true) {
         try {
+          // It's possible that an abort or disconnect operation could
+          // be followed by a `close` operation. The close would cause these
+          // operations, which use the DB, to throw an exception. Breaking the loop
+          // here prevents unnecessary potential (caught) exceptions.
+          if (aborted) {
+            break;
+          }
           // This is the first item in the FIFO CRUD queue.
           CrudEntry? nextCrudItem = await adapter.nextCrudItem();
           if (nextCrudItem != null) {
@@ -215,7 +220,7 @@ class StreamingSyncImplementation {
           checkedCrudItem = null;
           isolateLogger.warning('Data upload error', e, stacktrace);
           _updateStatus(uploading: false, uploadError: e);
-          await Future.delayed(retryDelay);
+          await _delayRetry();
           if (!isConnected) {
             // Exit the upload loop if the sync stream is no longer connected
             break;
@@ -487,6 +492,12 @@ class StreamingSyncImplementation {
       yield parseStreamingSyncLine(line as Map<String, dynamic>);
     }
   }
+
+  /// Delays the standard `retryDelay` Duration, but exists early if
+  /// an abort has been requested.
+  Future<void> _delayRetry() async {
+    await Future.any([Future.delayed(retryDelay), _abort!.onAbort]);
+  }
 }
 
 /// Attempt to give a basic summary of the error for cases where the full error

packages/powersync/test/upload_test.dart

@@ -66,7 +66,9 @@ void main() {
 
       powersync =
           await testUtils.setupPowerSync(path: path, logger: testWarningLogger);
-      powersync.retryDelay = Duration(milliseconds: 0);
+      // Use a short retry delay here.
+      // A zero retry delay makes this test unstable, since it expects `2` error logs later.
+      powersync.retryDelay = Duration(milliseconds: 100);
       var connector = TestConnector(credentialsCallback, uploadData);
       powersync.connect(connector: connector);