perf(sync-client): prefetch pull metadata pages

apps/obsidian-plugin/release-notes/next.md

@@ -6,6 +6,8 @@
 
 ## Changed
 
+- Reduce waiting between batches when downloading large remote vaults.
+
 - Keep uploading queued files while slower attachments finish, and prioritize incoming remote changes after active uploads settle.
 
 - Open self-hosted device sign-in pages through an external browser when they use localhost.

packages/sync-client/benchmarks/README.md

@@ -124,3 +124,39 @@ They do not replace real-server measurements for protocol or server changes,
 nor do they model Obsidian's Dexie persistence or mobile runtime. Push service
 tests separately verify continuous replenishment, bounded outstanding work,
 pull priority, and queue preservation and retry after pipeline failures.
+
+## Pull pagination latency scenarios
+
+Run the lightweight pull scenarios without generating the 1 GiB fixture:
+
+```sh
+pnpm -C packages/sync-client bench --run -t pull-notes \
+  --outputJson /tmp/synch-pull-current.json
+```
+
+The public `SyncEngine.syncNow()` pulls 500 distinct 4 KiB Markdown notes into
+an empty filesystem vault, using the real metadata/blob encryption and client
+verification paths. Fixture generation, engine setup, final file-size/hash
+verification, cursor/queue checks, and cleanup are outside the timed operation.
+Each profile runs one warmup and five measured iterations. The production page
+size and download concurrency are used without overrides.
+
+| Scenario | Delay per metadata page | Delay per blob GET |
+| --- | ---: | ---: |
+| `pull-notes-no-delay` | 0 ms | 0 ms |
+| `pull-notes-page-40ms` | 40 ms | 5 ms |
+| `pull-notes-page-120ms` | 120 ms | 5 ms |
+
+Delays are independent asynchronous waits added to actual local filesystem and
+crypto work. They isolate pagination latency; they do not simulate shared
+bandwidth, real Cloudflare/R2 timings, Obsidian/Dexie, or mobile performance.
+The larger page delay models a metadata endpoint slower than blob downloads.
+No performance threshold is asserted.
+
+For before/after comparisons, run **the same benchmark files** in isolated
+checkouts of both implementations, save a baseline with `--outputJson`, and add
+`--compare /tmp/synch-pull-baseline.json` to the current run. Run sequentially to
+avoid competing for CPU and disk. Compare means and sample variation, and retain
+the raw JSON and base commit alongside the result. These scenarios must be
+copied into older checkouts that predate them; comparing a missing scenario
+against a new scenario is not a performance comparison.

packages/sync-client/benchmarks/pull-notes-fixture.ts

@@ -0,0 +1,49 @@
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { createSyncCryptoContext, hashBytes } from "../src/index";
+import type { BenchmarkFixture, BenchmarkFixtureEntry } from "./benchmark-fixture";
+
+export const PULL_NOTES_FIXTURE_SPEC = { notes: 500, noteBytes: 4 * 1024 } as const;
+
+/** Shared encrypted inputs; every timed run gets its own empty filesystem vault. */
+export async function createPullNotesFixture(key: Uint8Array) {
+  const directory = await mkdtemp(join(tmpdir(), "synch-pull-notes-"));
+  const crypto = createSyncCryptoContext(key);
+  const baseline: BenchmarkFixtureEntry[] = [];
+  try {
+    for (let index = 0; index < PULL_NOTES_FIXTURE_SPEC.notes; index++) {
+      const entryId = `entry-${String(index).padStart(4, "0")}`;
+      const blobId = `${entryId}-blob`;
+      const path = `notes/${entryId}.md`;
+      const bytes = new TextEncoder().encode(
+        `# Note ${index}\n`.padEnd(PULL_NOTES_FIXTURE_SPEC.noteBytes, "x"),
+      );
+      const hash = await hashBytes(bytes);
+      const bytesFile = `${entryId}.md`;
+      const encryptedBlobFile = `${blobId}.bin`;
+      const bytesPath = join(directory, bytesFile);
+      const encryptedBlobPath = join(directory, encryptedBlobFile);
+      await writeFile(bytesPath, bytes);
+      await writeFile(encryptedBlobPath, await crypto.encryptBlob(bytes, { blobId }));
+      baseline.push({
+        entryId, blobId, path, size: bytes.byteLength, hash,
+        revision: 1, updatedSeq: index + 1, updatedAt: index + 1,
+        encryptedMetadata: await crypto.encryptMetadata({ path, hash }, {
+          entryId, revision: 1, op: "upsert", blobId,
+        }),
+        bytesFile, encryptedBlobFile, bytesPath, encryptedBlobPath,
+      });
+    }
+    const fixture: BenchmarkFixture = {
+      directory, filesDirectory: directory, baseline, incremental: baseline,
+      incrementalCursor: baseline.length,
+    };
+    return { fixture, dispose: () => rm(directory, { recursive: true, force: true }) };
+  } catch (error) {
+    await rm(directory, { recursive: true, force: true });
+    throw error;
+  } finally {
+    crypto.dispose();
+  }
+}

packages/sync-client/benchmarks/sync-client.bench.ts

@@ -20,6 +20,7 @@ import {
   type BenchmarkFixture,
   type BenchmarkFixtureEntry,
 } from "./benchmark-fixture";
+import { createPullNotesFixture } from "./pull-notes-fixture";
 import { PushMetrics } from "./push-metrics";
 import { createMixedPushFixture, MIXED_PUSH_FIXTURE_SPEC } from "./mixed-push-fixture";
 import { createBenchmarkVault } from "./benchmark-vault";
@@ -70,6 +71,8 @@ type BenchmarkRun = {
 };
 
 type TransportProfile = {
+  pageDelayMs?: number;
+  downloadDelayMs?: number;
   uploadDelayMs: number;
   commitDelayMs: number;
   attachmentExtraDelayMs: number;
@@ -92,6 +95,12 @@ const MIXED_PROFILES: Record<string, TransportProfile> = {
     attachmentExtraDelayMs: 800,
   },
 };
+const PULL_PROFILES: Record<string, TransportProfile> = {
+  "pull-notes-no-delay": NO_DELAY,
+  "pull-notes-page-40ms": { ...NO_DELAY, pageDelayMs: 40, downloadDelayMs: 5 },
+  "pull-notes-page-120ms": { ...NO_DELAY, pageDelayMs: 120, downloadDelayMs: 5 },
+};
+let pullNotesFixture: ReturnType<typeof createPullNotesFixture> | undefined;
 const measurements: Record<string, ReturnType<PushMetrics["snapshot"]>[]> = {};
 
 let fixturePromise: Promise<BenchmarkFixture> | null = null;
@@ -158,6 +167,7 @@ class BenchmarkServer {
       return { status: 200 };
     }
 
+    await delay(this.profile.downloadDelayMs ?? 0);
     const blobPath = this.blobFiles.get(blobId);
     if (!blobPath) {
       return { status: 404 };
@@ -191,6 +201,7 @@ class BenchmarkServer {
         return;
 
       case "list_entry_states":
+        await delay(this.profile.pageDelayMs ?? 0);
         socket.receive({
           type: "entry_states_listed",
           requestId,
@@ -385,6 +396,12 @@ describe("sync-client black-box scenarios", () => {
   registerScenario("initial-pull-1GiB", async () => createInitialPullRun(await getFixture()));
   registerScenario("incremental-pull-64MiB", async () => createIncrementalPullRun(await getFixture()));
   registerScenario("push-1GiB", async () => createPushRun(await getFixture()));
+  for (const [name, profile] of Object.entries(PULL_PROFILES)) {
+    registerScenario(name, async () => {
+      pullNotesFixture ??= createPullNotesFixture(REMOTE_VAULT_KEY);
+      return createInitialPullRun((await pullNotesFixture).fixture, profile);
+    });
+  }
   for (const [name, profile] of Object.entries(MIXED_PROFILES)) {
     registerScenario(name, () => createMixedPushRun(profile));
   }
@@ -444,6 +461,7 @@ function registerScenario(
 
 afterAll(async () => {
   await benchmarkCleanupPromise;
+  if (pullNotesFixture) await (await pullNotesFixture).dispose();
   const samples = Object.fromEntries(Object.entries(measurements).filter(([, runs]) => runs.length));
   if (Object.keys(samples).length) {
     console.log("Push measurements (means of measured iterations; milliseconds):");
@@ -473,8 +491,9 @@ async function getFixture(): Promise<BenchmarkFixture> {
 
 async function createInitialPullRun(
   fixture: BenchmarkFixture,
+  profile: TransportProfile = NO_DELAY,
 ): Promise<BenchmarkRun> {
-  const server = new BenchmarkServer(fixture.baseline, fixture.baseline.length);
+  const server = new BenchmarkServer(fixture.baseline, fixture.baseline.length, profile);
   const vaultHandle = await createBenchmarkVault(null);
   const store = createTestSyncStore(LOCAL_VAULT_ID);
   const harness = createEngineHarness(server, vaultHandle.adapter, store);

packages/sync-client/src/sync/engine/__tests__/pull-service/prefetch.test.ts

@@ -0,0 +1,152 @@
+import { describe, expect, it, vi } from "vitest";
+import { createTestSyncStore } from "../../../../test-support/in-memory-sync-store";
+import { SyncPullService } from "../../pull-service";
+import {
+  createCommit, createPullClient, createRealtimeSession, createToken,
+  createVaultAdapter, encryptRemoteMetadata, encryptTestBlob, hashText,
+  TEST_VAULT_KEY,
+} from "./helpers";
+
+function gate() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((done) => { resolve = done; });
+  return { promise, resolve };
+}
+
+async function setup() {
+  const store = createTestSyncStore();
+  const adapter = createVaultAdapter();
+  const commits = await Promise.all([1, 2, 3].map(async (id) => createCommit({
+    cursor: id,
+    entryId: `entry-${id}`,
+    blobId: `blob-${id}`,
+    encryptedMetadata: await encryptRemoteMetadata({
+      entryId: `entry-${id}`, revision: 1, blobId: `blob-${id}`,
+      path: `note-${id}.md`, hash: await hashText(`body-${id}`),
+    }),
+  })));
+  const session = createRealtimeSession({
+    pages: commits.map((commit, index) => ({
+      cursor: 3, hasMore: index < 2, commits: [commit],
+    })),
+  });
+  const client = createPullClient({ blobs: Object.fromEntries(await Promise.all(
+    [1, 2, 3].map(async (id) => [
+      `blob-${id}`, await encryptTestBlob(`blob-${id}`, new TextEncoder().encode(`body-${id}`)),
+    ]),
+  )) });
+  const service = new SyncPullService({
+    getApiBaseUrl: () => "http://127.0.0.1:8787",
+    getSyncToken: async () => createToken(),
+    getSyncStore: () => store,
+    getRemoteVaultKey: () => TEST_VAULT_KEY,
+    vaultAdapter: adapter, pullClient: client, applyWindowSize: 1,
+  });
+  return { store, adapter, session, client, service };
+}
+
+describe("SyncPullService page prefetch", () => {
+  it("fetches one page ahead during downloads and preserves pagination and checkpoint order", async () => {
+    const { store, adapter, session, client, service } = await setup();
+    const release = gate();
+    const list = vi.spyOn(session, "listEntryStates");
+    const download = client.downloadBlob.bind(client);
+    const downloads = vi.spyOn(client, "downloadBlob").mockImplementation(async (...args) => {
+      if (args[3] === "blob-1") await release.promise;
+      return await download(...args);
+    });
+    const pulling = service.pullOnce(session);
+    try {
+      await vi.waitFor(() => expect(downloads).toHaveBeenCalledTimes(1));
+      expect(list).toHaveBeenCalledTimes(2);
+      expect(list.mock.calls[1][0]).toMatchObject({
+        sinceCursor: 0, targetCursor: 3,
+        after: { updatedSeq: 1, entryId: "entry-1" },
+      });
+      expect(await store.getCursor()).toBe(0);
+      expect(await adapter.exists("note-2.md")).toBe(false);
+    } finally {
+      release.resolve();
+    }
+    await expect(pulling).resolves.toMatchObject({ cursor: 3, filesWritten: 3 });
+    expect(list).toHaveBeenCalledTimes(3);
+    expect(list.mock.calls[2][0]).toMatchObject({
+      sinceCursor: 0, targetCursor: 3,
+      after: { updatedSeq: 2, entryId: "entry-2" },
+    });
+    expect(adapter.text("note-3.md")).toBe("body-3");
+    await store.close();
+  });
+
+  it.each(["request", "metadata"])("keeps the applied checkpoint when prefetched %s fails", async (failure) => {
+    const { store, adapter, session, client, service } = await setup();
+    const release = gate();
+    const list = session.listEntryStates.bind(session);
+    const failureObserved = gate();
+    const error = new Error("injected page failure");
+    vi.spyOn(session, "listEntryStates").mockImplementation(async (request) => {
+      const page = await list(request);
+      if (request.after) {
+        failureObserved.resolve();
+        if (failure === "request") throw error;
+        page.entries[0].encryptedMetadata = "invalid";
+      }
+      return page;
+    });
+    const download = client.downloadBlob.bind(client);
+    vi.spyOn(client, "downloadBlob").mockImplementation(async (...args) => {
+      await release.promise;
+      return await download(...args);
+    });
+    const outcome = service.pullOnce(session).then(
+      () => ({ error: null }), (error: unknown) => ({ error }),
+    );
+    await failureObserved.promise;
+    // Give the prefetch time to reject while the current download is blocked.
+    await new Promise<void>((resolve) => setTimeout(resolve, 10));
+    expect(await store.getCursor()).toBe(0);
+    release.resolve();
+    const result = await outcome;
+    if (failure === "request") expect(result.error).toBe(error);
+    else expect(result.error).toBeInstanceOf(Error);
+    expect(await store.getCursor()).toBe(1);
+    expect(adapter.text("note-1.md")).toBe("body-1");
+    expect(await adapter.exists("note-2.md")).toBe(false);
+    await store.close();
+  });
+
+  it("drains an outstanding prefetch after apply fails and preserves the apply error", async () => {
+    const { store, session, client, service } = await setup();
+    const release = gate();
+    const list = session.listEntryStates.bind(session);
+    const calls = vi.spyOn(session, "listEntryStates").mockImplementation(async (request) => {
+      if (request.after) {
+        await release.promise;
+        throw new Error("injected prefetch failure");
+      }
+      return await list(request);
+    });
+    const error = new Error("injected download failure");
+    const failed = gate();
+    vi.spyOn(client, "downloadBlob").mockImplementation(async () => {
+      failed.resolve();
+      throw error;
+    });
+    let settled = false;
+    const outcome = service.pullOnce(session).then(
+      () => ({ error: null }), (error: unknown) => ({ error }),
+    ).finally(() => { settled = true; });
+    try {
+      await failed.promise;
+      await new Promise<void>((resolve) => setTimeout(resolve, 10));
+      expect(calls).toHaveBeenCalledTimes(2);
+      expect(settled).toBe(false);
+    } finally {
+      release.resolve();
+    }
+    expect((await outcome).error).toBe(error);
+    expect(await store.getCursor()).toBe(0);
+    expect(calls).toHaveBeenCalledTimes(2);
+    await store.close();
+  });
+});

packages/sync-client/src/sync/engine/pull-entry-state-applier.ts

@@ -134,18 +134,32 @@ export class PullEntryStateApplier {
   ): Promise<PullEntryStateManifestItem[]> {
     const remoteVaultKey = this.deps.getRemoteVaultKey();
 
-    return await mapWithConcurrency(
+    // Settle every metadata worker before returning a failed prefetched page.
+    const results = await mapWithConcurrency(
       states,
       this.deps.prepareConcurrency ?? DEFAULT_PREPARE_CONCURRENCY,
-      async (state) => ({
-        state,
-        metadata: await decryptSyncMetadata(
-          remoteVaultKey,
-          state.encryptedMetadata,
-          metadataContextFromRemoteState(state),
-        ),
-      }),
+      async (state) => {
+        try {
+          return {
+            ok: true as const,
+            item: {
+              state,
+              metadata: await decryptSyncMetadata(
+                remoteVaultKey,
+                state.encryptedMetadata,
+                metadataContextFromRemoteState(state),
+              ),
+            },
+          };
+        } catch (error) {
+          return { ok: false as const, error };
+        }
+      },
     );
+    return results.map((result) => {
+      if (!result.ok) throw result.error;
+      return result.item;
+    });
   }
 
   async applyEntryStates(