diff --git a/cli/console.ts b/cli/console.ts index 2f9bd60c6..50a069742 100644 --- a/cli/console.ts +++ b/cli/console.ts @@ -172,6 +172,10 @@ export function printError(errorText: string, indentCount: number = 0) { writeStdErr(errorOutput(errorText), indentCount); } +export function printWarning(warningText: string, indentCount: number = 0) { + writeStdErr(warningOutput(warningText), indentCount); +} + export function printInitResult(result: IInitResult) { if (result.dirsCreated && result.dirsCreated.length) { writeStdOut(successOutput("Directories successfully created:")); @@ -425,6 +429,8 @@ export function printExecutedAction( return; } case dataform.ActionResult.ExecutionStatus.SKIPPED: { + const skipReason = executedAction.tasks?.[0]?.errorMessage; + const skipSuffix = skipReason ? ` (${skipReason})` : ""; switch (executionAction.type) { case "table": { writeStdOut( @@ -432,7 +438,7 @@ export function printExecutedAction( executionAction.target, executionAction.tableType, executionAction.tasks.length === 0 - )}` + )}${skipSuffix}` ); return; } @@ -441,7 +447,7 @@ export function printExecutedAction( `${warningOutput("Skipping assertion execution: ")} ${assertionString( executionAction.target, executionAction.tasks.length === 0 - )}` + )}${skipSuffix}` ); return; } @@ -450,7 +456,7 @@ export function printExecutedAction( `${warningOutput("Skipping operation execution: ")} ${operationString( executionAction.target, executionAction.tasks.length === 0 - )}` + )}${skipSuffix}` ); return; } diff --git a/cli/index.ts b/cli/index.ts index 5a38a4d44..6d3507a5e 100644 --- a/cli/index.ts +++ b/cli/index.ts @@ -24,7 +24,8 @@ import { printInitCredsResult, printInitResult, printSuccess, - printTestResult + printTestResult, + printWarning } from "df/cli/console"; import { getBigQueryCredentials } from "df/cli/credentials"; import { @@ -230,11 +231,22 @@ const dotOutputOption: INamedOption = { const timeoutOption: INamedOption = { name: "timeout", + option: { + describe: "Duration to allow project compilation to complete. Examples: '1s', '10m', etc.", + type: "string", + default: null, + coerce: (rawTimeoutString: string | null) => + rawTimeoutString ? parseDuration(rawTimeoutString) : null + } +}; + +const executionTimeoutOption: INamedOption = { + name: "execution-timeout", option: { describe: - "Wall-clock deadline for the entire command. When it fires, in-flight work " + - "(including any running JiT compilation worker) is cancelled. Examples: " + - "'1s', '10m'.", + "Wall-clock deadline for the entire run (compile + all actions). When it fires, " + + "in-flight actions are cancelled and pending actions are skipped. Off by default. " + + "Examples: '10m', '2h'.", type: "string", default: null, coerce: (rawTimeoutString: string | null) => @@ -247,8 +259,9 @@ const jitTimeoutOption: INamedOption = { option: { describe: "Per-model JiT compilation worker timeout. Each action with jitCode gets " + - "its own fresh deadline; independent of --timeout. When unset, no per-model " + - "cap is applied and only --timeout bounds JiT work. Examples: '30s', '2m'.", + "its own fresh deadline; independent of --execution-timeout. When unset, no " + + "per-model cap is applied and only --execution-timeout bounds JiT work. " + + "Examples: '30s', '2m'.", type: "string", default: null, coerce: (rawTimeoutString: string | null) => @@ -689,6 +702,7 @@ export function runCli() { includeDependentsOption, jsonOutputOption, timeoutOption, + executionTimeoutOption, jitTimeoutOption, tagsOption, bigqueryJobLabelsOption, @@ -705,6 +719,16 @@ export function runCli() { ); return; } + if ( + !isJsonOutput && + argv[timeoutOption.name] != null && + argv[executionTimeoutOption.name] == null + ) { + printWarning( + "Note: --timeout only bounds project compilation. " + + "For a whole-run wall-clock deadline, use --execution-timeout.\n" + ); + } logger.log("Compiling...\n"); const compiledGraph = await compile({ projectDir: argv[projectDirOption.name], @@ -729,7 +753,7 @@ export function runCli() { includeDependencies: argv[includeDepsOption.name], includeDependents: argv[includeDependentsOption.name], tags: argv[tagsOption.name], - timeoutMillis: argv[timeoutOption.name] || undefined, + timeoutMillis: argv[executionTimeoutOption.name] || undefined, jitTimeoutMillis: argv[jitTimeoutOption.name] || undefined }, dbadapter @@ -838,6 +862,17 @@ export function runCli() { if (isJsonOutput) { print(prettyJsonStringify(runResult)); } + if (!isJsonOutput) { + if (runResult.status === dataform.RunResult.ExecutionStatus.TIMED_OUT) { + const executionTimeoutMillis = argv[executionTimeoutOption.name]; + const suffix = executionTimeoutMillis + ? ` after ${executionTimeoutMillis / 1000} seconds (--execution-timeout)` + : ""; + printError(`Run timed out${suffix}.`); + } else if (runResult.status === dataform.RunResult.ExecutionStatus.CANCELLED) { + printError("Run cancelled."); + } + } return runResult.status === dataform.RunResult.ExecutionStatus.SUCCESSFUL ? 0 : 1; } }, diff --git a/cli/index_run_e2e_test.ts b/cli/index_run_e2e_test.ts index a7031eeef..226e5833b 100644 --- a/cli/index_run_e2e_test.ts +++ b/cli/index_run_e2e_test.ts @@ -812,4 +812,57 @@ DROP SCHEMA IF EXISTS \`\${dataform.projectConfig.defaultDatabase}.\${dataform.p } }); }); + + test("--timeout on run emits deprecation notice pointing to --execution-timeout", async () => { + // The notice fires in the run handler before compile. Yargs validation + // requires workflow_settings.yaml to exist, but compile can fail after that + // — we only assert on stderr for the notice line. + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + `defaultProject: ${DEFAULT_DATABASE}\ndefaultLocation: ${DEFAULT_LOCATION}\n` + ); + + const runResult = await getProcessResult( + execFile(nodePath, [ + cliEntryPointPath, + "run", + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--timeout", + "30s" + ]) + ); + + expect(runResult.stderr).to.match( + /--timeout only bounds project compilation[\s\S]*use --execution-timeout/ + ); + }); + + test("--timeout on run does NOT emit notice when --execution-timeout is also set", async () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + `defaultProject: ${DEFAULT_DATABASE}\ndefaultLocation: ${DEFAULT_LOCATION}\n` + ); + + const runResult = await getProcessResult( + execFile(nodePath, [ + cliEntryPointPath, + "run", + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--timeout", + "30s", + "--execution-timeout", + "10m" + ]) + ); + + expect(runResult.stderr).to.not.match( + /--timeout only bounds project compilation/ + ); + }); }); diff --git a/cli/tests/jit/index_jit_runtime_test.ts b/cli/tests/jit/index_jit_runtime_test.ts index ca587c41e..eb27f5785 100644 --- a/cli/tests/jit/index_jit_runtime_test.ts +++ b/cli/tests/jit/index_jit_runtime_test.ts @@ -80,7 +80,7 @@ suite("JiT support runtime", ({ afterEach }) => { expect(runResult.stdout).to.include("Compilation timed out"); }); - test("Global --timeout cancels in-flight JiT worker", { timeout: 90000 }, async () => { + test("Global --execution-timeout cancels in-flight JiT worker", { timeout: 90000 }, async () => { const projectDir = tmpDirFixture.createNewTmpDir(); await setupJitProject(tmpDirFixture, projectDir); @@ -91,8 +91,8 @@ suite("JiT support runtime", ({ afterEach }) => { return "SELECT 1"; })` ); - // --timeout must exceed BQ schema-prep time; smaller values fire the - // timer before the JiT compile starts, leaving the action SKIPPED and + // --execution-timeout must exceed BQ schema-prep time; smaller values fire + // the timer before the JiT compile starts, leaving the action SKIPPED and // defeating the assertions below. const runResult = await getProcessResult( execFile(nodePath, [ @@ -104,7 +104,7 @@ suite("JiT support runtime", ({ afterEach }) => { "--dry-run", "--json", "--actions=hang_jit_global", - "--timeout=15s" + "--execution-timeout=15s" ], { timeout: 80000 }) ); diff --git a/cli/tests/jit/jit_run_test.ts b/cli/tests/jit/jit_run_test.ts index a1f4d4ede..d196c96fa 100644 --- a/cli/tests/jit/jit_run_test.ts +++ b/cli/tests/jit/jit_run_test.ts @@ -437,6 +437,46 @@ suite("run", () => { ); }); + test("Global timeout surfaces reason on skipped pending actions", async () => { + const { adapterInstance } = createMocks(); + + const executionGraph = createGraph([ + { + target: { database: "db", schema: "sch", name: "hang_jit" }, + type: "table", + tableType: "table", + jitCode: "async (jctx) => { /* hangs forever in mock */ }", + tasks: [] + }, + { + target: { database: "db", schema: "sch", name: "downstream" }, + type: "table", + tableType: "table", + jitCode: "async (jctx) => { return 'SELECT 1'; }", + tasks: [], + dependencyTargets: [{ database: "db", schema: "sch", name: "hang_jit" }] + } + ]); + executionGraph.runConfig.timeoutMillis = 200; + + const runner = new Runner(adapterInstance, executionGraph, { + jitCompiler: (req, pdir, adapter, client, timeoutMs, opts, onCancel) => + new Promise((resolve, reject) => { + onCancel(() => reject(new Error("Run cancelled while worker was in flight."))); + }) + }); + + const result = await runner.execute().result(); + + expect(result.status).equals(dataform.RunResult.ExecutionStatus.TIMED_OUT); + + const downstream = result.actions.find(a => a.target.name === "downstream"); + expect(downstream).to.not.equal(undefined); + expect(downstream.status).equals(dataform.ActionResult.ExecutionStatus.SKIPPED); + expect(downstream.tasks.length).to.be.greaterThan(0); + expect(downstream.tasks[0].errorMessage).to.match(/Run timed out after 0\.2 seconds/); + }); + test("JiT compilation cleans up CANCEL_EVENT listeners after each action", async () => { const { adapterInstance } = createMocks();