fix(server): reuse the compiler worker during plugin build batches

Amortize TypeScript startup across built-ins and queued builds, then retire the worker after one idle second. Carry preparation environments explicitly across worker reuse. Keep dependency-backed example coverage with its existing app fixture.
This commit is contained in:
Mohamed Boudra
2026-10-26 19:33:32 +01:00
parent 7c9a7c191e
commit 4b5c52ebe6
7 changed files with 173 additions and 55 deletions
@@ -1,4 +1,4 @@
import { parentPort, workerData } from "node:worker_threads";
import { parentPort } from "node:worker_threads";
import type { ZodIssue } from "zod";
import { compilePlugin, compileDirectoryPlugin } from "./compiler-build.js";
import { runPluginBuild } from "./preparation-build.js";
@@ -16,32 +16,43 @@ export type CompilerReply =
};
const port = parentPort!;
const task = workerData as CompilerTask;
// The compiler owner serializes requests; the port queues the first request
// until this entrypoint has loaded, including in source/tsx development.
port.on("message", (task: CompilerTask) => {
void run(task);
});
try {
let value: unknown;
switch (task.kind) {
case "entries":
value = await compilePlugin(task.entries);
break;
case "directory":
value = await compileDirectoryPlugin(task.directory, task.version, task.builtinId);
break;
case "prepare":
await runPluginBuild(task.directory, task.commands, {
info: (data, message) =>
port.postMessage({ kind: "log", data, message } satisfies CompilerReply),
});
break;
async function run(task: CompilerTask): Promise<void> {
try {
let value: unknown;
switch (task.kind) {
case "entries":
value = await compilePlugin(task.entries);
break;
case "directory":
value = await compileDirectoryPlugin(task.directory, task.version, task.builtinId);
break;
case "prepare":
await runPluginBuild(
task.directory,
task.commands,
{
info: (data, message) =>
port.postMessage({ kind: "log", data, message } satisfies CompilerReply),
},
task.environment,
);
break;
}
port.postMessage({ kind: "result", value } satisfies CompilerReply);
} catch (cause) {
const error = cause instanceof Error ? cause : new Error(String(cause));
port.postMessage({
kind: "error",
error,
...("issues" in error ? { issues: error.issues as ZodIssue[] } : {}),
...("errors" in error ? { errors: error.errors as unknown[] } : {}),
...("warnings" in error ? { warnings: error.warnings as unknown[] } : {}),
} satisfies CompilerReply);
}
port.postMessage({ kind: "result", value } satisfies CompilerReply);
} catch (cause) {
const error = cause instanceof Error ? cause : new Error(String(cause));
port.postMessage({
kind: "error",
error,
...("issues" in error ? { issues: error.issues as ZodIssue[] } : {}),
...("errors" in error ? { errors: error.errors as unknown[] } : {}),
...("warnings" in error ? { warnings: error.warnings as unknown[] } : {}),
} satisfies CompilerReply);
}
+89 -21
View File
@@ -16,7 +16,12 @@ export {
export type CompilerTask =
| { kind: "entries"; entries: Parameters<typeof build.compilePlugin>[0] }
| { kind: "directory"; directory: string; version: string; builtinId?: string }
| { kind: "prepare"; directory: string; commands: PluginManifest["build"] };
| {
kind: "prepare";
directory: string;
commands: PluginManifest["build"];
environment: NodeJS.ProcessEnv;
};
interface TaskResult {
entries: Awaited<ReturnType<typeof build.compilePlugin>>;
@@ -24,15 +29,41 @@ interface TaskResult {
prepare: void;
}
// One active build per daemon, including preparation. A fresh isolate per task
// releases TypeScript's ASTs and caches and inherits current command environment.
const compileQueue = pLimit(1);
const IDLE_TIMEOUT_MS = 1_000;
export function runCompilerTask<T extends CompilerTask>(
task: T,
onLog?: (data: Record<string, unknown>, message: string) => void,
): Promise<TaskResult[T["kind"]]> {
return compileQueue(async () => {
// Reuse the isolate during a build batch (notably daemon bootstrap), then release
// TypeScript's heap when idle. Only this owner admits work or stops the worker.
class PluginCompiler {
private readonly queue = pLimit(1);
private worker: Worker | null = null;
private idleTimer: ReturnType<typeof setTimeout> | null = null;
private stopping: Promise<unknown> = Promise.resolve();
run<T extends CompilerTask>(
task: T,
onLog?: (data: Record<string, unknown>, message: string) => void,
): Promise<TaskResult[T["kind"]]> {
return this.queue(async () => {
if (this.idleTimer) clearTimeout(this.idleTimer);
this.idleTimer = null;
await this.stopping;
const worker = this.worker ?? this.startWorker();
worker.ref();
try {
return await this.request<T>(worker, task, onLog);
} finally {
if (this.worker === worker) {
worker.unref();
this.idleTimer = setTimeout(() => this.stopWorker(worker), IDLE_TIMEOUT_MS);
this.idleTimer.unref();
} else {
await this.stopping;
}
}
});
}
private startWorker(): Worker {
const source = import.meta.url.endsWith(".ts");
const url = new URL(source ? "./compiler-worker.ts" : "./compiler-worker.js", import.meta.url);
// Node workers do not inherit tsx's in-process loader registration. Register
@@ -40,16 +71,40 @@ export function runCompilerTask<T extends CompilerTask>(
const worker = source
? new Worker(
`import(${JSON.stringify(pathToFileURL(createRequire(import.meta.url).resolve("tsx/esm/api")).href)}).then(({ register }) => { register(); return import(${JSON.stringify(url.href)}); });`,
{ eval: true, execArgv: [], workerData: task },
{ eval: true, execArgv: [] },
)
: new Worker(url, { execArgv: [], workerData: task });
: new Worker(url, { execArgv: [] });
this.worker = worker;
// Keep error/exit ownership while idle too. The next caller gets a fresh
// isolate after a crash; author build commands are never retried.
worker.on("error", () => this.stopWorker(worker));
worker.on("exit", () => {
if (this.worker === worker) this.stopWorker(worker);
});
return worker;
}
private stopWorker(worker: Worker): void {
if (this.worker !== worker) return;
this.worker = null;
if (this.idleTimer) clearTimeout(this.idleTimer);
this.idleTimer = null;
this.stopping = worker.terminate();
}
private async request<T extends CompilerTask>(
worker: Worker,
task: T,
onLog?: (data: Record<string, unknown>, message: string) => void,
): Promise<TaskResult[T["kind"]]> {
let onMessage: (reply: CompilerReply) => void;
let onError: (error: Error) => void;
let onExit: (code: number) => void;
try {
return await new Promise<TaskResult[T["kind"]]>((resolve, reject) => {
worker.on("error", reject);
worker.on("exit", (code) =>
reject(new Error(`Plugin compiler worker exited (code ${code})`)),
);
worker.on("message", (reply: CompilerReply) => {
onError = reject;
onExit = (code) => reject(new Error(`Plugin compiler worker exited (code ${code})`));
onMessage = (reply) => {
if (reply.kind === "log") {
onLog?.(reply.data, reply.message);
} else if (reply.kind === "result") {
@@ -60,14 +115,27 @@ export function runCompilerTask<T extends CompilerTask>(
if (reply.warnings) Object.assign(error, { warnings: reply.warnings });
reject(error);
}
});
};
worker.on("error", onError);
worker.on("exit", onExit);
worker.on("message", onMessage);
worker.postMessage(task, []);
});
} finally {
// A crash rejects only its task. The queue always advances and the next
// caller gets a new worker; no retries of author build commands.
await worker.terminate();
worker.off("message", onMessage!);
worker.off("error", onError!);
worker.off("exit", onExit!);
}
});
}
}
const compiler = new PluginCompiler();
export function runCompilerTask<T extends CompilerTask>(
task: T,
onLog?: (data: Record<string, unknown>, message: string) => void,
): Promise<TaskResult[T["kind"]]> {
return compiler.run(task, onLog);
}
export function compilePlugin(entries: Parameters<typeof build.compilePlugin>[0]) {
@@ -1612,7 +1612,7 @@ it("keeps reload and enable responsive while compiling a real plugin", async ()
Array.from({ length: 10_000 }, (_, index) => offset + index).join(",") + ",",
);
}
await entry.write("]; export default () => values;");
await entry.write("]; export default function contribute() { return () => values; }");
} finally {
await entry.close();
}
@@ -1643,6 +1643,15 @@ it("keeps reload and enable responsive while compiling a real plugin", async ()
expect(service.catalog()[0]?.clientBundle).toContain("349999");
expect(longest).toBeLessThan(20);
}
// Evaluate after both probes so parsing the returned code is not charged to compilation.
const factory = globalThis.eval(service.catalog()[0]!.clientBundle) as (
require: (specifier: string) => never,
) => { default: () => () => number[] };
const values = factory((specifier) => {
throw new Error(`Unexpected fixture dependency: ${specifier}`);
}).default()();
expect(values).toHaveLength(350_000);
expect(values.every((value, index) => value === index)).toBe(true);
} finally {
await service.stopAllPlugins();
}
@@ -8,10 +8,11 @@ export async function runPluginBuild(
directory: string,
commands: PluginManifest["build"],
logger: { info(data: Record<string, unknown>, message: string): void },
environment: NodeJS.ProcessEnv,
): Promise<void> {
for (const command of commands ?? []) {
logger.info({ directory, command }, "Running plugin build command");
const result = await run(command, directory);
const result = await run(command, directory, environment);
if (result.stdout)
logger.info({ directory, command, output: result.stdout }, "Plugin build stdout");
if (result.stderr)
@@ -29,11 +30,15 @@ export async function runPluginBuild(
}
}
function run(command: string[], directory: string): Promise<CommandResult> {
function run(
command: string[],
directory: string,
environment: NodeJS.ProcessEnv,
): Promise<CommandResult> {
return new Promise((resolve) => {
const executable = command[0]!;
const arguments_ = command.slice(1);
const child = spawnProcess(executable, arguments_, { cwd: directory });
const child = spawnProcess(executable, arguments_, { cwd: directory, env: environment });
let stdout = "";
let stderr = "";
let error: Error | undefined;
@@ -143,6 +143,28 @@ describe("runPluginBuild", () => {
]);
});
it("uses the caller's current environment across consecutive builds", async () => {
const directory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-build-"));
directories.push(directory);
const previous = process.env.PASEO_PLUGIN_TEST_BUILD_VALUE;
const command = [
process.execPath,
"-e",
'require("node:fs").writeFileSync(process.argv[1], process.env.PASEO_PLUGIN_TEST_BUILD_VALUE);',
];
try {
process.env.PASEO_PLUGIN_TEST_BUILD_VALUE = "first";
await runPluginBuild(directory, [[...command, "first.txt"]], createTestLogger());
process.env.PASEO_PLUGIN_TEST_BUILD_VALUE = "second";
await runPluginBuild(directory, [[...command, "second.txt"]], createTestLogger());
expect(await readFile(path.join(directory, "first.txt"), "utf8")).toBe("first");
expect(await readFile(path.join(directory, "second.txt"), "utf8")).toBe("second");
} finally {
if (previous === undefined) delete process.env.PASEO_PLUGIN_TEST_BUILD_VALUE;
else process.env.PASEO_PLUGIN_TEST_BUILD_VALUE = previous;
}
});
it("handles empty or undefined build commands gracefully", async () => {
const directory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-build-"));
directories.push(directory);
@@ -8,7 +8,8 @@ export async function runPluginBuild(
logger: pino.Logger,
): Promise<void> {
if (!commands?.length) return;
await runCompilerTask({ kind: "prepare", directory, commands }, (data, message) =>
logger.info(data, message),
await runCompilerTask(
{ kind: "prepare", directory, commands, environment: { ...process.env } },
(data, message) => logger.info(data, message),
);
}
@@ -2059,8 +2059,10 @@ export default function contribute(server) {
const examplesDirectory = fileURLToPath(
new URL("../../../../../plugin-examples/", import.meta.url),
);
// The Markdown example owns npm preparation in the real app fixture:
// packages/app/e2e/browser/plugin-file-panel-examples.spec.ts.
const examples = (await readdir(examplesDirectory, { withFileTypes: true }))
.filter((entry) => entry.isDirectory())
.filter((entry) => entry.isDirectory() && entry.name !== "markdown-panel")
.map((entry) => entry.name);
it.each(examples)("builds and loads the %s plugin example", async (example) => {