Test disconnect while fetching credentials

Author: simolus3

packages/powersync_core/lib/src/abort_controller.dart

@@ -25,7 +25,7 @@ class AbortController {
       _abortRequested.complete();
     }
 
-    await _abortCompleter.future;
+    await onCompletion;
   }
 
   /// Signal that an abort has completed.

packages/powersync_core/lib/src/database/powersync_db_mixin.dart

@@ -372,7 +372,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
         _abortActiveSync = null;
       } else {
         /// Wait for the abort to complete. Continue updating the sync status after completed
-        await disconnector.onAbort;
+        await disconnector.onCompletion;
       }
     }
   }

packages/powersync_core/test/streaming_sync_test.dart

@@ -9,6 +9,7 @@ import 'package:logging/logging.dart';
 import 'package:powersync_core/powersync_core.dart';
 import 'package:test/test.dart';
 
+import 'server/sync_server/in_memory_sync_server.dart';
 import 'test_server.dart';
 import 'utils/abstract_test_utils.dart';
 import 'utils/test_utils_impl.dart';
@@ -61,6 +62,40 @@ void main() {
       server.close();
     });
 
+    test('can disconnect in fetchCredentials', () async {
+      final service = MockSyncService();
+      final server = await createServer(mockSyncService: service);
+      final ignoreLogger = Logger.detached('powersync.test');
+
+      final pdb =
+          await testUtils.setupPowerSync(path: path, logger: ignoreLogger);
+      pdb.retryDelay = Duration(milliseconds: 50);
+      final connector = TestConnector(expectAsync0(() async {
+        return PowerSyncCredentials(endpoint: server.endpoint, token: 'token');
+      }));
+
+      await pdb.connect(connector: connector);
+      while (server.connectionCount != 1) {
+        await Future<void>.delayed(const Duration(milliseconds: 100));
+      }
+
+      service.addKeepAlive(60);
+
+      final didDisconnect = Completer<void>();
+
+      connector.fetchCredentialsCallback = expectAsync0(() async {
+        Zone.current.parent!.scheduleMicrotask(() {
+          didDisconnect.complete(pdb.disconnect());
+        });
+
+        throw 'deliberate disconnect';
+      });
+
+      service.addKeepAlive(0);
+      await didDisconnect.future;
+      expect(pdb.currentStatus.connected, isFalse);
+    });
+
     test('can connect as initial operation', () async {
       final server = await createServer();
       final ignoreLogger = Logger.detached('powersync.test');

packages/powersync_core/test/test_server.dart

@@ -8,18 +8,21 @@ import 'package:shelf/shelf.dart';
 import 'package:shelf/shelf_io.dart' as shelf_io;
 import 'package:shelf_router/shelf_router.dart';
 
-class TestServer {
+import 'server/sync_server/in_memory_sync_server.dart';
+
+final class TestServer {
   late HttpServer server;
   Router app = Router();
   int maxConnectionCount = 0;
   int tokenExpiresIn;
 
   TestServer({this.tokenExpiresIn = 65});
 
-  Future<void> init() async {
+  Future<void> init({MockSyncService? mockSyncService}) async {
     app.post('/sync/stream', handleSyncStream);
     // Open on an arbitrary open port
-    server = await shelf_io.serve(app.call, 'localhost', 0);
+    server = await shelf_io.serve(
+        mockSyncService?.router.call ?? app.call, 'localhost', 0);
   }
 
   String get endpoint {
@@ -34,6 +37,9 @@ class TestServer {
     return server.connectionsInfo();
   }
 
+  /// The default response if no [MockSyncService] has been passed to [init].
+  ///
+  /// This will emit keepalive messages frequently.
   Future<Response> handleSyncStream(Request request) async {
     maxConnectionCount = max(connectionCount, maxConnectionCount);
 
@@ -61,9 +67,9 @@ class TestServer {
   }
 }
 
-Future<TestServer> createServer() async {
+Future<TestServer> createServer({MockSyncService? mockSyncService}) async {
   var server = TestServer();
-  await server.init();
+  await server.init(mockSyncService: mockSyncService);
   return server;
 }
 

packages/powersync_core/test/utils/abstract_test_utils.dart

@@ -125,21 +125,21 @@ abstract class AbstractTestUtils {
 }
 
 class TestConnector extends PowerSyncBackendConnector {
-  final Future<PowerSyncCredentials> Function() _fetchCredentials;
-  final Future<void> Function(PowerSyncDatabase)? _uploadData;
+  Future<PowerSyncCredentials> Function() fetchCredentialsCallback;
+  Future<void> Function(PowerSyncDatabase)? uploadDataCallback;
 
-  TestConnector(this._fetchCredentials,
+  TestConnector(this.fetchCredentialsCallback,
       {Future<void> Function(PowerSyncDatabase)? uploadData})
-      : _uploadData = uploadData;
+      : uploadDataCallback = uploadData;
 
   @override
   Future<PowerSyncCredentials?> fetchCredentials() {
-    return _fetchCredentials();
+    return fetchCredentialsCallback();
   }
 
   @override
   Future<void> uploadData(PowerSyncDatabase database) async {
-    await _uploadData?.call(database);
+    await uploadDataCallback?.call(database);
   }
 }