Merge branch 'main' into chore/add-kysely-test-example

Author: DominicGBauer

.changeset/perfect-zebras-marry.md

@@ -0,0 +1,7 @@
+---
+'@powersync/common': patch
+'@powersync/web': patch
+'@powersync/react-native': patch
+---
+
+Improved `getCrudBatch` to use a default limit of 100 CRUD entries.

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -132,6 +132,8 @@ export const DEFAULT_POWERSYNC_DB_OPTIONS = {
   crudUploadThrottleMs: DEFAULT_CRUD_UPLOAD_THROTTLE_MS
 };
 
+export const DEFAULT_CRUD_BATCH_LIMIT = 100;
+
 /**
  * Requesting nested or recursive locks can block the application in some circumstances.
  * This default lock timeout will act as a failsafe to throw an error if a lock cannot
@@ -492,7 +494,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * data by transaction. One batch may contain data from multiple transactions,
    * and a single transaction may be split over multiple batches.
    */
-  async getCrudBatch(limit: number): Promise<CrudBatch | null> {
+  async getCrudBatch(limit: number = DEFAULT_CRUD_BATCH_LIMIT): Promise<CrudBatch | null> {
     const result = await this.getAll<CrudEntryJSON>(
       `SELECT id, tx_id, data FROM ${PSInternalTable.CRUD} ORDER BY id ASC LIMIT ?`,
       [limit + 1]

packages/powersync-op-sqlite/CHANGELOG.md

@@ -1,5 +1,11 @@
 # @powersync/op-sqlite
 
+## 0.0.6
+
+### Patch Changes
+
+- 2b614bc: Improved queueing for read connections
+
 ## 0.0.5
 
 ### Patch Changes

packages/powersync-op-sqlite/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@powersync/op-sqlite",
-  "version": "0.0.5",
+  "version": "0.0.6",
   "description": "PowerSync - sync Postgres or MongoDB with SQLite in your React Native app for offline-first and real-time data",
   "source": "./src/index.ts",
   "main": "./lib/commonjs/index.js",

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

@@ -35,10 +35,12 @@ export class OPSQLiteDBAdapter extends BaseObserver<DBAdapterListener> implement
 
   protected initialized: Promise<void>;
 
-  protected readConnections: OPSQLiteConnection[] | null;
+  protected readConnections: Array<{ busy: boolean; connection: OPSQLiteConnection }> | null;
 
   protected writeConnection: OPSQLiteConnection | null;
 
+  private readQueue: Array<() => void> = [];
+
   constructor(protected options: OPSQLiteAdapterOptions) {
     super();
     this.name = this.options.name;
@@ -88,7 +90,7 @@ export class OPSQLiteDBAdapter extends BaseObserver<DBAdapterListener> implement
       let dbName = './'.repeat(i + 1) + dbFilename;
       const conn = await this.openConnection(dbName);
       await conn.execute('PRAGMA query_only = true');
-      this.readConnections.push(conn);
+      this.readConnections.push({ busy: false, connection: conn });
     }
   }
 
@@ -145,36 +147,46 @@ export class OPSQLiteDBAdapter extends BaseObserver<DBAdapterListener> implement
   close() {
     this.initialized.then(() => {
       this.writeConnection!.close();
-      this.readConnections!.forEach((c) => c.close());
+      this.readConnections!.forEach((c) => c.connection.close());
     });
   }
 
   async readLock<T>(fn: (tx: OPSQLiteConnection) => Promise<T>, options?: DBLockOptions): Promise<T> {
     await this.initialized;
-    // TODO: Use async queues to handle multiple read connections
-    const sortedConnections = this.readConnections!.map((connection, index) => ({
-      lockKey: `${LockType.READ}-${index}`,
-      connection
-    })).sort((a, b) => {
-      const aBusy = this.locks.isBusy(a.lockKey);
-      const bBusy = this.locks.isBusy(b.lockKey);
-      // Sort by ones which are not busy
-      return aBusy > bBusy ? 1 : 0;
+    return new Promise(async (resolve, reject) => {
+      const execute = async () => {
+        // Find an available connection that is not busy
+        const availableConnection = this.readConnections!.find((conn) => !conn.busy);
+
+        // If we have an available connection, use it
+        if (availableConnection) {
+          availableConnection.busy = true;
+          try {
+            resolve(await fn(availableConnection.connection));
+          } catch (error) {
+            reject(error);
+          } finally {
+            availableConnection.busy = false;
+            // After query execution, process any queued tasks
+            this.processQueue();
+          }
+        } else {
+          // If no available connections, add to the queue
+          this.readQueue.push(execute);
+        }
+      };
+
+      execute();
     });
+  }
 
-    return new Promise(async (resolve, reject) => {
-      try {
-        await this.locks.acquire(
-          sortedConnections[0].lockKey,
-          async () => {
-            resolve(await fn(sortedConnections[0].connection));
-          },
-          { timeout: options?.timeoutMs }
-        );
-      } catch (ex) {
-        reject(ex);
+  private async processQueue(): Promise<void> {
+    if (this.readQueue.length > 0) {
+      const next = this.readQueue.shift();
+      if (next) {
+        next();
       }
-    });
+    }
   }
 
   async writeLock<T>(fn: (tx: OPSQLiteConnection) => Promise<T>, options?: DBLockOptions): Promise<T> {

packages/web/tests/crud.test.ts

@@ -1,9 +1,9 @@
-import { afterEach, beforeEach, describe, expect, it } from 'vitest';
 import { AbstractPowerSyncDatabase, Column, ColumnType, CrudEntry, Schema, Table, UpdateType } from '@powersync/common';
 import { PowerSyncDatabase } from '@powersync/web';
+import pDefer from 'p-defer';
 import { v4 as uuid } from 'uuid';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
 import { generateTestDb } from './utils/testDb';
-import pDefer from 'p-defer';
 
 const testId = '2290de4f-0488-4e50-abed-f8e8eb1d0b42';
 
@@ -311,4 +311,35 @@ describe('CRUD Tests', () => {
     await txPromise;
     expect(await r).toEqual(null);
   });
+
+  it('CRUD Batch Limits', async () => {
+    const initialBatch = await powersync.getCrudBatch();
+    expect(initialBatch, 'Initial CRUD batch should be null').null;
+
+    /**
+     * Create some items. Use Multiple transactions to demonstrate getCrudBatch does not
+     * group by transaction.
+     */
+    for (let i = 0; i < 2; i++) {
+      await powersync.writeTransaction(async (tx) => {
+        for (let j = 0; j < 51; j++) {
+          await tx.execute(`INSERT INTO assets(id, description) VALUES(uuid(), ?)`, [`test-${i}-${j}`]);
+        }
+      });
+    }
+
+    // This should contain CRUD entries for the first and second transaction
+    const smallBatch = await powersync.getCrudBatch(55);
+    expect(smallBatch, 'CRUD items should be present').exist;
+    expect(smallBatch?.crud.length, 'Should only be 55 CRUD items').eq(55);
+    expect(smallBatch?.haveMore, 'There should be more CRUD items pending').true;
+
+    const defaultBatch = await powersync.getCrudBatch();
+    expect(defaultBatch?.crud.length, 'Should use default limit').eq(100);
+    expect(defaultBatch?.haveMore, 'There should be more CRUD items pending').true;
+
+    const maxBatch = await powersync.getCrudBatch(1000);
+    expect(maxBatch?.crud.length, 'Should contain all created CRUD items').eq(102);
+    expect(maxBatch?.haveMore, 'There should not be any more pending CRUD items').false;
+  });
 });