Remove subscriptions when tab is removed

Author: simolus3

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

@@ -7,18 +7,15 @@ import {
 } from '@powersync/common';
 import * as Comlink from 'comlink';
 import { AbstractSharedSyncClientProvider } from '../../worker/sync/AbstractSharedSyncClientProvider';
-import {
-  ManualSharedSyncPayload,
-  SharedSyncClientEvent,
-  SharedSyncImplementation
-} from '../../worker/sync/SharedSyncImplementation';
+import { ManualSharedSyncPayload, SharedSyncClientEvent } from '../../worker/sync/SharedSyncImplementation';
 import { DEFAULT_CACHE_SIZE_KB, resolveWebSQLFlags, TemporaryStorageOption } from '../adapters/web-sql-flags';
 import { WebDBAdapter } from '../adapters/WebDBAdapter';
 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
@@ -111,6 +108,7 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
 
   protected isInitialized: Promise<void>;
   protected dbAdapter: WebDBAdapter;
+  private abortOnClose = new AbortController();
 
   constructor(options: SharedWebStreamingSyncImplementationOptions) {
     super(options);
@@ -192,6 +190,19 @@ 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();
+        });
+      }
+    });
   }
 
   /**
@@ -238,6 +249,7 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
       };
       this.messagePort.postMessage(closeMessagePayload);
     });
+    this.abortOnClose.abort();
 
     // Release the proxy
     this.syncManager[Comlink.releaseProxy]();

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

@@ -292,12 +292,12 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
    * Removes a message port client from this manager's managed
    * clients.
    */
-  async removePort(port: MessagePort) {
+  async removePort(port: WrappedSyncPort) {
     // Remove the port within a mutex context.
     // Warns if the port is not found. This should not happen in practice.
     // We return early if the port is not found.
     const { trackedPort, shouldReconnect } = await this.portMutex.runExclusive(async () => {
-      const index = this.ports.findIndex((p) => p.port == port);
+      const index = this.ports.findIndex((p) => p == port);
       if (index < 0) {
         this.logger.warn(`Could not remove port ${port} since it is not present in active ports.`);
         return {};
@@ -312,7 +312,7 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
        * not resolve. Abort them here.
        */
       [this.fetchCredentialsController, this.uploadDataController].forEach((abortController) => {
-        if (abortController?.activePort.port == port) {
+        if (abortController?.activePort == port) {
           abortController!.controller.abort(
             new AbortOperation('Closing pending requests after client port is removed')
           );
@@ -347,6 +347,10 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
     if (trackedPort.db) {
       await trackedPort.db.close();
     }
+
+    // Re-index subscriptions, the subscriptions of the removed port would no longer be considered.
+    this.collectActiveSubscriptions();
+
     // Release proxy
     return () => trackedPort.clientProvider[Comlink.releaseProxy]();
   }

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

@@ -7,6 +7,7 @@ import {
   WrappedSyncPort
 } from './SharedSyncImplementation';
 import { ILogLevel, PowerSyncConnectionOptions, SubscribedStream, SyncStatusOptions } from '@powersync/common';
+import { getNavigatorLocks } from '../../shared/navigator';
 
 /**
  * A client to the shared sync worker.
@@ -30,19 +31,37 @@ export class WorkerClient {
     this.port.addEventListener('message', async (event) => {
       const payload = event.data as ManualSharedSyncPayload;
       if (payload?.event == SharedSyncClientEvent.CLOSE_CLIENT) {
-        const release = await this.sync.removePort(this.port);
-        this.port.postMessage({
-          event: SharedSyncClientEvent.CLOSE_ACK,
-          data: {}
-        } satisfies ManualSharedSyncPayload);
-        release?.();
+        await this.removePort();
       }
     });
 
     this.resolvedPort = await this.sync.addPort(this.port);
     Comlink.expose(this, this.port);
   }
 
+  private async removePort() {
+    if (this.resolvedPort) {
+      const release = await this.sync.removePort(this.resolvedPort);
+      this.port.postMessage({
+        event: SharedSyncClientEvent.CLOSE_ACK,
+        data: {}
+      } satisfies ManualSharedSyncPayload);
+      release?.();
+    }
+  }
+
+  /**
+   * Called by a client after obtaining a lock with a random name.
+   *
+   * When the client tab is closed, its lock will be returned. So when the shared worker attempts to acquire the lock,
+   * it can consider the connection to be closed.
+   */
+  addLockBasedCloseSignal(name: string) {
+    getNavigatorLocks().request(name, async () => {
+      await this.removePort();
+    });
+  }
+
   setLogLevel(level: ILogLevel) {
     this.sync.setLogLevel(level);
   }