Opfs uploads (#505)

Co-authored-by: Christiaan Landman <chriz.ek@gmail.com>

Author: stevensJourney

.changeset/forty-moose-yell.md

@@ -0,0 +1,5 @@
+---
+'@powersync/web': patch
+---
+
+Fixed bug where using OPFS and reconnecting would cause upload triggers to fail.

.github/workflows/test.yml

@@ -7,7 +7,7 @@ on:
 jobs:
   test:
     name: Test Packages
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-xl
     steps:
       - uses: actions/checkout@v4
         with:

packages/web/src/worker/sync/SharedSyncImplementation.ts

@@ -180,6 +180,9 @@ export class SharedSyncImplementation
     await this.waitForReady();
     // This effectively queues connect and disconnect calls. Ensuring multiple tabs' requests are synchronized
     return getNavigatorLocks().request('shared-sync-connect', async () => {
+      if (!this.dbAdapter) {
+        await this.openInternalDB();
+      }
       this.syncStreamClient = this.generateStreamingImplementation();
       this.lastConnectOptions = options;
       this.syncStreamClient.registerListener({
@@ -231,12 +234,7 @@ export class SharedSyncImplementation
     }
 
     const trackedPort = this.ports[index];
-    if (trackedPort.db) {
-      trackedPort.db.close();
-    }
-
-    // Release proxy
-    trackedPort.clientProvider[Comlink.releaseProxy]();
+    // Remove from the list of active ports
     this.ports.splice(index, 1);
 
     /**
@@ -249,12 +247,25 @@ export class SharedSyncImplementation
       }
     });
 
-    if (this.dbAdapter == trackedPort.db && this.syncStreamClient) {
-      await this.disconnect();
-      // Ask for a new DB worker port handler
-      await this.openInternalDB();
-      await this.connect(this.lastConnectOptions);
+    const shouldReconnect = !!this.syncStreamClient;
+    if (this.dbAdapter && this.dbAdapter == trackedPort.db) {
+      if (shouldReconnect) {
+        await this.disconnect();
+      }
+
+      // Clearing the adapter will result in a new one being opened in connect
+      this.dbAdapter = null;
+
+      if (shouldReconnect) {
+        await this.connect(this.lastConnectOptions);
+      }
     }
+
+    if (trackedPort.db) {
+      trackedPort.db.close();
+    }
+    // Release proxy
+    trackedPort.clientProvider[Comlink.releaseProxy]();
   }
 
   triggerCrudUpload() {
@@ -334,6 +345,10 @@ export class SharedSyncImplementation
 
   protected async openInternalDB() {
     const lastClient = this.ports[this.ports.length - 1];
+    if (!lastClient) {
+      // Should not really happen in practice
+      throw new Error(`Could not open DB connection since no client is connected.`);
+    }
     const workerPort = await lastClient.clientProvider.getDBWorkerPort();
     const remote = Comlink.wrap<OpenAsyncDatabaseConnection>(workerPort);
     const identifier = this.syncParams!.dbParams.dbFilename;

packages/web/tests/main.test.ts

@@ -1,16 +1,22 @@
-import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { AbstractPowerSyncDatabase } from '@powersync/common';
+import { PowerSyncDatabase, WASQLiteOpenFactory, WASQLiteVFS } from '@powersync/web';
 import { v4 as uuid } from 'uuid';
-import { TestDatabase, generateTestDb } from './utils/testDb';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import { TestDatabase, generateTestDb, testSchema } from './utils/testDb';
 // TODO import tests from a common package
 
 describe('Basic', () => {
   let dbWithoutWebWorker: AbstractPowerSyncDatabase;
   let dbWithWebWorker: AbstractPowerSyncDatabase;
+  let dbWithOPFS: AbstractPowerSyncDatabase;
 
   beforeEach(() => {
     dbWithoutWebWorker = generateTestDb({ useWebWorker: false });
     dbWithWebWorker = generateTestDb();
+    dbWithOPFS = new PowerSyncDatabase({
+      database: new WASQLiteOpenFactory({ dbFilename: 'basic.sqlite', vfs: WASQLiteVFS.OPFSCoopSyncVFS }),
+      schema: testSchema
+    });
   });
 
   /**
@@ -19,13 +25,16 @@ describe('Basic', () => {
   const itWithDBs = (name: string, test: (db: AbstractPowerSyncDatabase) => Promise<void>) => {
     it(`${name} - with web worker`, () => test(dbWithWebWorker));
     it(`${name} - without web worker`, () => test(dbWithoutWebWorker));
+    it(`${name} - with OPFS`, () => test(dbWithOPFS));
   };
 
   afterEach(async () => {
     await dbWithWebWorker.disconnectAndClear();
     await dbWithWebWorker.close();
     await dbWithoutWebWorker.disconnectAndClear();
     await dbWithoutWebWorker.close();
+    await dbWithOPFS.disconnectAndClear();
+    await dbWithOPFS.close();
   });
 
   describe('executeQuery', () => {

packages/web/tests/stream.test.ts

@@ -1,3 +1,4 @@
+import { WASQLiteOpenFactory, WASQLiteVFS } from '@powersync/web';
 import Logger from 'js-logger';
 import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest';
 import { TestConnector } from './utils/MockStreamOpenFactory';
@@ -18,19 +19,36 @@ describe(
       name: string,
       test: (createConnectedDatabase: () => ReturnType<typeof generateConnectedDatabase>) => Promise<void>
     ) => {
-      const funcWithWebWorker = generateConnectedDatabase;
+      const funcWithWebWorker = () =>
+        generateConnectedDatabase({
+          powerSyncOptions: {
+            dbFilename: `test-stream-connection-worker.db`
+          }
+        });
       const funcWithoutWebWorker = () =>
         generateConnectedDatabase({
           powerSyncOptions: {
-            dbFilename: 'test-stream-connection-no-worker.db',
+            dbFilename: `test-stream-connection-no-worker.db`,
             flags: {
               useWebWorker: false
             }
           }
         });
 
-      it(`${name} - with web worker`, () => test(funcWithWebWorker));
-      it(`${name} - without web worker`, () => test(funcWithoutWebWorker));
+      it.sequential(`${name} - with web worker`, () => test(funcWithWebWorker));
+      it.sequential(`${name} - without web worker`, () => test(funcWithoutWebWorker));
+      it.sequential(`${name} - with OPFS`, () =>
+        test(() =>
+          generateConnectedDatabase({
+            powerSyncOptions: {
+              database: new WASQLiteOpenFactory({
+                dbFilename: `test-stream-connection-opfs.db`,
+                vfs: WASQLiteVFS.OPFSCoopSyncVFS
+              })
+            }
+          })
+        )
+      );
     };
 
     beforeAll(() => Logger.useDefaults());
@@ -101,7 +119,7 @@ describe(
       const throwCounter = 2;
       uploadSpy.mockImplementation(async (db) => {
         if (uploadCounter++ < throwCounter) {
-          throw new Error('No uploads yet');
+          throw new Error(`No uploads yet`);
         }
         // Now actually do the upload
         const tx = await db.getNextCrudTransaction();

packages/web/vitest.config.ts

@@ -46,10 +46,10 @@ const config: UserConfigExport = {
       instances: [
         {
           browser: 'chromium'
-        },
-        {
-          browser: 'firefox'
         }
+        // {
+        //   browser: 'firefox'
+        // }
         // This requires some additional work to get all tests passing
         // {
         //   browser: 'webkit'