mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-15 08:02:53 +00:00
fix(cli): keep script run waiting on long jobs and flaky log streaming (#11766)
* fix(cli): keep script run waiting on long jobs and flaky log streaming Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(cli): back off job result polling after an error Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(cli): keep skipped jobs successful and color early completions Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(cli): fall back to the polled result when the final job fetch fails Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
cd7813e354
commit
a2ee052e03
@@ -19,7 +19,7 @@ import * as log from "../../core/log.ts";
|
||||
import { sep as SEP } from "node:path";
|
||||
import * as path from "node:path";
|
||||
import { stringify as yamlStringify } from "yaml";
|
||||
import { deepEqual, getHeaders, isFileResource, isFilesetResource, readTextFile, readTextFileSync } from "../../utils/utils.ts";
|
||||
import { deepEqual, getHeaders, isFileResource, isFilesetResource, readTextFile, readTextFileSync, sleep } from "../../utils/utils.ts";
|
||||
import { detectAuthGatewayChallenge } from "../../utils/http_guards.ts";
|
||||
import * as wmill from "../../../gen/services.gen.ts";
|
||||
import * as specificItems from "../../core/specific_items.ts";
|
||||
@@ -1492,45 +1492,49 @@ async function run(
|
||||
await track_job(workspace.workspaceId, id);
|
||||
}
|
||||
|
||||
const MAX_RETRIES = 600; // ~60 seconds at 100ms intervals
|
||||
let retries = 0;
|
||||
while (retries < MAX_RETRIES) {
|
||||
try {
|
||||
const completedJob = await wmill.getCompletedJob({
|
||||
workspace: workspace.workspaceId,
|
||||
id,
|
||||
});
|
||||
|
||||
if (completedJob.success === false) {
|
||||
process.exitCode = 1;
|
||||
}
|
||||
|
||||
const result = completedJob.result ?? {};
|
||||
if (opts.silent) {
|
||||
console.log(JSON.stringify(result));
|
||||
} else {
|
||||
log.info(JSON.stringify(result, null, 2));
|
||||
}
|
||||
|
||||
break;
|
||||
} catch {
|
||||
retries++;
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
}
|
||||
let { result, success } = await pollForJobResult(workspace.workspaceId, id);
|
||||
// Prefer getCompletedJob: unlike the poller, it counts a skipped
|
||||
// (debounced) job as a success.
|
||||
try {
|
||||
const completedJob = await wmill.getCompletedJob({
|
||||
workspace: workspace.workspaceId,
|
||||
id,
|
||||
});
|
||||
result = completedJob.result;
|
||||
success = completedJob.success !== false;
|
||||
} catch {
|
||||
/* keep the poller's result */
|
||||
}
|
||||
if (retries >= MAX_RETRIES) {
|
||||
throw new Error(`Timed out waiting for job ${id} to complete`);
|
||||
if (!success) {
|
||||
process.exitCode = 1;
|
||||
}
|
||||
result = result ?? {};
|
||||
if (opts.silent) {
|
||||
console.log(JSON.stringify(result));
|
||||
} else {
|
||||
log.info(JSON.stringify(result, null, 2));
|
||||
}
|
||||
}
|
||||
|
||||
function logJobCompleted(success: boolean | undefined) {
|
||||
log.info("\n");
|
||||
log.info(
|
||||
success === false
|
||||
? colors.bold.underline.red("Job Completed")
|
||||
: colors.bold.underline.green("Job Completed")
|
||||
);
|
||||
log.info("\n");
|
||||
}
|
||||
|
||||
const JOB_UPDATES_POLL_INTERVAL_MS = 500;
|
||||
const MAX_CONSECUTIVE_JOB_UPDATE_ERRORS = 5;
|
||||
|
||||
export async function track_job(workspace: string, id: string) {
|
||||
try {
|
||||
const result = await wmill.getCompletedJob({ workspace, id });
|
||||
|
||||
log.info(result.logs);
|
||||
log.info("\n");
|
||||
log.info(colors.bold.underline.green("Job Completed"));
|
||||
log.info("\n");
|
||||
logJobCompleted(result.success);
|
||||
return;
|
||||
} catch {
|
||||
/* ignore */
|
||||
@@ -1538,15 +1542,13 @@ export async function track_job(workspace: string, id: string) {
|
||||
|
||||
log.info(colors.yellow("Waiting for Job " + id + " to start..."));
|
||||
|
||||
// Server-side character position (1-based, as returned in `log_offset`), not
|
||||
// a JS string length: UTF-16 lengths drift from it on non-BMP characters.
|
||||
let logOffset = 0;
|
||||
let running = false;
|
||||
let retry = 0;
|
||||
let errors = 0;
|
||||
while (true) {
|
||||
let updates: {
|
||||
running?: boolean | undefined;
|
||||
completed?: boolean | undefined;
|
||||
new_logs?: string | undefined;
|
||||
};
|
||||
let updates: Awaited<ReturnType<typeof wmill.getJobUpdates>>;
|
||||
try {
|
||||
updates = await wmill.getJobUpdates({
|
||||
workspace,
|
||||
@@ -1554,13 +1556,16 @@ export async function track_job(workspace: string, id: string) {
|
||||
logOffset,
|
||||
running,
|
||||
});
|
||||
} catch {
|
||||
retry++;
|
||||
if (retry > 3) {
|
||||
log.info("failed to get job updated. skipping log streaming.");
|
||||
errors = 0;
|
||||
} catch (e: any) {
|
||||
errors++;
|
||||
if (errors >= MAX_CONSECUTIVE_JOB_UPDATE_ERRORS) {
|
||||
log.warn(
|
||||
`Failed to get job updates (${e?.message ?? e}), log streaming stopped. Waiting for the job to complete...`
|
||||
);
|
||||
break;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 500));
|
||||
await sleep(JOB_UPDATES_POLL_INTERVAL_MS);
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -1571,11 +1576,10 @@ export async function track_job(workspace: string, id: string) {
|
||||
|
||||
if (updates.new_logs) {
|
||||
process.stdout.write(updates.new_logs);
|
||||
logOffset += updates.new_logs.length;
|
||||
}
|
||||
logOffset = updates.log_offset ?? logOffset;
|
||||
|
||||
if (updates.completed === true) {
|
||||
running = false;
|
||||
break;
|
||||
}
|
||||
|
||||
@@ -1583,24 +1587,25 @@ export async function track_job(workspace: string, id: string) {
|
||||
running = false;
|
||||
log.info(colors.yellow("Job suspended. Waiting for it to continue..."));
|
||||
}
|
||||
await sleep(JOB_UPDATES_POLL_INTERVAL_MS);
|
||||
}
|
||||
await new Promise((resolve, _) => setTimeout(() => resolve(undefined), 1000));
|
||||
|
||||
await pollForJobResult(workspace, id);
|
||||
|
||||
let success: boolean | undefined;
|
||||
try {
|
||||
const final_job = await wmill.getCompletedJob({ workspace, id });
|
||||
if ((final_job.logs?.length ?? -1) > logOffset) {
|
||||
log.info(final_job.logs!.substring(logOffset));
|
||||
success = final_job.success;
|
||||
const remaining = Array.from(final_job.logs ?? "")
|
||||
.slice(Math.max(logOffset - 1, 0))
|
||||
.join("");
|
||||
if (remaining) {
|
||||
log.info(remaining);
|
||||
}
|
||||
log.info("\n");
|
||||
if (final_job.success) {
|
||||
log.info(colors.bold.underline.green("Job Completed"));
|
||||
} else {
|
||||
log.info(colors.bold.underline.red("Job Completed"));
|
||||
}
|
||||
log.info("\n");
|
||||
} catch {
|
||||
log.info("Job appears to have completed, but no data can be retrieved");
|
||||
} catch (e: any) {
|
||||
log.warn(`Could not retrieve the job's remaining logs: ${e?.message ?? e}`);
|
||||
}
|
||||
logJobCompleted(success);
|
||||
}
|
||||
|
||||
export async function pollForJobResult(
|
||||
|
||||
@@ -135,8 +135,10 @@ export async function pollJobWithQueueLogging(
|
||||
);
|
||||
}
|
||||
|
||||
// Errors back off to the slow interval: at the fast one, the consecutive
|
||||
// error budget would be spent on a ~1s blip.
|
||||
const delayMs =
|
||||
Date.now() - startedAt < fastPollDurationMs
|
||||
consecutiveErrors === 0 && Date.now() - startedAt < fastPollDurationMs
|
||||
? fastPollIntervalMs
|
||||
: slowPollIntervalMs;
|
||||
await new Promise((resolve) => setTimeout(resolve, delayMs));
|
||||
|
||||
Reference in New Issue
Block a user