diff --git a/src/commands/actor/push-data.ts b/src/commands/actor/push-data.ts index b6ae01d57..01368a1fc 100644 --- a/src/commands/actor/push-data.ts +++ b/src/commands/actor/push-data.ts @@ -1,7 +1,7 @@ -import { cachedStdinInput } from '../../entrypoints/_shared.js'; import { APIFY_STORAGE_TYPES, getApifyStorageClient, getDefaultStorageId } from '../../lib/actor.js'; import { ApifyCommand } from '../../lib/command-framework/apify-command.js'; import { Args } from '../../lib/command-framework/args.js'; +import { readStdin } from '../../lib/commands/read-stdin.js'; import { error } from '../../lib/outputs.js'; export class ActorPushDataCommand extends ApifyCommand { @@ -34,7 +34,8 @@ export class ActorPushDataCommand extends ApifyCommand | Record[]; - const item = _item || cachedStdinInput; + // Piping items in is the documented way to use this command, so wait for the writer to close. + const item = _item || (await readStdin()); if (!item) { error({ message: 'No items were provided.' }); diff --git a/src/entrypoints/_shared.ts b/src/entrypoints/_shared.ts index 363801f71..a48884133 100644 --- a/src/entrypoints/_shared.ts +++ b/src/entrypoints/_shared.ts @@ -11,7 +11,6 @@ import type { BuiltApifyCommand } from '../lib/command-framework/apify-command.j import { commandRegistry, internalRunCommand } from '../lib/command-framework/apify-command.js'; import { CommandError } from '../lib/command-framework/CommandError.js'; import { renderMainHelpMenu } from '../lib/command-framework/help.js'; -import { readStdin } from '../lib/commands/read-stdin.js'; import { SUPPORTED_NODEJS_VERSION } from '../lib/consts.js'; import { useCLIMetadata } from '../lib/hooks/useCLIMetadata.js'; import { shouldSkipVersionCheck } from '../lib/hooks/useCLIVersionCheck.js'; @@ -19,8 +18,6 @@ import { useCommandSuggestions } from '../lib/hooks/useCommandSuggestions.js'; import { error } from '../lib/outputs.js'; import { cliDebugPrint } from '../lib/utils/cliDebugPrint.js'; -export const cachedStdinInput = await readStdin(); - const cliMetadata = useCLIMetadata(); export const USER_AGENT = `Apify CLI/${cliMetadata.version} (https://github.com/apify/apify-cli)`; diff --git a/src/lib/command-framework/CommandError.ts b/src/lib/command-framework/CommandError.ts index 1c8e55a44..fe23fab91 100644 --- a/src/lib/command-framework/CommandError.ts +++ b/src/lib/command-framework/CommandError.ts @@ -1,6 +1,6 @@ import chalk from 'chalk'; -import { cachedStdinInput } from '../../entrypoints/_shared.js'; +import { describeStdinRead } from '../commands/read-stdin.js'; import { useCLIMetadata } from '../hooks/useCLIMetadata.js'; import type { BuiltApifyCommand } from './apify-command.js'; import { selectiveRenderHelpForCommand } from './help.js'; @@ -225,7 +225,7 @@ export class CommandError extends Error { '', `- CLI version: \`${cliMetadata.fullVersionString}\``, `- CLI debug logs (process.env.APIFY_CLI_DEBUG): ${process.env.APIFY_CLI_DEBUG ? 'Enabled' : 'Disabled'}`, - `- Stdin data? ${cachedStdinInput ? 'Yes' : 'No'}`, + `- Stdin data? ${describeStdinRead()}`, ].join('\n'); } } diff --git a/src/lib/command-framework/apify-command.ts b/src/lib/command-framework/apify-command.ts index f6b94f60e..e22cc444a 100644 --- a/src/lib/command-framework/apify-command.ts +++ b/src/lib/command-framework/apify-command.ts @@ -9,8 +9,8 @@ import indentString from 'indent-string'; import widestLine from 'widest-line'; import wrapAnsi from 'wrap-ansi'; -import { cachedStdinInput } from '../../entrypoints/_shared.js'; import { selectProfile } from '../auth.js'; +import { readStdin } from '../commands/read-stdin.js'; import { keepStdoutClean } from '../exec.js'; import { detectAiAgent, detectCi, detectIsInteractive } from '../hooks/telemetry/detectEnvironment.js'; import type { TrackEventMap } from '../hooks/telemetry/trackEvent.js'; @@ -378,7 +378,7 @@ export abstract class ApifyCommand | undefined; +let readResult: Buffer | undefined; +let readFinished = false; +let readCutShort = false; + +interface ReadStdinOptions { + /** + * Stop at the first quiet gap instead of waiting for the writer to close stdin. Set it where + * stdin is a fallback the user never asked for, so the command cannot hang on a pipe it merely + * inherited (#1206). Leave it off for an explicit `-`, which waits for as long as the writer + * wants, the way `cat` does. + */ + implicit?: boolean; +} + +/** + * Reads stdin, at most once per process. Call it only when the command actually wants stdin data. + * The first call decides the options; later ones reuse its result. + */ +export async function readStdin(options: ReadStdinOptions = {}) { + readPromise ??= _readStdin(options).then((data) => { + readResult = data; + readFinished = true; + return data; + }); + + return readPromise; +} - const { hasData, waitDelay, stream } = dataRef; +/** + * Whether an implicit read gave up before the writer closed stdin. Explains both a payload that + * ends where nothing should end and a read that came back with nothing. + */ +export function stdinWasCutShort() { + return readCutShort; +} + +/** + * What a finished read found, for the bug report footer. Says `Not read` while no command has asked + * for stdin, which is most of them. + */ +export function describeStdinRead() { + if (!readFinished) { + return 'Not read'; + } + + return readResult ? 'Yes' : 'No'; +} + +async function _readStdin({ implicit }: ReadStdinOptions) { + const { hasData, waitDelay, stream } = await useStdin(); if (!hasData) { return; } + // An inherited pipe may send nothing and stay open, so waiting for the end never finishes + // (#1206). Keep a shorter platform deadline where there is one: it speeds up every spawned CLI. + const firstByteTimeout = implicit ? waitDelay || IMPLICIT_STDIN_IDLE_TIMEOUT_MILLIS : waitDelay; + const bufferChunks: Buffer[] = []; const controller = new AbortController(); let timeout: NodeJS.Timeout | null = null; - if (waitDelay) { - timeout = setTimeout(() => { - controller.abort(); - }, waitDelay).unref(); - } - - const onData = (chunk: Buffer) => { - bufferChunks.push(chunk); + const armTimeout = (delay: number) => { + if (delay) { + timeout = setTimeout(() => controller.abort(), delay).unref(); + } + }; - // If we got some data already, we can clear the timeout, as we will get more + const disarmTimeout = () => { if (timeout) { clearTimeout(timeout); timeout = null; } }; + armTimeout(firstByteTimeout); + + const onData = (chunk: Buffer) => { + bufferChunks.push(chunk); + + disarmTimeout(); + + // An implicit read has no other way to tell the writer is done, so every chunk restarts the + // clock. An explicit one waits for the real end of the stream. + if (implicit) { + armTimeout(IMPLICIT_STDIN_IDLE_TIMEOUT_MILLIS); + } + }; + stream.on('data', onData); try { @@ -41,24 +106,23 @@ export async function readStdin() { const casted = error as Error; if (casted.name === 'AbortError') { - return; + // An explicit read that runs out its deadline saw nothing at all, so it has nothing to give + // back. An implicit one keeps whatever arrived before stdin went quiet. + if (!implicit) { + return; + } + + // A socket's short first-byte guard fires for every spawned CLI, so only a read that + // outwaited the idle timeout counts as given up on. + readCutShort = bufferChunks.length > 0 || firstByteTimeout === IMPLICIT_STDIN_IDLE_TIMEOUT_MILLIS; } } finally { - // Stop reading from stdin so its open handle can't keep the event loop (and - // the CLI) alive after the command finishes (#1206). This only helps when the - // await above settles ('end' or the no-data abort). A writer that sends data - // but never closes stdin still hangs up there; that needs the lazy stdin - // reading discussed in #1206. + // Stop reading so an open stdin handle can't keep the CLI alive after the command (#1206). stream.off('data', onData); stream.pause(); } - if (timeout) { - clearTimeout(timeout); - } - - // Mark further uses of useStdin / readStdin as having no more data since we've read it all - dataRef.hasData = false; + disarmTimeout(); const concat = Buffer.concat(bufferChunks); diff --git a/src/lib/commands/resolve-input.ts b/src/lib/commands/resolve-input.ts index e849e1f95..970e58f9f 100644 --- a/src/lib/commands/resolve-input.ts +++ b/src/lib/commands/resolve-input.ts @@ -4,10 +4,10 @@ import process from 'node:process'; import mime from 'mime'; -import { cachedStdinInput } from '../../entrypoints/_shared.js'; import { CommandExitCodes } from '../consts.js'; -import { error } from '../outputs.js'; +import { error, warning } from '../outputs.js'; import { getLocalInput } from '../utils.js'; +import { readStdin, stdinWasCutShort } from './read-stdin.js'; interface InputOverrideOptions { schemaHint?: string; @@ -58,8 +58,18 @@ export async function getInputOverride( const { schemaHint } = options; if (!inputFlag && !inputFileFlag) { - // Try reading stdin - const stdin = cachedStdinInput; + // Nobody asked for stdin here, so it must not block: this command is reachable with a pipe + // it only inherited from whatever spawned it. + const stdin = await readStdin({ implicit: true }); + + // Giving up on a silent pipe is the only way not to hang on one, but a writer that was just + // slow would otherwise see its input vanish with nothing said. + if (!stdin && stdinWasCutShort()) { + warning({ + message: + 'Nothing arrived on standard input, so it was ignored. Use `--input-file=-` to wait for a slow writer.', + }); + } if (stdin) { try { @@ -76,9 +86,15 @@ export async function getInputOverride( input = parsed; source = 'stdin'; } catch (err) { + // Standard input is only waited on for as long as it keeps talking, so a writer that stalls + // mid-payload leaves behind JSON that ends nowhere. Say so, or the message blames the data. + const cutShortHint = stdinWasCutShort() + ? '\nStandard input went quiet before it closed, so only part of it was read. Use `--input-file=-` to wait for all of it.' + : ''; + error({ message: withSchemaHint( - `Cannot parse JSON input from standard input.\n ${(err as Error).message}`, + `Cannot parse JSON input from standard input.\n ${(err as Error).message}${cutShortHint}`, schemaHint, ), }); diff --git a/src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts b/src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts index dd2db6223..875b785c5 100644 --- a/src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts +++ b/src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts @@ -47,11 +47,13 @@ export function stdinCheckWrapper any>( }: StdinCheckWrapperOptions = {}, ): (...args: NewFunctionArgs) => Promise>> { return async (input, ...rest) => { - const { isTTY, hasData } = await useStdin(); + const { isTTY } = await useStdin(); const casted = input as StdinCheckWrapperInput>>; - if (isCI || (!isTTY && !hasData)) { + // Prompts need a terminal to read the answer from. Piped stdin is command input, not an + // answer source. + if (isCI || !isTTY) { if (typeof casted.providedConfirmFromStdin === 'undefined') { throw new Error(casted.errorMessageForStdin ?? errorMessageForStdin); } diff --git a/test/e2e/commands/stdin-held-open.test.ts b/test/e2e/commands/stdin-held-open.test.ts index 11d58db37..8d0a7c2d9 100644 --- a/test/e2e/commands/stdin-held-open.test.ts +++ b/test/e2e/commands/stdin-held-open.test.ts @@ -1,16 +1,13 @@ -import { mkdir, rm } from 'node:fs/promises'; +import { mkdir, open, readdir, readFile, rm, writeFile } from 'node:fs/promises'; import path from 'node:path'; -import { fileURLToPath } from 'node:url'; +import process from 'node:process'; import { execa } from 'execa'; +import { CliTestEnv, DistActor, DistApify } from '../__helpers__/run-cli.js'; import { TestTmpRoot } from '../__helpers__/tmp.js'; -const DistApify = fileURLToPath(new URL('../../../dist/apify.js', import.meta.url)); - -// How long the CLI gets to exit on its own. The command itself finishes in well -// under a second; the broken behavior never exits at all, so any finite deadline -// separates the two. Generous to absorb slow CI runners. +// The broken behavior never exits, so any finite deadline separates it. Generous for slow CI. const EXIT_DEADLINE_MS = 15_000; describe('[e2e] stdin held open (#1206)', () => { @@ -26,33 +23,163 @@ describe('[e2e] stdin held open (#1206)', () => { }); it('exits on its own when stdin is a pipe that never closes', async () => { - // The runCli helper cannot express this scenario: it either ignores stdin or - // writes input and closes it. execa with stdin: 'pipe' and no `input` keeps - // the child's stdin open for the child's whole lifetime (verified on execa 9), - // which is what spawned subprocesses (CI runners, agent shells) see. Before - // the fix, the startup stdin read left process.stdin flowing with a data - // listener attached, so the CLI printed all its output but never exited. + // `stdin: 'pipe'` with no `input` keeps the child's stdin open for its whole lifetime, the + // way a CI runner or agent shell does. runCli cannot express that. const result = await execa('node', [DistApify, 'run'], { cwd: emptyDir, reject: false, timeout: EXIT_DEADLINE_MS, stdin: 'pipe', - env: { - APIFY_CLI_DISABLE_TELEMETRY: '1', - APIFY_CLI_SKIP_UPDATE_CHECK: '1', - APIFY_DISABLE_KEYRING: '1', - }, + env: CliTestEnv, }); - // `timedOut` is the regression signal: the broken build completes the command - // (same stderr) but never exits, so execa kills it at the deadline. + // The broken build prints the same stderr but never exits. Keep this on `apify run`: + // --version and --help call process.exit(), which would mask a leaked stdin handle. expect(result.timedOut, `stderr: ${result.stderr}`).toBe(false); - - // Sanity: the command really ran and failed naturally (an empty dir is not an - // actor project). Keep this on `apify run`: --version, --help, and unknown - // commands all call process.exit(), which exits even with a leaked stdin - // handle and would mask the regression. expect(result.exitCode).toBe(1); expect(result.stderr).toContain('Actor is of an unknown format'); }); + + // A named pipe has no wait deadline, unlike the socket a spawned child gets, so the old eager + // read blocked forever. Opening it `r+` keeps a writer attached. Windows has no mkfifo. + it.skipIf(process.platform === 'win32')('exits on its own when stdin is a named pipe with no writer', async () => { + const fifo = path.join(emptyDir, 'stdin.fifo'); + await execa('mkfifo', [fifo]); + + const handle = await open(fifo, 'r+'); + + try { + const result = await execa('node', [DistApify, 'run'], { + cwd: emptyDir, + reject: false, + timeout: EXIT_DEADLINE_MS, + stdin: handle.fd, + env: CliTestEnv, + }); + + expect(result.timedOut, `stderr: ${result.stderr}`).toBe(false); + expect(result.exitCode).toBe(1); + expect(result.stderr).toContain('Actor is of an unknown format'); + } finally { + await handle.close(); + await rm(fifo, { force: true }); + } + }); + + // The tests above stop at the project check. A real project reaches `run`'s input resolution, + // the last path that still hung on a named pipe. + const createFifoActor = async (name: string) => { + const dir = path.join(emptyDir, name); + + await mkdir(path.join(dir, '.actor'), { recursive: true }); + await writeFile( + path.join(dir, '.actor', 'actor.json'), + JSON.stringify({ actorSpecification: 1, name, version: '0.0', buildTag: 'latest' }), + ); + await writeFile( + path.join(dir, 'package.json'), + JSON.stringify({ name, version: '0.0.1', type: 'module', scripts: { start: 'node main.js' } }), + ); + // Records the input to a file, not stdout, so the package manager cannot garble it. + await writeFile( + path.join(dir, 'main.js'), + [ + "import { readFile, writeFile } from 'node:fs/promises';", + "import path from 'node:path';", + "import process from 'node:process';", + "const store = path.join(process.env.APIFY_LOCAL_STORAGE_DIR ?? 'storage', 'key_value_stores', 'default');", + "const key = process.env.ACTOR_INPUT_KEY ?? 'INPUT';", + "const input = await readFile(path.join(store, `${key}.json`), 'utf8').catch(() => 'null');", + "await writeFile('SEEN_INPUT.json', input);", + ].join('\n'), + ); + + return dir; + }; + + const runWithFifoStdin = async (cwd: string, write?: string) => { + const fifo = path.join(cwd, 'stdin.fifo'); + await execa('mkfifo', [fifo]); + + // `r+` keeps a writer attached, so the pipe never reaches EOF even after the write below. + const handle = await open(fifo, 'r+'); + + try { + if (write) { + await handle.write(write); + } + + return await execa('node', [DistApify, 'run'], { + cwd, + reject: false, + timeout: EXIT_DEADLINE_MS, + stdin: handle.fd, + env: CliTestEnv, + }); + } finally { + await handle.close(); + await rm(fifo, { force: true }); + } + }; + + it.skipIf(process.platform === 'win32')('runs a real project when its named pipe stays silent', async () => { + const actorDir = await createFifoActor('silent-pipe-actor'); + + const result = await runWithFifoStdin(actorDir); + + expect(result.timedOut, `stderr: ${result.stderr}`).toBe(false); + expect(result.stderr).toContain('Nothing arrived on standard input'); + expect(await readFile(path.join(actorDir, 'SEEN_INPUT.json'), 'utf8')).toBe('null'); + }); + + it.skipIf(process.platform === 'win32')('uses what a named pipe sent, then stops waiting for more', async () => { + const actorDir = await createFifoActor('talking-pipe-actor'); + + const result = await runWithFifoStdin(actorDir, '{"fromStdin":true}'); + + expect(result.timedOut, `stderr: ${result.stderr}`).toBe(false); + expect(JSON.parse(await readFile(path.join(actorDir, 'SEEN_INPUT.json'), 'utf8'))).toEqual({ fromStdin: true }); + }); +}); + +describe('[e2e] piped stdin with a slow producer', () => { + const slowDir = path.join(TestTmpRoot, 'stdin-slow-producer'); + + beforeAll(async () => { + await rm(slowDir, { recursive: true, force: true }); + await mkdir(slowDir, { recursive: true }); + }); + + afterAll(async () => { + await rm(slowDir, { recursive: true, force: true }); + }); + + // A producer that scrapes or queries before it prints outlives any idle timeout. + it.skipIf(process.platform === 'win32')('waits for a producer that takes seconds to print', async () => { + const storageDir = path.join(slowDir, 'storage'); + + // `push-data` writes straight to the default dataset without creating it first. + await mkdir(path.join(storageDir, 'datasets', 'default'), { recursive: true }); + + const result = await execa( + 'sh', + ['-c', `(sleep 3; printf '%s' '{"slow":true}') | node "$1" push-data`, 'sh', DistActor], + { + cwd: slowDir, + reject: false, + timeout: EXIT_DEADLINE_MS, + env: { ...CliTestEnv, APIFY_LOCAL_STORAGE_DIR: storageDir }, + }, + ); + + expect(result.stderr).not.toContain('No item was provided'); + expect(result.exitCode).toBe(0); + + const dataset = path.join(storageDir, 'datasets', 'default'); + const items = await Promise.all( + (await readdir(dataset)).map(async (file) => JSON.parse(await readFile(path.join(dataset, file), 'utf8'))), + ); + + expect(items).toEqual([{ slow: true }]); + }); });