Skip to content
6 changes: 4 additions & 2 deletions src/commands/actor/push-data.ts
Original file line number Diff line number Diff line change
@@ -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> {
Expand Down Expand Up @@ -34,7 +34,9 @@ export class ActorPushDataCommand extends ApifyCommand<typeof ActorPushDataComma
async run() {
const { item: _item } = this.args;

const item = _item || cachedStdinInput;
// 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 item was provided.' });
Expand Down
6 changes: 4 additions & 2 deletions src/commands/datasets/push-items.ts
Original file line number Diff line number Diff line change
@@ -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';
Expand Down Expand Up @@ -55,7 +55,9 @@ export class DatasetsPushDataCommand extends ApifyCommand<typeof DatasetsPushDat

let parsedData: Record<string, unknown> | Record<string, unknown>[];

const item = _item || cachedStdinInput;
// 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.' });
Expand Down
3 changes: 0 additions & 3 deletions src/entrypoints/_shared.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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)`;
Expand Down
4 changes: 2 additions & 2 deletions src/lib/command-framework/CommandError.ts
Original file line number Diff line number Diff line change
@@ -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';
Expand Down Expand Up @@ -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');
}
}
Expand Down
18 changes: 10 additions & 8 deletions src/lib/command-framework/apify-command.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -368,7 +368,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) {
Expand All @@ -389,7 +389,7 @@ export abstract class ApifyCommand<T extends typeof BuiltApifyCommand = typeof B
return;
}

this._parseFlags(rawFlags, rawTokens);
await this._parseFlags(rawFlags, rawTokens);

try {
await this.run();
Expand Down Expand Up @@ -464,7 +464,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;
}
Expand Down Expand Up @@ -594,7 +594,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;
Expand Down Expand Up @@ -719,12 +719,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;
}
}

Expand Down
116 changes: 93 additions & 23 deletions src/lib/commands/read-stdin.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,37 +2,109 @@ import { once } from 'node:events';

import { useStdin } from '../hooks/useStdin.js';

export async function readStdin() {
const dataRef = await useStdin();
/**
* 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<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;
}

/**
* 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 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';
}

const { hasData, waitDelay, stream } = dataRef;
return readResult ? 'Yes' : 'No';
}

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 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[] = [];

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 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(IMPLICIT_STDIN_IDLE_TIMEOUT_MILLIS);
}
};

stream.on('data', onData);

try {
Expand All @@ -41,24 +113,22 @@ 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;
}

readCutShort = bufferChunks.length > 0;
}
} 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);

Expand Down
15 changes: 11 additions & 4 deletions src/lib/commands/resolve-input.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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, stdinWasCutShort } from './read-stdin.js';

interface InputOverrideOptions {
schemaHint?: string;
Expand Down Expand Up @@ -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 {
Expand All @@ -76,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,
),
});
Expand Down
6 changes: 4 additions & 2 deletions src/lib/hooks/user-confirmations/_stdinCheckWrapper.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
Loading
Loading