fix unit tests. Improve closing behaviour.

Author: stevensJourney

.changeset/little-bananas-fetch.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': minor
+---
+
+Added additional listeners for `closing` and `closed` events in `AbstractPowerSyncDatabase`.

.changeset/stale-dots-jog.md

@@ -0,0 +1,5 @@
+---
+'@powersync/web': minor
+---
+
+Improved query behaviour when client is closed. Pending requests will be aborted, future requests will be rejected with an Error. Fixed read and write lock requests not respecting timeout parameter.

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -125,7 +125,8 @@ export interface WatchOnChangeHandler {
 export interface PowerSyncDBListener extends StreamingSyncImplementationListener {
   initialized: () => void;
   schemaChanged: (schema: Schema) => void;
-  closing: () => void;
+  closing: () => Promise<void> | void;
+  closed: () => Promise<void> | void;
 }
 
 export interface PowerSyncCloseOptions {
@@ -532,7 +533,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       return;
     }
 
-    this.iterateListeners((cb) => cb.closing?.());
+    await this.iterateAsyncListeners(async (cb) => cb.closing?.());
 
     const { disconnect } = options;
     if (disconnect) {
@@ -542,6 +543,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     await this.syncStreamImplementation?.dispose();
     await this.database.close();
     this.closed = true;
+    await this.iterateAsyncListeners(async (cb) => cb.closed?.());
   }
 
   /**

packages/common/src/client/watched/processors/AbstractQueryProcessor.ts

@@ -119,8 +119,8 @@ export abstract class AbstractQueryProcessor<Data = unknown[]>
     const { db } = this.options;
 
     const disposeCloseListener = db.registerListener({
-      closed: async () => {
-        this.close();
+      closing: async () => {
+        await this.close();
       }
     });
 

packages/common/src/client/watched/processors/OnChangeQueryProcessor.ts

@@ -12,37 +12,6 @@ export interface OnChangeQueryProcessorOptions<Data> extends AbstractQueryProces
   comparator?: WatchedQueryComparator<Data>;
 }
 
-/**
- * @internal
- */
-export class ArrayComparator<Element> implements WatchedQueryComparator<Element[]> {
-  constructor(protected compareBy: (element: Element) => string) {}
-
-  checkEquality(current: Element[], previous: Element[]) {
-    if (current.length == 0 && previous.length == 0) {
-      return true;
-    }
-
-    if (current.length !== previous.length) {
-      return false;
-    }
-
-    const { compareBy } = this;
-
-    // At this point the lengths are equal
-    for (let i = 0; i < current.length; i++) {
-      const currentItem = compareBy(current[i]);
-      const previousItem = compareBy(previous[i]);
-
-      if (currentItem !== previousItem) {
-        return false;
-      }
-    }
-
-    return true;
-  }
-}
-
 /**
  * Uses the PowerSync onChange event to trigger watched queries.
  * Results are emitted on every change of the relevant tables.
@@ -71,9 +40,12 @@ export class OnChangeQueryProcessor<Data> extends AbstractQueryProcessor<Data> {
     db.onChangeWithCallback(
       {
         onChange: async () => {
+          if (this.closed) {
+            return;
+          }
           // This fires for each change of the relevant tables
           try {
-            if (this.reportFetching) {
+            if (this.reportFetching && !this.state.isFetching) {
               await this.updateState({ isFetching: true });
             }
 

packages/react/vitest.config.ts

@@ -32,7 +32,7 @@ const config: UserConfigExport = {
        */
       isolate: true,
       provider: 'playwright',
-      headless: false,
+      headless: true,
       instances: [
         {
           browser: 'chromium'

packages/vue/vitest.config.ts

@@ -32,7 +32,7 @@ const config: UserConfigExport = {
        */
       isolate: true,
       provider: 'playwright',
-      headless: false,
+      headless: true,
       instances: [
         {
           browser: 'chromium'

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

@@ -47,11 +47,18 @@ export class LockedAsyncDatabaseAdapter
   private _db: AsyncDatabaseConnection | null = null;
   protected _disposeTableChangeListener: (() => void) | null = null;
   private _config: ResolvedWebSQLOpenOptions | null = null;
+  protected pendingAbortControllers: Set<AbortController>;
+
+  closing: boolean;
+  closed: boolean;
 
   constructor(protected options: LockedAsyncDatabaseAdapterOptions) {
     super();
     this._dbIdentifier = options.name;
     this.logger = options.logger ?? createLogger(`LockedAsyncDatabaseAdapter - ${this._dbIdentifier}`);
+    this.pendingAbortControllers = new Set<AbortController>();
+    this.closed = false;
+    this.closing = false;
     // Set the name if provided. We can query for the name if not available yet
     this.debugMode = options.debugMode ?? false;
     if (this.debugMode) {
@@ -154,8 +161,11 @@ export class LockedAsyncDatabaseAdapter
    * tabs are still using it.
    */
   async close() {
+    this.closing = true;
     this._disposeTableChangeListener?.();
+    this.pendingAbortControllers.forEach((controller) => controller.abort('Closed'));
     await this.baseDB?.close?.();
+    this.closed = true;
   }
 
   async getAll<T>(sql: string, parameters?: any[] | undefined): Promise<T[]> {
@@ -175,20 +185,49 @@ export class LockedAsyncDatabaseAdapter
 
   async readLock<T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions | undefined): Promise<T> {
     await this.waitForInitialized();
-    return this.acquireLock(async () =>
-      fn(this.generateDBHelpers({ execute: this._execute, executeRaw: this._executeRaw }))
+    return this.acquireLock(
+      async () => fn(this.generateDBHelpers({ execute: this._execute, executeRaw: this._executeRaw })),
+      {
+        timeoutMs: options?.timeoutMs
+      }
     );
   }
 
   async writeLock<T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions | undefined): Promise<T> {
     await this.waitForInitialized();
-    return this.acquireLock(async () =>
-      fn(this.generateDBHelpers({ execute: this._execute, executeRaw: this._executeRaw }))
+    return this.acquireLock(
+      async () => fn(this.generateDBHelpers({ execute: this._execute, executeRaw: this._executeRaw })),
+      {
+        timeoutMs: options?.timeoutMs
+      }
     );
   }
 
-  protected acquireLock(callback: () => Promise<any>): Promise<any> {
-    return getNavigatorLocks().request(`db-lock-${this._dbIdentifier}`, callback);
+  protected async acquireLock(callback: () => Promise<any>, options?: { timeoutMs?: number }): Promise<any> {
+    await this.waitForInitialized();
+
+    if (this.closing) {
+      throw new Error(`Cannot acquire lock, the database is closing`);
+    }
+
+    const abortController = new AbortController();
+    this.pendingAbortControllers.add(abortController);
+    const { timeoutMs } = options ?? {};
+
+    const timoutId = timeoutMs
+      ? setTimeout(() => {
+          abortController.abort(`Timeout after ${timeoutMs}ms`);
+          this.pendingAbortControllers.delete(abortController);
+        }, timeoutMs)
+      : null;
+
+    return getNavigatorLocks().request(`db-lock-${this._dbIdentifier}`, { signal: abortController.signal }, () => {
+      this.pendingAbortControllers.delete(abortController);
+      if (timoutId) {
+        clearTimeout(timoutId);
+      }
+      return callback();
+    });
   }
 
   async readTransaction<T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions | undefined): Promise<T> {
@@ -286,6 +325,7 @@ export class LockedAsyncDatabaseAdapter
    */
   private _execute = async (sql: string, bindings?: any[]): Promise<QueryResult> => {
     await this.waitForInitialized();
+
     const result = await this.baseDB.execute(sql, bindings);
     return {
       ...result,

packages/web/tests/watch.test.ts

@@ -346,7 +346,7 @@ describe('Watch Tests', { sequential: true }, () => {
         });
       });
 
-    let state = await getNextState();
+    let state = watch.state;
     expect(state.isFetching).true;
     expect(state.isLoading).true;
 

packages/web/vitest.config.ts

@@ -50,7 +50,7 @@ const config: UserConfigExport = {
        */
       isolate: true,
       provider: 'playwright',
-      headless: false,
+      headless: true,
       instances: [
         {
           browser: 'chromium'