perf(sync-client): continuously drain pushes with pull priority

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 () => {

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<PendingMutationRow[]> {
+  async listDirtyEntries(limit?: number, excludedEntryIds?: ReadonlySet<string>): Promise<PendingMutationRow[]> {
+    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);

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.
 

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([

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<PendingMutationRow[]> {
+  async listDirtyEntries(limit?: number, excludedEntryIds?: ReadonlySet<string>): Promise<PendingMutationRow[]> {
     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);
     }

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.

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);

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<void>((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();
+  });
+});

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,

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<PushPendingMutationsResult>;
   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 {