Batch sync path and index tombstone scans

.changeset/batched-sync-path-resolution.md

@@ -0,0 +1,5 @@
+---
+"@cloudflare/dofs": patch
+---
+
+Resolve sync change paths in batches and index tombstone scans by `(op, rev)`.

packages/dofs/src/schema/core.ts

@@ -13,9 +13,11 @@
 // dirents leaf so the (parent, name) resolve read is covering
 // (no separate index needed). Bumped to 7 when `_vfs_sync_operations`
 // and `_vfs_sync_skips` landed, carrying the durable half of a
-// restartable pull or push. See `schema/migrations.ts` for the
-// migration list; `sync.ts` carries the fresh-install DDL.
-export const SCHEMA_VERSION = 7;
+// restartable pull or push. Bumped to 8 when `vfs_changes` gained
+// `vfs_changes_by_op_rev`, so tombstone scans can restrict on the rev
+// window. See `schema/migrations.ts` for the migration list; `sync.ts`
+// carries the fresh-install DDL.
+export const SCHEMA_VERSION = 8;
 export const ROOT_INODE = 1;
 
 export const CORE_STATEMENTS = [

packages/dofs/src/schema/index.test.ts

@@ -444,6 +444,74 @@ describe("initializeSchema", () => {
     expect(norm(tableSql("vfs_chunks"))).toBe(norm(freshSql("vfs_chunks")));
   });
 
+  it("creates vfs_changes_by_op_rev on a fresh DB and uses it for the tombstone scan", () => {
+    const storage = new SQLiteTestStorage();
+    const db = new Database(storage);
+    initializeSchema(db, () => 0);
+
+    const indexNames = db
+      .all<{ name: string }>("SELECT name FROM sqlite_master WHERE type = 'index'")
+      .map((r) => r.name);
+    expect(indexNames).toContain("vfs_changes_by_op_rev");
+
+    // The coalesce tombstone query must be able to restrict on rev
+    // rather than scanning the whole table. Before this index the plan
+    // was `SCAN vfs_changes USING INDEX vfs_changes_by_path`, which
+    // ignores the rev predicate entirely.
+    const plan = db
+      .all<{ detail: string }>(
+        `EXPLAIN QUERY PLAN
+         SELECT path, MAX(rev) AS rev FROM vfs_changes
+          WHERE rev > ? AND op = 'delete' GROUP BY path`,
+        0,
+      )
+      .map((r) => r.detail)
+      .join(" | ");
+    expect(plan).toContain("vfs_changes_by_op_rev");
+    expect(plan).not.toMatch(/SCAN vfs_changes(?! USING)/);
+  });
+
+  it("adds vfs_changes_by_op_rev on the v7 -> v8 upgrade, preserving tombstones", () => {
+    // Stage a v7-shape database: everything current except the new
+    // index. The migrator must add it without disturbing existing rows.
+    const storage = new SQLiteTestStorage();
+    const db = new Database(storage);
+
+    initializeSchema(db, () => 0);
+    db.run("INSERT INTO vfs_changes (rev, path, op) VALUES (?, ?, 'delete')", 5, "/a.txt");
+    db.run("INSERT INTO vfs_changes (rev, path, op) VALUES (?, ?, 'delete')", 7, "/b.txt");
+    db.run("INSERT INTO vfs_changes (rev, path, op) VALUES (?, ?, 'delete')", 9, "/a.txt");
+    const before = db.all("SELECT id, rev, path, op FROM vfs_changes ORDER BY id");
+
+    // Roll the database back to v7: drop the index and restamp.
+    db.run("DROP INDEX IF EXISTS vfs_changes_by_op_rev");
+    db.run("UPDATE vfs_meta SET v = 7 WHERE k = 'schema_version'");
+    expect(
+      db
+        .all<{ name: string }>("SELECT name FROM sqlite_master WHERE type = 'index'")
+        .map((r) => r.name),
+    ).not.toContain("vfs_changes_by_op_rev");
+
+    initializeSchema(db, () => 0);
+
+    expect(db.one<{ v: number }>("SELECT v FROM vfs_meta WHERE k = 'schema_version'")?.v).toBe(
+      SCHEMA_VERSION,
+    );
+    expect(
+      db
+        .all<{ name: string }>("SELECT name FROM sqlite_master WHERE type = 'index'")
+        .map((r) => r.name),
+    ).toContain("vfs_changes_by_op_rev");
+    expect(db.all("SELECT id, rev, path, op FROM vfs_changes ORDER BY id")).toEqual(before);
+
+    // The pre-existing path index must survive the upgrade too.
+    expect(
+      db
+        .all<{ name: string }>("SELECT name FROM sqlite_master WHERE type = 'index'")
+        .map((r) => r.name),
+    ).toContain("vfs_changes_by_path");
+  });
+
   it("upgrades a v6 database with the sync operation tables", () => {
     const storage = new SQLiteTestStorage();
     const db = new Database(storage);

packages/dofs/src/schema/migrations.ts

@@ -206,13 +206,29 @@ function v6_to_v7_sync_operations(db: Database): void {
   );
 }
 
+// v7 → v8 — add `vfs_changes_by_op_rev`. The push tick's tombstone
+// query filters `rev > ? AND op = 'delete'` and groups by path;
+// without an (op, rev) index the planner scans vfs_changes in path
+// order and never applies the rev predicate, reading the whole table
+// to return the few rows inside the watermark window.
+//
+// Index-only migration: no table is rewritten and no row is touched,
+// so existing tombstones carry through untouched. The CREATE is
+// duplicated in `sync.ts`'s fresh-install DDL; keep the two in
+// lockstep. IF NOT EXISTS keeps this safe if a database somehow
+// already has the index.
+function v7_to_v8_changes_op_rev_index(db: Database): void {
+  db.run(`CREATE INDEX IF NOT EXISTS vfs_changes_by_op_rev ON vfs_changes(op, rev)`);
+}
+
 export const MIGRATIONS: readonly Migration[] = [
   { from: 1, to: 2, migrator: v1_to_v2_add_mounts_mode },
   { from: 2, to: 3, migrator: v2_to_v3_add_size_column },
   { from: 3, to: 4, migrator: v3_to_v4_watermark_backend_column },
   { from: 4, to: 5, migrator: v4_to_v5_without_rowid },
   { from: 5, to: 6, migrator: v5_to_v6_push_cursor },
   { from: 6, to: 7, migrator: v6_to_v7_sync_operations },
+  { from: 7, to: 8, migrator: v7_to_v8_changes_op_rev_index },
 ] as const;
 
 // Apply every migration whose `from` matches the current version,

packages/dofs/src/schema/sync.ts

@@ -23,6 +23,23 @@ export const SYNC_STATEMENTS = [
   // index. Used on every recordDelete and on every push-tick that
   // processes tombstones.
   `CREATE INDEX IF NOT EXISTS vfs_changes_by_path ON vfs_changes(path, id DESC)`,
+  // coalesceChanges scans tombstones with
+  // `WHERE rev > ? AND op = 'delete' GROUP BY path`. Neither of the
+  // indexes above serves that: vfs_changes_by_rev(rev) can drive the
+  // range but leaves GROUP BY path to a sort, so the planner instead
+  // scans vfs_changes_by_path in path order and ignores the rev
+  // predicate entirely — reading the whole table to return the handful
+  // of rows in the watermark window.
+  //
+  // (op, rev) puts the equality column first and the range column
+  // second, which is the shape SQLite can drive both halves from. The
+  // GROUP BY still needs a temp b-tree, but the scan is now bounded by
+  // the rev window instead of the table.
+  //
+  // This only started to matter once node_modules entered the sync set:
+  // every reinstall appends thousands of tombstones and the table only
+  // grows. Added at schema v8; `schema/migrations.ts` owns the upgrade.
+  `CREATE INDEX IF NOT EXISTS vfs_changes_by_op_rev ON vfs_changes(op, rev)`,
   // Watermarks are keyed by (k, backend) so a workspace hosting
   // multiple backends keeps each backend's sync cursors
   // independent. The `backend` column was added at schema v3;

packages/dofs/src/sync/coalesce.ts

@@ -1,7 +1,7 @@
 import type { Database } from "../storage.js";
 import { type ChangeEntry, materialiseChange } from "./changes.js";
 import { isIgnored } from "./ignore.js";
-import { pathsOf } from "./paths.js";
+import { pathsOfMany } from "./paths.js";
 import { type ChangeCursor, compareChangeCursors } from "./watermarks.js";
 
 // Yield one ChangeEntry per path touched after `after`. Per-path
@@ -63,11 +63,19 @@ export async function* coalesceChanges(
           lowerRev,
           through.rev,
         );
+  // Resolve every touched inode to its path(s) in one batched pass.
+  // Doing this per inode issued O(N x depth) statements, which
+  // dominated the tick once node_modules entered the sync set.
+  const pathsByInode = pathsOfMany(
+    db,
+    touched.map((row) => row.inode),
+  );
+
   for (const { inode, rev } of touched) {
     // One inode can carry several hardlink names; every name has to
     // become a candidate so the wire materialises each, not just the
     // arbitrary one pathOf would return.
-    for (const path of pathsOf(db, inode)) {
+    for (const path of pathsByInode.get(inode) ?? []) {
       if (!inCursorWindow({ rev, path }, cursor, through)) continue;
       if (isIgnored(path, ignore)) continue;
       const prior = candidates.get(path);

packages/dofs/src/sync/paths.test.ts

@@ -0,0 +1,141 @@
+import { describe, expect, it } from "vitest";
+
+import { link } from "../fs/link.js";
+import { mkdir } from "../fs/mkdir.js";
+import { withDB } from "../fs/with-db.js";
+import { writeFile } from "../fs/writeFile.js";
+import { ROOT_INODE } from "../schema/index.js";
+import { type pathOf, pathsOf, pathsOfMany } from "./paths.js";
+
+// Read back the inode a path currently names. Tests need this to feed
+// pathsOfMany without going through resolveInode's symlink handling.
+function inodeOf(db: Parameters<typeof pathOf>[0], parentInode: number, name: string): number {
+  const row = db.one<{ child_inode: number }>(
+    "SELECT child_inode FROM vfs_dirents WHERE parent_inode = ? AND name = ?",
+    parentInode,
+    name,
+  );
+  if (row === undefined) throw new Error(`no dirent ${name} under ${parentInode}`);
+  return row.child_inode;
+}
+
+describe("pathsOfMany", () => {
+  it("returns an empty map for no inodes", async () => {
+    await withDB((db) => {
+      expect(pathsOfMany(db, [])).toEqual(new Map());
+    });
+  });
+
+  it("maps the root inode to /", async () => {
+    await withDB((db) => {
+      expect(pathsOfMany(db, [ROOT_INODE])).toEqual(new Map([[ROOT_INODE, ["/"]]]));
+    });
+  });
+
+  it("resolves a top-level entry", async () => {
+    await withDB(async (db) => {
+      await writeFile(db, "/a.txt", "x", {}, () => 1);
+      const inode = inodeOf(db, ROOT_INODE, "a.txt");
+      expect(pathsOfMany(db, [inode])).toEqual(new Map([[inode, ["/a.txt"]]]));
+    });
+  });
+
+  it("resolves a deeply nested entry", async () => {
+    await withDB(async (db) => {
+      mkdir(db, "/a/b/c/d", { recursive: true }, () => 1);
+      await writeFile(db, "/a/b/c/d/deep.txt", "x", {}, () => 2);
+      const a = inodeOf(db, ROOT_INODE, "a");
+      const b = inodeOf(db, a, "b");
+      const c = inodeOf(db, b, "c");
+      const d = inodeOf(db, c, "d");
+      const file = inodeOf(db, d, "deep.txt");
+      expect(pathsOfMany(db, [file])).toEqual(new Map([[file, ["/a/b/c/d/deep.txt"]]]));
+    });
+  });
+
+  it("returns every hardlink name for an inode, sorted", async () => {
+    await withDB(async (db) => {
+      mkdir(db, "/dir", {}, () => 1);
+      await writeFile(db, "/one.txt", "x", {}, () => 2);
+      link(db, "/one.txt", "/two.txt");
+      link(db, "/one.txt", "/dir/three.txt");
+      const inode = inodeOf(db, ROOT_INODE, "one.txt");
+      const got = pathsOfMany(db, [inode]).get(inode);
+      expect([...(got ?? [])].sort()).toEqual(["/dir/three.txt", "/one.txt", "/two.txt"]);
+    });
+  });
+
+  it("omits inodes that are unreachable from the root", async () => {
+    await withDB((db) => {
+      // 99999 has no dirent row at all.
+      expect(pathsOfMany(db, [99999])).toEqual(new Map());
+    });
+  });
+
+  it("resolves many inodes in one call, matching pathsOf exactly", async () => {
+    await withDB(async (db) => {
+      const inodes: number[] = [];
+      mkdir(db, "/pkg", { recursive: true }, () => 1);
+      for (let i = 0; i < 25; i++) {
+        mkdir(db, `/pkg/p${i}/dist`, { recursive: true }, () => 2);
+        await writeFile(db, `/pkg/p${i}/dist/index.js`, "x", {}, () => 3);
+      }
+      for (const row of db.all<{ inode: number }>("SELECT inode FROM vfs_nodes")) {
+        inodes.push(row.inode);
+      }
+
+      const batched = pathsOfMany(db, inodes);
+      for (const inode of inodes) {
+        const expected = pathsOf(db, inode);
+        const actual = batched.get(inode) ?? [];
+        expect([...actual].sort()).toEqual([...expected].sort());
+      }
+    });
+  });
+
+  it("agrees with pathsOf on a tree containing hardlinks", async () => {
+    await withDB(async (db) => {
+      mkdir(db, "/x/y", { recursive: true }, () => 1);
+      await writeFile(db, "/x/y/f.txt", "data", {}, () => 2);
+      link(db, "/x/y/f.txt", "/x/alias.txt");
+      const inodes = db.all<{ inode: number }>("SELECT inode FROM vfs_nodes").map((r) => r.inode);
+      const batched = pathsOfMany(db, inodes);
+      for (const inode of inodes) {
+        expect([...(batched.get(inode) ?? [])].sort()).toEqual([...pathsOf(db, inode)].sort());
+      }
+    });
+  });
+
+  it("issues a bounded number of SQL statements regardless of inode count", async () => {
+    await withDB(async (db) => {
+      mkdir(db, "/big", { recursive: true }, () => 1);
+      for (let i = 0; i < 60; i++) {
+        await writeFile(db, `/big/f${i}.txt`, "x", {}, () => 2);
+      }
+      const inodes = db.all<{ inode: number }>("SELECT inode FROM vfs_nodes").map((r) => r.inode);
+
+      let statements = 0;
+      const originalAll = db.all.bind(db);
+      const originalOne = db.one.bind(db);
+      (db as unknown as { all: unknown }).all = (...args: unknown[]) => {
+        statements += 1;
+        return (originalAll as (...a: unknown[]) => unknown)(...args);
+      };
+      (db as unknown as { one: unknown }).one = (...args: unknown[]) => {
+        statements += 1;
+        return (originalOne as (...a: unknown[]) => unknown)(...args);
+      };
+      try {
+        pathsOfMany(db, inodes);
+      } finally {
+        (db as unknown as { all: unknown }).all = originalAll;
+        (db as unknown as { one: unknown }).one = originalOne;
+      }
+
+      // The batched resolver must not scale its statement count with the
+      // number of inodes. pathsOf would issue >= inodes.length here.
+      expect(statements).toBeLessThan(5);
+      expect(inodes.length).toBeGreaterThan(60);
+    });
+  });
+});