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/4] 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/4] 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 }); + }); }); From e017c4ae859f56b7266a94c1316a1dc0ee44cf2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Richard=20Sol=C3=A1r?= Date: Wed, 23 Sep 2026 22:44:24 +0200 Subject: [PATCH 3/4] fix(cli): keep the socket stdin fast path, stop push-data hanging The implicit read replaced `waitDelay` with its own 2s deadline, so a spawned child's socket stdin lost the 50ms first-byte guard. Measured on `apify run` with a silent socket: 703ms before, 2637ms after. Keep `waitDelay` as the first-byte deadline where stdin has one, and use the 2s constant for the idle gap and for stdin that has no deadline at all. `push-data` and `push-items` fall back to stdin the same way and still hung on a silent named pipe. They now use the same deadline: 2.4s to `No item was provided.` instead of waiting forever. A writer that stalls mid-payload leaves JSON that ends nowhere, and the parse error blamed the data. `getInputOverride` now says stdin went quiet and points at `--input-file=-`. `peekStdin()` fed the bug report footer from a read that had not happened yet, so it always printed `No`. Replaced with `describeStdinRead()`, which distinguishes `Not read`. Co-Authored-By: Claude Opus 5 --- src/commands/actor/push-data.ts | 4 +- src/commands/datasets/push-items.ts | 4 +- src/lib/command-framework/CommandError.ts | 4 +- src/lib/commands/read-stdin.ts | 60 ++++++++++++------- src/lib/commands/resolve-input.ts | 10 +++- .../user-confirmations/_stdinCheckWrapper.ts | 2 +- test/e2e/commands/stdin-held-open.test.ts | 6 +- 7 files changed, 60 insertions(+), 30 deletions(-) diff --git a/src/commands/actor/push-data.ts b/src/commands/actor/push-data.ts index a36dac65b..7c9f054ca 100644 --- a/src/commands/actor/push-data.ts +++ b/src/commands/actor/push-data.ts @@ -34,7 +34,9 @@ export class ActorPushDataCommand extends ApifyCommand | Record[]; - const item = _item || (await readStdin()); + // Nobody asked for stdin when the item is missing, so it must not block on a pipe this + // process only inherited. + const item = _item || (await readStdin({ implicit: true })); if (!item) { error({ message: 'No items were provided.' }); diff --git a/src/lib/command-framework/CommandError.ts b/src/lib/command-framework/CommandError.ts index 3d3e66d00..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 { peekStdin } from '../commands/read-stdin.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? ${peekStdin() ? 'Yes' : 'No'}`, + `- Stdin data? ${describeStdinRead()}`, ].join('\n'); } } diff --git a/src/lib/commands/read-stdin.ts b/src/lib/commands/read-stdin.ts index a8431e140..f6ed64dbe 100644 --- a/src/lib/commands/read-stdin.ts +++ b/src/lib/commands/read-stdin.ts @@ -3,16 +3,18 @@ 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 + * How long an implicit read waits for stdin to say anything more. Long enough for a writer that has + * to fetch or compute its next 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; +let readFinished = false; +let readCutShort = false; -export interface ReadStdinOptions { +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 @@ -29,6 +31,7 @@ export interface ReadStdinOptions { export async function readStdin(options: ReadStdinOptions = {}) { readPromise ??= _readStdin(options).then((data) => { readResult = data; + readFinished = true; return data; }); @@ -36,11 +39,23 @@ export async function readStdin(options: ReadStdinOptions = {}) { } /** - * Stdin data from a completed read, without starting one. Undefined until a read finishes. For - * diagnostics only. + * Whether stdin went quiet mid-stream and the read gave up on the rest. Explains a payload that + * ends where nothing should end. */ -export function peekStdin() { - return readResult; +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) { @@ -50,10 +65,11 @@ async function _readStdin({ implicit }: ReadStdinOptions) { 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; + // `waitDelay` guards the first byte on a socket, and nothing else. An implicit read needs a + // deadline on every byte: 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). Where stdin already has a + // first-byte deadline, keep it — it is shorter, and dropping it slows down every spawned CLI. + const firstByteTimeout = implicit ? waitDelay || IMPLICIT_STDIN_IDLE_TIMEOUT_MILLIS : waitDelay; const bufferChunks: Buffer[] = []; @@ -61,9 +77,9 @@ async function _readStdin({ implicit }: ReadStdinOptions) { let timeout: NodeJS.Timeout | null = null; - const armTimeout = () => { - if (idleTimeout) { - timeout = setTimeout(() => controller.abort(), idleTimeout).unref(); + const armTimeout = (delay: number) => { + if (delay) { + timeout = setTimeout(() => controller.abort(), delay).unref(); } }; @@ -74,7 +90,7 @@ async function _readStdin({ implicit }: ReadStdinOptions) { } }; - armTimeout(); + armTimeout(firstByteTimeout); const onData = (chunk: Buffer) => { bufferChunks.push(chunk); @@ -85,7 +101,7 @@ async function _readStdin({ implicit }: ReadStdinOptions) { // 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(); + armTimeout(IMPLICIT_STDIN_IDLE_TIMEOUT_MILLIS); } }; @@ -96,10 +112,14 @@ async function _readStdin({ implicit }: ReadStdinOptions) { } catch (error) { const casted = error as Error; - // 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; + 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 (!implicit) { + return; + } + + readCutShort = bufferChunks.length > 0; } } finally { // Stop reading from stdin so its open handle can't keep the event loop (and diff --git a/src/lib/commands/resolve-input.ts b/src/lib/commands/resolve-input.ts index af5d965c6..61a345a95 100644 --- a/src/lib/commands/resolve-input.ts +++ b/src/lib/commands/resolve-input.ts @@ -7,7 +7,7 @@ import mime from 'mime'; import { CommandExitCodes } from '../consts.js'; import { error } from '../outputs.js'; import { getLocalInput } from '../utils.js'; -import { readStdin } from './read-stdin.js'; +import { readStdin, stdinWasCutShort } from './read-stdin.js'; interface InputOverrideOptions { schemaHint?: string; @@ -77,9 +77,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 e8a795258..875b785c5 100644 --- a/src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts +++ b/src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts @@ -52,7 +52,7 @@ export function stdinCheckWrapper any>( const casted = input as StdinCheckWrapperInput>>; // 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. + // 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 c0410d1a8..be61279a8 100644 --- a/test/e2e/commands/stdin-held-open.test.ts +++ b/test/e2e/commands/stdin-held-open.test.ts @@ -89,9 +89,9 @@ describe('[e2e] stdin held open (#1206)', () => { } }); - // 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. + // The tests above stop at the project check, well 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); From 81860946ade2a657a944b2caeec244a9b3b9a811 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Richard=20Sol=C3=A1r?= Date: Tue, 27 Oct 2026 12:50:21 +0100 Subject: [PATCH 4/4] fix(cli): wait for the writer where stdin is the documented input Piping is the documented main usage of `actor push-data` and `datasets push-items`, so an item-less call waits for EOF instead of giving up after 2s of quiet. A producer slower than that was dropping its data silently. An implicit read that gives up on a silent pipe now says so on stderr, so a slow writer's input does not vanish without a word. A spawned child's empty socket keeps its short deadline and stays quiet. Co-Authored-By: Claude Opus 5 --- src/commands/actor/push-data.ts | 5 +- src/commands/datasets/push-items.ts | 5 +- src/lib/commands/read-stdin.ts | 28 +++--- src/lib/commands/resolve-input.ts | 11 ++- test/e2e/commands/stdin-held-open.test.ts | 106 ++++++++++++---------- 5 files changed, 84 insertions(+), 71 deletions(-) diff --git a/src/commands/actor/push-data.ts b/src/commands/actor/push-data.ts index 7c9f054ca..01368a1fc 100644 --- a/src/commands/actor/push-data.ts +++ b/src/commands/actor/push-data.ts @@ -34,9 +34,8 @@ export class ActorPushDataCommand extends ApifyCommand | Record[]; - // Nobody asked for stdin when the item is missing, so it must not block on a pipe this - // process only inherited. - const item = _item || (await readStdin({ implicit: true })); + // 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/lib/commands/read-stdin.ts b/src/lib/commands/read-stdin.ts index f6ed64dbe..912b643a4 100644 --- a/src/lib/commands/read-stdin.ts +++ b/src/lib/commands/read-stdin.ts @@ -2,11 +2,7 @@ import { once } from 'node:events'; import { useStdin } from '../hooks/useStdin.js'; -/** - * How long an implicit read waits for stdin to say anything more. Long enough for a writer that has - * to fetch or compute its next bytes, short enough that a pipe with nothing behind it does not look - * like a hang. - */ +/** Long enough for a writer that has to fetch its next bytes, short enough not to look like a hang. */ const IMPLICIT_STDIN_IDLE_TIMEOUT_MILLIS = 2_000; let readPromise: Promise | undefined; @@ -39,8 +35,8 @@ export async function readStdin(options: ReadStdinOptions = {}) { } /** - * Whether stdin went quiet mid-stream and the read gave up on the rest. Explains a payload that - * ends where nothing should end. + * 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; @@ -65,10 +61,8 @@ async function _readStdin({ implicit }: ReadStdinOptions) { return; } - // `waitDelay` guards the first byte on a socket, and nothing else. An implicit read needs a - // deadline on every byte: 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). Where stdin already has a - // first-byte deadline, keep it — it is shorter, and dropping it slows down every spawned CLI. + // 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[] = []; @@ -97,9 +91,8 @@ async function _readStdin({ implicit }: ReadStdinOptions) { 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. + // 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); } @@ -119,11 +112,12 @@ async function _readStdin({ implicit }: ReadStdinOptions) { return; } - readCutShort = bufferChunks.length > 0; + // 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). + // Stop reading so an open stdin handle can't keep the CLI alive after the command (#1206). stream.off('data', onData); stream.pause(); } diff --git a/src/lib/commands/resolve-input.ts b/src/lib/commands/resolve-input.ts index 61a345a95..970e58f9f 100644 --- a/src/lib/commands/resolve-input.ts +++ b/src/lib/commands/resolve-input.ts @@ -5,7 +5,7 @@ import process from 'node:process'; import mime from 'mime'; 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'; @@ -62,6 +62,15 @@ export async function getInputOverride( // 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 { const parsed = JSON.parse(stdin.toString('utf8')); diff --git a/test/e2e/commands/stdin-held-open.test.ts b/test/e2e/commands/stdin-held-open.test.ts index be61279a8..8d0a7c2d9 100644 --- a/test/e2e/commands/stdin-held-open.test.ts +++ b/test/e2e/commands/stdin-held-open.test.ts @@ -1,17 +1,13 @@ -import { mkdir, open, readFile, rm, writeFile } from 'node:fs/promises'; +import { mkdir, open, readdir, readFile, rm, writeFile } from 'node:fs/promises'; import path from 'node:path'; import process from 'node:process'; -import { fileURLToPath } from 'node:url'; 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)', () => { @@ -27,40 +23,25 @@ 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 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. + // 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]); @@ -73,11 +54,7 @@ describe('[e2e] stdin held open (#1206)', () => { 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', - }, + env: CliTestEnv, }); expect(result.timedOut, `stderr: ${result.stderr}`).toBe(false); @@ -89,9 +66,8 @@ describe('[e2e] stdin held open (#1206)', () => { } }); - // The tests above stop at the project check, well 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. + // 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); @@ -104,8 +80,7 @@ describe('[e2e] stdin held open (#1206)', () => { 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. + // Records the input to a file, not stdout, so the package manager cannot garble it. await writeFile( path.join(dir, 'main.js'), [ @@ -126,8 +101,7 @@ describe('[e2e] stdin held open (#1206)', () => { 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. + // `r+` keeps a writer attached, so the pipe never reaches EOF even after the write below. const handle = await open(fifo, 'r+'); try { @@ -140,12 +114,7 @@ describe('[e2e] stdin held open (#1206)', () => { 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', - }, + env: CliTestEnv, }); } finally { await handle.close(); @@ -159,6 +128,7 @@ describe('[e2e] stdin held open (#1206)', () => { 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'); }); @@ -171,3 +141,45 @@ describe('[e2e] stdin held open (#1206)', () => { 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 }]); + }); +});