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; + +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, 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; + return data; + }); + + return readPromise; +} - const { hasData, waitDelay, stream } = dataRef; +/** + * 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({ 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 { @@ -40,25 +96,19 @@ export 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 { // 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(); } - 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..af5d965c6 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; @@ -58,8 +58,9 @@ 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 }); 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..c0410d1a8 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, 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'; @@ -55,4 +56,118 @@ 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 }); + } + }); + + // 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 }); + }); });