diff --git a/packages/powersync/CHANGELOG.md b/packages/powersync/CHANGELOG.md index f9250b10..b823b2b5 100644 --- a/packages/powersync/CHANGELOG.md +++ b/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 diff --git a/packages/powersync/lib/src/sync/connection_manager.dart b/packages/powersync/lib/src/sync/connection_manager.dart index 01dc0a63..97d98028 100644 --- a/packages/powersync/lib/src/sync/connection_manager.dart +++ b/packages/powersync/lib/src/sync/connection_manager.dart @@ -306,13 +306,15 @@ final class ConnectionManager { } Future 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() { diff --git a/packages/powersync/lib/src/sync/streaming_sync.dart b/packages/powersync/lib/src/sync/streaming_sync.dart index 3a4cf66c..e52ae8bc 100644 --- a/packages/powersync/lib/src/sync/streaming_sync.dart +++ b/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 _resolveClientId() async { diff --git a/packages/powersync/test/sync/in_memory_sync_test.dart b/packages/powersync/test/sync/in_memory_sync_test.dart index 52cba6a7..43482e6a 100644 --- a/packages/powersync/test/sync/in_memory_sync_test.dart +++ b/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(); + final stalledResponse = Completer(); + 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; + }); }); });