Merge branch 'dead-tab-detection' into fix/opfs-multitab-issue

Author: stevensJourney

packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -6,16 +6,16 @@ import {
   SyncStatusOptions
 } from '@powersync/common';
 import * as Comlink from 'comlink';
+import { getNavigatorLocks } from '../../shared/navigator';
 import { AbstractSharedSyncClientProvider } from '../../worker/sync/AbstractSharedSyncClientProvider';
 import { ManualSharedSyncPayload, SharedSyncClientEvent } from '../../worker/sync/SharedSyncImplementation';
-import { DEFAULT_CACHE_SIZE_KB, resolveWebSQLFlags, TemporaryStorageOption } from '../adapters/web-sql-flags';
+import { WorkerClient } from '../../worker/sync/WorkerClient';
 import { WebDBAdapter } from '../adapters/WebDBAdapter';
+import { DEFAULT_CACHE_SIZE_KB, TemporaryStorageOption, resolveWebSQLFlags } from '../adapters/web-sql-flags';
 import {
   WebStreamingSyncImplementation,
   WebStreamingSyncImplementationOptions
 } from './WebStreamingSyncImplementation';
-import { WorkerClient } from '../../worker/sync/WorkerClient';
-import { getNavigatorLocks } from '../../shared/navigator';
 
 /**
  * The shared worker will trigger methods on this side of the message port
@@ -160,6 +160,21 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
     const { crudUploadThrottleMs, identifier, retryDelayMs } = this.options;
     const flags = { ...this.webOptions.flags, workers: undefined };
 
+    // Request a random lock until this client is disposed. The name of the lock is sent to the shared worker, which
+    // will also attempt to acquire it. Since the lock is returned when the tab is closed, this allows the share worker
+    // to free resources associated with this tab.
+    // We take hold of this lock as soon-as-possible in order to cater for potentially closed tabs.
+    getNavigatorLocks().request(`tab-close-signal-${crypto.randomUUID()}`, async (lock) => {
+      if (!this.abortOnClose.signal.aborted) {
+        // Awaiting here ensures the worker is waiting for the lock
+        await this.syncManager.addLockBasedCloseSignal(lock!.name);
+
+        await new Promise<void>((r) => {
+          this.abortOnClose.signal.onabort = () => r();
+        });
+      }
+    });
+
     this.isInitialized = this.syncManager.setParams(
       {
         dbParams: this.dbAdapter.getConfiguration(),
@@ -190,19 +205,6 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
      * This performs bi-directional method calling.
      */
     Comlink.expose(this.clientProvider, this.messagePort);
-
-    // Request a random lock until this client is disposed. The name of the lock is sent to the shared worker, which
-    // will also attempt to acquire it. Since the lock is returned when the tab is closed, this allows the share worker
-    // to free resources associated with this tab.
-    getNavigatorLocks().request(`tab-close-signal-${crypto.randomUUID()}`, async (lock) => {
-      if (!this.abortOnClose.signal.aborted) {
-        this.syncManager.addLockBasedCloseSignal(lock!.name);
-
-        await new Promise<void>((r) => {
-          this.abortOnClose.signal.onabort = () => r();
-        });
-      }
-    });
   }
 
   /**

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

@@ -322,6 +322,22 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       });
 
       const shouldReconnect = !!this.connectionManager.syncStreamImplementation && this.ports.length > 0;
+
+      /**
+       * If the current database adapter is the one that is being closed, we need to disconnect from the backend.
+       * We can disconnect in the portMutex lock. This ensures the disconnect is not affected by potential other
+       * connect operations coming from other tabs.
+       */
+      if (this.dbAdapter && this.dbAdapter == trackedPort.db) {
+        this.logger.debug(`Disconnecting due to closed database: should reconnect: ${shouldReconnect}`);
+        this.dbAdapter = null;
+        // Unconditionally close the connection because the database it's writing to has just been closed.
+        // The connection has been closed previously, this might throw. We should be able to ignore it.
+        await this.connectionManager
+          .disconnect()
+          .catch((ex) => this.logger.warn('Error while disconnecting. Will attempt to reconnect.', ex));
+      }
+
       return {
         shouldReconnect,
         trackedPort
@@ -337,19 +353,8 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       await closeListener();
     }
 
-    if (this.dbAdapter && this.dbAdapter == trackedPort.db) {
-      // Unconditionally close the connection because the database it's writing to has just been closed.
-      // The connection has been closed previously, this might throw. We should be able to ignore it.
-      await this.connectionManager
-        .disconnect()
-        .catch((ex) => this.logger.warn('Error while disconnecting. Will attempt to reconnect.', ex));
-
-      // Clearing the adapter will result in a new one being opened in connect
-      this.dbAdapter = null;
-
-      if (shouldReconnect) {
-        await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions ?? {});
-      }
+    if (shouldReconnect) {
+      await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions ?? {});
     }
 
     // Re-index subscriptions, the subscriptions of the removed port would no longer be considered.
@@ -487,6 +492,11 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
         });
         lastClient.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.
+           */
           wrapped.close().catch((ex) => this.logger.warn('error closing database connection', ex));
           wrapped.markRemoteClosed();
         });

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

@@ -1,13 +1,13 @@
+import { ILogLevel, PowerSyncConnectionOptions, SubscribedStream, SyncStatusOptions } from '@powersync/common';
 import * as Comlink from 'comlink';
+import { getNavigatorLocks } from '../../shared/navigator';
 import {
   ManualSharedSyncPayload,
   SharedSyncClientEvent,
   SharedSyncImplementation,
   SharedSyncInitOptions,
   WrappedSyncPort
 } from './SharedSyncImplementation';
-import { ILogLevel, PowerSyncConnectionOptions, SubscribedStream, SyncStatusOptions } from '@powersync/common';
-import { getNavigatorLocks } from '../../shared/navigator';
 
 /**
  * A client to the shared sync worker.
@@ -21,7 +21,9 @@ export class WorkerClient {
   constructor(
     private readonly sync: SharedSyncImplementation,
     private readonly port: MessagePort
-  ) {}
+  ) {
+    Comlink.expose(this, this.port);
+  }
 
   async initialize() {
     /**
@@ -36,7 +38,6 @@ export class WorkerClient {
     });
 
     this.resolvedPort = await this.sync.addPort(this.port);
-    Comlink.expose(this, this.port);
   }
 
   private async removePort() {

packages/web/tests/mocks/MockWebRemote.ts

@@ -0,0 +1,56 @@
+import {
+  AbstractRemote,
+  AbstractRemoteOptions,
+  BSONImplementation,
+  DEFAULT_REMOTE_LOGGER,
+  FetchImplementation,
+  FetchImplementationProvider,
+  ILogger,
+  RemoteConnector
+} from '@powersync/common';
+
+/**
+ * Mock WebRemote that throws 401 Unauthorized errors for all HTTP requests.
+ * Used for testing error handling in the shared sync worker.
+ * Other tests may override this for managed streams.
+ */
+class MockFetchProvider extends FetchImplementationProvider {
+  getFetch(): FetchImplementation {
+    // Return a mock fetch that always returns 401
+    return async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
+      const response = new Response(null, {
+        status: 401,
+        statusText: 'Unauthorized'
+      });
+      return response;
+    };
+  }
+}
+
+export class WebRemote extends AbstractRemote {
+  private _bson: BSONImplementation | undefined;
+
+  constructor(
+    protected connector: RemoteConnector,
+    protected logger: ILogger = DEFAULT_REMOTE_LOGGER,
+    options?: Partial<AbstractRemoteOptions>
+  ) {
+    super(connector, logger, {
+      ...(options ?? {}),
+      fetchImplementation: options?.fetchImplementation ?? new MockFetchProvider()
+    });
+  }
+
+  getUserAgent(): string {
+    return 'powersync-web-mock';
+  }
+
+  async getBSON(): Promise<BSONImplementation> {
+    if (this._bson) {
+      return this._bson;
+    }
+    const { BSON } = await import('bson');
+    this._bson = BSON;
+    return this._bson;
+  }
+}