starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
sql-s3-streaming-issue
windmill/backend/tests/worker.rs
T
Ruben Fiszel 0dbd9c1231 perf: eliminate dual-connection DB pool contention across worker, queue, and api (#9798) 
* perf: eliminate dual-connection DB pool contention across worker, queue, and api

Reuse the held transaction (or move pool reads before begin()) instead of
checking out a second pool connection while a tx is open, extending the
fix from #9789/#7861. Targets the per-worker pool (max 5) hot paths plus
several server-pool API handlers.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix: pass owned pool to get_email_from_permissioned_as in http trigger handler

The generified signature takes impl PgExecutor; the http trigger handler
passed &db where db is already &DB, yielding &&Pool which does not impl
PgExecutor (only surfaced under the full feature set in CI).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix: keep RLS-exposed reads on the non-RLS pool and isolate flow-eval reads in a savepoint

Addresses review of the dual-connection sweep:

- worker_flow: wrap the stop_after_all_iters_if reads in a SAVEPOINT. The
  caller swallows the error and keeps using tx, so a DB read failure must
  not leave the outer transaction aborted (it would fail the later commit).
  Matches the previous pool-read semantics.

- Revert reads that were moved onto an RLS (user_db) transaction back to the
  non-RLS pool, since RLS row-visibility/role context can change results:
  push_scheduled_job (email/tag/settings lookups; reachable with a user_db
  tx from api-schedule/api-flows), push_inner native-retry dedicated_worker
  routing (RLS isolation variants), resources.rs app-namespace folder
  auto-create (non-admins must not be blocked), and the script archive/delete
  UPDATEs. Non-RLS db.begin() reuse and move-before-begin are kept.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test: failpoint proving the stop_after_all_iters_if savepoint isolates an aborted read

Adds a worker-crate failpoints feature and a data-driven hook: when the
stop_after_all_iters_if expr is the magic sentinel, the in-evaluation read runs
SELECT 1/0 to abort its (savepoint) transaction. The test asserts the flow still
completes (iteration marked failed) — which only holds if the savepoint keeps the
outer status-update transaction committable. Without the savepoint the abort would
poison the outer tx and the job would never complete.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 21:33:17 +00:00

5723 lines
199 KiB
Rust
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
use serde::de::DeserializeOwned;
use serial_test::serial;
#[cfg(feature = "enterprise")]
use chrono::Timelike;
use serde::Deserialize;
use serde_json::json;
use sqlx::{types::Uuid, Pool, Postgres};
#[cfg(feature = "enterprise")]
use tokio::time::{timeout, Duration};
#[cfg(feature = "python")]
use windmill_api_client::types::{CreateFlowBody, RawScript};
#[cfg(feature = "enterprise")]
use windmill_api_client::types::{EditSchedule, NewSchedule, ScriptArgs};
#[cfg(feature = "deno_core")]
use windmill_common::flows::InputTransform;
#[cfg(any(feature = "python", feature = "deno_core"))]
use windmill_common::flow_status::RestartedFrom;
use windmill_common::{
flows::FlowValue,
jobs::{JobPayload, RawCode},
scripts::ScriptLang,
};
use windmill_test_utils::*;
#[cfg(feature = "enterprise")]
use futures::StreamExt;
use windmill_common::flows::FlowModule;
use windmill_common::flows::FlowModuleValue;
// async fn _print_job(id: Uuid, db: &Pool<Postgres>) -> Result<(), anyhow::Error> {
// tracing::info!(
// "{:#?}",
// get_job_by_id(db.begin().await?, "test-workspace", id)
// .await?
// .0
// );
// Ok(())
// }
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_iteration(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{
"value": {
"type": "forloopflow",
"iterator": { "type": "javascript", "expr": "result.items" },
"skip_failures": false,
"modules": [{
"value": {
"input_transforms": {
"n": {
"type": "javascript",
"expr": "flow_input.iter.value",
},
},
"type": "rawscript",
"language": "python3",
"content": "def main(n):\n if 1 < n:\n raise StopIteration(n)",
},
}],
},
}],
}))
.unwrap();
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("items", json!([]))
.run_until_complete(&db, false, server.addr.port())
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([]));
/* Don't actually test that this does 257 jobs or that will take forever. */
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("items", json!((0..257).collect::<Vec<_>>()))
.run_until_complete(&db, false, server.addr.port())
.await
.json_result()
.unwrap();
assert!(matches!(result, serde_json::Value::Array(_)));
assert!(result[2]["error"]
.as_object()
.unwrap()
.get("message")
.unwrap()
.as_str()
.unwrap()
.contains("2"));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_iteration_parallel(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{
"value": {
"type": "forloopflow",
"iterator": { "type": "javascript", "expr": "result.items" },
"skip_failures": false,
"parallel": true,
"modules": [{
"value": {
"input_transforms": {
"n": {
"type": "javascript",
"expr": "flow_input.iter.value",
},
},
"type": "rawscript",
"language": "python3",
"content": "def main(n):\n if 1 < n:\n raise StopIteration(n)",
},
}],
},
}],
}))
.unwrap();
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("items", json!([]))
.run_until_complete(&db, false, server.addr.port())
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([]));
/* Don't actually test that this does 257 jobs or that will take forever. */
let job =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("items", json!((0..50).collect::<Vec<_>>()))
.run_until_complete(&db, false, server.addr.port())
.await;
// println!("{:#?}", job);
let result = job.json_result().unwrap();
assert!(matches!(result, serde_json::Value::Array(_)));
assert!(result[2]["error"]
.as_object()
.unwrap()
.get("message")
.unwrap()
.as_str()
.unwrap()
.contains("2"));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let numbers = "export function main() { return [1, 2, 3]; }";
let doubles = "export function main(n) { return n * 2; }";
let flow = {
use windmill_common::flows::{FlowModule, FlowModuleValue};
FlowValue {
modules: vec![
FlowModule {
id: "a".to_string(),
value: FlowModuleValue::RawScript {
input_transforms: Default::default(),
language: ScriptLang::Deno,
content: numbers.to_string(),
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
is_trigger: None,
assets: None,
}
.into(),
stop_after_if: Default::default(),
stop_after_all_iters_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
},
FlowModule {
id: "b".to_string(),
value: FlowModuleValue::ForloopFlow {
iterator: InputTransform::Javascript { expr: "result".to_string() },
skip_failures: false,
parallel: false,
squash: None,
parallelism: None,
modules: vec![FlowModule {
id: "c".to_string(),
value: FlowModuleValue::RawScript {
input_transforms: [(
"n".to_string(),
InputTransform::Javascript {
expr: "flow_input.iter.value".to_string(),
},
)]
.into(),
language: ScriptLang::Deno,
content: doubles.to_string(),
path: None,
lock: None,
tag: None,
concurrency_settings:
windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}
.into(),
stop_after_if: Default::default(),
stop_after_all_iters_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
}],
modules_node: None,
}
.into(),
stop_after_if: Default::default(),
stop_after_all_iters_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
},
],
same_worker: false,
..Default::default()
}
};
let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let port = server.addr.port();
for i in 0..50 {
println!("deno flow iteration: {}", i);
let job = run_job_in_new_worker_until_complete(&db, false, job.clone(), port).await;
// println!("job: {:#?}", job.flow_status);
let result = job.json_result().unwrap();
assert_eq!(result, serde_json::json!([2, 4, 6]), "iteration: {}", i);
}
Ok(())
}
#[cfg(feature = "python")]
#[sqlx::test(fixtures("base"))]
async fn test_identity(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{
"value": {
"type": "rawscript",
"language": "python3",
"content": "def main(): return 42",
}}, {
"value": {
"type": "identity",
},
}, {
"value": {
"type": "identity",
},
}, {
"value": {
"type": "identity",
},
}],
}))
.unwrap();
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.run_until_complete(&db, false, server.addr.port())
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!(42));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let write_file = r#"export async function main(loop: boolean, i: number, path: string) {
await Deno.writeTextFile(`./shared/${path}`, `${loop} ${i}`);
}"#
.to_string();
let flow = FlowValue {
modules: vec![
FlowModule {
id: "a".to_string(),
value: FlowModuleValue::RawScript {
input_transforms: [
(
"loop".to_string(),
InputTransform::Static { value: windmill_common::worker::to_raw_value(&false) },
),
("i".to_string(), InputTransform::Static { value: windmill_common::worker::to_raw_value(&1) }),
(
"path".to_string(),
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"outer.txt") },
),
]
.into(),
language: ScriptLang::Deno,
content: write_file.clone(),
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}.into(),
stop_after_if: Default::default(),
stop_after_all_iters_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
},
FlowModule {
id: "b".to_string(),
value: FlowModuleValue::ForloopFlow {
iterator: InputTransform::Static { value: windmill_common::worker::to_raw_value(&[1, 2, 3]) },
skip_failures: false,
parallel: false,
squash: None,
parallelism: None,
modules: vec![
FlowModule {
id: "d".to_string(),
value: FlowModuleValue::RawScript {
input_transforms: [
(
"i".to_string(),
InputTransform::Javascript {
expr: "flow_input.iter.value".to_string(),
},
),
(
"loop".to_string(),
InputTransform::Static { value: windmill_common::worker::to_raw_value(&true) },
),
(
"path".to_string(),
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"inner.txt") },
),
]
.into(),
language: ScriptLang::Deno,
content: write_file,
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}.into(),
stop_after_if: Default::default(),
stop_after_all_iters_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
},
FlowModule {
id: "e".to_string(),
value: FlowModuleValue::RawScript {
input_transforms: [(
"path".to_string(),
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"inner.txt") },
), (
"path2".to_string(),
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"outer.txt") },
)]
.into(),
language: ScriptLang::Deno,
content: r#"export async function main(path: string, path2: string) {
return await Deno.readTextFile(`./shared/${path}`) + "," + await Deno.readTextFile(`./shared/${path2}`);
}"#
.to_string(),
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}.into(),
stop_after_if: Default::default(),
stop_after_all_iters_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
},
],
modules_node: None,
}.into(),
stop_after_if: Default::default(),
stop_after_all_iters_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
},
FlowModule {
id: "c".to_string(),
value: FlowModuleValue::RawScript {
input_transforms: [
(
"loops".to_string(),
InputTransform::Javascript { expr: "results.b".to_string() },
),
(
"path".to_string(),
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"outer.txt") },
),
(
"path2".to_string(),
InputTransform::Static { value: windmill_common::worker::to_raw_value(&"inner.txt") },
),
]
.into(),
language: ScriptLang::Deno,
content: r#"export async function main(path: string, loops: string[], path2: string) {
return await Deno.readTextFile(`./shared/${path}`) + "," + loops + "," + await Deno.readTextFile(`./shared/${path2}`);
}"#
.to_string(),
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}.into(),
stop_after_if: Default::default(),
stop_after_all_iters_if: Default::default(),
summary: Default::default(),
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
delete_after_use: None,
delete_after_secs: None,
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
},
],
same_worker: true,
..Default::default()
};
let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, job.clone(), server.addr.port())
.await
.json_result()
.unwrap();
assert_eq!(
result,
serde_json::json!("false 1,true 1,false 1,true 2,false 1,true 3,false 1,true 3")
);
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_flow_result_by_id(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server: ApiServer = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return 42 }",
}
},
{
"value": {
"branches": [
{
"modules": [{
"value": {
"branches": [{"modules": [ {
"id": "d",
"value": {
"input_transforms": {"v": {"type": "javascript", "expr": "results.a"}},
"type": "rawscript",
"language": "deno",
"content": "export function main(v){ return v }",
}
},]}],
"type": "branchall",
}
}],
}],
"type": "branchall",
},
}
],
}))
.unwrap();
let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, job.clone(), port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([[42]]));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_stop_after_if(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
// let server = ApiServer::start(db.clone()).await?;
// let port = server.addr.port();
let port = 123;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"id": "a",
"value": {
"input_transforms": { "n": { "type": "javascript", "expr": "flow_input.n" } },
"type": "rawscript",
"language": "python3",
"content": "def main(n): return n",
},
"stop_after_if": {
"expr": "result < 0",
"skip_if_stopped": false,
},
},
{
"id": "b",
"value": {
"input_transforms": { "n": { "type": "javascript", "expr": "results.a" } },
"type": "rawscript",
"language": "python3",
"content": "def main(n): return f'last step saw {n}'",
},
},
],
}))
.unwrap();
let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = RunJob::from(job.clone())
.arg("n", json!(123))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(json!("last step saw 123"), result);
let cjob = RunJob::from(job.clone())
.arg("n", json!(-123))
.run_until_complete(&db, false, port)
.await;
let result = cjob.json_result().unwrap();
assert_eq!(json!(-123), result);
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_stop_after_if_nested(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
// let server = ApiServer::start(db.clone()).await?;
// let port = server.addr.port();
let port = 123;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"id": "a",
"value": {
"branches": [{"modules": [{
"id": "b",
"value": {
"input_transforms": { "n": { "type": "javascript", "expr": "flow_input.n" } },
"type": "rawscript",
"language": "python3",
"content": "def main(n): return n",
},
"stop_after_if": {
"expr": "result < 0",
"skip_if_stopped": false,
}}]}],
"type": "branchall",
"parallel": false,
},
},
{
"id": "c",
"value": {
"input_transforms": { "n": { "type": "javascript", "expr": "results.a" } },
"type": "rawscript",
"language": "python3",
"content": "def main(n): return f'last step saw {n}'",
},
},
],
}))
.unwrap();
let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = RunJob::from(job.clone())
.arg("n", json!(123))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(json!("last step saw [123]"), result);
let cjob = RunJob::from(job.clone())
.arg("n", json!(-123))
.run_until_complete(&db, false, port)
.await;
let result = cjob.json_result().unwrap();
assert_eq!(json!([-123]), result);
Ok(())
}
#[cfg(all(feature = "deno_core", feature = "python"))]
#[sqlx::test(fixtures("base"))]
async fn test_python_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let numbers = "def main(): return [1, 2, 3]";
let doubles = "def main(n): return n * 2";
let flow: FlowValue = serde_json::from_value(serde_json::json!( {
"modules": [
{
"value": {
"type": "rawscript",
"language": "python3",
"content": numbers,
},
},
{
"value": {
"type": "forloopflow",
"iterator": { "type": "javascript", "expr": "result" },
"skip_failures": false,
"modules": [{
"value": {
"input_transforms": {
"n": {
"type": "javascript",
"expr": "flow_input.iter.value",
},
},
"type": "rawscript",
"language": "python3",
"content": doubles,
},
}],
},
},
],
}))
.unwrap();
for i in 0..10 {
println!("python flow iteration: {}", i);
let result = run_job_in_new_worker_until_complete(
&db,
false,
JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None },
port,
)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([2, 4, 6]), "iteration: {i}");
}
Ok(())
}
#[cfg(feature = "python")]
#[sqlx::test(fixtures("base"))]
async fn test_python_flow_2(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"value": {
"input_transforms": {},
"type": "rawscript",
"content": "import wmill\ndef main(): return \"Hello\"",
"language": "python3"
},
}
]
}))
.unwrap();
for i in 0..10 {
println!("python flow iteration: {}", i);
let result = run_job_in_new_worker_until_complete(
&db,
false,
JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None },
port,
)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("Hello"), "iteration: {i}");
}
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_go_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
package inner
import "fmt"
func main(derp string) (string, error) {
fmt.Println("Hello, 世界")
return fmt.Sprintf("hello %s", derp), nil
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Go,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("derp", json!("world"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("hello world"));
Ok(())
}
#[cfg(feature = "rust")]
#[sqlx::test(fixtures("base"))]
async fn test_rust_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
fn main(world: String) -> Result<String, String> {
println!("Which world to greet today?");
Ok(format!("Hello {}!", world))
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Rust,
cache_ignore_s3_path: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
cache_ttl: None,
dedicated_worker: None,
modules: None,
tag: None,
}))
.arg("world", json!("Hyrule"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("Hello Hyrule!"));
Ok(())
}
#[cfg(feature = "csharp")]
#[sqlx::test(fixtures("base"))]
async fn test_csharp_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
using System;
class Script
{
public static string Main(string world, int b = 2)
{
Console.WriteLine($"Hello {world} - {b}. This is a log line");
return $"Hello {world} - {b}";
}
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::CSharp,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("world", json!("Arakis"))
.arg("b", json!(3))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("Hello Arakis - 3"));
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_bash_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
msg="$1"
echo "hello $msg"
"#
.to_owned();
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Bash,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("msg", json!("world"))
.run_until_complete(&db, false, port)
.await;
assert_eq!(job.json_result(), Some(json!("hello world")));
Ok(())
}
#[sqlx::test(fixtures("base", "wmill_cli_test"))]
async fn test_bash_wmill_variable_get(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
// The bash script uses wmill CLI to get the variable value.
// The worker sets WM_TOKEN, WM_WORKSPACE, and BASE_INTERNAL_URL as env vars,
// and the CLI auto-configures from them when no workspace is explicitly set.
// We point WMILL_CONFIG_DIR to a clean temp dir so no local active workspace interferes.
let content = r#"
export WMILL_CONFIG_DIR=$(mktemp -d)
result=$(wmill variable get "u/test-user/test_var" --json | jq -r .value)
echo "$result"
"#
.to_owned();
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Bash,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.run_until_complete(&db, false, port)
.await;
assert_eq!(job.json_result(), Some(json!("hello from variable")));
Ok(())
}
#[sqlx::test(fixtures("base", "wmill_cli_test"))]
async fn test_bash_wmill_resource_get(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
// The bash script uses wmill CLI to get the resource value.
// We point WMILL_CONFIG_DIR to a clean temp dir so no local active workspace interferes.
let content = r#"
export WMILL_CONFIG_DIR=$(mktemp -d)
result=$(wmill resource get "u/test-user/test_res" --json | jq -c .value)
echo "$result"
"#
.to_owned();
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Bash,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.run_until_complete(&db, false, port)
.await;
// Bash echo outputs are returned as strings, so the JSON is a string value
assert_eq!(
job.json_result(),
Some(json!("{\"host\":\"localhost\",\"port\":5432}"))
);
Ok(())
}
#[cfg(feature = "rlang")]
#[sqlx::test(fixtures("base"))]
async fn test_r_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
main <- function(msg) {
return(paste("hello", msg))
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Rlang,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("msg", json!("world"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, json!("hello world"));
Ok(())
}
#[cfg(feature = "rlang")]
#[sqlx::test(fixtures("base", "wmill_cli_test"))]
async fn test_r_get_variable(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
main <- function() {
return(get_variable("u/test-user/test_var"))
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Rlang,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, json!("hello from variable"));
Ok(())
}
#[cfg(feature = "rlang")]
#[sqlx::test(fixtures("base", "wmill_cli_test"))]
async fn test_r_get_resource(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
main <- function() {
return(get_resource("u/test-user/test_res"))
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Rlang,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, json!({"host": "localhost", "port": 5432}));
Ok(())
}
#[cfg(feature = "nu")]
#[sqlx::test(fixtures("base"))]
async fn test_nu_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
def main [ msg: string ] {
"hello " + $msg
}
"#
.to_owned();
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Nu,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("msg", json!("world"))
.run_until_complete(&db, false, port)
.await;
assert_eq!(job.json_result(), Some(json!("hello world")));
Ok(())
}
#[cfg(feature = "nu")]
#[sqlx::test(fixtures("base"))]
async fn test_nu_job_full(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
def main [
# Required
## Primitive
a
b: any
c: bool
d: float
e: datetime
f: string
j: nothing
## Nesting
g: record
h: list<string>
i: table
# Optional
m?
n = "foo"
o: any = "foo"
p?: any
# TODO: ...x
] {
0
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Nu,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("a", json!("3"))
.arg("b", json!("null"))
.arg("c", json!(true))
.arg("d", json!(3.0))
.arg("e", json!("2024-09-24T10:00:00.000Z"))
.arg("f", json!("str"))
.arg("j", json!(null))
.arg("g", json!({"a": 32}))
.arg("h", json!(["foo"]))
.arg(
"i",
json!([
{"a": 1, "b": "foo", "c": true},
{"a": 2, "b": "baz", "c": false}
]),
)
.arg("n", json!("baz"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!(0));
Ok(())
}
#[cfg(feature = "java")]
#[sqlx::test(fixtures("base"))]
async fn test_java_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
public class Main {
public static Object main(
// Primitive
int a,
float b,
// Objects
Integer age,
Float d
){
return "hello world";
}
}
"#
.to_owned();
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Java,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("a", json!(3))
.arg("b", json!(3.0))
.arg("age", json!(30))
.arg("d", json!(3.0))
.run_until_complete(&db, false, port)
.await;
assert_eq!(job.json_result(), Some(json!("hello world")));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_nativets_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
export async function main(name: string): Promise<string> {
return `hello ${name}`;
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Nativets,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("name", json!("world"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("hello world"));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_nativets_job_with_args(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
export async function main(a: number, b: number): Promise<number> {
return a + b;
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Nativets,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("a", json!(3))
.arg("b", json!(7))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!(10));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_nativets_job_object_return(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
export async function main(items: string[]): Promise<{ count: number; items: string[] }> {
return { count: items.length, items };
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Nativets,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("items", json!(["a", "b", "c"]))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, json!({"count": 3, "items": ["a", "b", "c"]}));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_nativets_job_datetime(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
// nativets passes Date-typed args as strings (no auto-conversion unlike Bun/Deno)
let content = r#"
export async function main(a: Date) {
return typeof a;
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Nativets,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("string"));
Ok(())
}
#[sqlx::test(fixtures("base"))]
#[serial(pg_cache)]
async fn test_postgresql_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
windmill_worker::pg_executor::clear_pg_cache().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
-- $1 name
SELECT 'hello ' || $1::text AS result;
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("name", json!("world"))
.arg(
"database",
json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"}),
)
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, json!([{"result": "hello world"}]));
Ok(())
}
#[sqlx::test(fixtures("base"))]
#[serial(pg_cache)]
async fn test_postgresql_cached_connection_resets_session(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
use std::sync::atomic::Ordering;
use windmill_worker::pg_executor::{clear_pg_cache, CACHE_HITS};
initialize_tracing().await;
// Clear stale connections from prior tests.
clear_pg_cache().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let hits_before = CACHE_HITS.load(Ordering::Relaxed);
let db_arg = json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"});
let make_pg_job = |content: String| {
RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(
)
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("database", db_arg.clone())
};
// Run 3 jobs against the same database resource so that at least the
// second→third transition exercises the cached-connection path.
//
// Job 1: warm up — creates and (if eligible) caches the connection.
let result1 = make_pg_job("SELECT 1 as n;".into())
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result1, json!([{"n": 1}]));
// Job 2: mutate session state — set a custom search_path.
let result2 = make_pg_job(
"SET search_path TO pg_catalog;\nSELECT current_setting('search_path') as sp;".into(),
)
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(
result2,
json!([{"sp": "pg_catalog"}]),
"second job should see the custom search_path it just set"
);
// Job 3: read search_path — RESET ALL (on cached conn) should have restored
// the default.
let result3 = make_pg_job("SELECT current_setting('search_path') as sp;".into())
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
let sp = result3[0]["sp"].as_str().unwrap();
assert_ne!(
sp, "pg_catalog",
"search_path must be reset between jobs, got: {sp}"
);
// Verify the cached-connection path was actually exercised.
let hits_after = CACHE_HITS.load(Ordering::Relaxed);
assert!(
hits_after > hits_before,
"expected at least one cache hit across the 3 jobs, got {hits_before} -> {hits_after}"
);
Ok(())
}
/// Runs multiple PG jobs through a SINGLE worker (like production) to verify
/// that SET ROLE / search_path changes do not leak across jobs.
#[sqlx::test(fixtures("base"))]
#[serial(pg_cache)]
async fn test_postgresql_single_worker_session_isolation(db: Pool<Postgres>) -> anyhow::Result<()> {
use std::sync::atomic::Ordering;
use windmill_common::worker::Connection;
use windmill_worker::pg_executor::{clear_pg_cache, CACHE_HITS};
initialize_tracing().await;
clear_pg_cache().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let hits_before = CACHE_HITS.load(Ordering::Relaxed);
let db_arg = json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"});
let make_pg_job = |content: String| {
RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(
)
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("database", db_arg.clone())
};
// Push all 3 jobs into the queue BEFORE starting the worker, so a single
// worker processes them all sequentially — matching production behaviour.
let id1 = make_pg_job("SELECT current_user as cu;".into())
.push(&db)
.await;
let id2 = make_pg_job("SET ROLE postgres;\nSET search_path TO pg_catalog;\nSELECT current_user as cu, current_setting('search_path') as sp;".into())
.push(&db)
.await;
let id3 =
make_pg_job("SELECT current_user as cu, current_setting('search_path') as sp;".into())
.push(&db)
.await;
// Run ONE worker that processes all three jobs.
let ids = [id1, id2, id3];
let listener = listen_for_completed_jobs(&db).await;
in_test_worker(
Connection::Sql(db.clone()),
async {
use futures::StreamExt;
let mut remaining = ids
.iter()
.copied()
.collect::<std::collections::HashSet<_>>();
let mut listener = listener;
while !remaining.is_empty() {
if let Some(id) = listener.next().await {
remaining.remove(&id);
}
}
},
port,
)
.await;
let r1 = completed_job(id1, &db).await.json_result().unwrap();
let r2 = completed_job(id2, &db).await.json_result().unwrap();
let r3 = completed_job(id3, &db).await.json_result().unwrap();
// Job 1: baseline — should be the connecting user (postgres)
assert_eq!(r1[0]["cu"], "postgres");
// Job 2: SET ROLE + SET search_path took effect within the job
assert_eq!(r2[0]["sp"], "pg_catalog");
// Job 3: neither the role nor the search_path should leak from job 2
let cu3 = r3[0]["cu"].as_str().unwrap();
let sp3 = r3[0]["sp"].as_str().unwrap();
assert_eq!(
cu3, "postgres",
"SET ROLE must not leak across jobs, got: {cu3}"
);
assert_ne!(
sp3, "pg_catalog",
"search_path must not leak across jobs, got: {sp3}"
);
// Verify caching actually happened (jobs 2 and 3 should hit the cache).
let hits_after = CACHE_HITS.load(Ordering::Relaxed);
assert!(
hits_after > hits_before,
"expected cache hits, got {hits_before} -> {hits_after}"
);
Ok(())
}
/// Runs 100 varied PG jobs through a single worker, verifying every job
/// succeeds, the cache is used for 99 of them, and session state never leaks.
#[sqlx::test(fixtures("base"))]
#[serial(pg_cache)]
async fn test_postgresql_100_jobs_cached(db: Pool<Postgres>) -> anyhow::Result<()> {
use std::sync::atomic::Ordering;
use windmill_common::worker::Connection;
use windmill_worker::pg_executor::{clear_pg_cache, CACHE_HITS};
initialize_tracing().await;
clear_pg_cache().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let hits_before = CACHE_HITS.load(Ordering::Relaxed);
let db_arg = json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"});
let make_pg_job = |content: String| {
RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(
)
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("database", db_arg.clone())
};
const N: usize = 100;
let mut ids = Vec::with_capacity(N);
// Push 100 varied jobs — mix of session-mutating and read-only queries.
for i in 0..N {
let content = match i % 5 {
// Plain SELECT with different values each time
0 => format!("SELECT {} as n, current_user as cu, current_setting('search_path') as sp;", i),
// SET search_path then read it
1 => format!("SET search_path TO pg_catalog;\nSELECT {} as n, current_setting('search_path') as sp;", i),
// SET ROLE then read it
2 => "SET ROLE postgres;\nSELECT current_user as cu;".to_string(),
// Multi-statement
3 => format!("SELECT 1;\nSELECT {} as n;", i),
// Read-only with math
_ => format!("SELECT {} + {} as n;", i, i * 2),
};
ids.push(make_pg_job(content).push(&db).await);
}
// Run ONE worker for all 100 jobs.
let listener = listen_for_completed_jobs(&db).await;
let id_set = ids
.iter()
.copied()
.collect::<std::collections::HashSet<_>>();
in_test_worker(
Connection::Sql(db.clone()),
async {
use futures::StreamExt;
let mut remaining = id_set;
let mut listener = listener;
while !remaining.is_empty() {
if let Some(id) = listener.next().await {
remaining.remove(&id);
}
}
},
port,
)
.await;
// Verify all 100 jobs succeeded.
for (i, id) in ids.iter().enumerate() {
let cjob = completed_job(*id, &db).await;
assert!(cjob.success, "job {i} (id={id}) failed: {:?}", cjob.result);
}
// Spot-check: jobs that followed a SET search_path should NOT see pg_catalog.
// Pattern: job i%5==1 sets search_path, job i%5==2 follows — should be clean.
for i in (2..N).step_by(5) {
let cjob = completed_job(ids[i], &db).await;
let result = cjob.json_result().unwrap();
let cu = result[0]["cu"].as_str().unwrap_or("N/A");
assert_eq!(cu, "postgres", "job {i}: SET ROLE leaked, got {cu}");
}
for i in (0..N).step_by(5) {
let cjob = completed_job(ids[i], &db).await;
let result = cjob.json_result().unwrap();
let sp = result[0]["sp"].as_str().unwrap_or("N/A");
assert_ne!(sp, "pg_catalog", "job {i}: search_path leaked, got {sp}");
}
// Verify caching: first job creates the connection, remaining 99 should hit cache.
let hits_after = CACHE_HITS.load(Ordering::Relaxed);
let new_hits = hits_after - hits_before;
assert!(
new_hits >= (N as u64 - 1),
"expected at least {} cache hits, got {new_hits}",
N - 1
);
Ok(())
}
/// Cover the (Value × arg_t) combinations that #8988 broke. Each shape mirrors
/// what the windmill-client SDK or a hand-written PG script can emit:
///
/// - bare `$N` with no inline cast and no `-- $N name (type)` declaration
/// (parser defaults the otyp to "text"). The user's value can be any JSON
/// shape; the eventual column type is whatever the SQL context implies.
/// - inline `$N::TYPE` casts (the SDK's default for bare `${value}`).
/// - `CAST($N AS T)` syntax (the SDK strips its own cast when this pattern
/// surrounds the value, so the parser sees a bare `$N`).
/// - explicit declaration: `-- $N name (type)`.
///
/// Pre-fix, the dispatch asserted `Type::TEXT` for parser-defaulted args, so
/// e.g. a `Value::Bool` bound to `Box<bool>` failed at the encoder with
/// "cannot convert between the Rust type `bool` and the Postgres type `text`"
/// before the query ever reached the server.
#[sqlx::test(fixtures("base"))]
#[serial(pg_cache)]
async fn test_postgresql_arg_type_combinations(db: Pool<Postgres>) -> anyhow::Result<()> {
use windmill_worker::pg_executor::clear_pg_cache;
initialize_tracing().await;
clear_pg_cache().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let db_arg = json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"});
// Build a fresh schema for this test so we don't trip over prior runs.
let setup = r#"
DROP SCHEMA IF EXISTS wm_pg_arg_combo_test CASCADE;
CREATE SCHEMA wm_pg_arg_combo_test;
CREATE TABLE wm_pg_arg_combo_test.bugbool (flag bool);
CREATE TABLE wm_pg_arg_combo_test.sdkbug (n int, f double precision);
CREATE TABLE wm_pg_arg_combo_test.bugmix (id int, name text, payload jsonb, tags text[]);
CREATE TABLE wm_pg_arg_combo_test.allcols (
c_bool bool,
c_int2 smallint,
c_int4 int,
c_int8 bigint,
c_float4 real,
c_float8 double precision,
c_numeric numeric,
c_text text,
c_varchar varchar(64),
c_uuid uuid,
c_date date,
c_time time,
c_ts timestamp,
c_tstz timestamptz,
c_json json,
c_jsonb jsonb,
c_int_arr int[],
c_text_arr text[]
);
CREATE TYPE wm_pg_arg_combo_test.color AS ENUM ('red','green','blue');
CREATE TABLE wm_pg_arg_combo_test.enumtbl (c wm_pg_arg_combo_test.color);
"#;
RunJob::from(JobPayload::Code(RawCode {
hash: None,
content: setup.to_owned(),
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("database", db_arg.clone())
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
// Each case: (name, content, args, expected_result)
let cases: Vec<(&str, String, serde_json::Value, serde_json::Value)> = vec![
// === parser-default "text" otyp (bare $N), value drives the binding ===
(
"bool via CAST AS bool (parser default text)",
"-- $1 arg1\nINSERT INTO wm_pg_arg_combo_test.bugbool VALUES (CAST($1 AS bool)) RETURNING flag".to_owned(),
json!({"arg1": true}),
json!([{"flag": true}]),
),
(
"bare $1 with bool into bool col",
"-- $1 arg1\nINSERT INTO wm_pg_arg_combo_test.bugbool VALUES ($1) RETURNING flag".to_owned(),
json!({"arg1": false}),
json!([{"flag": false}]),
),
(
"bare $1 with bool into text via implicit cast bool->text",
"-- $1 arg1\nSELECT $1::text AS s".to_owned(),
json!({"arg1": true}),
json!([{"s": "true"}]),
),
(
"bare $1 with int into text via implicit cast int->text",
"-- $1 arg1\nSELECT $1::text AS s".to_owned(),
json!({"arg1": 42}),
json!([{"s": "42"}]),
),
(
"object via CAST AS jsonb (parser default text)",
"-- $1 arg1\nSELECT CAST($1 AS jsonb) AS v".to_owned(),
json!({"arg1": {"k": 1}}),
json!([{"v": {"k": 1}}]),
),
(
"string '42' via CAST AS int (parser default text)",
"-- $1 arg1\nSELECT CAST($1 AS int) AS v".to_owned(),
json!({"arg1": "42"}),
json!([{"v": 42}]),
),
(
"NULL into bool col via CAST",
"-- $1 arg1\nINSERT INTO wm_pg_arg_combo_test.bugbool VALUES (CAST($1 AS bool)) RETURNING flag".to_owned(),
json!({"arg1": null}),
json!([{"flag": null}]),
),
// === SDK happy path — inline ::TYPE injected by the client ===
(
"SDK shape: $1::BIGINT, $2::DOUBLE PRECISION",
"-- $1 arg1\n-- $2 arg2\nINSERT INTO wm_pg_arg_combo_test.sdkbug VALUES ($1::BIGINT, $2::DOUBLE PRECISION) RETURNING n, f".to_owned(),
json!({"arg1": 42, "arg2": 3.14}),
json!([{"n": 42, "f": 3.14}]),
),
(
"SDK shape: $1::TEXT with int target via explicit ::int",
"-- $1 arg1\nSELECT $1::TEXT::int AS v".to_owned(),
json!({"arg1": "7"}),
json!([{"v": 7}]),
),
// === explicit declaration -- $N name (type) ===
(
"explicit decl (text)",
"-- $1 arg1 (text)\nSELECT $1 AS v".to_owned(),
json!({"arg1": "hello"}),
json!([{"v": "hello"}]),
),
(
"explicit decl (jsonb) with object",
"-- $1 arg1 (jsonb)\nSELECT $1 AS v".to_owned(),
json!({"arg1": {"k": [1, 2]}}),
json!([{"v": {"k": [1, 2]}}]),
),
// === mixed args: int, text, jsonb, text[] ===
(
"mixed: int + text + jsonb + text[]",
"-- $1 arg1\n-- $2 arg2\n-- $3 arg3\n-- $4 arg4\nINSERT INTO wm_pg_arg_combo_test.bugmix VALUES ($1::int, $2::text, $3::jsonb, $4::text[]) RETURNING *".to_owned(),
json!({"arg1": 7, "arg2": "hello", "arg3": {"k": 1}, "arg4": ["a", "b"]}),
json!([{"id": 7, "name": "hello", "payload": {"k": 1}, "tags": ["a", "b"]}]),
),
// === Every PG type, SDK happy path (inline ::TYPE) ===
(
"all PG types via inline casts",
r#"-- $1 arg1
-- $2 arg2
-- $3 arg3
-- $4 arg4
-- $5 arg5
-- $6 arg6
-- $7 arg7
-- $8 arg8
-- $9 arg9
-- $10 arg10
-- $11 arg11
-- $12 arg12
-- $13 arg13
-- $14 arg14
-- $15 arg15
-- $16 arg16
-- $17 arg17
-- $18 arg18
INSERT INTO wm_pg_arg_combo_test.allcols VALUES (
$1::bool, $2::int2, $3::int4, $4::int8,
$5::real, $6::double precision, $7::numeric,
$8::text, $9::varchar,
$10::uuid, $11::date, $12::time, $13::timestamp, $14::timestamptz,
$15::json, $16::jsonb,
$17::int[], $18::text[]
) RETURNING c_bool, c_int4, c_int8, c_text, c_uuid, c_int_arr"#.to_owned(),
json!({
"arg1": true, "arg2": 1, "arg3": 2, "arg4": 3,
"arg5": 1.5, "arg6": 2.5, "arg7": 3.14,
"arg8": "hello", "arg9": "varhello",
"arg10": "550e8400-e29b-41d4-a716-446655440000",
"arg11": "2024-01-15", "arg12": "10:30:00",
"arg13": "2024-01-15T10:30:00", "arg14": "2024-01-15T10:30:00Z",
"arg15": {"k": 1}, "arg16": {"k": 2},
"arg17": [1,2,3], "arg18": ["a","b"]
}),
json!([{"c_bool": true, "c_int4": 2, "c_int8": 3, "c_text": "hello",
"c_uuid": "550e8400-e29b-41d4-a716-446655440000",
"c_int_arr": [1,2,3]}]),
),
// === Every PG type, declaration-style otyp (Python SDK shape) ===
(
"all PG types via -- $N name (TYPE) decls",
r#"-- $1 arg1 (bool)
-- $2 arg2 (int2)
-- $3 arg3 (int4)
-- $4 arg4 (int8)
-- $5 arg5 (real)
-- $6 arg6 (float8)
-- $7 arg7 (numeric)
-- $8 arg8 (text)
-- $9 arg9 (varchar)
-- $10 arg10 (uuid)
-- $11 arg11 (date)
-- $12 arg12 (time)
-- $13 arg13 (timestamp)
-- $14 arg14 (timestamptz)
-- $15 arg15 (json)
-- $16 arg16 (jsonb)
-- $17 arg17 (int[])
-- $18 arg18 (text[])
INSERT INTO wm_pg_arg_combo_test.allcols VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18
) RETURNING c_bool, c_int4, c_int8, c_text, c_int_arr, c_text_arr"#.to_owned(),
json!({
"arg1": false, "arg2": 4, "arg3": 5, "arg4": 6,
"arg5": 1.5, "arg6": 2.5, "arg7": 3.14,
"arg8": "world", "arg9": "varworld",
"arg10": "550e8400-e29b-41d4-a716-446655440000",
"arg11": "2024-01-15", "arg12": "10:30:00",
"arg13": "2024-01-15T10:30:00", "arg14": "2024-01-15T10:30:00Z",
"arg15": [1,2], "arg16": [3,4],
"arg17": [10,20], "arg18": ["x","y"]
}),
json!([{"c_bool": false, "c_int4": 5, "c_int8": 6, "c_text": "world",
"c_int_arr": [10,20], "c_text_arr": ["x","y"]}]),
),
// === Edge values per type ===
("int8 negative", "-- $1 arg1\nSELECT $1::int8 AS v".to_owned(), json!({"arg1": -42}), json!([{"v": -42}])),
("int8 zero", "-- $1 arg1\nSELECT $1::int8 AS v".to_owned(), json!({"arg1": 0}), json!([{"v": 0}])),
("int4 max", "-- $1 arg1\nSELECT $1::int4 AS v".to_owned(), json!({"arg1": 2147483647i64}), json!([{"v": 2147483647i64}])),
("int8 max", "-- $1 arg1\nSELECT $1::int8 AS v".to_owned(), json!({"arg1": 9223372036854775807i64}), json!([{"v": 9223372036854775807i64}])),
("float8 fraction", "-- $1 arg1\nSELECT $1::float8 AS v".to_owned(), json!({"arg1": 0.1 + 0.2}), json!([{"v": 0.1 + 0.2}])),
("empty string", "-- $1 arg1\nSELECT $1::text AS v".to_owned(), json!({"arg1": ""}), json!([{"v": ""}])),
("empty array", "-- $1 arg1\nSELECT $1::int[] AS v".to_owned(), json!({"arg1": []}), json!([{"v": []}])),
("empty object", "-- $1 arg1\nSELECT $1::jsonb AS v".to_owned(), json!({"arg1": {}}), json!([{"v": {}}])),
// === Bool roundtrip across every shape ===
("Bool/bare $1 → bool col", "-- $1 arg1\nSELECT $1 AS v".to_owned(), json!({"arg1": true}), json!([{"v": true}])),
("Bool/inline ::bool", "-- $1 arg1\nSELECT $1::bool AS v".to_owned(), json!({"arg1": true}), json!([{"v": true}])),
("Bool/decl (bool)", "-- $1 arg1 (bool)\nSELECT $1 AS v".to_owned(), json!({"arg1": true}), json!([{"v": true}])),
("Bool/CAST AS bool", "-- $1 arg1\nSELECT CAST($1 AS bool) AS v".to_owned(), json!({"arg1": false}), json!([{"v": false}])),
("Bool/CAST AS bool inside SELECT","-- $1 arg1\nSELECT CAST($1 AS bool) AS v WHERE CAST($1 AS bool) IS NOT NULL".to_owned(), json!({"arg1": true}), json!([{"v": true}])),
// === Object/Array via CAST AS jsonb (regression: non-text otyp via CAST) ===
("Object via CAST AS jsonb", "-- $1 arg1\nSELECT CAST($1 AS jsonb) AS v".to_owned(), json!({"arg1": {"a":1,"b":[2,3]}}), json!([{"v": {"a":1,"b":[2,3]}}])),
("Array via CAST AS jsonb", "-- $1 arg1\nSELECT CAST($1 AS jsonb) AS v".to_owned(), json!({"arg1": [1,"two",{"three":3}]}), json!([{"v": [1,"two",{"three":3}]}])),
("Object inline ::json", "-- $1 arg1\nSELECT $1::json AS v".to_owned(), json!({"arg1": {"k":1}}), json!([{"v": {"k":1}}])),
("Object decl (jsonb)", "-- $1 arg1 (jsonb)\nSELECT $1 AS v".to_owned(), json!({"arg1": {"k":1}}), json!([{"v": {"k":1}}])),
// === Numbers: implicit/explicit casts to text ===
("Number/CAST AS text", "-- $1 arg1\nSELECT CAST($1 AS text) AS v".to_owned(), json!({"arg1": 42}), json!([{"v": "42"}])),
("Number/inline ::text", "-- $1 arg1\nSELECT $1::text AS v".to_owned(), json!({"arg1": 42}), json!([{"v": "42"}])),
("Float/CAST AS text", "-- $1 arg1\nSELECT CAST($1 AS text) AS v".to_owned(), json!({"arg1": 3.14}), json!([{"v": "3.14"}])),
("Negative/CAST AS int8", "-- $1 arg1\nSELECT CAST($1 AS int8) AS v".to_owned(), json!({"arg1": -1}), json!([{"v": -1}])),
// === Strings parsed into typed targets ===
("String '42' as int", "-- $1 arg1\nSELECT $1::int AS v".to_owned(), json!({"arg1": "42"}), json!([{"v": 42}])),
("String '42' as bigint", "-- $1 arg1\nSELECT $1::bigint AS v".to_owned(), json!({"arg1": "42"}), json!([{"v": 42}])),
("String '1.5' as real", "-- $1 arg1\nSELECT $1::real AS v".to_owned(), json!({"arg1": "1.5"}), json!([{"v": 1.5}])),
("String uuid", "-- $1 arg1\nSELECT $1::uuid AS v".to_owned(), json!({"arg1": "550e8400-e29b-41d4-a716-446655440000"}), json!([{"v": "550e8400-e29b-41d4-a716-446655440000"}])),
// === NULL handling for every type ===
("Null/bool", "-- $1 arg1\nSELECT CAST($1 AS bool) AS v".to_owned(), json!({"arg1": null}), json!([{"v": null}])),
("Null/int", "-- $1 arg1\nSELECT CAST($1 AS int) AS v".to_owned(), json!({"arg1": null}), json!([{"v": null}])),
("Null/bigint", "-- $1 arg1\nSELECT CAST($1 AS bigint) AS v".to_owned(), json!({"arg1": null}), json!([{"v": null}])),
("Null/text", "-- $1 arg1\nSELECT $1::text AS v".to_owned(), json!({"arg1": null}), json!([{"v": null}])),
("Null/jsonb", "-- $1 arg1\nSELECT CAST($1 AS jsonb) AS v".to_owned(), json!({"arg1": null}), json!([{"v": null}])),
("Null/uuid", "-- $1 arg1\nSELECT CAST($1 AS uuid) AS v".to_owned(), json!({"arg1": null}), json!([{"v": null}])),
("Null/timestamp", "-- $1 arg1\nSELECT CAST($1 AS timestamp) AS v".to_owned(), json!({"arg1": null}), json!([{"v": null}])),
("Null/date", "-- $1 arg1\nSELECT CAST($1 AS date) AS v".to_owned(), json!({"arg1": null}), json!([{"v": null}])),
// === Multi-statement (datatable scripts often DROP/CREATE/INSERT) ===
(
"multi-statement: DROP/CREATE/INSERT",
r#"DROP TABLE IF EXISTS wm_pg_arg_combo_test.tmp_multi;
CREATE TABLE wm_pg_arg_combo_test.tmp_multi (n int, b bool);
-- $1 arg1
-- $2 arg2
INSERT INTO wm_pg_arg_combo_test.tmp_multi VALUES ($1::int, $2::bool) RETURNING *"#.to_owned(),
json!({"arg1": 10, "arg2": true}),
json!([{"n": 10, "b": true}]),
),
// === Same arg used in multiple positions (parser reorders) ===
(
"same arg twice",
"-- $1 arg1\nSELECT $1::int + $1::int AS v".to_owned(),
json!({"arg1": 5}),
json!([{"v": 10}]),
),
// === Custom enum (unrecognised arg_t → prepare fallback path) ===
// The unrecognised-arg_t fallback works when the user formats the
// value as the enum's text representation themselves, so the binding
// never has to encode a Rust String *as* an enum (which the
// tokio-postgres `ToSql` impls don't support):
(
"custom enum: text representation cast in SQL",
r#"-- $1 arg1
INSERT INTO wm_pg_arg_combo_test.enumtbl
VALUES (CAST($1::text AS wm_pg_arg_combo_test.color))
RETURNING c::text AS c"#
.to_owned(),
json!({"arg1": "green"}),
json!([{"c": "green"}]),
),
// === Sparse positional placeholders ($5, $50) ===
// The pre-fix `String::replace($5 → $1)` chain mangled `$50` into
// `$10`, breaking sparse-index queries. The regex-based renumbering
// handles them as distinct units.
(
"sparse $5 / $50 renumbering",
"-- $5 arg5\n-- $50 arg50\nSELECT $5::int AS a, $50::int AS b".to_owned(),
json!({"arg5": 5, "arg50": 50}),
json!([{"a": 5, "b": 50}]),
),
(
"sparse $5 / $50 reversed in SQL",
"-- $5 arg5\n-- $50 arg50\nSELECT $50::int AS a, $5::int AS b".to_owned(),
json!({"arg5": 5, "arg50": 50}),
json!([{"a": 50, "b": 5}]),
),
(
"sparse same arg used twice + sparse",
"-- $5 arg5\n-- $50 arg50\nSELECT $5::int + $5::int AS a, $50::int AS b".to_owned(),
json!({"arg5": 7, "arg50": 50}),
json!([{"a": 14, "b": 50}]),
),
// === Explicit (text) decl + non-string value: should coerce ===
// Without the otyp_inferred flag, the executor would bind the value's
// natural type (INT8/BOOL) and the WHERE comparison `text = int8` /
// `text = bool` would fail with "operator does not exist". With the
// flag, the parser tells the executor "user committed to text" and
// the value is JSON-stringified so `text = text` works.
(
"decl (text) + Number used in WHERE text comparison",
r#"-- $1 arg1 (text)
SELECT name FROM (VALUES ('42'::text)) AS t(name) WHERE name = $1"#
.to_owned(),
json!({"arg1": 42}),
json!([{"name": "42"}]),
),
(
"decl (text) + Bool used in WHERE text comparison",
r#"-- $1 arg1 (text)
SELECT name FROM (VALUES ('true'::text)) AS t(name) WHERE name = $1"#
.to_owned(),
json!({"arg1": true}),
json!([{"name": "true"}]),
),
(
"decl (varchar) + Number used in WHERE",
r#"-- $1 arg1 (varchar)
SELECT name FROM (VALUES ('99'::varchar)) AS t(name) WHERE name = $1"#
.to_owned(),
json!({"arg1": 99}),
json!([{"name": "99"}]),
),
// === Bare $N (parser-default text) + non-string value: bind native ===
// The user wrote no annotation — we bind the value's natural type so
// it works against whatever column the SQL eventually targets.
(
"bare $1 + Bool into bool col",
"-- $1 arg1\nSELECT $1 = true AS v".to_owned(),
json!({"arg1": true}),
json!([{"v": true}]),
),
// === Custom enum (Kind::Enum) — round-trip via AnyTextValue ===
// Pre-fix this failed at the encoder ("cannot convert String → color")
// because vanilla tokio_postgres' ToSql/FromSql for String reject
// Kind::Enum. The wrapper accepts enum kinds in both directions.
(
"enum: explicit ::wm_pg_arg_combo_test.color cast",
r#"-- $1 arg1
INSERT INTO wm_pg_arg_combo_test.enumtbl
VALUES ($1::wm_pg_arg_combo_test.color) RETURNING c"#
.to_owned(),
json!({"arg1": "blue"}),
json!([{"c": "blue"}]),
),
(
"enum: SELECT a literal value cast to enum",
"-- $1 arg1\nSELECT $1::wm_pg_arg_combo_test.color AS c".to_owned(),
json!({"arg1": "red"}),
json!([{"c": "red"}]),
),
// === Extended String→numeric/real/double/oid/bool arms (#10) ===
(
"String '3.14' → numeric",
"-- $1 arg1\nSELECT $1::numeric AS v".to_owned(),
json!({"arg1": "3.14"}),
json!([{"v": 3.14}]),
),
(
"String '1.5' → real",
"-- $1 arg1\nSELECT $1::real AS v".to_owned(),
json!({"arg1": "1.5"}),
json!([{"v": 1.5}]),
),
(
"String '2.5' → double",
"-- $1 arg1\nSELECT $1::double precision AS v".to_owned(),
json!({"arg1": "2.5"}),
json!([{"v": 2.5}]),
),
(
"String 'true' → bool",
"-- $1 arg1\nSELECT $1::bool AS v".to_owned(),
json!({"arg1": "true"}),
json!([{"v": true}]),
),
(
"String 't' → bool",
"-- $1 arg1\nSELECT $1::bool AS v".to_owned(),
json!({"arg1": "t"}),
json!([{"v": true}]),
),
(
"String '0' → bool false",
"-- $1 arg1\nSELECT $1::bool AS v".to_owned(),
json!({"arg1": "0"}),
json!([{"v": false}]),
),
(
"String '42' → oid",
"-- $1 arg1\nSELECT $1::oid AS v".to_owned(),
json!({"arg1": "42"}),
json!([{"v": 42}]),
),
// === String literals containing $N must NOT be renumbered ===
// Sparse positional args force a rewrite pass; the literal
// `'price: $5'` and the comment `-- mention $5` must survive intact.
(
"renumber must skip $N inside string literal",
r#"-- $5 arg5
-- $50 arg50
SELECT 'price: $5' AS lbl, $5::int + $50::int AS sum"#
.to_owned(),
json!({"arg5": 1, "arg50": 2}),
json!([{"lbl": "price: $5", "sum": 3}]),
),
// === Multi-word PG type names with [] array suffix ===
// Pre-fix: `transform_types_with_spaces` returned a `&str` alias
// ("double" / "varchar" / "timestamptz" / …) and dropped the trailing
// `[]`, so the dispatch routed `Value::Array` through `Type::JSONB`
// and the server failed with "cannot cast type jsonb to <T>[]".
(
"multi-word array: double precision[]",
"-- $1 a\nSELECT $1::double precision[] AS v".to_owned(),
json!({"a": [1.5, 2.5]}),
json!([{"v": [1.5, 2.5]}]),
),
(
"multi-word array: character varying[]",
"-- $1 a\nSELECT $1::character varying[] AS v".to_owned(),
json!({"a": ["x", "y"]}),
json!([{"v": ["x", "y"]}]),
),
(
"multi-word array: timestamp without time zone[]",
"-- $1 a\nSELECT $1::timestamp without time zone[] AS v".to_owned(),
json!({"a": ["2024-01-15T10:30:00"]}),
json!([{"v": ["2024-01-15T10:30:00"]}]),
),
// === Stringified primitives in array args ===
// Mirror the scalar `Value::String → <numeric type>` arms so values
// sent as e.g. `["1.5", "2.5"]` for `numeric[]` (typical for bulk-
// loading via `unnest` or BigInt-stringified arrays) round-trip
// instead of erroring with "Mixed types in array".
(
"numeric[] from stringified decimals",
"-- $1 a\nSELECT $1::numeric[] AS v".to_owned(),
json!({"a": ["1.5", "2.5"]}),
json!([{"v": [1.5, 2.5]}]),
),
(
"int[] from stringified ints",
"-- $1 a\nSELECT $1::int[] AS v".to_owned(),
json!({"a": ["1", "2", "3"]}),
json!([{"v": [1, 2, 3]}]),
),
(
"bool[] from stringified bools",
"-- $1 a\nSELECT $1::bool[] AS v".to_owned(),
json!({"a": ["true", "f", "yes"]}),
json!([{"v": [true, false, true]}]),
),
];
for (name, content, args, expected) in cases {
let mut job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(
)
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("database", db_arg.clone());
for (k, v) in args.as_object().unwrap() {
job = job.arg(k, v.clone());
}
let result = job
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap_or_else(|| panic!("case '{name}': no json result"));
assert_eq!(result, expected, "case '{name}' mismatch");
}
Ok(())
}
/// Pooler-safety regression test for #8988: when every arg has a resolvable
/// otyp (the common SDK case), the dispatch must use unnamed prepared
/// statements (`query_typed_raw`) and *must not* leak named statements
/// (`s0, s1, ...`) on the cached connection. Behind a transaction-mode pooler
/// (PgBouncer / Supabase pooler / RDS Proxy), accumulated names get dropped
/// when the prepare and execute land on different backend connections, which
/// is what produced the original "prepared statement \"sN\" does not exist"
/// errors.
#[sqlx::test(fixtures("base"))]
#[serial(pg_cache)]
async fn test_postgresql_no_named_statements_after_typed_args(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
use windmill_worker::pg_executor::clear_pg_cache;
initialize_tracing().await;
clear_pg_cache().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let db_arg = json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"});
let make_pg_job = |content: String| {
RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(
)
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("database", db_arg.clone())
};
// Run several SDK-shape queries (parameterised, all args resolvable) on the
// same cached connection. None of them should land in `pg_prepared_statements`.
for i in 0..5 {
let result = make_pg_job(format!(
"-- $1 arg1\n-- $2 arg2\nSELECT $1::BIGINT AS a, $2::TEXT AS b, {} AS i;",
i
))
.arg("arg1", json!(i))
.arg("arg2", json!(format!("v{i}")))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(
result,
json!([{"a": i, "b": format!("v{i}"), "i": i}]),
"iteration {i}"
);
}
// Use the *same* connection (cached one) to peek at pg_prepared_statements.
// Anything matching the tokio-postgres "s\d+" naming would mean the
// pooler-unsafe `prepare + query_raw` path was taken.
let probe = make_pg_job(
"SELECT count(*)::int AS n FROM pg_prepared_statements WHERE name ~ '^s[0-9]+$'".to_owned(),
)
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
let n = probe[0]["n"].as_i64().unwrap_or(-1);
assert_eq!(
n, 0,
"expected no leaked named prepared statements, found {n}"
);
Ok(())
}
/// Exercise the `prepare + query_raw` fallback path. The dispatch takes this
/// path when `otyp_to_pg_type` doesn't recognise the parser-derived arg_t —
/// typical for custom enums, domains, and extension types.
///
/// Vanilla `tokio_postgres`'s `ToSql for String` doesn't actually accept
/// `Kind::Enum` / `Kind::Domain`, so an end-to-end happy-path test of an
/// arbitrary custom type isn't possible without `postgres-derive`. What we
/// CAN lock in here is *which dispatch path runs*: when the arg_t is
/// unrecognised, the prepare path must be taken (the server resolves the
/// param type from the cast and the binding then errors at the encoder).
/// If a regression accidentally routes unrecognised types through
/// `query_typed_raw`, the failure mode flips: instead of "cannot convert
/// `String` to `<custom_type>`" we'd see "cannot convert `String` to
/// `text`" (because we'd assert TEXT). The error-text check below catches
/// that flip.
#[sqlx::test(fixtures("base"))]
#[serial(pg_cache)]
async fn test_postgresql_prepare_fallback_for_unrecognised_arg_t(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
use windmill_worker::pg_executor::clear_pg_cache;
initialize_tracing().await;
clear_pg_cache().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let db_arg = json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"});
let make_pg_job = |content: String| {
RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(
)
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("database", db_arg.clone())
};
// Set up a custom enum in a dedicated schema so the test is self-contained.
let setup = r#"
DROP SCHEMA IF EXISTS fallback_test CASCADE;
CREATE SCHEMA fallback_test;
CREATE TYPE fallback_test.color AS ENUM ('red', 'green', 'blue');
"#;
make_pg_job(setup.to_owned())
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
// Inline cast `::fallback_test.color` — the parser only captures
// `fallback_test` (regex stops at the dot), which otyp_to_pg_type won't
// recognise. Dispatch must take the prepare fallback path.
//
// Thanks to the AnyTextValue ToSql/FromSql wrapper, this case now
// round-trips end-to-end (the wrapper accepts Kind::Enum on both
// directions). Pre-fix, vanilla tokio_postgres rejected String → color
// and the user had to write `CAST($1::text AS color)` as a workaround.
let result = make_pg_job("-- $1 arg1\nSELECT $1::fallback_test.color AS c".to_owned())
.arg("arg1", json!("red"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, json!([{"c": "red"}]));
// Reading enum columns also goes through AnyTextValue's FromSql impl on
// the result side, so the value comes back as a JSON string.
let result = make_pg_job(
r#"-- $1 arg1
SELECT $1::fallback_test.color AS c1,
'green'::fallback_test.color AS c2"#
.to_owned(),
)
.arg("arg1", json!("blue"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, json!([{"c1": "blue", "c2": "green"}]));
Ok(())
}
/// Regression: custom enum / domain queries must not error with `prepared
/// statement "sN" does not exist` when run on a cached connection.
///
/// Root cause: when we used `DISCARD ALL` to reset the cached connection
/// between jobs, the included `DEALLOCATE ALL` deallocated *every* prepared
/// statement server-side — including the typeinfo statements that
/// tokio_postgres caches per-client to resolve custom-type Oids. The Rust
/// client still held `Statement` objects whose names the server had
/// forgotten, so the next custom-type query failed.
///
/// Fix: switched the cached-connection probe to `RESET ALL; UNLISTEN *;
/// CLOSE ALL;` which covers windmill's session-isolation needs (GUC reset,
/// listen channels, open cursors) without nuking the prepared-statement
/// cache. This test runs the failing pattern (enum query → domain query on
/// the same cached connection, several times) to lock the behaviour in.
#[sqlx::test(fixtures("base"))]
#[serial(pg_cache)]
async fn test_postgresql_custom_types_on_cached_connection(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
use windmill_worker::pg_executor::clear_pg_cache;
initialize_tracing().await;
clear_pg_cache().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let db_arg = json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"});
let make_pg_job = |content: String| {
RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(
)
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("database", db_arg.clone())
};
// Set up custom enum + domain types in a dedicated schema (avoid the
// `pg_*` reserved prefix for user schemas).
let setup = r#"
DROP SCHEMA IF EXISTS wm_pg_cached_test CASCADE;
CREATE SCHEMA wm_pg_cached_test;
CREATE TYPE wm_pg_cached_test.color AS ENUM ('red', 'green', 'blue');
CREATE DOMAIN wm_pg_cached_test.short_name AS TEXT CHECK (length(VALUE) BETWEEN 1 AND 10);
"#;
make_pg_job(setup.to_owned())
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
// Run a long alternating sequence of enum + domain queries on the same
// cached connection. Each query goes through the prepare-fallback path
// (otyp_to_pg_type doesn't recognise these custom-type names) and needs
// tokio_postgres' typeinfo cache to resolve the Oids server-side. Pre-fix
// this would fail intermittently with `prepared statement "sN" does not
// exist` after the first cached-conn reuse.
for i in 0..10 {
let r = make_pg_job("-- $1 arg1\nSELECT $1::wm_pg_cached_test.color AS c".to_owned())
.arg("arg1", json!(if i % 2 == 0 { "red" } else { "blue" }))
.run_until_complete(&db, false, port)
.await;
assert!(
r.success,
"iter {i} enum query failed (was the DISCARD ALL bug); result: {:?}",
r.result
);
let r = make_pg_job("-- $1 arg1\nSELECT $1::wm_pg_cached_test.short_name AS s".to_owned())
.arg("arg1", json!(format!("v{i}")))
.run_until_complete(&db, false, port)
.await;
assert!(
r.success,
"iter {i} domain query failed (was the DISCARD ALL bug); result: {:?}",
r.result
);
}
Ok(())
}
/// Security regression: a previous job that did `SET ROLE` or `SET SESSION
/// AUTHORIZATION` to a different role must not leak that role into the next
/// job that reuses the cached connection.
///
/// This is the case `RESET ALL` alone does *not* cover — neither SET ROLE
/// nor SET SESSION AUTHORIZATION are GUC parameters, so they survive
/// `RESET ALL`. We rely on `RESET SESSION AUTHORIZATION` (which subsumes
/// `RESET ROLE`) explicitly being part of the cached-connection probe.
///
/// The pre-existing `test_postgresql_single_worker_session_isolation` test
/// only did `SET ROLE postgres` (the connecting user), so the leak was
/// invisible — this test catches it by switching to a *different* role.
#[sqlx::test(fixtures("base"))]
#[serial(pg_cache)]
async fn test_postgresql_set_role_does_not_leak_across_cached_connection(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
use windmill_worker::pg_executor::clear_pg_cache;
initialize_tracing().await;
clear_pg_cache().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let db_arg = json!({"host": "localhost", "port": 5432, "dbname": "windmill", "user": "postgres", "password": "changeme"});
let make_pg_job = |content: String| {
RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Postgresql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(
)
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("database", db_arg.clone())
};
// Create a non-postgres role to switch to. Idempotent so the test survives
// re-runs against the same DB.
make_pg_job(
"DO $$ BEGIN \
IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'wm_isolation_test_role') THEN \
CREATE ROLE wm_isolation_test_role; \
END IF; \
END $$"
.to_owned(),
)
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
// Job 1: SET ROLE to a different role.
let r1 = make_pg_job(
"SET ROLE wm_isolation_test_role; SELECT current_user AS cu, session_user AS su".to_owned(),
)
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(r1[0]["cu"], "wm_isolation_test_role");
assert_eq!(r1[0]["su"], "postgres");
// Job 2 (cached connection reuse): role MUST be back to the connecting
// user. Pre-fix with `RESET ALL` alone, this would still see
// `wm_isolation_test_role` because RESET ALL doesn't cover SET ROLE.
let r2 = make_pg_job("SELECT current_user AS cu, session_user AS su".to_owned())
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(
r2[0]["cu"], "postgres",
"SET ROLE leaked across cached connection: current_user is {}",
r2[0]["cu"]
);
// Job 3: SET SESSION AUTHORIZATION (changes both current_user and
// session_user — RESET ALL does NOT touch this either).
let r3 = make_pg_job(
"SET SESSION AUTHORIZATION wm_isolation_test_role; \
SELECT current_user AS cu, session_user AS su"
.to_owned(),
)
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(r3[0]["cu"], "wm_isolation_test_role");
assert_eq!(r3[0]["su"], "wm_isolation_test_role");
// Job 4 (cached): both must be restored to the connecting user.
let r4 = make_pg_job("SELECT current_user AS cu, session_user AS su".to_owned())
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(
r4[0]["cu"], "postgres",
"SET SESSION AUTHORIZATION leaked across cached connection: current_user is {}",
r4[0]["cu"]
);
assert_eq!(
r4[0]["su"], "postgres",
"SET SESSION AUTHORIZATION leaked across cached connection: session_user is {}",
r4[0]["su"]
);
Ok(())
}
#[cfg(feature = "mysql")]
#[sqlx::test(fixtures("base"))]
async fn test_mysql_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
-- ? name (varchar)
SELECT ? AS result;
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Mysql,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("name", json!("world"))
.arg(
"database",
json!({"host": "localhost", "port": 3306, "user": "root", "password": "changeme", "database": "windmill_test"}),
)
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, json!([{"result": "world"}]));
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_bunnative_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
export async function main(name: string): Promise<string> {
return `hello ${name}`;
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Bunnative,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("name", json!("world"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("hello world"));
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_powershell_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
param($msg)
Write-Output "hello $msg"
"#
.to_owned();
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Powershell,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("msg", json!("world"))
.run_until_complete(&db, false, port)
.await;
assert_eq!(job.json_result(), Some(json!("hello world")));
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_powershell_param_block_with_attributes(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
param(
[Parameter(Mandatory=$true)]
[string]$Name,
[int]$Count = 3
)
Write-Output "$Name-$Count"
"#
.to_owned();
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Powershell,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("Name", json!("test"))
.arg("Count", json!(7))
.run_until_complete(&db, false, port)
.await;
assert_eq!(job.json_result(), Some(json!("test-7")));
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_powershell_error_caught(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
// Script with param block that throws an error — verifies the catch block works
let content = r#"
param($x)
throw "intentional error"
"#
.to_owned();
let job = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Powershell,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("x", json!(1))
.run_until_complete(&db, false, port)
.await;
assert!(!job.success, "job should fail on thrown error");
let result_str = serde_json::to_string(&job.result).unwrap_or_default();
assert!(
result_str.contains("An error occurred:"),
"catch block should output 'An error occurred:', got: {result_str}"
);
assert!(
result_str.contains("intentional error"),
"catch block should output the error message, got: {result_str}"
);
// Verify the catch block doesn't leak "Write-Output" as literal text
// (regression from the old broken line continuation in strict_termination_end)
let after_marker = result_str.split("An error occurred:").nth(1).unwrap_or("");
assert!(
!after_marker.starts_with("\\nWrite-Output"),
"catch block should not output literal 'Write-Output' text, got: {result_str}"
);
Ok(())
}
#[cfg(feature = "php")]
#[sqlx::test(fixtures("base"))]
async fn test_php_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
<?php
function main(string $name): string {
return "hello " . $name;
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Php,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("name", json!("world"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("hello world"));
Ok(())
}
#[cfg(feature = "ruby")]
#[sqlx::test(fixtures("base"))]
async fn test_ruby_job(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
def main(name)
"hello #{name}"
end
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Ruby,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("name", json!("world"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("hello world"));
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_bun_job_datetime(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
export async function main(a: Date) {
return typeof a;
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Bun,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("object"));
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_deno_job_datetime(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
export async function main(a: Date) {
return typeof a;
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Deno,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("object"));
Ok(())
}
/// Test that full .npmrc content works for deno jobs with private registries.
/// Requires:
/// - `TEST_NPMRC` environment variable set to the full .npmrc content
#[cfg(feature = "private_registry_test")]
#[sqlx::test(fixtures("base"))]
async fn test_deno_job_private_npmrc(db: Pool<Postgres>) -> anyhow::Result<()> {
use windmill_worker::NPMRC;
let npmrc_content = std::env::var("TEST_NPMRC")
.expect("TEST_NPMRC must be set when running private_registry_test");
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
{
let mut npmrc = NPMRC.write().await;
*npmrc = Some(npmrc_content.clone());
}
let content = r#"
import { greet } from "npm:@windmill-test/private-pkg";
export function main(name: string) {
return greet(name);
}
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
language: ScriptLang::Deno,
lock: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
modules: None,
tag: None,
}))
.arg("name", json!("World"))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
{
let mut npmrc = NPMRC.write().await;
*npmrc = None;
}
assert_eq!(
result,
serde_json::json!("Hello from private package, World!")
);
Ok(())
}
#[cfg(feature = "python")]
#[sqlx::test(fixtures("base"))]
async fn test_python_job_datetime_and_bytes(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let content = r#"
from datetime import datetime
def main(a: datetime, b: bytes):
return (isinstance(a, datetime), isinstance(b, bytes))
"#
.to_owned();
let result = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content,
path: None,
lock: None,
language: ScriptLang::Python3,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
modules: None,
tag: None,
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.arg("b", json!("dGVzdA=="))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([true, true]));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_empty_loop_1(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"id": "a",
"value": {
"type": "forloopflow",
"iterator": { "type": "static", "value": [] },
"modules": [
{
"value": {
"input_transforms": {
"n": {
"type": "javascript",
"expr": "flow_input.iter.value",
},
},
"type": "rawscript",
"language": "python3",
"content": "def main(n): return n",
},
}
],
},
},
{
"value": {
"input_transforms": {
"items": {
"type": "javascript",
"expr": "results.a",
},
},
"type": "rawscript",
"language": "python3",
"content": "def main(items): return sum(items)",
},
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!(0));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_invalid_first_step(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"value": {
"type": "forloopflow",
"iterator": { "type": "javascript", "expr": "flow_input" },
"modules": [
{
"value": {
"type": "identity",
},
}
],
},
},
{
"value": {
"type": "identity",
},
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let job = run_job_in_new_worker_until_complete(&db, false, flow, port).await;
assert!(
serde_json::to_string(&job.json_result().unwrap()).unwrap().contains("Expected an array value in the iterator expression, found: invalid type: map, expected a sequence at line 1 column 0")
);
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_empty_loop_2(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"value": {
"type": "forloopflow",
"iterator": { "type": "static", "value": [] },
"modules": [
{
"value": {
"input_transforms": {
"n": {
"type": "javascript",
"expr": "flow_input.iter.value",
},
},
"type": "rawscript",
"language": "python3",
"content": "def main(n): return n",
},
}
],
},
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([]));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_step_after_loop(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"id": "a",
"value": {
"type": "forloopflow",
"iterator": { "type": "static", "value": [2,3,4] },
"modules": [
{
"value": {
"input_transforms": {
"n": {
"type": "javascript",
"expr": "flow_input.iter.value",
},
},
"type": "rawscript",
"language": "python3",
"content": "def main(n): return n",
} ,
}
],
},
},
{
"value": {
"input_transforms": {
"items": {
"type": "javascript",
"expr": "results.a",
},
},
"type": "rawscript",
"language": "python3",
"content": "def main(items): return sum(items)",
},
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!(9));
Ok(())
}
fn module_add_item_to_list(i: i32, id: &str) -> serde_json::Value {
json!({
"id": format!("id_{}", i.to_string().replace("-", "_")),
"value": {
"input_transforms": {
"array": {
"type": "javascript",
"expr": format!("results.{id}"),
},
"i": {
"type": "static",
"value": json!(i),
}
},
"type": "rawscript",
"language": "deno",
"content": "export function main(array, i){ array.push(i); return array }",
}
})
}
fn module_failure() -> serde_json::Value {
json!({
"value": {
"input_transforms": {},
"type": "rawscript",
"language": "deno",
"content": "export function main(){ throw Error('failure') }",
}
})
}
#[sqlx::test(fixtures("base"))]
async fn test_branchone_simple(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [1] }",
}
},
{
"value": {
"branches": [],
"default": [module_add_item_to_list(2, "a")],
"type": "branchone",
}
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([1, 2]));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_branchone_with_cond(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [1] }",
}
},
{
"value": {
"branches": [{"expr": "results.a[0] == 1", "modules": [module_add_item_to_list(3, "a")]}],
"default": [module_add_item_to_list(2, "a")],
"type": "branchone",
}
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([1, 3]));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_branchall_sequential(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [1] }",
}
},
{
"value": {
"branches": [
{"modules": [module_add_item_to_list(2, "a")]},
{"modules": [module_add_item_to_list(3, "a")]}],
"type": "branchall",
"parallel": true,
}
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([[1, 2], [1, 3]]));
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_branchall_simple(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [1] }",
}
},
{
"value": {
"branches": [
{"modules": [module_add_item_to_list(2, "a")]},
{"modules": [module_add_item_to_list(3, "a")]}],
"type": "branchall",
}
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([[1, 2], [1, 3]]));
Ok(())
}
#[derive(Deserialize)]
struct ErrorResult {
error: NamedError,
}
#[derive(Deserialize)]
struct NamedError {
name: String,
}
#[sqlx::test(fixtures("base"))]
async fn test_branchall_skip_failure(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [1] }",
}
},
{
"value": {
"branches": [
{"modules": [module_failure()], "skip_failure": false},
{"modules": [module_add_item_to_list(3, "a")]}],
"type": "branchall",
}
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(
serde_json::from_value::<ErrorResult>(result.get(0).unwrap().clone())
.unwrap()
.error
.name,
"Error"
);
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [1] }",
}
},
{
"value": {
"branches": [
{"modules": [module_failure()], "skip_failure": true},
{"modules": [module_add_item_to_list(2, "a")]}
],
"type": "branchall",
}
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(
serde_json::from_value::<ErrorResult>(result.get(0).unwrap().clone())
.unwrap()
.error
.name,
"Error"
);
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_branchone_nested(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [] }",
}
},
module_add_item_to_list(1, "a"),
{
"id": "b",
"value": {
"branches": [
{
"expr": "false",
"modules": []
},
{
"expr": "true",
"modules": [ {
"value": {
"branches": [
{
"expr": "false",
"modules": []
}],
"default": [module_add_item_to_list(2, "id_1")],
"type": "branchone",
}
}]
},
],
"default": [module_add_item_to_list(-4, "id_1")],
"type": "branchone",
}
},
module_add_item_to_list(3, "b"),
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([1, 2, 3]));
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_branchall_nested(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main(){ return [1] }",
}
},
{
"value": {
"branches": [
{
"modules": [
{
"id": "b",
"value": {
"branches": [
{"modules": [module_add_item_to_list(2, "a")]},
{"modules": [module_add_item_to_list(3, "a")]}],
"type": "branchall",
}
}, {
"value": {
"branches": [
{"modules": [module_add_item_to_list(4, "b")]},
{"modules": [module_add_item_to_list(5, "b")]}],
"type": "branchall",
}
}
]
},
{"modules": [module_add_item_to_list(6, "a")]}],
// "parallel": false,
"type": "branchall",
}
},
],
}))
.unwrap();
let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let result = run_job_in_new_worker_until_complete(&db, false, flow, port)
.await
.json_result()
.unwrap();
println!("{:#?}", result);
assert_eq!(
result,
serde_json::json!([[[[1, 2], [1, 3], 4], [[1, 2], [1, 3], 5]], [1, 6]])
);
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_failure_module(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{
"id": "a",
"value": {
"input_transforms": {
"l": { "type": "javascript", "expr": "[]", },
"n": { "type": "javascript", "expr": "flow_input.n", },
},
"type": "rawscript",
"language": "deno",
"content": "export function main(n, l) { if (n == 0) throw Error(JSON.stringify(l)); return { l: [...l, 0] } }",
},
}, {
"id": "b",
"value": {
"input_transforms": {
"l": { "type": "javascript", "expr": "results.a.l", },
"n": { "type": "javascript", "expr": "flow_input.n", },
},
"type": "rawscript",
"language": "deno",
"content": "export function main(n, l) { if (n == 1) throw Error(JSON.stringify(l)); return { l: [...l, 1] } }",
},
}, {
"value": {
"input_transforms": {
"l": { "type": "javascript", "expr": "results.b.l", },
"n": { "type": "javascript", "expr": "flow_input.n", },
},
"type": "rawscript",
"language": "deno",
"content": "export function main(n, l) { if (n == 2) throw Error(JSON.stringify(l)); return { l: [...l, 2] } }",
},
}],
"failure_module": {
"value": {
"input_transforms": { "error": { "type": "javascript", "expr": "previous_result", } },
"type": "rawscript",
"language": "deno",
"content": "export function main(error) { return { 'from failure module': error } }",
}
},
}))
.unwrap();
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("n", json!(0))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert!(result["from failure module"]["error"]
.as_object()
.unwrap()
.get("message")
.unwrap()
.as_str()
.unwrap()
.contains("[]"));
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("n", json!(1))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert!(result["from failure module"]["error"]
.as_object()
.unwrap()
.get("message")
.unwrap()
.as_str()
.unwrap()
.contains("[0]"));
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("n", json!(2))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert!(result["from failure module"]["error"]
.as_object()
.unwrap()
.get("message")
.unwrap()
.as_str()
.unwrap()
.contains("[0,1]"));
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("n", json!(3))
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(json!({ "l": [0, 1, 2] }), result);
Ok(())
}
/// Push `flow`, run it on a real worker until its first step is running, then simulate
/// `monitor::handle_zombie_jobs` reaping that step unrecoverably (its worker crashed/OOM'd)
/// by calling `handle_job_error(..., unrecoverable = true, ...)` exactly as the monitor does.
/// Returns the flow's completed result.
#[cfg(feature = "deno_core")]
async fn run_flow_until_step_running_then_fail_unrecoverably(
db: &Pool<Postgres>,
port: u16,
flow: FlowValue,
) -> serde_json::Value {
use std::sync::atomic::AtomicU16;
use std::sync::Arc;
use tokio::sync::mpsc;
use windmill_common::auth::create_token_for_owner;
use windmill_common::client::AuthedClient;
use windmill_common::KillpillSender;
use windmill_queue::{get_queued_job_v2, MiniCompletedJob, SameWorkerPayload};
use windmill_worker::{JobCompletedSender, SameWorkerSender};
let flow_id =
RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None })
.push(db)
.await;
let db_ = db.clone();
in_test_worker(
db,
async move {
let db = db_;
// Wait for the first step to be running on the worker.
let step_job = loop {
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
let running = sqlx::query_scalar!(
"SELECT q.id FROM v2_job_queue q JOIN v2_job j USING (id)
WHERE j.parent_job = $1 AND q.running = true",
flow_id
)
.fetch_optional(&db)
.await
.unwrap();
if let Some(step_id) = running {
if let Some(job) = get_queued_job_v2(&db, &step_id).await.unwrap() {
break job;
}
}
};
// The dummy `same_worker_tx` mirrors the monitor, which has no live worker channel.
let (sw_tx, _sw_rx) = mpsc::channel::<SameWorkerPayload>(1);
let sw_tx = SameWorkerSender(sw_tx, Arc::new(AtomicU16::new(0)));
let (jc_tx, _jc_rx) = JobCompletedSender::new_never_used();
let (_kp_tx, kp_rx) = KillpillSender::new(1);
let token = create_token_for_owner(
&db,
"test-workspace",
"u/test-user",
"",
100,
"",
&Uuid::nil(),
None,
None,
)
.await
.unwrap();
let client = AuthedClient::new(
format!("http://localhost:{port}"),
"test-workspace".to_string(),
token,
None,
);
windmill_worker::result_processor::handle_job_error(
&db,
&client,
&MiniCompletedJob::from(step_job),
0,
None,
windmill_common::error::Error::ExecutionErr(
"simulated worker OOM crash".to_string(),
),
true, // unrecoverable
Some(&sw_tx),
"",
"test-monitor",
jc_tx,
&kp_rx,
)
.await;
// The flow should now run its failure module and complete.
loop {
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
let done = sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM v2_job_completed WHERE id = $1)",
flow_id
)
.fetch_one(&db)
.await
.unwrap()
.unwrap_or(false);
if done {
break;
}
}
},
port,
)
.await;
completed_job(flow_id, db).await.json_result().unwrap()
}
/// The hanging-forever first step: only ever completed by the simulated zombie handler.
#[cfg(feature = "deno_core")]
fn hanging_step_value() -> serde_json::Value {
serde_json::json!({
"input_transforms": {},
"type": "rawscript",
"language": "deno",
"content": "export async function main() { await new Promise((r) => setTimeout(r, 600000)); }",
})
}
/// The error handler module that marks itself so tests can assert it ran.
#[cfg(feature = "deno_core")]
fn marker_failure_module() -> serde_json::Value {
serde_json::json!({
"value": {
"input_transforms": { "error": { "type": "javascript", "expr": "previous_result", } },
"type": "rawscript",
"language": "deno",
"content": "export function main(error) { return { handled_unrecoverable: true, error } }",
}
})
}
/// Regression test for WIN-2070: a flow step that fails *unrecoverably* — e.g. its worker was
/// OOM-killed and the failure is surfaced by the zombie job handler — must still trigger the
/// flow's error handler (failure module). Previously, `unrecoverable` failures silently
/// completed the flow with an error and skipped the failure module entirely.
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_failure_module_triggered_on_unrecoverable_failure(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{ "id": "a", "value": hanging_step_value() }],
"failure_module": marker_failure_module(),
}))
.unwrap();
let result = run_flow_until_step_running_then_fail_unrecoverably(&db, port, flow).await;
server.close().await.unwrap();
assert_eq!(
result["handled_unrecoverable"],
json!(true),
"failure module (flow error handler) should run for an unrecoverable step failure, got: {result}"
);
Ok(())
}
/// WIN-2070: an unrecoverable failure must NOT be retried even when the step has a retry
/// policy — the original worker is gone, so retrying is pointless. It should fall straight
/// through to the error handler (failure module) instead.
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_unrecoverable_failure_skips_retry_runs_failure_module(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{
"id": "a",
"value": hanging_step_value(),
"retry": { "constant": { "attempts": 5, "seconds": 0 } },
}],
"failure_module": marker_failure_module(),
}))
.unwrap();
let result = run_flow_until_step_running_then_fail_unrecoverably(&db, port, flow).await;
server.close().await.unwrap();
assert_eq!(
result["handled_unrecoverable"],
json!(true),
"unrecoverable failure should skip retry and run the failure module, got: {result}"
);
Ok(())
}
/// WIN-2070: an unrecoverable failure on a `continue_on_error` step must still route to the
/// error handler rather than silently advancing to the next step (which would hide the worker
/// death). A normal failure on a `continue_on_error` step is tolerated and the flow continues;
/// a worker crash/OOM is not.
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_unrecoverable_failure_on_continue_on_error_runs_failure_module(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
// Step 'a' hangs (and tolerates failures via continue_on_error); step 'b' would run next
// if the unrecoverable failure were (wrongly) tolerated.
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"id": "a",
"value": hanging_step_value(),
"continue_on_error": true,
},
{
"id": "b",
"value": {
"input_transforms": {},
"type": "rawscript",
"language": "deno",
"content": "export function main() { return { ran_b: true }; }",
},
},
],
"failure_module": marker_failure_module(),
}))
.unwrap();
let result = run_flow_until_step_running_then_fail_unrecoverably(&db, port, flow).await;
server.close().await.unwrap();
assert_eq!(
result["handled_unrecoverable"],
json!(true),
"unrecoverable failure on a continue_on_error step should run the failure module, got: {result}"
);
assert!(
result.get("ran_b").is_none(),
"the step after a continue_on_error step must NOT run on an unrecoverable failure, got: {result}"
);
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_run_wait_result_early_return_with_failure_module(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"input_transforms": {},
"content": "export function main() { throw new Error('boom'); }",
},
}],
"failure_module": {
"value": {
"type": "rawscript",
"language": "deno",
"input_transforms": {},
"content": "export function main() { return { recovered: true } }",
},
},
}))
.unwrap();
let completed =
RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None })
.run_until_complete(&db, false, port)
.await;
// Sanity: flow result is the failure_module's output.
assert_eq!(
json!({ "recovered": true }),
completed.json_result().unwrap()
);
let read_body = |response: axum::response::Response| async move {
let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
serde_json::from_slice::<serde_json::Value>(&bytes).unwrap()
};
// has_failure_module=false: legacy behavior — return the early_return node's failure.
let resp_a_only = windmill_api::jobs::run_wait_result(
&db,
completed.id,
"test-workspace",
Some("a".to_string()),
false,
"test-user",
)
.await
.unwrap();
let body_a_only = read_body(resp_a_only).await;
let body_a_only_str = body_a_only.to_string();
assert!(
body_a_only_str.contains("boom"),
"expected node 'a' failure, got {body_a_only_str}",
);
assert!(
!body_a_only_str.contains("recovered"),
"expected node 'a' failure, not failure_module output; got {body_a_only_str}",
);
// has_failure_module=true: skip the early_return node's failure and return the
// failure_module's recovered result instead.
let resp_with_fm = windmill_api::jobs::run_wait_result(
&db,
completed.id,
"test-workspace",
Some("a".to_string()),
true,
"test-user",
)
.await
.unwrap();
let body_with_fm = read_body(resp_with_fm).await;
assert_eq!(json!({ "recovered": true }), body_with_fm);
Ok(())
}
#[cfg(feature = "python")]
#[sqlx::test(fixtures("base"))]
async fn test_flow_lock_all(db: Pool<Postgres>) -> anyhow::Result<()> {
use futures::StreamExt;
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: windmill_api_client::types::OpenFlow = serde_json::from_value(serde_json::json!({
"summary": "",
"description": "",
"value": {
"modules": [
{
"id": "a",
"value": {
"lock": null,
"path": null,
"type": "rawscript",
"content": "import wmill\n\ndef main():\n return \"Test\"\n",
"language": "python3",
"input_transforms": {}
},
"summary": null,
"stop_after_if": null,
"input_transforms": {}
},
{
"id": "b",
"value": {
"lock": null,
"path": null,
"type": "rawscript",
"content": "import * as wmill from \"https://deno.land/x/windmill@v1.50.0/mod.ts\"\n\nexport async function main() {\n return wmill\n}\n",
"language": "deno",
"input_transforms": {}
},
"summary": null,
"stop_after_if": null,
"input_transforms": {}
},
{
"id": "c",
"value": {
"lock": null,
"path": null,
"type": "rawscript",
"content": "package inner\n\nimport (\n\t\"fmt\"\n\t\"rsc.io/quote\"\n wmill \"github.com/windmill-labs/windmill-go-client\"\n)\n\n// the main must return (interface{}, error)\n\nfunc main() (interface{}, error) {\n\tfmt.Println(\"Hello, World\")\n // v, _ := wmill.GetVariable(\"g/all/pretty_secret\")\n return \"Test\"\n}\n",
"language": "go",
"input_transforms": {}
},
"summary": null,
"stop_after_if": null,
"input_transforms": {}
},
{
"id": "d",
"value": {
"lock": null,
"path": null,
"type": "rawscript",
"content": "\n# the last line of the stdout is the return value\necho \"Hello $msg\"\n",
"language": "bash",
"input_transforms": {}
},
"summary": null,
"stop_after_if": null,
"input_transforms": {}
}
],
"failure_module": null
},
"schema": {
"type": "object",
"$schema": "https://json-schema.org/draft/2020-12/schema",
"required": [],
"properties": {}
}
}))
.unwrap();
let client = windmill_api_client::create_client(
&format!("http://localhost:{port}"),
"SECRET_TOKEN".to_owned(),
);
client
.create_flow(
"test-workspace",
&CreateFlowBody {
open_flow_w_path: windmill_api_client::types::OpenFlowWPath {
open_flow: flow,
path: "g/all/flow_lock_all".to_owned(),
tag: None,
ws_error_handler_muted: None,
priority: None,
dedicated_worker: None,
timeout: None,
visible_to_runner_only: None,
on_behalf_of_email: None,
},
deployment_message: None,
draft_only: None,
},
)
.await
.unwrap();
let mut str = listen_for_completed_jobs(&db).await;
let listen_first_job = str.next();
in_test_worker(&db, listen_first_job, port).await;
let modules = client
.get_flow_by_path("test-workspace", "g/all/flow_lock_all", None)
.await
.unwrap()
.open_flow
.value
.modules;
modules.into_iter()
.for_each(|m| {
assert!(matches!(
m.value,
windmill_api_client::types::FlowModuleValue::RawScript(RawScript {
language: windmill_api_client::types::RawScriptLanguage::Bash,
lock: Some(ref lock),
..
}) if lock.is_empty())
|| matches!(
m.value,
windmill_api_client::types::FlowModuleValue::RawScript(RawScript{
language: windmill_api_client::types::RawScriptLanguage::Go | windmill_api_client::types::RawScriptLanguage::Python3 | windmill_api_client::types::RawScriptLanguage::Deno,
lock: Some(ref lock),
..
}) if !lock.is_empty()),
"{:?}", m.value
);
});
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base"))]
async fn test_complex_flow_restart(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "go",
"content": "package inner\nimport (\n\t\"fmt\"\n\t\"math/rand\"\n)\nfunc main(max int) (interface{}, error) {\n\tresult := rand.Intn(max) + 1\n\tfmt.Printf(\"Number generated: '%d'\", result)\n\treturn result, nil\n}",
"input_transforms": {
"max": {
"type": "static",
"value": json!(20),
},
}
},
"summary": "Generate random number in [1, 20]"
},
{
"id": "b",
"value":
{
"type": "branchall",
"branches":
[
{
"modules":
[
{
"id": "d",
"value":
{
"type": "branchone",
"default":
[
{
"id": "f",
"value":
{
"type": "rawscript",
"content": "package inner\nimport \"math/rand\"\nfunc main(x int) (interface{}, error) {\n\treturn rand.Intn(x) + 1, nil\n}",
"language": "go",
"input_transforms":
{
"x":
{
"expr": "results.a",
"type": "javascript"
}
}
},
"summary": "Rand N in [1; x]"
}
],
"branches":
[
{
"expr": "results.a < flow_input.max / 2",
"modules":
[
{
"id": "e",
"value":
{
"type": "rawscript",
"content": "package inner\nimport \"math/rand\"\nfunc main(x int) (interface{}, error) {\n\treturn rand.Intn(x * 2) + 1, nil\n}\n",
"language": "go",
"input_transforms":
{
"x":
{
"expr": "results.a",
"type": "javascript"
}
}
},
"summary": "Rand N in [1; x*2]"
}
],
"summary": "N in first half"
}
]
},
"summary": ""
}
],
"summary": "Process x",
"parallel": true,
"skip_failure": false
},
{
"modules":
[
{
"id": "c",
"value":
{
"type": "rawscript",
"content": "package inner\nfunc main(x int) (interface{}, error) {\n\treturn x, nil\n}",
"language": "go",
"input_transforms":
{
"x":
{
"expr": "results.a",
"type": "javascript"
}
}
},
"summary": "Identity"
}
],
"summary": "Do nothing",
"parallel": true,
"skip_failure": false
}
],
"parallel": false
},
"summary": ""
},
{
"id": "g",
"value":
{
"tag": "",
"type": "rawscript",
"content": "package inner\nimport \"fmt\"\nfunc main(x []int) (interface{}, error) {\n\tfmt.Printf(\"Results: %v\", x)\n\treturn x, nil\n}\n",
"language": "go",
"input_transforms":
{
"x":
{
"expr": "results.b",
"type": "javascript"
}
}
},
"summary": "Print results - This will get the results from the prior step directly"
},
{
"id": "h",
"value":
{
"tag": "",
"type": "rawscript",
"content": "package inner\nimport (\n\t\"fmt\"\n\t\"slices\"\n)\nfunc main(x []int) (interface{}, error) {\n\tresult := slices.Max(x)\n\tfmt.Printf(\"Result is %d\", result)\n\treturn result, nil\n}",
"language": "go",
"input_transforms":
{
"x":
{
"expr": "results.b",
"type": "javascript"
}
}
},
"summary": "Choose max - this will get results.b querying get_result_by_id on the backend"
}
],
}))
.unwrap();
let first_run_result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.run_until_complete(&db, false, port)
.await;
let restarted_flow_result = RunJob::from(JobPayload::RawFlow {
value: flow.clone(),
path: None,
restarted_from: Some(RestartedFrom {
flow_job_id: first_run_result.id,
step_id: "h".to_owned(),
branch_or_iteration_n: None,
flow_version: None,
..Default::default()
}),
})
.run_until_complete(&db, false, port)
.await;
let first_run_result_int =
serde_json::from_value::<i32>(first_run_result.json_result().unwrap())
.expect("first_run_result was not an int");
let restarted_flow_result_int =
serde_json::from_value::<i32>(restarted_flow_result.json_result().unwrap())
.expect("restarted_flow_result was not an int");
assert_eq!(first_run_result_int, restarted_flow_result_int);
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_rust_client(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
windmill_api_client::create_client(
&format!("http://localhost:{port}"),
"SECRET_TOKEN".to_string(),
)
.list_workspaces()
.await
.unwrap();
Ok(())
}
#[cfg(all(feature = "enterprise", feature = "private"))]
#[sqlx::test(fixtures("base", "schedule"))]
async fn test_script_schedule_handlers(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let client = windmill_api_client::create_client(
&format!("http://localhost:{port}"),
"SECRET_TOKEN".to_string(),
);
let mut args = std::collections::HashMap::new();
args.insert("fail".to_string(), json!(true));
let now = chrono::Utc::now();
// add 5 seconds to now
let then = now
.checked_add_signed(chrono::Duration::try_seconds(5).unwrap())
.unwrap();
let schedule = NewSchedule {
args: ScriptArgs::from(args),
enabled: Some(true),
is_flow: false,
on_failure: Some("script/f/system/schedule_error_handler".to_string()),
on_failure_times: None,
on_failure_exact: None,
on_failure_extra_args: None,
on_recovery: Some("script/f/system/schedule_recovery_handler".to_string()),
on_recovery_times: None,
on_recovery_extra_args: None,
on_success: None,
on_success_extra_args: None,
path: "f/system/failing_script_schedule".to_string(),
script_path: "f/system/failing_script".to_string(),
timezone: "UTC".to_string(),
schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
ws_error_handler_muted: None,
retry: None,
no_flow_overlap: None,
summary: None,
tag: None,
cron_version: None,
description: None,
};
let _ = client.create_schedule("test-workspace", &schedule).await;
let mut str = listen_for_completed_jobs(&db).await;
let db2 = db.clone();
in_test_worker(
&db,
async move {
str.next().await; // completed error job
let uuid = timeout(Duration::from_millis(5000), str.next()).await; // error handler
if uuid.is_err() {
panic!("schedule error handler was not run within 5 s");
}
let uuid = uuid.unwrap().unwrap();
let completed_job = sqlx::query!(
"SELECT runnable_path as script_path FROM v2_job WHERE id = $1",
uuid
)
.fetch_one(&db2)
.await
.unwrap();
if completed_job.script_path.is_none()
|| completed_job.script_path != Some("f/system/schedule_error_handler".to_string())
{
panic!(
"a script was run after main job execution but was not schedule error handler"
);
}
},
port,
)
.await;
let mut args = std::collections::HashMap::new();
args.insert("fail".to_string(), json!(false));
let now = chrono::Utc::now();
let then = now
.checked_add_signed(chrono::Duration::try_seconds(5).unwrap())
.unwrap();
client
.update_schedule(
"test-workspace",
"f/system/failing_script_schedule",
&EditSchedule {
args: ScriptArgs::from(args),
on_failure: Some("script/f/system/schedule_error_handler".to_string()),
on_failure_times: None,
on_failure_exact: None,
on_failure_extra_args: None,
on_recovery: Some("script/f/system/schedule_recovery_handler".to_string()),
on_recovery_times: None,
on_recovery_extra_args: None,
on_success: None,
on_success_extra_args: None,
timezone: "UTC".to_string(),
schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
ws_error_handler_muted: None,
retry: None,
summary: None,
no_flow_overlap: None,
tag: None,
cron_version: None,
description: None,
},
)
.await
.unwrap();
let mut str = listen_for_completed_jobs(&db).await;
let db2 = db.clone();
in_test_worker(
&db,
async move {
str.next().await; // completed working job
let uuid = timeout(Duration::from_millis(5000), str.next()).await; // recovery handler
if uuid.is_err() {
panic!("schedule recovery handler was not run within 5 s");
}
let uuid = uuid.unwrap().unwrap();
let completed_job =
sqlx::query!("SELECT runnable_path as script_path FROM v2_job WHERE id = $1", uuid)
.fetch_one(&db2)
.await
.unwrap();
if completed_job.script_path.is_none()
|| completed_job.script_path
!= Some("f/system/schedule_recovery_handler".to_string())
{
panic!("a script was run after main job execution but was not schedule recovery handler");
}
},
port,
)
.await;
Ok(())
}
#[cfg(all(feature = "enterprise", feature = "private"))]
#[sqlx::test(fixtures("base", "schedule"))]
async fn test_flow_schedule_handlers(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let client = windmill_api_client::create_client(
&format!("http://localhost:{port}"),
"SECRET_TOKEN".to_string(),
);
let mut args = std::collections::HashMap::new();
args.insert("fail".to_string(), json!(true));
let now = chrono::Utc::now();
// add 5 seconds to now
let then = now
.checked_add_signed(chrono::Duration::try_seconds(5).unwrap())
.unwrap();
let schedule = NewSchedule {
args: ScriptArgs::from(args),
enabled: Some(true),
is_flow: true,
on_failure: Some("script/f/system/schedule_error_handler".to_string()),
on_failure_times: None,
on_failure_exact: None,
on_failure_extra_args: None,
on_recovery: Some("script/f/system/schedule_recovery_handler".to_string()),
on_recovery_times: None,
on_recovery_extra_args: None,
on_success: None,
on_success_extra_args: None,
path: "f/system/failing_flow_schedule".to_string(),
script_path: "f/system/failing_flow".to_string(),
timezone: "UTC".to_string(),
schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
ws_error_handler_muted: None,
retry: None,
no_flow_overlap: None,
summary: None,
tag: None,
cron_version: None,
description: None,
};
let _ = client.create_schedule("test-workspace", &schedule).await;
let mut str = listen_for_completed_jobs(&db).await;
let db2 = db.clone();
in_test_worker(
&db,
async move {
str.next().await; // completed error step
str.next().await; // completed error flow
let uuid = timeout(Duration::from_millis(5000), str.next()).await; // error handler
if uuid.is_err() {
panic!("schedule error handler was not run within 5 s");
}
let uuid = uuid.unwrap().unwrap();
let completed_job = sqlx::query!(
"SELECT runnable_path as script_path FROM v2_job WHERE id = $1",
uuid
)
.fetch_one(&db2)
.await
.unwrap();
if completed_job.script_path.is_none()
|| completed_job.script_path != Some("f/system/schedule_error_handler".to_string())
{
panic!(
"a script was run after main job execution but was not schedule error handler"
);
}
},
port,
)
.await;
let mut args = std::collections::HashMap::new();
args.insert("fail".to_string(), json!(false));
let now = chrono::Utc::now();
let then = now
.checked_add_signed(chrono::Duration::try_seconds(5).unwrap())
.unwrap();
client
.update_schedule(
"test-workspace",
"f/system/failing_flow_schedule",
&EditSchedule {
args: ScriptArgs::from(args),
on_failure: Some("script/f/system/schedule_error_handler".to_string()),
on_failure_times: None,
on_failure_exact: None,
on_failure_extra_args: None,
on_recovery: Some("script/f/system/schedule_recovery_handler".to_string()),
on_recovery_times: None,
on_recovery_extra_args: None,
on_success: None,
on_success_extra_args: None,
timezone: "UTC".to_string(),
schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
ws_error_handler_muted: None,
retry: None,
summary: None,
no_flow_overlap: None,
tag: None,
cron_version: None,
description: None,
},
)
.await
.unwrap();
let mut str = listen_for_completed_jobs(&db).await;
let db2 = db.clone();
in_test_worker(
&db,
async move {
str.next().await; // completed working step
str.next().await; // completed working flow
let uuid = timeout(Duration::from_millis(5000), str.next()).await; // recovery handler
if uuid.is_err() {
panic!("schedule recovery handler was not run within 5 s");
}
let uuid = uuid.unwrap().unwrap();
let completed_job =
sqlx::query!("SELECT runnable_path as script_path FROM v2_job WHERE id = $1", uuid)
.fetch_one(&db2)
.await
.unwrap();
if completed_job.script_path.is_none()
|| completed_job.script_path
!= Some("f/system/schedule_recovery_handler".to_string())
{
panic!("a script was run after main job execution but was not schedule recovery handler");
}
},
port,
)
.await;
Ok(())
}
#[sqlx::test(fixtures("base", "relative_bun"))]
async fn test_relative_imports_bun(db: Pool<Postgres>) -> anyhow::Result<()> {
let content = r#"
import { main as test1 } from "/f/system/same_folder_script.ts";
import { main as test2 } from "./same_folder_script.ts";
import { main as test3 } from "/f/system_relative/different_folder_script.ts";
import { main as test4 } from "../system_relative/different_folder_script.ts";
export async function main() {
return [test1(), test2(), test3(), test4()];
}
"#
.to_string();
run_deployed_relative_imports(&db, content.clone(), ScriptLang::Bun).await?;
run_preview_relative_imports(&db, content, ScriptLang::Bun).await?;
Ok(())
}
#[cfg(feature = "deno_core")]
#[sqlx::test(fixtures("base", "relative_bun"))]
async fn test_nested_imports_bun(db: Pool<Postgres>) -> anyhow::Result<()> {
let content = r#"
import { main as test } from "/f/system_relative/nested_script.ts";
export async function main() {
return test();
}
"#
.to_string();
run_deployed_relative_imports(&db, content.clone(), ScriptLang::Bun).await?;
run_preview_relative_imports(&db, content, ScriptLang::Bun).await?;
Ok(())
}
#[sqlx::test(fixtures("base", "relative_deno"))]
async fn test_relative_imports_deno(db: Pool<Postgres>) -> anyhow::Result<()> {
let content = r#"
import { main as test1 } from "/f/system/same_folder_script.ts";
import { main as test2 } from "./same_folder_script.ts";
import { main as test3 } from "/f/system_relative/different_folder_script.ts";
import { main as test4 } from "../system_relative/different_folder_script.ts";
export async function main() {
return [test1(), test2(), test3(), test4()];
}
"#
.to_string();
run_deployed_relative_imports(&db, content.clone(), ScriptLang::Deno).await?;
run_preview_relative_imports(&db, content, ScriptLang::Deno).await?;
Ok(())
}
#[sqlx::test(fixtures("base", "relative_deno"))]
async fn test_nested_imports_deno(db: Pool<Postgres>) -> anyhow::Result<()> {
let content = r#"
import { main as test } from "/f/system_relative/nested_script.ts";
export async function main() {
return test();
}
"#
.to_string();
run_deployed_relative_imports(&db, content.clone(), ScriptLang::Deno).await?;
run_preview_relative_imports(&db, content, ScriptLang::Deno).await?;
Ok(())
}
#[sqlx::test(fixtures("base", "result_format"))]
async fn test_result_format(db: Pool<Postgres>) -> anyhow::Result<()> {
let ordered_result_job_id = "1eecb96a-c8b0-4a3d-b1b6-087878c55e41";
set_jwt_secret().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let token = windmill_common::auth::create_token_for_owner(
&db,
"test-workspace",
"u/test-user",
"",
100,
"",
&Uuid::nil(),
None,
None,
)
.await
.unwrap();
#[derive(Debug, Deserialize)]
struct JobResponse {
result: Option<Box<serde_json::value::RawValue>>,
}
async fn get_result<T: DeserializeOwned>(url: String) -> T {
reqwest::get(url)
.await
.unwrap()
.error_for_status()
.unwrap()
.json()
.await
.unwrap()
}
let correct_result = r#"[{"b":"first","a":"second"}]"#;
let job_response: JobResponse = get_result(format!("http://localhost:{port}/api/w/test-workspace/jobs_u/get/{ordered_result_job_id}?token={token}&no_logs=true")).await;
assert_eq!(job_response.result.unwrap().get(), correct_result);
let job_response: JobResponse = get_result(format!("http://localhost:{port}/api/w/test-workspace/jobs_u/completed/get_result_maybe/{ordered_result_job_id}?token={token}&no_logs=true")).await;
assert_eq!(job_response.result.unwrap().get(), correct_result);
let job_result: Box<serde_json::value::RawValue> = get_result(format!("http://localhost:{port}/api/w/test-workspace/jobs_u/completed/get_result/{ordered_result_job_id}?token={token}&no_logs=true")).await;
assert_eq!(job_result.get(), correct_result);
let response = windmill_api::jobs::run_wait_result(
&db,
Uuid::parse_str(ordered_result_job_id).unwrap(),
"test-workspace",
None,
false,
"test-user",
)
.await
.unwrap();
let result: Box<serde_json::value::RawValue> = serde_json::from_slice(
&axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap(),
)
.unwrap();
assert_eq!(result.get(), correct_result);
Ok(())
}
#[sqlx::test(fixtures("base"))]
async fn test_job_labels(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let db = &db;
let test = |original_labels: &'static [&'static str]| async move {
let job = RunJob::from(JobPayload::RawFlow {
value: serde_json::from_value(json!({
"modules": [{
"id": "a",
"value": {
"type": "rawscript",
"content": r#"export function main(world: string) {
const greet = `Hello ${world}!`;
console.log(greet)
return { greet, wm_labels: ["yolo", "greet", "greet", world] };
}"#,
"language": "deno",
"input_transforms": {
"world": { "type": "javascript", "expr": "flow_input.world" }
}
}
}],
"schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"properties": { "world": { "type": "string" } },
"type": "object",
"order": [ "world" ]
}
}))
.unwrap(),
path: None,
restarted_from: None,
})
.arg("world", json!("you"))
.run_until_complete_with(db, false, port, |id| async move {
sqlx::query!(
"UPDATE v2_job SET labels = $2 WHERE id = $1 AND $2::TEXT[] IS NOT NULL",
id,
original_labels as &[&str],
)
.execute(db)
.await
.unwrap();
})
.await;
let result = job.json_result().unwrap();
assert_eq!(result.get("greet"), Some(&json!("Hello you!")));
let labels = sqlx::query_scalar!("SELECT labels FROM v2_job WHERE id = $1", job.id)
.fetch_one(db)
.await
.unwrap();
let mut expected_labels = original_labels
.iter()
.chain(&["yolo", "greet", "you"])
.map(ToString::to_string)
.collect::<Vec<_>>();
expected_labels.sort();
assert_eq!(labels, Some(expected_labels));
};
test(&[]).await;
test(&["z", "a", "x"]).await;
Ok(())
}
#[cfg(feature = "python")]
const WORKFLOW_AS_CODE: &str = r#"
from wmill import task
import pandas as pd
import numpy as np
@task()
def heavy_compute(n: int):
df = pd.DataFrame(np.random.randn(100, 4), columns=list('ABCD'))
return df.sum().sum()
@task
def send_result(res: int, email: str):
print(f"Sending result {res} to {email}")
return "OK"
def main(n: int):
l = []
for i in range(n):
l.append(heavy_compute(i))
print(l)
return [send_result(sum(l), "example@example.com"), n]
"#;
#[cfg(feature = "python")]
#[sqlx::test(fixtures("base", "hello"))]
async fn test_workflow_as_code(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
// workflow as code require at least 2 workers:
let db = &db;
in_test_worker(
db,
async move {
let job = Box::pin(
RunJob::from(JobPayload::Code(RawCode {
language: ScriptLang::Python3,
content: WORKFLOW_AS_CODE.into(),
tag: None,
..RawCode::default()
}))
.arg("n", json!(3))
.run_until_complete(db, false, port),
)
.await;
assert_eq!(job.json_result().unwrap(), json!(["OK", 3]));
let workflow_as_code_status = sqlx::query_scalar!(
"SELECT workflow_as_code_status FROM v2_job_completed WHERE id = $1",
job.id
)
.fetch_one(db)
.await
.unwrap()
.unwrap();
#[derive(Deserialize)]
#[allow(dead_code)]
struct WorkflowJobStatus {
name: String,
started_at: String,
scheduled_for: String,
duration_ms: i64,
}
let workflow_as_code_status: std::collections::HashMap<String, WorkflowJobStatus> =
serde_json::from_value(workflow_as_code_status).unwrap();
let uuids = sqlx::query_scalar!("SELECT id FROM v2_job WHERE parent_job = $1", job.id)
.fetch_all(db)
.await
.unwrap();
assert_eq!(uuids.len(), 4);
for uuid in uuids {
let status = workflow_as_code_status.get(&uuid.to_string());
assert!(status.is_some());
assert!(
status.unwrap().name == "send_result"
|| status.unwrap().name == "heavy_compute"
);
}
},
port,
)
.await;
Ok(())
}
#[cfg(feature = "duckdb")]
#[sqlx::test(fixtures("base"))]
async fn test_duckdb_ffi(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let content = "-- result_collection=last_statement_first_row_scalar\nSELECT 'Hello world!';";
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{
"value": {
"type": "rawscript",
"language": "duckdb",
"content": content,
},
}],
}))
.unwrap();
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.run_until_complete(&db, false, server.addr.port())
.await
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("Hello world!"));
Ok(())
}
/// Test that flow substeps with tags that are not available for the workspace fail.
/// This validates that `check_tag_available_for_workspace_internal` is properly called
/// when pushing jobs from worker_flow.
#[sqlx::test(fixtures("base"))]
async fn test_flow_substep_tag_availability_check(db: Pool<Postgres>) -> anyhow::Result<()> {
use windmill_common::worker::{
CustomTags, SpecificTagData, SpecificTagType, CUSTOM_TAGS_PER_WORKSPACE,
};
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
// Set up a restricted tag that is only available to "other-workspace" (not "test-workspace")
CUSTOM_TAGS_PER_WORKSPACE.store(std::sync::Arc::new(CustomTags {
global: vec![],
specific: std::collections::HashMap::from([(
"restricted-tag".to_string(),
SpecificTagData {
tag_type: SpecificTagType::NoneExcept,
workspaces: vec!["other-workspace".to_string()],
},
)]),
}));
// Create a flow with a substep that uses the restricted tag
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [{
"id": "a",
"value": {
"type": "rawscript",
"language": "deno",
"content": "export function main() { return 42; }",
"tag": "restricted-tag",
},
}],
}))
.unwrap();
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.email("test2@windmill.dev")
.run_until_complete(&db, false, server.addr.port())
.await;
// The flow should fail because the tag is not available for test-workspace
assert!(
!result.success,
"Flow should have failed due to unavailable tag"
);
let result_json = result.json_result();
assert!(result_json.is_some(), "Result should have error details");
let error_result = result_json.unwrap();
let error_message = error_result["error"]["message"].as_str().unwrap_or("");
// Verify the error is about tag availability
assert!(
error_message.contains("restricted-tag") || error_message.contains("tag"),
"Error message should mention the tag issue: {}",
error_message
);
// Clean up: reset custom tags
CUSTOM_TAGS_PER_WORKSPACE.store(std::sync::Arc::new(CustomTags::default()));
Ok(())
}
#[cfg(all(feature = "quickjs", feature = "python"))]
#[sqlx::test(fixtures("base"))]
async fn test_whileloop_propagates_inner_iterator_eval_failure(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
initialize_tracing().await;
// Regression test: while-loop with skip_failures=false whose body has
// a successful step followed by a for-loop whose iterator expression
// throws. Previously the iterator-eval failure was silently swallowed
// because `handle_flow` returning `Err` left the local `success` set to
// the prior step's outcome (true). The parent while-loop received
// success=true and kept iterating despite the inner failure.
let port = 123;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"id": "outer",
"value": {
"type": "whileloopflow",
"skip_failures": false,
"modules": [
{
"id": "prev",
"value": {
"input_transforms": {
"i": {
"type": "javascript",
"expr": "flow_input.iter.index",
},
},
"type": "rawscript",
"language": "python3",
"content": "def main(i): return i",
},
},
{
"id": "inner",
"value": {
"type": "forloopflow",
// results.prev is null, so `.does_not_exist` throws.
"iterator": {
"type": "javascript",
"expr": "results.prev.does_not_exist",
},
"skip_failures": false,
"parallel": false,
"modules": [{
"id": "leaf",
"value": {
"type": "rawscript",
"language": "python3",
"content": "def main(): return 1",
},
}],
},
},
],
},
// bound iteration count so without the fix the test fails fast
// with success=true (rather than running forever); with the fix
// the loop stops at iter 0 with success=false.
"stop_after_if": {
"expr": "result >= 2",
"skip_if_stopped": false,
},
},
],
}))
.unwrap();
let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let cjob = RunJob::from(job).run_until_complete(&db, false, port).await;
assert!(
!cjob.success,
"flow should fail when inner forloop iterator throws inside a while-loop with skip_failures=false"
);
Ok(())
}
#[cfg(all(feature = "quickjs", feature = "python"))]
#[sqlx::test(fixtures("base"))]
async fn test_stop_after_all_iters_if_bad_expr_parallel_branchall(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
initialize_tracing().await;
let port = 123;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"id": "a",
"value": {
"branches": [
{"modules": [{
"id": "b",
"value": {
"input_transforms": { "n": { "type": "javascript", "expr": "flow_input.n" } },
"type": "rawscript",
"language": "python3",
"content": "def main(n): return n",
},
}]}
],
"type": "branchall",
"parallel": true,
},
"stop_after_all_iters_if": {
"expr": "invalid!!!syntax",
"skip_if_stopped": false,
},
},
],
}))
.unwrap();
let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let cjob = RunJob::from(job)
.arg("n", json!(42))
.run_until_complete(&db, false, port)
.await;
assert!(
!cjob.success,
"flow should fail when stop_after_all_iters_if has bad expression"
);
let result = cjob.json_result().unwrap();
let error_msg = result["error"]["message"].as_str().unwrap_or("");
assert!(
error_msg.contains("stop_after_all_iters_if"),
"error should mention stop_after_all_iters_if, got: {error_msg}"
);
Ok(())
}
#[cfg(all(feature = "quickjs", feature = "python"))]
#[sqlx::test(fixtures("base"))]
async fn test_stop_after_all_iters_if_bad_expr_parallel_forloop(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
initialize_tracing().await;
let port = 123;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"id": "a",
"value": {
"type": "forloopflow",
"iterator": { "type": "javascript", "expr": "result.items" },
"skip_failures": false,
"parallel": true,
"modules": [{
"value": {
"input_transforms": {
"n": { "type": "javascript", "expr": "flow_input.iter.value" },
},
"type": "rawscript",
"language": "python3",
"content": "def main(n): return n",
},
}],
},
"stop_after_all_iters_if": {
"expr": "invalid!!!syntax",
"skip_if_stopped": false,
},
},
],
}))
.unwrap();
let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
let cjob = RunJob::from(job)
.arg("items", json!([1, 2, 3]))
.run_until_complete(&db, false, port)
.await;
assert!(
!cjob.success,
"flow should fail when stop_after_all_iters_if has bad expression"
);
let result = cjob.json_result().unwrap();
let error_msg = result["error"]["message"].as_str().unwrap_or("");
assert!(
error_msg.contains("stop_after_all_iters_if"),
"error should mention stop_after_all_iters_if, got: {error_msg}"
);
Ok(())
}
// Regression for the savepoint added around `evaluate_stop_after_all_iters_if`.
// The failpoint makes the in-evaluation DB read fail with a transaction-aborting
// error (SELECT 1/0). The caller swallows that error and keeps using the outer
// status-update transaction (later reads + commit). Without the savepoint the
// outer transaction would be aborted and the commit would fail, so the flow job
// would never be marked completed (the worker would error/retry). With the
// savepoint the read failure is isolated, the iteration is marked failed, and
// the flow completes — which is what this test asserts.
#[cfg(all(feature = "failpoints", feature = "quickjs", feature = "python"))]
#[sqlx::test(fixtures("base"))]
async fn test_stop_after_all_iters_if_db_error_isolated_by_savepoint(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
initialize_tracing().await;
let port = 123;
let flow: FlowValue = serde_json::from_value(serde_json::json!({
"modules": [
{
"id": "a",
"value": {
"type": "forloopflow",
"iterator": { "type": "javascript", "expr": "result.items" },
"skip_failures": false,
"parallel": true,
"modules": [{
"value": {
"input_transforms": {
"n": { "type": "javascript", "expr": "flow_input.iter.value" },
},
"type": "rawscript",
"language": "python3",
"content": "def main(n): return n",
},
}],
},
"stop_after_all_iters_if": {
"expr": "__wm_failpoint_abort_tx__",
"skip_if_stopped": false,
},
},
],
}))
.unwrap();
let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None };
// If the savepoint failed to isolate the aborted read, the status-update
// transaction would be poisoned and the job would never complete, so this
// call would hang until the worker times out rather than returning.
let cjob = RunJob::from(job)
.arg("items", json!([1, 2, 3]))
.run_until_complete(&db, false, port)
.await;
assert!(
!cjob.success,
"iteration should be marked failed after the injected read error"
);
let result = cjob.json_result().unwrap();
let error_msg = result["error"]["message"].as_str().unwrap_or("");
assert!(
error_msg.contains("stop_after_all_iters_if"),
"error should mention stop_after_all_iters_if, got: {error_msg}"
);
Ok(())
}
#[cfg(all(feature = "quickjs", feature = "python"))]
#[sqlx::test(fixtures("base"))]
async fn test_results_length_in_input_transform(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
// Step a returns a list, step b accesses results.a.length via input transform.
// This tests that the handle_full_regex fast path falls through to QuickJS
// when the SQL JSON path operator can't resolve JS properties like .length.
let flow: FlowValue = serde_json::from_value(json!({
"modules": [
{
"id": "a",
"value": {
"type": "rawscript",
"language": "python3",
"content": "def main(): return [10, 20, 30]",
},
},
{
"id": "b",
"value": {
"input_transforms": {
"v": { "type": "javascript", "expr": "results.a.length" },
},
"type": "rawscript",
"language": "python3",
"content": "def main(v): return v",
},
},
],
}))
.unwrap();
let result =
RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None })
.run_until_complete(&db, false, port)
.await
.json_result()
.unwrap();
assert_eq!(
result,
json!(3),
"results.a.length should resolve to 3, not null"
);
Ok(())
}
Reference in New Issue View Git Blame Copy Permalink