mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-24 03:15:46 -06:00
feat(node-host): supervise durable worker launches (#122829)
* feat(node-host): add worker launch supervision * fix(node-host): harden worker lifecycle ownership * fix(node-host): harden worker execution boundary * fix(node-host): preserve worker trust settings * chore(plugin-sdk): refresh worker lifecycle baselines * docs(plan): track runner implementation slices * test(node-host): await runtime shutdown owner
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
93f5e0f1f6
commit
6b9bea84f0
@@ -0,0 +1,692 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import { stableStringify } from "@openclaw/normalization-core";
|
||||
import { resolveStateDir } from "../config/paths.js";
|
||||
import { formatErrorMessage } from "../infra/errors.js";
|
||||
import { isPathInside } from "../infra/path-guards.js";
|
||||
import { redactToolPayloadText } from "../logging/redact.js";
|
||||
import {
|
||||
redactRegisteredSecretValues,
|
||||
registerSecretValueForRedaction,
|
||||
} from "../logging/secret-redaction-registry.js";
|
||||
import {
|
||||
appendCapturedOutput,
|
||||
createCapturedOutputBuffers,
|
||||
finalizeCapturedOutput,
|
||||
} from "../process/exec-output.js";
|
||||
import { signalProcessTree } from "../process/kill-tree.js";
|
||||
import { createChildAdapter } from "../process/supervisor/adapters/child.js";
|
||||
import { truncateUtf8Suffix } from "../utils/utf8-truncate.js";
|
||||
import {
|
||||
parseWorkerLaunchDescriptor,
|
||||
type WorkerLaunchDescriptor,
|
||||
} from "../worker/launch-descriptor.js";
|
||||
import { snapshotNodeWorkerEnv } from "./node-worker-environment.js";
|
||||
import {
|
||||
NodeWorkerLaunchStore,
|
||||
type NodeWorkerLaunchReceipt,
|
||||
type NodeWorkerTerminalState,
|
||||
} from "./node-worker-launch-store.js";
|
||||
import {
|
||||
inspectNodeWorkerProcessIdentity,
|
||||
requireNodeWorkerProcessIdentity,
|
||||
type NodeWorkerProcessIdentity,
|
||||
} from "./node-worker-process-identity.js";
|
||||
|
||||
const STDOUT_MAX_BYTES = 64 * 1024;
|
||||
const STDERR_MAX_BYTES = 4 * 1024;
|
||||
const STOP_GRACE_MS = 1_000;
|
||||
const FORCE_STOP_WAIT_MS = 4_000;
|
||||
const RECOVERY_POLL_MS = 25;
|
||||
const GATEWAY_NAMESPACE_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/u;
|
||||
const BUNDLE_HASH_PATTERN = /^[a-f0-9]{64}$/u;
|
||||
|
||||
type NodeWorkerLaunchInput = {
|
||||
launchId: string;
|
||||
gatewayNamespace: string;
|
||||
bundleHash: string;
|
||||
placementGeneration: number;
|
||||
descriptor: WorkerLaunchDescriptor;
|
||||
};
|
||||
|
||||
type ChildAdapter = Awaited<ReturnType<typeof createChildAdapter>>;
|
||||
type StopState = Extract<NodeWorkerTerminalState, "cancelled" | "interrupted">;
|
||||
type OwnedTreeState = "live" | "dead" | "unknown";
|
||||
type CredentialScrubber = {
|
||||
maxRepresentationBytes: number;
|
||||
scrub: (text: string) => string;
|
||||
};
|
||||
type ActiveBase = {
|
||||
launchId: string;
|
||||
planHash: string;
|
||||
supervisor: NodeWorkerProcessIdentity;
|
||||
worker: NodeWorkerProcessIdentity;
|
||||
};
|
||||
type RunningChild = ActiveBase & {
|
||||
state: "running";
|
||||
adapter: ChildAdapter;
|
||||
done: Promise<void>;
|
||||
journalReady: Promise<void>;
|
||||
releaseJournal: () => void;
|
||||
scrubber: CredentialScrubber;
|
||||
stopState?: StopState;
|
||||
};
|
||||
type TerminalOutcome = Readonly<{
|
||||
state: NodeWorkerTerminalState;
|
||||
resultJson?: string;
|
||||
errorText?: string;
|
||||
}>;
|
||||
type ObservedTerminal = ActiveBase & {
|
||||
state: "observed";
|
||||
outcome: TerminalOutcome;
|
||||
persistenceError?: unknown;
|
||||
};
|
||||
type ActiveOwnership = RunningChild | ObservedTerminal;
|
||||
|
||||
function nodeWorkerPlanHash(params: {
|
||||
bundleHash: string;
|
||||
descriptor: WorkerLaunchDescriptor;
|
||||
gatewayNamespace: string;
|
||||
placementGeneration: number;
|
||||
}): string {
|
||||
return createHash("sha256").update(stableStringify(params)).digest("hex");
|
||||
}
|
||||
|
||||
function resolveWorkerEntry(params: {
|
||||
bundleRoot: string;
|
||||
bundleHash: string;
|
||||
gatewayNamespace: string;
|
||||
}): string {
|
||||
const root = fs.realpathSync.native(params.bundleRoot);
|
||||
const bundle = fs.realpathSync.native(
|
||||
path.join(root, params.gatewayNamespace, "bundles", params.bundleHash),
|
||||
);
|
||||
if (!isPathInside(root, bundle)) {
|
||||
throw new Error("node worker bundle resolves outside its configured root");
|
||||
}
|
||||
const entry = fs.realpathSync.native(path.join(bundle, "openclaw.mjs"));
|
||||
if (!isPathInside(bundle, entry) || !fs.statSync(entry).isFile()) {
|
||||
throw new Error("node worker entry must be a regular file inside its bundle");
|
||||
}
|
||||
return entry;
|
||||
}
|
||||
|
||||
function createCredentialScrubber(credential: string): CredentialScrubber {
|
||||
const representations = new Set([
|
||||
credential,
|
||||
encodeURIComponent(credential),
|
||||
JSON.stringify(credential).slice(1, -1),
|
||||
]);
|
||||
const ordered = [...representations].toSorted((left, right) => right.length - left.length);
|
||||
return {
|
||||
maxRepresentationBytes: Math.max(
|
||||
...ordered.map((representation) => Buffer.byteLength(representation, "utf8")),
|
||||
),
|
||||
scrub: (text) => {
|
||||
let scrubbed = text;
|
||||
for (const representation of ordered) {
|
||||
scrubbed = scrubbed.replaceAll(representation, "[REDACTED]");
|
||||
}
|
||||
return scrubbed;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function redactLaunchText(value: string, scrubCredential: (text: string) => string): string {
|
||||
const launchRedacted = scrubCredential(value);
|
||||
const exactRedacted = redactRegisteredSecretValues(launchRedacted, () => "[REDACTED]");
|
||||
return redactToolPayloadText(exactRedacted);
|
||||
}
|
||||
|
||||
function sanitizeDiagnostic(
|
||||
value: string,
|
||||
fallback: string,
|
||||
scrubCredential: (text: string) => string,
|
||||
): string {
|
||||
const oneLine = redactLaunchText(value, scrubCredential).replace(/\s+/gu, " ").trim();
|
||||
return truncateUtf8Suffix(oneLine || fallback, STDERR_MAX_BYTES);
|
||||
}
|
||||
|
||||
function successfulResult(
|
||||
stdout: ReturnType<typeof createCapturedOutputBuffers>,
|
||||
scrubCredential: (text: string) => string,
|
||||
): string {
|
||||
if (stdout.truncatedBytes > 0) {
|
||||
throw new Error(`worker stdout exceeded ${STDOUT_MAX_BYTES} bytes`);
|
||||
}
|
||||
const raw = finalizeCapturedOutput(stdout, "head", true).toString("utf8").trim();
|
||||
const redacted = redactLaunchText(raw, scrubCredential);
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(redacted) as unknown;
|
||||
} catch (error) {
|
||||
throw new Error("worker returned invalid JSON output", { cause: error });
|
||||
}
|
||||
const result = JSON.stringify(parsed);
|
||||
if (Buffer.byteLength(result, "utf8") > STDOUT_MAX_BYTES) {
|
||||
throw new Error(`worker result exceeded ${STDOUT_MAX_BYTES} bytes`);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function inspectPosixProcessGroup(pid: number): OwnedTreeState {
|
||||
try {
|
||||
process.kill(-pid, 0);
|
||||
return "live";
|
||||
} catch (error) {
|
||||
const code = (error as NodeJS.ErrnoException).code;
|
||||
return code === "ESRCH" ? "dead" : "unknown";
|
||||
}
|
||||
}
|
||||
|
||||
function inspectOwnedWorkerTree(worker: NodeWorkerProcessIdentity): OwnedTreeState {
|
||||
const root = inspectNodeWorkerProcessIdentity(worker);
|
||||
if (root === "reused") {
|
||||
return "dead";
|
||||
}
|
||||
if (root === "live") {
|
||||
return "live";
|
||||
}
|
||||
if (root === "unknown") {
|
||||
return "unknown";
|
||||
}
|
||||
return process.platform === "win32" ? "dead" : inspectPosixProcessGroup(worker.pid);
|
||||
}
|
||||
|
||||
async function signalOwnedWorkerTree(
|
||||
worker: NodeWorkerProcessIdentity,
|
||||
signal: "SIGTERM" | "SIGKILL",
|
||||
): Promise<void> {
|
||||
const root = inspectNodeWorkerProcessIdentity(worker);
|
||||
if (root === "reused" || root === "unknown") {
|
||||
return;
|
||||
}
|
||||
await new Promise<void>((resolve) => {
|
||||
signalProcessTree(worker.pid, signal, { detached: true, onComplete: resolve });
|
||||
});
|
||||
}
|
||||
|
||||
async function waitForOwnedWorkerTreeDeath(
|
||||
worker: NodeWorkerProcessIdentity,
|
||||
timeoutMs: number,
|
||||
): Promise<OwnedTreeState> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
let state = inspectOwnedWorkerTree(worker);
|
||||
while (state === "live" && Date.now() < deadline) {
|
||||
await delay(RECOVERY_POLL_MS);
|
||||
state = inspectOwnedWorkerTree(worker);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
/** Owns worker process groups, lifetime gates, and the durable node-host launch journal. */
|
||||
class NodeWorkerSupervisor {
|
||||
private readonly active = new Map<string, ActiveOwnership>();
|
||||
private readonly starting = new Map<string, Promise<NodeWorkerLaunchReceipt>>();
|
||||
private readonly bundleRoot: string;
|
||||
private readonly store: NodeWorkerLaunchStore;
|
||||
private readonly workerEnv: NodeJS.ProcessEnv;
|
||||
private supervisorIdentity?: NodeWorkerProcessIdentity;
|
||||
private closed = false;
|
||||
private closePromise?: Promise<void>;
|
||||
|
||||
constructor(options: { bundleRoot?: string; env?: NodeJS.ProcessEnv } = {}) {
|
||||
const env = options.env ?? process.env;
|
||||
this.bundleRoot = path.resolve(
|
||||
options.bundleRoot ?? path.join(resolveStateDir(env), "node-host"),
|
||||
);
|
||||
this.store = new NodeWorkerLaunchStore({ env });
|
||||
this.workerEnv = snapshotNodeWorkerEnv(env);
|
||||
}
|
||||
|
||||
private requireSupervisorIdentity(): NodeWorkerProcessIdentity {
|
||||
return (this.supervisorIdentity ??= requireNodeWorkerProcessIdentity(process.pid));
|
||||
}
|
||||
|
||||
async launch(input: NodeWorkerLaunchInput): Promise<NodeWorkerLaunchReceipt> {
|
||||
if (!GATEWAY_NAMESPACE_PATTERN.test(input.gatewayNamespace)) {
|
||||
throw new Error("gateway namespace must be a safe bounded path component");
|
||||
}
|
||||
if (!BUNDLE_HASH_PATTERN.test(input.bundleHash)) {
|
||||
throw new Error("node worker bundle hash must be 64 lowercase hexadecimal characters");
|
||||
}
|
||||
if (!Number.isSafeInteger(input.placementGeneration) || input.placementGeneration < 0) {
|
||||
throw new Error("node worker placement generation must be a non-negative safe integer");
|
||||
}
|
||||
const descriptor = parseWorkerLaunchDescriptor(structuredClone(input.descriptor));
|
||||
if (descriptor.admission.handshake.bundleHash !== input.bundleHash) {
|
||||
throw new Error("node worker descriptor bundle hash does not match the launch bundle");
|
||||
}
|
||||
const planHash = nodeWorkerPlanHash({
|
||||
bundleHash: input.bundleHash,
|
||||
descriptor,
|
||||
gatewayNamespace: input.gatewayNamespace,
|
||||
placementGeneration: input.placementGeneration,
|
||||
});
|
||||
const local = this.active.get(input.launchId);
|
||||
if (local) {
|
||||
if (local.planHash !== planHash) {
|
||||
throw new Error(`node worker launch ${input.launchId} was replayed with a different plan`);
|
||||
}
|
||||
if (local.state === "observed") {
|
||||
return this.reconcileActiveTerminal(local);
|
||||
}
|
||||
const receipt = this.store.get(input.launchId);
|
||||
if (receipt) {
|
||||
return receipt;
|
||||
}
|
||||
}
|
||||
if (this.closed) {
|
||||
throw new Error("node worker supervisor is closed");
|
||||
}
|
||||
const supervisor = this.requireSupervisorIdentity();
|
||||
const claim = this.store.claim(
|
||||
{
|
||||
launchId: input.launchId,
|
||||
planHash,
|
||||
gatewayNamespace: input.gatewayNamespace,
|
||||
environmentId: descriptor.admission.environmentId,
|
||||
sessionId: descriptor.admission.sessionId,
|
||||
ownerEpoch: descriptor.admission.ownerEpoch,
|
||||
placementGeneration: input.placementGeneration,
|
||||
runId: descriptor.assignment.runId,
|
||||
},
|
||||
supervisor,
|
||||
);
|
||||
if (claim.action === "recover") {
|
||||
return await this.recoverRunning(claim.receipt);
|
||||
}
|
||||
if (claim.action === "replay") {
|
||||
const replay = this.active.get(input.launchId);
|
||||
if (replay?.planHash === planHash && replay.state === "observed") {
|
||||
return this.reconcileActiveTerminal(replay);
|
||||
}
|
||||
const startup = this.starting.get(input.launchId);
|
||||
return startup && claim.receipt.state === "pending" ? await startup : claim.receipt;
|
||||
}
|
||||
const startup = this.startClaimed({ input, descriptor, planHash, supervisor });
|
||||
this.starting.set(input.launchId, startup);
|
||||
try {
|
||||
return await startup;
|
||||
} finally {
|
||||
if (this.starting.get(input.launchId) === startup) {
|
||||
this.starting.delete(input.launchId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async status(launchId: string): Promise<NodeWorkerLaunchReceipt | undefined> {
|
||||
const active = this.active.get(launchId);
|
||||
if (active?.state === "observed") {
|
||||
return this.reconcileActiveTerminal(active);
|
||||
}
|
||||
return this.store.get(launchId);
|
||||
}
|
||||
|
||||
async cancel(launchId: string): Promise<NodeWorkerLaunchReceipt | undefined> {
|
||||
const active = this.active.get(launchId);
|
||||
if (active) {
|
||||
if (active.state === "running") {
|
||||
await this.stopChild(active, "cancelled");
|
||||
}
|
||||
const observed = this.active.get(launchId);
|
||||
if (observed?.state === "observed") {
|
||||
return this.reconcileActiveTerminal(observed);
|
||||
}
|
||||
return this.store.get(launchId);
|
||||
}
|
||||
const startup = this.starting.get(launchId);
|
||||
const receipt = this.store.get(launchId);
|
||||
if (!receipt || receipt.state === "completed" || receipt.state === "failed") {
|
||||
return receipt;
|
||||
}
|
||||
if (receipt.state === "interrupted" || receipt.state === "cancelled") {
|
||||
return receipt;
|
||||
}
|
||||
if (!startup || receipt.state !== "pending" || receipt.supervisor.pid !== process.pid) {
|
||||
return receipt;
|
||||
}
|
||||
const cancelled = this.store.finish({
|
||||
launchId,
|
||||
planHash: receipt.planHash,
|
||||
supervisor: this.requireSupervisorIdentity(),
|
||||
worker: null,
|
||||
state: "cancelled",
|
||||
errorText: "node worker launch cancelled",
|
||||
});
|
||||
await startup;
|
||||
return this.store.get(launchId) ?? cancelled;
|
||||
}
|
||||
|
||||
close(): Promise<void> {
|
||||
if (this.closePromise) {
|
||||
return this.closePromise;
|
||||
}
|
||||
this.closed = true;
|
||||
const operation = (async () => {
|
||||
await Promise.allSettled(this.starting.values());
|
||||
await Promise.all(
|
||||
[...this.active.values()]
|
||||
.filter((active): active is RunningChild => active.state === "running")
|
||||
.map(async (active) => await this.stopChild(active, "interrupted")),
|
||||
);
|
||||
const errors: unknown[] = [];
|
||||
for (const active of this.active.values()) {
|
||||
if (active.state !== "observed") {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
this.reconcileActiveTerminal(active);
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
}
|
||||
if (errors.length === 1) {
|
||||
throw errors[0];
|
||||
}
|
||||
if (errors.length > 1) {
|
||||
throw new AggregateError(errors, "node worker terminal reconciliation failed");
|
||||
}
|
||||
})();
|
||||
const closePromise = operation.finally(() => {
|
||||
if (this.closePromise === closePromise) {
|
||||
this.closePromise = undefined;
|
||||
}
|
||||
});
|
||||
this.closePromise = closePromise;
|
||||
return closePromise;
|
||||
}
|
||||
|
||||
private reconcileActiveTerminal(active: ObservedTerminal): NodeWorkerLaunchReceipt {
|
||||
try {
|
||||
const receipt = this.store.finish({
|
||||
launchId: active.launchId,
|
||||
planHash: active.planHash,
|
||||
supervisor: active.supervisor,
|
||||
worker: active.worker,
|
||||
...active.outcome,
|
||||
});
|
||||
if (receipt.state === "pending" || receipt.state === "running") {
|
||||
throw new Error(`node worker launch ${active.launchId} terminal state was not persisted`);
|
||||
}
|
||||
if (this.active.get(active.launchId) === active) {
|
||||
this.active.delete(active.launchId);
|
||||
}
|
||||
return receipt;
|
||||
} catch (error) {
|
||||
active.persistenceError = error;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async recoverRunning(receipt: NodeWorkerLaunchReceipt): Promise<NodeWorkerLaunchReceipt> {
|
||||
if (receipt.state !== "running" || !receipt.worker) {
|
||||
return receipt;
|
||||
}
|
||||
const previousSupervisor = inspectNodeWorkerProcessIdentity(receipt.supervisor);
|
||||
if (previousSupervisor !== "dead" && previousSupervisor !== "reused") {
|
||||
return this.store.get(receipt.launchId) ?? receipt;
|
||||
}
|
||||
let workerState = inspectOwnedWorkerTree(receipt.worker);
|
||||
if (workerState === "unknown") {
|
||||
return this.store.get(receipt.launchId) ?? receipt;
|
||||
}
|
||||
if (workerState === "live") {
|
||||
await signalOwnedWorkerTree(receipt.worker, "SIGTERM");
|
||||
workerState = await waitForOwnedWorkerTreeDeath(receipt.worker, STOP_GRACE_MS);
|
||||
}
|
||||
if (workerState === "live") {
|
||||
await signalOwnedWorkerTree(receipt.worker, "SIGKILL");
|
||||
workerState = await waitForOwnedWorkerTreeDeath(receipt.worker, FORCE_STOP_WAIT_MS);
|
||||
}
|
||||
if (workerState !== "dead") {
|
||||
return this.store.get(receipt.launchId) ?? receipt;
|
||||
}
|
||||
return this.store.finish({
|
||||
launchId: receipt.launchId,
|
||||
planHash: receipt.planHash,
|
||||
supervisor: receipt.supervisor,
|
||||
worker: receipt.worker,
|
||||
state: "interrupted",
|
||||
errorText: "node host stopped before the worker launch completed",
|
||||
});
|
||||
}
|
||||
|
||||
private async startClaimed(params: {
|
||||
input: NodeWorkerLaunchInput;
|
||||
descriptor: WorkerLaunchDescriptor;
|
||||
planHash: string;
|
||||
supervisor: NodeWorkerProcessIdentity;
|
||||
}): Promise<NodeWorkerLaunchReceipt> {
|
||||
const credential = params.descriptor.admission.credential;
|
||||
const scrubber = createCredentialScrubber(credential);
|
||||
registerSecretValueForRedaction(credential);
|
||||
let adapter: ChildAdapter;
|
||||
try {
|
||||
const entry = resolveWorkerEntry({
|
||||
bundleRoot: this.bundleRoot,
|
||||
bundleHash: params.input.bundleHash,
|
||||
gatewayNamespace: params.input.gatewayNamespace,
|
||||
});
|
||||
adapter = await createChildAdapter({
|
||||
argv: [process.execPath, entry, "worker", "--internal-worker-ipc"],
|
||||
env: this.workerEnv,
|
||||
exactEnv: true,
|
||||
ownedWorker: true,
|
||||
input: JSON.stringify(params.descriptor),
|
||||
});
|
||||
} catch (error) {
|
||||
return this.store.finish({
|
||||
launchId: params.input.launchId,
|
||||
planHash: params.planHash,
|
||||
supervisor: params.supervisor,
|
||||
worker: null,
|
||||
state: "failed",
|
||||
errorText: sanitizeDiagnostic(
|
||||
formatErrorMessage(error),
|
||||
"node worker spawn failed",
|
||||
scrubber.scrub,
|
||||
),
|
||||
});
|
||||
}
|
||||
if (!adapter.pid) {
|
||||
adapter.kill("SIGKILL");
|
||||
adapter.dispose();
|
||||
return this.store.finish({
|
||||
launchId: params.input.launchId,
|
||||
planHash: params.planHash,
|
||||
supervisor: params.supervisor,
|
||||
worker: null,
|
||||
state: "failed",
|
||||
errorText: "node worker spawn did not return a process id",
|
||||
});
|
||||
}
|
||||
let worker: NodeWorkerProcessIdentity;
|
||||
try {
|
||||
worker = requireNodeWorkerProcessIdentity(adapter.pid);
|
||||
} catch (error) {
|
||||
adapter.kill("SIGKILL");
|
||||
await adapter.wait().catch(() => undefined);
|
||||
adapter.dispose();
|
||||
return this.store.finish({
|
||||
launchId: params.input.launchId,
|
||||
planHash: params.planHash,
|
||||
supervisor: params.supervisor,
|
||||
worker: null,
|
||||
state: "failed",
|
||||
errorText: sanitizeDiagnostic(
|
||||
formatErrorMessage(error),
|
||||
"node worker process identity unavailable",
|
||||
scrubber.scrub,
|
||||
),
|
||||
});
|
||||
}
|
||||
let journalReleased = false;
|
||||
let releaseJournalPromise!: () => void;
|
||||
const journalReady = new Promise<void>((resolve) => {
|
||||
releaseJournalPromise = resolve;
|
||||
});
|
||||
const releaseJournal = () => {
|
||||
if (!journalReleased) {
|
||||
journalReleased = true;
|
||||
releaseJournalPromise();
|
||||
}
|
||||
};
|
||||
const active = {
|
||||
state: "running",
|
||||
adapter,
|
||||
journalReady,
|
||||
launchId: params.input.launchId,
|
||||
planHash: params.planHash,
|
||||
releaseJournal,
|
||||
scrubber,
|
||||
supervisor: params.supervisor,
|
||||
worker,
|
||||
} as RunningChild;
|
||||
active.done = this.observeChild(active);
|
||||
this.active.set(active.launchId, active);
|
||||
void active.done.catch(() => undefined);
|
||||
let running: NodeWorkerLaunchReceipt;
|
||||
try {
|
||||
running = this.store.markRunning({
|
||||
launchId: active.launchId,
|
||||
planHash: active.planHash,
|
||||
supervisor: params.supervisor,
|
||||
worker,
|
||||
});
|
||||
} catch (error) {
|
||||
active.releaseJournal();
|
||||
await this.stopChild(active, "interrupted").catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
active.releaseJournal();
|
||||
if (running.state === "cancelled" || running.state === "interrupted") {
|
||||
await this.stopChild(active, running.state);
|
||||
return this.store.get(active.launchId) ?? running;
|
||||
}
|
||||
if (running.state !== "running") {
|
||||
adapter.closeStartGate?.();
|
||||
return running;
|
||||
}
|
||||
if (this.closed) {
|
||||
await this.stopChild(active, "interrupted");
|
||||
return this.store.get(active.launchId) ?? running;
|
||||
}
|
||||
try {
|
||||
await adapter.openStartGate?.();
|
||||
} catch {
|
||||
await this.stopChild(active, "interrupted");
|
||||
return this.store.get(active.launchId) ?? running;
|
||||
}
|
||||
return running;
|
||||
}
|
||||
|
||||
private async observeChild(active: RunningChild): Promise<void> {
|
||||
const stdout = createCapturedOutputBuffers();
|
||||
const stderr = createCapturedOutputBuffers();
|
||||
active.adapter.onStdout((chunk) =>
|
||||
appendCapturedOutput(stdout, chunk, STDOUT_MAX_BYTES, "head"),
|
||||
);
|
||||
active.adapter.onStderr((chunk) =>
|
||||
appendCapturedOutput(
|
||||
stderr,
|
||||
chunk,
|
||||
STDERR_MAX_BYTES + active.scrubber.maxRepresentationBytes,
|
||||
"tail",
|
||||
),
|
||||
);
|
||||
let outcome: TerminalOutcome;
|
||||
try {
|
||||
const exit = await active.adapter.wait();
|
||||
await active.journalReady;
|
||||
if (active.stopState) {
|
||||
outcome = Object.freeze({
|
||||
state: active.stopState,
|
||||
errorText:
|
||||
active.stopState === "cancelled"
|
||||
? "node worker launch cancelled"
|
||||
: "node worker launch interrupted during node-host shutdown",
|
||||
});
|
||||
} else if (exit.code === 0 && exit.signal === null) {
|
||||
try {
|
||||
outcome = Object.freeze({
|
||||
state: "completed",
|
||||
resultJson: successfulResult(stdout, active.scrubber.scrub),
|
||||
});
|
||||
} catch (error) {
|
||||
outcome = Object.freeze({
|
||||
state: "failed",
|
||||
errorText: sanitizeDiagnostic(
|
||||
formatErrorMessage(error),
|
||||
"invalid worker result",
|
||||
active.scrubber.scrub,
|
||||
),
|
||||
});
|
||||
}
|
||||
} else {
|
||||
const detail = finalizeCapturedOutput(stderr, "tail", true).toString("utf8");
|
||||
const exitLabel = exit.signal ? `signal ${exit.signal}` : `exit code ${String(exit.code)}`;
|
||||
outcome = Object.freeze({
|
||||
state: "failed",
|
||||
errorText: sanitizeDiagnostic(
|
||||
`node worker failed with ${exitLabel}${detail ? `: ${detail}` : ""}`,
|
||||
"node worker failed",
|
||||
active.scrubber.scrub,
|
||||
),
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
await active.journalReady;
|
||||
outcome = Object.freeze({
|
||||
state: active.stopState ?? "failed",
|
||||
errorText: sanitizeDiagnostic(
|
||||
formatErrorMessage(error),
|
||||
"node worker wait failed",
|
||||
active.scrubber.scrub,
|
||||
),
|
||||
});
|
||||
} finally {
|
||||
active.adapter.dispose();
|
||||
}
|
||||
const observed: ObservedTerminal = {
|
||||
state: "observed",
|
||||
launchId: active.launchId,
|
||||
planHash: active.planHash,
|
||||
supervisor: active.supervisor,
|
||||
worker: active.worker,
|
||||
outcome,
|
||||
};
|
||||
if (this.active.get(active.launchId) !== active) {
|
||||
return;
|
||||
}
|
||||
this.active.set(active.launchId, observed);
|
||||
try {
|
||||
this.reconcileActiveTerminal(observed);
|
||||
} catch {
|
||||
// The observed outcome stays owned in memory for the next supervisor operation.
|
||||
}
|
||||
}
|
||||
|
||||
private async stopChild(active: RunningChild, state: StopState): Promise<void> {
|
||||
active.stopState ??= state;
|
||||
active.adapter.kill("SIGTERM");
|
||||
const forceKill = setTimeout(() => active.adapter.kill("SIGKILL"), STOP_GRACE_MS);
|
||||
forceKill.unref?.();
|
||||
try {
|
||||
await active.done;
|
||||
} finally {
|
||||
clearTimeout(forceKill);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function createNodeWorkerSupervisor(
|
||||
options: {
|
||||
bundleRoot?: string;
|
||||
env?: NodeJS.ProcessEnv;
|
||||
} = {},
|
||||
): NodeWorkerSupervisor {
|
||||
return new NodeWorkerSupervisor(options);
|
||||
}
|
||||
Reference in New Issue
Block a user