fix(cli): read stdin lazily so open pipes don't hang the CLI

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<typeof ActorPushDataCommand> {
@@ -34,7 +34,8 @@ export class ActorPushDataCommand extends ApifyCommand<typeof ActorPushDataComma
 	async run() {
 		const { item: _item } = this.args;
 
-		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 item was provided.' });

src/commands/datasets/push-items.ts

@@ -1,9 +1,9 @@
 import type { ApifyApiError } from 'apify-client';
 import chalk from 'chalk';
 
-import { cachedStdinInput } from '../../entrypoints/_shared.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 { tryToGetDataset } from '../../lib/commands/storages.js';
 import { error, success } from '../../lib/outputs.js';
 import { getLoggedClientOrThrow } from '../../lib/utils.js';
@@ -57,7 +57,8 @@ export class DatasetsPushDataCommand extends ApifyCommand<typeof DatasetsPushDat
 
 		let parsedData: Record<string, unknown> | Record<string, unknown>[];
 
-		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.' });

src/entrypoints/_shared.ts

@@ -11,16 +11,13 @@ 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';
 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)`;

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');
 			}
 		}

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<T extends typeof BuiltApifyCommand = typeof B
 							this.args[camelCasedName] = String(rawArg);
 
 							if (rawArg === '-' && builderData.stdin) {
-								this.args[camelCasedName] = this._handleStdin(builderData.stdin);
+								this.args[camelCasedName] = await this._handleStdin(builderData.stdin);
 							}
 
 							if (builderData.catchAll) {
@@ -399,7 +399,7 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
 			return;
 		}
 
-		this._parseFlags(rawFlags, rawTokens);
+		await this._parseFlags(rawFlags, rawTokens);
 
 		try {
 			if (this.ctor.enableProfileFlag && typeof rawFlags.profile === 'string') {
@@ -479,7 +479,7 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
 		return flagKey;
 	}
 
-	private _parseFlags(rawFlags: ParseResult['values'], rawTokens: ParseResult['tokens']) {
+	private async _parseFlags(rawFlags: ParseResult['values'], rawTokens: ParseResult['tokens']) {
 		if (!this.ctor.flags) {
 			return;
 		}
@@ -609,7 +609,7 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
 
 							flagThatUsedStdin = baseFlagName;
 
-							this.flags[camelCasedName] = this._handleStdin(builderData.stdin);
+							this.flags[camelCasedName] = await this._handleStdin(builderData.stdin);
 						}
 
 						break;
@@ -734,12 +734,14 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
 		});
 	}
 
-	private _handleStdin(mode: StdinMode) {
+	private async _handleStdin(mode: StdinMode) {
+		const input = await readStdin();
+
 		switch (mode) {
 			case StdinMode.Stringified:
-				return (cachedStdinInput?.toString('utf8') ?? '').trim();
+				return (input?.toString('utf8') ?? '').trim();
 			default:
-				return cachedStdinInput;
+				return input;
 		}
 	}
 

src/lib/commands/read-stdin.ts

@@ -2,37 +2,102 @@ import { once } from 'node:events';
 
 import { useStdin } from '../hooks/useStdin.js';
 
-export async function readStdin() {
-	const dataRef = await useStdin();
+/** 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<Buffer | undefined> | 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);
 

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,
 					),
 				});

src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts

@@ -47,11 +47,13 @@ export function stdinCheckWrapper<Fn extends (...args: any[]) => any>(
 	}: StdinCheckWrapperOptions = {},
 ): (...args: NewFunctionArgs<Fn>) => Promise<Awaited<ReturnType<Fn>>> {
 	return async (input, ...rest) => {
-		const { isTTY, hasData } = await useStdin();
+		const { isTTY } = await useStdin();
 
 		const casted = input as StdinCheckWrapperInput<Awaited<ReturnType<Fn>>>;
 
-		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);
 			}