test: acceptance of the paged restore, and its benchmark

README.md

@@ -645,6 +645,30 @@ Conditions are disclosed in the output (rows, batch/transactions, source WAL byt
 process-wide (it includes the Node baseline), so the meaningful claim is that it stays **flat** as the
 database grows — see the `SCALE` mode, which shows WAL bytes and peak RSS at two sizes.
 
+[`bench/bench-paged-restore.ts`](bench/bench-paged-restore.ts) compares a full restore with a
+[paged restore](#restore-page-by-page) as the database grows. The replica is read through a simulated
+network (`LATENCY_MS` per request, one link of `BANDWIDTH_MBPS`), and a worker thread reads a random row
+as soon as the paged restore resolves:
+
+```bash
+pnpm test:paged-vfs   # builds the paged VFS
+pnpm bench:paged
+SIZES=16,64 LATENCY_MS=5 BANDWIDTH_MBPS=500 pnpm bench:paged
+```
+
+On an Apple M4 Max, Node 24.18, with 20 ms per request and 100 MB/s:
+
+| Database | Full restore | Paged: open | Paged: first response | Paged: every page |
+| --- | --- | --- | --- | --- |
+| 16 MiB | 1.31 s | 298 ms | 335 ms | 510 ms |
+| 64 MiB | 4.30 s | 318 ms | 373 ms | 951 ms |
+| 257 MiB | 16.03 s | 343 ms | 370 ms | 2.78 s |
+| 1027 MiB | 64.02 s | 447 ms | 528 ms | 10.09 s |
+
+The first response of a paged restore barely grows with the database: before it resolves, it reads the
+page indexes of the restore plan (a few bytes per page) and the schema, and a query then waits only for
+the pages it reads. Most of the time to open is listing the ten levels of the replica, one request each.
+
 ### Compaction & retention
 
 Over time `sync()` writes one level-0 LTX file per transaction. To keep the replica bounded and

bench/bench-paged-restore.ts

@@ -0,0 +1,188 @@
+/**
+ * Time to first response of a paged restore, against a full restore, as the
+ * database grows.
+ *
+ * For each size, a database of incompressible rows (about one per page) is
+ * replicated to a file replica, which is then read through a simulated
+ * network: each request waits LATENCY_MS, and the bytes share one link of
+ * BANDWIDTH_MBPS, as reads from S3 would. Then the database is restored:
+ *
+ *   - full:  restore() of the replica's own path, until it resolves, after
+ *            which a query can run;
+ *   - paged: restorePaged(), until it resolves (open), until a query from a
+ *            worker thread returns a random row (first response), and until
+ *            every page is local (filled).
+ *
+ * Tunables (env):
+ *   SIZES           database sizes, in MiB                  (default 16,64,256,1024)
+ *   LATENCY_MS      latency of each request, in ms          (default 20)
+ *   BANDWIDTH_MBPS  bandwidth of the link, in MB/s          (default 100)
+ *
+ * Run (the paged VFS must be built, as pnpm test:paged-vfs does):
+ *   pnpm bench:paged
+ *   SIZES=16,64 LATENCY_MS=5 pnpm bench:paged
+ */
+
+import { DatabaseSync } from 'node:sqlite';
+import { mkdtempSync } from 'node:fs';
+import { rm, stat } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { Readable } from 'node:stream';
+import { setTimeout as sleep } from 'node:timers/promises';
+import { Worker } from 'node:worker_threads';
+
+import { databaseUri } from '@platformatic/sqlite-js-stream-paged-vfs';
+
+import {
+  FileReplicaClient,
+  Replica,
+  type LTXFilesOptions,
+  type OpenLTXFileOptions,
+  type ReplicaClient,
+} from '../src/index.ts';
+
+const SIZES = (process.env.SIZES ?? '16,64,256,1024').split(',').map(Number);
+const LATENCY_MS = Number(process.env.LATENCY_MS ?? 20);
+const BANDWIDTH_MBPS = Number(process.env.BANDWIDTH_MBPS ?? 100);
+
+/**
+ * Reads `client` through a simulated network: every request (a read or a
+ * listing) waits `latencyMs`, and the bytes read share one link of
+ * `megabytesPerSecond`, whatever the number of concurrent reads.
+ */
+function network(client: ReplicaClient, latencyMs: number, megabytesPerSecond: number): ReplicaClient {
+  const bytesPerMs = (megabytesPerSecond * 1e6) / 1000;
+  let linkFree = 0;
+  const transfer = async (bytes: number): Promise<void> => {
+    const now = performance.now();
+    linkFree = Math.max(now, linkFree) + bytes / bytesPerMs;
+    if (linkFree - now >= 1) await sleep(linkFree - now);
+  };
+
+  return new Proxy(client, {
+    get(target, prop) {
+      if (prop === 'openLtxFile') {
+        return async (opts: OpenLTXFileOptions) => {
+          await sleep(latencyMs);
+          const stream = await target.openLtxFile(opts);
+          return Readable.from(
+            (async function* () {
+              for await (const chunk of stream) {
+                await transfer(chunk.length);
+                yield chunk;
+              }
+            })(),
+          );
+        };
+      }
+      if (prop === 'ltxFiles') {
+        return (opts: LTXFilesOptions) =>
+          (async function* () {
+            await sleep(latencyMs);
+            yield* target.ltxFiles(opts);
+          })();
+      }
+      const value = Reflect.get(target, prop, target);
+      return typeof value === 'function' ? value.bind(target) : value;
+    },
+  });
+}
+
+/** Replicates a database of about `mib` MiB, then changes a few rows twice. Returns its number of rows. */
+async function replicate(dir: string, client: ReplicaClient, mib: number): Promise<number> {
+  const path = join(dir, 'source.db');
+  const rows = mib * 256; // one row of 3,000 random bytes per 4 KiB page
+  const db = new DatabaseSync(path);
+  db.exec('PRAGMA journal_mode=WAL; PRAGMA wal_autocheckpoint=0; PRAGMA synchronous=OFF');
+  db.exec('CREATE TABLE t (id INTEGER PRIMARY KEY, v BLOB)');
+  db.exec(`WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${rows}) INSERT INTO t SELECT i, randomblob(3000) FROM n`);
+  // The first sync reads the database file rather than a WAL of the same size
+  db.exec('PRAGMA wal_checkpoint(TRUNCATE)');
+
+  const replica = new Replica(path, client);
+  await replica.sync();
+  for (let i = 0; i < 2; i++) {
+    db.exec(`UPDATE t SET v = randomblob(3000) WHERE id % 1000 = ${i}`);
+    await replica.sync();
+  }
+  replica.close();
+  db.close();
+  return rows;
+}
+
+/** A worker thread which opens the database and reads one row when told, as a Durable Object host would. */
+function queryWorker(): { ready: Promise<void>; query: (uri: string, id: number) => Promise<void>; close: () => Promise<number> } {
+  const worker = new Worker(
+    `
+      const { parentPort } = require('node:worker_threads')
+      const { DatabaseSync } = require('node:sqlite')
+      parentPort.on('message', ({ uri, id }) => {
+        const db = new DatabaseSync(uri)
+        const row = db.prepare('SELECT length(v) AS n FROM t WHERE id = ?').get(id)
+        db.close()
+        parentPort.postMessage(row)
+      })
+    `,
+    { eval: true },
+  );
+  return {
+    ready: new Promise((resolve) => worker.once('online', () => resolve())),
+    query: (uri, id) =>
+      new Promise((resolve, reject) => {
+        worker.once('message', (row: { n: number } | undefined) => (row?.n === 3000 ? resolve() : reject(new Error(`unexpected row ${JSON.stringify(row)}`))));
+        worker.postMessage({ uri, id });
+      }),
+    close: () => worker.terminate(),
+  };
+}
+
+function seconds(ms: number): string {
+  return ms < 1000 ? `${Math.round(ms)} ms` : `${(ms / 1000).toFixed(2)} s`;
+}
+
+console.log(`Paged restore benchmark: Node ${process.version}, ${LATENCY_MS} ms per request, ${BANDWIDTH_MBPS} MB/s`);
+console.log('');
+console.log('| Database | Full restore | Paged: open | Paged: first response | Paged: every page |');
+console.log('| --- | --- | --- | --- | --- |');
+
+for (const mib of SIZES) {
+  const dir = mkdtempSync(join(tmpdir(), 'bench-paged-'));
+  try {
+    const backing = new FileReplicaClient(join(dir, 'replica'));
+    const rows = await replicate(dir, backing, mib);
+    const client = network(backing, LATENCY_MS, BANDWIDTH_MBPS);
+
+    // Full restore
+    const fullPath = join(dir, 'full.db');
+    const full = new Replica(fullPath, client);
+    let start = performance.now();
+    await full.restore({ outputPath: fullPath });
+    const fullMs = performance.now() - start;
+    full.close();
+    const size = (await stat(fullPath)).size;
+    await rm(fullPath, { force: true });
+
+    // Paged restore, queried from another thread as soon as it resolves
+    const pagedPath = join(dir, 'paged.db');
+    const paged = new Replica(pagedPath, client);
+    const worker = queryWorker();
+    await worker.ready;
+    start = performance.now();
+    const result = await paged.restorePaged();
+    if (!result.paged) throw new Error('the paged VFS cannot be used: build it with pnpm test:paged-vfs');
+    const openMs = performance.now() - start;
+    await worker.query(databaseUri(pagedPath), 1 + Math.floor(Math.random() * rows));
+    const firstMs = performance.now() - start;
+    await result.filled;
+    const filledMs = performance.now() - start;
+    await worker.close();
+    paged.close();
+
+    console.log(
+      `| ${(size / 1024 / 1024).toFixed(0)} MiB | ${seconds(fullMs)} | ${seconds(openMs)} | ${seconds(firstMs)} | ${seconds(filledMs)} |`,
+    );
+  } finally {
+    await rm(dir, { recursive: true, force: true });
+  }
+}

package.json

@@ -52,7 +52,8 @@
     "test:all": "pnpm test && pnpm -r --filter '!@platformatic/sqlite-js-stream-paged-vfs' test",
     "test:paged-vfs": "pnpm --filter @platformatic/sqlite-js-stream-paged-vfs build:native && pnpm --filter @platformatic/sqlite-js-stream-paged-vfs build && pnpm --filter @platformatic/sqlite-js-stream-paged-vfs test && node --test test/integration/paged-restore.test.ts",
     "typecheck:all": "pnpm typecheck",
-    "bench": "node bench/bench-sync-restore.ts"
+    "bench": "node bench/bench-sync-restore.ts",
+    "bench:paged": "node bench/bench-paged-restore.ts"
   },
   "dependencies": {
     "lz4js": "^0.2.0"

test/integration/paged-restore.test.ts

@@ -7,6 +7,7 @@ import { once } from 'node:events';
 import { existsSync } from 'node:fs';
 import { readFile } from 'node:fs/promises';
 import { join } from 'node:path';
+import { Readable } from 'node:stream';
 
 import {
   Decoder,
@@ -41,11 +42,11 @@ const Rows = 2000;
 const Tables = 80;
 
 /**
- * A database of about 2,000 pages, whose schema spans several pages and has
+ * A database of about `rows` pages, whose schema spans several pages and has
  * records larger than a page (so overflow pages), replicated as a level-9
  * snapshot followed by level-0 files which change some of its pages.
  */
-async function replicate(t: TestContext, client: ReplicaClient): Promise<DatabaseSync> {
+async function replicate(t: TestContext, client: ReplicaClient, rows = Rows): Promise<DatabaseSync> {
   const path = join(makeTempDir('paged-src-'), 'src.db');
   const src = new DatabaseSync(path);
   t.after(() => src.close());
@@ -56,7 +57,7 @@ async function replicate(t: TestContext, client: ReplicaClient): Promise<Databas
   for (let i = 0; i < Tables; i++) src.exec(`CREATE TABLE extra_${i} (a INTEGER, b TEXT, c BLOB, d REAL)`);
   const columns = Array.from({ length: 400 }, (_, i) => `column_with_a_long_name_${i} TEXT`).join(', ');
   src.exec(`CREATE TABLE wide (${columns})`);
-  src.exec(`WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${Rows}) INSERT INTO t SELECT i, randomblob(3000) FROM n`);
+  src.exec(`WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${rows}) INSERT INTO t SELECT i, randomblob(3000) FROM n`);
   src.exec('COMMIT');
 
   const replica = new Replica(path, client);
@@ -78,28 +79,40 @@ async function replicate(t: TestContext, client: ReplicaClient): Promise<Databas
 /**
  * A client whose ranged reads of at least `minSize` bytes (those of the
  * background fill, not those for the pages queries wait for) wait for
- * `release()`.
+ * `release()`. It records the reads, and counts the bytes they return.
  */
-function gated(client: ReplicaClient, minSize = 256 * 1024): { client: ReplicaClient; release: () => void; reads: OpenLTXFileOptions[] } {
+function gated(
+  client: ReplicaClient,
+  minSize = 256 * 1024,
+): { client: ReplicaClient; release: () => void; reads: OpenLTXFileOptions[]; bytes: () => number } {
   let release!: () => void;
   const open = new Promise<void>((resolve) => {
     release = resolve;
   });
   const reads: OpenLTXFileOptions[] = [];
+  let bytes = 0;
   const proxy = new Proxy(client, {
     get(target, prop) {
       if (prop === 'openLtxFile') {
         return async (opts: OpenLTXFileOptions) => {
           reads.push(opts);
           if (opts.size >= minSize) await open;
-          return target.openLtxFile(opts);
+          const stream = await target.openLtxFile(opts);
+          return Readable.from(
+            (async function* () {
+              for await (const chunk of stream) {
+                bytes += chunk.length;
+                yield chunk;
+              }
+            })(),
+          );
         };
       }
       const value = Reflect.get(target, prop, target);
       return typeof value === 'function' ? value.bind(target) : value;
     },
   });
-  return { client: proxy, release, reads };
+  return { client: proxy, release, reads, bytes: () => bytes };
 }
 
 interface SqlWorker {
@@ -251,6 +264,62 @@ test('replication goes on while a paged restore fills the database', { skip }, a
   await db.close();
 });
 
+test('once every page is restored, nothing is read from the replica', { skip }, async (t) => {
+  const backing = new FileReplicaClient(makeTempDir('paged-repl-'));
+  const src = await replicate(t, backing);
+  const { client, release, reads } = gated(backing);
+  release();
+
+  const path = join(makeTempDir('paged-dst-'), 'db.db');
+  const replica = new Replica(path, client);
+  t.after(() => replica.close());
+  const result = await replica.restorePaged();
+  await result.filled;
+  reads.length = 0;
+
+  // Queries read every page, writes are replicated, and a snapshot reads the database again
+  const db = sqlWorker(t, path, true);
+  assert.deepEqual(await db.all('PRAGMA integrity_check'), [{ integrity_check: 'ok' }]);
+  assert.deepEqual(
+    await db.all('SELECT count(*) AS n, sum(length(v)) AS size FROM t'),
+    plain(src.prepare('SELECT count(*) AS n, sum(length(v)) AS size FROM t').all()),
+  );
+  await db.exec("INSERT INTO t (v) VALUES (x'02')");
+  assert.equal((await replica.sync())?.maxTXID, 7n);
+  assert.equal((await replica.snapshot())?.maxTXID, 7n);
+  await db.close();
+
+  assert.deepEqual(reads, []);
+});
+
+test('the reads before a paged restore resolves barely grow with the database (acceptance)', { skip }, async (t) => {
+  // What is read before the database can be opened: the headers of two files, the page indexes of the plan (a few
+  // bytes per page), and the schema with its read-ahead. Queries then read only the pages they need, so the time to
+  // the first response does not grow with the size of the database the way a full restore does.
+  const measure = async (rows: number): Promise<{ size: number; before: number; reads: number }> => {
+    const backing = new FileReplicaClient(makeTempDir('paged-repl-'));
+    await replicate(t, backing, rows);
+    const { client, release, bytes, reads } = gated(backing);
+    t.after(release);
+
+    const path = join(makeTempDir('paged-dst-'), 'db.db');
+    const replica = new Replica(path, client);
+    t.after(() => replica.close());
+    const result = await replica.restorePaged();
+    const before = { before: bytes(), reads: reads.length };
+    release();
+    await result.filled;
+    return { size: (await readFile(path)).length, ...before };
+  };
+
+  const small = await measure(2000);
+  const large = await measure(8000);
+  assert.ok(large.size > 3.5 * small.size, `the large database (${large.size} bytes) is 4 times the small one (${small.size})`);
+  assert.ok(large.before < 2 * small.before, `read before opening: ${large.before} bytes for ${large.size}, ${small.before} for ${small.size}`);
+  assert.ok(large.before < large.size / 50, `read before opening: ${large.before} bytes, under 2% of ${large.size}`);
+  t.diagnostic(`read before opening: ${small.before} of ${small.size} bytes, ${large.before} of ${large.size} bytes`);
+});
+
 test('closing the replica stops a paged restore, and the database is restored again', { skip }, async (t) => {
   const backing = new FileReplicaClient(makeTempDir('paged-repl-'));
   await replicate(t, backing);