From 59a59e1345e23a9f3d052edcc7e2e8df4e60323d Mon Sep 17 00:00:00 2001 From: Miya Date: Mon, 26 Oct 2026 01:33:27 +0100 Subject: [PATCH 1/2] perf(sdk): start the authored Node probe while the flow loads The standalone CLI imported the flow and hashed its Surface package, and only then spawned the Node probe and SHA-256 hashed the ~110 MB executable for the first authority check before root admission. Start that probe and hash before the flow load, without blocking it. It only fills #681's memo under the same rules (same candidate by path and realpath, unchanged stat fingerprint, a file not rewritten while hashed), and the first authority check awaits it and still validates the memo. A failed or unusable prewarm leaves the probe, and its refusal, to that check. The async probe keeps the synchronous probe's contract: exit 0, 4096 output bytes, 10 s. The authority code moves to authored-node-authority.ts; the runner was at the 500-line limit. Co-Authored-By: Claude Opus 5.5 --- packages/sdk/src/authored-node-authority.ts | 158 ++++++++++++++++++ packages/sdk/src/authored-node-runner.ts | 106 +++--------- packages/sdk/src/authored-root.ts | 4 +- packages/sdk/src/cli/direct-run.ts | 4 + .../sdk/tests/authored-node-prewarm.test.ts | 93 +++++++++++ .../sdk/tests/authored-node-runtime.test.ts | 23 +++ 6 files changed, 303 insertions(+), 85 deletions(-) create mode 100644 packages/sdk/src/authored-node-authority.ts create mode 100644 packages/sdk/tests/authored-node-prewarm.test.ts diff --git a/packages/sdk/src/authored-node-authority.ts b/packages/sdk/src/authored-node-authority.ts new file mode 100644 index 00000000..8a2353da --- /dev/null +++ b/packages/sdk/src/authored-node-authority.ts @@ -0,0 +1,158 @@ +import { spawn, spawnSync } from 'node:child_process'; +import { createHash } from 'node:crypto'; +import { accessSync, constants as fsConstants, createReadStream, readFileSync, realpathSync, statSync } from 'node:fs'; +import { delimiter, isAbsolute, join } from 'node:path'; +import { AuthoredFlowExecutionError } from './authored-flow-error.js'; + +export const authoredNodeRefusal = (): AuthoredFlowExecutionError => new AuthoredFlowExecutionError( + 'unsupported_promise_lifecycle', + 'authored execution requires the embedded Node runner and Node >=22.14 on PATH (or an absolute FLOWS_AUTHORED_NODE); no workflow body was executed', +); +export interface NodeAuthority { path: string; version: string; executableSha256: string } + +const PROBE_ARGS = ['--input-type=module', '--eval', ` +import {createHook} from 'node:async_hooks'; +import {stripTypeScriptTypes,register} from 'node:module'; +const [major,minor]=process.versions.node.split('.').map(Number); +let init=false,resolve=false; +const h=createHook({init(_id,type){if(type==='PROMISE')init=true},promiseResolve(){resolve=true}}).enable(); +new Promise(r=>r());h.disable(); +if(process.versions.bun || major<22 || (major===22 && minor<14) || !init || !resolve || typeof stripTypeScriptTypes!=='function' || typeof register!=='function') process.exit(2); +process.stdout.write(JSON.stringify({path:process.execPath,version:process.versions.node})); +`]; +const PROBE_TIMEOUT_MS = 10_000, PROBE_MAX_BYTES = 4096; + +/** + * One authored run asks for the Node authority before root admission and + * again before spawning the body. Each probe spawns Node and hashes its whole + * executable (~100 MB), which a cold sandbox pays on the startup path twice. + * Reuse the previous answer only while the same candidate is selected (the + * override or the first `node` on PATH, by path and by realpath, so a + * retargeted symlink re-probes) and the executable it resolved is unchanged. + * The spawned child still hashes the executable it actually runs and the + * parent compares that report with this authority, so a binary replaced + * after the probe is refused, never trusted. + */ +let probedAuthority: { selection: string; fingerprint: string; authority: NodeAuthority } | undefined; +function executableFingerprint(path: string): string { + const stat = statSync(path, { bigint: true }); + return [stat.dev, stat.ino, stat.size, stat.mtimeNs, stat.ctimeNs].join(':'); +} +/** The candidate `spawnSync(override ?? 'node')` would run, by path and realpath. */ +function selectedCandidate(override: string | undefined): string | undefined { + const directories = (process.env['PATH'] ?? '').split(delimiter); + // A relative PATH entry resolves against the working directory; do not + // guess which candidate the spawn would pick, just probe every time. + if (override === undefined && directories.some(directory => !isAbsolute(directory))) return undefined; + const candidates = override !== undefined ? [override] : directories.map(directory => join(directory, 'node')); + for (const candidate of candidates) { + try { + if (!statSync(candidate).isFile()) continue; + accessSync(candidate, fsConstants.X_OK); + return `${candidate}\0${realpathSync(candidate)}`; + } catch { /* Not this candidate. */ } + } + return undefined; +} + +type Probe = { readonly command: string; readonly selection: string | undefined }; +/** The remembered authority while it is still valid, otherwise what to probe. */ +function rememberedOrProbe(): NodeAuthority | Probe { + const override = process.env['FLOWS_AUTHORED_NODE']; + if (override !== undefined && !isAbsolute(override)) throw authoredNodeRefusal(); + const selection = selectedCandidate(override); + if (selection !== undefined && probedAuthority?.selection === selection) { + try { + if (executableFingerprint(probedAuthority.authority.path) === probedAuthority.fingerprint) { + return probedAuthority.authority; + } + } catch { /* A vanished executable is re-probed, and refused there. */ } + } + probedAuthority = undefined; + return { command: override ?? 'node', selection }; +} + +/** The probed Node's real path, before its bytes are hashed. */ +function probedNode(stdout: string): { path: string; version: string; fingerprint: string } { + const value = JSON.parse(stdout) as { path: string; version: string }; + if (!isAbsolute(value.path) || !/^\d+\.\d+\.\d+$/.test(value.version)) throw authoredNodeRefusal(); + const [major, minor] = value.version.split('.').map(Number); + if (major! < 22 || (major === 22 && minor! < 14)) throw authoredNodeRefusal(); + const path = realpathSync(value.path); + return { path, version: value.version, fingerprint: executableFingerprint(path) }; +} + +function remember(selection: string | undefined, fingerprint: string, authority: NodeAuthority): NodeAuthority { + // Reuse only an answer for a candidate that could be identified, and not + // one whose file was rewritten while it was being hashed. + if (selection !== undefined && executableFingerprint(authority.path) === fingerprint) { + probedAuthority = { selection, fingerprint, authority }; + } + return authority; +} + +export function nodeAuthority(): NodeAuthority { + const request = rememberedOrProbe(); + if (!('command' in request)) return request; + const probe = spawnSync(request.command, PROBE_ARGS, { + encoding: 'utf8', timeout: PROBE_TIMEOUT_MS, maxBuffer: PROBE_MAX_BYTES, env: process.env, + }); + if (probe.error || probe.status !== 0) throw authoredNodeRefusal(); + try { + const { path, version, fingerprint } = probedNode(probe.stdout); + const executableSha256 = createHash('sha256').update(readFileSync(path)).digest('hex'); + return remember(request.selection, fingerprint, { path, version, executableSha256 }); + } catch { throw authoredNodeRefusal(); } +} + +/** + * The CLI loads the flow (importing it and hashing its Surface package) + * before it first asks for the Node authority. A prewarm runs the same probe + * and hash concurrently with that load, without blocking it, and only fills + * the memo above under the same rules. It never answers by itself: + * {@link awaitNodeAuthority} still validates the memo, and a failed or + * unusable prewarm leaves the probe and its refusal to that call. + */ +let prewarm: Promise | undefined; +export function prewarmNodeAuthority(): void { + prewarm ??= probeInBackground().catch(() => { /* A miss: the next call probes, and refuses, itself. */ }); +} + +export async function awaitNodeAuthority(): Promise { + await prewarm; + return nodeAuthority(); +} + +async function probeInBackground(): Promise { + const request = rememberedOrProbe(); + // Nothing a later call could reuse: an unidentifiable candidate is never remembered. + if (!('command' in request) || request.selection === undefined) return; + const { path, version, fingerprint } = probedNode(await probeOutput(request.command)); + remember(request.selection, fingerprint, { path, version, executableSha256: await fileSha256(path) }); +} + +/** `spawnSync`'s contract above, without blocking: exit 0, bounded output and time. */ +function probeOutput(command: string): Promise { + return new Promise((resolve, reject) => { + const child = spawn(command, PROBE_ARGS, { env: process.env, stdio: ['ignore', 'pipe', 'ignore'] }); + let stdout = ''; + const fail = (error: Error): void => { child.kill('SIGKILL'); reject(error); }; + const timer = setTimeout(() => fail(new Error('node probe timed out')), PROBE_TIMEOUT_MS); + child.stdout.setEncoding('utf8'); + child.stdout.on('data', (chunk: string) => { + stdout += chunk; + if (Buffer.byteLength(stdout) > PROBE_MAX_BYTES) fail(new Error('node probe output exceeded limit')); + }); + child.once('error', error => { clearTimeout(timer); reject(error); }); + child.once('close', code => { + clearTimeout(timer); + if (code === 0) resolve(stdout); else reject(new Error('node probe failed')); + }); + }); +} + +async function fileSha256(path: string): Promise { + const digest = createHash('sha256'); + for await (const chunk of createReadStream(path, { highWaterMark: 1 << 20 })) digest.update(chunk as Buffer); + return digest.digest('hex'); +} diff --git a/packages/sdk/src/authored-node-runner.ts b/packages/sdk/src/authored-node-runner.ts index 281ab63a..a213767b 100644 --- a/packages/sdk/src/authored-node-runner.ts +++ b/packages/sdk/src/authored-node-runner.ts @@ -1,11 +1,10 @@ import { JournalClient } from './journal-client.js'; import { FLOW_READ_BUDGET_MS } from './journal-read-policy.js'; -import { spawn, spawnSync } from 'node:child_process'; +import { spawn } from 'node:child_process'; import { createHash, createHmac, randomBytes, timingSafeEqual } from 'node:crypto'; import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; -import { accessSync, constants as fsConstants, readFileSync, realpathSync, statSync } from 'node:fs'; import { tmpdir } from 'node:os'; -import { delimiter, isAbsolute, join } from 'node:path'; +import { join } from 'node:path'; import type { Readable } from 'node:stream'; import type { AuthoredRootMetadata } from './authored-root.js'; import type { AuthoredExecutionRuntime, AuthoredFlowExecutionResult, ExecuteAuthoredFlowOptions } from './authored-flow-executor.js'; @@ -19,6 +18,9 @@ import type { } from './failure-kinds.js'; import { HUMAN_WAIT_ID } from './authored-human.js'; import { assertAuthoredPromiseHooks } from './authored-runtime-capability.js'; +import { + authoredNodeRefusal as refusal, awaitNodeAuthority, nodeAuthority, prewarmNodeAuthority, type NodeAuthority, +} from './authored-node-authority.js'; import { settlingCompletion } from './step-settling.js'; import { readAuthoredStepIndex } from './authored-step-index.js'; import { isTransportRetries } from './validate.js'; @@ -30,93 +32,31 @@ export function installAuthoredNodeSource(source: string): void { embeddedSource = source; } const hash = (value: Uint8Array | string): string => createHash('sha256').update(value).digest('hex'); -const refusal = (): AuthoredFlowExecutionError => new AuthoredFlowExecutionError( - 'unsupported_promise_lifecycle', - 'authored execution requires the embedded Node runner and Node >=22.14 on PATH (or an absolute FLOWS_AUTHORED_NODE); no workflow body was executed', -); -interface NodeAuthority { path: string; version: string; executableSha256: string } -/** - * One authored run asks for the Node authority before root admission and - * again before spawning the body. Each probe spawns Node and hashes its whole - * executable (~100 MB), which a cold sandbox pays on the startup path twice. - * Reuse the previous answer only while the same candidate is selected (the - * override or the first `node` on PATH, by path and by realpath, so a - * retargeted symlink re-probes) and the executable it resolved is unchanged. - * The spawned child still hashes the executable it actually runs and the - * parent compares that report with this authority, so a binary replaced - * after the probe is refused, never trusted. - */ -let probedAuthority: { selection: string; fingerprint: string; authority: NodeAuthority } | undefined; -function executableFingerprint(path: string): string { - const stat = statSync(path, { bigint: true }); - return [stat.dev, stat.ino, stat.size, stat.mtimeNs, stat.ctimeNs].join(':'); -} -/** The candidate `spawnSync(override ?? 'node')` would run, by path and realpath. */ -function selectedCandidate(override: string | undefined): string | undefined { - const directories = (process.env['PATH'] ?? '').split(delimiter); - // A relative PATH entry resolves against the working directory; do not - // guess which candidate the spawn would pick, just probe every time. - if (override === undefined && directories.some(directory => !isAbsolute(directory))) return undefined; - const candidates = override !== undefined ? [override] : directories.map(directory => join(directory, 'node')); - for (const candidate of candidates) { - try { - if (!statSync(candidate).isFile()) continue; - accessSync(candidate, fsConstants.X_OK); - return `${candidate}\0${realpathSync(candidate)}`; - } catch { /* Not this candidate. */ } - } - return undefined; -} -function nodeAuthority(): NodeAuthority { - if (embeddedSource === undefined) throw refusal(); - const override = process.env['FLOWS_AUTHORED_NODE']; - if (override !== undefined && !isAbsolute(override)) throw refusal(); - const selection = selectedCandidate(override); - if (selection !== undefined && probedAuthority?.selection === selection) { - try { - if (executableFingerprint(probedAuthority.authority.path) === probedAuthority.fingerprint) { - return probedAuthority.authority; - } - } catch { /* A vanished executable is re-probed, and refused there. */ } - } - probedAuthority = undefined; - const probe = spawnSync(override ?? 'node', ['--input-type=module', '--eval', ` -import {createHook} from 'node:async_hooks'; -import {stripTypeScriptTypes,register} from 'node:module'; -const [major,minor]=process.versions.node.split('.').map(Number); -let init=false,resolve=false; -const h=createHook({init(_id,type){if(type==='PROMISE')init=true},promiseResolve(){resolve=true}}).enable(); -new Promise(r=>r());h.disable(); -if(process.versions.bun || major<22 || (major===22 && minor<14) || !init || !resolve || typeof stripTypeScriptTypes!=='function' || typeof register!=='function') process.exit(2); -process.stdout.write(JSON.stringify({path:process.execPath,version:process.versions.node})); -`], { encoding: 'utf8', timeout: 10_000, maxBuffer: 4096, env: process.env }); - if (probe.error || probe.status !== 0) throw refusal(); - try { - const value = JSON.parse(probe.stdout) as { path: string; version: string }; - if (!isAbsolute(value.path) || !/^\d+\.\d+\.\d+$/.test(value.version)) throw refusal(); - const [major, minor] = value.version.split('.').map(Number); - if (major! < 22 || (major === 22 && minor! < 14)) throw refusal(); - const path = realpathSync(value.path); - const fingerprint = executableFingerprint(path); - const authority = { path, version: value.version, executableSha256: hash(readFileSync(path)) }; - // Reuse only an answer for a candidate that could be identified, and not - // one whose file was rewritten while it was being hashed. - if (selection !== undefined && executableFingerprint(path) === fingerprint) { - probedAuthority = { selection, fingerprint, authority }; - } - return authority; - } catch { throw refusal(); } -} /** Test seam: resolve the authored Node authority exactly as the runner does. */ export function authoredNodeAuthorityForTest(): Readonly { + if (embeddedSource === undefined) throw refusal(); return nodeAuthority(); } +async function embeddedNodeAuthority(): Promise { + if (embeddedSource === undefined) throw refusal(); + return await awaitNodeAuthority(); +} + +/** + * Start the Node probe and executable hash now, so they overlap loading the + * flow. Only the standalone CLI runs authored bodies in Node; elsewhere this + * does nothing. + */ +export function prewarmAuthoredRuntime(): void { + if (process.versions['bun'] !== undefined && embeddedSource !== undefined) prewarmNodeAuthority(); +} + /** Called before root admission/resume; declarative commands never reach it. */ -export function assertAuthoredRuntimeAvailable(): void { +export async function assertAuthoredRuntimeAvailable(): Promise { if (process.versions['bun'] === undefined) assertAuthoredPromiseHooks(); - else nodeAuthority(); + else await embeddedNodeAuthority(); } /** Bun retains the root lease and local workers. Only the authored body uses Node. */ @@ -126,7 +66,7 @@ export async function runAuthoredInNode( rootRunId: string, options: ExecuteAuthoredFlowOptions, ): Promise { - const authority = nodeAuthority(); + const authority = await embeddedNodeAuthority(); const source = embeddedSource!; const runtime: AuthoredExecutionRuntime = { kind: 'node', version: authority.version, diff --git a/packages/sdk/src/authored-root.ts b/packages/sdk/src/authored-root.ts index 983e4c99..2beea14e 100644 --- a/packages/sdk/src/authored-root.ts +++ b/packages/sdk/src/authored-root.ts @@ -87,7 +87,7 @@ export async function executeDurableAuthoredFlow( input: unknown, options: DurableAuthoredOptions, ): Promise { - assertAuthoredRuntimeAvailable(); + await assertAuthoredRuntimeAvailable(); const source = await readFile(loaded.sourcePath); const sources = await Promise.all(loaded.graph.map(async node => Object.freeze({ path: node.path, @@ -195,7 +195,7 @@ export async function resumeDurableAuthoredFlow( options.lifecycle?.onRunReceipt?.({ runId: rootRunId, flow: metadata.flowName }); return completed; } - assertAuthoredRuntimeAvailable(); + await assertAuthoredRuntimeAvailable(); const loaded = await loadPinnedAuthoredSource(metadata); // Source loading itself can alter a process-wide intrinsic. Refuse before // run.resume dispatches a pinned parallel root in that environment. diff --git a/packages/sdk/src/cli/direct-run.ts b/packages/sdk/src/cli/direct-run.ts index d9b1d0df..5e874ced 100644 --- a/packages/sdk/src/cli/direct-run.ts +++ b/packages/sdk/src/cli/direct-run.ts @@ -14,6 +14,7 @@ import { AuthoredFlowExecutionError, } from '../authored-flow-executor.js'; import { executeDurableAuthoredFlow } from '../authored-root.js'; +import { prewarmAuthoredRuntime } from '../authored-node-runner.js'; import { AuthoredHumanParked } from '../authored-flow-error.js'; import { AuthoredFlowLoadError } from '../authored-flow-loader.js'; import { DirectInputError, parseDirectInput } from '../direct-input.js'; @@ -60,6 +61,9 @@ export async function runDirectFlow( }; } + // Loading the flow below imports it and hashes its Surface package; let + // the Node probe and executable hash run meanwhile instead of after it. + prewarmAuthoredRuntime(); // Declared triggers are knowable before any daemon or step is started. // Importing the authored module is unavoidable here — trigger sources // are only observable after `flow(...).on(webhook(...))` has run — but diff --git a/packages/sdk/tests/authored-node-prewarm.test.ts b/packages/sdk/tests/authored-node-prewarm.test.ts new file mode 100644 index 00000000..d77061ea --- /dev/null +++ b/packages/sdk/tests/authored-node-prewarm.test.ts @@ -0,0 +1,93 @@ +import { createHash } from 'node:crypto'; +import { chmodSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; + +// The CLI starts the Node probe and executable hash while it loads the flow. +// The candidate here is a stand-in that answers the probe (optionally slowly, +// wrongly or verbosely) and records each invocation. +let directory: string, probes: string, previousOverride: string | undefined; +type Authority = typeof import('../src/authored-node-authority.js'); +let authority: Authority; + +beforeAll(() => { + directory = mkdtempSync(join(tmpdir(), 'authored-node-prewarm-')); + probes = join(directory, 'probes'); + previousOverride = process.env['FLOWS_AUTHORED_NODE']; +}); + +beforeEach(async () => { + writeFileSync(probes, ''); + // The prewarm and its memo are per process; give every case a fresh module. + vi.resetModules(); + authority = await import('../src/authored-node-authority.js'); +}); + +afterEach(() => { + if (previousOverride === undefined) delete process.env['FLOWS_AUTHORED_NODE']; + else process.env['FLOWS_AUTHORED_NODE'] = previousOverride; +}); + +afterAll(() => rmSync(directory, { recursive: true, force: true })); + +const probeLog = (): string[] => readFileSync(probes, 'utf8').split('\n').filter(Boolean); + +function candidate(name: string, { delaySeconds = 0, exit = 0, padding = 0 } = {}): string { + const path = join(directory, name); + const answer = JSON.stringify({ path, version: '22.14.0' }) + ' '.repeat(padding); + writeFileSync(path, `#!/bin/sh\necho ${name} >> ${JSON.stringify(probes)}\nsleep ${delaySeconds}\n` + + `printf '%s' '${answer}'\nexit ${exit}\n`); + chmodSync(path, 0o755); + process.env['FLOWS_AUTHORED_NODE'] = path; + return path; +} + +describe('authored Node authority prewarm', () => { + it('probes and hashes once; the later call reuses the same authority', async () => { + const path = candidate('node-a'); + authority.prewarmNodeAuthority(); + const resolved = await authority.awaitNodeAuthority(); + expect(probeLog()).toEqual(['node-a']); + expect(resolved).toEqual({ + path: realpathSync(path), version: '22.14.0', + executableSha256: createHash('sha256').update(readFileSync(path)).digest('hex'), + }); + expect(authority.nodeAuthority()).toBe(resolved); + expect(probeLog()).toEqual(['node-a']); + }); + + it('returns at once and leaves the event loop free while the probe runs', async () => { + candidate('node-slow', { delaySeconds: 0.4 }); + const started = performance.now(); + authority.prewarmNodeAuthority(); + expect(performance.now() - started).toBeLessThan(150); + let ticks = 0; + const ticker = setInterval(() => { ticks += 1; }, 20); + try { await authority.awaitNodeAuthority(); } finally { clearInterval(ticker); } + expect(ticks).toBeGreaterThan(5); + expect(probeLog()).toEqual(['node-slow']); + }); + + it('probes again when a different candidate is selected by the time it is needed', async () => { + candidate('node-before'); + authority.prewarmNodeAuthority(); + const after = candidate('node-after'); + expect((await authority.awaitNodeAuthority()).path).toBe(realpathSync(after)); + expect(probeLog()).toEqual(['node-before', 'node-after']); + }); + + it('leaves a failed probe to the call, which probes again and refuses', async () => { + candidate('node-broken', { exit: 2 }); + authority.prewarmNodeAuthority(); + await expect(authority.awaitNodeAuthority()).rejects.toThrow('requires the embedded Node runner'); + expect(probeLog()).toEqual(['node-broken', 'node-broken']); + }); + + it('does not trust a probe answer over the output limit the synchronous probe enforces', async () => { + candidate('node-verbose', { padding: 5000 }); + authority.prewarmNodeAuthority(); + await expect(authority.awaitNodeAuthority()).rejects.toThrow('requires the embedded Node runner'); + expect(probeLog()).toEqual(['node-verbose', 'node-verbose']); + }); +}); diff --git a/packages/sdk/tests/authored-node-runtime.test.ts b/packages/sdk/tests/authored-node-runtime.test.ts index 9c7a6bba..a97e15b1 100644 --- a/packages/sdk/tests/authored-node-runtime.test.ts +++ b/packages/sdk/tests/authored-node-runtime.test.ts @@ -93,6 +93,29 @@ async function entries(directory: string, runId: string) { } describe('Bun 1.4.x standalone → native Node authored lifecycle', () => { + it('starts the Node probe while it imports the flow, and probes once per run', () => { + const f = fixture(''); + const node = join(f.directory, 'node-probe-wrapper'); + // Only the probe runs the selected candidate; the body runs the real path it reports. + writeFileSync(node, `#!/bin/sh\necho probe >> node-probes\nexec ${JSON.stringify(process.execPath)} "$@"\n`); + chmodSync(node, 0o755); + writeFileSync(join(f.directory, 'case.flow.ts'), `import { flow } from '@relayflows/surface'; +import { existsSync, writeFileSync } from 'node:fs'; +if (process.versions.bun) { + // Bun evaluates this while it loads the flow: has the Node probe begun? + const deadline = Date.now() + 5000; + while (!existsSync('node-probes') && Date.now() < deadline) await new Promise(r => setTimeout(r, 10)); + writeFileSync('probe-began-during-load', String(existsSync('node-probes'))); +} +export default flow('runtime-case', async f => { await f.run('touch ran'); f.done('success'); }); +`); + const run = f.invoke(['run', 'case.flow.ts', '--input', '{}'], { FLOWS_AUTHORED_NODE: node }); + expect(run.status, run.stderr + run.stdout).toBe(0); + expect(readFileSync(join(f.directory, 'probe-began-during-load'), 'utf8')).toBe('true'); + expect(readFileSync(join(f.directory, 'node-probes'), 'utf8')).toBe('probe\n'); + expect(existsSync(join(f.directory, 'ran'))).toBe(true); + }, 120_000); + it('runs a certified Promise.all group in parallel through the embedded Node runner', async () => { const f = fixture(''); writeFileSync(join(f.directory, 'case.flow.ts'), `import { flow } from '@relayflows/surface'; From 14a6137612a6dd48777e426df0acca39b15f3717 Mon Sep 17 00:00:00 2001 From: Miya Date: Mon, 26 Oct 2026 02:03:34 +0100 Subject: [PATCH 2/2] fix(sdk): bound probe stderr, stop an unneeded prewarm, hash Node on a worker Review (#684): - The async probe discarded stderr, while spawnSync's maxBuffer bounds each stream. A candidate writing over 4096 bytes to stderr passed the prewarm and was refused without it. Bound stderr the same way. - A run refused before it needs Node returned while the prewarm probe was still running, so a slow Node candidate kept the CLI alive up to the 10 s probe limit. runDirectFlow now stops any prewarm still running when it returns: the probe is killed, its pipes closed, the hash worker ended. Also hash the executable on a worker thread. The chunked main-thread hash shared Bun's main thread with the flow load and slowed it (#684 timings). Co-Authored-By: Claude Opus 5.5 --- packages/sdk/src/authored-node-authority.ts | 111 +++++++++++++----- packages/sdk/src/authored-node-runner.ts | 8 +- packages/sdk/src/cli/direct-run.ts | 16 ++- .../sdk/tests/authored-node-prewarm.test.ts | 31 ++++- .../sdk/tests/authored-node-runtime.test.ts | 21 ++++ 5 files changed, 154 insertions(+), 33 deletions(-) diff --git a/packages/sdk/src/authored-node-authority.ts b/packages/sdk/src/authored-node-authority.ts index 8a2353da..b36aaf6f 100644 --- a/packages/sdk/src/authored-node-authority.ts +++ b/packages/sdk/src/authored-node-authority.ts @@ -1,7 +1,8 @@ import { spawn, spawnSync } from 'node:child_process'; import { createHash } from 'node:crypto'; -import { accessSync, constants as fsConstants, createReadStream, readFileSync, realpathSync, statSync } from 'node:fs'; +import { accessSync, constants as fsConstants, readFileSync, realpathSync, statSync } from 'node:fs'; import { delimiter, isAbsolute, join } from 'node:path'; +import { Worker } from 'node:worker_threads'; import { AuthoredFlowExecutionError } from './authored-flow-error.js'; export const authoredNodeRefusal = (): AuthoredFlowExecutionError => new AuthoredFlowExecutionError( @@ -108,51 +109,101 @@ export function nodeAuthority(): NodeAuthority { /** * The CLI loads the flow (importing it and hashing its Surface package) * before it first asks for the Node authority. A prewarm runs the same probe - * and hash concurrently with that load, without blocking it, and only fills - * the memo above under the same rules. It never answers by itself: - * {@link awaitNodeAuthority} still validates the memo, and a failed or - * unusable prewarm leaves the probe and its refusal to that call. + * concurrently with that load, hashes the executable on a worker thread so + * the load keeps the main thread, and only fills the memo above under the + * same rules. It never answers by itself: {@link awaitNodeAuthority} still + * validates the memo, and a failed, stopped or unusable prewarm leaves the + * probe and its refusal to that call. */ -let prewarm: Promise | undefined; +let prewarm: { readonly done: Promise; readonly stop: AbortController } | undefined; export function prewarmNodeAuthority(): void { - prewarm ??= probeInBackground().catch(() => { /* A miss: the next call probes, and refuses, itself. */ }); + if (prewarm !== undefined) return; + const stop = new AbortController(); + prewarm = { stop, done: probeInBackground(stop.signal).catch(() => { /* A miss: the next call probes, and refuses, itself. */ }) }; +} + +/** Stop a prewarm that is still running, so a run that ends early does not wait for it. */ +export async function stopNodeAuthorityPrewarm(): Promise { + prewarm?.stop.abort(); + await prewarm?.done; } export async function awaitNodeAuthority(): Promise { - await prewarm; + await prewarm?.done; return nodeAuthority(); } -async function probeInBackground(): Promise { +async function probeInBackground(signal: AbortSignal): Promise { const request = rememberedOrProbe(); // Nothing a later call could reuse: an unidentifiable candidate is never remembered. if (!('command' in request) || request.selection === undefined) return; - const { path, version, fingerprint } = probedNode(await probeOutput(request.command)); - remember(request.selection, fingerprint, { path, version, executableSha256: await fileSha256(path) }); + const { path, version, fingerprint } = probedNode(await probeOutput(request.command, signal)); + remember(request.selection, fingerprint, { path, version, executableSha256: await fileSha256(path, signal) }); } -/** `spawnSync`'s contract above, without blocking: exit 0, bounded output and time. */ -function probeOutput(command: string): Promise { +/** + * `spawnSync`'s contract above, without blocking: exit 0 within the time + * limit, and at most PROBE_MAX_BYTES on stdout and on stderr (spawnSync's + * maxBuffer bounds each stream). + */ +function probeOutput(command: string, signal: AbortSignal): Promise { return new Promise((resolve, reject) => { - const child = spawn(command, PROBE_ARGS, { env: process.env, stdio: ['ignore', 'pipe', 'ignore'] }); - let stdout = ''; - const fail = (error: Error): void => { child.kill('SIGKILL'); reject(error); }; - const timer = setTimeout(() => fail(new Error('node probe timed out')), PROBE_TIMEOUT_MS); - child.stdout.setEncoding('utf8'); - child.stdout.on('data', (chunk: string) => { - stdout += chunk; - if (Buffer.byteLength(stdout) > PROBE_MAX_BYTES) fail(new Error('node probe output exceeded limit')); - }); - child.once('error', error => { clearTimeout(timer); reject(error); }); - child.once('close', code => { + if (signal.aborted) { reject(new Error('node probe stopped')); return; } + const child = spawn(command, PROBE_ARGS, { env: process.env, stdio: ['ignore', 'pipe', 'pipe'] }); + const stdout: Buffer[] = []; + const bytes = { stdout: 0, stderr: 0 }; + let settled = false; + const finish = (error: Error | undefined): void => { + if (settled) return; + settled = true; clearTimeout(timer); - if (code === 0) resolve(stdout); else reject(new Error('node probe failed')); - }); + signal.removeEventListener('abort', stopped); + if (error === undefined) { resolve(Buffer.concat(stdout).toString('utf8')); return; } + // A grandchild may hold the pipes open; close this side so nothing waits on them. + child.kill('SIGKILL'); child.stdout.destroy(); child.stderr.destroy(); + reject(error); + }; + const stopped = (): void => finish(new Error('node probe stopped')); + const timer = setTimeout(() => finish(new Error('node probe timed out')), PROBE_TIMEOUT_MS); + signal.addEventListener('abort', stopped, { once: true }); + const bounded = (stream: 'stdout' | 'stderr') => (chunk: Buffer): void => { + bytes[stream] += chunk.length; + if (bytes[stream] > PROBE_MAX_BYTES) finish(new Error(`node probe ${stream} exceeded limit`)); + else if (stream === 'stdout') stdout.push(chunk); + }; + child.stdout.on('data', bounded('stdout')); + child.stderr.on('data', bounded('stderr')); + child.once('error', error => finish(error)); + child.once('close', code => finish(code === 0 ? undefined : new Error('node probe failed'))); }); } -async function fileSha256(path: string): Promise { - const digest = createHash('sha256'); - for await (const chunk of createReadStream(path, { highWaterMark: 1 << 20 })) digest.update(chunk as Buffer); - return digest.digest('hex'); +/** SHA-256 of a file, computed on a worker thread. */ +const HASH_WORKER = ` +const { parentPort, workerData } = require('node:worker_threads'); +const { createHash } = require('node:crypto'); +const { readFileSync } = require('node:fs'); +try { parentPort.postMessage({ sha256: createHash('sha256').update(readFileSync(workerData)).digest('hex') }); } +catch (error) { parentPort.postMessage({ error: String(error) }); } +`; +function fileSha256(path: string, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal.aborted) { reject(new Error('node hash stopped')); return; } + const worker = new Worker(HASH_WORKER, { eval: true, workerData: path }); + let settled = false; + const finish = (error: Error | undefined, sha256?: string): void => { + if (settled) return; + settled = true; + signal.removeEventListener('abort', stopped); + void worker.terminate(); + if (error === undefined && sha256 !== undefined && /^[a-f0-9]{64}$/.test(sha256)) resolve(sha256); + else reject(error ?? new Error('node hash failed')); + }; + const stopped = (): void => finish(new Error('node hash stopped')); + signal.addEventListener('abort', stopped, { once: true }); + worker.once('message', (message: { sha256?: string; error?: string }) => + finish(message.error === undefined ? undefined : new Error(message.error), message.sha256)); + worker.once('error', error => finish(error)); + worker.once('exit', () => finish(new Error('node hash worker exited'))); + }); } diff --git a/packages/sdk/src/authored-node-runner.ts b/packages/sdk/src/authored-node-runner.ts index a213767b..f4db5d83 100644 --- a/packages/sdk/src/authored-node-runner.ts +++ b/packages/sdk/src/authored-node-runner.ts @@ -19,7 +19,8 @@ import type { import { HUMAN_WAIT_ID } from './authored-human.js'; import { assertAuthoredPromiseHooks } from './authored-runtime-capability.js'; import { - authoredNodeRefusal as refusal, awaitNodeAuthority, nodeAuthority, prewarmNodeAuthority, type NodeAuthority, + authoredNodeRefusal as refusal, awaitNodeAuthority, nodeAuthority, prewarmNodeAuthority, stopNodeAuthorityPrewarm, + type NodeAuthority, } from './authored-node-authority.js'; import { settlingCompletion } from './step-settling.js'; import { readAuthoredStepIndex } from './authored-step-index.js'; @@ -53,6 +54,11 @@ export function prewarmAuthoredRuntime(): void { if (process.versions['bun'] !== undefined && embeddedSource !== undefined) prewarmNodeAuthority(); } +/** A run that ends before it needs Node must not wait for the prewarm to finish. */ +export async function stopAuthoredRuntimePrewarm(): Promise { + await stopNodeAuthorityPrewarm(); +} + /** Called before root admission/resume; declarative commands never reach it. */ export async function assertAuthoredRuntimeAvailable(): Promise { if (process.versions['bun'] === undefined) assertAuthoredPromiseHooks(); diff --git a/packages/sdk/src/cli/direct-run.ts b/packages/sdk/src/cli/direct-run.ts index 5e874ced..e5eae80f 100644 --- a/packages/sdk/src/cli/direct-run.ts +++ b/packages/sdk/src/cli/direct-run.ts @@ -14,7 +14,7 @@ import { AuthoredFlowExecutionError, } from '../authored-flow-executor.js'; import { executeDurableAuthoredFlow } from '../authored-root.js'; -import { prewarmAuthoredRuntime } from '../authored-node-runner.js'; +import { prewarmAuthoredRuntime, stopAuthoredRuntimePrewarm } from '../authored-node-runner.js'; import { AuthoredHumanParked } from '../authored-flow-error.js'; import { AuthoredFlowLoadError } from '../authored-flow-loader.js'; import { DirectInputError, parseDirectInput } from '../direct-input.js'; @@ -43,6 +43,20 @@ export async function runDirectFlow( inputArgument: string | undefined, dataDir: string, options: RunLifecycleOptions = {}, +): Promise { + try { + return await runDirectFlowPrewarmed(path, inputArgument, dataDir, options); + } finally { + // A refused or failed run may return before anything awaits the prewarm. + await stopAuthoredRuntimePrewarm(); + } +} + +async function runDirectFlowPrewarmed( + path: string, + inputArgument: string | undefined, + dataDir: string, + options: RunLifecycleOptions, ): Promise { // Consume and close the agent-only descriptor before authored source is // imported by trigger preflight. Authored code never receives this object. diff --git a/packages/sdk/tests/authored-node-prewarm.test.ts b/packages/sdk/tests/authored-node-prewarm.test.ts index d77061ea..ec4d8691 100644 --- a/packages/sdk/tests/authored-node-prewarm.test.ts +++ b/packages/sdk/tests/authored-node-prewarm.test.ts @@ -8,6 +8,15 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } // The candidate here is a stand-in that answers the probe (optionally slowly, // wrongly or verbosely) and records each invocation. let directory: string, probes: string, previousOverride: string | undefined; +// The prewarm hashes on a worker thread so the flow load keeps the main thread. +const workers = vi.hoisted(() => ({ constructed: 0 })); +vi.mock('node:worker_threads', async (importOriginal) => { + const actual = await importOriginal(); + class CountedWorker extends actual.Worker { + constructor(...args: ConstructorParameters) { super(...args); workers.constructed += 1; } + } + return { ...actual, Worker: CountedWorker }; +}); type Authority = typeof import('../src/authored-node-authority.js'); let authority: Authority; @@ -19,6 +28,7 @@ beforeAll(() => { beforeEach(async () => { writeFileSync(probes, ''); + workers.constructed = 0; // The prewarm and its memo are per process; give every case a fresh module. vi.resetModules(); authority = await import('../src/authored-node-authority.js'); @@ -33,10 +43,11 @@ afterAll(() => rmSync(directory, { recursive: true, force: true })); const probeLog = (): string[] => readFileSync(probes, 'utf8').split('\n').filter(Boolean); -function candidate(name: string, { delaySeconds = 0, exit = 0, padding = 0 } = {}): string { +function candidate(name: string, { delaySeconds = 0, exit = 0, padding = 0, stderrBytes = 0 } = {}): string { const path = join(directory, name); const answer = JSON.stringify({ path, version: '22.14.0' }) + ' '.repeat(padding); writeFileSync(path, `#!/bin/sh\necho ${name} >> ${JSON.stringify(probes)}\nsleep ${delaySeconds}\n` + + (stderrBytes > 0 ? `head -c ${stderrBytes} /dev/zero | tr '\\0' x >&2\n` : '') + `printf '%s' '${answer}'\nexit ${exit}\n`); chmodSync(path, 0o755); process.env['FLOWS_AUTHORED_NODE'] = path; @@ -55,6 +66,7 @@ describe('authored Node authority prewarm', () => { }); expect(authority.nodeAuthority()).toBe(resolved); expect(probeLog()).toEqual(['node-a']); + expect(workers.constructed).toBe(1); }); it('returns at once and leaves the event loop free while the probe runs', async () => { @@ -90,4 +102,21 @@ describe('authored Node authority prewarm', () => { await expect(authority.awaitNodeAuthority()).rejects.toThrow('requires the embedded Node runner'); expect(probeLog()).toEqual(['node-verbose', 'node-verbose']); }); + + it('does not trust a probe whose stderr is over the limit the synchronous probe enforces', async () => { + candidate('node-noisy', { stderrBytes: 5000 }); + authority.prewarmNodeAuthority(); + await expect(authority.awaitNodeAuthority()).rejects.toThrow('requires the embedded Node runner'); + expect(probeLog()).toEqual(['node-noisy', 'node-noisy']); + }); + + it('stops a slow probe at once when the run no longer needs it', async () => { + candidate('node-stuck', { delaySeconds: 5 }); + authority.prewarmNodeAuthority(); + while (probeLog().length === 0) await new Promise(resolve => setTimeout(resolve, 10)); + const started = performance.now(); + await authority.stopNodeAuthorityPrewarm(); + expect(performance.now() - started).toBeLessThan(1000); + expect(probeLog()).toEqual(['node-stuck']); + }); }); diff --git a/packages/sdk/tests/authored-node-runtime.test.ts b/packages/sdk/tests/authored-node-runtime.test.ts index a97e15b1..a988c3b2 100644 --- a/packages/sdk/tests/authored-node-runtime.test.ts +++ b/packages/sdk/tests/authored-node-runtime.test.ts @@ -93,6 +93,27 @@ async function entries(directory: string, runId: string) { } describe('Bun 1.4.x standalone → native Node authored lifecycle', () => { + it('does not wait for a slow Node probe when the flow is refused before it needs Node', () => { + const f = fixture(''); + const node = join(f.directory, 'slow-node-wrapper'); + writeFileSync(node, `#!/bin/sh\necho probe >> node-probes\nsleep 8\nexec ${JSON.stringify(process.execPath)} "$@"\n`); + chmodSync(node, 0o755); + // The import fails only once the slow probe is under way, so the refusal races a live probe. + writeFileSync(join(f.directory, 'case.flow.ts'), `import { flow } from '@relayflows/surface'; +import { existsSync } from 'node:fs'; +const deadline = Date.now() + 5000; +while (!existsSync('node-probes') && Date.now() < deadline) await new Promise(r => setTimeout(r, 10)); +throw new Error('refused after the probe began'); +export default flow('runtime-case', async f => { f.done('success'); }); +`); + const started = Date.now(); + const run = f.invoke(['run', 'case.flow.ts', '--input', '{}'], { FLOWS_AUTHORED_NODE: node }); + expect(run.status, run.stderr + run.stdout).toBe(2); + expect(run.stdout).toContain('refused after the probe began'); + expect(readFileSync(join(f.directory, 'node-probes'), 'utf8')).toBe('probe\n'); + expect(Date.now() - started).toBeLessThan(5000); + }, 60_000); + it('starts the Node probe while it imports the flow, and probes once per run', () => { const f = fixture(''); const node = join(f.directory, 'node-probe-wrapper');