debug wal

Author: simolus3

packages/shared-internals/src/utils/mutex.ts

@@ -49,7 +49,7 @@ export class Semaphore<T> {
     if (waiter == this.lastWaiter) this.lastWaiter = prev;
   }
 
-  private requestPermits(amount: number, abort?: AbortSignal): Promise<{ items: T[]; release: UnlockFn }> {
+  requestPermits(amount: number, abort?: AbortSignal): Promise<{ items: T[]; release: UnlockFn }> {
     if (amount <= 0 || amount > this.size) {
       throw new Error(`Invalid amount of items requested (${amount}), must be between 1 and ${this.size}`);
     }

packages/web/package.json

@@ -23,6 +23,10 @@
       "react-native": "./dist/index.react_native_web.js",
       "default": "./lib/index.js"
     },
+    "./in-memory-wal-experiment": {
+      "types": "./lib/db/adapters/memory/client.d.ts",
+      "default": "./lib/db/adapters/memory/client.js"
+    },
     "./bundled_worker": {
       "types": "./lib/worker/worker.d.ts",
       "default": "./dist/worker/worker.js"

packages/web/src/db/adapters/memory/client.ts

@@ -1,6 +1,6 @@
 import * as Comlink from 'comlink';
 import { DBAdapter, DBLockOptions, LockContext, RawQueryResult } from '@powersync/common';
-import { DatabaseServer, emptyWalState, WalIndexChange, WriteAheadBuffers, WriteAheadState } from './shared.js';
+import { applyWalChanges, DatabaseServer, emptyWalState, WalIndexChange, WriteAheadBuffers } from './shared.js';
 import { Mutex, Semaphore } from '@powersync/shared-internals';
 
 function createWriteAheadLogBuffers(): WriteAheadBuffers {
@@ -23,6 +23,7 @@ export class InMemoryWriteAheadLogPool extends DBAdapter {
   readonly #rawWorkers: PoolWorker[] = [];
   readonly #workers: Semaphore<PoolWorker>;
   readonly #writeLock = new Mutex();
+  readonly #walState = emptyWalState();
 
   constructor(options: InMemoryOptions) {
     super();
@@ -58,20 +59,50 @@ export class InMemoryWriteAheadLogPool extends DBAdapter {
     }, options);
   }
 
+  #checkpoint() {
+    const walBuffer = this.#buffers.writeAheadLog;
+    const databaseBuffer = this.#buffers.database;
+    const newFileSize = this.#walState.fileSize;
+    if (databaseBuffer.byteLength < newFileSize) {
+      databaseBuffer.grow(newFileSize);
+    }
+
+    for (const [pageOffset, overlayEntry] of this.#walState.overlay.entries()) {
+      const source = new Uint8Array(walBuffer, overlayEntry.logOffset, overlayEntry.size);
+      new Uint8Array(databaseBuffer, pageOffset).set(source);
+    }
+
+    const cleared: WalIndexChange = { cleared: true, fileSize: newFileSize, walEnd: 0, added: [] };
+    applyWalChanges(this.#walState, cleared);
+    for (const worker of this.#rawWorkers) {
+      worker.addChanges(cleared);
+    }
+  }
+
   writeLock<T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions): Promise<T> {
     return this.#writeLock.runExclusive(() => {
       return this.#withWorker(async (worker) => {
         try {
           return await fn(worker);
         } finally {
           const changes = await worker.takeWalChanges();
+          applyWalChanges(this.#walState, changes);
 
           if (changes.walEnd > 4096 * 128) {
-          }
-
-          for (const otherWorker of this.#rawWorkers) {
-            if (otherWorker !== worker) {
-              otherWorker.addChanges(changes);
+            // Checkpoint. This can't run concurrently to anything else, so acquire remaining workers.
+            const remainingWorkers = this.#workers.size - 1;
+            if (remainingWorkers) {
+              const { release } = await this.#workers.requestPermits(this.#workers.size - 1);
+              this.#checkpoint();
+              release();
+            } else {
+              this.#checkpoint();
+            }
+          } else {
+            for (const otherWorker of this.#rawWorkers) {
+              if (otherWorker !== worker) {
+                otherWorker.addChanges(changes);
+              }
             }
           }
         }
@@ -113,7 +144,10 @@ class PoolWorker extends LockContext {
   constructor(buffers: WriteAheadBuffers) {
     super();
     this.#buffers = buffers;
-    this.#worker = new Worker('./worker.js', { type: 'module' });
+    this.#worker = new Worker(new URL('./worker.js', import.meta.url), { type: 'module' });
+    this.#worker.onerror = (e) => {
+      console.error('Worker error', e);
+    };
     this.#server = Comlink.wrap(this.#worker);
   }
 

packages/web/src/db/adapters/memory/shared.ts

@@ -14,14 +14,19 @@ export interface WriteAheadState {
   /**
    * A map of database position offsets to offsets in the WAL logs at which position contents of the page are stored.
    */
-  overlay: Map<number, number>;
+  overlay: Map<number, WalOverlayEntry>;
+}
+
+export interface WalOverlayEntry {
+  logOffset: number;
+  size: number;
 }
 
 export interface WalIndexChange {
   fileSize: number;
   walEnd: number;
 
-  added: number[];
+  added: (number | WalOverlayEntry)[];
   cleared: boolean;
 }
 
@@ -37,29 +42,17 @@ export function emptyWalState(): WriteAheadState {
   return { fileSize: 0, walEnd: 0, overlay: new Map() };
 }
 
-/**
-
-Main tab:
-
-state:
- - mutex around writer
- - semaphore around readers
+export function applyWalChanges(state: WriteAheadState, changes: WalIndexChange) {
+  state.fileSize = changes.fileSize;
+  state.walEnd = changes.walEnd;
 
-to start read:
- - acquire from semaphore.
- - send a wal patch if necessary.
- - ... use!
+  if (changes.cleared) {
+    state.overlay.clear();
+  }
 
-to start write:
-  - acquire from mutex
-  - ... use, updating write-ahead log offset
-  - update overlay index if offset has changed
-  - if len(overlay) > 500
-    - acquire all readers
-    - checkpoint, incrementing epoch
-
-Reader:
-
- - obtain overlay from main tab
-
- */
+  for (let i = 0; i < changes.added.length; i += 2) {
+    const dbOffset = changes.added[i] as number;
+    const walOffset = changes.added[i + 1] as WalOverlayEntry;
+    state.overlay.set(dbOffset, walOffset);
+  }
+}

packages/web/src/db/adapters/memory/vfs.ts

@@ -1,7 +1,7 @@
 // @ts-ignore
 import { FacadeVFS } from '@journeyapps/wa-sqlite/src/FacadeVFS.js';
 import * as VFS from '@journeyapps/wa-sqlite/src/VFS.js';
-import { emptyWalState, WalIndexChange, WriteAheadBuffers, WriteAheadState } from './shared.js';
+import { emptyWalState, WalIndexChange, WalOverlayEntry, WriteAheadBuffers, WriteAheadState } from './shared.js';
 
 const mainDbSentinel = Symbol();
 
@@ -10,7 +10,7 @@ export class InMemoryWriteAheadLog extends FacadeVFS {
   #files = new Map<string, LocalFile | typeof mainDbSentinel>();
 
   #tx: WriteAheadTransaction | undefined = undefined;
-  #newOverlayPagesToSendToMainTab = new Map<number, number>();
+  #newOverlayPagesToSendToMainTab = new Map<number, WalOverlayEntry>();
 
   writeAheadState: WriteAheadState = emptyWalState();
 
@@ -23,7 +23,7 @@ export class InMemoryWriteAheadLog extends FacadeVFS {
   }
 
   takeChanges(): WalIndexChange {
-    const added: number[] = [];
+    const added: (number | WalOverlayEntry)[] = [];
     this.#newOverlayPagesToSendToMainTab.forEach((v, k) => {
       added.push(k);
       added.push(v);
@@ -40,7 +40,7 @@ export class InMemoryWriteAheadLog extends FacadeVFS {
 
   jAccess(zName: string, _pFlags: number, pResOut: DataView) {
     const file = this.#files.get(zName);
-    pResOut.setInt32(0, file ? 1 : 0, true);
+    pResOut.setInt32(0, file != null ? 1 : 0, true);
     return VFS.SQLITE_OK;
   }
 
@@ -97,7 +97,12 @@ export class InMemoryWriteAheadLog extends FacadeVFS {
       const page = overlay.get(offset < 100 ? 0 : offset);
 
       if (page != null) {
-        const source = new Uint8Array(this.buffers.writeAheadLog, offset < 100 ? page + offset : page, readableBytes);
+        const pageOffset = page.logOffset;
+        const source = new Uint8Array(
+          this.buffers.writeAheadLog,
+          offset < 100 ? pageOffset + offset : pageOffset,
+          readableBytes
+        );
         data.set(source);
       } else {
         // Page is not in WAL overlay, read directly from underlying in-memory buffer.
@@ -114,7 +119,7 @@ export class InMemoryWriteAheadLog extends FacadeVFS {
       }
     }
 
-    if (endOffset < data.byteLength) {
+    if (readableBytes < data.byteLength) {
       data.fill(0, readableBytes); // Fill rest with zeroes.
       return VFS.SQLITE_IOERR_SHORT_READ;
     }
@@ -179,6 +184,15 @@ export class InMemoryWriteAheadLog extends FacadeVFS {
     return VFS.SQLITE_OK;
   }
 
+  jDelete(name: string) {
+    if (name === '/database') {
+      return VFS.SQLITE_IOERR_DELETE;
+    }
+
+    this.#files.delete(name);
+    return VFS.SQLITE_OK;
+  }
+
   jDeviceCharacteristics(): number {
     return VFS.SQLITE_IOCAP_UNDELETABLE_WHEN_OPEN | VFS.SQLITE_IOCAP_BATCH_ATOMIC;
   }
@@ -208,17 +222,17 @@ export class InMemoryWriteAheadLog extends FacadeVFS {
   #commit(tx: WriteAheadTransaction) {
     this.writeAheadState.walEnd = tx.walEndOffset;
     this.writeAheadState.fileSize = tx.fileSize;
-    tx.changedPages.forEach((walOffset, databaseOffset) => {
-      this.#newOverlayPagesToSendToMainTab.set(databaseOffset, walOffset);
-      this.writeAheadState.overlay.set(databaseOffset, walOffset);
+    tx.changedPages.forEach((walEntry, databaseOffset) => {
+      this.#newOverlayPagesToSendToMainTab.set(databaseOffset, walEntry);
+      this.writeAheadState.overlay.set(databaseOffset, walEntry);
     });
   }
 }
 
 class WriteAheadTransaction {
   walEndOffset: number;
   fileSize: number;
-  changedPages = new Map<number, number>();
+  changedPages = new Map<number, WalOverlayEntry>();
 
   constructor(
     endOffset: number,
@@ -241,7 +255,7 @@ class WriteAheadTransaction {
     new Uint8Array(wal, currentEnd, data.length).set(data);
 
     this.walEndOffset = newEnd;
-    this.changedPages.set(offset, currentEnd);
+    this.changedPages.set(offset, { logOffset: currentEnd, size: data.length });
     return true;
   }
 

packages/web/src/db/adapters/memory/worker.ts

@@ -1,5 +1,5 @@
 import * as Comlink from 'comlink';
-import { DatabaseServer, WalIndexChange, WriteAheadBuffers } from './shared.js';
+import { applyWalChanges, DatabaseServer, WalIndexChange, WriteAheadBuffers } from './shared.js';
 import { RawQueryResult } from '@powersync/common';
 import { InMemoryWriteAheadLog } from './vfs.js';
 import { RawSqliteConnection } from '../wa-sqlite/RawSqliteConnection.js';
@@ -30,23 +30,11 @@ class MemoryDatabaseServer implements DatabaseServer {
 
   async updateWalState(overlay: WalIndexChange): Promise<void> {
     const currentState = this.#vfs.writeAheadState;
-    currentState.fileSize = overlay.fileSize;
-    currentState.walEnd = overlay.walEnd;
-
-    if (overlay.cleared) {
-      currentState.overlay.clear();
-    }
-
-    for (let i = 0; i < overlay.added.length; i += 2) {
-      const dbOffset = overlay.added[i];
-      const walOffset = overlay.added[i + 1];
-      currentState.overlay.set(dbOffset, walOffset);
-    }
+    applyWalChanges(currentState, overlay);
   }
 
   async executeRaw(query: string, params?: any[] | undefined): Promise<RawQueryResult> {
-    const results = await this.#connection.executeRaw(query, params);
-    return results[0];
+    return await this.#connection.execute(query, params);
   }
 
   async takeWalChanges(): Promise<WalIndexChange> {

packages/web/src/db/adapters/wa-sqlite/RawSqliteConnection.ts

@@ -61,7 +61,7 @@ export class RawSqliteConnection {
   }
 
   async initWithModule(module: any, vfs: any) {
-    const api = await this.openSQLiteAPI(module, vfs);
+    const api = (this._sqliteAPI = await this.openSQLiteAPI(module, vfs));
     this.db = await api.open_v2(
       this.options.filename,
       this.options.readonly ? 1 /* SQLITE_OPEN_READONLY */ : 6 /* SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE */

packages/web/tests/main.test.ts

@@ -1,4 +1,5 @@
 import { PowerSyncDatabase, WASQLiteOpenFactory, WASQLiteVFS } from '@powersync/web';
+import { InMemoryWriteAheadLogPool } from '@powersync/web/in-memory-wal-experiment';
 import { v4 as uuid } from 'uuid';
 import { describe, expect, it } from 'vitest';
 import { TEST_SCHEMA, TestDatabase } from './utils/test-schema.js';
@@ -94,6 +95,13 @@ describe('Basic - with in-memory', () => {
       })
     )
   );
+
+  describe(
+    'wal',
+    describeBasicTests(() =>
+      generateTestDb({ schema: TEST_SCHEMA, opened: new InMemoryWriteAheadLogPool({ numWorkers: 1 }) })
+    )
+  );
 });
 
 function describeBasicTests(generateDB: () => PowerSyncDatabase) {

packages/web/vitest.config.ts

@@ -17,6 +17,7 @@ const config: ViteUserConfigExport = {
        * first. This is required due to the format of Webworker URIs
        * they link to `.js` files.
        */
+      '@powersync/web/in-memory-wal-experiment': path.resolve(__dirname, './lib/db/adapters/memory/client.js'),
       '@powersync/web': path.resolve(__dirname, './lib'),
       // Mock WebRemote to throw 401 errors for all HTTP requests in tests
       '../../db/sync/WebRemote.js': path.resolve(__dirname, './tests/mocks/MockWebRemote.ts')