diff --git a/docs/product/output-conventions.md b/docs/product/output-conventions.md index 5ead36eb..84e5b9fd 100644 --- a/docs/product/output-conventions.md +++ b/docs/product/output-conventions.md @@ -37,6 +37,7 @@ command: - in human mode, the child inherits stdin, stdout, and stderr - in JSON mode, stdin remains inherited while the child's stdout and stderr are routed to the CLI's diagnostic stderr stream +- when the command transforms the child's output (`transformOutput`), stdin remains inherited and the engine reads the child's stdout and stderr, passes each through the command's transform, and writes the result to the same place the two cases above would have: the child's stdout to stdout and its stderr to stderr in human and markdown format, and both to stderr in JSON mode. The engine also replaces the credential it injected into the child's environment with `[redacted]`; without `transformOutput` it does not. - JSON stdout remains an NDJSON event stream with exactly one terminal result frame - a non-zero child status is preserved as the process exit code and is diff --git a/packages/cli-engine/README.md b/packages/cli-engine/README.md index 419a0d4a..eb661c54 100644 --- a/packages/cli-engine/README.md +++ b/packages/cli-engine/README.md @@ -75,4 +75,38 @@ A statement is the command's own declared input, unlike a `--confirm` value, whi Each flag answers one question. A run that succeeds with a flag nothing consumed fails with `CLI.CONSENT_UNUSED`, so a mistyped subject or a second answer to one question is never ignored. The same holds for a `--confirm` token no consent asked for, on a command that declares statements (other commands still ignore a leftover `--confirm`; see [#339](https://github.com/prisma/prisma-cli/issues/339)). A refusal also names any given statement flag that matches none of its questions, and any `--confirm` token, so the user hears of it before answering a prompt. That check runs when the handler returns, after the command has acted. A command that asks everything in one batch passes `statements(questions, { last: true })` to get the same failure right after the questions are answered, before it does anything with them. +## Transforming a child's output + +A command that declares `maySpawn` runs a child process with `ctx.spawn`. By default the child writes straight to the terminal, or, with `--json`, to stderr. To change the text before it is written, for example to redact secrets, pass `transformOutput`: + +```ts +if (!ctx.spawnOptionNames?.includes("transformOutput")) { + throw new Error("This command needs @prisma/cli-engine 0.10.0 or later."); +} +const result = await ctx.spawn({ + command: "alchemy", + args: ["deploy"], + transformOutput: (stream) => redactor(stream), +}); +``` + +The engine calls `transformOutput` once for `"stdout"` and once for `"stderr"` before the child starts. It pipes each stream, decoded as UTF-8, through the `TransformStream` you return, and writes what comes out where the child's output would have gone without it: + +- human and markdown format: the child's stdout to stdout, its stderr to stderr; +- json format: both to stderr, so stdout carries only the engine's NDJSON output. + +Each stream keeps its own order; there is no order between the two. The engine reads the child only as fast as the terminal takes the text, so a slow reader slows the child down instead of filling memory. When a stream ends, the transform's `flush` runs and its output is written before `ctx.spawn` settles, including when the child was aborted or failed. A transform can therefore hold text back between chunks, as a redactor that matches across chunk boundaries must. Hold back only what could still become a secret, the longest end of the text that is a start of one, and not a fixed number of characters: a child that prints a prompt with no newline and waits for input must show the prompt at once. The engine never splits a UTF-16 surrogate pair across two writes, so a transform that cuts text by length does not corrupt an emoji. + +When the command declares `needs: { credentials: "child" }`, the engine puts the access token in the child's `PRISMA_SERVICE_TOKEN`. The command never sees that value, so when the command passes `transformOutput`, the engine replaces the token with `[redacted]` itself, before your transform runs. Without `transformOutput` the engine does not read the child's output and does not redact the token, in any format, json included. + +The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If a transform throws or errors, or the destination fails, the engine stops reading that stream, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. With `PRISMA_DEBUG=1`, a child output stream whose read fails, or that a grandchild keeps open with no output after the child exits, leaves a line saying its later output was dropped. A transform or destination failure leaves no such line, because `ctx.spawn` rejects with its error. A slow reader does not cause this: the grace period counts only time the engine spends waiting for output that does not come. + +### Checking for support + +Hosts older than 0.10.0 ignore spawn options they do not know, so on those hosts a child would run with its output untransformed. `ctx.spawnOptionNames` lists the option names `ctx.spawn` accepts; it is absent on those hosts, and its type is optional so the compiler makes you check it. Check it before spawning, as above. From 0.10.0 on, `ctx.spawn` refuses any option name it does not know, so a future option cannot be ignored silently. + +### Spawn adapters + +A spawn adapter supports `transformOutput` by starting the child with stdout and stderr piped when the request's `output` is `"engine"`, and returning both as `stdout` and `stderr` on the child. If the adapter returns no streams, the engine kills the child with SIGKILL at once and `ctx.spawn` rejects. An adapter written before `"engine"` existed has usually started that child with inherited output by then, so some of it can reach the terminal before the kill. This check catches a wrongly wired adapter; it does not make redaction safe with one. + Part of [prisma/prisma-cli](https://github.com/prisma/prisma-cli). diff --git a/packages/cli-engine/package.json b/packages/cli-engine/package.json index 02033abf..428fad4b 100644 --- a/packages/cli-engine/package.json +++ b/packages/cli-engine/package.json @@ -1,6 +1,6 @@ { "name": "@prisma/cli-engine", - "version": "0.9.0", + "version": "0.10.0", "description": "The execution engine of the unified Prisma CLI.", "type": "module", "exports": { diff --git a/packages/cli-engine/src/context.ts b/packages/cli-engine/src/context.ts index 357ef066..570e311b 100644 --- a/packages/cli-engine/src/context.ts +++ b/packages/cli-engine/src/context.ts @@ -63,9 +63,11 @@ export interface CommandContext< readonly api: ManagementApiClient; /** - * Hands the terminal to a child process and resolves when it ends. - * The child inherits stdio and runs in this process's own group - * (POSIX) or console (Windows), so Ctrl-C reaches it natively. + * Runs a child process and resolves when it ends. The child inherits + * stdin and runs in this process's own group (POSIX) or console + * (Windows), so Ctrl-C reaches it natively. Its stdout and stderr go + * where the output format sends them (see output-conventions.md), + * passing through `transformOutput` when the handler gives one. * * While a child is live the engine neither aborts nor exits on a * delivered signal: it records signals and replays them into its @@ -85,6 +87,14 @@ export interface CommandContext< */ readonly spawn: (options: SpawnOptions) => Promise; + /** + * The option names ctx.spawn accepts on this host. ctx.spawn refuses + * any other name. Absent on hosts older than 0.10.0, which ignore + * options they do not know, so a command that depends on an option + * checks for it here before spawning. + */ + readonly spawnOptionNames?: readonly (keyof SpawnOptions)[]; + /** * How the run's most recent completed child ended, or undefined when * none has run. The engine records every child ctx.spawn returns, so diff --git a/packages/cli-engine/src/execution/command-context.ts b/packages/cli-engine/src/execution/command-context.ts index e3f27187..b8ea105e 100644 --- a/packages/cli-engine/src/execution/command-context.ts +++ b/packages/cli-engine/src/execution/command-context.ts @@ -15,6 +15,7 @@ import { } from "../presentation"; import { type Diagnostic, notOk, okVoid } from "../protocol"; import type { LoadedConfigFile, OutputStream } from "../runtime"; +import { SPAWN_OPTION_NAMES } from "../spawn"; import { buildManagementApiClient } from "./api-client"; import { constructionError } from "./command-tree"; import type { Invocation, RunState } from "./engine"; @@ -178,6 +179,7 @@ export function makeContext( return api; }, spawn: makeSpawn(invocation, def), + spawnOptionNames: SPAWN_OPTION_NAMES, lastChild: () => state.lastChild, report: (event) => reportEvent(invocation, event), prompt: makePromptSurface(invocation), diff --git a/packages/cli-engine/src/execution/spawn.ts b/packages/cli-engine/src/execution/spawn.ts index 723d9461..6ceeca1c 100644 --- a/packages/cli-engine/src/execution/spawn.ts +++ b/packages/cli-engine/src/execution/spawn.ts @@ -6,13 +6,16 @@ import { } from "../credential-manager"; import type { EngineEvent } from "../events"; import { CliStructuredError } from "../protocol"; +import type { OutputStream } from "../runtime"; import type { ChildResult, + ChildStreamName, SpawnChild, SpawnedChild, SpawnOptions, SpawnRequest, } from "../spawn"; +import { SPAWN_OPTION_NAMES } from "../spawn"; import { constructionError } from "./command-tree"; import { makeDebugLog } from "./debug"; import type { Invocation } from "./engine"; @@ -73,6 +76,14 @@ export function makeSpawn( `command '${state.commandId}' called ctx.spawn without declaring maySpawn`, ); } + const unknown = Object.keys(options).filter( + (name) => !(SPAWN_OPTION_NAMES as readonly string[]).includes(name), + ); + if (unknown.length > 0) { + throw constructionError( + `command '${state.commandId}' passed ctx.spawn options it does not accept: ${unknown.join(", ")}`, + ); + } if (state.delegatedTerminal !== undefined) { throw constructionError( `command '${state.commandId}' called ctx.spawn while a child was still live (one live child per run)`, @@ -133,17 +144,26 @@ async function runDelegated( const state = invocation.state; const debug = makeDebugLog(invocation.runtime); try { + const transforms = makeTransforms(options); + const env = await composeChildEnv(invocation, def, options.env); const request: SpawnRequest = { command: options.command, args: options.args ?? [], cwd: options.cwd ?? invocation.runtime.cwd, - env: await composeChildEnv(invocation, def, options.env), - output: state.format === "json" ? "diagnostic" : "inherit", + env, + output: childOutput(transforms, state.format === "json"), }; debug( `spawn: ${request.command} ${request.args.join(" ")} (cwd ${request.cwd})`, ); - const result = await runChild(invocation, terminal, spawnChild, request); + const result = await runChild( + invocation, + terminal, + spawnChild, + request, + transforms, + injectedSecrets(def, env), + ); debug( `spawn ended: exitCode=${String(result.exitCode)} signal=${String(result.signal)} recorded=[${terminal.recorded.join(",")}]`, ); @@ -161,6 +181,31 @@ async function runDelegated( } } +type ChildTransforms = Readonly< + Record> +>; + +function makeTransforms(options: SpawnOptions): ChildTransforms | undefined { + const transformOutput = options.transformOutput; + if (transformOutput === undefined) { + return undefined; + } + return { + stdout: transformOutput("stdout"), + stderr: transformOutput("stderr"), + }; +} + +function childOutput( + transforms: ChildTransforms | undefined, + structured: boolean, +): SpawnRequest["output"] { + if (transforms !== undefined) { + return "engine"; + } + return structured ? "diagnostic" : "inherit"; +} + /** A handler that resolved or threw with its child still live: the * engine ends the child on the abort ladder and waits for it, because * the engine outliving the child is not negotiable. */ @@ -176,6 +221,8 @@ async function runChild( terminal: DelegatedTerminal, spawnChild: SpawnChild, request: SpawnRequest, + transforms: ChildTransforms | undefined, + secrets: readonly string[], ): Promise { let child: SpawnedChild; try { @@ -187,17 +234,236 @@ async function runChild( if (terminal.recorded.includes("SIGTERM") || terminal.recorded.length > 1) { child.kill("SIGTERM"); } + if ( + transforms !== undefined && + (child.stdout === undefined || child.stderr === undefined) + ) { + return await refuseUnpipedChild(invocation, child); + } const ended = new AbortController(); armTerminationLadder(invocation, terminal, child, ended.signal); try { - return recordCompletedChild(invocation, await child.ended); - } catch (cause) { - throw spawnFailedError(request.command, cause); + const relay = + transforms === undefined + ? undefined + : relayOutput(invocation, terminal, child, transforms, secrets); + let result: ChildResult; + try { + result = recordCompletedChild(invocation, await child.ended); + } catch (cause) { + terminal.termination.abort(); + await relay; + throw spawnFailedError(request.command, cause); + } + const failure = await relay; + if (failure !== undefined) { + throw failure.cause; + } + return result; } finally { ended.abort(); } } +/** The adapter ignored output "engine", so the child is writing + * wherever the adapter sent it. Nothing was promised to a child the + * engine cannot read, so it gets no grace period. */ +async function refuseUnpipedChild( + invocation: Invocation, + child: SpawnedChild, +): Promise { + makeDebugLog(invocation.runtime)( + "spawn: the adapter returned no stdout and stderr for output engine; SIGKILL", + ); + child.kill("SIGKILL"); + const ended = await child.ended.catch(() => undefined); + if (ended !== undefined) { + recordCompletedChild(invocation, ended); + } + throw new Error( + "@prisma/cli-engine: ctx.spawn was given transformOutput but the Runtime's spawn adapter returned no piped stdout and stderr", + ); +} + +/** Pipes each child stream through its transform to the stream the + * default output mode would have used. A failing transform ends the + * child on the abort ladder and is returned, so ctx.spawn rejects with + * it; the other stream keeps flowing until the child ends. */ +async function relayOutput( + invocation: Invocation, + terminal: DelegatedTerminal, + child: SpawnedChild, + transforms: ChildTransforms, + secrets: readonly string[], +): Promise<{ readonly cause: unknown } | undefined> { + const runtime = invocation.runtime; + const debug = makeDebugLog(runtime); + const structured = invocation.state.format === "json"; + let failure: { readonly cause: unknown } | undefined; + const relay = async (stream: ChildStreamName): Promise => { + const source = stream === "stdout" ? child.stdout : child.stderr; + if (source === undefined) { + return; + } + const destination = + stream === "stdout" && !structured ? runtime.stdout : runtime.stderr; + try { + await relayStream( + source, + [...secrets.map(redactingTransform), transforms[stream]], + destination, + (cause) => { + debug( + `spawn: the child's ${stream} failed, so its later output was dropped: ${errorMessage(cause)}`, + ); + }, + ); + } catch (cause) { + failure ??= { cause }; + terminal.termination.abort(); + } + }; + await Promise.all([relay("stdout"), relay("stderr")]); + return failure; +} + +/** Reads `source` only as fast as `destination` drains, so a slow + * destination makes the child block on its own writes. A source that + * fails ends normally, so the transform still flushes. Rejects when + * the transform or the destination fails. */ +export async function relayStream( + source: AsyncIterable, + transforms: readonly TransformStream[], + destination: OutputStream, + onSourceError: (cause: unknown) => void, +): Promise { + await [...transforms, wholeCharacters()] + .reduce( + (readable, transform) => readable.pipeThrough(transform), + readableFrom(endOnError(source, onSourceError)), + ) + .pipeTo( + new WritableStream({ + write: async (text) => { + destination.write(text); + await destination.drained?.(); + }, + }), + ); +} + +/** ReadableStream.from, which Bun does not have. A high-water mark of + * zero reads the source only when the pipe asks for a chunk. */ +function readableFrom(source: AsyncIterable): ReadableStream { + const chunks = source[Symbol.asyncIterator](); + return new ReadableStream( + { + pull: async (controller) => { + const next = await chunks.next(); + if (next.done === true) { + controller.close(); + } else { + controller.enqueue(next.value); + } + }, + cancel: async () => { + await chunks.return?.(); + }, + }, + { highWaterMark: 0 }, + ); +} + +/** The credential the engine put in the child's environment. The + * command never sees it, so only the engine can redact it. */ +function injectedSecrets( + def: AnyCommand, + env: Readonly>, +): readonly string[] { + if (def.needs.credentials !== "child") { + return []; + } + const token = env[SERVICE_TOKEN_ENV_VAR]; + return token === undefined || token === "" ? [] : [token]; +} + +/** Replaces `secret` with "[redacted]" across chunk boundaries. It + * holds back only an end of the text that could still become the + * secret, so a prompt with no newline is written at once. */ +function redactingTransform(secret: string): TransformStream { + let held = ""; + return new TransformStream({ + transform: (chunk, controller) => { + const text = (held + chunk).replaceAll(secret, "[redacted]"); + const cut = text.length - partialMatchLength(text, secret); + held = text.slice(cut); + if (cut > 0) { + controller.enqueue(text.slice(0, cut)); + } + }, + flush: (controller) => { + if (held !== "") { + controller.enqueue(held); + } + }, + }); +} + +/** The length of the longest end of `text` that is a start of `secret`. */ +function partialMatchLength(text: string, secret: string): number { + for ( + let length = Math.min(text.length, secret.length - 1); + length > 0; + length -= 1 + ) { + if (text.endsWith(secret.slice(0, length))) { + return length; + } + } + return 0; +} + +/** Holds a trailing high surrogate until its pair arrives: a lone half + * written on its own is encoded as U+FFFD. */ +function wholeCharacters(): TransformStream { + let held = ""; + return new TransformStream({ + transform: (chunk, controller) => { + const text = held + chunk; + const cut = endsWithHighSurrogate(text) ? text.length - 1 : text.length; + held = text.slice(cut); + if (cut > 0) { + controller.enqueue(text.slice(0, cut)); + } + }, + flush: (controller) => { + if (held !== "") { + controller.enqueue(held); + } + }, + }); +} + +function endsWithHighSurrogate(text: string): boolean { + const last = text.charCodeAt(text.length - 1); + return last >= 0xd800 && last <= 0xdbff; +} + +async function* endOnError( + source: AsyncIterable, + onError: (cause: unknown) => void, +): AsyncGenerator { + try { + yield* source; + } catch (cause) { + onError(cause); + } +} + +function errorMessage(cause: unknown): string { + return cause instanceof Error ? cause.message : String(cause); +} + /** The engine mints the result the handler sees and keeps it on the * run: the settlement then reads the child's status from the engine's * own record rather than from anything the handler hands back, and a diff --git a/packages/cli-engine/src/exports/index.ts b/packages/cli-engine/src/exports/index.ts index 8dc5b13c..020f6ee7 100644 --- a/packages/cli-engine/src/exports/index.ts +++ b/packages/cli-engine/src/exports/index.ts @@ -140,6 +140,7 @@ export { export type { EngineCommandSnapshot, RunSummary } from "../run-summary"; export { type Host, + type HostOutputStream, type HostProcess, type InputStream, type LoadedConfig, @@ -151,6 +152,7 @@ export { export { type ChildResult, type ChildStatusSettlement, + type ChildStreamName, type ExitWithChildStatusOptions, exitWithChildStatus, type SpawnChild, diff --git a/packages/cli-engine/src/runtime.ts b/packages/cli-engine/src/runtime.ts index 7bbb620e..1771b70b 100644 --- a/packages/cli-engine/src/runtime.ts +++ b/packages/cli-engine/src/runtime.ts @@ -12,6 +12,10 @@ export interface OutputStream { * The engine reads it at render time rather than caching it, so a * terminal resized mid-run is respected by the next thing drawn. */ readonly columns?: number; + /** Resolves once the stream can take more text without buffering it + * in memory, and rejects when the stream fails. Absent means the + * stream never asks the writer to wait. */ + readonly drained?: () => Promise; } /** @@ -155,6 +159,24 @@ export interface Host { readonly arch: string; } +type HostStreamEvent = "drain" | "error" | "close"; +type HostStreamListener = (cause?: unknown) => void; + +/** The parts of a Node writable the bin reads off process.stdout and + * process.stderr. A stream that can wait for drain has both `once` and + * `off`, so waiting never leaves a listener behind. */ +export type HostOutputStream = { + write(text: string): unknown; + isTTY?: boolean; + writableNeedDrain?: boolean; +} & ( + | { + once(event: HostStreamEvent, listener: HostStreamListener): unknown; + off(event: HostStreamEvent, listener: HostStreamListener): unknown; + } + | { once?: undefined; off?: undefined } +); + /** * The minimal process surface a bin adapts a Runtime from — Node's * `process` satisfies it structurally. The engine never reads it; it @@ -168,12 +190,8 @@ export interface HostProcess { readonly platform: string; readonly arch: string; cwd(): string; - readonly stdout: { write(text: string): unknown; isTTY?: boolean }; - readonly stderr: { - write(text: string): unknown; - isTTY?: boolean; - columns?: number; - }; + readonly stdout: HostOutputStream; + readonly stderr: HostOutputStream & { columns?: number }; readonly stdin: { isTTY?: boolean; setRawMode?(enabled: boolean): unknown; diff --git a/packages/cli-engine/src/spawn.ts b/packages/cli-engine/src/spawn.ts index bd36f396..e284191e 100644 --- a/packages/cli-engine/src/spawn.ts +++ b/packages/cli-engine/src/spawn.ts @@ -19,10 +19,18 @@ export interface SpawnRequest { * NDJSON on stdout in that mode and cannot detect an adapter that * ignores this field — inherited child stdout would silently corrupt * the stream. Adapters written before this field existed must be - * updated. */ - readonly output: "inherit" | "diagnostic"; + * updated. "engine" keeps stdin inherited and pipes stdout and stderr + * back to the engine: the adapter MUST return them as + * SpawnedChild.stdout and SpawnedChild.stderr. The engine SIGKILLs a + * child returned without them, but an adapter that predates "engine" + * has already started that child with inherited output, so some of + * it can reach the terminal before the kill. */ + readonly output: "inherit" | "diagnostic" | "engine"; } +/** Names one of the child's two output streams. */ +export type ChildStreamName = "stdout" | "stderr"; + /** How a child ended. A signal-killed child carries `signal` and a null * `exitCode`, and is an abort rather than a failure: exitWithChildStatus * settles it as one, and a handler that reads the result itself branches @@ -38,6 +46,15 @@ export interface SpawnedChild { * launched at all; the engine phrases that as CLI.SPAWN_FAILED. */ readonly ended: Promise; readonly kill: (signal: "SIGTERM" | "SIGKILL") => void; + /** The child's piped output, decoded as UTF-8. Required when the + * request's `output` is "engine", absent otherwise. Each ends when + * the child closes it. After the child exits, an adapter ends a + * stream that the engine has waited on for a grace period with no + * output, so a grandchild holding the pipe open cannot hold the run. + * Time the engine spends not reading does not count, so a slow + * destination does not lose the child's last output. */ + readonly stdout?: AsyncIterable; + readonly stderr?: AsyncIterable; } /** @@ -45,7 +62,9 @@ export interface SpawnedChild { * inherited stdio, in the caller's own process group (POSIX) / console * (Windows). In structured mode stdin remains inherited while stdout and * stderr are routed to the host's diagnostic stream, preserving framed - * stdout. Neither mode detaches or opens a new console. + * stdout. In engine mode stdin remains inherited and both output + * streams are returned to the engine. No mode detaches or opens a new + * console. */ export type SpawnChild = (request: SpawnRequest) => SpawnedChild; @@ -58,8 +77,36 @@ export interface SpawnOptions { /** Added to, and overriding, the invocation environment. The engine's * credential variables are applied last and cannot be overridden. */ readonly env?: Readonly>; + /** + * Passes the child's output through a transform before the engine + * writes it. The engine calls this once per stream before the child + * starts, pipes that stream through the returned transform as UTF-8 + * text, and writes the result where the child's output goes without + * it. The transform's `flush` runs when the stream ends, before + * ctx.spawn settles. If a transform throws or errors, the engine ends + * the child as on abort and ctx.spawn rejects with that error. The + * engine replaces the credential it injects into the child's + * environment with `[redacted]` before the transform sees the text. + * Hosts older than 0.10.0 ignore this option: check + * `ctx.spawnOptionNames?.includes("transformOutput")` first. + */ + readonly transformOutput?: ( + stream: ChildStreamName, + ) => TransformStream; } +const ACCEPTED_SPAWN_OPTIONS = { + command: true, + args: true, + cwd: true, + env: true, + transformOutput: true, +} as const satisfies Record; + +/** Every option ctx.spawn accepts; it refuses any other. */ +export const SPAWN_OPTION_NAMES: readonly (keyof SpawnOptions)[] = + Object.freeze(Object.keys(ACCEPTED_SPAWN_OPTIONS) as (keyof SpawnOptions)[]); + export const CHILD_STATUS: unique symbol = Symbol.for( "@prisma/cli-engine.childStatus", ); diff --git a/packages/cli-engine/src/testing.ts b/packages/cli-engine/src/testing.ts index a90b0cf9..11d12539 100644 --- a/packages/cli-engine/src/testing.ts +++ b/packages/cli-engine/src/testing.ts @@ -18,20 +18,72 @@ import type { PackageManagerId, PackageManagerRunner } from "./package-manager"; import type { PresentedResult } from "./presentation"; import type { RunSummary } from "./run-summary"; import type { Host, Runtime } from "./runtime"; -import type { ChildResult, SpawnChild, SpawnRequest } from "./spawn"; +import type { + ChildResult, + ChildStreamName, + SpawnChild, + SpawnRequest, +} from "./spawn"; import type { TelemetryPayload } from "./telemetry/payload"; import type { TelemetryDeclaration } from "./telemetry/report"; /** * What a scripted fake child does. `nextKill` resolves with each signal * the engine delivers, so a script can model a child that ignores - * SIGTERM and only dies on SIGKILL. + * SIGTERM and only dies on SIGKILL. `write` emits output when the + * request's `output` is "engine" and is ignored otherwise; both streams + * end when the program returns. */ export type ScriptedChildProgram = ( request: SpawnRequest, - child: { readonly nextKill: () => Promise<"SIGTERM" | "SIGKILL"> }, + child: { + readonly nextKill: () => Promise<"SIGTERM" | "SIGKILL">; + readonly write: (stream: ChildStreamName, chunk: string) => void; + }, ) => ChildResult | Promise; +interface ScriptedStream extends AsyncIterable { + push(chunk: string): void; + end(): void; +} + +function scriptedStream(): ScriptedStream { + const chunks: string[] = []; + let ended = false; + let wake: (() => void) | undefined; + const notify = () => { + const waiting = wake; + wake = undefined; + waiting?.(); + }; + return { + push: (chunk) => { + chunks.push(chunk); + notify(); + }, + end: () => { + ended = true; + notify(); + }, + [Symbol.asyncIterator]: () => { + const next = async (): Promise> => { + const chunk = chunks.shift(); + if (chunk !== undefined) { + return { value: chunk, done: false }; + } + if (ended) { + return { value: undefined, done: true }; + } + await new Promise((resolve) => { + wake = resolve; + }); + return next(); + }; + return { next }; + }, + }; +} + /** One ctx.spawn call, as the harness saw it. */ export interface SpawnRecord { readonly command: string; @@ -53,19 +105,33 @@ function scriptedSpawn(program: ScriptedChildProgram): SpawnChild { return (request) => { const delivered: ("SIGTERM" | "SIGKILL")[] = []; const waiting: Array<(signal: "SIGTERM" | "SIGKILL") => void> = []; + const piped = + request.output === "engine" + ? { stdout: scriptedStream(), stderr: scriptedStream() } + : undefined; return { - ended: (async () => - program(request, { - nextKill: async () => { - const queued = delivered.shift(); - return ( - queued ?? - (await new Promise<"SIGTERM" | "SIGKILL">((resolve) => { - waiting.push(resolve); - })) - ); - }, - }))(), + ended: (async () => { + try { + return await program(request, { + nextKill: async () => { + const queued = delivered.shift(); + return ( + queued ?? + (await new Promise<"SIGTERM" | "SIGKILL">((resolve) => { + waiting.push(resolve); + })) + ); + }, + write: (stream, chunk) => { + piped?.[stream].push(chunk); + }, + }); + } finally { + piped?.stdout.end(); + piped?.stderr.end(); + } + })(), + ...piped, kill: (signal) => { const waiter = waiting.shift(); if (waiter === undefined) { @@ -99,6 +165,8 @@ function recordingSpawn( record.kills.push(signal); child.kill(signal); }, + stdout: child.stdout, + stderr: child.stderr, }; }; } diff --git a/packages/cli-engine/tests/fixtures/child.mjs b/packages/cli-engine/tests/fixtures/child.mjs index e4680b93..90276971 100644 --- a/packages/cli-engine/tests/fixtures/child.mjs +++ b/packages/cli-engine/tests/fixtures/child.mjs @@ -40,6 +40,22 @@ switch (mode) { idle(); writeFileSync(value, "ready"); break; + case "chatter": { + const pause = () => new Promise((resolve) => setTimeout(resolve, 50)); + process.stdout.write("out-1\n"); + await pause(); + process.stderr.write("err-1\n"); + await pause(); + process.stdout.write("out-2\n"); + await pause(); + process.exitCode = Number(value); + break; + } + case "chatter-ignore-term": + process.on("SIGTERM", () => {}); + setInterval(() => process.stdout.write("tick\n"), 20); + writeFileSync(value, String(process.pid)); + break; default: process.exit(9); } diff --git a/packages/cli-engine/tests/spawn-real-child.test.ts b/packages/cli-engine/tests/spawn-real-child.test.ts index eb0e1301..7639addb 100644 --- a/packages/cli-engine/tests/spawn-real-child.test.ts +++ b/packages/cli-engine/tests/spawn-real-child.test.ts @@ -12,6 +12,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { fileURLToPath } from "node:url"; import { + type ChildStreamName, defineCommand, exitWithChildStatus, type SpawnChild, @@ -21,6 +22,7 @@ import { createTestCli } from "@prisma/cli-engine/testing"; import { describe, expect, test } from "vitest"; const NODE = process.execPath; +const TICKS = /^(tick\n)+$/; const CHILD = fileURLToPath(new URL("./fixtures/child.mjs", import.meta.url)); const HOST = fileURLToPath( new URL("./fixtures/spawn-host.mjs", import.meta.url), @@ -46,6 +48,57 @@ const realSpawn: SpawnChild = (request) => { }; }; +/** Engine mode with both output streams returned to the engine. The + * shipped adapter's engine mode is tested end to end in + * packages/cli/tests/spawn-engine-output.test.ts. */ +const pipingSpawn: SpawnChild = (request) => { + const child = spawn(request.command, [...request.args], { + cwd: request.cwd, + env: request.env, + stdio: ["ignore", "pipe", "pipe"], + }); + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + return { + ended: new Promise((resolve, reject) => { + child.on("error", reject); + child.on("close", (exitCode, signal) => resolve({ exitCode, signal })); + }), + kill: (signal) => { + child.kill(signal); + }, + stdout: child.stdout, + stderr: child.stderr, + }; +}; + +type Chunk = readonly [stream: ChildStreamName, chunk: string]; + +function pipingCommand(chunks: Chunk[], ...childArgs: string[]) { + return defineCommand({ + help: { summary: "Transforms the fixture child's output" }, + maySpawn: true, + handler: async (_args, ctx) => { + await ctx.spawn({ + command: NODE, + args: [CHILD, ...childArgs], + transformOutput: recording(chunks), + }); + return ok(exitWithChildStatus()); + }, + }); +} + +function recording(chunks: Chunk[]) { + return (stream: ChildStreamName) => + new TransformStream({ + transform: (chunk, controller) => { + chunks.push([stream, chunk]); + controller.enqueue(chunk); + }, + }); +} + function childCommand(...childArgs: string[]) { return defineCommand({ help: { summary: "Runs the fixture child" }, @@ -79,7 +132,7 @@ function termAfterReady(adapter: SpawnChild, ready: string): SpawnChild { return (request) => { const child = adapter(request); return { - ended: child.ended, + ...child, kill: (signal) => { if (signal === "SIGTERM") { waitForFile(ready, 2_000) @@ -257,6 +310,75 @@ describe.skipIf(process.platform === "win32")( expect(result.spawns[0].kills).toEqual(["SIGTERM", "SIGKILL"]); expect(result.exitCode).toBe(137); }); + + test("transformed output keeps each stream's order, and the exit code passes through", async () => { + const chunks: Chunk[] = []; + const cli = createTestCli({ + commands: { converge: pipingCommand(chunks, "chatter", "4") }, + spawn: pipingSpawn, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.exitCode).toBe(4); + const text = (stream: ChildStreamName) => + chunks + .filter(([name]) => name === stream) + .map(([, chunk]) => chunk) + .join(""); + expect(text("stdout")).toBe("out-1\nout-2\n"); + expect(text("stderr")).toBe("err-1\n"); + expect(result.stdout).toBe("out-1\nout-2\n"); + expect(result.stderr).toBe("err-1\n"); + }); + + test("a piped child that ignores SIGTERM is killed on abort, leaving no orphan", async () => { + const dir = scratch(); + const pidFile = join(dir, "pid"); + const chunks: Chunk[] = []; + const controller = new AbortController(); + const watcher = defineCommand({ + help: { summary: "Aborts while a piped child runs" }, + maySpawn: true, + handler: async (_args, ctx) => { + ctx.report({ kind: "status", subject: "run", status: "ready" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + await ctx.spawn({ + command: NODE, + args: [CHILD, "chatter-ignore-term", pidFile], + transformOutput: recording(chunks), + }); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: watcher }, + spawn: termAfterReady(pipingSpawn, pidFile), + delay: async () => { + await waitForFile(pidFile); + await new Promise((resolve) => setTimeout(resolve, 300)); + }, + }); + + const result = await cli.run(["converge"], { + abort: controller.signal, + onEvent: (event) => { + if (event.kind === "status") { + controller.abort(); + } + }, + }); + + expect(result.spawns[0].kills).toEqual(["SIGTERM", "SIGKILL"]); + expect(result.exitCode).toBe(137); + expect(chunks.length).toBeGreaterThan(0); + expect( + chunks.every( + ([stream, chunk]) => stream === "stdout" && TICKS.test(chunk), + ), + ).toBe(true); + expect(alive(Number(readFileSync(pidFile, "utf8")))).toBe(false); + }); }, 20_000, ); diff --git a/packages/cli-engine/tests/spawn.test.ts b/packages/cli-engine/tests/spawn.test.ts index d11d97de..6530b6ae 100644 --- a/packages/cli-engine/tests/spawn.test.ts +++ b/packages/cli-engine/tests/spawn.test.ts @@ -10,8 +10,9 @@ import { defineSessionCommand, exitWithChildStatus, type Runtime, + type SpawnOptions, } from "@prisma/cli-engine"; -import { ok } from "@prisma/cli-engine/protocol"; +import { type CliStructuredError, ok } from "@prisma/cli-engine/protocol"; import { createTestCli, mintTestJwt, @@ -19,6 +20,7 @@ import { type SpawnRecord, } from "@prisma/cli-engine/testing"; import { describe, expect, test } from "vitest"; +import { relayStream } from "../src/execution/spawn"; const NOW = new Date("2030-01-01T00:00:00.000Z"); const CLOCK = () => NOW; @@ -1808,3 +1810,759 @@ function controllableRuntime() { }, }; } + +/** Replaces `secret` across chunk boundaries by holding back only an + * end of the text that could still become the secret. */ +function redacting(secret: string): () => TransformStream { + return () => { + let held = ""; + return new TransformStream({ + transform: (chunk, controller) => { + const text = (held + chunk).replaceAll(secret, "[redacted]"); + let keep = Math.min(text.length, secret.length - 1); + while (keep > 0 && !text.endsWith(secret.slice(0, keep))) { + keep -= 1; + } + held = text.slice(text.length - keep); + if (text.length > keep) { + controller.enqueue(text.slice(0, text.length - keep)); + } + }, + flush: (controller) => { + if (held !== "") { + controller.enqueue(held); + } + }, + }); + }; +} + +const upperCase = () => + new TransformStream({ + transform: (chunk, controller) => { + controller.enqueue(chunk.toUpperCase()); + }, + }); + +function transformingCommand( + transformOutput: SpawnOptions["transformOutput"], + after?: () => void, +) { + return defineCommand({ + help: { summary: "Transforms a child's output" }, + maySpawn: true, + handler: async (_args, ctx) => { + await ctx.spawn({ + command: "alchemy", + args: ["deploy"], + transformOutput, + }); + after?.(); + return ok(exitWithChildStatus()); + }, + }); +} + +const chatter: ScriptedChildProgram = async (_request, child) => { + child.write("stdout", "one "); + child.write("stdout", "two\n"); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stderr", "warn\n"); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stdout", "three\n"); + return { exitCode: 0, signal: null }; +}; + +describe("transformed child output", () => { + test("human mode writes each transformed stream where the child would have written it", async () => { + const cli = createTestCli({ + commands: { converge: transformingCommand(upperCase) }, + now: CLOCK, + spawnScript: chatter, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.exitCode).toBe(0); + expect(result.spawns[0]?.output).toBe("engine"); + expect(result.stdout).toBe("ONE TWO\nTHREE\n"); + expect(result.stderr).toBe("WARN\n"); + }); + + test("markdown mode follows the inherited handoff too", async () => { + const cli = createTestCli({ + commands: { converge: transformingCommand(upperCase) }, + now: CLOCK, + spawnScript: chatter, + }); + + const result = await cli.run(["converge", "--format", "markdown"]); + + expect(result.stdout).toBe("ONE TWO\nTHREE\n"); + expect(result.stderr).toBe("WARN\n"); + }); + + test("json mode writes both streams to stderr and keeps stdout to the result frame", async () => { + const cli = createTestCli({ + commands: { converge: transformingCommand(upperCase) }, + now: CLOCK, + spawnScript: chatter, + }); + + const result = await cli.run(["converge", "--json"]); + + expect(result.spawns[0]?.output).toBe("engine"); + expect(result.stderr).toBe("ONE TWO\nWARN\nTHREE\n"); + expect(result.json).toHaveLength(1); + expect(result.json[0]).toMatchObject({ kind: "result" }); + }); + + test("a held-back tail is flushed before ctx.spawn resolves", async () => { + const order: string[] = []; + const cli = createTestCli({ + commands: { + converge: transformingCommand(redacting("hunter2"), () => { + order.push("resolved"); + }), + }, + now: CLOCK, + spawnScript: async (_request, child) => { + child.write("stdout", "token=hun"); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stdout", "ter2 done"); + return { exitCode: 0, signal: null }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.stdout).toBe("token=[redacted] done"); + expect(order).toEqual(["resolved"]); + }); + + test("the child's exit code still settles the run", async () => { + const cli = createTestCli({ + commands: { converge: transformingCommand(upperCase) }, + now: CLOCK, + spawnScript: (_request, child) => { + child.write("stderr", "failed\n"); + return { exitCode: 3, signal: null }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.exitCode).toBe(3); + expect(result.stderr).toBe("FAILED\n"); + }); + + test("a SIGTERM delivered while transforming is forwarded, and the held-back tail is still written", async () => { + const controller = new AbortController(); + const cli = createTestCli({ + commands: { converge: transformingCommand(redacting("hunter2")) }, + now: CLOCK, + spawnScript: async (_request, child) => { + child.write("stdout", "working on hunt"); + controller.abort("SIGTERM"); + await child.nextKill(); + return { exitCode: null, signal: "SIGTERM" }; + }, + }); + + const result = await cli.run(["converge"], { abort: controller.signal }); + + expect(result.spawns[0]?.kills).toEqual(["SIGTERM"]); + expect(result.stdout).toBe("working on hunt"); + expect(result.exitCode).toBe(143); + }); + + test("an abort runs the termination ladder on a child whose output is transformed", async () => { + const controller = new AbortController(); + const late = defineCommand({ + help: { summary: "Aborts while a child runs" }, + maySpawn: true, + handler: async (_args, ctx) => { + ctx.report({ kind: "status", subject: "run", status: "ready" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + await ctx.spawn({ command: "alchemy", transformOutput: upperCase }); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { late }, + now: CLOCK, + spawnScript: async (_request, child) => { + child.write("stdout", "still deploying\n"); + await child.nextKill(); + await child.nextKill(); + return { exitCode: null, signal: "SIGKILL" }; + }, + }); + + const result = await cli.run(["late"], { + abort: controller.signal, + onEvent: (event) => { + if (event.kind === "status") { + controller.abort(); + } + }, + }); + + expect(result.spawns[0]?.kills).toEqual(["SIGTERM", "SIGKILL"]); + expect(result.stdout).toBe("STILL DEPLOYING\n"); + expect(result.exitCode).toBe(137); + }); + + test("a throwing transform ends the child, and ctx.spawn rejects with its error", async () => { + let caught: unknown; + const command = defineCommand({ + help: { summary: "Fails while transforming a child's output" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: "alchemy", + transformOutput: (stream) => + new TransformStream({ + transform: (chunk, controller) => { + if (stream === "stdout" && chunk === "second") { + throw new Error("redaction failed"); + } + controller.enqueue(chunk); + }, + }), + }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + delay: (_ms, signal) => signalDone(signal), + spawnScript: async (_request, child) => { + child.write("stdout", "first"); + child.write("stdout", "second"); + child.write("stdout", "third"); + await child.nextKill(); + child.write("stderr", "stopping"); + return { exitCode: null, signal: "SIGTERM" }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.spawns[0]?.kills).toEqual(["SIGTERM"]); + expect(result.stdout).toBe("first"); + expect(result.stderr).toBe("stopping"); + expect((caught as Error).message).toBe("redaction failed"); + }); + + test("a transform whose promise rejects later ends the child, and ctx.spawn rejects with its error", async () => { + let caught: unknown; + const command = defineCommand({ + help: { summary: "Fails asynchronously while transforming" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: "alchemy", + transformOutput: () => + new TransformStream({ + transform: async (chunk, controller) => { + await new Promise((resolve) => setTimeout(resolve, 0)); + if (chunk === "second") { + throw new Error("redaction failed later"); + } + controller.enqueue(chunk); + }, + }), + }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + delay: (_ms, signal) => signalDone(signal), + spawnScript: async (_request, child) => { + child.write("stdout", "first"); + child.write("stdout", "second"); + await child.nextKill(); + return { exitCode: null, signal: "SIGTERM" }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.spawns[0]?.kills).toEqual(["SIGTERM"]); + expect(result.stdout).toBe("first"); + expect((caught as Error).message).toBe("redaction failed later"); + }); + + test("a throwing transformOutput rejects ctx.spawn before any child starts", async () => { + let caught: unknown; + const command = defineCommand({ + help: { summary: "Fails to build its transform" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: "alchemy", + transformOutput: () => { + throw new Error("no redactor"); + }, + }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ commands: { converge: command }, now: CLOCK }); + + const result = await cli.run(["converge"]); + + expect(result.spawns).toEqual([]); + expect((caught as Error).message).toBe("no redactor"); + }); + + test("a child the adapter could not keep running still has its held-back tail flushed before ctx.spawn rejects", async () => { + let caught: unknown; + const command = defineCommand({ + help: { summary: "Spawns a child whose adapter fails" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: "alchemy", + transformOutput: redacting("hunter2"), + }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + delay: (_ms, signal) => signalDone(signal), + spawnScript: (_request, child) => { + child.write("stdout", "partial hun"); + throw new Error("kill EPERM"); + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.stdout).toBe("partial hun"); + expect((caught as CliStructuredError).code).toBe("CLI.SPAWN_FAILED"); + }); + + test("an adapter that returns no streams has its child killed at once", async () => { + let caught: unknown; + const kills: string[] = []; + const command = defineCommand({ + help: { summary: "Asks for transformed output" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ command: "alchemy", transformOutput: upperCase }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + spawn: () => { + let end: (signal: "SIGTERM" | "SIGKILL") => void = () => {}; + return { + ended: new Promise((resolve) => { + end = (signal) => resolve({ exitCode: null, signal }); + }), + kill: (signal) => { + kills.push(signal); + end(signal); + }, + }; + }, + }); + + await cli.run(["converge"]); + + expect(kills).toEqual(["SIGKILL"]); + expect((caught as Error).message).toContain( + "returned no piped stdout and stderr", + ); + }); + + test("a stream that fails leaves a debug line, and the transform still flushes", async () => { + const failing: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "before hun"; + throw new Error("the pipe was closed"); + }, + }; + const cli = createTestCli({ + commands: { converge: transformingCommand(redacting("hunter2")) }, + now: CLOCK, + spawn: () => ({ + ended: Promise.resolve({ exitCode: 0, signal: null }), + kill: () => {}, + stdout: failing, + stderr: (async function* () {})(), + }), + }); + + const result = await cli.run(["converge", "--format", "human"], { + env: { PRISMA_DEBUG: "1" }, + }); + + expect(result.stdout).toBe("before hun"); + expect(result.stderr).toContain( + "spawn: the child's stdout failed, so its later output was dropped: the pipe was closed", + ); + }); + + test("the credential the engine injects is redacted before the command's transform sees it", async () => { + const token = jwtExpiringIn(3600, "ws_1"); + const seenByCommand: string[] = []; + const command = defineCommand({ + help: { + summary: "Hands credentials to a child and transforms its output", + }, + maySpawn: true, + needs: { credentials: "child" }, + handler: async (_args, ctx) => { + await ctx.spawn({ + command: "alchemy", + transformOutput: () => + new TransformStream({ + transform: (chunk, controller) => { + seenByCommand.push(chunk); + controller.enqueue(chunk); + }, + }), + }); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + sessions: [ + { + workspaceId: "ws_1", + workspaceName: undefined, + credential: { token, refreshToken: "r", expiresAt: undefined }, + }, + ], + selectedWorkspaceId: "ws_1", + spawnScript: async (request, child) => { + const injected = request.env.PRISMA_SERVICE_TOKEN ?? ""; + const middle = Math.floor(injected.length / 2); + child.write("stdout", `token=${injected.slice(0, middle)}`); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stdout", `${injected.slice(middle)} end`); + child.write("stderr", `again ${injected}`); + return { exitCode: 0, signal: null }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.stdout).toBe("token=[redacted] end"); + expect(result.stderr).toBe("again [redacted]"); + expect(seenByCommand.join("")).not.toContain(token); + }); + + test("a prompt with no newline reaches the command's transform before the child exits", async () => { + const token = jwtExpiringIn(3600, "ws_1"); + let promptSeen: () => void = () => {}; + const seen = new Promise((resolve) => { + promptSeen = () => resolve(true); + }); + let seenBeforeExit = false; + const command = defineCommand({ + help: { summary: "Hands credentials to a child that asks a question" }, + maySpawn: true, + needs: { credentials: "child" }, + handler: async (_args, ctx) => { + await ctx.spawn({ + command: "alchemy", + transformOutput: () => + new TransformStream({ + transform: (chunk, controller) => { + if (chunk.includes("(y/N) ")) { + promptSeen(); + } + controller.enqueue(chunk); + }, + }), + }); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + sessions: [ + { + workspaceId: "ws_1", + workspaceName: undefined, + credential: { token, refreshToken: "r", expiresAt: undefined }, + }, + ], + selectedWorkspaceId: "ws_1", + spawnScript: async (_request, child) => { + child.write("stdout", "Deploy to production? (y/N) "); + seenBeforeExit = await Promise.race([ + seen, + new Promise((resolve) => setTimeout(resolve, 500, false)), + ]); + return { exitCode: 0, signal: null }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(seenBeforeExit).toBe(true); + expect(result.stdout).toBe("Deploy to production? (y/N) "); + }); + + test("ctx.spawnOptionNames names transformOutput", async () => { + let names: readonly string[] | undefined; + const command = defineCommand({ + help: { summary: "Checks spawn support" }, + handler: async (_args, ctx) => { + names = ctx.spawnOptionNames; + return ok( + ctx.present( + { data: null }, + { + human: () => [], + stdout: () => [], + json: () => null, + next: () => [], + }, + ), + ); + }, + }); + const cli = createTestCli({ commands: { check: command }, now: CLOCK }); + + await cli.run(["check"]); + + expect(names).toContain("transformOutput"); + }); + + test("ctx.spawn refuses an option it does not know", async () => { + let caught: unknown; + const command = defineCommand({ + help: { summary: "Passes an unknown spawn option" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: "alchemy", + onOutput: () => {}, + } as SpawnOptions); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ commands: { converge: command }, now: CLOCK }); + + const result = await cli.run(["converge"]); + + expect(result.spawns).toEqual([]); + expect((caught as Error).message).toContain( + "passed ctx.spawn options it does not accept: onOutput", + ); + }); + + test("without transformOutput the default handoffs are unchanged", async () => { + const cli = createTestCli({ commands: { converge }, now: CLOCK }); + + const human = await cli.run(["converge"]); + const json = await cli.run(["converge", "--json"]); + + expect(human.spawns[0]?.output).toBe("inherit"); + expect(json.spawns[0]?.output).toBe("diagnostic"); + }); +}); + +describe("relayStream", () => { + test("writes a prompt with no newline at once through a redacting transform", async () => { + let finish: () => void = () => {}; + const finished = new Promise((resolve) => { + finish = resolve; + }); + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "Deploy to production? (y/N) "; + await finished; + yield "token hun"; + yield "ter2\n"; + }, + }; + const written: string[] = []; + const relaying = relayStream( + source, + [redacting("hunter2")()], + { + write: (text) => { + written.push(text); + }, + }, + () => {}, + ); + + await new Promise((resolve) => setTimeout(resolve, 20)); + const writtenBeforeEnd = written.join(""); + finish(); + await relaying; + + expect(writtenBeforeEnd).toBe("Deploy to production? (y/N) "); + expect(written.join("")).toBe( + "Deploy to production? (y/N) token [redacted]\n", + ); + }); + + test("works where ReadableStream.from does not exist, as in Bun", async () => { + const from = Object.getOwnPropertyDescriptor(ReadableStream, "from"); + Reflect.deleteProperty(ReadableStream, "from"); + try { + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "one hun"; + yield "ter2 two\n"; + }, + }; + const written: string[] = []; + + await relayStream( + source, + [redacting("hunter2")()], + { + write: (text) => { + written.push(text); + }, + }, + () => {}, + ); + + expect(written.join("")).toBe("one [redacted] two\n"); + } finally { + if (from !== undefined) { + Object.defineProperty(ReadableStream, "from", from); + } + } + }); + + test("rejects when the destination fails while the relay waits for it to drain", async () => { + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "one\n"; + yield "two\n"; + }, + }; + const written: string[] = []; + + await expect( + relayStream( + source, + [], + { + write: (text) => { + written.push(text); + }, + drained: () => Promise.reject(new Error("EPIPE")), + }, + () => {}, + ), + ).rejects.toThrow("EPIPE"); + expect(written).toEqual(["one\n"]); + }); + + test("never splits a surrogate pair across two writes", async () => { + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "hello \u{1F600}"; + yield " bye"; + }, + }; + const splitAtSeven = new TransformStream({ + transform: (chunk, controller) => { + controller.enqueue(chunk.slice(0, 7)); + if (chunk.length > 7) { + controller.enqueue(chunk.slice(7)); + } + }, + }); + const written: string[] = []; + + await relayStream( + source, + [splitAtSeven], + { + write: (text) => { + written.push(text); + }, + }, + () => {}, + ); + + expect(written.join("")).toBe("hello \u{1F600} bye"); + for (const text of written) { + expect(Buffer.from(text, "utf8").toString("utf8")).toBe(text); + } + }); + + test("does not read ahead of a destination that has not drained", async () => { + let pulled = 0; + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + for (let index = 0; index < 100; index += 1) { + pulled += 1; + yield `chunk ${index}\n`; + } + }, + }; + let release: () => void = () => {}; + const blocked = new Promise((resolve) => { + release = resolve; + }); + const written: string[] = []; + const relaying = relayStream( + source, + [new TransformStream()], + { + write: (text) => { + written.push(text); + }, + drained: () => blocked, + }, + () => {}, + ); + + await new Promise((resolve) => setTimeout(resolve, 20)); + const pulledWhileBlocked = pulled; + release(); + await relaying; + + expect(pulledWhileBlocked).toBeLessThan(10); + expect(written).toHaveLength(100); + }); +}); diff --git a/packages/cli/package.json b/packages/cli/package.json index d8dfbe90..5b6ffab3 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -49,7 +49,7 @@ }, "dependencies": { "@manypkg/tools": "^2.1.2", - "@prisma/cli-engine": "workspace:0.9.0", + "@prisma/cli-engine": "workspace:0.10.0", "@prisma/composer-cli": "0.36.0", "@prisma/compute-sdk": "0.43.0", "@prisma/management-api-sdk": "1.80.0", diff --git a/packages/cli/scripts/conformance.ts b/packages/cli/scripts/conformance.ts index 3403120d..370eb7ce 100644 --- a/packages/cli/scripts/conformance.ts +++ b/packages/cli/scripts/conformance.ts @@ -109,7 +109,24 @@ async function tarball(): Promise { shellPackage: "@prisma/cli", enginePackage: "@prisma/cli-engine", familyPackages: ["@prisma/composer-cli", "@prisma/orm-toolchain"], - exceptions: [], + exceptions: [ + { + familyPackage: "@prisma/composer-cli", + familyPin: "0.9.0", + shellPin: "0.10.0", + reason: "engine 0.10.0 must publish before composer-cli can peer it", + removeWhen: + "composer-cli releases peering 0.10.0 and the follow-up bump PR pins that release", + }, + { + familyPackage: "@prisma/orm-toolchain", + familyPin: "0.9.0", + shellPin: "0.10.0", + reason: "engine 0.10.0 must publish before orm-toolchain can peer it", + removeWhen: + "orm-toolchain releases peering 0.10.0 and the follow-up bump PR pins that release", + }, + ], channel: CHANNEL, sandboxDir: join(WORK_DIR, "sandbox"), }, diff --git a/packages/cli/src/runtime.ts b/packages/cli/src/runtime.ts index 4ba5599f..3b4a7b1c 100644 --- a/packages/cli/src/runtime.ts +++ b/packages/cli/src/runtime.ts @@ -1,6 +1,7 @@ import { fstatSync } from "node:fs"; import { fileURLToPath } from "node:url"; import { + type HostOutputStream, type HostProcess, type InputStream, type LoadedConfig, @@ -146,6 +147,35 @@ export function makeStdin(proc: HostProcess): InputStream { }; } +/** A stream that closes instead of draining resolves too, so a writer + * waiting on it is never stuck. A stream that errors rejects, so the + * writer stops instead of writing to a failed stream. */ +function drained(stream: HostOutputStream): Promise { + const { once, off } = stream; + if (stream.writableNeedDrain !== true || once === undefined) { + return Promise.resolve(); + } + return new Promise((resolve, reject) => { + const settle = (outcome: () => void) => { + off.call(stream, "drain", onDrain); + off.call(stream, "error", onError); + off.call(stream, "close", onDrain); + outcome(); + }; + const onDrain = () => { + settle(resolve); + }; + const onError = (cause?: unknown) => { + settle(() => { + reject(cause instanceof Error ? cause : new Error(String(cause))); + }); + }; + once.call(stream, "drain", onDrain); + once.call(stream, "error", onError); + once.call(stream, "close", onDrain); + }); +} + export async function assembleRuntime(proc: HostProcess): Promise { const stdin = makeStdin(proc); warnOnDeprecatedStateFileEnvVar(proc); @@ -156,11 +186,13 @@ export async function assembleRuntime(proc: HostProcess): Promise { write: (text) => { proc.stdout.write(text); }, + drained: () => drained(proc.stdout), }, stderr: { write: (text) => { proc.stderr.write(text); }, + drained: () => drained(proc.stderr), get columns() { return proc.stderr.columns; }, diff --git a/packages/cli/src/spawn.ts b/packages/cli/src/spawn.ts index 15b8a132..c2bae5b1 100644 --- a/packages/cli/src/spawn.ts +++ b/packages/cli/src/spawn.ts @@ -1,6 +1,7 @@ +import type { ChildProcess } from "node:child_process"; import { type Readable, Writable } from "node:stream"; import { pipeline } from "node:stream/promises"; -import type { ChildResult, SpawnChild } from "@prisma/cli-engine"; +import type { ChildResult, SpawnChild, SpawnedChild } from "@prisma/cli-engine"; import spawn from "cross-spawn"; interface DiagnosticStream { @@ -27,8 +28,9 @@ export interface SpawnChildOptions { /** * The engine's spawn seam, adapted through cross-spawn. Human mode * inherits stdio; structured mode pipes both child output streams to - * diagnostics. Neither mode detaches or opens a new console, so the child - * stays in this process's group (POSIX) or console (Windows). + * diagnostics; engine mode returns both to the engine. No mode detaches or + * opens a new console, so the child stays in this process's group + * (POSIX) or console (Windows). * * The child's own status settles the run: `ended` resolves from the * process `exit` event, waits for the diagnostic relay only up to the @@ -45,7 +47,8 @@ export function makeSpawnChild( const child = spawn(request.command, [...request.args], { cwd: request.cwd, env: request.env, - stdio: structured ? ["inherit", "pipe", "pipe"] : "inherit", + stdio: + request.output === "inherit" ? "inherit" : ["inherit", "pipe", "pipe"], }); const processEnded = new Promise((resolve, reject) => { child.on("error", reject); @@ -53,6 +56,12 @@ export function makeSpawnChild( resolve({ exitCode, signal }); }); }); + const kill = (signal: "SIGTERM" | "SIGKILL") => { + child.kill(signal); + }; + if (request.output === "engine") { + return engineReadChild(child, kill, processEnded, drainGraceMs); + } if (!structured) { return { ended: processEnded, @@ -83,6 +92,99 @@ export function makeSpawnChild( }; } +/** The engine reads both pipes. A grandchild that inherited them can + * hold them open after the child exits, so a pipe the engine has + * waited on with no output for the drain grace is closed, with an + * error that says why. Time the engine spends not reading, because + * its destination is slow, does not count. */ +function engineReadChild( + child: ChildProcess, + kill: SpawnedChild["kill"], + processEnded: Promise, + drainGraceMs: number, +): SpawnedChild { + const exited = processEnded.then( + () => undefined, + () => undefined, + ); + const sources = [child.stdout, child.stderr].filter( + (source): source is Readable => source !== null, + ); + const closed = Promise.all( + sources.map((source) => { + source.setEncoding("utf8"); + return new Promise((resolve) => source.once("close", resolve)); + }), + ); + const readable = (source: Readable | null) => + source === null ? undefined : closeWhenIdle(source, exited, drainGraceMs); + const stdout = readable(child.stdout); + const stderr = readable(child.stderr); + return { + ended: processEnded.then( + async (result) => { + await closed; + return result; + }, + async (cause: unknown) => { + await closed; + throw cause; + }, + ), + kill, + ...(stdout === undefined ? {} : { stdout }), + ...(stderr === undefined ? {} : { stderr }), + }; +} + +function closeWhenIdle( + source: Readable, + exited: Promise, + drainGraceMs: number, +): AsyncIterable { + let childExited = false; + let waiting = false; + let deadline: ReturnType | undefined; + const arm = () => { + if (childExited && waiting && deadline === undefined) { + deadline = setTimeout(() => { + source.destroy( + new Error( + `the pipe stayed open with no output for ${drainGraceMs}ms after the child exited, so it was closed`, + ), + ); + }, drainGraceMs); + } + }; + const disarm = () => { + clearTimeout(deadline); + deadline = undefined; + }; + void exited.then(() => { + childExited = true; + arm(); + }); + return { + [Symbol.asyncIterator]: () => { + const chunks: AsyncIterator = source[Symbol.asyncIterator](); + return { + next: async () => { + waiting = true; + arm(); + try { + return await chunks.next(); + } finally { + waiting = false; + disarm(); + } + }, + return: async () => + (await chunks.return?.()) ?? { done: true, value: undefined }, + }; + }, + }; +} + /** Best-effort relay: a forwarding failure never rejects, so the child's * real status still settles the run when the diagnostic sink dies. */ function forwardStructuredOutput( diff --git a/packages/cli/tests/runtime-drained.test.ts b/packages/cli/tests/runtime-drained.test.ts new file mode 100644 index 00000000..a7662cba --- /dev/null +++ b/packages/cli/tests/runtime-drained.test.ts @@ -0,0 +1,108 @@ +import { EventEmitter } from "node:events"; +import { describe, expect, test } from "vitest"; +import { assembleRuntime, type HostProcess } from "../src/runtime"; + +class HostStream extends EventEmitter { + writableNeedDrain = true; + write(_text: string): boolean { + return false; + } +} + +function hostStreams() { + return { stdout: new HostStream(), stderr: new HostStream() }; +} + +async function runtimeWith(streams: ReturnType) { + const proc: HostProcess = { + argv: ["node", "bin.js"], + version: "v22.18.0", + versions: { node: "22.18.0" }, + platform: "linux", + arch: "x64", + env: {}, + cwd: () => "/projects/acme", + stdout: streams.stdout, + stderr: streams.stderr, + stdin: { + async *[Symbol.asyncIterator]() {}, + }, + on: () => undefined, + off: () => undefined, + exit: (code: number): never => { + throw new Error(`process.exit(${code}) reached the test`); + }, + }; + return await assembleRuntime(proc); +} + +async function settledWithin(promise: Promise, ms: number) { + let settled = false; + void promise.then( + () => { + settled = true; + }, + () => { + settled = true; + }, + ); + await new Promise((resolve) => setTimeout(resolve, ms)); + return settled; +} + +const LISTENED = ["drain", "error", "close"] as const; + +describe.each(["stdout", "stderr"] as const)("runtime.%s.drained", (name) => { + test("stays pending until the host stream emits drain, then resolves", async () => { + const streams = hostStreams(); + const host = streams[name]; + const runtime = await runtimeWith(streams); + + const waiting = runtime[name].drained?.() ?? Promise.resolve(); + + expect(await settledWithin(waiting, 20)).toBe(false); + host.writableNeedDrain = false; + host.emit("drain"); + await expect(waiting).resolves.toBeUndefined(); + for (const event of LISTENED) { + expect(host.listenerCount(event)).toBe(0); + } + }); + + test("resolves when the host stream closes instead of draining", async () => { + const streams = hostStreams(); + const host = streams[name]; + const runtime = await runtimeWith(streams); + + const waiting = runtime[name].drained?.() ?? Promise.resolve(); + host.emit("close"); + + await expect(waiting).resolves.toBeUndefined(); + for (const event of LISTENED) { + expect(host.listenerCount(event)).toBe(0); + } + }); + + test("rejects when the host stream fails", async () => { + const streams = hostStreams(); + const host = streams[name]; + const runtime = await runtimeWith(streams); + + const waiting = runtime[name].drained?.() ?? Promise.resolve(); + host.emit("error", new Error("EPIPE")); + + await expect(waiting).rejects.toThrow("EPIPE"); + for (const event of LISTENED) { + expect(host.listenerCount(event)).toBe(0); + } + }); + + test("resolves at once when the host stream does not need to drain", async () => { + const streams = hostStreams(); + streams[name].writableNeedDrain = false; + const runtime = await runtimeWith(streams); + + await expect(runtime[name].drained?.()).resolves.toBeUndefined(); + expect(streams[name].listenerCount("drain")).toBe(0); + }); +}); diff --git a/packages/cli/tests/spawn-adapter.test.ts b/packages/cli/tests/spawn-adapter.test.ts index 05a09a1c..a403b7f2 100644 --- a/packages/cli/tests/spawn-adapter.test.ts +++ b/packages/cli/tests/spawn-adapter.test.ts @@ -46,6 +46,19 @@ beforeEach(() => { diagnosticText = ""; }); +async function readAll( + source: AsyncIterable | undefined, +): Promise { + if (source === undefined) { + throw new Error("the adapter returned no piped stream"); + } + let text = ""; + for await (const chunk of source) { + text += chunk; + } + return text; +} + async function waitForFile(path: string, timeoutMs = 10_000): Promise { const deadline = Date.now() + timeoutMs; while (!existsSync(path)) { @@ -335,4 +348,59 @@ describe("the shipped spawn adapter", () => { ).toBe(true); } }); + + test("engine mode returns both decoded streams to the engine and writes nothing itself", async () => { + const child = spawnChild({ + command: NODE, + args: [ + "-e", + "process.stdout.write(Buffer.from([0xf0, 0x9f])); setTimeout(() => { process.stdout.write(Buffer.from([0x98, 0x80])); process.stderr.write('child-err'); process.exit(5); }, 50)", + ], + cwd: process.cwd(), + env: process.env, + output: "engine", + }); + + const [stdout, stderr, result] = await Promise.all([ + readAll(child.stdout), + readAll(child.stderr), + child.ended, + ]); + + expect(spawnOptionsSeen.at(-1)?.stdio).toEqual(["inherit", "pipe", "pipe"]); + expect(result).toEqual({ exitCode: 5, signal: null }); + expect(stdout).toBe("😀"); + expect(stderr).toBe("child-err"); + expect(diagnosticText).toBe(""); + }); + + test.skipIf(process.platform === "win32")( + "engine mode: a grandchild holding the pipes does not block settlement past the drain grace", + async () => { + const gracedSpawn = makeSpawnChild( + { write: () => {} }, + { drainGraceMs: 200 }, + ); + const startedAt = Date.now(); + const child = gracedSpawn({ + command: NODE, + args: [ + "-e", + `require("node:child_process").spawn(process.execPath, ["-e", "setTimeout(() => {}, 10000)"], { detached: true, stdio: ["ignore", "inherit", "ignore"] }).unref();`, + ], + cwd: process.cwd(), + env: process.env, + output: "engine", + }); + const reading = Promise.all([ + readAll(child.stdout).catch(() => ""), + readAll(child.stderr).catch(() => ""), + ]); + + await expect(child.ended).resolves.toEqual({ exitCode: 0, signal: null }); + await reading; + expect(Date.now() - startedAt).toBeLessThan(3_000); + }, + 20_000, + ); }); diff --git a/packages/cli/tests/spawn-engine-output.test.ts b/packages/cli/tests/spawn-engine-output.test.ts new file mode 100644 index 00000000..ad0d18d3 --- /dev/null +++ b/packages/cli/tests/spawn-engine-output.test.ts @@ -0,0 +1,211 @@ +/** + * The engine and the SHIPPED spawn adapter together, with a real child, + * when the command transforms the child's output. + */ +import { + type ChildStreamName, + defineCommand, + exitWithChildStatus, + type SpawnOptions, +} from "@prisma/cli-engine"; +import { ok } from "@prisma/cli-engine/protocol"; +import { createTestCli } from "@prisma/cli-engine/testing"; +import { describe, expect, test } from "vitest"; +import { makeSpawnChild } from "../src/spawn"; + +const NODE = process.execPath; +const READY_THEN_TICKS = /^ready\n(tick\n)*$/; + +function redacting(secret: string) { + return (_stream: ChildStreamName) => { + let held = ""; + return new TransformStream({ + transform: (chunk, controller) => { + const text = (held + chunk).replaceAll(secret, "[redacted]"); + let keep = Math.min(text.length, secret.length - 1); + while (keep > 0 && !text.endsWith(secret.slice(0, keep))) { + keep -= 1; + } + held = text.slice(text.length - keep); + if (text.length > keep) { + controller.enqueue(text.slice(0, text.length - keep)); + } + }, + flush: (controller) => { + if (held !== "") { + controller.enqueue(held); + } + }, + }); + }; +} + +function nodeCommand( + script: string, + transformOutput: SpawnOptions["transformOutput"], + onError?: (cause: unknown) => void, +) { + return defineCommand({ + help: { summary: "Runs a node child and transforms its output" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: NODE, + args: ["-e", script], + transformOutput, + }); + } catch (cause) { + onError?.(cause); + } + return ok(exitWithChildStatus()); + }, + }); +} + +function cliFor(command: ReturnType) { + return createTestCli({ + commands: { converge: command }, + spawn: makeSpawnChild({ write: () => {} }, { drainGraceMs: 200 }), + }); +} + +const CHATTER = ` +const pause = () => new Promise((resolve) => setTimeout(resolve, 30)); +(async () => { + process.stdout.write("out-1 hunt"); + await pause(); + process.stderr.write("err-1\\n"); + await pause(); + process.stdout.write("er2\\nout-2\\n"); + await pause(); + process.exitCode = 4; +})(); +`; + +describe.skipIf(process.platform === "win32")( + "transformed output through the shipped spawn adapter", + () => { + test("each stream is transformed in order and written where the child would have written it", async () => { + const cli = cliFor(nodeCommand(CHATTER, redacting("hunter2"))); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.spawns[0]?.output).toBe("engine"); + expect(result.exitCode).toBe(4); + expect(result.stdout).toBe("out-1 [redacted]\nout-2\n"); + expect(result.stderr).toBe("err-1\n"); + }); + + test("json mode writes both streams to stderr", async () => { + const cli = cliFor(nodeCommand(CHATTER, redacting("hunter2"))); + + const result = await cli.run(["converge", "--json"]); + + expect(result.json).toHaveLength(1); + expect(result.stderr.replace("err-1\n", "")).toBe( + "out-1 [redacted]\nout-2\n", + ); + expect(result.stderr).toContain("err-1\n"); + }); + + test("a SIGTERM ends the child, and its output written so far still arrives", async () => { + const controller = new AbortController(); + const abortOnReady = (_stream: ChildStreamName) => + new TransformStream({ + transform: (chunk, transformController) => { + transformController.enqueue(chunk); + if (chunk.includes("ready\n")) { + controller.abort("SIGTERM"); + } + }, + }); + const script = + "process.stdout.write('ready\\n'); setInterval(() => process.stdout.write('tick\\n'), 10);"; + const cli = cliFor(nodeCommand(script, abortOnReady)); + + const result = await cli.run(["converge"], { abort: controller.signal }); + + expect(result.spawns[0]?.kills[0]).toBe("SIGTERM"); + expect(result.exitCode).toBe(143); + expect(result.stdout).toMatch(READY_THEN_TICKS); + }); + + test("a throwing transform ends a child that ignores SIGTERM, and ctx.spawn rejects with its error", async () => { + let caught: unknown; + const script = ` +process.on("SIGTERM", () => {}); +process.stdout.write("ready\\n"); +setInterval(() => process.stdout.write("tick\\n"), 10); +`; + const cli = cliFor( + nodeCommand( + script, + () => + new TransformStream({ + transform: (chunk, controller) => { + if (chunk.includes("tick")) { + throw new Error("redaction failed"); + } + controller.enqueue(chunk); + }, + }), + (cause) => { + caught = cause; + }, + ), + ); + + const result = await cli.run(["converge", "--format", "human"]); + + expect((caught as Error).message).toBe("redaction failed"); + expect(result.spawns[0]?.kills).toEqual(["SIGTERM", "SIGKILL"]); + expect(result.stdout).toBe("ready\n"); + expect(result.exitCode).toBe(137); + }); + + test("output dropped at the drain grace leaves a debug line", async () => { + const script = `require("node:child_process").spawn(process.execPath, ["-e", "setTimeout(() => {}, 10000)"], { detached: true, stdio: ["ignore", "inherit", "ignore"] }).unref(); process.stdout.write("done\\n");`; + const cli = cliFor(nodeCommand(script, redacting("hunter2"))); + + const result = await cli.run(["converge", "--format", "human"], { + env: { PRISMA_DEBUG: "1" }, + }); + + expect(result.exitCode).toBe(0); + expect(result.stdout).toBe("done\n"); + expect(result.stderr).toContain( + "spawn: the child's stdout failed, so its later output was dropped: the pipe stayed open with no output for 200ms after the child exited, so it was closed", + ); + }); + + test("a slow reader still gets the child's last output after the drain grace", async () => { + const script = `process.stdout.write("x".repeat(300_000) + "LAST-LINE\\n");`; + const child = makeSpawnChild( + { write: () => {} }, + { drainGraceMs: 200 }, + )({ + command: NODE, + args: ["-e", script], + cwd: process.cwd(), + env: process.env, + output: "engine", + }); + let received = ""; + for await (const chunk of child.stdout ?? []) { + received += chunk; + await new Promise((resolve) => setTimeout(resolve, 150)); + } + let errors = ""; + for await (const chunk of child.stderr ?? []) { + errors += chunk; + } + + expect(await child.ended).toEqual({ exitCode: 0, signal: null }); + expect(errors).toBe(""); + expect(received.length).toBe(300_000 + "LAST-LINE\n".length); + expect(received.endsWith("LAST-LINE\n")).toBe(true); + }); + }, + 20_000, +); diff --git a/packages/prisma/package.json b/packages/prisma/package.json index 3af55153..eeafc3a0 100644 --- a/packages/prisma/package.json +++ b/packages/prisma/package.json @@ -50,7 +50,7 @@ }, "dependencies": { "@manypkg/tools": "^2.1.2", - "@prisma/cli-engine": "workspace:0.9.0", + "@prisma/cli-engine": "workspace:0.10.0", "@prisma/composer-cli": "0.36.0", "@prisma/compute-sdk": "0.43.0", "@prisma/management-api-sdk": "1.80.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9688e7e4..15d790ba 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -27,7 +27,7 @@ importers: specifier: ^2.1.2 version: 2.1.2 '@prisma/cli-engine': - specifier: workspace:0.9.0 + specifier: workspace:0.10.0 version: link:../cli-engine '@prisma/composer-cli': specifier: 0.36.0 @@ -210,7 +210,7 @@ importers: specifier: ^2.1.2 version: 2.1.2 '@prisma/cli-engine': - specifier: workspace:0.9.0 + specifier: workspace:0.10.0 version: link:../cli-engine '@prisma/composer-cli': specifier: 0.36.0