From bae10cde649559997be8c93c4d0c5b29c6a6f04d Mon Sep 17 00:00:00 2001 From: hhhjin Date: Sat, 5 Sep 2026 19:05:08 +0900 Subject: [PATCH] perf(sync-client): prefetch pull metadata pages --- apps/obsidian-plugin/release-notes/next.md | 2 + packages/sync-client/benchmarks/README.md | 36 +++++ .../benchmarks/pull-notes-fixture.ts | 49 ++++++ .../benchmarks/sync-client.bench.ts | 21 ++- .../__tests__/pull-service/prefetch.test.ts | 152 ++++++++++++++++++ .../sync/engine/pull-entry-state-applier.ts | 32 ++-- .../src/sync/engine/pull-service.ts | 93 +++++++---- packages/sync-client/vitest.bench.config.mts | 8 +- 8 files changed, 345 insertions(+), 48 deletions(-) create mode 100644 packages/sync-client/benchmarks/pull-notes-fixture.ts create mode 100644 packages/sync-client/src/sync/engine/__tests__/pull-service/prefetch.test.ts diff --git a/apps/obsidian-plugin/release-notes/next.md b/apps/obsidian-plugin/release-notes/next.md index e3b78f8c..9350acf2 100644 --- a/apps/obsidian-plugin/release-notes/next.md +++ b/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. diff --git a/packages/sync-client/benchmarks/README.md b/packages/sync-client/benchmarks/README.md index 21e691da..ec0e0970 100644 --- a/packages/sync-client/benchmarks/README.md +++ b/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. diff --git a/packages/sync-client/benchmarks/pull-notes-fixture.ts b/packages/sync-client/benchmarks/pull-notes-fixture.ts new file mode 100644 index 00000000..62fb84f2 --- /dev/null +++ b/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(); + } +} diff --git a/packages/sync-client/benchmarks/sync-client.bench.ts b/packages/sync-client/benchmarks/sync-client.bench.ts index c3d94288..231eef56 100644 --- a/packages/sync-client/benchmarks/sync-client.bench.ts +++ b/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 = { attachmentExtraDelayMs: 800, }, }; +const PULL_PROFILES: Record = { + "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 | undefined; const measurements: Record[]> = {}; let fixturePromise: Promise | 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 { async function createInitialPullRun( fixture: BenchmarkFixture, + profile: TransportProfile = NO_DELAY, ): Promise { - 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); diff --git a/packages/sync-client/src/sync/engine/__tests__/pull-service/prefetch.test.ts b/packages/sync-client/src/sync/engine/__tests__/pull-service/prefetch.test.ts new file mode 100644 index 00000000..ed176934 --- /dev/null +++ b/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((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((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((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(); + }); +}); diff --git a/packages/sync-client/src/sync/engine/pull-entry-state-applier.ts b/packages/sync-client/src/sync/engine/pull-entry-state-applier.ts index d767f65f..ad39376f 100644 --- a/packages/sync-client/src/sync/engine/pull-entry-state-applier.ts +++ b/packages/sync-client/src/sync/engine/pull-entry-state-applier.ts @@ -134,18 +134,32 @@ export class PullEntryStateApplier { ): Promise { 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( diff --git a/packages/sync-client/src/sync/engine/pull-service.ts b/packages/sync-client/src/sync/engine/pull-service.ts index d39f427b..03f69e2f 100644 --- a/packages/sync-client/src/sync/engine/pull-service.ts +++ b/packages/sync-client/src/sync/engine/pull-service.ts @@ -110,7 +110,6 @@ export class SyncPullService { const progress = new SyncWorkProgress("pull"); const received = new Set(); await onProgress(progress.snapshot()); - let after: { updatedSeq: number; entryId: string } | null = null; let window: PullEntryStateManifestItem[] = []; const applyWindowSize = normalizePositiveInteger( this.deps.applyWindowSize, @@ -123,14 +122,16 @@ export class SyncPullService { conflictsCreated: 0, }; - while (hasMore) { + const preparePage = async ( + target: number | null, + after: { updatedSeq: number; entryId: string } | null, + ) => { const page = await session.listEntryStates({ sinceCursor: requestCursor, - targetCursor, + targetCursor: target, after, limit: DEFAULT_PULL_BATCH, }); - targetCursor = page.targetCursor; const entries = page.entries.filter((entry) => { const key = stateKey(entry); if (received.has(key)) return false; @@ -138,39 +139,63 @@ export class SyncPullService { return true; }); const items = await this.entryStateApplier.createManifestItems(entries); - progress.register(items.map(manifestKey)); - window.push(...items); - after = page.nextAfter; - hasMore = page.hasMore; - if (!hasMore) progress.seal(); - await onProgress(progress.snapshot()); + return { page, items }; + }; + // Observe errors immediately, but propagate them only when consuming the page. + const startPage = ( + target: number | null, + after: { updatedSeq: number; entryId: string } | null, + ) => preparePage(target, after).then( + (value) => ({ ok: true as const, value }), + (error: unknown) => ({ ok: false as const, error }), + ); + let pendingPage = startPage(null, null); - if (window.length >= applyWindowSize || !hasMore) { - const appliedWindow = window; - const applied = await this.entryStateApplier.applyManifestWindow( - store, - token, - window, - { - finalWindow: !hasMore, - }, - ); - totals.entriesApplied += applied.entriesApplied; - totals.filesWritten += applied.filesWritten; - totals.filesDeleted += applied.filesDeleted; - totals.conflictsCreated += applied.conflictsCreated; - window = applied.deferred; - cursor = await this.checkpointAppliedWindow( - store, - session, - cursor, - appliedWindow, - applied.deferred, - hasMore ? null : targetCursor, - ); - progress.complete(applied.completedStates.map(stateKey)); + try { + while (hasMore) { + const result = await pendingPage; + if (!result.ok) throw result.error; + const { page, items } = result.value; + targetCursor = page.targetCursor; + hasMore = page.hasMore; + // Only metadata is prefetched. Planning, vault writes and checkpoints stay + // ordered, and at most one page is prepared ahead of the current window. + if (hasMore) pendingPage = startPage(targetCursor, page.nextAfter); + progress.register(items.map(manifestKey)); + window.push(...items); + if (!hasMore) progress.seal(); await onProgress(progress.snapshot()); + + if (window.length >= applyWindowSize || !hasMore) { + const appliedWindow = window; + const applied = await this.entryStateApplier.applyManifestWindow( + store, + token, + window, + { + finalWindow: !hasMore, + }, + ); + totals.entriesApplied += applied.entriesApplied; + totals.filesWritten += applied.filesWritten; + totals.filesDeleted += applied.filesDeleted; + totals.conflictsCreated += applied.conflictsCreated; + window = applied.deferred; + cursor = await this.checkpointAppliedWindow( + store, + session, + cursor, + appliedWindow, + applied.deferred, + hasMore ? null : targetCursor, + ); + progress.complete(applied.completedStates.map(stateKey)); + await onProgress(progress.snapshot()); + } } + } finally { + // Do not let background work outlive pullOnce (or its crypto/session). + await pendingPage; } if (window.length > 0) { diff --git a/packages/sync-client/vitest.bench.config.mts b/packages/sync-client/vitest.bench.config.mts index 1cc53bb6..46b7e91f 100644 --- a/packages/sync-client/vitest.bench.config.mts +++ b/packages/sync-client/vitest.bench.config.mts @@ -3,9 +3,9 @@ import { defineConfig } from "vitest/config"; export default defineConfig({ test: { environment: "node", - }, - benchmark: { - include: ["benchmarks/**/*.bench.ts"], - includeSamples: true, + benchmark: { + include: ["benchmarks/**/*.bench.ts"], + includeSamples: true, + }, }, });