Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 20 additions & 6 deletions packages/host-service/src/workers/WorkerTaskRunner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,12 @@

import { randomUUID } from "node:crypto";
import { Worker } from "node:worker_threads";
import type {
SerializedWorkerError,
WorkerShutdownRequestMessage,
WorkerTaskRequestMessage,
WorkerTaskResponseMessage,
import {
isWorkerTaskPhaseMessage,
type SerializedWorkerError,
type WorkerShutdownRequestMessage,
type WorkerTaskRequestMessage,
type WorkerTaskResponseMessage,
} from "./worker-task-protocol.ts";

export class WorkerTaskError extends Error {
Expand Down Expand Up @@ -81,6 +82,8 @@ interface QueuedTask {
timeoutMs: number;
timeoutHandle?: NodeJS.Timeout;
slotId?: number;
/** Last phase the handler reported, if it reports any. */
phase?: string;
}

export const DEFAULT_TIMEOUT_MS = 60_000;
Expand Down Expand Up @@ -402,6 +405,13 @@ export class WorkerTaskRunner {
const slot = this.workerSlots.get(slotId);
if (!slot) return;

if (isWorkerTaskPhaseMessage(message)) {
if (slot.activeTaskId !== message.taskId) return;
const active = this.tasks.get(message.taskId);
if (active) active.phase = message.phase;
return;
}

if (!this.isWorkerResultMessage(message)) {
return;
}
Expand Down Expand Up @@ -476,10 +486,14 @@ export class WorkerTaskRunner {
const task = this.tasks.get(taskId);
if (!task) return;

// Name the phase when the handler reported one: the timer knows only
// that the budget expired, and the worker is retired right below, so
// this is the sole record of what was still running.
const phaseSuffix = task.phase ? ` in phase "${task.phase}"` : "";
this.rejectTask(
taskId,
new WorkerTaskError(
`[${this.name}] Task "${task.taskType}" timed out after ${task.timeoutMs}ms`,
`[${this.name}] Task "${task.taskType}" timed out after ${task.timeoutMs}ms${phaseSuffix}`,
),
);

Expand Down
7 changes: 6 additions & 1 deletion packages/host-service/src/workers/define-worker-task.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,15 @@
// clone-safe and free of DB/event-bus/native imports (see
// no-native-worker-imports.test.ts).

/** Names the step a multi-step handler is about to start. The pool keeps the
* last value and names it in the task's timeout error, so a handler that
* hangs says where it hung. Single-step handlers ignore it. */
export type ReportTaskPhase = (phase: string) => void;

export interface WorkerTaskDefinition<TInput, TResult> {
/** Namespaced as "<domain>/<task>", e.g. "git/getStatusSnapshot". */
type: string;
handler: (input: TInput) => Promise<TResult>;
handler: (input: TInput, reportPhase?: ReportTaskPhase) => Promise<TResult>;
}

export function defineWorkerTask<TInput, TResult>(
Expand Down
139 changes: 138 additions & 1 deletion packages/host-service/src/workers/host-worker-pool.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,11 @@ import * as os from "node:os";
import * as path from "node:path";
import { defineWorkerTask } from "./define-worker-task.ts";
import { HostWorkerPool } from "./host-worker-pool.ts";
import { gitStatusSnapshotTask } from "./tasks/git.ts";
import {
gitStatusSnapshotTask,
gitWorktreeRemoveTask,
gitWorktreeStateTask,
} from "./tasks/git.ts";

const WORKER_ENTRY = path.resolve(import.meta.dirname, "host-worker.ts");
const CRASH_WORKER = path.resolve(
Expand Down Expand Up @@ -68,6 +72,65 @@ function makeFixtureRepo(): string {
return dir;
}

// A `git` shimmed onto the front of PATH that blocks on one subcommand:
// `gitEnv` is the spawned process's whole environment, so this is how a real
// task can be stalled in a chosen phase without stalling real git. The rest
// of PATH stays real — the shim needs `sleep`.
//
// It blocks on a sentinel rather than for a fixed duration: a `sleep` long
// enough to survive a loaded machine is also a `sleep` left running after the
// test, and one short enough to clean up can expire mid-assertion and let the
// task succeed. Waiting on its own existence gives an unbounded hang that
// afterEach's fixture cleanup releases within ~100ms.
function makeHangingGitShim(hangOn: "remove" | "list" | "status" | "none"): {
dir: string;
gitEnv: { PATH: string };
} {
const dir = fs.mkdtempSync(path.join(os.tmpdir(), "host-worker-gitshim-"));
fixtureDirs.push(dir);
const shim = path.join(dir, "git");
const hang = `while [ -f "${shim}" ]; do sleep 0.1; done`;
fs.writeFileSync(
shim,
`#!/bin/sh
case "$*" in
*"worktree remove"*) ${hangOn === "remove" ? hang : ""} ;;
*"worktree list"*) ${hangOn === "list" ? hang : ""} ;;
status*) ${hangOn === "status" ? hang : ""} ;;
esac
exit 0
`,
{ mode: 0o755 },
);
return { dir, gitEnv: { PATH: `${dir}:/usr/bin:/bin` } };
}

function shimInput(shim: { dir: string; gitEnv: { PATH: string } }) {
return {
repoPath: shim.dir,
worktreePath: path.join(shim.dir, "wt"),
gitEnv: shim.gitEnv,
};
}

// Boot the worker before a timed run. A cold worker compiles the whole task
// module graph first (~300ms idle, more under load), which can burn a short
// budget before the handler runs at all — the task would then time out having
// reported nothing and the phase assertions would flake. Warm, the first
// phase is reported within a message hop of dispatch.
async function warmPool(pool: HostWorkerPool): Promise<void> {
await pool.run(gitWorktreeRemoveTask, shimInput(makeHangingGitShim("none")));
}

async function rejectionOf(promise: Promise<unknown>): Promise<Error> {
const error = await promise.then(
() => null,
(reason: Error) => reason,
);
if (!error) throw new Error("expected a rejection, got a resolved value");
return error;
}

describe("HostWorkerPool", () => {
test("worker result matches inline handler result (parity)", async () => {
const worktreePath = makeFixtureRepo();
Expand All @@ -88,6 +151,80 @@ describe("HostWorkerPool", () => {
]);
});

// HOST-SERVICE-17 / HOST-SERVICE-47: `git/removeWorktree` blows its budget
// in the field and the timeout named only the budget, so the steps that
// stall for unrelated reasons were indistinguishable. These drive the real
// task through the real pool and worker and assert, on the exact message
// the caller sees, that it says which step hung.
test("worker timeout names the phase that was running (worktree-remove)", async () => {
const shim = makeHangingGitShim("remove");
const pool = makePool();
await warmPool(pool);

const error = await rejectionOf(
pool.run(gitWorktreeRemoveTask, shimInput(shim), { timeoutMs: 1_000 }),
);

expect(error.message).toBe(
'[host-worker] Task "git/removeWorktree" timed out after 1000ms in phase "worktree-remove"',
);
});

test("worker timeout names the phase that was running (worktree-list)", async () => {
// Same task, same budget, same shim — only the stalling subcommand
// differs. A label that tracked the task rather than the step, or one
// left over from the step before, fails here.
const shim = makeHangingGitShim("list");
const pool = makePool();
await warmPool(pool);

const error = await rejectionOf(
pool.run(gitWorktreeRemoveTask, shimInput(shim), { timeoutMs: 1_000 }),
);

expect(error.message).toBe(
'[host-worker] Task "git/removeWorktree" timed out after 1000ms in phase "worktree-list"',
);
});

test("a task that reports no phase keeps its timeout message unchanged", async () => {
// The other half of the contract: this adds information to a failure,
// it does not change one. `git/worktreeState` reports no phases, so its
// timeout must read exactly as it did before.
const shim = makeHangingGitShim("status");
const pool = makePool();
await warmPool(pool);

const error = await rejectionOf(
pool.run(
gitWorktreeStateTask,
{ worktreePath: shim.dir, gitEnv: shim.gitEnv },
{ timeoutMs: 1_000 },
),
);

expect(error.message).toBe(
'[host-worker] Task "git/worktreeState" timed out after 1000ms',
);
});

test("inline timeout names the phase too", async () => {
// The pool degrades to inline on a missing bundle or crash-looping
// workers; that path enforces the same budget and must stay as
// diagnosable as the worker one.
const shim = makeHangingGitShim("remove");
const pool = makePool({ scriptPathResolver: () => null });
expect(pool.getMode()).toBe("inline");

const error = await rejectionOf(
pool.run(gitWorktreeRemoveTask, shimInput(shim), { timeoutMs: 1_000 }),
);

expect(error.message).toBe(
'[host-worker] Task "git/removeWorktree" timed out after 1000ms in phase "worktree-remove" (inline)',
);
});

test("unknown task type rejects with the worker's error", async () => {
const pool = makePool();
const bogus = defineWorkerTask<Record<string, never>, never>({
Expand Down
12 changes: 10 additions & 2 deletions packages/host-service/src/workers/host-worker-pool.ts
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,8 @@ export class HostWorkerPool {

return new Promise<TResult>((resolve, reject) => {
let settled = false;
// Same phase record as the worker path, minus the message hop.
let phase: string | undefined;
const settle = (fn: () => void) => {
if (settled) return;
settled = true;
Expand All @@ -178,7 +180,9 @@ export class HostWorkerPool {
settle(() =>
reject(
new WorkerTaskError(
`[host-worker] Task "${def.type}" timed out after ${timeoutMs}ms (inline)`,
`[host-worker] Task "${def.type}" timed out after ${timeoutMs}ms${
phase ? ` in phase "${phase}"` : ""
} (inline)`,
),
),
),
Expand All @@ -193,7 +197,11 @@ export class HostWorkerPool {
// Promise.resolve() wrapper: a synchronously-throwing handler must
// route through settle() too, or the timer and abort listener leak.
Promise.resolve()
.then(() => def.handler(input))
.then(() =>
def.handler(input, (reported) => {
phase = reported;
}),
)
.then((result) => settle(() => resolve(result)))
.catch((error) => settle(() => reject(error)));
});
Expand Down
10 changes: 9 additions & 1 deletion packages/host-service/src/workers/host-worker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import { killAndReapTrackedChildren } from "./worker-child-tracker.ts";
import {
isWorkerShutdownRequestMessage,
serializeWorkerError,
type WorkerTaskPhaseMessage,
type WorkerTaskRequestMessage,
} from "./worker-task-protocol.ts";

Expand Down Expand Up @@ -56,7 +57,14 @@ parentPort.on("message", async (message: unknown) => {
if (!def) {
throw new Error(`unknown worker task type: ${task.taskType}`);
}
const result = await def.handler(task.payload);
const result = await def.handler(task.payload, (phase) => {
const phaseMessage: WorkerTaskPhaseMessage = {
kind: "phase",
taskId: task.taskId,
phase,
};
parentPort?.postMessage(phaseMessage);
});
parentPort?.postMessage({
kind: "result",
taskId: task.taskId,
Expand Down
13 changes: 12 additions & 1 deletion packages/host-service/src/workers/tasks/git.ts
Original file line number Diff line number Diff line change
Expand Up @@ -200,21 +200,32 @@ export const gitWorktreeRemoveTask = defineWorkerTask<
{ stillRegistered: boolean }
>({
type: "git/removeWorktree",
handler: async ({ repoPath, worktreePath, gitEnv }) => {
// This task outlives its caller's budget in the field (HOST-SERVICE-17,
// HOST-SERVICE-47) and the timeout named only the budget. Its steps stall
// for unrelated reasons, so each announces itself before starting and the
// pool names the last one in the timeout error.
handler: async ({ repoPath, worktreePath, gitEnv }, reportPhase) => {
// Labelled from the first statement so every moment of the handler
// falls under some phase — an unlabelled timeout would be
// indistinguishable from one reported by a build without this.
reportPhase?.("resolve-path");
const git = createUserSimpleGit(repoPath).env(gitEnv);
// Remove against git's canonical path so a symlinked stored path
// (macOS `/var` → `/private/var`) still matches its registration.
// `realpathSync.native` is a blocking syscall, hence its own phase.
const target = normalizeWorktreePath(worktreePath);
// Best-effort: the registry read below is authoritative, not the
// command's locale- and version-dependent exit text. `--force --force`
// also unregisters a worktree whose directory is already gone, so no
// separate prune (which would clobber other stale worktrees' metadata)
// is needed.
reportPhase?.("worktree-remove");
await git
.raw(["worktree", "remove", "--force", "--force", target])
.catch(() => {});
// A `worktree list` failure throws out of the task: the post-remove
// state is unknown and the caller must not treat it as removed.
reportPhase?.("worktree-list");
const raw = await git.raw(["worktree", "list", "--porcelain"]);
return {
stillRegistered: parseWorktreeList(raw).some(
Expand Down
25 changes: 25 additions & 0 deletions packages/host-service/src/workers/worker-task-protocol.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,31 @@ export function isWorkerShutdownRequestMessage(
);
}

/** Worker → main thread: the handler has entered a named phase.
*
* The task budget is enforced by a timer in the parent, which otherwise has
* no idea which step of a multi-step handler was in flight when it expired —
* and the worker is retired straight afterwards, so there is nothing left to
* ask. This carries that one fact out ahead of the hang. Best-effort: a phase
* that arrives after the timeout is ignored. */
export interface WorkerTaskPhaseMessage {
kind: "phase";
taskId: string;
phase: string;
}

export function isWorkerTaskPhaseMessage(
message: unknown,
): message is WorkerTaskPhaseMessage {
if (!message || typeof message !== "object") return false;
const candidate = message as Partial<WorkerTaskPhaseMessage>;
return (
candidate.kind === "phase" &&
typeof candidate.taskId === "string" &&
typeof candidate.phase === "string"
);
}

export type WorkerTaskResponseMessage =
| {
kind: "result";
Expand Down
34 changes: 34 additions & 0 deletions packages/host-service/test/workspace-cleanup.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -568,6 +568,40 @@ describe("workspaceCleanup.destroy cleanup ordering", () => {
);
});

test("worktree-removal timeout carries its phase into the reported error", async () => {
// The Sentry event for HOST-SERVICE-17 / -47 is this TRPCError, and its
// message is the pool's timeout text verbatim — so a phase named there
// is a phase named in the report. Without this link the label would
// stop at the pool and never reach anyone triaging.
const poolTimeout = new WorkerTaskError(
'[host-worker] Task "git/removeWorktree" timed out after 120000ms in phase "worktree-remove"',
);
const ctx = makeCtx({
workspace: {
id: "ws-1",
projectId: "p-1",
worktreePath: "/branch/wt",
branch: "feature",
},
project: { id: "p-1", repoPath: "/repo" },
removeWorktree: () => Promise.reject(poolTimeout),
});
const caller = workspaceCleanupRouter.createCaller(ctx);

const error = await caller
.destroy({ workspaceId: "ws-1", deleteBranch: false, force: true })
.then(() => null)
.catch((err: Error) => err);

// Built from the pool error rather than restated, so the two halves
// cannot drift; the pool's exact wording is pinned by the phase tests
// in src/workers/host-worker-pool.test.ts.
expect(error?.message).toBe(
`Failed to verify worktree removal at /branch/wt: ${poolTimeout.message}`,
);
expect(error?.message).toContain('in phase "worktree-remove"');
});

test("preflight pool timeout fails closed instead of skipping the dirty check", async () => {
const ctx = makeCtx({
workspace: {
Expand Down
Loading