diff --git a/.changeset/batched-sync-path-resolution.md b/.changeset/batched-sync-path-resolution.md new file mode 100644 index 00000000..5dd0c02b --- /dev/null +++ b/.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)`. diff --git a/packages/dofs/src/schema/core.ts b/packages/dofs/src/schema/core.ts index a4f514da..922071d8 100644 --- a/packages/dofs/src/schema/core.ts +++ b/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 = [ diff --git a/packages/dofs/src/schema/index.test.ts b/packages/dofs/src/schema/index.test.ts index 3d1ca85e..ad5755b0 100644 --- a/packages/dofs/src/schema/index.test.ts +++ b/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); diff --git a/packages/dofs/src/schema/migrations.ts b/packages/dofs/src/schema/migrations.ts index 3a5e14d1..68eb6492 100644 --- a/packages/dofs/src/schema/migrations.ts +++ b/packages/dofs/src/schema/migrations.ts @@ -206,6 +206,21 @@ 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 }, @@ -213,6 +228,7 @@ export const MIGRATIONS: readonly Migration[] = [ { 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, diff --git a/packages/dofs/src/schema/sync.ts b/packages/dofs/src/schema/sync.ts index 902b4e9e..fdfd73e4 100644 --- a/packages/dofs/src/schema/sync.ts +++ b/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; diff --git a/packages/dofs/src/sync/coalesce.ts b/packages/dofs/src/sync/coalesce.ts index bfd4b304..d6aeb62e 100644 --- a/packages/dofs/src/sync/coalesce.ts +++ b/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); diff --git a/packages/dofs/src/sync/paths.test.ts b/packages/dofs/src/sync/paths.test.ts new file mode 100644 index 00000000..3dd156a4 --- /dev/null +++ b/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[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); + }); + }); +}); diff --git a/packages/dofs/src/sync/paths.ts b/packages/dofs/src/sync/paths.ts index d68a1e14..cb75a3aa 100644 --- a/packages/dofs/src/sync/paths.ts +++ b/packages/dofs/src/sync/paths.ts @@ -44,3 +44,122 @@ export function pathsOf(db: Database, inode: number): string[] { } return paths; } + +// How many inodes to resolve per CTE round. SQLite's parameter limit +// (SQLITE_MAX_VARIABLE_NUMBER, 999 on conservative builds) caps a bound +// json array's practical size; we bind one JSON string, but keeping the +// batches bounded also keeps the recursive walk's working set small. +const PATHS_BATCH_SIZE = 512; + +// Batched `pathsOf`. Resolves every inode in `inodes` to all of its +// hardlink names in a fixed number of statements rather than one +// dirent lookup per ancestor per inode. +// +// coalesceChanges calls this once per push tick with the entire set of +// revved inodes. The per-inode version issued O(N x depth) statements +// — ~74k round-trips for a 20k-node node_modules tree, which dominated +// the tick (~30s) even though every one of those lookups was already a +// covering-index hit. The cost was the statement count, not the index. +// +// Shape: seed one row per (target, dirent) so hardlinks fan out, then +// walk `child_inode -> parent_inode` upward, carrying `target` along. +// Rows are ordered deepest-segment-first per target and reassembled in +// JS. An inode with no dirent row (unreachable, or the root) produces +// no seed row and is simply absent from the result — same contract as +// pathsOf returning [] / pathOf returning null. +export function pathsOfMany(db: Database, inodes: readonly number[]): Map { + const out = new Map(); + if (inodes.length === 0) return out; + + // Deduplicate and pull the root out; it has no dirent row. + const unique: number[] = []; + const seen = new Set(); + for (const inode of inodes) { + if (seen.has(inode)) continue; + seen.add(inode); + if (inode === ROOT_INODE) { + out.set(ROOT_INODE, ["/"]); + continue; + } + unique.push(inode); + } + if (unique.length === 0) return out; + + for (let start = 0; start < unique.length; start += PATHS_BATCH_SIZE) { + const batch = unique.slice(start, start + PATHS_BATCH_SIZE); + collectBatch(db, batch, out); + } + return out; +} + +interface SegmentRow { + target: number; + link: string; + depth: number; + name: string; + parent_inode: number; +} + +function collectBatch(db: Database, batch: number[], out: Map): void { + // `link` distinguishes the hardlink names of one target: each seed + // dirent starts a separate upward walk, and every row on that walk + // carries its seed's identity so segments regroup correctly. Two + // hardlinks can share a parent directory ("/one.txt" and "/two.txt" + // both sit under the root), so the seed's parent inode alone is not + // a unique key — the seed's (parent_inode, name) pair is. + const rows = db.all( + `WITH RECURSIVE + targets(inode) AS ( + SELECT value FROM json_each(?) + ), + walk(target, link, depth, name, parent_inode) AS ( + SELECT t.inode, d.parent_inode || '/' || d.name, 0, d.name, d.parent_inode + FROM targets t + JOIN vfs_dirents d ON d.child_inode = t.inode + UNION ALL + SELECT w.target, w.link, w.depth + 1, d.name, d.parent_inode + FROM walk w + JOIN vfs_dirents d ON d.child_inode = w.parent_inode + WHERE w.parent_inode <> ? + ) + SELECT target, link, depth, name, parent_inode + FROM walk + ORDER BY target, link, depth DESC`, + JSON.stringify(batch), + ROOT_INODE, + ); + + // Group by (target, link). Rows arrive root-most segment first, so + // appending in order builds the path left to right. + let currentTarget: number | undefined; + let currentLink: string | undefined; + let segments: string[] = []; + let reachedRoot = false; + + const flush = () => { + if (currentTarget === undefined) return; + // Only emit paths whose walk actually terminated at the root. A + // walk that ran out of dirent rows describes an unreachable inode. + if (reachedRoot && segments.length > 0) { + const path = `/${segments.join("/")}`; + const existing = out.get(currentTarget); + if (existing === undefined) out.set(currentTarget, [path]); + else existing.push(path); + } + segments = []; + reachedRoot = false; + }; + + for (const row of rows) { + if (row.target !== currentTarget || row.link !== currentLink) { + flush(); + currentTarget = row.target; + currentLink = row.link; + } + segments.push(row.name); + // The deepest row of a completed walk is the one whose parent is + // the root; depth 0 is the target's own dirent. + if (row.parent_inode === ROOT_INODE) reachedRoot = true; + } + flush(); +}