test(server): isolate reconciliation responsiveness from runner work

The in-process probe charges unrelated runner callbacks against the service budget. Measure the cold workload in plain Node while continuous parent-side CPU guards isolation; retain the 20ms limit, exact asynchronous filesystem count, and archival results.
This commit is contained in:
Mohamed Boudra
2026-10-30 17:40:19 +01:00
parent b5e8601066
commit 1557765860
3 changed files with 260 additions and 150 deletions
@@ -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<typeof deferred>;
release: ReturnType<typeof deferred>;
} | null = null;
const holdNextArchive = () => {
archiveGate = { started: deferred(), release: deferred() };
return archiveGate;
};
const projects = new Map<string, PersistedProjectRecord>();
const workspaces = new Map<string, PersistedWorkspaceRecord>();
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<void>; resolve(): void } {
let resolve!: () => void;
const promise = new Promise<void>((accept) => {
resolve = accept;
});
return { promise, resolve };
}
@@ -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();
}
@@ -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<typeof deferred>;
release: ReturnType<typeof deferred>;
} | null = null;
const holdNextArchive = () => {
archiveGate = { started: deferred(), release: deferred() };
return archiveGate;
};
const projects = new Map<string, PersistedProjectRecord>();
const workspaces = new Map<string, PersistedWorkspaceRecord>();
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);