Fix readTransaction errors on node and op-sqlite drivers (#941)

Author: simolus3

.changeset/pink-elephants-repair.md

@@ -0,0 +1,7 @@
+---
+'@powersync/node': patch
+'@powersync/common': patch
+'@powersync/op-sqlite': patch
+---
+
+Fix `attempt to write a readonly database` error in `readTransaction`.

packages/common/src/db/DBAdapter.ts

@@ -69,7 +69,17 @@ export interface SqlExecutor {
   executeBatch: (query: string, params?: any[][]) => Promise<QueryResult>;
 }
 
-export interface LockContext extends SqlExecutor, DBGetUtils {}
+export interface LockContext extends SqlExecutor, DBGetUtils {
+  /**
+   * How the connection has been opened.
+   *
+   * `writer` indicates that the lock context is capable of writing to the database.
+   * `queryOnly` indicates that the lock context has been opened in a readwrite mode, but a `PRAGMA query_only = TRUE`
+   * disabled writes.
+   * `readOnly` indicates that the lock context has been opened by passing `SQLITE_OPEN_READONLY` to `sqlite3_open_v2`.
+   */
+  connectionType?: 'writer' | 'queryOnly' | 'readOnly';
+}
 
 /**
  * Implements {@link DBGetUtils} on a {@link SqlRunner}.
@@ -263,8 +273,16 @@ class TransactionImplementation extends DBGetUtilsDefaultMixin(BaseTransaction)
   static async runWith<T>(ctx: LockContext, fn: (tx: Transaction) => Promise<T>): Promise<T> {
     let tx = new TransactionImplementation(ctx);
 
+    // For write transactions, use BEGIN IMMEDIATE to immediately obtain a write lock on the database (instead of doing
+    // that on the first statement). If we have a genuine read-only connection, we also use BEGIN IMMEDIATE there: In
+    // WAL mode, that ensures we pin the current state of the database (instead of the state at the first statement in
+    // the transaction). But if we have a "fake" read-only connection implemented through `pragma query_only = true`, we
+    // can't use this trick because it would attempt to lock the connection. So there, we use a regular `BEGIN`
+    // statement.
+    const useBeginImmediate = ctx.connectionType != 'queryOnly';
+
     try {
-      await ctx.execute('BEGIN IMMEDIATE');
+      await ctx.execute(useBeginImmediate ? 'BEGIN IMMEDIATE' : 'BEGIN');
 
       const result = await fn(tx);
       await tx.commit();

packages/node/package.json

@@ -74,6 +74,7 @@
     "@powersync/drizzle-driver": "workspace:*",
     "@types/node": "catalog:",
     "better-sqlite3": "^12.2.0",
+    "@types/better-sqlite3": "^7.6.13",
     "bson": "catalog:",
     "drizzle-orm": "catalog:",
     "js-logger": "catalog:",

packages/node/src/db/BetterSqliteWorker.ts

@@ -64,7 +64,7 @@ class BlockingAsyncDatabase implements AsyncDatabase {
 
 export async function openDatabase(worker: PowerSyncWorkerOptions, options: AsyncDatabaseOpenOptions) {
   const BetterSQLite3Database = await worker.loadBetterSqlite3();
-  const baseDB = new BetterSQLite3Database(options.path);
+  const baseDB = new BetterSQLite3Database(options.path, { readonly: !options.isWriter });
   baseDB.loadExtension(worker.extensionPath(), 'sqlite3_powersync_init');
 
   const asyncDb = new BlockingAsyncDatabase(baseDB);

packages/node/src/db/NodeSqliteWorker.ts

@@ -56,7 +56,7 @@ export async function openDatabase(worker: PowerSyncWorkerOptions, options: Asyn
   // end, since that would make us incompatible with older Node.JS versions.
   const { DatabaseSync } = await dynamicImport('node:sqlite');
 
-  const baseDB = new DatabaseSync(options.path, { allowExtension: true });
+  const baseDB = new DatabaseSync(options.path, { allowExtension: true, readOnly: !options.isWriter });
   baseDB.loadExtension(worker.extensionPath(), 'sqlite3_powersync_init');
 
   return new BlockingNodeDatabase(baseDB, options.isWriter);

packages/node/src/db/RemoteConnection.ts

@@ -19,7 +19,12 @@ class BaseRemoteConnection implements SqlExecutor {
 
   private readonly notifyWorkerClosed = new AbortController();
 
-  constructor(worker: Worker, comlink: Remote<AsyncDatabaseOpener>, database: Remote<AsyncDatabase>) {
+  constructor(
+    worker: Worker,
+    comlink: Remote<AsyncDatabaseOpener>,
+    database: Remote<AsyncDatabase>,
+    private readonly readonly: boolean
+  ) {
     this.worker = worker;
     this.comlink = comlink;
     this.database = database;
@@ -29,6 +34,10 @@ class BaseRemoteConnection implements SqlExecutor {
     });
   }
 
+  public get connectionType() {
+    return this.readonly ? 'readOnly' : 'writer';
+  }
+
   /**
    * Runs the inner function, but appends the stack trace where this function was called. This is useful for workers
    * because stack traces from worker errors are otherwise unrelated to the application issue that has caused them.

packages/node/src/db/WorkerConnectionPool.ts

@@ -130,13 +130,11 @@ export class WorkerConnectionPool extends BaseObserver<DBAdapterListener> implem
         await database.execute("SELECT powersync_update_hooks('install');", []);
       }
 
-      const connection = new RemoteConnection(worker, comlink, database);
+      const connection = new RemoteConnection(worker, comlink, database, !isWriter);
       if (this.options.initializeConnection) {
         await this.options.initializeConnection(connection, isWriter);
       }
-      if (!isWriter) {
-        await connection.execute('pragma query_only = true');
-      } else {
+      if (isWriter) {
         // We only need to enable this on the writer connection.
         // We can get `database is locked` errors if we enable this on concurrently opening read connections.
         await connection.execute('pragma journal_mode = WAL');

packages/node/tests/PowerSyncDatabase.test.ts

@@ -252,3 +252,25 @@ databaseTest('execute batch', async ({ database }) => {
   await database.executeBatch('INSERT INTO users (id, name) VALUES (uuid(), ?)', [['a'], ['b'], ['c']]);
   expect(await database.getAll('SELECT * FROM users')).toHaveLength(3);
 });
+
+databaseTest('read transaction', async ({ database }) => {
+  await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['before tx']);
+
+  let completeHasRead: () => void, completeDidWrite: () => void;
+  const hasReadTx = new Promise<void>((resolve) => (completeHasRead = resolve));
+  const didWrite = new Promise<void>((resolve) => (completeDidWrite = resolve));
+
+  const listsInReadTx = database.readTransaction(async (tx) => {
+    completeHasRead();
+    await didWrite;
+
+    // Because this transaction was started before the write, we shouldn't see it in here.
+    return await tx.getAll<{ name: string }>('SELECT name FROM lists');
+  });
+
+  await hasReadTx;
+  await database.execute('INSERT INTO lists (id, name) VALUES (uuid(), ?)', ['after tx']);
+  completeDidWrite!();
+
+  expect(await listsInReadTx).toEqual([{ name: 'before tx' }]);
+});

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

@@ -13,6 +13,7 @@ import {
 
 export type OPSQLiteConnectionOptions = {
   baseDB: DB;
+  readonly: boolean;
 };
 
 export type OPSQLiteUpdateNotification = {
@@ -116,6 +117,10 @@ class OPSQLiteExecutor extends BaseObserver<DBAdapterListener> implements Omit<S
 }
 
 export class OPSQLiteConnection extends DBGetUtilsDefaultMixin(OPSQLiteExecutor) implements LockContext {
+  get connectionType() {
+    return this.options.readonly ? 'queryOnly' : 'writer';
+  }
+
   async refreshSchema() {
     await this.get("PRAGMA table_info('sqlite_master')");
   }

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

@@ -51,7 +51,7 @@ class OPSQLiteConnectionPool extends BaseObserver<DBAdapterListener> implements
       this.options.sqliteOptions!;
     const dbFilename = this.options.name;
 
-    const underlyingWriteConnection = await this.openConnection(dbFilename);
+    const underlyingWriteConnection = await this.openConnection(false, dbFilename);
 
     const baseStatements = [
       `PRAGMA busy_timeout = ${lockTimeoutMs}`,
@@ -90,7 +90,7 @@ class OPSQLiteConnectionPool extends BaseObserver<DBAdapterListener> implements
 
     const underlyingReadConnections = [];
     for (let i = 0; i < READ_CONNECTIONS; i++) {
-      const conn = await this.openConnection(dbFilename);
+      const conn = await this.openConnection(true, dbFilename);
       for (let statement of readConnectionStatements) {
         await conn.execute(statement);
       }
@@ -101,7 +101,7 @@ class OPSQLiteConnectionPool extends BaseObserver<DBAdapterListener> implements
     this.readConnections = new Semaphore(underlyingReadConnections);
   }
 
-  protected async openConnection(filenameOverride?: string): Promise<OPSQLiteConnection> {
+  protected async openConnection(readonly: boolean, filenameOverride?: string): Promise<OPSQLiteConnection> {
     const dbFilename = filenameOverride ?? this.options.name;
     const DB: DB = this.openDatabase(dbFilename, this.options.sqliteOptions?.encryptionKey ?? undefined);
 
@@ -112,7 +112,8 @@ class OPSQLiteConnectionPool extends BaseObserver<DBAdapterListener> implements
     await DB.execute('SELECT powersync_init()');
 
     return new OPSQLiteConnection({
-      baseDB: DB
+      baseDB: DB,
+      readonly
     });
   }