fix: Fix sync once hanging forever when opened again too quickly after closing (#98)

packages/serverpod_offline_sync_client/lib/src/crdt/client_sync.dart

@@ -75,7 +75,12 @@ class CrdtSyncClient {
             mode: CrdtSyncPeerMode.follower,
           )
           .listen(
-            outboundChanges.addIfNotClosed,
+            (event) {
+              outboundChanges.addIfNotClosed(event);
+              if (once && event is CrdtSyncClose) {
+                unawaited(outboundChanges.closeOrSkip());
+              }
+            },
             onDone: doneCompleter.completeOrSkip,
             onError: doneCompleter.completeErrorOrSkip,
             cancelOnError: true,
@@ -117,8 +122,15 @@ extension on StreamController<CrdtSyncStreamEvent> {
   }
 
   Future<void> closeOrSkip() async {
-    if (isClosed) return;
-    await close();
+    if (!isClosed) {
+      unawaited(close());
+    }
+    // The done event is only delivered once the stream has a listener. If the
+    // method stream never listened (e.g. opening it failed), awaiting [done]
+    // would hang forever.
+    if (hasListener) {
+      await done;
+    }
   }
 }
 

test/serverpod_offline_sync_test_server/test/integration/sync/sync_flow_test.dart

@@ -85,6 +85,29 @@ void main() {
           },
         );
 
+        test(
+          'when an unauthenticated client calls syncOnce '
+          'then the unauthorized error is surfaced without waiting for stream teardown.',
+          () async {
+            final unauthenticatedClient = client.Client(
+              'http://localhost:${rawServerSession.server.port}',
+            );
+
+            await expectLater(
+              unauthenticatedClient.crdt
+                  .syncOnce(clientSession)
+                  .timeout(const Duration(seconds: 3)),
+              throwsA(
+                predicate<Object>(
+                  (error) =>
+                      error.toString() ==
+                      'ServerpodClientException: Unauthorized, statusCode = 401',
+                ),
+              ),
+            );
+          },
+        );
+
         test(
           'when client syncOnce is called repeatedly on a reused client '
           'then the rounds complete promptly instead of stalling on teardown.',
@@ -132,6 +155,27 @@ void main() {
           },
         );
 
+        test(
+          'when a running syncContinuously session is cancelled '
+          'then method stream teardown does not close the shared websocket.',
+          () async {
+            final stderr = await captureStderr(() async {
+              final syncSession = testClient.crdt.syncContinuously(clientSession);
+
+              await Future<void>.delayed(const Duration(milliseconds: 300));
+              await syncSession.cancel();
+              await syncSession.done;
+              await Future<void>.delayed(const Duration(milliseconds: 100));
+            });
+
+            expect(stderr, isNot(contains('WebSocketConnectionClosed')));
+            expect(
+              stderr,
+              isNot(contains('Message posted when web socket connection is closed')),
+            );
+          },
+        );
+
         test(
           'when client syncContinuously runs '
           'then neither side keeps sending frames while idle.',
@@ -847,6 +891,26 @@ void main() {
             await expectLater(syncSession.done, completes);
           },
         );
+
+        test(
+          'when a running syncContinuously session is cancelled '
+          'then method stream teardown does not close the shared websocket.',
+          () async {
+            final stderr = await captureStderr(() async {
+              final syncSession = testClient.crdt.syncContinuously(clientSession);
+
+              await syncSession.cancel();
+              await syncSession.done;
+              await Future<void>.delayed(const Duration(milliseconds: 100));
+            });
+
+            expect(stderr, isNot(contains('WebSocketConnectionClosed')));
+            expect(
+              stderr,
+              isNot(contains('Message posted when web socket connection is closed')),
+            );
+          },
+        );
       });
 
       group('and an inserted server person', () {