diff --git a/apps/cli/src/host/sqlite-store/store.test.ts b/apps/cli/src/host/sqlite-store/store.test.ts index 8e980b05..9ca7eb50 100644 --- a/apps/cli/src/host/sqlite-store/store.test.ts +++ b/apps/cli/src/host/sqlite-store/store.test.ts @@ -177,6 +177,8 @@ describe("SqliteSyncStore basics", () => { expect(all.map((mutation) => mutation.entryId)).toEqual(["e1", "e2", "e3"]); const limited = await store.listDirtyEntries(2); expect(limited.map((mutation) => mutation.entryId)).toEqual(["e1", "e2"]); + expect((await store.listDirtyEntries(2, new Set(["e1"]))) + .map((mutation) => mutation.entryId)).toEqual(["e2", "e3"]); }); it("tracks blocked mutations separately and unblocks them", async () => { diff --git a/apps/cli/src/host/sqlite-store/store.ts b/apps/cli/src/host/sqlite-store/store.ts index 6868e33e..7bd2efdc 100644 --- a/apps/cli/src/host/sqlite-store/store.ts +++ b/apps/cli/src/host/sqlite-store/store.ts @@ -409,15 +409,13 @@ export class SqliteSyncStore implements SyncStore { return row ? toPendingMutationRow(row) : null; } - async listDirtyEntries(limit?: number): Promise { + async listDirtyEntries(limit?: number, excludedEntryIds?: ReadonlySet): Promise { + const excluded = [...(excludedEntryIds ?? [])]; const sql = `SELECT * FROM entries - WHERE pending_status = 'pending' + WHERE pending_status = 'pending'${excluded.length ? ` AND entry_id NOT IN (${excluded.map(() => "?").join(",")})` : ""} ORDER BY pending_created_at ASC, entry_id ASC${limit === undefined ? "" : " LIMIT ?"}`; - const rows = ( - limit === undefined - ? this.prepare(sql).all() - : this.prepare(sql).all(limit) - ) as unknown as SqlEntryRow[]; + const parameters = limit === undefined ? excluded : [...excluded, limit]; + const rows = this.prepare(sql).all(...parameters) as unknown as SqlEntryRow[]; return rows .map((row) => toPendingMutationRow(fromSqlRow(row))) .filter(isPresent); diff --git a/apps/obsidian-plugin/release-notes/next.md b/apps/obsidian-plugin/release-notes/next.md index 72160274..e3b78f8c 100644 --- a/apps/obsidian-plugin/release-notes/next.md +++ b/apps/obsidian-plugin/release-notes/next.md @@ -6,7 +6,7 @@ ## Changed -- Sync uploaded notes sooner while slower attachments are still uploading. +- 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/apps/obsidian-plugin/src/adapters/dexie-store.test.ts b/apps/obsidian-plugin/src/adapters/dexie-store.test.ts index f8eb00af..394f232b 100644 --- a/apps/obsidian-plugin/src/adapters/dexie-store.test.ts +++ b/apps/obsidian-plugin/src/adapters/dexie-store.test.ts @@ -260,6 +260,9 @@ describe("DexieSyncStore", () => { "mutation-middle", ]); + expect((await store.listDirtyEntries(2, new Set(["entry-early"]))) + .map((entry) => entry.entryId)).toEqual(["entry-middle", "entry-late"]); + await store.clearDirtyEntryByMutationId("mutation-middle"); expect((await store.listDirtyEntries()).map((entry) => entry.mutationId)).toEqual([ diff --git a/apps/obsidian-plugin/src/adapters/dexie-store/store.ts b/apps/obsidian-plugin/src/adapters/dexie-store/store.ts index 57fefccb..abda36c4 100644 --- a/apps/obsidian-plugin/src/adapters/dexie-store/store.ts +++ b/apps/obsidian-plugin/src/adapters/dexie-store/store.ts @@ -349,13 +349,16 @@ export class DexieSyncStore implements SyncStore { return row ? toPendingMutationRow(row) : null; } - async listDirtyEntries(limit?: number): Promise { + async listDirtyEntries(limit?: number, excludedEntryIds?: ReadonlySet): Promise { let collection = this.db.entries .where("[pendingStatus+pendingCreatedAt+entryId]") .between( ["pending", MIN_PENDING_CREATED_AT, ""], ["pending", [], []], ); + if (excludedEntryIds?.size) { + collection = collection.filter((row) => !excludedEntryIds.has(row.entryId)); + } if (limit !== undefined) { collection = collection.limit(limit); } diff --git a/packages/sync-client/benchmarks/README.md b/packages/sync-client/benchmarks/README.md index 633f6755..21e691da 100644 --- a/packages/sync-client/benchmarks/README.md +++ b/packages/sync-client/benchmarks/README.md @@ -122,5 +122,5 @@ per-file service times: The new scenarios complement the existing 1 GiB client throughput benchmark. 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 early completion, bounded upload concurrency, and queue -preservation and retry after pipeline failures. +tests separately verify continuous replenishment, bounded outstanding work, +pull priority, and queue preservation and retry after pipeline failures. diff --git a/packages/sync-client/src/sync/engine/__tests__/auto-sync/local-changes.test.ts b/packages/sync-client/src/sync/engine/__tests__/auto-sync/local-changes.test.ts index 76549055..1c8ec196 100644 --- a/packages/sync-client/src/sync/engine/__tests__/auto-sync/local-changes.test.ts +++ b/packages/sync-client/src/sync/engine/__tests__/auto-sync/local-changes.test.ts @@ -172,7 +172,7 @@ describe("SyncAutoLoop local changes", () => { await vi.advanceTimersByTimeAsync(100); expect(pushPendingMutations).toHaveBeenCalledTimes(1); - expect(pushPendingMutations).toHaveBeenCalledWith(session); + expect(pushPendingMutations).toHaveBeenCalledWith(session, expect.any(Function)); expect(pullOnce).toHaveBeenCalledTimes(0); autoLoop.stop(); await store.close(); @@ -226,7 +226,7 @@ describe("SyncAutoLoop local changes", () => { await Promise.resolve(); expect(unblockFileSizeBlockedMutations).toHaveBeenCalledWith(session); - expect(pushPendingMutations).toHaveBeenCalledWith(session); + expect(pushPendingMutations).toHaveBeenCalledWith(session, expect.any(Function)); expect(pullOnce).toHaveBeenCalledTimes(0); autoLoop.stop(); await store.close(); @@ -271,7 +271,7 @@ describe("SyncAutoLoop local changes", () => { }, ); await vi.waitFor(() => { - expect(pushPendingMutations).toHaveBeenCalledWith(session); + expect(pushPendingMutations).toHaveBeenCalledWith(session, expect.any(Function)); }); expect(unblockFileSizeBlockedMutations).toHaveBeenCalledTimes(2); diff --git a/packages/sync-client/src/sync/engine/__tests__/push-service/continuous-push.test.ts b/packages/sync-client/src/sync/engine/__tests__/push-service/continuous-push.test.ts new file mode 100644 index 00000000..42f626c6 --- /dev/null +++ b/packages/sync-client/src/sync/engine/__tests__/push-service/continuous-push.test.ts @@ -0,0 +1,181 @@ +import { describe, expect, it, vi } from "vitest"; +import { createTestSyncStore } from "../../../../test-support/in-memory-sync-store"; +import { encodeUtf8, hashBytes } from "../../../core/content"; +import { SyncAutoLoop } from "../../auto-sync"; +import { PushNoProgressError, SyncPushService } from "../../push-service"; +import { createRealtimeClient } from "../auto-sync/helpers"; +import { + createPushSession, createToken, encryptMutationMetadata, + ignoreProgress, TEST_VAULT_KEY, +} from "./helpers"; + +function gate() { + let resolve!: () => void; + const promise = new Promise((done) => { resolve = done; }); + return { promise, resolve }; +} + +async function fixture(count: number) { + const store = createTestSyncStore(); + const body = encodeUtf8("body"); + const hash = await hashBytes(body); + for (let index = 0; index < count; index++) { + await store.markEntryDirty({ + mutationId: `mutation-${index}`, entryId: `entry-${index}`, + op: "upsert", baseRevision: 0, blobId: `blob-${index}`, hash, + encryptedMetadata: await encryptMutationMetadata({ + entryId: `entry-${index}`, baseRevision: 0, op: "upsert", + blobId: `blob-${index}`, path: `file-${index}.md`, hash, + }), + createdAt: index, + }); + } + let cursor = 0; + const session = createPushSession(async (mutation) => ({ + cursor: ++cursor, entryId: mutation.entryId, revision: mutation.baseRevision + 1, + })); + const deps = { + getApiBaseUrl: () => "http://127.0.0.1:8787", + getSyncToken: async () => createToken(), + getSyncStore: () => store, + getRemoteVaultKey: () => TEST_VAULT_KEY, + fileReader: { async readBytes() { return body; } }, + }; + return { store, session, deps }; +} + +describe("continuous push", () => { + it("replenishes beyond the first 100 entries before a slow upload finishes", async () => { + const { store, session, deps } = await fixture(125); + const slow = gate(); + let active = 0; + let maxActive = 0; + let owned = 0; + let maxOwned = 0; + const completed: string[] = []; + const service = new SyncPushService({ + ...deps, + blobClient: { + async uploadBlob(_url, _token, _vault, id) { + active++; + maxActive = Math.max(maxActive, active); + try { if (id === "blob-0") await slow.promise; } + finally { active--; } + }, + }, + onFileSyncStarted() { owned++; maxOwned = Math.max(maxOwned, owned); }, + onFileSyncCompleted({ path }) { owned--; completed.push(path); }, + }); + const push = service.pushPendingMutations(session); + try { + await vi.waitFor(() => expect(completed).toHaveLength(124)); + expect(completed).not.toContain("file-0.md"); + expect(maxActive).toBeLessThanOrEqual(12); + expect(maxOwned).toBeLessThanOrEqual(100); + } finally { slow.resolve(); } + expect(await push).toMatchObject({ mutationsPushed: 125, hasMore: false }); + expect(await store.getCursor()).toBe(125); + await store.close(); + }); + + it.each(["pull", "stop"] as const)("joins started work before %s and preserves the remaining queue", async (action) => { + const { store, session, deps } = await fixture(125); + const uploads = gate(); + let started = 0; + let completed = 0; + let pulled = false; + const service = new SyncPushService({ + ...deps, + blobClient: { + async uploadBlob() { + started++; + await uploads.promise; + }, + }, + onFileSyncCompleted() { completed++; }, + }); + const loop = new SyncAutoLoop({ + ...deps, + realtimeClient: createRealtimeClient(undefined, (next) => { + next.commitMutations = session.commitMutations; + }), + pushPendingMutations: (next, shouldYield) => + service.pushPendingMutations(next, ignoreProgress, shouldYield), + async pullOnce() { + expect(started).toBe(12); + expect(completed).toBe(12); + expect(await store.getCursor()).toBe(12); + expect(await store.listDirtyEntries()).toHaveLength(113); + pulled = true; + }, + }); + await loop.start(); + loop.notifyLocalChange(); + loop.flushDebouncedPush(); + const drain = loop.waitForInFlightDrain(); + try { + await vi.waitFor(() => expect(started).toBe(12)); + if (action === "pull") loop.requestPull(13); + else loop.stop(); + expect(pulled).toBe(false); + uploads.resolve(); + await drain; + expect(pulled).toBe(action === "pull"); + expect(completed).toBe(action === "pull" ? 125 : 12); + expect(await store.listDirtyEntries()).toHaveLength(action === "pull" ? 0 : 113); + } finally { + uploads.resolve(); + loop.stop(); + await drain; + await store.close(); + } + }); + + it("yields without starting files when pull is already pending", async () => { + const { store, session, deps } = await fixture(1); + const uploadBlob = vi.fn(async () => {}); + const service = new SyncPushService({ ...deps, blobClient: { uploadBlob } }); + expect(await service.pushPendingMutations(session, ignoreProgress, () => true)) + .toMatchObject({ mutationsPushed: 0, hasMore: true }); + expect(uploadBlob).not.toHaveBeenCalled(); + await store.close(); + }); + + it("does not start a store selection if pull arrives during its read", async () => { + const { store, session, deps } = await fixture(1); + const selection = gate(); + let reading = false; + let yielding = false; + const list = store.listDirtyEntries.bind(store); + vi.spyOn(store, "listDirtyEntries").mockImplementationOnce(async (...args) => { + reading = true; + await selection.promise; + return list(...args); + }); + const uploadBlob = vi.fn(async () => {}); + const service = new SyncPushService({ ...deps, blobClient: { uploadBlob } }); + const push = service.pushPendingMutations(session, ignoreProgress, () => yielding); + try { + await vi.waitFor(() => expect(reading).toBe(true)); + yielding = true; + } finally { selection.resolve(); } + expect(await push).toMatchObject({ mutationsPushed: 0, hasMore: true }); + expect(uploadBlob).not.toHaveBeenCalled(); + await store.close(); + }); + + it("returns a retryable failure if a changing file repeatedly prevents progress", async () => { + const { store, session, deps } = await fixture(1); + let reads = 0; + const service = new SyncPushService({ + ...deps, + fileReader: { async readBytes() { return encodeUtf8(`edit-${++reads}`); } }, + blobClient: { async uploadBlob() { throw new Error("changed content must be requeued"); } }, + }); + await expect(service.pushPendingMutations(session)).rejects.toBeInstanceOf(PushNoProgressError); + expect(reads).toBeLessThan(10); + expect(await store.listDirtyEntries()).toHaveLength(1); + expect(await store.getCursor()).toBe(0); + await store.close(); + }); +}); diff --git a/packages/sync-client/src/sync/engine/__tests__/push-service/drain-batching.test.ts b/packages/sync-client/src/sync/engine/__tests__/push-service/drain-batching.test.ts index 86744f1b..f641f02f 100644 --- a/packages/sync-client/src/sync/engine/__tests__/push-service/drain-batching.test.ts +++ b/packages/sync-client/src/sync/engine/__tests__/push-service/drain-batching.test.ts @@ -17,7 +17,7 @@ import { } from "./helpers"; describe("SyncPushService drain: batching", () => { - it("counts only selected mutations and preserves pending work after the drain limit", async () => { + it("drains more than 1000 mutations and reports all completed work", async () => { const store = createTestSyncStore(); const mutationCount = 1_001; const body = new TextEncoder().encode("body"); @@ -68,8 +68,8 @@ describe("SyncPushService drain: batching", () => { const result = await service.pushPendingMutations(session); - expect(result.mutationsPushed).toBe(1_000); - expect(result.hasMore).toBe(true); + expect(result.mutationsPushed).toBe(mutationCount); + expect(result.hasMore).toBe(false); expect(progressUpdates[0]).toEqual({ direction: "push", totalKnown: false, completedEntries: 0, diff --git a/packages/sync-client/src/sync/engine/auto-sync.ts b/packages/sync-client/src/sync/engine/auto-sync.ts index a411b5fc..5273f4cc 100644 --- a/packages/sync-client/src/sync/engine/auto-sync.ts +++ b/packages/sync-client/src/sync/engine/auto-sync.ts @@ -31,6 +31,7 @@ export interface SyncAutoLoopDeps { getSyncStore: () => SyncCursorStore | null; pushPendingMutations: ( session: SyncRealtimeSession, + shouldYield: () => boolean, ) => Promise; unblockFileSizeBlockedMutations?: ( session: SyncRealtimeSession, @@ -632,7 +633,9 @@ export class SyncAutoLoop { if (!session) { throw new Error("Sync realtime session is not connected."); } - const pushResult = await this.deps.pushPendingMutations(session); + const pushResult = await this.deps.pushPendingMutations(session, () => + !this.isActive() || this.pendingWork.pullTargetCursor !== null, + ); pushCompleted = true; if (pushResult.stopReason === "storage_quota_exceeded") { try { diff --git a/packages/sync-client/src/sync/engine/push-preparation-pipeline.ts b/packages/sync-client/src/sync/engine/push-preparation-pipeline.ts index 3a1ba915..ec36e276 100644 --- a/packages/sync-client/src/sync/engine/push-preparation-pipeline.ts +++ b/packages/sync-client/src/sync/engine/push-preparation-pipeline.ts @@ -1,76 +1,104 @@ export const PUSH_BATCH_SIZE = 100; -// Coalesce completions during slower uploads; flush immediately once the -// selection is fully prepared so fast local transports pay no timer delay. const COMMIT_COALESCE_MS = 100; -/** - * Prepare one bounded selection of dirty entries while the consumer commits - * completed batches. The caller must finish this selection before reading the - * next one, so a newer mutation for the same entry cannot overtake its commit. - */ -export async function* preparePushBatches( - items: T[], +/** Own entries until the consumer finishes committing and applying the batch. */ +export async function* preparePushBatches( + load: (limit: number, excluded: ReadonlySet) => Promise, concurrency: number, prepare: (item: T) => Promise, + shouldYield: () => boolean, ): AsyncGenerator { - const ready: Array<{ index: number; value: U }> = []; + const owned = new Set(); + const waiting: Array<{ item: T; index: number }> = []; + const ready: Array<{ item: T; index: number; value: U }> = []; + const jobs = new Set>(); const normalized = Number.isFinite(concurrency) ? Math.floor(concurrency) : 1; - const workerCount = Math.max(0, Math.min(Math.max(1, normalized), items.length)); + const workerCount = Math.min(PUSH_BATCH_SIZE, Math.max(1, normalized)); let nextIndex = 0; - let activeWorkers = workerCount; let stopped = false; + let yielding = false; + let sourceEmpty = false; let failure: { error: unknown } | undefined; let wake: (() => void) | undefined; let timer: ReturnType | undefined; let flushReady = false; - const workers = Promise.all(Array.from({ length: workerCount }, async () => { - try { - while (!stopped && nextIndex < items.length) { - const index = nextIndex++; - const value = await prepare(items[index]!); - if (stopped) break; - ready.push({ index, value }); - if (timer === undefined) { - timer = setTimeout(() => { - flushReady = true; - wake?.(); - }, COMMIT_COALESCE_MS); - } + function throwIfFailed(): void { + if (failure) throw failure.error; + } + + function updateAndCheckSupplyStop(): boolean { + // Once requested, yielding remains active for the rest of this drain. + yielding ||= shouldYield(); + return stopped || yielding; + } + + function pump(): void { + while (!updateAndCheckSupplyStop() && jobs.size < workerCount && waiting.length > 0) { + const work = waiting.shift()!; + const job = Promise.resolve().then(() => prepare(work.item)).then( + (value) => { + ready.push({ ...work, value }); + if (timer === undefined && !stopped) { + timer = setTimeout(() => { + flushReady = true; + wake?.(); + }, COMMIT_COALESCE_MS); + } + }, + (error: unknown) => { + failure ??= { error }; + stopped = true; + }, + ).finally(() => { + jobs.delete(job); + pump(); wake?.(); - } - } catch (error) { - failure ??= { error }; - stopped = true; - } finally { - activeWorkers -= 1; - wake?.(); + }); + jobs.add(job); } - })); + } try { while (true) { - if (failure) throw failure.error; + throwIfFailed(); + if (!updateAndCheckSupplyStop() && !sourceEmpty && owned.size < PUSH_BATCH_SIZE) { + const items = await load(PUSH_BATCH_SIZE - owned.size, owned); + // A pull can arrive during the store read. Do not start its results. + if (!updateAndCheckSupplyStop()) { + sourceEmpty = items.length === 0; + for (const item of items) { + owned.add(item.entryId); + waiting.push({ item, index: nextIndex++ }); + } + pump(); + } + } + throwIfFailed(); + const preparationFinished = jobs.size === 0 && (yielding || waiting.length === 0); if (ready.length > 0 && - (flushReady || ready.length >= PUSH_BATCH_SIZE || activeWorkers === 0)) { + (flushReady || ready.length >= PUSH_BATCH_SIZE || preparationFinished)) { clearTimeout(timer); timer = undefined; flushReady = false; - // Preserve queue order among ready entries without waiting for slow ones. ready.sort((left, right) => left.index - right.index); - yield ready.splice(0, PUSH_BATCH_SIZE).map(({ value }) => value); + const batch = ready.splice(0, PUSH_BATCH_SIZE); + yield batch.map(({ value }) => value); + for (const { item } of batch) owned.delete(item.entryId); + sourceEmpty = false; continue; } - if (activeWorkers === 0) break; - await new Promise((resolve) => { wake = resolve; }); + if (preparationFinished && (yielding || sourceEmpty)) break; + await new Promise((resolve) => { + wake = resolve; + }); wake = undefined; } } finally { stopped = true; clearTimeout(timer); - // Uploads cannot be cancelled through the blob client. Join them before the - // caller flushes the store or disposes the shared crypto context, including - // when a commit fails or the consumer stops on a rejected mutation. - await workers; + // No cancellation is available on the blob client. Join started operations + // before the caller disposes crypto or allows pull to mutate the store. + await Promise.all(jobs); } } diff --git a/packages/sync-client/src/sync/engine/push-service.ts b/packages/sync-client/src/sync/engine/push-service.ts index 7cf2a592..33093ea4 100644 --- a/packages/sync-client/src/sync/engine/push-service.ts +++ b/packages/sync-client/src/sync/engine/push-service.ts @@ -1,4 +1,4 @@ -import { preparePushBatches, PUSH_BATCH_SIZE } from "./push-preparation-pipeline"; +import { preparePushBatches } from "./push-preparation-pipeline"; import { SyncWorkProgress } from "./work-progress"; import type { SyncOperationProgress } from "../runtime/user-visible-status"; import type { SyncBlobClient } from "../remote/blob-client"; @@ -35,7 +35,6 @@ import { } from "./push-mutation-committer"; import { metadataContextFromMutation } from "./push-mutation-shared"; -const DEFAULT_PUSH_DRAIN_LIMIT = 1_000; const DEFAULT_PUSH_PREPARE_CONCURRENCY = 12; export interface SyncPushServiceDeps extends SyncContentRuntimeDeps { @@ -99,6 +98,7 @@ export class SyncPushService { async pushPendingMutations( session: SyncRealtimeSession, onProgress = this.deps.onProgress ?? (async (_progress: SyncOperationProgress) => {}), + shouldYield: () => boolean = () => false, ): Promise { const store = this.deps.getSyncStore(); if (!store) { @@ -119,7 +119,14 @@ export class SyncPushService { let fileSizeBlocked = 0; let shouldPullAfterPush = false; const acceptedCursors: number[] = []; - let processedMutations = 0; + // Allow one immediate retry after requeueing; repeated churn must use the + // auto loop's retry backoff instead of keeping an unbounded drain alive. + const requeuedEntries = new Set(); + let requeueLimitReached = false; + const recordRequeue = (entryId: string) => { + if (requeuedEntries.has(entryId)) requeueLimitReached = true; + requeuedEntries.add(entryId); + }; let hasMore = false; let stopAfterCurrentBatch = false; let stopReason: PushPendingMutationsResult["stopReason"]; @@ -131,220 +138,206 @@ export class SyncPushService { syncCryptoContext, ); try { - while (processedMutations < DEFAULT_PUSH_DRAIN_LIMIT) { - const remainingBudget = DEFAULT_PUSH_DRAIN_LIMIT - processedMutations; - const pending = await store.listDirtyEntries( - Math.min(PUSH_BATCH_SIZE, remainingBudget), - ); - if (pending.length === 0) { - hasMore = false; - break; - } - - progress.register(pending.map((mutation) => mutation.mutationId)); - for await (const preparedMutations of this.preparePendingMutations( - mutationCommitter, - syncCryptoContext, - store, - token, - session, - pending, - )) { - const committable: Array<{ - mutation: (typeof preparedMutations)[number]["mutation"]; - prepared: PreparedPushMutation; - path: string; - }> = []; - - for (const { mutation, prepared, path } of preparedMutations) { - processedMutations += 1; - - if (!prepared) { - mutationsRequeued += 1; - this.deps.onFileSyncFailed?.({ - operation: mutation.op, - path, - reason: "requeued", - }); - continue; - } - if ("skipped" in prepared) { - this.deps.onFileSyncFailed?.({ - operation: mutation.op, - path, - reason: prepared.reason, - }); - if (prepared.reason === "file_too_large") { - fileSizeBlocked += 1; - } - if (prepared.reason === "storage_quota_exceeded") { - stopAfterCurrentBatch = true; - stopReason = "storage_quota_exceeded"; - break; - } - continue; - } - - committable.push({ mutation, prepared, path }); + for await (const preparedMutations of this.preparePendingMutations( + mutationCommitter, + syncCryptoContext, + store, + token, + session, + progress, + () => shouldYield() || shouldPullAfterPush || requeueLimitReached, + )) { + const committable: Array<{ + mutation: (typeof preparedMutations)[number]["mutation"]; + prepared: PreparedPushMutation; + path: string; + }> = []; + + for (const { mutation, prepared, path } of preparedMutations) { + if (!prepared) { + mutationsRequeued += 1; + recordRequeue(mutation.entryId); + this.deps.onFileSyncFailed?.({ + operation: mutation.op, + path, + reason: "requeued", + }); + continue; } - - if (committable.length === 0) { - await store.flush(); - await onProgress(progress.snapshot()); - if (stopAfterCurrentBatch) { + if ("skipped" in prepared) { + this.deps.onFileSyncFailed?.({ + operation: mutation.op, + path, + reason: prepared.reason, + }); + if (prepared.reason === "file_too_large") { + fileSizeBlocked += 1; + } + if (prepared.reason === "storage_quota_exceeded") { + stopAfterCurrentBatch = true; + stopReason = "storage_quota_exceeded"; break; } continue; } - let committed; - try { - committed = await session.commitMutations( - committable.map(({ prepared }) => prepared.commitPayload), - ); - } catch (error) { - for (const { mutation, path } of committable) { - this.deps.onFileSyncFailed?.({ - operation: mutation.op, - path, - reason: "commit_failed", - }); - } - throw error; + committable.push({ mutation, prepared, path }); + } + + if (committable.length === 0) { + await store.flush(); + await onProgress(progress.snapshot()); + if (stopAfterCurrentBatch) { + break; } - const resultsByMutationId = new Map( - committed.results.map((result) => [result.mutationId, result]), + continue; + } + + let committed; + try { + committed = await session.commitMutations( + committable.map(({ prepared }) => prepared.commitPayload), ); + } catch (error) { + for (const { mutation, path } of committable) { + this.deps.onFileSyncFailed?.({ + operation: mutation.op, + path, + reason: "commit_failed", + }); + } + throw error; + } + const resultsByMutationId = new Map( + committed.results.map((result) => [result.mutationId, result]), + ); - const acceptedPushMutations: AcceptedPushMutationRow[] = []; - const acceptedFiles: Array<{ - operation: "upsert" | "delete"; - path: string; - revision: number; - }> = []; - const rejectedPushMutations: Array<{ - mutation: (typeof committable)[number]["mutation"]; - result: Extract; - }> = []; - for (const { mutation, prepared, path } of committable) { - const batchResult = resultsByMutationId.get(mutation.mutationId); - if (!batchResult) { - throw new Error(`Commit batch did not include ${mutation.mutationId}.`); - } + const acceptedPushMutations: AcceptedPushMutationRow[] = []; + const acceptedFiles: Array<{ + operation: "upsert" | "delete"; + path: string; + revision: number; + }> = []; + const rejectedPushMutations: Array<{ + mutation: (typeof committable)[number]["mutation"]; + result: Extract; + path: string; + }> = []; + for (const { mutation, prepared, path } of committable) { + const batchResult = resultsByMutationId.get(mutation.mutationId); + if (!batchResult) { + throw new Error(`Commit batch did not include ${mutation.mutationId}.`); + } - if (batchResult.status === "accepted") { - const acceptedPushMutation = - await mutationCommitter.buildAcceptedPushMutation( - mutation, - prepared, - batchResult, - ); - acceptedPushMutations.push(acceptedPushMutation); - if (acceptedPushMutation.remoteBlobId) { - // The coordinator made this blob live as part of accepting the - // mutation. A replay after a local apply failure is idempotent, - // and a redundant upload is rejected before reaching storage. - this.remotelyStagedBlobIds.delete(acceptedPushMutation.remoteBlobId); - } - cursor = Math.max(cursor, batchResult.cursor); - acceptedCursors.push(batchResult.cursor); - acceptedFiles.push({ - operation: mutation.op, - path, - revision: batchResult.revision, - }); - filesCreatedOrUpdated += mutation.op === "upsert" ? 1 : 0; - filesDeleted += mutation.op === "delete" ? 1 : 0; - mutationsPushed += 1; - continue; + if (batchResult.status === "accepted") { + const acceptedPushMutation = + await mutationCommitter.buildAcceptedPushMutation( + mutation, + prepared, + batchResult, + ); + acceptedPushMutations.push(acceptedPushMutation); + if (acceptedPushMutation.remoteBlobId) { + // The coordinator made this blob live as part of accepting the + // mutation. A replay after a local apply failure is idempotent, + // and a redundant upload is rejected before reaching storage. + this.remotelyStagedBlobIds.delete(acceptedPushMutation.remoteBlobId); } + cursor = Math.max(cursor, batchResult.cursor); + acceptedCursors.push(batchResult.cursor); + acceptedFiles.push({ + operation: mutation.op, + path, + revision: batchResult.revision, + }); + filesCreatedOrUpdated += mutation.op === "upsert" ? 1 : 0; + filesDeleted += mutation.op === "delete" ? 1 : 0; + mutationsPushed += 1; + requeuedEntries.delete(mutation.entryId); + continue; + } + + rejectedPushMutations.push({ mutation, result: batchResult, path }); + } - rejectedPushMutations.push({ mutation, result: batchResult }); + try { + await store.applyAcceptedPushBatch(acceptedPushMutations, { + remoteVaultKey, + }); + } catch (error) { + for (const accepted of acceptedFiles) { + this.deps.onFileSyncFailed?.({ + operation: accepted.operation, + path: accepted.path, + reason: "local_commit_failed", + }); } + throw error; + } + await store.flush(); + progress.complete(acceptedPushMutations.map(({ mutation }) => mutation.mutationId)); + for (const accepted of acceptedFiles) { + this.deps.onFileSyncCompleted?.(accepted); + } + mutationCommitter.forgetRemotelyStagedBlobsIfMissing( + rejectedPushMutations.map(({ mutation, result }) => ({ + blobId: mutation.blobId, + error: result, + })), + ); + + for (const { mutation, result: batchResult, path } of rejectedPushMutations) { + let result; try { - await store.applyAcceptedPushBatch(acceptedPushMutations, { - remoteVaultKey, - }); + result = await mutationCommitter.handleRejectedPreparedMutation( + store, + mutation, + batchResult, + ); } catch (error) { - for (const accepted of acceptedFiles) { - this.deps.onFileSyncFailed?.({ - operation: accepted.operation, - path: accepted.path, - reason: "local_commit_failed", - }); - } + this.deps.onFileSyncFailed?.({ + operation: mutation.op, + path, + reason: "rejected", + }); throw error; } - await store.flush(); - progress.complete(acceptedPushMutations.map(({ mutation }) => mutation.mutationId)); - for (const accepted of acceptedFiles) { - this.deps.onFileSyncCompleted?.(accepted); + conflictsCreated += result.conflictsCreated; + shouldPullAfterPush = shouldPullAfterPush || result.shouldPullAfterPush; + + if (result.status === "stale") { + this.deps.onFileSyncFailed?.({ + operation: mutation.op, + path, + reason: "stale_revision", + }); + mutationsRequeued += 1; + recordRequeue(mutation.entryId); + stopAfterCurrentBatch = true; + continue; } - - mutationCommitter.forgetRemotelyStagedBlobsIfMissing( - rejectedPushMutations.map(({ mutation, result }) => ({ - blobId: mutation.blobId, - error: result, - })), - ); - - for (const { mutation, result: batchResult } of rejectedPushMutations) { - const path = committable.find( - (item) => item.mutation.mutationId === mutation.mutationId, - )?.path ?? ""; - let result; - try { - result = await mutationCommitter.handleRejectedPreparedMutation( - store, - mutation, - batchResult, - ); - } catch (error) { - this.deps.onFileSyncFailed?.({ - operation: mutation.op, - path, - reason: "rejected", - }); - throw error; - } - conflictsCreated += result.conflictsCreated; - shouldPullAfterPush = shouldPullAfterPush || result.shouldPullAfterPush; - - if (result.status === "stale") { - this.deps.onFileSyncFailed?.({ - operation: mutation.op, - path, - reason: "stale_revision", - }); - mutationsRequeued += 1; - stopAfterCurrentBatch = true; - continue; - } - if (result.status === "requeued") { - this.deps.onFileSyncFailed?.({ - operation: mutation.op, - path, - reason: "requeued", - }); - mutationsRequeued += 1; - continue; - } - if (result.status === "conflict") { - this.deps.onFileSyncFailed?.({ - operation: mutation.op, - path, - reason: "conflict", - }); - continue; - } + if (result.status === "requeued") { + this.deps.onFileSyncFailed?.({ + operation: mutation.op, + path, + reason: "requeued", + }); + mutationsRequeued += 1; + recordRequeue(mutation.entryId); + continue; } - await store.flush(); - await onProgress(progress.snapshot()); - if (stopAfterCurrentBatch) { - break; + if (result.status === "conflict") { + this.deps.onFileSyncFailed?.({ + operation: mutation.op, + path, + reason: "conflict", + }); + continue; } } + await store.flush(); + await onProgress(progress.snapshot()); if (stopAfterCurrentBatch) { break; } @@ -366,6 +359,10 @@ export class SyncPushService { await store.flush(); } + if (requeueLimitReached && !shouldYield() && !shouldPullAfterPush) { + throw new PushNoProgressError(); + } + progress.seal(); await onProgress(progress.snapshot()); @@ -441,18 +438,20 @@ export class SyncPushService { store: SyncPushStore, token: SyncTokenResponse, session: SyncRealtimeSession, - pending: PendingMutationRow[], + progress: SyncWorkProgress, + shouldYield: () => boolean, ): AsyncGenerator< Array<{ - mutation: (typeof pending)[number]; + mutation: PendingMutationRow; prepared: Awaited>; path: string; }> > { return preparePushBatches( - pending, + (limit, excluded) => store.listDirtyEntries(limit, excluded), this.deps.prepareConcurrency ?? DEFAULT_PUSH_PREPARE_CONCURRENCY, async (mutation) => { + progress.register([mutation.mutationId]); let path = ""; try { path = ( @@ -481,10 +480,18 @@ export class SyncPushService { throw error; } }, + shouldYield, ); } } +export class PushNoProgressError extends Error { + constructor() { + super("Push repeatedly requeued an entry without accepting it."); + this.name = "PushNoProgressError"; + } +} + function getContiguousAcceptedCursor( currentCursor: number, acceptedCursors: number[], diff --git a/packages/sync-client/src/sync/runtime/sync-engine.ts b/packages/sync-client/src/sync/runtime/sync-engine.ts index 78077567..31b27641 100644 --- a/packages/sync-client/src/sync/runtime/sync-engine.ts +++ b/packages/sync-client/src/sync/runtime/sync-engine.ts @@ -229,9 +229,9 @@ export class SyncEngine { ? this.deps.createWebSocket(url, protocols) : new WebSocket(url, protocols), }), - pushPendingMutations: async (session) => + pushPendingMutations: async (session, shouldYield) => await this.withSyncActivity("push", async (report) => { - return await this.syncPushService.pushPendingMutations(session, report); + return await this.syncPushService.pushPendingMutations(session, report, shouldYield); }), unblockFileSizeBlockedMutations: async (session) => await this.withSyncActivity("local", async () => { diff --git a/packages/sync-client/src/sync/store/ports.ts b/packages/sync-client/src/sync/store/ports.ts index a22a0443..57d0b40b 100644 --- a/packages/sync-client/src/sync/store/ports.ts +++ b/packages/sync-client/src/sync/store/ports.ts @@ -63,7 +63,8 @@ export interface SyncMutationStore { options?: MarkEntryDirtyOptions, ): Promise; getDirtyEntryMutation(entryId: string): Promise; - listDirtyEntries(limit?: number): Promise; + /** Exclude owned entries before applying the limit, preserving queue order. */ + listDirtyEntries(limit?: number, excludedEntryIds?: ReadonlySet): Promise; listBlockedDirtyEntriesByReason( reason: PendingMutationBlockedReason, ): Promise; diff --git a/packages/sync-client/src/test-support/in-memory-sync-store.ts b/packages/sync-client/src/test-support/in-memory-sync-store.ts index 5cada4ed..dcc99149 100644 --- a/packages/sync-client/src/test-support/in-memory-sync-store.ts +++ b/packages/sync-client/src/test-support/in-memory-sync-store.ts @@ -249,10 +249,10 @@ export class InMemorySyncStore implements SyncStore { return cloneMutation(this.entries.get(entryId)?.dirty ?? null); } - async listDirtyEntries(limit?: number): Promise { + async listDirtyEntries(limit?: number, excludedEntryIds?: ReadonlySet): Promise { const mutations = [...this.entries.values()] .flatMap((state) => - state.dirty && (state.dirty.status ?? "pending") === "pending" + !excludedEntryIds?.has(state.entryId) && state.dirty && (state.dirty.status ?? "pending") === "pending" ? [cloneMutation(state.dirty)] : [], )