Files
windmill/backend/windmill-common/src/error.rs
T
Ruben Fiszel a6d4390790 feat: workflow-as-code (WAC) v2 (#8172)
* feat: workflow-as-code v2 with @task decorator API

Replace ctx.step("name", "script") API with @task decorators where
functions are called directly. Users no longer need to pass WorkflowCtx
or use string-based step names/script paths.

Python: @task decorator with contextvars-based implicit context
TypeScript: task() wrapper with module-level context variable
Parsers: detect @task function calls instead of ctx.step() calls
Worker: updated wrappers to set implicit context

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: WAC v2 checkpoint/replay with _executing_key child dispatch

- Rust-side orchestration: parent dispatches child jobs, suspends, resumes on completion
- _executing_key in checkpoint tells child which step to execute directly
- task() throws StepSuspend(mode="step_complete") after executing target step
- result_processor handles child completion and updates parent checkpoint
- WacGraph.svelte for runtime execution visualization
- Sequential and parallel workflows tested end-to-end

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: WAC v2 bundle cache, globalThis ctx sharing, description optional

- Disable bun bundle caching for WAC v2 scripts (wrapper needs
  windmill-client from node_modules, not available in bundle mode)
- Use Reflect.set/get(globalThis, "__wmill_wf_ctx") to share workflow
  context across dual module instances (wrapper vs user script)
- Never-resolving thenable for non-matching steps in child job mode
  prevents Promise.all race conditions
- Make description field optional in NewScript API (defaults to "")

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: add step() primitive for inline checkpointed steps

step() executes a function inline (no child job) and persists the result
to the checkpoint. On replay, the cached value is returned — ensuring
deterministic behavior for non-deterministic operations like Date.now()
or Math.random().

- TypeScript: step(name, fn) — executes inline, throws StepSuspend with
  mode "inline_checkpoint" to persist before continuing
- Rust: InlineCheckpoint variant in WacOutput, saves to checkpoint and
  resets running=false for immediate re-pickup (no zombie wait)
- Shared step counter between task() and step() via _allocKey()

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: add Python WAC v2 support with task(), step(), workflow()

- Python SDK: WorkflowCtx with _executing_key child mode, _alloc_key
  shared counter, _run_inline_step for step(), _execute_directly and
  _never_resolve for child mode, step() async function
- Python executor: WAC v2 detection, checkpoint.json writing, WAC
  wrapper.py generation calling _run_workflow(), post-execution hook
  into shared handle_wac_v2_output()
- Make handle_wac_v2_output pub so both bun and python executors share
  the same dispatch/suspend/inline-checkpoint logic
- 17 Python tests covering dispatch, replay, parallel, conditional,
  inline checkpoint, and child mode

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* chore: update sqlx prepared queries

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: WacGraph Tooltip→Popover, simplify wacToFlow parsers

- Fix type error: Tooltip doesn't accept text snippet, use Popover
- Extract shared helpers for task matching and block collection
- Replace linear tasks.find() with Map lookups
- Remove mutable module-level counter

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: Box::pin WAC v2 output handler to prevent stack overflow

handle_python_job's async state machine was too large when combined
with handle_wac_v2_output. Box::pin heap-allocates the future.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: merge WAC v1 and v2 task decorators to preserve backward compat

The v2 @task decorator was shadowing the v1 one, breaking WAC v1
scripts that rely on HTTP-based dispatch via /workflow_as_code/ API.

The merged decorator handles three modes:
- v2: inside @workflow context → checkpoint/replay dispatch
- v1: WM_JOB_ID set, no @workflow → HTTP API dispatch + wait_job
- standalone: no Windmill env → execute function body directly

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: skip no_main_func detection for WAC v2 scripts in TS and Python parsers

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: prevent empty/noop dispatch causing infinite requeue loop

- Validate steps.len() > 0 in WAC dispatch handler (issue 3)
- Replace noop StepSuspend throw with never-resolving promise so it
  can't reach the backend as an empty dispatch (issue 4)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: Python task wrapper now converts positional args to kwargs in v2 mode

Previously only **kwargs were passed to _next_step(), silently dropping
positional arguments. Extract shared _merge_args() helper used by both
v1 and v2 paths.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: replace unwrap() with proper error propagation in WAC arg serialization

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: add workspace_id filter to v2_job queries in WAC dispatch

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: prevent race condition in WAC child dispatch

Restructure dispatch to save checkpoint + suspend parent + seed child
checkpoints in a single transaction BEFORE pushing child jobs. This
ensures a fast child can't complete before the parent is suspended.

Also wrap InlineCheckpoint save + running reset in a transaction to
prevent corrupted state on crash.

Use ULID for pre-generated child job IDs (consistent with rest of API).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: include step key and child job ID in WAC error propagation

Move step_key lookup before the success check so failed child errors
include which task failed, the child job ID, and the original error.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* docs: document WAC determinism contract and step dispatch semantics

- Document that workflow functions must be deterministic across replays
- Document that WacStepDispatch.script/args are metadata, not dispatch targets
- Add comments on counter-based key allocation

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: tighten WAC v2 detection to reduce false positives

Replace naive substring matching with line-aware checks that skip
comments and look for specific patterns:
- TS: import from "windmill-client" containing workflow/task
- Python: @workflow and @task decorators with wmill import

Extracted shared helpers in wac_executor.rs used by both executors.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: show failed steps in WacGraph when workflow completes with errors

When flowDone is true and a pending step isn't in completedSteps,
mark it as 'failed' instead of 'running'. The failed state CSS and
XCircle icon were already defined but never triggered.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: unsuspend and fail parent when WAC child push fails

Previously if a child push failed mid-batch, the parent remained
suspended with suspend = num_steps but fewer children, hanging until
the 14-day timeout. Now the push loop catches errors and unsuspends
the parent before returning the error.

Also adds source hash validation: if the script content changes between
replays, the job fails with a clear error instead of silently feeding
stale checkpoint data into wrong steps.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: clear suspend_until when unsuspending WAC parent

Set suspend_until = NULL alongside suspend = 0 in both the child
failure and all-children-complete paths, so the parent doesn't rely
on subtle pull query invariants to be re-picked-up.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* test: add exhaustive edge case tests for WAC v2 SDK

fix: make TS task wrapper non-async to fix unawaited task flush

The async wrapper caused microtask-based thenable auto-resolution that
fired .then() and threw StepSuspend before _flushPending() could capture
unawaited steps — making the flush mechanism completely broken. Now the
thenable is returned directly without async wrapping. Backward compatible
with v1 (all code paths still return awaitables).

Tests added (59 TS + 66 Python) covering: full sequential lifecycle,
step after parallel, parallel after parallel, conditional on step result,
empty/single-task workflows, 10+ steps, falsy value preservation, inline
steps, mixed step/task, unawaited flush, child mode with parallel,
key determinism, large parallel groups, and complex mixed patterns.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: atomic checkpoint updates to prevent parallel child race condition

Replace read-modify-write pattern in handle_wac_child_completion with
atomic SQL operations:
- completed_steps merged via jsonb_set(... || jsonb_build_object(...))
  so concurrent children on different workers don't overwrite each other
- suspend counter decremented atomically with RETURNING to determine
  "all done" condition (instead of checking completed_steps in memory)
- suspend_until cleared in the same atomic decrement statement

Before this fix, two parallel children completing simultaneously could
both load the same checkpoint, each add their step, and save — the
second write would overwrite the first, silently losing a child result
and leaving the parent suspended forever.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: cancel already-pushed children on partial WAC dispatch failure

When pushing child jobs sequentially, if pushing child N fails, children
1..N-1 are already running. Previously the error handler only unsuspended
the parent, leaving orphaned children that would complete and corrupt the
checkpoint state (decrementing suspend on an already-unsuspended parent,
potentially causing duplicate step execution on re-run).

Now on partial failure:
1. Cancel all already-pushed children (prevents them from completing
   and corrupting checkpoint state)
2. Clear pending_steps from checkpoint (so parent doesn't think
   children are outstanding on re-run)
3. Then unsuspend parent (so the error propagates)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: skip WAC duration write and child check for non-WAC parents

The duration write to workflow_as_code_status was running for every
non-flow child with a parent (error handlers, success handlers,
run_script children), even though it was only intended for WAC jobs.

Add WHERE workflow_as_code_status IS NOT NULL to skip non-WAC parents
entirely. Piggyback RETURNING pending_steps.job_ids on the same query
so WAC v2 child completion needs zero extra DB round-trips on the
success path.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: seed child checkpoint in same transaction as push

The child checkpoint insert was happening before the child job was
pushed, violating the FK constraint on v2_job_status. Move it into
the push transaction so the job row exists and the child can't be
picked up before its checkpoint is ready.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: set running=false when WAC parent suspends for child dispatch

The parent job kept running=true after suspending, so workers wouldn't
pick it up when children completed and suspend reached 0. The parent
only advanced when the zombie job detector reset it (~90s). Now the
dispatch suspend sets running=false so the parent is immediately
eligible for pickup.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: WAC parent suspend/unsuspend lifecycle

Keep running=true when suspending the parent so the normal pull query
(WHERE running=false) never picks it up. Keep suspend_until non-null
when decrementing suspend to 0 so the suspended pull query
(WHERE suspend_until IS NOT NULL AND suspend<=0) picks it up.

Previously: setting running=false caused infinite restart loops because
the normal pull query has no suspend check and would immediately re-pick
the parent. Clearing suspend_until on the last child prevented the
suspended pull from ever seeing it, requiring the 90s zombie detector.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: add approval primitive, flow child completion, timeline fixes for WAC v2

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: add error propagation, task options, sleep, and parallel for WAC v2

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* test: fix python SDK tests to use name-based keys and add new test coverage

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: address WAC v2 review findings (sleep timing, error marker, atomicity)

- Fix sleep using suspend=1 instead of 0 to enforce actual delay
- Add approval/sleep resume injection to Python executor
- Fix TS SDK concurrency_limit mapping (was reading wrong property)
- Namespace error marker as __wmill_error to avoid user data collision
- Wrap child completion SQL in transaction for atomicity
- Decrement suspend even when step key is missing (prevents hang)
- Expand TASK_RE to handle export const, let, var, generics
- Validate step key uniqueness before dispatch
- Log warning on checkpoint deserialization failure
- Remove unimplemented delete_after_use from SDKs
- Add TaskError exception class to Python SDK with diagnostic context
- Fix extra positional args handling and add functools.wraps
- Improve getParamNames to handle typed/destructured params

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* sqlx

* sqlx

* test: add WAC v1 e2e integration tests for TS and Python

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: revert fake test versions in typescript-client

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: remove unused WacGraph component and strip wacToFlow to isWorkflowAsCode

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: extract shared approval/sleep resume logic into wac_executor

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-09 19:39:24 +00:00

319 lines
10 KiB
Rust

/*
* Author: Ruben Fiszel
* Copyright: Windmill Labs, Inc 2022
* This file and its contents are licensed under the AGPLv3 License.
* Please see the included NOTICE for copyright information and
* LICENSE-AGPL for a copy of the license.
*/
use std::panic::Location;
use axum::body::Body;
use axum::response::Response;
use axum::{response::IntoResponse, response::Json};
use hyper::StatusCode;
use sqlx::migrate::MigrateError;
use thiserror::Error;
use tokio::io;
pub type Result<T> = std::result::Result<T, Error>;
pub type JsonResult<T> = std::result::Result<Json<T>, Error>;
#[derive(Debug, Error)]
pub enum Error {
#[error("Bad gateway: {0}")]
BadGateway(String),
#[error("Bad config: {0}")]
BadConfig(String),
#[error("Connecting to database: {0}")]
ConnectingToDatabase(String),
#[error("Not found: {0}")]
NotFound(String),
#[error("Not authorized: {0}")]
NotAuthorized(String),
#[error("Metric not found: {0}")]
MetricNotFound(String),
#[error("Permission denied: {0}")]
PermissionDenied(String),
#[error("Require Admin privileges for {0}")]
RequireAdmin(String),
#[error("{0}")]
ExecutionErr(String),
#[error("{0}")]
ResultTooLarge(String),
#[error("IoErr: {error:#} @{location:#}")]
IoErr { error: io::Error, location: String },
#[error("Utf8Err: {error:#} @{location:#}")]
Utf8Err { error: std::string::FromUtf8Error, location: String },
#[error("UuidErr: {error:#} @{location:#}")]
UuidErr { error: uuid::Error, location: String },
#[error("SqlErr: {error:#} @{location:#}")]
SqlErr { error: sqlx::Error, location: String },
#[error("SerdeJson: {error:#} @{location:#}")]
SerdeJson { error: serde_json::Error, location: String },
#[error("Bad request: {0}")]
BadRequest(String),
#[error("Quota exceeded: {0}")]
QuotaExceeded(String),
#[error("Internal: {0}")]
InternalErr(String),
#[error("Internal: {message} @{location}")]
InternalErrLoc { message: String, location: String },
#[error("Internal: {0}: {1}")]
InternalErrAt(&'static Location<'static>, String),
#[error("HexErr: {error:#} @{location:#}")]
HexErr { error: hex::FromHexError, location: String },
#[error("Migrating database: {0}")]
DatabaseMigration(#[from] MigrateError),
#[error("Non-zero exit status for {0}: {1}")]
ExitStatus(String, i32),
#[error("ExecutionRawError: {0}")]
ExecutionRawError(Box<serde_json::value::RawValue>),
#[error("Error: {error:#} @{location:#}")]
Anyhow { error: anyhow::Error, location: String },
#[error("Error: {0:#?}")]
JsonErr(serde_json::Value),
#[error("{0}")]
AIError(String),
#[error("{0}")]
AlreadyCompleted(String),
#[error("WAC job suspended: {0}")]
WacSuspended(String),
#[error("Find python error: {0}")]
FindPythonError(String),
#[error("Problem with arguments: {0}")]
ArgumentErr(String),
#[error("{1}")]
Generic(StatusCode, String),
#[error("{feature} is unavailable due to some workers being behind. Do not use the feature or make sure all workers run at least {min_version}")]
WorkersAreBehind { feature: String, min_version: String },
#[error(
"Breaking change was introduced in v{version} ({feature}). Follow this migration guide: {guide_url}"
)]
MigrationNeeded { version: String, feature: String, guide_url: url::Url },
#[error("{0} is unavailable. It is possible for this worker to be behind.")]
FeatureUnavailable(String),
}
impl Error {
pub fn name(&self) -> &str {
match self {
Self::ExecutionErr(_) => "ExecutionErr",
Self::ResultTooLarge(_) => "ResultTooLarge",
Self::BadRequest(_) => "BadRequest",
Self::QuotaExceeded(_) => "QuotaExceeded",
Self::InternalErr(_) => "InternalErr",
Self::InternalErrLoc { .. } => "InternalErr",
Self::InternalErrAt(_, _) => "InternalErr",
Self::Anyhow { .. } => "Anyhow",
Self::JsonErr(_) => "JsonErr",
Self::AIError(_) => "AIError",
Self::AlreadyCompleted(_) => "AlreadyCompleted",
Self::WacSuspended(_) => "WacSuspended",
Self::FindPythonError(_) => "FindPythonError",
Self::ArgumentErr(_) => "ArgumentErr",
Self::Generic(_, _) => "Generic",
Self::IoErr { .. } => "IoErr",
Self::Utf8Err { .. } => "Utf8Err",
Self::UuidErr { .. } => "UuidErr",
Self::SqlErr { .. } => "SqlErr",
Self::SerdeJson { .. } => "SerdeJson",
Self::HexErr { .. } => "HexErr",
Self::DatabaseMigration(_) => "DatabaseMigration",
Self::ExitStatus(_, _) => "ExitStatus",
Self::ExecutionRawError(_) => "ExecutionRawError",
Self::BadGateway(_) => "BadGateway",
Self::BadConfig(_) => "BadConfig",
Self::ConnectingToDatabase(_) => "ConnectingToDatabase",
Self::NotFound(_) => "NotFound",
Self::NotAuthorized(_) => "NotAuthorized",
Self::MetricNotFound(_) => "MetricNotFound",
Self::PermissionDenied(_) => "PermissionDenied",
_ => "InternalErr",
}
}
}
fn prettify_location(location: &'static Location<'static>) -> String {
location
.to_string()
.split("/")
.last()
.unwrap_or("unknown")
.to_string()
}
impl From<anyhow::Error> for Error {
#[track_caller]
fn from(e: anyhow::Error) -> Self {
Self::Anyhow { error: e, location: prettify_location(std::panic::Location::caller()) }
}
}
impl From<sqlx::Error> for Error {
#[track_caller]
fn from(e: sqlx::Error) -> Self {
Self::SqlErr { error: e, location: prettify_location(std::panic::Location::caller()) }
}
}
impl From<uuid::Error> for Error {
#[track_caller]
fn from(e: uuid::Error) -> Self {
Self::UuidErr { error: e, location: prettify_location(std::panic::Location::caller()) }
}
}
impl From<std::string::FromUtf8Error> for Error {
#[track_caller]
fn from(e: std::string::FromUtf8Error) -> Self {
Self::Utf8Err { error: e, location: prettify_location(std::panic::Location::caller()) }
}
}
impl From<io::Error> for Error {
#[track_caller]
fn from(e: io::Error) -> Self {
Self::IoErr { error: e, location: prettify_location(std::panic::Location::caller()) }
}
}
impl From<hex::FromHexError> for Error {
#[track_caller]
fn from(e: hex::FromHexError) -> Self {
Self::HexErr { error: e, location: prettify_location(std::panic::Location::caller()) }
}
}
impl From<serde_json::Error> for Error {
#[track_caller]
fn from(e: serde_json::Error) -> Self {
Self::SerdeJson { error: e, location: prettify_location(std::panic::Location::caller()) }
}
}
impl From<url::ParseError> for Error {
#[track_caller]
fn from(e: url::ParseError) -> Self {
Self::ArgumentErr(format!("Cannot parse provided url. \ne: {e}"))
}
}
impl From<tokio::time::error::Elapsed> for Error {
fn from(value: tokio::time::error::Elapsed) -> Self {
Self::InternalErr(value.to_string())
}
}
impl From<semver::Error> for Error {
fn from(e: semver::Error) -> Self {
Self::ArgumentErr(format!("Cannot parse provided semver: {e}"))
}
}
impl Error {
/// https://docs.rs/anyhow/1/anyhow/struct.Error.html#display-representations
pub fn alt(&self) -> String {
format!("{:#}", self)
}
pub fn dbg(&self) -> String {
format!("{:?}", self)
}
pub fn relocate_internal(self, loc: &'static Location<'static>) -> Self {
match self {
Self::InternalErrLoc { message, .. }
| Self::InternalErrAt(_, message)
| Self::InternalErr(message) => Self::InternalErrAt(loc, message),
_ => self,
}
}
#[track_caller]
pub fn internal_err<T: AsRef<str>>(msg: T) -> Self {
Self::InternalErrLoc {
message: msg.as_ref().to_string(),
location: prettify_location(std::panic::Location::caller()),
}
}
}
pub fn relocate_internal(loc: &'static Location<'static>) -> impl FnOnce(Error) -> Error {
move |e| e.relocate_internal(loc)
}
pub fn to_anyhow<T: 'static + std::error::Error + Send + Sync>(e: T) -> anyhow::Error {
From::from(e)
}
impl IntoResponse for Error {
fn into_response(self) -> axum::response::Response {
let status = match self {
Self::NotFound(_) => axum::http::StatusCode::NOT_FOUND,
Self::NotAuthorized(_) => axum::http::StatusCode::UNAUTHORIZED,
Self::RequireAdmin(_) | Self::PermissionDenied(_) => axum::http::StatusCode::FORBIDDEN,
Self::SqlErr { .. }
| Self::BadRequest(_)
| Self::AIError(_)
| Self::QuotaExceeded(_) => axum::http::StatusCode::BAD_REQUEST,
Self::BadGateway(_) => axum::http::StatusCode::BAD_GATEWAY,
Self::Generic(status_code, _) => status_code,
_ => axum::http::StatusCode::INTERNAL_SERVER_ERROR,
};
let e = &self;
if matches!(status, axum::http::StatusCode::NOT_FOUND) {
tracing::warn!(message = e.to_string());
} else {
tracing::error!(message = e.to_string(), error = ?e);
};
let body = Body::from(e.to_string());
axum::response::Response::builder()
.header("Content-Type", "text/plain")
.status(status)
.body(body)
.unwrap()
}
}
pub trait OrElseNotFound<T> {
fn or_else_not_found(self, s: impl ToString) -> Result<T>;
}
impl<T> OrElseNotFound<T> for Option<T> {
fn or_else_not_found(self, s: impl ToString) -> Result<T> {
self.ok_or_else(|| Error::NotFound(s.to_string()))
}
}
// Make our own error that wraps `anyhow::Error`.
pub struct AppError(anyhow::Error);
// Tell axum how to convert `AppError` into a response.
impl IntoResponse for AppError {
fn into_response(self) -> Response {
let body = Body::from(self.0.to_string());
tracing::error!(error = self.0.to_string());
axum::response::Response::builder()
.header("Content-Type", "text/plain")
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(body)
.unwrap()
}
}
// This enables using `?` on functions that return `Result<_, anyhow::Error>` to turn them into
// `Result<_, AppError>`. That way you don't need to do that manually
impl<E> From<E> for AppError
where
E: Into<anyhow::Error>,
{
fn from(err: E) -> Self {
Self(err.into())
}
}