From 2b8ebae73bab21bc908c7b13ee2c576441c201ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Richard=20Sol=C3=A1r?= Date: Fri, 14 Aug 2026 12:55:13 +0200 Subject: [PATCH 1/2] fix(cli): read stdin lazily so open pipes don't hang the CLI (#1206) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every command paid for stdin: `_shared.ts` did `await readStdin()` at module scope, so `apify --version` blocked on a stream it never reads. With a named pipe (no wait deadline, unlike the socket a spawned child gets) the read never ended and the command never even ran. Read stdin only when a command asks for it — a `-` arg/flag, `apify run` without `--input`, `push-data`/`push-items` without an item. Result is memoized, since stdin can only be drained once. Confirmations previously keyed off `hasData`, which the eager read flipped to false once stdin was drained. Without that read the flag stays true, so `apify < /dev/null` would try to prompt on a non-TTY; gate on `isTTY` instead. Commands that genuinely consume stdin still wait for the writer to close, same as `cat`. Co-Authored-By: Claude Opus 5 --- src/commands/actor/push-data.ts | 4 +-- src/commands/datasets/push-items.ts | 4 +-- src/entrypoints/_shared.ts | 3 -- src/lib/command-framework/CommandError.ts | 4 +-- src/lib/command-framework/apify-command.ts | 18 +++++----- src/lib/commands/read-stdin.ts | 34 +++++++++++++----- src/lib/commands/resolve-input.ts | 4 +-- .../user-confirmations/_stdinCheckWrapper.ts | 6 ++-- test/e2e/commands/stdin-held-open.test.ts | 35 ++++++++++++++++++- 9 files changed, 81 insertions(+), 31 deletions(-) diff --git a/src/commands/actor/push-data.ts b/src/commands/actor/push-data.ts index b6ae01d57..a36dac65b 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,7 @@ export class ActorPushDataCommand extends ApifyCommand | Record[]; - const item = _item || cachedStdinInput; + 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..3d3e66d00 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 { peekStdin } 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? ${peekStdin() ? 'Yes' : 'No'}`, ].join('\n'); } } diff --git a/src/lib/command-framework/apify-command.ts b/src/lib/command-framework/apify-command.ts index cfede72ad..d2f66acda 100644 --- a/src/lib/command-framework/apify-command.ts +++ b/src/lib/command-framework/apify-command.ts @@ -9,7 +9,7 @@ import indentString from 'indent-string'; import widestLine from 'widest-line'; import wrapAnsi from 'wrap-ansi'; -import { cachedStdinInput } from '../../entrypoints/_shared.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'; @@ -368,7 +368,7 @@ export abstract class ApifyCommand | undefined; +let readResult: Buffer | undefined; + +/** + * Reads stdin to its end, at most once per process. Callers must only call this when the command + * actually wants stdin data — a pipe that stays open never ends, so this waits for as long as the + * writer keeps it open (#1206). + */ export async function readStdin() { - const dataRef = await useStdin(); + readPromise ??= _readStdin().then((data) => { + readResult = data; + return data; + }); - const { hasData, waitDelay, stream } = dataRef; + return readPromise; +} + +/** + * Stdin data read so far, without triggering a read. For diagnostics only. + */ +export function peekStdin() { + return readResult; +} + +async function _readStdin() { + const { hasData, waitDelay, stream } = await useStdin(); if (!hasData) { return; @@ -45,10 +67,7 @@ export async function readStdin() { } } 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. + // the CLI) alive after the command finishes (#1206). stream.off('data', onData); stream.pause(); } @@ -57,9 +76,6 @@ export async function readStdin() { clearTimeout(timeout); } - // Mark further uses of useStdin / readStdin as having no more data since we've read it all - dataRef.hasData = false; - const concat = Buffer.concat(bufferChunks); if (concat.length) { diff --git a/src/lib/commands/resolve-input.ts b/src/lib/commands/resolve-input.ts index e849e1f95..26b438e56 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 { getLocalInput } from '../utils.js'; +import { readStdin } from './read-stdin.js'; interface InputOverrideOptions { schemaHint?: string; @@ -59,7 +59,7 @@ export async function getInputOverride( if (!inputFlag && !inputFileFlag) { // Try reading stdin - const stdin = cachedStdinInput; + const stdin = await readStdin(); if (stdin) { try { diff --git a/src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts b/src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts index dd2db6223..e8a795258 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 — before stdin became lazy (#1206) it was always drained by then anyway. + 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..29bc32b78 100644 --- a/test/e2e/commands/stdin-held-open.test.ts +++ b/test/e2e/commands/stdin-held-open.test.ts @@ -1,5 +1,6 @@ -import { mkdir, rm } from 'node:fs/promises'; +import { mkdir, open, rm } from 'node:fs/promises'; import path from 'node:path'; +import process from 'node:process'; import { fileURLToPath } from 'node:url'; import { execa } from 'execa'; @@ -55,4 +56,36 @@ describe('[e2e] stdin held open (#1206)', () => { expect(result.exitCode).toBe(1); expect(result.stderr).toContain('Actor is of an unknown format'); }); + + // A named pipe is the harsher case: unlike the socket a spawned child gets, it has no wait + // deadline, so the old eager startup read blocked forever and the command never even ran. + // Opening the FIFO read-write keeps a writer attached, so it never reaches EOF, with no + // second process to manage. 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: { + APIFY_CLI_DISABLE_TELEMETRY: '1', + APIFY_CLI_SKIP_UPDATE_CHECK: '1', + APIFY_DISABLE_KEYRING: '1', + }, + }); + + 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 }); + } + }); }); From c76a6167d8f8890df05b822ea66d63e7c6245312 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Richard=20Sol=C3=A1r?= Date: Thu, 3 Sep 2026 23:15:15 +0200 Subject: [PATCH 2/2] fix(cli): stop implicit stdin reads from hanging on an open pipe `apify run` without `--input` falls back to stdin, and waited there for an end that a named pipe never reaches. A pipe the process merely inherited hung the command forever, and so did a writer that sent data and kept the pipe open. `actors call` and `actors start` share the path. Reads that nobody asked for now stop at the first quiet gap: each chunk restarts a 2s clock, and when stdin goes silent the CLI uses what arrived. An explicit `-` is unchanged and still waits for the writer to close. The cost is a writer slower than ~2.5s to its first byte, whose input is now ignored. Truncation mid-stream stays loud, as the JSON fails to parse. Co-Authored-By: Claude Opus 5 --- src/lib/commands/read-stdin.ts | 74 ++++++++++++++------ src/lib/commands/resolve-input.ts | 5 +- test/e2e/commands/stdin-held-open.test.ts | 84 ++++++++++++++++++++++- 3 files changed, 140 insertions(+), 23 deletions(-) diff --git a/src/lib/commands/read-stdin.ts b/src/lib/commands/read-stdin.ts index 318ebaf8c..a8431e140 100644 --- a/src/lib/commands/read-stdin.ts +++ b/src/lib/commands/read-stdin.ts @@ -2,16 +2,32 @@ import { once } from 'node:events'; import { useStdin } from '../hooks/useStdin.js'; +/** + * How long an implicit read waits for stdin to say something. Long enough for a writer that has to + * fetch or compute its first bytes, short enough that a pipe with nothing behind it does not look + * like a hang. + */ +const IMPLICIT_STDIN_IDLE_TIMEOUT_MILLIS = 2_000; + let readPromise: Promise | undefined; let readResult: Buffer | undefined; +export 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 to its end, at most once per process. Callers must only call this when the command - * actually wants stdin data — a pipe that stays open never ends, so this waits for as long as the - * writer keeps it open (#1206). + * 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() { - readPromise ??= _readStdin().then((data) => { +export async function readStdin(options: ReadStdinOptions = {}) { + readPromise ??= _readStdin(options).then((data) => { readResult = data; return data; }); @@ -20,41 +36,59 @@ export async function readStdin() { } /** - * Stdin data read so far, without triggering a read. For diagnostics only. + * Stdin data from a completed read, without starting one. Undefined until a read finishes. For + * diagnostics only. */ export function peekStdin() { return readResult; } -async function _readStdin() { +async function _readStdin({ implicit }: ReadStdinOptions) { const { hasData, waitDelay, stream } = await useStdin(); if (!hasData) { return; } + // `waitDelay` guards the first byte on a socket, and nothing else. An implicit read needs more + // than that: an inherited pipe may send nothing at all, and may stay open after it does, so + // waiting for the end of the stream never finishes (#1206). + const idleTimeout = implicit ? 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 = () => { + if (idleTimeout) { + timeout = setTimeout(() => controller.abort(), idleTimeout).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(); + + const onData = (chunk: Buffer) => { + bufferChunks.push(chunk); + + disarmTimeout(); + + // An implicit read has no other way to tell that the writer is done, so every chunk restarts + // the clock. An explicit one waits for the real end of the stream, and its deadline only ever + // guarded the first byte. + if (implicit) { + armTimeout(); + } + }; + stream.on('data', onData); try { @@ -62,7 +96,9 @@ async function _readStdin() { } catch (error) { const casted = error as Error; - if (casted.name === 'AbortError') { + // 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 (casted.name === 'AbortError' && !implicit) { return; } } finally { @@ -72,9 +108,7 @@ async function _readStdin() { stream.pause(); } - if (timeout) { - clearTimeout(timeout); - } + disarmTimeout(); const concat = Buffer.concat(bufferChunks); diff --git a/src/lib/commands/resolve-input.ts b/src/lib/commands/resolve-input.ts index 26b438e56..af5d965c6 100644 --- a/src/lib/commands/resolve-input.ts +++ b/src/lib/commands/resolve-input.ts @@ -58,8 +58,9 @@ export async function getInputOverride( const { schemaHint } = options; if (!inputFlag && !inputFileFlag) { - // Try reading stdin - const stdin = await readStdin(); + // 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 }); if (stdin) { try { diff --git a/test/e2e/commands/stdin-held-open.test.ts b/test/e2e/commands/stdin-held-open.test.ts index 29bc32b78..c0410d1a8 100644 --- a/test/e2e/commands/stdin-held-open.test.ts +++ b/test/e2e/commands/stdin-held-open.test.ts @@ -1,4 +1,4 @@ -import { mkdir, open, rm } from 'node:fs/promises'; +import { mkdir, open, readFile, rm, writeFile } from 'node:fs/promises'; import path from 'node:path'; import process from 'node:process'; import { fileURLToPath } from 'node:url'; @@ -88,4 +88,86 @@ describe('[e2e] stdin held open (#1206)', () => { await rm(fifo, { force: true }); } }); + + // The tests above stop at the project check, ~100 lines before `apify run` resolves its input. + // A real project gets that far, and there `run` reads stdin when `--input` is missing — 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 the CLI handed over, so the test can read it back from disk. Going through + // a file rather than stdout keeps this independent of how the package manager forwards output. + 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 for the whole test, so the pipe never reaches EOF even after + // the data below is written. + 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: { + APIFY_CLI_DISABLE_TELEMETRY: '1', + APIFY_CLI_SKIP_UPDATE_CHECK: '1', + APIFY_CLI_SKIP_RENTAL_SUNSET_NOTICE: '1', + APIFY_DISABLE_KEYRING: '1', + }, + }); + } 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(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 }); + }); });