Fix handling worker errors

powersync-ja · Jan 16, 2025 · 1ab9b38 · 1ab9b38
1 parent 67e5e44
commit 1ab9b38
Show file tree

Hide file tree

Showing 5 changed files with 89 additions and 54 deletions.
diff --git a/sqlite3_web/CHANGELOG.md b/sqlite3_web/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.2.2
+
+- Recover from worker errors at startup.
+
 ## 0.2.1
 
 - Add `WebSqlite.deleteDatabase` to delete databases.

diff --git a/sqlite3_web/lib/src/channel.dart b/sqlite3_web/lib/src/channel.dart
@@ -103,7 +103,9 @@ abstract class ProtocolChannel {
   final Map<int, Completer<Response>> _responses = {};
 
   ProtocolChannel(this._channel) {
-    _channel.stream.listen(_handleIncoming);
+    _channel.stream.listen(_handleIncoming, onError: (e) {
+      close(e);
+    });
   }
 
   Future<void> get closed => _channel.sink.done;
@@ -165,8 +167,14 @@ abstract class ProtocolChannel {
 
   void handleNotification(Notification notification);
 
-  Future<void> close() async {
+  Future<void> close([Object? error]) async {
     await _channel.sink.close();
+
+    for (final response in _responses.values) {
+      response.completeError(
+          StateError('Channel closed before receiving response: $error'));
+    }
+    _responses.clear();
   }
 }
 

diff --git a/sqlite3_web/lib/src/client.dart b/sqlite3_web/lib/src/client.dart
@@ -255,16 +255,10 @@ final class DatabaseClient implements WebSqlite {
 
   Future<void> _startDedicated() async {
     if (globalContext.has('Worker')) {
-      final Worker dedicated;
-      try {
-        dedicated = Worker(
-          workerUri.toString().toJS,
-          WorkerOptions(name: 'sqlite3_worker'),
-        );
-      } on Object {
-        _missingFeatures.add(MissingBrowserFeature.dedicatedWorkers);
-        return;
-      }
+      final dedicated = Worker(
+        workerUri.toString().toJS,
+        WorkerOptions(name: 'sqlite3_worker'),
+      );
 
       final (endpoint, channel) = await createChannel();
       ConnectRequest(endpoint: endpoint, requestId: 0).sendToWorker(dedicated);
@@ -278,14 +272,8 @@ final class DatabaseClient implements WebSqlite {
 
   Future<void> _startShared() async {
     if (globalContext.has('SharedWorker')) {
-      final SharedWorker shared;
-      try {
-        shared = SharedWorker(workerUri.toString().toJS);
-        shared.port.start();
-      } on Object {
-        _missingFeatures.add(MissingBrowserFeature.sharedWorkers);
-        return;
-      }
+      final shared = SharedWorker(workerUri.toString().toJS);
+      shared.port.start();
 
       final (endpoint, channel) = await createChannel();
       ConnectRequest(endpoint: endpoint, requestId: 0).sendToPort(shared.port);
@@ -348,15 +336,22 @@ final class DatabaseClient implements WebSqlite {
     final available = <(StorageMode, AccessMode)>[];
     var workersReportedIndexedDbSupport = false;
 
-    if (_connectionToDedicated case final connection?) {
-      final response = await connection.sendRequest(
-        CompatibilityCheck(
-          requestId: 0,
-          type: MessageType.dedicatedCompatibilityCheck,
-          databaseName: databaseName,
-        ),
-        MessageType.simpleSuccessResponse,
-      );
+    Future<void> dedicatedCompatibilityCheck(
+        WorkerConnection connection) async {
+      SimpleSuccessResponse response;
+      try {
+        response = await connection.sendRequest(
+          CompatibilityCheck(
+            requestId: 0,
+            type: MessageType.dedicatedCompatibilityCheck,
+            databaseName: databaseName,
+          ),
+          MessageType.simpleSuccessResponse,
+        );
+      } on Object {
+        return;
+      }
+
       final result = CompatibilityResult.fromJS(response.response as JSObject);
       existing.addAll(result.existingDatabases);
       available.add((StorageMode.inMemory, AccessMode.throughDedicatedWorker));
@@ -390,15 +385,21 @@ final class DatabaseClient implements WebSqlite {
       }
     }
 
-    if (_connectionToShared case final connection?) {
-      final response = await connection.sendRequest(
-        CompatibilityCheck(
-          requestId: 0,
-          type: MessageType.sharedCompatibilityCheck,
-          databaseName: databaseName,
-        ),
-        MessageType.simpleSuccessResponse,
-      );
+    Future<void> sharedCompatibilityCheck(WorkerConnection connection) async {
+      SimpleSuccessResponse response;
+      try {
+        response = await connection.sendRequest(
+          CompatibilityCheck(
+            requestId: 0,
+            type: MessageType.sharedCompatibilityCheck,
+            databaseName: databaseName,
+          ),
+          MessageType.simpleSuccessResponse,
+        );
+      } on Object {
+        return;
+      }
+
       final result = CompatibilityResult.fromJS(response.response as JSObject);
 
       if (result.canUseIndexedDb) {
@@ -423,6 +424,13 @@ final class DatabaseClient implements WebSqlite {
       }
     }
 
+    if (_connectionToDedicated case final dedicated?) {
+      await dedicatedCompatibilityCheck(dedicated);
+    }
+    if (_connectionToShared case final shared?) {
+      await sharedCompatibilityCheck(shared);
+    }
+
     available.add((StorageMode.inMemory, AccessMode.inCurrentContext));
     if (workersReportedIndexedDbSupport || await checkIndexedDbSupport()) {
       // If the workers can use IndexedDb, so can we.

diff --git a/sqlite3_web/pubspec.yaml b/sqlite3_web/pubspec.yaml
@@ -1,6 +1,6 @@
 name: sqlite3_web
 description: Utilities to simplify accessing sqlite3 on the web, with automated feature detection.
-version: 0.2.1
+version: 0.2.2
 homepage: https://github.com/simolus3/sqlite3.dart/tree/main/sqlite3_web
 repository: https://github.com/simolus3/sqlite3.dart
 

diff --git a/sqlite3_web/test/integration_test.dart b/sqlite3_web/test/integration_test.dart
@@ -108,21 +108,36 @@ void main() {
       });
 
       setUp(() async {
-        final rawDriver = await createDriver(
-          spec: browser.isChromium ? WebDriverSpec.JsonWire : WebDriverSpec.W3c,
-          uri: browser.driverUri,
-          desired: {
-            'goog:chromeOptions': {
-              'args': [
-                '--headless=new',
-                '--disable-search-engine-choice-screen',
-              ],
-            },
-            'moz:firefoxOptions': {
-              'args': ['-headless']
-            },
-          },
-        );
+        late WebDriver rawDriver;
+        for (var i = 0; i < 3; i++) {
+          try {
+            rawDriver = await createDriver(
+              spec: browser.isChromium
+                  ? WebDriverSpec.JsonWire
+                  : WebDriverSpec.W3c,
+              uri: browser.driverUri,
+              desired: {
+                'goog:chromeOptions': {
+                  'args': [
+                    '--headless=new',
+                    '--disable-search-engine-choice-screen',
+                  ],
+                },
+                'moz:firefoxOptions': {
+                  'args': ['-headless']
+                },
+              },
+            );
+            break;
+          } on SocketException {
+            // webdriver server taking a bit longer to start up...
+            if (i == 2) {
+              rethrow;
+            }
+
+            await Future.delayed(const Duration(milliseconds: 500));
+          }
+        }
 
         // logs.get() isn't supported on Firefox
         if (browser != Browser.firefox) {