diff --git a/packages/server/src/server/test-utils/reconciliation-registries.ts b/packages/server/src/server/test-utils/reconciliation-registries.ts new file mode 100644 index 0000000000..7d409842df --- /dev/null +++ b/packages/server/src/server/test-utils/reconciliation-registries.ts @@ -0,0 +1,117 @@ +import type pino from "pino"; +import { + createPersistedProjectRecord, + type PersistedProjectRecord, + type PersistedWorkspaceRecord, + type ProjectRegistry, + type 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; +} + +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-utils/reconciliation-responsiveness.ts b/packages/server/src/server/test-utils/reconciliation-responsiveness.ts new file mode 100644 index 0000000000..a97019f72e --- /dev/null +++ b/packages/server/src/server/test-utils/reconciliation-responsiveness.ts @@ -0,0 +1,86 @@ +import assert from "node:assert/strict"; +import { createHook, executionAsyncResource } from "node:async_hooks"; +import path from "node:path"; +import { setImmediate as nextTurn } from "node:timers/promises"; +import { MessagePort } from "node:worker_threads"; +import { measureMainThreadWork } from "../../test-utils/main-thread-work.js"; +import { deriveProjectKey } from "../project-key.js"; +import { WorkspaceReconciliationService } from "../workspace-reconciliation-service.js"; +import { + createPersistedProjectRecord, + createPersistedWorkspaceRecord, +} from "../workspace-registry.js"; +import { createTestRegistries, createTestLogger } from "./reconciliation-registries.js"; + +// Bundle before launch: neither the runner nor a TypeScript loader belongs to +// the measured thread. Keep fixture construction and its garbage outside the +// cold run; GC caused by reconciliation itself remains in the budget. +const rootPath = process.argv[2]; +assert.ok(rootPath, "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(), +}); +assert.ok(global.gc, "Launch the probe with --expose-gc"); +global.gc(); +await nextTurn(); +let workerMessages = 0; +const activity = createHook({ + before() { + if (executionAsyncResource() instanceof MessagePort) workerMessages++; + }, +}).enable(); +try { + const measured = await measureMainThreadWork("workspace reconciliation main-thread work", () => + service.runOnce(), + ); + activity.disable(); + process.stdout.write( + JSON.stringify({ + blockingMs: measured.blockingMs, + filesystemRequests: measured.filesystemRequests, + yields: measured.yields, + changesAppliedCount: measured.result.changesApplied.length, + allArchived: [...workspaces.values()].every((workspace) => workspace.archivedAt !== null), + workerMessages, + }), + ); +} finally { + activity.disable(); + 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..7758f2c78d 100644 --- a/packages/server/src/server/workspace-reconciliation-service.test.ts +++ b/packages/server/src/server/workspace-reconciliation-service.test.ts @@ -1,8 +1,11 @@ -import { measureMainThreadWork } from "../test-utils/main-thread-work.js"; -import { execFileSync } from "node:child_process"; +import { execFile, execFileSync } from "node:child_process"; import { mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; +import { build } from "esbuild"; +import { createTestRegistries, createTestLogger } from "./test-utils/reconciliation-registries.js"; import type { ProjectCheckoutLitePayload } from "@getpaseo/protocol/messages"; import type pino from "pino"; import { afterEach, describe, expect, test } from "vitest"; @@ -10,12 +13,6 @@ import { createPersistedProjectRecord, createPersistedWorkspaceRecord, } from "./workspace-registry.js"; -import type { - PersistedProjectRecord, - PersistedWorkspaceRecord, - ProjectRegistry, - WorkspaceRegistry, -} from "./workspace-registry.js"; import { type ReconciliationChange, WorkspaceReconciliationService, @@ -31,107 +28,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; @@ -626,52 +522,63 @@ 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(), + const probePath = path.join(rootPath, "reconciliation-responsiveness.mjs"); + await build({ + entryPoints: [ + fileURLToPath(new URL("./test-utils/reconciliation-responsiveness.ts", import.meta.url)), + ], + outfile: probePath, + bundle: true, + platform: "node", + format: "esm", }); - try { - const { result, blockingMs, filesystemRequests } = await measureMainThreadWork( - "workspace reconciliation main-thread work", - () => service.runOnce(), + // This CPU belongs to the runner, not to reconciliation. Keep it present + // throughout the read so a later loop turn cannot contaminate the budget. + let running = true; + const unrelatedWork = () => { + const threadCpuUsage: (previous?: NodeJS.CpuUsage) => NodeJS.CpuUsage = Reflect.get( + process, + "threadCpuUsage", ); + const start = threadCpuUsage(); + let elapsed: NodeJS.CpuUsage; + do { + elapsed = threadCpuUsage(start); + } while (elapsed.user + elapsed.system < 60_000); + if (running) unrelated = setImmediate(unrelatedWork); + }; + let unrelated = setImmediate(unrelatedWork); + try { + const { stdout, stderr } = await promisify(execFile)( + process.execPath, + ["--expose-gc", probePath, rootPath], + { timeout: 15_000 }, + ); + process.stderr.write(stderr); + const { + changesAppliedCount, + blockingMs, + filesystemRequests, + yields, + allArchived, + workerMessages, + } = JSON.parse(stdout) as { + changesAppliedCount: number; + blockingMs: number; + filesystemRequests: number; + yields: number; + allArchived: boolean; + workerMessages: number; + }; expect(blockingMs).toBeLessThan(20); expect(filesystemRequests).toBe(20001); - expect(result.changesApplied).toHaveLength(20000); - expect([...workspaces.values()].every((workspace) => workspace.archivedAt !== null)).toBe( - true, - ); + expect(changesAppliedCount).toBe(20000); + expect(allArchived).toBe(true); + expect(yields).toBeGreaterThan(0); + expect(workerMessages).toBe(0); } finally { - service.dispose(); + running = false; + clearImmediate(unrelated); } }, 30000);