From a427124ac9ff5574119a611e6934019438f77bc1 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Tue, 29 Oct 2024 19:10:30 +0200
Subject: [PATCH 01/16] trigger crud upload on reconnect

---
 packages/powersync/lib/src/streaming_sync.dart | 16 +++++++++++++++-
 1 file changed, 15 insertions(+), 1 deletion(-)

diff --git a/packages/powersync/lib/src/streaming_sync.dart b/packages/powersync/lib/src/streaming_sync.dart
index 8f2515fa..fc39c1f8 100644
--- a/packages/powersync/lib/src/streaming_sync.dart
+++ b/packages/powersync/lib/src/streaming_sync.dart
@@ -1,6 +1,7 @@
 import 'dart:async';
 import 'dart:convert' as convert;
 
+import 'package:async/async.dart';
 import 'package:http/http.dart' as http;
 import 'package:powersync/src/abort_controller.dart';
 import 'package:powersync/src/exceptions.dart';
@@ -27,6 +28,11 @@ class StreamingSyncImplementation {
 
   final Future<void> Function() uploadCrud;
 
+  // An internal controller which is used to trigger CRUD uploads internally
+  // e.g. when reconnecting.
+  final StreamController<Null> _internalCrudTriggerController =
+      StreamController<Null>.broadcast();
+
   final Stream crudUpdateTriggerStream;
 
   final StreamController<SyncStatus> _statusStreamController =
@@ -83,6 +89,7 @@ class StreamingSyncImplementation {
     // However, we still need to close the underlying stream explicitly, otherwise
     // the break will wait for the next line of data received on the stream.
     _localPingController.add(null);
+    await _internalCrudTriggerController.close();
     // According to the documentation, the behavior is undefined when calling
     // close() while requests are pending. However, this is no other
     // known way to cancel open streams, and this appears to end the stream with
@@ -92,6 +99,9 @@ class StreamingSyncImplementation {
     if (_safeToClose) {
       _client.close();
     }
+
+    await _internalCrudTriggerController.close();
+
     // wait for completeAbort() to be called
     await future;
 
@@ -155,7 +165,8 @@ class StreamingSyncImplementation {
   Future<void> crudLoop() async {
     await uploadAllCrud();
 
-    await for (var _ in crudUpdateTriggerStream) {
+    await for (var _ in StreamGroup.merge(
+        [crudUpdateTriggerStream, _internalCrudTriggerController.stream])) {
       if (_abort?.aborted == true) {
         break;
       }
@@ -298,6 +309,9 @@ class StreamingSyncImplementation {
     Future<void>? credentialsInvalidation;
     bool haveInvalidated = false;
 
+    // Trigger a CRUD upload on reconnect
+    _internalCrudTriggerController.add(null);
+
     await for (var line in merged) {
       if (aborted) {
         break;

From ce5d99ae767cb017d8d1118b9882a7d40c51b791 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Tue, 29 Oct 2024 19:10:56 +0200
Subject: [PATCH 02/16] wip: mock server

---
 packages/powersync/test/connected_test.dart   | 63 +++++++++++++++++++
 .../server/sync_server/mock_sync_server.dart  | 57 +++++++++++++++++
 2 files changed, 120 insertions(+)
 create mode 100644 packages/powersync/test/connected_test.dart
 create mode 100644 packages/powersync/test/server/sync_server/mock_sync_server.dart

diff --git a/packages/powersync/test/connected_test.dart b/packages/powersync/test/connected_test.dart
new file mode 100644
index 00000000..91875cf0
--- /dev/null
+++ b/packages/powersync/test/connected_test.dart
@@ -0,0 +1,63 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:http/http.dart' as http;
+import 'package:test/test.dart';
+
+import 'server/sync_server/mock_sync_server.dart';
+
+void main() {
+  late TestHttpServerHelper testServer;
+
+  setUp(() async {
+    testServer = TestHttpServerHelper();
+    await testServer.start();
+  });
+
+  tearDown(() async {
+    await testServer.stop();
+  });
+
+  test('should receive events from the sync stream without waiting for close',
+      () async {
+    final client = http.Client();
+    final request =
+        http.Request('POST', testServer.uri.replace(path: '/sync/stream'));
+    request.headers['Content-Type'] = 'application/json';
+
+    // Send the request and get the response stream
+    final responseStream = await client.send(request);
+
+    final expectedEvents = ['event1', 'event2', 'event3'];
+    final receivedEvents = <String>[];
+    final completer = Completer<void>();
+
+    // Listen to the response stream for real-time processing of incoming events
+    final subscription = responseStream.stream
+        .transform(utf8.decoder)
+        .transform(LineSplitter())
+        .listen(
+      (event) {
+        receivedEvents.add(event);
+        if (receivedEvents.length == expectedEvents.length) {
+          completer.complete(); // Complete once all events are received
+        }
+      },
+      onError: (e) => completer.completeError(e),
+    );
+
+    // Programmatically trigger events on the server
+    for (final event in expectedEvents) {
+      testServer.addEvent('$event\n');
+      await Future.delayed(
+          Duration(milliseconds: 100)); // Small delay for each event
+    }
+
+    // Wait for the events to be received
+    await completer.future.timeout(Duration(seconds: 5));
+    await subscription.cancel();
+    client.close();
+
+    expect(receivedEvents.toSet().containsAll(expectedEvents.toSet()), isTrue);
+  });
+}
diff --git a/packages/powersync/test/server/sync_server/mock_sync_server.dart b/packages/powersync/test/server/sync_server/mock_sync_server.dart
new file mode 100644
index 00000000..9effd01c
--- /dev/null
+++ b/packages/powersync/test/server/sync_server/mock_sync_server.dart
@@ -0,0 +1,57 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:shelf/shelf.dart';
+import 'package:shelf/shelf_io.dart' as io;
+import 'package:shelf_router/shelf_router.dart';
+
+class TestHttpServerHelper {
+  final StreamController<String> _controller = StreamController<String>();
+  late HttpServer _server;
+  // late Timer _timer;
+
+  Uri get uri => Uri.parse('http://localhost:${_server.port}');
+
+  // Start the HTTP server
+  Future<void> start() async {
+    final router = Router()
+      ..post('/sync/stream', (Request request) async {
+        // Respond immediately with a stream
+        return Response.ok(_controller.stream.transform(utf8.encoder),
+            headers: {
+              'Content-Type': 'text/event-stream',
+              'Cache-Control': 'no-cache',
+              'Connection': 'keep-alive',
+              'Transfer-Encoding': 'identity', // Use chunked transfer encoding
+            },
+            context: {
+              "shelf.io.buffer_output": false
+            });
+      });
+
+    // Sending a newline gets the stream going and resolving on the client side
+    _controller.add("\n");
+
+    // // Add data. The mock stream does not seem to resolve without data
+    // _timer = Timer.periodic(Duration(seconds: 1), (Timer timer) {
+    //   // This code will execute every second
+    //   _controller.add('{ "token_expires_in": 3600}\n');
+    // });
+
+    _server = await io.serve(router, 'localhost', 0);
+    print('Test server running at ${_server.address}:${_server.port}');
+  }
+
+  // Programmatically add data to the stream
+  void addEvent(String data) {
+    _controller.add(data);
+  }
+
+  // Stop the HTTP server
+  Future<void> stop() async {
+    // _timer.cancel();
+    await _controller.close();
+    await _server.close();
+  }
+}

From 0267e98b2263189957b25da469302e83eb2f2f9c Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 09:43:32 +0200
Subject: [PATCH 03/16] add native tests

---
 packages/powersync/test/connected_test.dart   | 149 +++++++++++++-----
 .../server/sync_server/mock_sync_server.dart  |  29 ++--
 .../powersync/test/streaming_sync_test.dart   |  11 +-
 3 files changed, 128 insertions(+), 61 deletions(-)

diff --git a/packages/powersync/test/connected_test.dart b/packages/powersync/test/connected_test.dart
index 91875cf0..2e29d713 100644
--- a/packages/powersync/test/connected_test.dart
+++ b/packages/powersync/test/connected_test.dart
@@ -1,63 +1,126 @@
 import 'dart:async';
-import 'dart:convert';
 
-import 'package:http/http.dart' as http;
+import 'package:powersync/powersync.dart';
 import 'package:test/test.dart';
 
 import 'server/sync_server/mock_sync_server.dart';
+import 'streaming_sync_test.dart';
+import 'utils/abstract_test_utils.dart';
+import 'utils/test_utils_impl.dart';
+
+final testUtils = TestUtils();
 
 void main() {
   late TestHttpServerHelper testServer;
+  late String path;
 
   setUp(() async {
+    path = testUtils.dbPath();
     testServer = TestHttpServerHelper();
     await testServer.start();
   });
 
   tearDown(() async {
+    await testUtils.cleanDb(path: path);
     await testServer.stop();
   });
 
-  test('should receive events from the sync stream without waiting for close',
-      () async {
-    final client = http.Client();
-    final request =
-        http.Request('POST', testServer.uri.replace(path: '/sync/stream'));
-    request.headers['Content-Type'] = 'application/json';
-
-    // Send the request and get the response stream
-    final responseStream = await client.send(request);
-
-    final expectedEvents = ['event1', 'event2', 'event3'];
-    final receivedEvents = <String>[];
-    final completer = Completer<void>();
-
-    // Listen to the response stream for real-time processing of incoming events
-    final subscription = responseStream.stream
-        .transform(utf8.decoder)
-        .transform(LineSplitter())
-        .listen(
-      (event) {
-        receivedEvents.add(event);
-        if (receivedEvents.length == expectedEvents.length) {
-          completer.complete(); // Complete once all events are received
-        }
-      },
-      onError: (e) => completer.completeError(e),
-    );
-
-    // Programmatically trigger events on the server
-    for (final event in expectedEvents) {
-      testServer.addEvent('$event\n');
-      await Future.delayed(
-          Duration(milliseconds: 100)); // Small delay for each event
-    }
-
-    // Wait for the events to be received
-    await completer.future.timeout(Duration(seconds: 5));
-    await subscription.cancel();
-    client.close();
-
-    expect(receivedEvents.toSet().containsAll(expectedEvents.toSet()), isTrue);
+  test('should connect to mock PowerSync instance', () async {
+    final connector = TestConnector(() async {
+      return PowerSyncCredentials(
+          endpoint: testServer.uri.toString(),
+          token: 'token not used here',
+          expiresAt: DateTime.now());
+    });
+
+    final db = PowerSyncDatabase.withFactory(
+        await testUtils.testFactory(path: path),
+        schema: defaultSchema,
+        maxReaders: 3);
+    await db.initialize();
+
+    final connectedCompleter = Completer();
+
+    db.statusStream.listen((status) {
+      if (status.connected) {
+        connectedCompleter.complete();
+      }
+    });
+
+    // Add a basic command for the test server to send
+    testServer.addEvent('{"token_expires_in": 3600}\n');
+
+    await db.connect(connector: connector);
+    await connectedCompleter.future;
+
+    expect(db.connected, isTrue);
+  });
+
+  test('should trigger uploads when connection is established', () async {
+    int uploadCounter = 0;
+    Completer uploadTriggeredCompleter = Completer();
+
+    final connector = TestConnector(() async {
+      return PowerSyncCredentials(
+          endpoint: testServer.uri.toString(),
+          token: 'token not used here',
+          expiresAt: DateTime.now());
+    }, uploadData: (database) async {
+      uploadCounter++;
+      uploadTriggeredCompleter.complete();
+      throw Exception('No uploads occur here');
+    });
+
+    final db = PowerSyncDatabase.withFactory(
+        await testUtils.testFactory(path: path),
+        schema: defaultSchema,
+        maxReaders: 3);
+    await db.initialize();
+
+    // Create an item which should trigger an upload.
+    await db.execute(
+        'INSERT INTO customers (id, name) VALUES (uuid(), ?)', ['steven']);
+
+    // Create a new completer to await the next upload
+    uploadTriggeredCompleter = Completer();
+
+    // Connect the PowerSync instance
+    final connectedCompleter = Completer();
+    // The first connection attempt will fail
+    final connectedErroredCompleter = Completer();
+
+    db.statusStream.listen((status) {
+      // print('status updated: ${status.connected}, ${status.downloadError}');
+      if (status.connected) {
+        connectedCompleter.complete();
+      }
+      if (status.downloadError != null &&
+          !connectedErroredCompleter.isCompleted) {
+        connectedErroredCompleter.complete();
+      }
+    });
+
+    // The first command will not be valid, this simulates a failed connection
+    testServer.addEvent('asdf\n');
+    await db.connect(connector: connector);
+
+    // The connect operation should have triggered an upload (even though it fails to connect)
+    await uploadTriggeredCompleter.future;
+    expect(uploadCounter, equals(1));
+    // Create a new completer for the next iteration
+    uploadTriggeredCompleter = Completer();
+
+    // Connection attempt should initially fail
+    await connectedErroredCompleter.future;
+    expect(db.currentStatus.anyError, isNotNull);
+
+    // Now send a valid command. Which will result in successful connection
+    await testServer.clearEvents();
+    testServer.addEvent('{"token_expires_in": 3600}\n');
+    await connectedCompleter.future;
+    expect(db.connected, isTrue);
+
+    await uploadTriggeredCompleter.future;
+    expect(uploadCounter, equals(2));
   });
 }
diff --git a/packages/powersync/test/server/sync_server/mock_sync_server.dart b/packages/powersync/test/server/sync_server/mock_sync_server.dart
index 9effd01c..f1e90141 100644
--- a/packages/powersync/test/server/sync_server/mock_sync_server.dart
+++ b/packages/powersync/test/server/sync_server/mock_sync_server.dart
@@ -6,14 +6,15 @@ import 'package:shelf/shelf.dart';
 import 'package:shelf/shelf_io.dart' as io;
 import 'package:shelf_router/shelf_router.dart';
 
+// A basic Mock PowerSync service server which queues commands
+// which clients can receive via connecting to the `/sync/stream` route.
+// This assumes only one client will ever be connected at a time.
 class TestHttpServerHelper {
-  final StreamController<String> _controller = StreamController<String>();
+  // Use a queued stream to make tests easier.
+  StreamController<String> _controller = StreamController<String>();
   late HttpServer _server;
-  // late Timer _timer;
-
   Uri get uri => Uri.parse('http://localhost:${_server.port}');
 
-  // Start the HTTP server
   Future<void> start() async {
     final router = Router()
       ..post('/sync/stream', (Request request) async {
@@ -30,27 +31,23 @@ class TestHttpServerHelper {
             });
       });
 
-    // Sending a newline gets the stream going and resolving on the client side
-    _controller.add("\n");
-
-    // // Add data. The mock stream does not seem to resolve without data
-    // _timer = Timer.periodic(Duration(seconds: 1), (Timer timer) {
-    //   // This code will execute every second
-    //   _controller.add('{ "token_expires_in": 3600}\n');
-    // });
-
     _server = await io.serve(router, 'localhost', 0);
     print('Test server running at ${_server.address}:${_server.port}');
   }
 
-  // Programmatically add data to the stream
+  // Queue events which will be sent to connected clients.
   void addEvent(String data) {
     _controller.add(data);
   }
 
-  // Stop the HTTP server
+  // Clear events. We rely on a buffered controller here. Create a new controller
+  // in order to clear the buffer.
+  Future<void> clearEvents() async {
+    await _controller.close();
+    _controller = StreamController<String>();
+  }
+
   Future<void> stop() async {
-    // _timer.cancel();
     await _controller.close();
     await _server.close();
   }
diff --git a/packages/powersync/test/streaming_sync_test.dart b/packages/powersync/test/streaming_sync_test.dart
index 6d1e9e0d..0b5df6d6 100644
--- a/packages/powersync/test/streaming_sync_test.dart
+++ b/packages/powersync/test/streaming_sync_test.dart
@@ -13,8 +13,11 @@ final testUtils = TestUtils();
 
 class TestConnector extends PowerSyncBackendConnector {
   final Function _fetchCredentials;
+  final Future<void> Function(PowerSyncDatabase)? _uploadData;
 
-  TestConnector(this._fetchCredentials);
+  TestConnector(this._fetchCredentials,
+      {Future<void> Function(PowerSyncDatabase)? uploadData})
+      : _uploadData = uploadData;
 
   @override
   Future<PowerSyncCredentials?> fetchCredentials() {
@@ -22,7 +25,11 @@ class TestConnector extends PowerSyncBackendConnector {
   }
 
   @override
-  Future<void> uploadData(PowerSyncDatabase database) async {}
+  Future<void> uploadData(PowerSyncDatabase database) async {
+    if (_uploadData != null) {
+      await _uploadData(database);
+    }
+  }
 }
 
 void main() {

From b07b034aa642410d3853212c3212b34e59a098e5 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 09:49:32 +0200
Subject: [PATCH 04/16] skip web test

---
 packages/powersync/test/connected_test.dart | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/packages/powersync/test/connected_test.dart b/packages/powersync/test/connected_test.dart
index 2e29d713..de5bc4eb 100644
--- a/packages/powersync/test/connected_test.dart
+++ b/packages/powersync/test/connected_test.dart
@@ -1,3 +1,6 @@
+@TestOn('!browser')
+// This test uses a local server which is possible to control in Web via hybrid main,
+// but this makes the test significantly more complex.
 import 'dart:async';
 
 import 'package:powersync/powersync.dart';

From 4df902cd07e0e3f383f462679c7d23fc8b0745cd Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 09:56:49 +0200
Subject: [PATCH 05/16] cleanup

---
 packages/powersync/lib/src/streaming_sync.dart | 1 -
 packages/powersync/test/connected_test.dart    | 1 -
 2 files changed, 2 deletions(-)

diff --git a/packages/powersync/lib/src/streaming_sync.dart b/packages/powersync/lib/src/streaming_sync.dart
index fc39c1f8..6f8cd78e 100644
--- a/packages/powersync/lib/src/streaming_sync.dart
+++ b/packages/powersync/lib/src/streaming_sync.dart
@@ -89,7 +89,6 @@ class StreamingSyncImplementation {
     // However, we still need to close the underlying stream explicitly, otherwise
     // the break will wait for the next line of data received on the stream.
     _localPingController.add(null);
-    await _internalCrudTriggerController.close();
     // According to the documentation, the behavior is undefined when calling
     // close() while requests are pending. However, this is no other
     // known way to cancel open streams, and this appears to end the stream with
diff --git a/packages/powersync/test/connected_test.dart b/packages/powersync/test/connected_test.dart
index de5bc4eb..e454e134 100644
--- a/packages/powersync/test/connected_test.dart
+++ b/packages/powersync/test/connected_test.dart
@@ -93,7 +93,6 @@ void main() {
     final connectedErroredCompleter = Completer();
 
     db.statusStream.listen((status) {
-      // print('status updated: ${status.connected}, ${status.downloadError}');
       if (status.connected) {
         connectedCompleter.complete();
       }

From ed582bb78972a9e27878451af84ff00d275fd6b8 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 10:08:49 +0200
Subject: [PATCH 06/16] cleanup

---
 packages/powersync/lib/src/streaming_sync.dart | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/packages/powersync/lib/src/streaming_sync.dart b/packages/powersync/lib/src/streaming_sync.dart
index 6f8cd78e..021971a7 100644
--- a/packages/powersync/lib/src/streaming_sync.dart
+++ b/packages/powersync/lib/src/streaming_sync.dart
@@ -30,6 +30,8 @@ class StreamingSyncImplementation {
 
   // An internal controller which is used to trigger CRUD uploads internally
   // e.g. when reconnecting.
+  // This is only a broadcast stream since the `crudLoop` method is public
+  // and could potentially be called multiple times externally.
   final StreamController<Null> _internalCrudTriggerController =
       StreamController<Null>.broadcast();
 
@@ -164,6 +166,11 @@ class StreamingSyncImplementation {
   Future<void> crudLoop() async {
     await uploadAllCrud();
 
+    // Trigger a CRUD upload whenever the upstream trigger fires
+    // as-well-as whenever the sync stream reconnects.
+    // 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.
     await for (var _ in StreamGroup.merge(
         [crudUpdateTriggerStream, _internalCrudTriggerController.stream])) {
       if (_abort?.aborted == true) {

From dd32fdfe0668135cccdcb22084b3ae9ed1a5ab4a Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 10:10:35 +0200
Subject: [PATCH 07/16] more cleanup

---
 packages/powersync/lib/src/streaming_sync.dart | 2 +-
 packages/powersync/test/connected_test.dart    | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/powersync/lib/src/streaming_sync.dart b/packages/powersync/lib/src/streaming_sync.dart
index 021971a7..24c3fb10 100644
--- a/packages/powersync/lib/src/streaming_sync.dart
+++ b/packages/powersync/lib/src/streaming_sync.dart
@@ -30,7 +30,7 @@ class StreamingSyncImplementation {
 
   // An internal controller which is used to trigger CRUD uploads internally
   // e.g. when reconnecting.
-  // This is only a broadcast stream since the `crudLoop` method is public
+  // This is only a broadcast controller since the `crudLoop` method is public
   // and could potentially be called multiple times externally.
   final StreamController<Null> _internalCrudTriggerController =
       StreamController<Null>.broadcast();
diff --git a/packages/powersync/test/connected_test.dart b/packages/powersync/test/connected_test.dart
index e454e134..5d31bb61 100644
--- a/packages/powersync/test/connected_test.dart
+++ b/packages/powersync/test/connected_test.dart
@@ -59,7 +59,7 @@ void main() {
     expect(db.connected, isTrue);
   });
 
-  test('should trigger uploads when connection is established', () async {
+  test('should trigger uploads when connection is re-established', () async {
     int uploadCounter = 0;
     Completer uploadTriggeredCompleter = Completer();
 

From 5e359078e5cae7fe3d159b41c3660d980f211c33 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 10:22:43 +0200
Subject: [PATCH 08/16] remove comment

---
 .../powersync/test/server/sync_server/mock_sync_server.dart     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/powersync/test/server/sync_server/mock_sync_server.dart b/packages/powersync/test/server/sync_server/mock_sync_server.dart
index f1e90141..473d6a6e 100644
--- a/packages/powersync/test/server/sync_server/mock_sync_server.dart
+++ b/packages/powersync/test/server/sync_server/mock_sync_server.dart
@@ -24,7 +24,7 @@ class TestHttpServerHelper {
               'Content-Type': 'text/event-stream',
               'Cache-Control': 'no-cache',
               'Connection': 'keep-alive',
-              'Transfer-Encoding': 'identity', // Use chunked transfer encoding
+              'Transfer-Encoding': 'identity',
             },
             context: {
               "shelf.io.buffer_output": false

From fa49b64eb8d34722c8fdeb93b85ef44239e939c2 Mon Sep 17 00:00:00 2001
From: stevensJourney <51082125+stevensJourney@users.noreply.github.com>
Date: Wed, 30 Oct 2024 10:35:13 +0200
Subject: [PATCH 09/16] Update packages/powersync/test/streaming_sync_test.dart

Co-authored-by: Mughees Khan <mugikhan@gmail.com>
---
 packages/powersync/test/streaming_sync_test.dart | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/packages/powersync/test/streaming_sync_test.dart b/packages/powersync/test/streaming_sync_test.dart
index 0b5df6d6..92b03dd3 100644
--- a/packages/powersync/test/streaming_sync_test.dart
+++ b/packages/powersync/test/streaming_sync_test.dart
@@ -26,9 +26,7 @@ class TestConnector extends PowerSyncBackendConnector {
 
   @override
   Future<void> uploadData(PowerSyncDatabase database) async {
-    if (_uploadData != null) {
-      await _uploadData(database);
-    }
+    await _uploadData?.call(database);
   }
 }
 

From bf8cd83a7810208d304c77995776d1d04cd07a5d Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 10:43:23 +0200
Subject: [PATCH 10/16] lint fixes. Use inhouse mergeStreams

---
 packages/powersync/lib/src/streaming_sync.dart                 | 3 +--
 .../powersync/test/server/sync_server/mock_sync_server.dart    | 2 +-
 2 files changed, 2 insertions(+), 3 deletions(-)

diff --git a/packages/powersync/lib/src/streaming_sync.dart b/packages/powersync/lib/src/streaming_sync.dart
index 24c3fb10..ac05030b 100644
--- a/packages/powersync/lib/src/streaming_sync.dart
+++ b/packages/powersync/lib/src/streaming_sync.dart
@@ -1,7 +1,6 @@
 import 'dart:async';
 import 'dart:convert' as convert;
 
-import 'package:async/async.dart';
 import 'package:http/http.dart' as http;
 import 'package:powersync/src/abort_controller.dart';
 import 'package:powersync/src/exceptions.dart';
@@ -171,7 +170,7 @@ 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.
-    await for (var _ in StreamGroup.merge(
+    await for (var _ in mergeStreams(
         [crudUpdateTriggerStream, _internalCrudTriggerController.stream])) {
       if (_abort?.aborted == true) {
         break;
diff --git a/packages/powersync/test/server/sync_server/mock_sync_server.dart b/packages/powersync/test/server/sync_server/mock_sync_server.dart
index 473d6a6e..cf0e8d84 100644
--- a/packages/powersync/test/server/sync_server/mock_sync_server.dart
+++ b/packages/powersync/test/server/sync_server/mock_sync_server.dart
@@ -31,7 +31,7 @@ class TestHttpServerHelper {
             });
       });
 
-    _server = await io.serve(router, 'localhost', 0);
+    _server = await io.serve(router.call, 'localhost', 0);
     print('Test server running at ${_server.address}:${_server.port}');
   }
 

From f42b9a8f8ba093a44c321c7a71825e58b016b1c2 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 12:34:35 +0200
Subject: [PATCH 11/16] fix tests

---
 .../powersync/lib/src/streaming_sync.dart     |   3 +
 packages/powersync/test/connected_test.dart   | 219 ++++++++++--------
 2 files changed, 119 insertions(+), 103 deletions(-)

diff --git a/packages/powersync/lib/src/streaming_sync.dart b/packages/powersync/lib/src/streaming_sync.dart
index ac05030b..43651636 100644
--- a/packages/powersync/lib/src/streaming_sync.dart
+++ b/packages/powersync/lib/src/streaming_sync.dart
@@ -158,6 +158,9 @@ class StreamingSyncImplementation {
         }
       }
     } finally {
+      // The client should no longer be connected at this point
+      // Adding this update allows for breaking out of an in-progress updateLoop
+      _updateStatus(connected: false);
       _abort!.completeAbort();
     }
   }
diff --git a/packages/powersync/test/connected_test.dart b/packages/powersync/test/connected_test.dart
index 5d31bb61..8491829c 100644
--- a/packages/powersync/test/connected_test.dart
+++ b/packages/powersync/test/connected_test.dart
@@ -14,115 +14,128 @@ import 'utils/test_utils_impl.dart';
 final testUtils = TestUtils();
 
 void main() {
-  late TestHttpServerHelper testServer;
-  late String path;
-
-  setUp(() async {
-    path = testUtils.dbPath();
-    testServer = TestHttpServerHelper();
-    await testServer.start();
-  });
-
-  tearDown(() async {
-    await testUtils.cleanDb(path: path);
-    await testServer.stop();
-  });
-
-  test('should connect to mock PowerSync instance', () async {
-    final connector = TestConnector(() async {
-      return PowerSyncCredentials(
-          endpoint: testServer.uri.toString(),
-          token: 'token not used here',
-          expiresAt: DateTime.now());
+  group('connected tests', () {
+    late String path;
+    setUp(() async {
+      path = testUtils.dbPath();
     });
 
-    final db = PowerSyncDatabase.withFactory(
-        await testUtils.testFactory(path: path),
-        schema: defaultSchema,
-        maxReaders: 3);
-    await db.initialize();
-
-    final connectedCompleter = Completer();
-
-    db.statusStream.listen((status) {
-      if (status.connected) {
-        connectedCompleter.complete();
-      }
+    tearDown(() async {
+      // await testUtils.cleanDb(path: path);
     });
 
-    // Add a basic command for the test server to send
-    testServer.addEvent('{"token_expires_in": 3600}\n');
-
-    await db.connect(connector: connector);
-    await connectedCompleter.future;
-
-    expect(db.connected, isTrue);
-  });
-
-  test('should trigger uploads when connection is re-established', () async {
-    int uploadCounter = 0;
-    Completer uploadTriggeredCompleter = Completer();
-
-    final connector = TestConnector(() async {
-      return PowerSyncCredentials(
-          endpoint: testServer.uri.toString(),
-          token: 'token not used here',
-          expiresAt: DateTime.now());
-    }, uploadData: (database) async {
-      uploadCounter++;
-      uploadTriggeredCompleter.complete();
-      throw Exception('No uploads occur here');
+    createTestServer() async {
+      final testServer = TestHttpServerHelper();
+      await testServer.start();
+      return testServer;
+    }
+
+    test('should connect to mock PowerSync instance', () async {
+      final testServer = await createTestServer();
+      // Cant do this inside the createTestServer function unfortunately
+      addTearDown(() => testServer.stop());
+
+      final connector = TestConnector(() async {
+        return PowerSyncCredentials(
+            endpoint: testServer.uri.toString(),
+            token: 'token not used here',
+            expiresAt: DateTime.now());
+      });
+
+      final db = PowerSyncDatabase.withFactory(
+          await testUtils.testFactory(path: path),
+          schema: defaultSchema,
+          maxReaders: 3);
+      await db.initialize();
+
+      final connectedCompleter = Completer();
+
+      db.statusStream.listen((status) {
+        if (status.connected) {
+          connectedCompleter.complete();
+        }
+      });
+
+      // Add a basic command for the test server to send
+      testServer.addEvent('{"token_expires_in": 3600}\n');
+
+      await db.connect(connector: connector);
+      await connectedCompleter.future;
+
+      expect(db.connected, isTrue);
+      await db.disconnect();
     });
 
-    final db = PowerSyncDatabase.withFactory(
-        await testUtils.testFactory(path: path),
-        schema: defaultSchema,
-        maxReaders: 3);
-    await db.initialize();
-
-    // Create an item which should trigger an upload.
-    await db.execute(
-        'INSERT INTO customers (id, name) VALUES (uuid(), ?)', ['steven']);
-
-    // Create a new completer to await the next upload
-    uploadTriggeredCompleter = Completer();
-
-    // Connect the PowerSync instance
-    final connectedCompleter = Completer();
-    // The first connection attempt will fail
-    final connectedErroredCompleter = Completer();
-
-    db.statusStream.listen((status) {
-      if (status.connected) {
-        connectedCompleter.complete();
-      }
-      if (status.downloadError != null &&
-          !connectedErroredCompleter.isCompleted) {
-        connectedErroredCompleter.complete();
-      }
+    test('should trigger uploads when connection is re-established', () async {
+      int uploadCounter = 0;
+      Completer uploadTriggeredCompleter = Completer();
+      final testServer = await createTestServer();
+      // Cant do this inside the createTestServer function unfortunately
+      addTearDown(() => testServer.stop());
+
+      final connector = TestConnector(() async {
+        return PowerSyncCredentials(
+            endpoint: testServer.uri.toString(),
+            token: 'token not used here',
+            expiresAt: DateTime.now());
+      }, uploadData: (database) async {
+        uploadCounter++;
+        uploadTriggeredCompleter.complete();
+        throw Exception('No uploads occur here');
+      });
+
+      final db = PowerSyncDatabase.withFactory(
+          await testUtils.testFactory(path: path),
+          schema: defaultSchema,
+          maxReaders: 3);
+      await db.initialize();
+
+      // Create an item which should trigger an upload.
+      await db.execute(
+          'INSERT INTO customers (id, name) VALUES (uuid(), ?)', ['steven']);
+
+      // Create a new completer to await the next upload
+      uploadTriggeredCompleter = Completer();
+
+      // Connect the PowerSync instance
+      final connectedCompleter = Completer();
+      // The first connection attempt will fail
+      final connectedErroredCompleter = Completer();
+
+      db.statusStream.listen((status) {
+        if (status.connected && !connectedCompleter.isCompleted) {
+          connectedCompleter.complete();
+        }
+        if (status.downloadError != null &&
+            !connectedErroredCompleter.isCompleted) {
+          connectedErroredCompleter.complete();
+        }
+      });
+
+      // The first command will not be valid, this simulates a failed connection
+      testServer.addEvent('asdf\n');
+      await db.connect(connector: connector);
+
+      // The connect operation should have triggered an upload (even though it fails to connect)
+      await uploadTriggeredCompleter.future;
+      expect(uploadCounter, equals(1));
+      // Create a new completer for the next iteration
+      uploadTriggeredCompleter = Completer();
+
+      // Connection attempt should initially fail
+      await connectedErroredCompleter.future;
+      expect(db.currentStatus.anyError, isNotNull);
+
+      // Now send a valid command. Which will result in successful connection
+      await testServer.clearEvents();
+      testServer.addEvent('{"token_expires_in": 3600}\n');
+      await connectedCompleter.future;
+      expect(db.connected, isTrue);
+
+      await uploadTriggeredCompleter.future;
+      expect(uploadCounter, equals(2));
+
+      await db.disconnect();
     });
-
-    // The first command will not be valid, this simulates a failed connection
-    testServer.addEvent('asdf\n');
-    await db.connect(connector: connector);
-
-    // The connect operation should have triggered an upload (even though it fails to connect)
-    await uploadTriggeredCompleter.future;
-    expect(uploadCounter, equals(1));
-    // Create a new completer for the next iteration
-    uploadTriggeredCompleter = Completer();
-
-    // Connection attempt should initially fail
-    await connectedErroredCompleter.future;
-    expect(db.currentStatus.anyError, isNotNull);
-
-    // Now send a valid command. Which will result in successful connection
-    await testServer.clearEvents();
-    testServer.addEvent('{"token_expires_in": 3600}\n');
-    await connectedCompleter.future;
-    expect(db.connected, isTrue);
-
-    await uploadTriggeredCompleter.future;
-    expect(uploadCounter, equals(2));
   });
 }

From 0d557cbefd904f99e0b92994a5719b11cedeb733 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 12:37:30 +0200
Subject: [PATCH 12/16] automatic teardown of server

---
 packages/powersync/test/connected_test.dart | 7 +------
 1 file changed, 1 insertion(+), 6 deletions(-)

diff --git a/packages/powersync/test/connected_test.dart b/packages/powersync/test/connected_test.dart
index 8491829c..7825e08a 100644
--- a/packages/powersync/test/connected_test.dart
+++ b/packages/powersync/test/connected_test.dart
@@ -27,14 +27,12 @@ void main() {
     createTestServer() async {
       final testServer = TestHttpServerHelper();
       await testServer.start();
+      addTearDown(() => testServer.stop());
       return testServer;
     }
 
     test('should connect to mock PowerSync instance', () async {
       final testServer = await createTestServer();
-      // Cant do this inside the createTestServer function unfortunately
-      addTearDown(() => testServer.stop());
-
       final connector = TestConnector(() async {
         return PowerSyncCredentials(
             endpoint: testServer.uri.toString(),
@@ -70,9 +68,6 @@ void main() {
       int uploadCounter = 0;
       Completer uploadTriggeredCompleter = Completer();
       final testServer = await createTestServer();
-      // Cant do this inside the createTestServer function unfortunately
-      addTearDown(() => testServer.stop());
-
       final connector = TestConnector(() async {
         return PowerSyncCredentials(
             endpoint: testServer.uri.toString(),

From b7713499aaf3e1583687a8c5d94c8dd001596e57 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Wed, 30 Oct 2024 17:13:46 +0200
Subject: [PATCH 13/16] use test name in logs for better readability

---
 .../powersync/test/utils/abstract_test_utils.dart   | 13 +++++++------
 1 file changed, 7 insertions(+), 6 deletions(-)

diff --git a/packages/powersync/test/utils/abstract_test_utils.dart b/packages/powersync/test/utils/abstract_test_utils.dart
index efae5964..773dafb6 100644
--- a/packages/powersync/test/utils/abstract_test_utils.dart
+++ b/packages/powersync/test/utils/abstract_test_utils.dart
@@ -26,8 +26,8 @@ final testLogger = _makeTestLogger();
 
 final testWarningLogger = _makeTestLogger(level: Level.WARNING);
 
-Logger _makeTestLogger({Level level = Level.ALL}) {
-  final logger = Logger.detached('PowerSync Tests');
+Logger _makeTestLogger({Level level = Level.ALL, String? name}) {
+  final logger = Logger.detached(name ?? 'PowerSync Tests');
   logger.level = level;
   logger.onRecord.listen((record) {
     print(
@@ -53,11 +53,11 @@ Logger _makeTestLogger({Level level = Level.ALL}) {
 }
 
 abstract class AbstractTestUtils {
+  String get _testName => Invoker.current!.liveTest.test.name;
+
   String dbPath() {
-    final test = Invoker.current!.liveTest;
-    var testName = test.test.name;
     var testShortName =
-        testName.replaceAll(RegExp(r'[\s\./]'), '_').toLowerCase();
+        _testName.replaceAll(RegExp(r'[\s\./]'), '_').toLowerCase();
     var dbName = "test-db/$testShortName.db";
     return dbName;
   }
@@ -74,7 +74,8 @@ abstract class AbstractTestUtils {
   Future<PowerSyncDatabase> setupPowerSync(
       {String? path, Schema? schema, Logger? logger}) async {
     final db = PowerSyncDatabase.withFactory(await testFactory(path: path),
-        schema: schema ?? defaultSchema, logger: logger ?? testLogger);
+        schema: schema ?? defaultSchema,
+        logger: logger ?? _makeTestLogger(name: _testName));
     await db.initialize();
     return db;
   }

From 6a4e6500adff26c5ab2f0aa378ae0abd119d4824 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Thu, 31 Oct 2024 09:19:33 +0200
Subject: [PATCH 14/16] exit crud loop on disconnect faster

---
 .../powersync/lib/src/streaming_sync.dart     | 21 ++++++++++++++-----
 packages/powersync/test/upload_test.dart      |  4 +++-
 2 files changed, 19 insertions(+), 6 deletions(-)

diff --git a/packages/powersync/lib/src/streaming_sync.dart b/packages/powersync/lib/src/streaming_sync.dart
index 43651636..4fcb68b4 100644
--- a/packages/powersync/lib/src/streaming_sync.dart
+++ b/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
diff --git a/packages/powersync/test/upload_test.dart b/packages/powersync/test/upload_test.dart
index 20ad04f7..c3fa603e 100644
--- a/packages/powersync/test/upload_test.dart
+++ b/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);
 

From 33399263ed4a417bf2a0a6b742f82983a620dc3e Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Thu, 31 Oct 2024 09:31:20 +0200
Subject: [PATCH 15/16] code cleanup

---
 packages/powersync/lib/src/streaming_sync.dart               | 5 +----
 packages/powersync/test/connected_test.dart                  | 2 +-
 .../powersync/test/server/sync_server/mock_sync_server.dart  | 3 +--
 3 files changed, 3 insertions(+), 7 deletions(-)

diff --git a/packages/powersync/lib/src/streaming_sync.dart b/packages/powersync/lib/src/streaming_sync.dart
index 4fcb68b4..57a5ff5b 100644
--- a/packages/powersync/lib/src/streaming_sync.dart
+++ b/packages/powersync/lib/src/streaming_sync.dart
@@ -158,9 +158,6 @@ class StreamingSyncImplementation {
         }
       }
     } finally {
-      // The client should no longer be connected at this point
-      // Adding this update allows for breaking out of an in-progress updateLoop
-      _updateStatus(connected: false);
       _abort!.completeAbort();
     }
   }
@@ -493,7 +490,7 @@ class StreamingSyncImplementation {
     }
   }
 
-  /// Delays the standard `retryDelay` Duration, but exists early if
+  /// Delays the standard `retryDelay` Duration, but exits early if
   /// an abort has been requested.
   Future<void> _delayRetry() async {
     await Future.any([Future.delayed(retryDelay), _abort!.onAbort]);
diff --git a/packages/powersync/test/connected_test.dart b/packages/powersync/test/connected_test.dart
index 7825e08a..0f6d3b71 100644
--- a/packages/powersync/test/connected_test.dart
+++ b/packages/powersync/test/connected_test.dart
@@ -21,7 +21,7 @@ void main() {
     });
 
     tearDown(() async {
-      // await testUtils.cleanDb(path: path);
+      await testUtils.cleanDb(path: path);
     });
 
     createTestServer() async {
diff --git a/packages/powersync/test/server/sync_server/mock_sync_server.dart b/packages/powersync/test/server/sync_server/mock_sync_server.dart
index cf0e8d84..0f1c8f49 100644
--- a/packages/powersync/test/server/sync_server/mock_sync_server.dart
+++ b/packages/powersync/test/server/sync_server/mock_sync_server.dart
@@ -21,10 +21,9 @@ class TestHttpServerHelper {
         // Respond immediately with a stream
         return Response.ok(_controller.stream.transform(utf8.encoder),
             headers: {
-              'Content-Type': 'text/event-stream',
+              'Content-Type': 'application/x-ndjson',
               'Cache-Control': 'no-cache',
               'Connection': 'keep-alive',
-              'Transfer-Encoding': 'identity',
             },
             context: {
               "shelf.io.buffer_output": false

From c92e7f5f40ea85f8871208b90077d0b1a4a27a99 Mon Sep 17 00:00:00 2001
From: Steven Ontong <steven@journeyapps.com>
Date: Thu, 31 Oct 2024 10:43:49 +0200
Subject: [PATCH 16/16] chore(release): publish packages

 - powersync@1.8.9
 - powersync_attachments_helper@0.6.13
 - powersync_flutter_libs@0.4.2
---
 CHANGELOG.md                                  | 31 +++++++++++++++++++
 demos/django-todolist/pubspec.yaml            |  2 +-
 demos/supabase-anonymous-auth/pubspec.yaml    |  2 +-
 .../supabase-edge-function-auth/pubspec.yaml  |  2 +-
 demos/supabase-simple-chat/pubspec.yaml       |  2 +-
 demos/supabase-todolist-drift/pubspec.yaml    |  4 +--
 .../pubspec.yaml                              |  2 +-
 demos/supabase-todolist/pubspec.yaml          |  4 +--
 packages/powersync/CHANGELOG.md               |  4 +++
 packages/powersync/lib/src/version.dart       |  2 +-
 packages/powersync/pubspec.yaml               |  4 +--
 .../powersync_attachments_helper/CHANGELOG.md |  4 +++
 .../powersync_attachments_helper/pubspec.yaml |  4 +--
 packages/powersync_flutter_libs/CHANGELOG.md  |  4 +++
 packages/powersync_flutter_libs/pubspec.yaml  |  2 +-
 15 files changed, 58 insertions(+), 15 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 69f4ee0e..ce8714ce 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,37 @@
 All notable changes to this project will be documented in this file.
 See [Conventional Commits](https://conventionalcommits.org) for commit guidelines.
 
+## 2024-10-31
+
+### Changes
+
+---
+
+Packages with breaking changes:
+
+ - There are no breaking changes in this release.
+
+Packages with other changes:
+
+ - [`powersync` - `v1.8.9`](#powersync---v189)
+ - [`powersync_attachments_helper` - `v0.6.13`](#powersync_attachments_helper---v0613)
+ - [`powersync_flutter_libs` - `v0.4.2`](#powersync_flutter_libs---v042)
+
+---
+
+#### `powersync` - `v1.8.9`
+
+ - **FIX**: Issue where CRUD uploads were not triggered when the SDK reconnected to the PowerSync service after being offline.
+
+#### `powersync_attachments_helper` - `v0.6.13`
+
+ - Update a dependency to the latest release.
+
+#### `powersync_flutter_libs` - `v0.4.2`
+
+ - Update a dependency to the latest release.
+
+
 ## 2024-10-21
 
 ### Changes
diff --git a/demos/django-todolist/pubspec.yaml b/demos/django-todolist/pubspec.yaml
index 4a9bd9af..317acd12 100644
--- a/demos/django-todolist/pubspec.yaml
+++ b/demos/django-todolist/pubspec.yaml
@@ -10,7 +10,7 @@ environment:
 dependencies:
   flutter:
     sdk: flutter
-  powersync: ^1.8.8
+  powersync: ^1.8.9
   path_provider: ^2.1.1
   path: ^1.8.3
   logging: ^1.2.0
diff --git a/demos/supabase-anonymous-auth/pubspec.yaml b/demos/supabase-anonymous-auth/pubspec.yaml
index 982f10ee..6550e3e1 100644
--- a/demos/supabase-anonymous-auth/pubspec.yaml
+++ b/demos/supabase-anonymous-auth/pubspec.yaml
@@ -11,7 +11,7 @@ dependencies:
   flutter:
     sdk: flutter
 
-  powersync: ^1.8.8
+  powersync: ^1.8.9
   path_provider: ^2.1.1
   supabase_flutter: ^2.0.2
   path: ^1.8.3
diff --git a/demos/supabase-edge-function-auth/pubspec.yaml b/demos/supabase-edge-function-auth/pubspec.yaml
index 6bd0a91b..4a9d4985 100644
--- a/demos/supabase-edge-function-auth/pubspec.yaml
+++ b/demos/supabase-edge-function-auth/pubspec.yaml
@@ -11,7 +11,7 @@ dependencies:
   flutter:
     sdk: flutter
 
-  powersync: ^1.8.8
+  powersync: ^1.8.9
   path_provider: ^2.1.1
   supabase_flutter: ^2.0.2
   path: ^1.8.3
diff --git a/demos/supabase-simple-chat/pubspec.yaml b/demos/supabase-simple-chat/pubspec.yaml
index 9968ad8b..f0a1c9c6 100644
--- a/demos/supabase-simple-chat/pubspec.yaml
+++ b/demos/supabase-simple-chat/pubspec.yaml
@@ -37,7 +37,7 @@ dependencies:
 
   supabase_flutter: ^2.0.2
   timeago: ^3.6.0
-  powersync: ^1.8.8
+  powersync: ^1.8.9
   path_provider: ^2.1.1
   path: ^1.8.3
   logging: ^1.2.0
diff --git a/demos/supabase-todolist-drift/pubspec.yaml b/demos/supabase-todolist-drift/pubspec.yaml
index 630b43a9..bf77f1a9 100644
--- a/demos/supabase-todolist-drift/pubspec.yaml
+++ b/demos/supabase-todolist-drift/pubspec.yaml
@@ -9,8 +9,8 @@ environment:
 dependencies:
   flutter:
     sdk: flutter
-  powersync_attachments_helper: ^0.6.12
-  powersync: ^1.8.8
+  powersync_attachments_helper: ^0.6.13
+  powersync: ^1.8.9
   path_provider: ^2.1.1
   supabase_flutter: ^2.0.1
   path: ^1.8.3
diff --git a/demos/supabase-todolist-optional-sync/pubspec.yaml b/demos/supabase-todolist-optional-sync/pubspec.yaml
index 280d17fb..94f83b9d 100644
--- a/demos/supabase-todolist-optional-sync/pubspec.yaml
+++ b/demos/supabase-todolist-optional-sync/pubspec.yaml
@@ -10,7 +10,7 @@ environment:
 dependencies:
   flutter:
     sdk: flutter
-  powersync: ^1.8.8
+  powersync: ^1.8.9
   path_provider: ^2.1.1
   supabase_flutter: ^2.0.1
   path: ^1.8.3
diff --git a/demos/supabase-todolist/pubspec.yaml b/demos/supabase-todolist/pubspec.yaml
index d4d64314..3e34efb7 100644
--- a/demos/supabase-todolist/pubspec.yaml
+++ b/demos/supabase-todolist/pubspec.yaml
@@ -10,8 +10,8 @@ environment:
 dependencies:
   flutter:
     sdk: flutter
-  powersync_attachments_helper: ^0.6.12
-  powersync: ^1.8.8
+  powersync_attachments_helper: ^0.6.13
+  powersync: ^1.8.9
   path_provider: ^2.1.1
   supabase_flutter: ^2.0.1
   path: ^1.8.3
diff --git a/packages/powersync/CHANGELOG.md b/packages/powersync/CHANGELOG.md
index de2020b5..e8733016 100644
--- a/packages/powersync/CHANGELOG.md
+++ b/packages/powersync/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 1.8.9
+
+ - **FIX**: issue where CRUD uploads were not triggered when the SDK reconnected to the PowerSync service after being offline.
+
 ## 1.8.8
 
 - Update dependency `powersync_flutter_libs`
diff --git a/packages/powersync/lib/src/version.dart b/packages/powersync/lib/src/version.dart
index 40b7b1a6..11c304d0 100644
--- a/packages/powersync/lib/src/version.dart
+++ b/packages/powersync/lib/src/version.dart
@@ -1 +1 @@
-const String libraryVersion = '1.8.8';
+const String libraryVersion = '1.8.9';
diff --git a/packages/powersync/pubspec.yaml b/packages/powersync/pubspec.yaml
index f614c937..1cb800d9 100644
--- a/packages/powersync/pubspec.yaml
+++ b/packages/powersync/pubspec.yaml
@@ -1,5 +1,5 @@
 name: powersync
-version: 1.8.8
+version: 1.8.9
 homepage: https://powersync.com
 repository: https://github.com/powersync-ja/powersync.dart
 description: PowerSync Flutter SDK - sync engine for building local-first apps.
@@ -16,7 +16,7 @@ dependencies:
   sqlite3: ^2.4.6
   universal_io: ^2.0.0
   sqlite3_flutter_libs: ^0.5.23
-  powersync_flutter_libs: ^0.4.1
+  powersync_flutter_libs: ^0.4.2
   meta: ^1.0.0
   http: ^1.1.0
   uuid: ^4.2.0
diff --git a/packages/powersync_attachments_helper/CHANGELOG.md b/packages/powersync_attachments_helper/CHANGELOG.md
index 8444c333..fd46da63 100644
--- a/packages/powersync_attachments_helper/CHANGELOG.md
+++ b/packages/powersync_attachments_helper/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.6.13
+
+ - Update a dependency to the latest release.
+
 ## 0.6.12
 
  - Update a dependency to the latest release.
diff --git a/packages/powersync_attachments_helper/pubspec.yaml b/packages/powersync_attachments_helper/pubspec.yaml
index 49a19eb4..e38e05c5 100644
--- a/packages/powersync_attachments_helper/pubspec.yaml
+++ b/packages/powersync_attachments_helper/pubspec.yaml
@@ -1,6 +1,6 @@
 name: powersync_attachments_helper
 description: A helper library for handling attachments when using PowerSync.
-version: 0.6.12
+version: 0.6.13
 repository: https://github.com/powersync-ja/powersync.dart
 homepage: https://www.powersync.com/
 environment:
@@ -10,7 +10,7 @@ dependencies:
   flutter:
     sdk: flutter
 
-  powersync: ^1.8.8
+  powersync: ^1.8.9
   logging: ^1.2.0
   sqlite_async: ^0.9.1
   path_provider: ^2.0.13
diff --git a/packages/powersync_flutter_libs/CHANGELOG.md b/packages/powersync_flutter_libs/CHANGELOG.md
index f5b7e354..d14eb195 100644
--- a/packages/powersync_flutter_libs/CHANGELOG.md
+++ b/packages/powersync_flutter_libs/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.4.2
+
+ - Update a dependency to the latest release.
+
 ## 0.4.1
 
  - powersync-sqlite-core v0.3.4
diff --git a/packages/powersync_flutter_libs/pubspec.yaml b/packages/powersync_flutter_libs/pubspec.yaml
index 03bc8950..cb50688b 100644
--- a/packages/powersync_flutter_libs/pubspec.yaml
+++ b/packages/powersync_flutter_libs/pubspec.yaml
@@ -1,6 +1,6 @@
 name: powersync_flutter_libs
 description: PowerSync core binaries for the PowerSync Flutter SDK. Needs to be included for Flutter apps.
-version: 0.4.1
+version: 0.4.2
 repository: https://github.com/powersync-ja/powersync.dart
 homepage: https://www.powersync.com/