feat: restore a database page by page, on demand

CHANGELOG.md

@@ -8,6 +8,8 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Changed
 
+- `S3ReplicaClient.openLtxFile()` reads a range with a single `GetObject`, without
+  the `HeadObject` it sent first, which halves the requests of a paged restore.
 - The first `sync()` after a restore into the replica's own path (as `ensureExists()` does) uploads
   only the changes made since the restore, instead of a snapshot of the whole database. The restore
   writes the first WAL of the restored database: one commit frame holding page 1 unchanged, under
@@ -34,6 +36,18 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Added
 
+- `Replica.restorePaged()` restores the replica's own database page by page, with
+  `@platformatic/sqlite-js-stream-paged-vfs` (a new optional peer dependency).
+  It resolves once the database can be opened through the VFS; queries from
+  other threads read the pages they need on demand, with read-ahead, while the
+  others are filled in the background (`result.filled`). Replication goes on
+  meanwhile: `sync()` and `snapshot()` read missing pages first, and retention
+  keeps the LTX files the restore reads. A restore stopped by `close()` or a
+  crash leaves the database incomplete: `localPos()` is 0, `sync()` and
+  `snapshot()` throw `ErrIncompleteDatabase`, and `ensureExists()` restores it
+  again. It restores fully below `minSize` bytes, or when the VFS cannot be used.
+  New exports: `PagedRestoreOptions`, `PagedRestoreResult`,
+  `ErrIncompleteDatabase`, `ErrPagedRestoreClosed`, `ErrPagedVfsUnavailable`.
 - `@platformatic/sqlite-js-stream-paged-vfs`, a new package: a SQLite VFS for
   `node:sqlite` that restores a database page by page on demand. Queries run
   while the database downloads; a read of a missing page waits until the

README.md

@@ -47,7 +47,8 @@ Importing the addon registers the `s3` URL scheme with
 **`@platformatic/sqlite-js-stream-paged-vfs`** is a SQLite VFS for `node:sqlite`
 which reads the pages of a database on demand, so that an application can query
 a large database while it is still being restored. It is a native library, with
-prebuilt binaries for macOS and Linux. See its
+prebuilt binaries for macOS and Linux. `replica.restorePaged()` uses it when it is
+installed (see [Restore page by page](#restore-page-by-page)). See its
 [README](packages/sqlite-js-stream-paged-vfs/README.md).
 
 ```bash
@@ -148,6 +149,47 @@ closing (which deletes it), makes the next `sync()` fall back to a snapshot.
 With `wal_autocheckpoint` disabled and checkpoints run through
 `replica.checkpoint()` (see below), syncs stay incremental from the restore on.
 
+### Restore page by page
+
+A full restore downloads the whole database before it can be opened. With
+`@platformatic/sqlite-js-stream-paged-vfs` installed, `restorePaged()` restores
+the replica's own database page by page instead: it resolves once the database
+can be opened, and queries read the pages they need on demand while the others
+are downloaded in the background.
+
+```ts
+import { databaseUri } from '@platformatic/sqlite-js-stream-paged-vfs';
+
+const result = await replica.restorePaged({ minSize: 64 * 1024 * 1024 });
+// In a worker thread, not this one:
+const db = new DatabaseSync(databaseUri('/data/app.db'));
+
+await result.filled; // every page is local
+```
+
+- The database is created with every page missing. A query that reads a missing
+  page waits while it is read from the replica, with the next pages of the same
+  LTX file (`readAhead`, 32 by default). Page 1 and the schema are read before
+  `restorePaged()` resolves.
+- Open the database with the URI from `databaseUri()`, from **another thread**
+  than the one that called `restorePaged()`: that thread serves the pages, so a
+  query from it would wait for itself (such reads fail instead). Every
+  connection that writes the database must use the VFS.
+- Replication goes on meanwhile. `sync()` and `snapshot()` read the pages they
+  need first, and retention keeps the LTX files the restore still reads.
+- `result.filled` resolves once every page is local. `replica.close()`, or
+  restoring the database again, stops the restore: the database stays
+  incomplete. So does a crash. An incomplete database has a `localPos()` of 0,
+  and `sync()`/`snapshot()` throw `ErrIncompleteDatabase` until it is restored
+  again (`ensureExists()` does).
+- It restores fully, like `restore()`, when the LTX files to download add up
+  to less than `minSize` bytes, or when the paged VFS cannot be used
+  (`result.unavailable` holds the reason: the package is not installed, has no
+  binary for the platform, or Node.js runs with `--permission`).
+  `result.paged` tells which.
+- `onMiss(pgno, pages, durationMs)` is called for each page a query waited for,
+  for metrics.
+
 ### Manage many databases
 
 ```ts

docs/ARCHITECTURE.md

@@ -159,6 +159,49 @@ by an earlier restore is ignored.
 Because restore never materializes the full merged LTX (or the whole DB) in memory, peak memory is
 bounded by one page buffer plus the (small) page index, regardless of the database size.
 
+### Replica.restorePaged()
+
+`restorePaged()` (`src/paged-restore.ts`) makes the replica's own database usable before it is
+downloaded, with the SQLite VFS of `@platformatic/sqlite-js-stream-paged-vfs`, an optional peer
+dependency imported only here. The VFS keeps, for each registered database, a bitmap of the pages
+present in a sparse file; a read of a missing page asks the thread that registered the database for
+it and waits, and it never fills a page SQLite wrote meanwhile (see that package's README).
+
+1. `CalcRestorePlan` as for a full restore. When its files add up to less than `minSize`, or the
+   VFS cannot be loaded, it restores fully instead.
+2. The headers of the first and last files give the page size and the final page count. The page
+   index of every file (read concurrently) gives, for each page, the file holding its newest
+   version and the offset and size of its frame there, kept in typed arrays.
+3. The marker `<db>-litestream/paged` is written, the database file and its `-wal`/`-shm` are
+   removed, and the database is registered with the VFS, which creates it sparse at its final size.
+4. Page 1 and the schema table (a b-tree rooted at page 1, with the overflow pages of its records)
+   are filled, walking the b-tree from page 1. The replica's read connection parses the schema and
+   runs on the thread that serves the pages, where a read of a missing page fails rather than
+   waits: these are the only pages of the database file it reads.
+5. The restore writes the first WAL and the restore record, as a full restore does, and resolves.
+
+From then on:
+
+- **Requests.** The VFS asks for a page; it is read with the next frames of the same file, up to
+  `readAhead`, in one ranged read, decoded and filled. A page the background fill is reading is
+  read again in this smaller read, so that a query never waits for a large one. A failed read is
+  retried twice, then the VFS fails the waiting reads with `SQLITE_IOERR_READ`.
+- **Background fill.** Passes over the missing pages in page order, grouping them by file into
+  ranged reads of at most 4 MiB (spanning frames of superseded pages up to 256 KiB), four at a
+  time. Failed reads are retried in the next pass, with a backoff. Pages no file has are filled
+  with zeros, as a full restore leaves them. Once the VFS reports every page present, the marker is
+  removed and `filled` resolves.
+- **Replication.** `LocalDatabase` reads database pages from the file only after the page loader
+  made them present: a page being read waits for that read, a missing one is read with read-ahead.
+  The replica's read connection opens the database through the VFS. `deleteFiles()` keeps the
+  files of the plan until the restore is done, whatever the retention policy.
+- **Stopping.** `close()`, or another restore, closes the VFS registration: reads of missing pages
+  fail, and `filled` rejects with `ErrPagedRestoreClosed`. The marker stays, as after a crash: a
+  replica without a paged restore in progress sees an incomplete database, reports a local
+  position of 0, and refuses to `sync()` or `snapshot()` it (`ErrIncompleteDatabase`), so that its
+  holes are never uploaded as zeros. A full restore removes the marker once the database is
+  complete.
+
 ### Replica.follow()
 
 `follow()` is litestream's `restore -f` continuous mode. It first restores the latest backup to

package.json

@@ -50,14 +50,23 @@
     "golden:go": "cd tools/golden && go run .",
     "pretest:all": "pnpm build",
     "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 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"
   },
   "dependencies": {
     "lz4js": "^0.2.0"
   },
+  "peerDependencies": {
+    "@platformatic/sqlite-js-stream-paged-vfs": "workspace:^"
+  },
+  "peerDependenciesMeta": {
+    "@platformatic/sqlite-js-stream-paged-vfs": {
+      "optional": true
+    }
+  },
   "devDependencies": {
+    "@platformatic/sqlite-js-stream-paged-vfs": "workspace:*",
     "@types/node": "^26.0.0",
     "eslint": "^9.39.5",
     "neostandard": "^0.14.0",

packages/backend-s3/src/replica-client.ts

@@ -557,19 +557,10 @@ export class S3ReplicaClient implements ReplicaClient {
       }
     }
 
-    // Need the object size to build a range and to handle offset >= size.
-    let head;
-    try {
-      head = await this.client.send(new HeadObjectCommand({ Bucket: this.bucket, Key: key }));
-    } catch (err) {
-      throw new LTXError('open', key, opts.level, opts.minTXID, opts.maxTXID, err);
-    }
-    const len = Number(head.ContentLength ?? 0);
-    if (opts.offset >= len) return Readable.from([]);
-
-    const end = opts.size > 0 ? Math.min(opts.offset + opts.size - 1, len - 1) : len - 1;
-    if (end < opts.offset) return Readable.from([]);
-
+    // A range may end past the end of the object, and an open range reads to
+    // its end, so no HeadObject round-trip is needed: a range starting past the
+    // end fails with 416, and reads nothing.
+    const end = opts.size > 0 ? String(opts.offset + opts.size - 1) : '';
     try {
       const res = await this.client.send(
         new GetObjectCommand({
@@ -580,6 +571,7 @@ export class S3ReplicaClient implements ReplicaClient {
       );
       return res.Body as Readable;
     } catch (err) {
+      if (isInvalidRange(err)) return Readable.from([]);
       throw new LTXError('open', key, opts.level, opts.minTXID, opts.maxTXID, err);
     }
   }
@@ -997,3 +989,9 @@ export function newReplicaClientFromURL(
   }
   return new S3ReplicaClient(opts);
 }
+
+/** Whether a GetObject failed because its range starts past the end of the object. */
+function isInvalidRange(err: unknown): boolean {
+  const e = err as { name?: string; Code?: string; $metadata?: { httpStatusCode?: number } };
+  return e?.name === 'InvalidRange' || e?.Code === 'InvalidRange' || e?.$metadata?.httpStatusCode === 416;
+}

packages/backend-s3/test/helpers/memory-s3.ts

@@ -198,7 +198,9 @@ export class MemoryS3 {
       const m = /^bytes=(\d+)-(\d*)$/.exec(cmd.input.Range);
       if (m) {
         const start = Number(m[1]);
-        const end = m[2] === '' ? obj.body.length - 1 : Number(m[2]);
+        // Like S3: a range starting past the end fails, one ending past it is cut
+        if (start >= obj.body.length) throw invalidRange(cmd.input.Range);
+        const end = m[2] === '' ? obj.body.length - 1 : Math.min(Number(m[2]), obj.body.length - 1);
         body = obj.body.subarray(start, end + 1);
       }
     }
@@ -273,6 +275,13 @@ function notFound(key: string): Error {
   return err;
 }
 
+function invalidRange(range: string): Error {
+  const err = new Error(`The requested range is not satisfiable: ${range}`);
+  (err as any).name = 'InvalidRange';
+  (err as any).$metadata = { httpStatusCode: 416 };
+  return err;
+}
+
 function noSuchUpload(uploadId: string): Error {
   const err = new Error(`No Such Upload: ${uploadId}`);
   (err as any).name = 'NoSuchUpload';

packages/backend-s3/test/s3.test.ts

@@ -16,6 +16,7 @@ import {
   S3ReplicaClient,
   type S3ReplicaClientOptions,
 } from '../src/index.ts';
+import { GetObjectCommand } from '@aws-sdk/client-s3';
 import { MemoryS3 } from './helpers/memory-s3.ts';
 import { createWALDB } from './helpers/db.ts';
 
@@ -205,6 +206,29 @@ test('openLtxFile returns empty stream when offset >= size', async () => {
   assert.equal(buf.length, 0);
 });
 
+test('openLtxFile reads a range with one request, without HeadObject', async () => {
+  const { client, s3 } = makeS3({ bucket: 'b' });
+  const { body } = makeLtxBody(5);
+  s3.objects.set('b/ltx/0/0000000000000001-0000000000000001.ltx', {
+    body: Buffer.from(body),
+    metadata: {},
+    lastModified: new Date(),
+  });
+  const sent: unknown[] = [];
+  s3.onSend = (command) => {
+    sent.push(command);
+  };
+  const read = async (offset: number, size: number): Promise<number[]> =>
+    [...new Uint8Array(await collect(await client.openLtxFile({ level: 0, minTXID: 1n, maxTXID: 1n, offset, size })))];
+
+  // Open ranges read to the end, and ranges ending past it are cut
+  assert.deepEqual(await read(102, 0), [2, 3, 4]);
+  assert.deepEqual(await read(103, 100), [3, 4]);
+  assert.deepEqual(await read(105, 0), []);
+  assert.equal(sent.length, 3);
+  assert.ok(sent.every((command) => command instanceof GetObjectCommand));
+});
+
 test('openLtxFile throws for a missing object', async () => {
   const { client } = makeS3({ bucket: 'b' });
   await assert.rejects(

packages/sqlite-js-stream-paged-vfs/README.md

@@ -9,7 +9,10 @@ the other pages in the background. Once every page is present, the database is
 a regular SQLite database and reads are local only.
 
 It lets an application open a large database restored from a replica right away,
-instead of after downloading all of it.
+instead of after downloading all of it. To restore a database replicated with
+[`@platformatic/sqlite-js-stream`](https://github.com/platformatic/sqlite-js-stream#restore-page-by-page),
+call `replica.restorePaged()`: it uses this package to fetch each page from the
+replica when it is read, and fills the others in the background.
 
 > Status: experimental.
 

src/db.ts

@@ -43,7 +43,7 @@ import {
 } from './wal/index.ts';
 import { SnapshotLevel } from './replica-client.ts';
 import { mkdirAll } from './util/fs.ts';
-import { DatabaseError } from './errors.ts';
+import { DatabaseError, ErrIncompleteDatabase } from './errors.ts';
 
 /**
  * ReaderConnection is the minimal SQLite connection used to pin a WAL read
@@ -63,6 +63,12 @@ export interface ReaderConnection {
  */
 export type OpenReader = (path: string) => ReaderConnection;
 
+/**
+ * PageLoader makes a page present in the database file, while a paged restore
+ * fills it. It returns undefined when the page is already present.
+ */
+export type PageLoader = (pgno: number) => Promise<void> | undefined;
+
 /**
  * CheckpointConnection is the minimal read-write SQLite connection used by
  * {@link runCheckpoint} to run `PRAGMA wal_checkpoint`. A
@@ -283,6 +289,7 @@ export class LocalDatabase {
   readonly pageSize: number;
   private conn: ReaderConnection | null = null;
   private openReader: OpenReader;
+  private pageLoader: PageLoader | null = null;
 
   private constructor(path: string, metaPath: string, pageSize: number, openReader: OpenReader) {
     this.path = path;
@@ -389,6 +396,47 @@ export class LocalDatabase {
     }
   }
 
+  /**
+   * PagedPath returns the path of the file which exists while a paged restore
+   * of the database has not filled every page.
+   */
+  pagedPath(): string {
+    return nodePath.join(this.metaPath, 'paged');
+  }
+
+  /** Reports whether a paged restore of the database did not fill every page. */
+  async incomplete(): Promise<boolean> {
+    try {
+      await stat(this.pagedPath());
+      return true;
+    } catch {
+      return false;
+    }
+  }
+
+  /**
+   * SetPageLoader sets the function which makes a page present in the
+   * database file before it is read from it, while a paged restore fills the
+   * database, or removes it. Without it, an incomplete database is never read.
+   */
+  setPageLoader(loader: PageLoader | null): void {
+    this.pageLoader = loader;
+  }
+
+  /** Throws if the database is incomplete and no page loader can fill it. */
+  private async assertComplete(): Promise<void> {
+    if (this.pageLoader === null && (await this.incomplete())) {
+      throw new ErrIncompleteDatabase(this.path);
+    }
+  }
+
+  /** Reads page `pgno` from the database file into `page`. */
+  private async readDbPage(handle: FileHandle, page: Uint8Array, pgno: number): Promise<void> {
+    const load = this.pageLoader?.(pgno);
+    if (load !== undefined) await load;
+    await readExact(handle, page, (pgno - 1) * page.length);
+  }
+
   /** LTXDir returns the local LTX root directory. */
   ltxDir(): string {
     return nodePath.join(this.metaPath, 'ltx');
@@ -500,6 +548,7 @@ export class LocalDatabase {
 
   /** Sync builds a level-0 LTX file and writes it to the local cache. */
   async sync(): Promise<FileInfo | null> {
+    await this.assertComplete();
     const pos = await this.pos();
     const txID = pos + 1n;
 
@@ -676,7 +725,7 @@ export class LocalDatabase {
       // Growth pages (not in pageMap) come directly from the database file.
       if (!pageMap.has(pgno)) {
         const page = new Uint8Array(pageSize);
-        await readExact(dbHandle, page, (pgno - 1) * pageSize);
+        await this.readDbPage(dbHandle, page, pgno);
         enc.encodePage({ pgno, flags: 0 }, page);
         continue;
       }
@@ -723,6 +772,7 @@ export class LocalDatabase {
    * a snapshot would cover no data (empty database).
    */
   async snapshot(): Promise<FileInfo | null> {
+    await this.assertComplete();
     const pos = await this.pos();
     const maxTXID = pos === 0n ? 1n : pos;
 
@@ -819,7 +869,7 @@ export class LocalDatabase {
       return new Uint8Array(walBuf.subarray(start, start + pageSize));
     }
     const page = new Uint8Array(pageSize);
-    await readExact(dbHandle, page, (pgno - 1) * pageSize);
+    await this.readDbPage(dbHandle, page, pgno);
     return page;
   }