Fire and forget close operations.

Serialize all sync status errors.

Author: stevensJourney

package.json

@@ -49,5 +49,10 @@
     "rollup-plugin-dts": "^6.2.1",
     "typescript": "^5.7.2",
     "vitest": "^3.2.4"
+  },
+  "pnpm": {
+    "overrides": {
+      "@journeyapps/wa-sqlite": "0.0.0-dev-20251201120934"
+    }
   }
 }

packages/common/src/db/crud/SyncStatus.ts

@@ -1,7 +1,7 @@
-import { CoreStreamSubscription } from '../../client/sync/stream/core-instruction.js';
 import { SyncClientImplementation } from '../../client/sync/stream/AbstractStreamingSyncImplementation.js';
-import { InternalProgressInformation, ProgressWithOperations, SyncProgress } from './SyncProgress.js';
+import { CoreStreamSubscription } from '../../client/sync/stream/core-instruction.js';
 import { SyncStreamDescription, SyncSubscriptionDescription } from '../../client/sync/sync-streams.js';
+import { InternalProgressInformation, ProgressWithOperations, SyncProgress } from './SyncProgress.js';
 
 export type SyncDataFlowStatus = Partial<{
   downloading: boolean;
@@ -250,13 +250,28 @@ export class SyncStatus {
     return {
       connected: this.connected,
       connecting: this.connecting,
-      dataFlow: this.dataFlowStatus,
+      dataFlow: {
+        ...this.dataFlowStatus,
+        uploadError: this.serializeError(this.dataFlowStatus.uploadError),
+        downloadError: this.serializeError(this.dataFlowStatus.downloadError)
+      },
       lastSyncedAt: this.lastSyncedAt,
       hasSynced: this.hasSynced,
       priorityStatusEntries: this.priorityStatusEntries
     };
   }
 
+  protected serializeError(error?: Error) {
+    if (typeof error == 'undefined') {
+      return undefined;
+    }
+    return {
+      name: error.name,
+      message: error.message,
+      stack: error.stack
+    };
+  }
+
   private static comparePriorities(a: SyncPriorityStatus, b: SyncPriorityStatus) {
     return b.priority - a.priority; // Reverse because higher priorities have lower numbers
   }

packages/web/src/db/adapters/WorkerWrappedAsyncDatabaseConnection.ts

@@ -80,14 +80,17 @@ export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLO
     return this.withRemote(() => this.baseConnection.isAutoCommit());
   }
 
-  private withRemote<T>(workerPromise: () => Promise<T>): Promise<T> {
+  private withRemote<T>(workerPromise: () => Promise<T>, fireActionOnAbort = false): Promise<T> {
     const controller = this.notifyRemoteClosed;
     if (controller) {
       return new Promise((resolve, reject) => {
         if (controller.signal.aborted) {
           reject(new ConnectionClosedError('Called operation on closed remote'));
-          // Don't run the operation if we're going to reject
-          return;
+          if (!fireActionOnAbort) {
+            // Don't run the operation if we're going to reject
+            // We might want to fire-and-forget the operation in some cases (like a close operation)
+            return;
+          }
         }
 
         function handleAbort() {
@@ -172,7 +175,8 @@ export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLO
     // Abort any pending lock requests.
     this.lockAbortController.abort();
     try {
-      await this.withRemote(() => this.baseConnection.close());
+      // fire and forget the close operation
+      await this.withRemote(() => this.baseConnection.close(), true);
     } finally {
       this.options.remote[Comlink.releaseProxy]();
       this.options.onClose?.();

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

@@ -184,6 +184,16 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
     });
   }
 
+  /**
+   * In some very rare cases a specific tab might not respond to requests.
+   * This returns a random port which is not closing.
+   */
+  protected async getRandomWrappedPort(): Promise<WrappedSyncPort | undefined> {
+    return await this.portMutex.runExclusive(() => {
+      return this.ports[Math.floor(Math.random() * this.ports.length)];
+    });
+  }
+
   async waitForStatus(status: SyncStatusOptions): Promise<void> {
     return this.withSyncImplementation(async (sync) => {
       return sync.waitForStatus(status);
@@ -487,12 +497,17 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
   protected async openInternalDB() {
     while (true) {
       try {
-        const lastClient = await this.getLastWrappedPort();
-        if (!lastClient) {
+        const client = await this.getRandomWrappedPort();
+        if (!client) {
           // Should not really happen in practice
           throw new Error(`Could not open DB connection since no client is connected.`);
         }
 
+        // Fail-safe timeout for opening a database connection.
+        const timeout = setTimeout(() => {
+          abortController.abort();
+        }, 10_000);
+
         /**
          * Handle cases where the client might close while opening a connection.
          */
@@ -502,21 +517,20 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
         };
 
         const removeCloseListener = () => {
-          const index = lastClient.closeListeners.indexOf(closeListener);
+          const index = client.closeListeners.indexOf(closeListener);
           if (index >= 0) {
-            lastClient.closeListeners.splice(index, 1);
+            client.closeListeners.splice(index, 1);
           }
         };
 
-        lastClient.closeListeners.push(closeListener);
+        client.closeListeners.push(closeListener);
 
-        const workerPort = await withAbort(
-          () => lastClient.clientProvider.getDBWorkerPort(),
-          abortController.signal
-        ).catch((ex) => {
-          removeCloseListener();
-          throw ex;
-        });
+        const workerPort = await withAbort(() => client.clientProvider.getDBWorkerPort(), abortController.signal).catch(
+          (ex) => {
+            removeCloseListener();
+            throw ex;
+          }
+        );
 
         const remote = Comlink.wrap<OpenAsyncDatabaseConnection>(workerPort);
         const identifier = this.syncParams!.dbParams.dbFilename;
@@ -532,6 +546,8 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
           removeCloseListener();
         });
 
+        clearTimeout(timeout);
+
         const wrapped = new WorkerWrappedAsyncDatabaseConnection({
           remote,
           baseConnection: db,
@@ -540,15 +556,15 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
           // that and ensure pending requests are aborted when the tab is closed.
           remoteCanCloseUnexpectedly: true
         });
-        lastClient.closeListeners.push(async () => {
+        client.closeListeners.push(async () => {
           this.logger.info('Aborting open connection because associated tab closed.');
           /**
            * Don't await this close operation. It might never resolve if the tab is closed.
-           * We run the close operation first, before marking the remote as closed. This gives the database some chance
-           * to close the connection.
+           * We mark the remote as closed first, this will reject any pending requests.
+           * We then call close. The close operation is configured to fire-and-forget, the main promise will reject immediately.
            */
-          wrapped.close().catch((ex) => this.logger.warn('error closing database connection', ex));
           wrapped.markRemoteClosed();
+          wrapped.close().catch((ex) => this.logger.warn('error closing database connection', ex));
         });
 
         return wrapped;

packages/web/tests/error_serialization.test.ts

@@ -0,0 +1,43 @@
+import { SyncStreamConnectionMethod } from '@powersync/common';
+import { describe, expect } from 'vitest';
+import { sharedMockSyncServiceTest } from './utils/mockSyncServiceTest';
+
+/**
+ * Test to verify that Error instances are properly serialized when passed through MessagePorts.
+ * When errors occur in the shared worker and are reported via statusChanged, they should
+ * be properly serialized and deserialized to appear in the sync status.
+ */
+describe('Error Serialization through MessagePorts', { sequential: true }, () => {
+  sharedMockSyncServiceTest(
+    'should serialize and deserialize Error in sync status when connection fails',
+    { timeout: 10_000 },
+    async ({ context: { database, mockService } }) => {
+      await mockService.setAutomaticResponse({
+        status: 401,
+        headers: { 'Content-Type': 'application/json' },
+        bodyLines: ['Unauthorized']
+      });
+
+      // Start connection attempt
+      await database.connect(
+        {
+          fetchCredentials: async () => {
+            return {
+              endpoint: 'http://localhost:3000',
+              token: 'test-token'
+            };
+          },
+          uploadData: async () => {}
+        },
+        {
+          connectionMethod: SyncStreamConnectionMethod.HTTP
+        }
+      );
+
+      expect(database.currentStatus.dataFlowStatus?.downloadError).toBeDefined();
+      expect(database.currentStatus.dataFlowStatus?.downloadError?.name).toBe('Error');
+      expect(database.currentStatus.dataFlowStatus?.downloadError?.message).toBe('HTTP : "Unauthorized"\n');
+      expect(database.currentStatus.dataFlowStatus?.downloadError?.stack).toBeDefined();
+    }
+  );
+});

packages/web/tests/mockSyncServiceExample.test.ts

@@ -17,9 +17,9 @@ describe('Mock Sync Service Example', { timeout: 100000 }, () => {
   sharedMockSyncServiceTest(
     'should allow mocking sync responses in shared worker',
     { timeout: 100000 },
-    async ({ context: { database, connect } }) => {
+    async ({ context: { database, connect, mockService } }) => {
       // Call connect to start the sync worker and get the sync service
-      const { syncService, syncRequestId } = await connect();
+      const { syncRequestId } = await connect();
 
       // Push a checkpoint with buckets (following node test pattern)
       const checkpoint: StreamingSyncCheckpoint = {
@@ -37,10 +37,10 @@ describe('Mock Sync Service Example', { timeout: 100000 }, () => {
         }
       };
 
-      await syncService.pushBodyLine(syncRequestId, checkpoint);
+      await mockService.pushBodyLine(syncRequestId, checkpoint);
 
       // The connect call should resolve by now
-      await syncService.pushBodyLine(syncRequestId, {
+      await mockService.pushBodyLine(syncRequestId, {
         data: {
           bucket: 'a',
           data: [
@@ -57,14 +57,14 @@ describe('Mock Sync Service Example', { timeout: 100000 }, () => {
       });
 
       // Push checkpoint_complete to finish the sync
-      await syncService.pushBodyLine(syncRequestId, {
+      await mockService.pushBodyLine(syncRequestId, {
         checkpoint_complete: {
           last_op_id: '1'
         }
       });
 
       // Complete the response
-      await syncService.completeResponse(syncRequestId);
+      await mockService.completeResponse(syncRequestId);
 
       // Wait for sync to complete and verify the data was saved
       await vi.waitFor(async () => {

packages/web/tests/multiple_instances.test.ts

@@ -243,7 +243,7 @@ describe('Multiple Instances', { sequential: true }, () => {
 
   sharedMockSyncServiceTest(
     'should trigger uploads from last connected clients',
-    async ({ context: { database, openDatabase, connect, connector } }) => {
+    async ({ context: { database, openDatabase, connect, connector, mockService } }) => {
       const secondDatabase = openDatabase();
 
       expect(database.currentStatus.connected).false;
@@ -258,7 +258,7 @@ describe('Multiple Instances', { sequential: true }, () => {
       await database.execute('INSERT into lists (id, name) VALUES (uuid(), ?)', ['steven']);
 
       // connect from the first database
-      const { syncService } = await connect();
+      await connect();
 
       await vi.waitFor(() => expect(database.currentStatus.connected).true);
 
@@ -275,13 +275,13 @@ describe('Multiple Instances', { sequential: true }, () => {
       const secondConnectPromise = secondDatabase.connect(secondConnector);
       let _pendingRequestId: string;
       await vi.waitFor(async () => {
-        const requests = await syncService.getPendingRequests();
+        const requests = await mockService.getPendingRequests();
         expect(requests.length).toBeGreaterThan(0);
         _pendingRequestId = requests[0].id;
       });
       const pendingRequestId = _pendingRequestId!;
-      await syncService.createResponse(pendingRequestId, 200, { 'Content-Type': 'application/json' });
-      await syncService.pushBodyLine(pendingRequestId, {
+      await mockService.createResponse(pendingRequestId, 200, { 'Content-Type': 'application/json' });
+      await mockService.pushBodyLine(pendingRequestId, {
         token_expires_in: 10000000
       });
       await secondConnectPromise;

packages/web/tests/utils/mockSyncServiceTest.ts

@@ -41,7 +41,6 @@ export function createTestConnector(): MockedTestConnector {
  * Result of calling the connect function
  */
 export interface ConnectResult {
-  syncService: MockSyncService;
   syncRequestId: string;
 }
 
@@ -63,6 +62,7 @@ export const sharedMockSyncServiceTest = test.extend<{
     database: PowerSyncDatabase;
     databaseName: string;
     openDatabase: (customConfig?: Partial<WebPowerSyncDatabaseOptions>) => PowerSyncDatabase;
+    mockService: MockSyncService;
   };
 }>({
   context: async ({}, use) => {
@@ -98,6 +98,15 @@ export const sharedMockSyncServiceTest = test.extend<{
 
     const database = openDatabase();
 
+    // Get the identifier from the database.name property
+    const identifier = database.database.name;
+
+    // Connect to the shared worker to get the mock service
+    const mockService = await getMockSyncServiceFromWorker(identifier);
+    if (!mockService) {
+      throw new Error('Mock service not available');
+    }
+
     const connector = createTestConnector();
 
     const connectFn = async (customConnector?: PowerSyncBackendConnector): Promise<ConnectResult> => {
@@ -116,18 +125,6 @@ export const sharedMockSyncServiceTest = test.extend<{
         { timeout: 1000 }
       );
 
-      // Get the identifier from the database.name property
-      const identifier = database.database.name;
-
-      // Connect to the shared worker to get the mock service
-      const mockService = await getMockSyncServiceFromWorker(identifier);
-
-      if (!mockService) {
-        throw new Error(
-          'Mock service not available - ensure enableMultiTabs is true and running in a test environment'
-        );
-      }
-
       let _syncRequestId: string;
       await vi.waitFor(async () => {
         const requests = await mockService.getPendingRequests();
@@ -147,7 +144,6 @@ export const sharedMockSyncServiceTest = test.extend<{
       await connectionPromise;
 
       return {
-        syncService: mockService,
         syncRequestId
       };
     };
@@ -157,7 +153,8 @@ export const sharedMockSyncServiceTest = test.extend<{
       connect: connectFn,
       database,
       databaseName: dbFilename,
-      openDatabase
+      openDatabase,
+      mockService
     });
   }
 });