Listen for database close events to catch closed items earlier.

Author: stevensJourney

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

@@ -132,6 +132,27 @@ export class LockedAsyncDatabaseAdapter
     this.requiresHolds = (this._config as ResolvedWASQLiteOpenFactoryOptions).vfs == WASQLiteVFS.IDBBatchAtomicVFS;
   }
 
+  protected _reOpen() {
+    this.databaseOpenPromise = this.openInternalDB().finally(() => {
+      this.databaseOpenPromise = null;
+    });
+    return this.databaseOpenPromise;
+  }
+
+  /**
+   * Re-opens the underlying database.
+   * Returns a pending operation if one is already in progress.
+   */
+  async reOpenInternalDB(): Promise<void> {
+    if (!this.options.reOpenOnConnectionClosed) {
+      throw new Error(`Cannot re-open underlying database, reOpenOnConnectionClosed is not enabled`);
+    }
+    if (this.databaseOpenPromise) {
+      return this.databaseOpenPromise;
+    }
+    return this._reOpen();
+  }
+
   protected async _init() {
     await this.openInternalDB();
     this.iterateListeners((cb) => cb.initialized?.());
@@ -277,9 +298,7 @@ export class LockedAsyncDatabaseAdapter
           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;
-              });
+              this.reOpenInternalDB();
             }
           }
           throw ex;

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

@@ -1,3 +1,4 @@
+import { BaseObserver } from '@powersync/common';
 import * as Comlink from 'comlink';
 import {
   AsyncDatabaseConnection,
@@ -24,17 +25,23 @@ export type WrappedWorkerConnectionOptions<Config extends ResolvedWebSQLOpenOpti
   onClose?: () => void;
 };
 
+export type WorkerWrappedAsyncDatabaseConnectionListener = {
+  closing: () => void;
+};
 /**
  * Wraps a provided instance of {@link AsyncDatabaseConnection}, providing necessary proxy
  * functions for worker listeners.
  */
 export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLOpenOptions = ResolvedWebSQLOpenOptions>
+  extends BaseObserver<WorkerWrappedAsyncDatabaseConnectionListener>
   implements AsyncDatabaseConnection
 {
   protected lockAbortController = new AbortController();
   protected notifyRemoteClosed: AbortController | undefined;
 
   constructor(protected options: WrappedWorkerConnectionOptions<Config>) {
+    super();
+
     if (options.remoteCanCloseUnexpectedly) {
       this.notifyRemoteClosed = new AbortController();
     }
@@ -169,6 +176,7 @@ export class WorkerWrappedAsyncDatabaseConnection<Config extends ResolvedWebSQLO
     } finally {
       this.options.remote[Comlink.releaseProxy]();
       this.options.onClose?.();
+      this.iterateListeners((l) => l.closing?.());
     }
   }
 

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

@@ -247,7 +247,13 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
       name: params.dbParams.dbFilename,
       openConnection: async () => {
         // Gets a connection from the clients when a new connection is requested.
-        return await this.openInternalDB();
+        const db = await this.openInternalDB();
+        db.registerListener({
+          closing: () => {
+            lockedAdapter.reOpenInternalDB();
+          }
+        });
+        return db;
       },
       logger: this.logger,
       reOpenOnConnectionClosed: true