Don't block disconnect on pending checkpoint requests

packages/powersync/CHANGELOG.md

@@ -2,6 +2,8 @@
 
 - Update PowerSync SQLite core extension to 0.5.4 ([release notes](https://github.com/powersync-ja/powersync-sqlite-core/releases/tag/v0.5.4)).
 - The DevTools extension is now based on the shared [PowerSync DevTools](https://releases.powersync.com/announcements/powersync-devtools-inspect-your-app-s-live-database). This improves the SQL inspector, reports per-bucket progress, allows subscribing to Sync Streams and has a cleaner UI.
+- Fix `disconnect()`, `disconnectAndClear()` and `close()` hanging while a `requestCheckpoint()` call waits for an
+  unresponsive service. Pending checkpoint requests now fail with a `CheckpointRequestException` on disconnect.
 
 ## 2.5.0
 

packages/powersync/lib/src/sync/connection_manager.dart

@@ -306,13 +306,15 @@ final class ConnectionManager {
   }
 
   Future<CheckpointRequest> requestCheckpoint() async {
-    return _activeGroup.syncConnectMutex.lock(() async {
-      final client = _abortActiveSync?.$2;
+    // Only hold the lock to read the active client. Awaiting the request while
+    // holding it would block disconnect() until the service responds.
+    final client = await _activeGroup.syncConnectMutex.lock(() async {
       checkConnectedWithRequestsMode();
-
-      final checkpoint = await client!.requestCheckpoint();
-      return CheckpointRequestImpl(checkpoint, this);
+      return _abortActiveSync!.$2;
     });
+
+    final checkpoint = await client.requestCheckpoint();
+    return CheckpointRequestImpl(checkpoint, this);
   }
 
   void close() {

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

@@ -184,9 +184,22 @@ class StreamingSyncImplementation implements StreamingSync {
       throw checkpoint.disabled;
     }
 
-    final neverAbort = AbortController();
-    final checkpointId = await _requestNextCheckpointFromService(neverAbort);
-    return Int64.parse(checkpointId);
+    // Tie the request to the active sync iteration, so that disconnecting
+    // aborts it instead of waiting for the service to respond.
+    final abort = _abort;
+    if (abort == null || abort.aborted) {
+      throw checkpoint.disconnected;
+    }
+
+    try {
+      final checkpointId = await _requestNextCheckpointFromService(abort);
+      return Int64.parse(checkpointId);
+    } catch (e) {
+      if (abort.isAbortException(e)) {
+        throw checkpoint.disconnected;
+      }
+      rethrow;
+    }
   }
 
   Future<String> _resolveClientId() async {

packages/powersync/test/sync/in_memory_sync_test.dart

@@ -1307,6 +1307,34 @@ void _declareTests(String name, bool bson) {
           abort.complete();
           await wasAborted;
         });
+
+        test('disconnect aborts pending requests', () async {
+          await waitForConnection(options: options);
+          await syncService.waitForCheckpointRequest(
+            () => syncService.lastCheckpointRequest == 1,
+          );
+
+          final requestReceived = Completer<void>();
+          final stalledResponse = Completer<void>();
+          syncService.beforeCheckpointRequestResponse = () {
+            if (!requestReceived.isCompleted) requestReceived.complete();
+            // Simulate a service that accepts the request but doesn't respond.
+            return stalledResponse.future;
+          };
+
+          final failureExpectation = expectLater(
+            database.requestCheckpoint(),
+            throwsA(disconnected),
+          );
+          await requestReceived.future;
+
+          try {
+            await database.disconnect().timeout(const Duration(seconds: 5));
+          } finally {
+            stalledResponse.complete();
+          }
+          await failureExpectation;
+        });
       });
     });