diff --git a/packages/server/src/server/test-fixtures/reconciliation-registries.ts b/packages/server/src/server/test-fixtures/reconciliation-registries.ts new file mode 100644 index 0000000000..2d020ec23f --- /dev/null +++ b/packages/server/src/server/test-fixtures/reconciliation-registries.ts @@ -0,0 +1,117 @@ +import type pino from "pino"; +import { createPersistedProjectRecord } from "../workspace-registry.js"; +import type { + PersistedProjectRecord, + PersistedWorkspaceRecord, + ProjectRegistry, + WorkspaceRegistry, +} from "../workspace-registry.js"; + +export function createTestRegistries() { + let archiveGate: { + started: ReturnType; + release: ReturnType; + } | null = null; + const holdNextArchive = () => { + archiveGate = { started: deferred(), release: deferred() }; + return archiveGate; + }; + const projects = new Map(); + const workspaces = new Map(); + + const projectRegistry: ProjectRegistry = { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => Array.from(projects.values()), + get: async (id: string) => projects.get(id) ?? null, + getOrCreateActiveByRoot: async (input) => { + const existing = Array.from(projects.values()).find( + (project) => !project.archivedAt && project.rootPath === input.rootPath, + ); + if (existing) return existing; + const record = createPersistedProjectRecord({ + projectId: `prj_${projects.size}`, + rootPath: input.rootPath, + kind: input.kind, + displayName: input.displayName, + createdAt: input.timestamp, + updatedAt: input.timestamp, + }); + projects.set(record.projectId, record); + return record; + }, + update: async (id, updater) => { + const current = projects.get(id); + if (!current) return null; + const updated = updater(current); + projects.set(id, updated); + return updated; + }, + upsert: async (record: PersistedProjectRecord) => { + projects.set(record.projectId, record); + }, + archive: async (id: string, archivedAt: string) => { + const existing = projects.get(id); + if (existing) { + projects.set(id, { ...existing, archivedAt, updatedAt: archivedAt }); + } + }, + remove: async (id: string) => { + projects.delete(id); + }, + }; + + const workspaceRegistry: WorkspaceRegistry = { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => Array.from(workspaces.values()), + get: async (id: string) => workspaces.get(id) ?? null, + update: async (id, updater) => { + const existing = workspaces.get(id); + if (!existing) return null; + const updated = updater(existing); + workspaces.set(id, updated); + return updated; + }, + upsert: async (record: PersistedWorkspaceRecord) => { + workspaces.set(record.workspaceId, record); + }, + archive: async (id: string, archivedAt: string) => { + const gate = archiveGate; + archiveGate = null; + if (gate) { + gate.started.resolve(); + await gate.release.promise; + } + const existing = workspaces.get(id); + if (!existing || existing.archivedAt) return false; + workspaces.set(id, { ...existing, archivedAt, updatedAt: archivedAt }); + return true; + }, + remove: async (id: string) => { + workspaces.delete(id); + }, + }; + + return { projects, workspaces, projectRegistry, workspaceRegistry, holdNextArchive }; +} + +export function createTestLogger() { + const logger = { + child: () => logger, + trace: () => undefined, + debug: () => undefined, + info: () => undefined, + warn: () => undefined, + error: () => undefined, + }; + return logger as unknown as pino.Logger; +} + +export function deferred(): { promise: Promise; resolve(): void } { + let resolve!: () => void; + const promise = new Promise((accept) => { + resolve = accept; + }); + return { promise, resolve }; +} diff --git a/packages/server/src/server/test-fixtures/reconciliation-responsiveness.ts b/packages/server/src/server/test-fixtures/reconciliation-responsiveness.ts new file mode 100644 index 0000000000..9f85cfb48f --- /dev/null +++ b/packages/server/src/server/test-fixtures/reconciliation-responsiveness.ts @@ -0,0 +1,71 @@ +import path from "node:path"; +import { setImmediate as nextTurn } from "node:timers/promises"; +import { measureMainThreadWork } from "../../test-utils/main-thread-work.js"; +import { createTestRegistries, createTestLogger } from "./reconciliation-registries.js"; +import { + createPersistedProjectRecord, + createPersistedWorkspaceRecord, +} from "../workspace-registry.js"; +import { WorkspaceReconciliationService } from "../workspace-reconciliation-service.js"; +import { deriveProjectKey } from "../project-key.js"; + +const rootPath = process.argv[2]; +if (!rootPath) throw new Error("Project root is required"); +const timestamp = "2025-01-01T00:00:00.000Z"; +const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries(); +projects.set( + "p1", + createPersistedProjectRecord({ + projectId: "p1", + rootPath, + projectKey: deriveProjectKey({ + rootPath, + remoteUrl: null, + worktreeRoot: null, + mainRepoRoot: null, + }), + kind: "non_git", + displayName: "project", + createdAt: timestamp, + updatedAt: timestamp, + }), +); +for (let i = 0; i < 20000; i++) { + const workspaceId = `w${i}`; + workspaces.set( + workspaceId, + createPersistedWorkspaceRecord({ + workspaceId, + projectId: "p1", + cwd: path.join(rootPath, workspaceId), + kind: "directory", + displayName: workspaceId, + createdAt: timestamp, + updatedAt: timestamp, + }), + ); +} +const service = new WorkspaceReconciliationService({ + projectRegistry, + workspaceRegistry, + logger: createTestLogger(), +}); + +// Compile before launch and measure in plain Node, outside the test runner's +// thread and TypeScript loader. Keep the first real reconciliation cold. +await nextTurn(); +try { + const measured = await measureMainThreadWork("workspace reconciliation main-thread work", () => + service.runOnce(), + ); + process.stdout.write( + JSON.stringify({ + blockingMs: measured.blockingMs, + filesystemRequests: measured.filesystemRequests, + changesApplied: measured.result.changesApplied.length, + allArchived: [...workspaces.values()].every((workspace) => workspace.archivedAt !== null), + }), + ); +} finally { + service.dispose(); +} diff --git a/packages/server/src/server/workspace-reconciliation-service.test.ts b/packages/server/src/server/workspace-reconciliation-service.test.ts index 0e9f5a2c2d..7ef187d16e 100644 --- a/packages/server/src/server/workspace-reconciliation-service.test.ts +++ b/packages/server/src/server/workspace-reconciliation-service.test.ts @@ -1,27 +1,45 @@ -import { measureMainThreadWork } from "../test-utils/main-thread-work.js"; -import { execFileSync } from "node:child_process"; +import { execFile, execFileSync } from "node:child_process"; +import { promisify } from "node:util"; +import { fileURLToPath } from "node:url"; +import { build } from "esbuild"; +import { + createTestRegistries, + createTestLogger, + deferred, +} from "./test-fixtures/reconciliation-registries.js"; import { mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages"; import type pino from "pino"; -import { afterEach, describe, expect, test } from "vitest"; +import { afterAll, beforeAll, afterEach, describe, expect, test } from "vitest"; import { createPersistedProjectRecord, createPersistedWorkspaceRecord, } from "./workspace-registry.js"; -import type { - PersistedProjectRecord, - PersistedWorkspaceRecord, - ProjectRegistry, - WorkspaceRegistry, -} from "./workspace-registry.js"; import { type ReconciliationChange, WorkspaceReconciliationService, } from "./workspace-reconciliation-service.js"; import { deriveProjectKey } from "./project-key.js"; +let probeDirectory: string; +let probePath: string; +beforeAll(async () => { + probeDirectory = mkdtempSync(path.join(tmpdir(), "reconciliation-probe-")); + probePath = path.join(probeDirectory, "reconciliation.mjs"); + await build({ + entryPoints: [ + fileURLToPath(new URL("./test-fixtures/reconciliation-responsiveness.ts", import.meta.url)), + ], + outfile: probePath, + bundle: true, + platform: "node", + format: "esm", + }); +}); +afterAll(() => rmSync(probeDirectory, { recursive: true, force: true })); + function canonicalLocalProjectKey(rootPath: string): string { return deriveProjectKey({ rootPath, @@ -31,107 +49,6 @@ function canonicalLocalProjectKey(rootPath: string): string { }); } -function createTestRegistries() { - let archiveGate: { - started: ReturnType; - release: ReturnType; - } | null = null; - const holdNextArchive = () => { - archiveGate = { started: deferred(), release: deferred() }; - return archiveGate; - }; - const projects = new Map(); - const workspaces = new Map(); - - const projectRegistry: ProjectRegistry = { - initialize: async () => {}, - existsOnDisk: async () => true, - list: async () => Array.from(projects.values()), - get: async (id: string) => projects.get(id) ?? null, - getOrCreateActiveByRoot: async (input) => { - const existing = Array.from(projects.values()).find( - (project) => !project.archivedAt && project.rootPath === input.rootPath, - ); - if (existing) return existing; - const record = createPersistedProjectRecord({ - projectId: `prj_${projects.size}`, - rootPath: input.rootPath, - kind: input.kind, - displayName: input.displayName, - createdAt: input.timestamp, - updatedAt: input.timestamp, - }); - projects.set(record.projectId, record); - return record; - }, - update: async (id, updater) => { - const current = projects.get(id); - if (!current) return null; - const updated = updater(current); - projects.set(id, updated); - return updated; - }, - upsert: async (record: PersistedProjectRecord) => { - projects.set(record.projectId, record); - }, - archive: async (id: string, archivedAt: string) => { - const existing = projects.get(id); - if (existing) { - projects.set(id, { ...existing, archivedAt, updatedAt: archivedAt }); - } - }, - remove: async (id: string) => { - projects.delete(id); - }, - }; - - const workspaceRegistry: WorkspaceRegistry = { - initialize: async () => {}, - existsOnDisk: async () => true, - list: async () => Array.from(workspaces.values()), - get: async (id: string) => workspaces.get(id) ?? null, - update: async (id, updater) => { - const existing = workspaces.get(id); - if (!existing) return null; - const updated = updater(existing); - workspaces.set(id, updated); - return updated; - }, - upsert: async (record: PersistedWorkspaceRecord) => { - workspaces.set(record.workspaceId, record); - }, - archive: async (id: string, archivedAt: string) => { - const gate = archiveGate; - archiveGate = null; - if (gate) { - gate.started.resolve(); - await gate.release.promise; - } - const existing = workspaces.get(id); - if (!existing || existing.archivedAt) return false; - workspaces.set(id, { ...existing, archivedAt, updatedAt: archivedAt }); - return true; - }, - remove: async (id: string) => { - workspaces.delete(id); - }, - }; - - return { projects, workspaces, projectRegistry, workspaceRegistry, holdNextArchive }; -} - -function createTestLogger() { - const logger = { - child: () => logger, - trace: () => undefined, - debug: () => undefined, - info: () => undefined, - warn: () => undefined, - error: () => undefined, - }; - return logger as unknown as pino.Logger; -} - interface CapturedLogRecord { message: string; payload: unknown; @@ -210,14 +127,6 @@ function createCheckout( }; } -function deferred(): { promise: Promise; resolve(): void } { - let resolve!: () => void; - const promise = new Promise((accept) => { - resolve = accept; - }); - return { promise, resolve }; -} - class TestCheckouts { readonly reads: string[] = []; private readonly checkouts = new Map(); @@ -626,53 +535,30 @@ describe("WorkspaceReconciliationService", () => { test("keeps the event loop responsive while inspecting 20000 real workspace paths", async () => { const rootPath = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-many-"))); tempDirs.push(rootPath); - const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries(); - projects.set( - "p1", - createPersistedProjectRecord({ - projectId: "p1", - rootPath, - projectKey: canonicalLocalProjectKey(rootPath), - kind: "non_git", - displayName: "project", - createdAt: timestamp, - updatedAt: timestamp, - }), - ); - for (let i = 0; i < 20000; i++) { - const workspaceId = `w${i}`; - workspaces.set( - workspaceId, - createPersistedWorkspaceRecord({ - workspaceId, - projectId: "p1", - cwd: path.join(rootPath, workspaceId), - kind: "directory", - displayName: workspaceId, - createdAt: timestamp, - updatedAt: timestamp, - }), - ); - } - const service = new WorkspaceReconciliationService({ - projectRegistry, - workspaceRegistry, - logger: createTestLogger(), + // Unrelated runner work must not consume the reconciliation budget. + const unrelatedWork = new Promise((resolve) => { + setImmediate(() => { + const cpu = Reflect.get(process, "threadCpuUsage") as typeof process.cpuUsage; + const start = cpu(); + while (cpu(start).user + cpu(start).system < 80_000) {} + resolve(); + }); }); - try { - const { result, blockingMs, filesystemRequests } = await measureMainThreadWork( - "workspace reconciliation main-thread work", - () => service.runOnce(), - ); - expect(blockingMs).toBeLessThan(20); - expect(filesystemRequests).toBe(20001); - expect(result.changesApplied).toHaveLength(20000); - expect([...workspaces.values()].every((workspace) => workspace.archivedAt !== null)).toBe( - true, - ); - } finally { - service.dispose(); - } + const { stdout, stderr } = await promisify(execFile)(process.execPath, [probePath, rootPath], { + timeout: 25_000, + }); + process.stderr.write(stderr); + const { changesApplied, blockingMs, filesystemRequests, allArchived } = JSON.parse(stdout) as { + changesApplied: number; + blockingMs: number; + filesystemRequests: number; + allArchived: boolean; + }; + await unrelatedWork; + expect(blockingMs).toBeLessThan(20); + expect(filesystemRequests).toBe(20001); + expect(changesApplied).toBe(20000); + expect(allArchived).toBe(true); }, 30000); test("preserves workspace archival that lands during boot reconciliation", async () => {