Rejecting pending read/write operations when the database is closed for op-sqlite.

Author: Chriztiaan

.changeset/khaki-bottles-sniff.md

@@ -0,0 +1,5 @@
+---
+'@powersync/op-sqlite': patch
+---
+
+Rejecting pending read/write operations when the database is closed.

packages/powersync-op-sqlite/src/db/OPSqliteAdapter.ts

@@ -32,6 +32,7 @@ export class OPSQLiteDBAdapter extends BaseObserver<DBAdapterListener> implement
   protected writeConnection: OPSQLiteConnection | null;
 
   private readQueue: Array<() => void> = [];
+  private abortController: AbortController;
 
   constructor(protected options: OPSQLiteAdapterOptions) {
     super();
@@ -40,6 +41,7 @@ export class OPSQLiteDBAdapter extends BaseObserver<DBAdapterListener> implement
     this.locks = new Lock();
     this.readConnections = null;
     this.writeConnection = null;
+    this.abortController = new AbortController();
     this.initialized = this.init();
   }
 
@@ -155,6 +157,10 @@ export class OPSQLiteDBAdapter extends BaseObserver<DBAdapterListener> implement
 
   close() {
     this.initialized.then(() => {
+      // Abort any pending operations
+      this.abortController.abort();
+      this.readQueue = [];
+
       this.writeConnection!.close();
       this.readConnections!.forEach((c) => c.connection.close());
     });
@@ -203,17 +209,30 @@ export class OPSQLiteDBAdapter extends BaseObserver<DBAdapterListener> implement
 
     return new Promise(async (resolve, reject) => {
       try {
+        // Set up abort signal listener
+        const abortListener = () => {
+          reject(new Error('Database connection was closed'));
+        };
+        this.abortController.signal.addEventListener('abort', abortListener);
+
         await this.locks
           .acquire(
             LockType.WRITE,
             async () => {
+              // Check if operation was aborted before executing
+              if (this.abortController.signal.aborted) {
+                reject(new Error('Database connection was closed'));
+              }
               resolve(await fn(this.writeConnection!));
             },
             { timeout: options?.timeoutMs }
           )
           .then(() => {
             // flush updates once a write lock has been released
             this.writeConnection!.flushUpdates();
+          })
+          .finally(() => {
+            this.abortController.signal.removeEventListener('abort', abortListener);
           });
       } catch (ex) {
         reject(ex);

tools/powersynctests/README.md

@@ -46,7 +46,7 @@ pnpm android
 ### iOS
 
 ```sh
-detox test --configuration ios.sim.debug
+pnpm detox test --configuration ios.sim.debug
 ```
 
 ### Android

tools/powersynctests/src/tests/queries.test.ts

@@ -575,42 +575,39 @@ export function registerBaseTests() {
       expect(duration).lessThan(2000);
     });
 
-    //     // TODO: Behaviour differs between RNQS and OP-sqlite. Might need to fix this one in the OP-sqlite package
-    // it('Should handle multiple closes', async () => {
-    //   // // Bulk insert 50000 rows without using a transaction
-    //   const bulkInsertCommands = [];
-    //   const statement = `INSERT INTO t1(id, c) VALUES(?, ?)`;
-
-    //   for (let i = 0; i < 50000; i++) {
-    //     bulkInsertCommands.push([[i + 1, `value${i + 1}`]]);
-    //   }
-
-    //   await db.executeBatch(statement, bulkInsertCommands);
-    //   db.close();
-
-    //   for (let i = 1; i < 10; i++) {
-    //     db = createDatabase();
-
-    //     // ensure a regular query works
-    //     const pExecute = await db.execute(`SELECT * FROM t1 `);
-    //     expect(pExecute.rows?.length).to.equal(50000);
-
-    //     // Queue a bunch of write locks, these will fail due to the db being closed
-    //     // before they are accepted.
-    //     const tests = [
-    //       db.execute(`SELECT * FROM t1 `),
-    //       db.execute(`SELECT * FROM t1 `),
-    //       db.execute(`SELECT * FROM t1 `),
-    //       db.execute(`SELECT * FROM t1 `),
-    //     ];
-
-    //     await db.close();
-
-    //     // const results = await Promise.allSettled(tests);
-    //     // expect(results.map(r => r.status)).deep.equal(
-    //     //   Array(tests.length).fill('rejected'),
-    //     // );
-    //   }
-    // });
+    it('Should handle multiple closes', async () => {
+      // Bulk insert 10000 rows without using a transaction
+      const bulkInsertCommands = [];
+      const statement = `INSERT INTO t1(id, c) VALUES(uuid(), ?)`;
+
+      for (let i = 0; i < 10000; i++) {
+        bulkInsertCommands.push([[`value${i + 1}`]]);
+      }
+
+      await db.executeBatch(statement, bulkInsertCommands);
+      db.close();
+
+      for (let i = 1; i < 10; i++) {
+        db = createDatabase();
+
+        // ensure a regular query works
+        const pExecute = await db.execute(`SELECT * FROM t1 `);
+        expect(pExecute.rows?.length).to.equal(10000);
+
+        // Queue a bunch of write locks, these will fail due to the db being closed
+        // before they are accepted.
+        const tests = [
+          db.execute(`SELECT * FROM t1 `),
+          db.execute(`SELECT * FROM t1 `),
+          db.execute(`SELECT * FROM t1 `),
+          db.execute(`SELECT * FROM t1 `)
+        ];
+
+        db.close();
+
+        const results = await Promise.allSettled(tests);
+        expect(results.map((r) => r.status)).deep.equal(Array(tests.length).fill('rejected'));
+      }
+    });
   });
 }