Re-open database as soon as it's closed. Trigger uploads if database has closed.

Author: stevensJourney

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

@@ -17,6 +17,13 @@ export type ProxiedQueryResult = Omit<QueryResult, 'rows'> & {
  */
 export type OnTableChangeCallback = (event: BatchedUpdateNotification) => void;
 
+export class ConnectionClosedError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = 'ConnectionClosedError';
+  }
+}
+
 /**
  * @internal
  * An async Database connection which provides basic async SQL methods.

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

@@ -11,12 +11,9 @@ import {
   type ILogger
 } from '@powersync/common';
 import { getNavigatorLocks } from '../..//shared/navigator';
-import { AsyncDatabaseConnection } from './AsyncDatabaseConnection';
+import { AsyncDatabaseConnection, ConnectionClosedError } from './AsyncDatabaseConnection';
 import { SharedConnectionWorker, WebDBAdapter } from './WebDBAdapter';
-import {
-  WorkerConnectionClosedError,
-  WorkerWrappedAsyncDatabaseConnection
-} from './WorkerWrappedAsyncDatabaseConnection';
+import { WorkerWrappedAsyncDatabaseConnection } from './WorkerWrappedAsyncDatabaseConnection';
 import { WASQLiteVFS } from './wa-sqlite/WASQLiteConnection';
 import { ResolvedWASQLiteOpenFactoryOptions } from './wa-sqlite/WASQLiteOpenFactory';
 import { ResolvedWebSQLOpenOptions } from './web-sql-flags';
@@ -30,10 +27,15 @@ export interface LockedAsyncDatabaseAdapterOptions {
   debugMode?: boolean;
   logger?: ILogger;
   defaultLockTimeoutMs?: number;
+  reOpenOnConnectionClosed?: boolean;
 }
 
 export type LockedAsyncDatabaseAdapterListener = DBAdapterListener & {
   initialized?: () => void;
+  /**
+   * Fired when the database is re-opened after being closed.
+   */
+  databaseReOpened?: () => void;
 };
 
 /**
@@ -56,6 +58,7 @@ export class LockedAsyncDatabaseAdapter
   protected pendingAbortControllers: Set<AbortController>;
   protected requiresHolds: boolean | null;
   protected requiresReOpen: boolean;
+  protected databaseOpenPromise: Promise<void> | null = null;
 
   closing: boolean;
   closed: boolean;
@@ -116,10 +119,15 @@ export class LockedAsyncDatabaseAdapter
     this._disposeTableChangeListener?.();
     this._disposeTableChangeListener = null;
 
+    const isReOpen = !!this._db;
+
     this._db = await this.options.openConnection();
     await this._db.init();
     this._config = await this._db.getConfig();
     await this.registerOnChangeListener(this._db);
+    if (isReOpen) {
+      this.iterateListeners((cb) => cb.databaseReOpened?.());
+    }
     /**
      * This is only required for the long-lived shared IndexedDB connections.
      */
@@ -253,18 +261,28 @@ export class LockedAsyncDatabaseAdapter
         }
         let holdId: string | null = null;
         try {
-          if (this.requiresReOpen) {
-            this.logger.debug('Re-opening database');
-            await this.openInternalDB();
-            this.logger.debug('Database re-opened');
-            this.requiresReOpen = false;
+          // The database is being opened in the background. Wait for it here.
+          if (this.databaseOpenPromise) {
+            try {
+              await this.databaseOpenPromise;
+            } catch (ex) {
+              // This will cause a retry of opening the database.
+              const wrappedError = new ConnectionClosedError('Could not open database');
+              wrappedError.cause = ex;
+              throw wrappedError;
+            }
           }
 
           holdId = this.requiresHolds ? await this.baseDB.markHold() : null;
           return await callback();
         } catch (ex) {
-          if (ex instanceof WorkerConnectionClosedError) {
-            this.requiresReOpen = true;
+          if (ex instanceof ConnectionClosedError) {
+            if (this.options.reOpenOnConnectionClosed && !this.databaseOpenPromise && !this.closing) {
+              // Immediately re-open the database. We need to miss as little table updates as possible.
+              this.databaseOpenPromise = this.openInternalDB().finally(() => {
+                this.databaseOpenPromise = null;
+              });
+            }
           }
           throw ex;
         } finally {

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

@@ -1,6 +1,7 @@
 import * as Comlink from 'comlink';
 import {
   AsyncDatabaseConnection,
+  ConnectionClosedError,
   OnTableChangeCallback,
   OpenAsyncDatabaseConnection,
   ProxiedQueryResult
@@ -23,13 +24,6 @@ export type WrappedWorkerConnectionOptions<Config extends ResolvedWebSQLOpenOpti
   onClose?: () => void;
 };
 
-export class WorkerConnectionClosedError extends Error {
-  constructor(message: string) {
-    super(message);
-    this.name = 'WorkerConnectionClosedError';
-  }
-}
-
 /**
  * Wraps a provided instance of {@link AsyncDatabaseConnection}, providing necessary proxy
  * functions for worker listeners.
@@ -84,13 +78,13 @@ export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLO
     if (controller) {
       return new Promise((resolve, reject) => {
         if (controller.signal.aborted) {
-          reject(new WorkerConnectionClosedError('Called operation on closed remote'));
+          reject(new ConnectionClosedError('Called operation on closed remote'));
           // Don't run the operation if we're going to reject
           return;
         }
 
         function handleAbort() {
-          reject(new WorkerConnectionClosedError('Remote peer closed with request in flight'));
+          reject(new ConnectionClosedError('Remote peer closed with request in flight'));
         }
 
         function completePromise(action: () => void) {

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

@@ -249,11 +249,20 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
           // Gets a connection from the clients when a new connection is requested.
           return await this.openInternalDB();
         },
-        logger: this.logger
+        logger: this.logger,
+        reOpenOnConnectionClosed: true
       });
       this.distributedDB = lockedAdapter;
       await lockedAdapter.init();
 
+      lockedAdapter.registerListener({
+        databaseReOpened: () => {
+          // We may have missed some table updates while the database was closed.
+          // We can poke the crud in case we missed any updates.
+          this.connectionManager.syncStreamImplementation?.triggerCrudUpload();
+        }
+      });
+
       self.onerror = (event) => {
         // Share any uncaught events on the broadcast logger
         this.logger.error('Uncaught exception in PowerSync shared sync worker', event);
@@ -480,7 +489,31 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
           throw new Error(`Could not open DB connection since no client is connected.`);
         }
 
-        const workerPort = await withTimeout(() => lastClient.clientProvider.getDBWorkerPort(), 5_000);
+        /**
+         * Handle cases where the client might close while opening a connection.
+         */
+        const abortController = new AbortController();
+        const closeListener = () => {
+          abortController.abort();
+        };
+
+        const removeCloseListener = () => {
+          const index = lastClient.closeListeners.indexOf(closeListener);
+          if (index >= 0) {
+            lastClient.closeListeners.splice(index, 1);
+          }
+        };
+
+        lastClient.closeListeners.push(closeListener);
+
+        const workerPort = await withAbort(
+          () => lastClient.clientProvider.getDBWorkerPort(),
+          abortController.signal
+        ).catch((ex) => {
+          removeCloseListener();
+          throw ex;
+        });
+
         const remote = Comlink.wrap<OpenAsyncDatabaseConnection>(workerPort);
         const identifier = this.syncParams!.dbParams.dbFilename;
 
@@ -490,7 +523,10 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
          * We typically execute the closeListeners using the portMutex in a different context.
          * We can't rely on the closeListeners to abort the operation if the tab is closed.
          */
-        const db = await withTimeout(() => remote(this.syncParams!.dbParams), 5_000);
+        const db = await withAbort(() => remote(this.syncParams!.dbParams), abortController.signal).finally(() => {
+          // We can remove the close listener here since we no longer need it past this point.
+          removeCloseListener();
+        });
 
         const wrapped = new WorkerWrappedAsyncDatabaseConnection({
           remote,
@@ -543,16 +579,29 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
 }
 
 /**
- * Runs the action with a timeout. If the action takes longer than the timeout, the promise will be rejected.
+ * Runs the action with an abort controller.
  */
-function withTimeout<T>(action: () => Promise<T>, timeoutMs: number): Promise<T> {
+function withAbort<T>(action: () => Promise<T>, signal: AbortSignal): Promise<T> {
   return new Promise((resolve, reject) => {
-    const timeout = setTimeout(() => {
-      reject(new Error('Timeout waiting for action'));
-    }, timeoutMs);
+    if (signal.aborted) {
+      reject(new AbortOperation('Operation aborted by abort controller'));
+      return;
+    }
+
+    function handleAbort() {
+      signal.removeEventListener('abort', handleAbort);
+      reject(new AbortOperation('Operation aborted by abort controller'));
+    }
+
+    signal.addEventListener('abort', handleAbort, { once: true });
+
+    function completePromise(action: () => void) {
+      signal.removeEventListener('abort', handleAbort);
+      action();
+    }
+
     action()
-      .then(resolve)
-      .catch(reject)
-      .finally(() => clearTimeout(timeout));
+      .then((data) => completePromise(() => resolve(data)))
+      .catch((e) => completePromise(() => reject(e)));
   });
 }