starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-19 16:03:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into home-tight-tags-runs-triggers

Browse Source
This commit is contained in:
Diego Imbert
2026-10-07 23:31:52 +02:00
committed by GitHub
parent 521be8ecc1 f40eea2a5f
commit f4ecfe5fa3
43 changed files with 1375 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e.json → backend/.sqlx/query-58ea7e2c6b07ffc7ab7387b2ebb542cba3a5532c64a9594d2e0ccdf21dee90d0.json
Generated
+3 -3
View File
@@ -1,11 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS (SELECT 1 FROM v2_job_queue WHERE id = $1 AND workspace_id = $2 AND running)",
"query": "SELECT canceled_by IS NOT NULL FROM v2_job_queue WHERE id = $1 AND workspace_id = $2 AND running",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"name": "?column?",
"type_info": "Bool"
}
],
@@ -19,5 +19,5 @@
null
]
},
"hash": "ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e"
"hash": "58ea7e2c6b07ffc7ab7387b2ebb542cba3a5532c64a9594d2e0ccdf21dee90d0"
}
backend/.sqlx/query-91fdd565bd210f8aa4c6d5f5e309573aefaeb5dbc6ac07f5e664bc8fe2cb2683.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "coalesce",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "91fdd565bd210f8aa4c6d5f5e309573aefaeb5dbc6ac07f5e664bc8fe2cb2683"
}
backend/.sqlx/query-a6d355b3121f199f3827c05aaa7c8f780ab4c9d3d373d3cd0bb489e0d173040c.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1 FOR UPDATE",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "coalesce",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "a6d355b3121f199f3827c05aaa7c8f780ab4c9d3d373d3cd0bb489e0d173040c"
}
backend/Cargo.lock
Generated
+79 -79
View File
@@ -14829,7 +14829,7 @@ dependencies = [
[[package]]
name = "windmill"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-nats",
@@ -14918,7 +14918,7 @@ dependencies = [
[[package]]
name = "windmill-ai"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"async-stream",
"async-trait",
@@ -14952,7 +14952,7 @@ dependencies = [
[[package]]
name = "windmill-alerting"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -14965,7 +14965,7 @@ dependencies = [
[[package]]
name = "windmill-api"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"argon2",
@@ -15105,7 +15105,7 @@ dependencies = [
[[package]]
name = "windmill-api-agent-workers"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15128,7 +15128,7 @@ dependencies = [
[[package]]
name = "windmill-api-assets"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15145,7 +15145,7 @@ dependencies = [
[[package]]
name = "windmill-api-auth"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"axum 0.8.9",
@@ -15171,7 +15171,7 @@ dependencies = [
[[package]]
name = "windmill-api-client"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"reqwest 0.12.28",
"serde",
@@ -15181,7 +15181,7 @@ dependencies = [
[[package]]
name = "windmill-api-configs"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15198,7 +15198,7 @@ dependencies = [
[[package]]
name = "windmill-api-debug"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"base64 0.22.1",
@@ -15220,7 +15220,7 @@ dependencies = [
[[package]]
name = "windmill-api-embeddings"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"axum 0.8.9",
@@ -15243,7 +15243,7 @@ dependencies = [
[[package]]
name = "windmill-api-flow-conversations"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15259,7 +15259,7 @@ dependencies = [
[[package]]
name = "windmill-api-flows"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15281,7 +15281,7 @@ dependencies = [
[[package]]
name = "windmill-api-groups"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15303,7 +15303,7 @@ dependencies = [
[[package]]
name = "windmill-api-inputs"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15317,7 +15317,7 @@ dependencies = [
[[package]]
name = "windmill-api-integration-tests"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-nats",
@@ -15353,7 +15353,7 @@ dependencies = [
[[package]]
name = "windmill-api-jobs"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"axum 0.8.9",
@@ -15379,7 +15379,7 @@ dependencies = [
[[package]]
name = "windmill-api-npm-proxy"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15407,7 +15407,7 @@ dependencies = [
[[package]]
name = "windmill-api-openapi"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"axum 0.8.9",
@@ -15429,7 +15429,7 @@ dependencies = [
[[package]]
name = "windmill-api-schedule"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15449,7 +15449,7 @@ dependencies = [
[[package]]
name = "windmill-api-scripts"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"axum 0.8.9",
@@ -15491,7 +15491,7 @@ dependencies = [
[[package]]
name = "windmill-api-settings"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"axum 0.8.9",
@@ -15520,7 +15520,7 @@ dependencies = [
[[package]]
name = "windmill-api-sse"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"lazy_static",
"serde",
@@ -15532,7 +15532,7 @@ dependencies = [
[[package]]
name = "windmill-api-users"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"argon2",
"axum 0.8.9",
@@ -15556,7 +15556,7 @@ dependencies = [
[[package]]
name = "windmill-api-workers"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15570,7 +15570,7 @@ dependencies = [
[[package]]
name = "windmill-api-workspaces"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"axum 0.8.9",
"chrono",
@@ -15609,7 +15609,7 @@ dependencies = [
[[package]]
name = "windmill-audit"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"chrono",
"lazy_static",
@@ -15624,7 +15624,7 @@ dependencies = [
[[package]]
name = "windmill-autoscaling"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"axum 0.8.9",
@@ -15643,7 +15643,7 @@ dependencies = [
[[package]]
name = "windmill-common"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"aes-gcm",
"aho-corasick",
@@ -15757,7 +15757,7 @@ dependencies = [
[[package]]
name = "windmill-dep-map"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"chrono",
"futures",
@@ -15777,7 +15777,7 @@ dependencies = [
[[package]]
name = "windmill-git-sync"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"regex",
"serde",
@@ -15794,7 +15794,7 @@ dependencies = [
[[package]]
name = "windmill-indexer"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"astral-tokio-tar",
@@ -15821,7 +15821,7 @@ dependencies = [
[[package]]
name = "windmill-jseval"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"futures",
@@ -15838,7 +15838,7 @@ dependencies = [
[[package]]
name = "windmill-macros"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"itertools 0.14.0",
"lazy_static",
@@ -15854,7 +15854,7 @@ dependencies = [
[[package]]
name = "windmill-mcp"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -15875,7 +15875,7 @@ dependencies = [
[[package]]
name = "windmill-native-triggers"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -15906,7 +15906,7 @@ dependencies = [
[[package]]
name = "windmill-oauth"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"arc-swap",
@@ -15931,7 +15931,7 @@ dependencies = [
[[package]]
name = "windmill-object-store"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-stream",
@@ -15967,7 +15967,7 @@ dependencies = [
[[package]]
name = "windmill-operator"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"futures",
@@ -15985,7 +15985,7 @@ dependencies = [
[[package]]
name = "windmill-parser"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"convert_case 0.6.0",
"serde",
@@ -15994,7 +15994,7 @@ dependencies = [
[[package]]
name = "windmill-parser-bash"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -16006,7 +16006,7 @@ dependencies = [
[[package]]
name = "windmill-parser-csharp"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde_json",
@@ -16018,7 +16018,7 @@ dependencies = [
[[package]]
name = "windmill-parser-go"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"gosyn",
@@ -16030,7 +16030,7 @@ dependencies = [
[[package]]
name = "windmill-parser-graphql"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -16042,7 +16042,7 @@ dependencies = [
[[package]]
name = "windmill-parser-java"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde_json",
@@ -16054,7 +16054,7 @@ dependencies = [
[[package]]
name = "windmill-parser-nu"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"nu-parser",
@@ -16065,7 +16065,7 @@ dependencies = [
[[package]]
name = "windmill-parser-php"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"itertools 0.14.0",
@@ -16076,7 +16076,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"itertools 0.14.0",
@@ -16088,7 +16088,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py-asset"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"rustpython-ast",
@@ -16099,7 +16099,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py-imports"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-recursion",
@@ -16121,7 +16121,7 @@ dependencies = [
[[package]]
name = "windmill-parser-r"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde_json",
@@ -16133,7 +16133,7 @@ dependencies = [
[[package]]
name = "windmill-parser-ruby"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -16147,7 +16147,7 @@ dependencies = [
[[package]]
name = "windmill-parser-rust"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"convert_case 0.6.0",
@@ -16164,7 +16164,7 @@ dependencies = [
[[package]]
name = "windmill-parser-sql"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -16177,7 +16177,7 @@ dependencies = [
[[package]]
name = "windmill-parser-sql-asset"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde",
@@ -16189,7 +16189,7 @@ dependencies = [
[[package]]
name = "windmill-parser-ts"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -16207,7 +16207,7 @@ dependencies = [
[[package]]
name = "windmill-parser-ts-asset"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde-wasm-bindgen",
@@ -16223,7 +16223,7 @@ dependencies = [
[[package]]
name = "windmill-parser-wac"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"rustpython-ast",
@@ -16239,7 +16239,7 @@ dependencies = [
[[package]]
name = "windmill-parser-yaml"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -16253,7 +16253,7 @@ dependencies = [
[[package]]
name = "windmill-queue"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-recursion",
@@ -16292,7 +16292,7 @@ dependencies = [
[[package]]
name = "windmill-runtime-nativets"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"const_format",
@@ -16332,7 +16332,7 @@ dependencies = [
[[package]]
name = "windmill-sql-datatype-parser-wasm"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"getrandom 0.3.4",
"wasm-bindgen",
@@ -16343,7 +16343,7 @@ dependencies = [
[[package]]
name = "windmill-store"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-recursion",
@@ -16378,7 +16378,7 @@ dependencies = [
[[package]]
name = "windmill-test-utils"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16402,7 +16402,7 @@ dependencies = [
[[package]]
name = "windmill-trigger"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16435,7 +16435,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-amqp"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16462,7 +16462,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-azure"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16495,7 +16495,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-email"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16515,7 +16515,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-gcp"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16549,7 +16549,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-http"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16586,7 +16586,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-kafka"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16609,7 +16609,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-mqtt"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16633,7 +16633,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-nats"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-nats",
@@ -16657,7 +16657,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-postgres"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16692,7 +16692,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-sqs"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16720,7 +16720,7 @@ dependencies = [
[[package]]
name = "windmill-trigger-websocket"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-trait",
@@ -16745,7 +16745,7 @@ dependencies = [
[[package]]
name = "windmill-types"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"bitflags 2.13.2",
@@ -16764,7 +16764,7 @@ dependencies = [
[[package]]
name = "windmill-worker"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-once-cell",
@@ -16883,7 +16883,7 @@ dependencies = [
[[package]]
name = "windmill-worker-volumes"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"bytes",
"futures",
backend/Cargo.toml
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "windmill"
version = "1.826.0"
version = "1.827.0"
authors.workspace = true
edition.workspace = true
@@ -88,7 +88,7 @@ members = [
exclude = ["./windmill-duckdb-ffi-internal", "./parsers/windmill-parser-wasm"]
[workspace.package]
version = "1.826.0"
version = "1.827.0"
authors = ["Ruben Fiszel <ruben@windmill.dev>"]
edition = "2021"
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
acb254ac509aff244849c09e50a8d39aae185138
1600d49ad5513af8b54ad65688d3d1c7edc977d9
backend/parsers/windmill-parser-wasm/Cargo.lock
Generated
+24 -24
View File
@@ -6216,7 +6216,7 @@ checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
[[package]]
name = "windmill-common"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"aho-corasick",
"anyhow",
@@ -6302,7 +6302,7 @@ dependencies = [
[[package]]
name = "windmill-macros"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"proc-macro2",
"quote",
@@ -6314,7 +6314,7 @@ dependencies = [
[[package]]
name = "windmill-parser"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"convert_case",
"serde",
@@ -6323,7 +6323,7 @@ dependencies = [
[[package]]
name = "windmill-parser-bash"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -6335,7 +6335,7 @@ dependencies = [
[[package]]
name = "windmill-parser-csharp"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde_json",
@@ -6347,7 +6347,7 @@ dependencies = [
[[package]]
name = "windmill-parser-go"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"gosyn",
@@ -6359,7 +6359,7 @@ dependencies = [
[[package]]
name = "windmill-parser-graphql"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -6371,7 +6371,7 @@ dependencies = [
[[package]]
name = "windmill-parser-java"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde_json",
@@ -6383,7 +6383,7 @@ dependencies = [
[[package]]
name = "windmill-parser-nu"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"nu-parser",
@@ -6394,7 +6394,7 @@ dependencies = [
[[package]]
name = "windmill-parser-php"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"itertools 0.14.0",
@@ -6405,7 +6405,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"itertools 0.14.0",
@@ -6417,7 +6417,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py-asset"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"rustpython-ast",
@@ -6428,7 +6428,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py-imports"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"async-recursion",
@@ -6450,7 +6450,7 @@ dependencies = [
[[package]]
name = "windmill-parser-r"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde_json",
@@ -6462,7 +6462,7 @@ dependencies = [
[[package]]
name = "windmill-parser-ruby"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -6476,7 +6476,7 @@ dependencies = [
[[package]]
name = "windmill-parser-rust"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"convert_case",
@@ -6493,7 +6493,7 @@ dependencies = [
[[package]]
name = "windmill-parser-sql"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -6506,7 +6506,7 @@ dependencies = [
[[package]]
name = "windmill-parser-sql-asset"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde",
@@ -6518,7 +6518,7 @@ dependencies = [
[[package]]
name = "windmill-parser-ts"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -6536,7 +6536,7 @@ dependencies = [
[[package]]
name = "windmill-parser-ts-asset"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"serde-wasm-bindgen",
@@ -6552,7 +6552,7 @@ dependencies = [
[[package]]
name = "windmill-parser-wac"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"rustpython-ast",
@@ -6568,7 +6568,7 @@ dependencies = [
[[package]]
name = "windmill-parser-wasm"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"getrandom 0.2.17",
@@ -6600,7 +6600,7 @@ dependencies = [
[[package]]
name = "windmill-parser-yaml"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -6614,7 +6614,7 @@ dependencies = [
[[package]]
name = "windmill-types"
version = "1.826.0"
version = "1.827.0"
dependencies = [
"anyhow",
"bitflags",
backend/parsers/windmill-parser-wasm/Cargo.toml
+1 -1
View File
@@ -12,7 +12,7 @@ resolver = "2"
members = ["."]
[workspace.package]
version = "1.826.0"
version = "1.827.0"
edition = "2021"
authors = ["Ruben Fiszel <ruben@windmill.dev>"]
backend/src/monitor.rs
+8 -1
View File
@@ -427,6 +427,8 @@ pub async fn initial_load(
native_mode,
// an agent worker never reads its group's config, only its token
object_store_cache_config: None,
// nor whether it is paused: the server refuses it jobs instead
paused: false,
}));
}
}
@@ -5689,8 +5691,13 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b
.as_ref()
.is_some_and(|dws| !dws.is_empty());
// Pausing or resuming must not restart the workers, dedicated ones included: a reload
// whose only difference is the flag skips every restart below.
let only_pause_changed = wc.paused != config.paused
&& WorkerConfig { paused: wc.paused, ..config.clone() } == **wc;
if **wc != config || has_dedicated {
if kill_if_change {
if kill_if_change && !only_pause_changed {
if has_dedicated
|| wc.dedicated_worker != config.dedicated_worker
|| wc.dedicated_workers != config.dedicated_workers
backend/tests/job_token_scopes.rs
+17
View File
@@ -410,5 +410,22 @@ async fn test_oidc_claim_carries_the_presented_scopes(db: Pool<Postgres>) -> any
)?;
assert_eq!(claims["job_token_scopes"], expected, "{job}");
}
// A soft-canceled job is still `running` until its worker kills it.
sqlx::query("UPDATE v2_job_queue SET canceled_by = 'test-user' WHERE id = $1")
.bind(Uuid::parse_str(OIDC_JOB)?)
.execute(&db)
.await?;
let resp = client
.post(format!("{base}/oidc/token/test-audience"))
.bearer_auth(job_token(&db, OIDC_JOB).await?)
.send()
.await?;
assert_eq!(
resp.status(),
StatusCode::FORBIDDEN,
"{}",
resp.text().await?
);
Ok(())
}
backend/tests/worker_group_pause.rs
+71
View File
@@ -0,0 +1,71 @@
//! In its own test binary: it flips the process-wide `WORKER_CONFIG`, which would stall every
//! other worker test sharing the process.
use futures::StreamExt;
use sqlx::{Pool, Postgres};
use std::time::Duration;
use windmill_common::{
jobs::{JobPayload, RawCode},
scripts::ScriptLang,
worker::{Connection, WORKER_CONFIG},
};
use windmill_test_utils::*;
fn set_paused(paused: bool) {
let mut wc = (**WORKER_CONFIG.load()).clone();
wc.paused = paused;
WORKER_CONFIG.store(std::sync::Arc::new(wc));
}
#[sqlx::test(fixtures("base"))]
async fn paused_worker_leaves_jobs_queued_until_resumed(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
set_paused(true);
let id = RunJob::from(JobPayload::Code(RawCode {
hash: None,
content: "echo done".to_string(),
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,
}))
.push(&db)
.await;
let mut completed = listen_for_completed_jobs(&db).await;
in_test_worker(
Connection::Sql(db.clone()),
async {
tokio::time::sleep(Duration::from_secs(3)).await;
let running =
sqlx::query_scalar::<_, bool>("SELECT running FROM v2_job_queue WHERE id = $1")
.bind(id)
.fetch_optional(&db)
.await
.unwrap();
assert_eq!(
running,
Some(false),
"a paused worker must not pull the job"
);
set_paused(false);
while completed.next().await != Some(id) {}
},
port,
)
.await;
assert!(completed_job(id, &db).await.success);
Ok(())
}
backend/windmill-api-configs/src/lib.rs
+26 -2
View File
@@ -224,12 +224,13 @@ async fn update_config(
#[cfg(not(feature = "enterprise"))]
let config = if name.starts_with("worker__") {
// In CE, only allow setting worker_tags, cache_clear, init_bash, and native_mode
// In CE, only allow setting worker_tags, cache_clear, init_bash, native_mode and paused
serde_json::json!({
"worker_tags": config.get("worker_tags"),
"cache_clear": config.get("cache_clear"),
"init_bash": config.get("init_bash"),
"native_mode": config.get("native_mode")
"native_mode": config.get("native_mode"),
"paused": config.get("paused")
})
} else {
config
@@ -272,6 +273,18 @@ async fn update_config(
}
let mut tx = db.begin().await?;
let was_paused = if name.starts_with("worker__") {
sqlx::query_scalar!(
"SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1 FOR UPDATE",
&name
)
.fetch_optional(&mut *tx)
.await?
.flatten()
.unwrap_or(false)
} else {
false
};
sqlx::query!(
"INSERT INTO config (name, config) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET config = EXCLUDED.config",
&name,
@@ -291,6 +304,17 @@ async fn update_config(
)
.await?;
tx.commit().await?;
if name.starts_with("worker__") {
let paused = config.get("paused").and_then(|v| v.as_bool()) == Some(true);
if paused != was_paused {
windmill_common::feature_usage::log_feature_usage(
"worker_group",
"pause_toggle",
if paused { "pause" } else { "resume" },
);
}
}
Ok(format!("Updated config {name}"))
}
backend/windmill-api/openapi.yaml
+1 -1
View File
@@ -1,7 +1,7 @@
openapi: "3.0.3"
info:
version: 1.826.0
version: 1.827.0
title: Windmill API
contact:
backend/windmill-common/src/instance_config.rs
+3
View File
@@ -963,6 +963,9 @@ pub struct WorkerGroupConfig {
pub autoscaling: Option<AutoscalingConfig>,
#[serde(skip_serializing_if = "Option::is_none")]
pub native_mode: Option<bool>,
/// Workers of the group pull no new job while this is set; queued jobs wait.
#[serde(skip_serializing_if = "Option::is_none")]
pub paused: Option<bool>,
/// Object store this group's dependency cache uses instead of the instance one. Same shape
/// as the instance `object_store_cache_config`.
#[serde(skip_serializing_if = "Option::is_none")]
backend/windmill-common/src/log_context.rs
+1
View File
@@ -49,6 +49,7 @@ pub struct LogContext {
// Worker / job span (windmill-worker/src/worker.rs)
pub worker: Option<String>,
pub worker_group: Option<String>,
pub hostname: Option<String>,
pub tag: Option<String>,
pub job_id: Option<String>,
backend/windmill-common/src/worker.rs
+59 -7
View File
@@ -477,6 +477,7 @@ lazy_static::lazy_static! {
env_vars: Default::default(),
native_mode: false,
object_store_cache_config: Default::default(),
paused: false,
});
pub static ref WORKER_PULL_QUERIES: arc_swap::ArcSwap<Vec<String>> = arc_swap::ArcSwap::from_pointee(vec![]);
@@ -844,11 +845,14 @@ pub fn make_suspended_pull_query(tags: &[String]) -> String {
}
// pub async fn make_suspended
pub async fn store_suspended_pull_query(wc: &WorkerConfig) {
if wc.worker_tags.len() == 0 {
tracing::error!("Empty tags in worker tags, skipping");
return;
}
let query = make_suspended_pull_query(&wc.worker_tags);
// An empty tag list must clear the query: keeping the previous one would leave a reloaded
// worker pulling suspended jobs for tags it no longer listens to.
let query = if wc.worker_tags.is_empty() {
tracing::error!("Empty tags in worker tags, pulling no suspended jobs");
String::new()
} else {
make_suspended_pull_query(&wc.worker_tags)
};
WORKER_SUSPENDED_PULL_QUERY.store(std::sync::Arc::new(query));
}
@@ -922,6 +926,42 @@ pub async fn store_pull_query(wc: &WorkerConfig) {
WORKER_PULL_QUERIES_FAIRNESS.store(std::sync::Arc::new(fairness_queries));
}
const WORKER_GROUP_PAUSED_CACHE_TTL: Duration = Duration::from_secs(10);
lazy_static::lazy_static! {
static ref WORKER_GROUP_PAUSED_CACHE: quick_cache::sync::Cache<String, (bool, std::time::Instant)> =
quick_cache::sync::Cache::new(1000);
}
/// Whether the config of `worker_group` is paused, as the server sees it. A worker that reads
/// its own config knows from `WORKER_CONFIG`; this is for the server deciding on behalf of
/// workers that do not, and is cached since they ask on every pull.
pub async fn is_worker_group_paused(db: &DB, worker_group: &str) -> bool {
if let Some((paused, at)) = WORKER_GROUP_PAUSED_CACHE.get(worker_group) {
if at.elapsed() < WORKER_GROUP_PAUSED_CACHE_TTL {
return paused;
}
}
let paused = match sqlx::query_scalar!(
"SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1",
format!("worker__{worker_group}")
)
.fetch_optional(db)
.await
{
Ok(paused) => paused.flatten().unwrap_or(false),
Err(e) => {
tracing::error!("Could not read whether worker group {worker_group} is paused: {e:#}");
return false;
}
};
WORKER_GROUP_PAUSED_CACHE.insert(
worker_group.to_string(),
(paused, std::time::Instant::now()),
);
paused
}
lazy_static::lazy_static! {
pub static ref WINDMILL_DIR: String = {
let dir = std::env::var("WINDMILL_DIR")
@@ -2470,6 +2510,12 @@ pub async fn load_worker_config(
.collect();
priority_tags_sorted.push(PriorityTags { priority: 0, tags: Vec::from_iter(all_tags_set) }); // push the tags that were not listed as high priority with a priority = 0
priority_tags_sorted.sort_by_key(|elt| Reverse(elt.priority)); // sort by priority DESC
// The buckets come out of hash maps: without a fixed order two loads of the same
// config compare unequal, and a reload reads that as a config change.
for bucket in priority_tags_sorted.iter_mut() {
bucket.tags.sort();
}
} else {
// if no priority is used, push all tags with a priority to 0
priority_tags_sorted.push(PriorityTags { priority: 0, tags: worker_tags.clone() });
@@ -2539,6 +2585,7 @@ pub async fn load_worker_config(
env_vars: resolved_env_vars,
native_mode,
object_store_cache_config: config.object_store_cache_config,
paused: config.paused.unwrap_or(false),
})
}
@@ -2629,6 +2676,7 @@ pub struct WorkerConfigOpt {
pub env_vars_allowlist: Option<Vec<String>>,
pub native_mode: Option<bool>,
pub object_store_cache_config: Option<serde_json::Value>,
pub paused: Option<bool>,
}
impl Default for WorkerConfigOpt {
@@ -2648,6 +2696,7 @@ impl Default for WorkerConfigOpt {
env_vars_allowlist: Default::default(),
native_mode: Default::default(),
object_store_cache_config: Default::default(),
paused: Default::default(),
}
}
}
@@ -2670,12 +2719,15 @@ pub struct WorkerConfig {
/// in the group config. Raw JSON: `windmill-common` cannot depend on the object store crate
/// that parses it, and comparing the raw value is what tells a reload the store changed.
pub object_store_cache_config: Option<serde_json::Value>,
/// The group pulls no job from the queue: its workers finish what they hold, keep pinging
/// and leave queued jobs for when it is resumed.
pub paused: bool,
}
impl std::fmt::Debug for WorkerConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "WorkerConfig {{ worker_tags: {:?}, priority_tags_sorted: {:?}, dedicated_worker: {:?}, dedicated_workers: {:?}, init_bash: {:?}, periodic_script_bash: {:?}, periodic_script_interval_seconds: {:?}, cache_clear: {:?}, additional_python_paths: {:?}, pip_local_dependencies: {:?}, env_vars: {:?}, native_mode: {:?}, object_store_cache_config: {} }}",
self.worker_tags, self.priority_tags_sorted, self.dedicated_worker, self.dedicated_workers, self.init_bash, self.periodic_script_bash, self.periodic_script_interval_seconds, self.cache_clear, self.additional_python_paths, self.pip_local_dependencies, self.env_vars.iter().map(|(k, v)| format!("{}: {}{} ({} chars)", k, &v[..3.min(v.len())], "***", v.len())).collect::<Vec<String>>().join(", "), self.native_mode,
write!(f, "WorkerConfig {{ worker_tags: {:?}, priority_tags_sorted: {:?}, dedicated_worker: {:?}, dedicated_workers: {:?}, init_bash: {:?}, periodic_script_bash: {:?}, periodic_script_interval_seconds: {:?}, cache_clear: {:?}, additional_python_paths: {:?}, pip_local_dependencies: {:?}, env_vars: {:?}, native_mode: {:?}, paused: {:?}, object_store_cache_config: {} }}",
self.worker_tags, self.priority_tags_sorted, self.dedicated_worker, self.dedicated_workers, self.init_bash, self.periodic_script_bash, self.periodic_script_interval_seconds, self.cache_clear, self.additional_python_paths, self.pip_local_dependencies, self.env_vars.iter().map(|(k, v)| format!("{}: {}{} ({} chars)", k, &v[..3.min(v.len())], "***", v.len())).collect::<Vec<String>>().join(", "), self.native_mode, self.paused,
// holds bucket credentials
self.object_store_cache_config.as_ref().map(|_| "***").unwrap_or("None"))
}
backend/windmill-runtime-nativets/src/lib.rs
+2 -2
View File
@@ -929,8 +929,8 @@ pub async fn eval_fetch_timeout(
for line in logged.as_deref().unwrap_or(&log).lines() {
tracing::info!(
target: "windmill:job_log",
job_id = ?job_id,
workspace_id = ?w_id_for_tracing,
job_id = %job_id,
workspace_id = %w_id_for_tracing,
"{line}"
);
if *OTEL_JOB_LOGS {
backend/windmill-worker/nsjail/build.csharp.config.proto
+128
View File
@@ -0,0 +1,128 @@
name: "csharp restore/build"
mode: ONCE
hostname: "csharp"
log_level: ERROR
time_limit: {TIMEOUT}
disable_rl: true
cwd: "{JOB_DIR}"
clone_newnet: false
clone_newuser: {CLONE_NEWUSER}
skip_setsid: true
keep_caps: false
keep_env: true
mount_proc: true
mount {
src: "/bin"
dst: "/bin"
is_bind: true
}
mount {
src: "/lib"
dst: "/lib"
is_bind: true
}
mount {
src: "/lib64"
dst: "/lib64"
is_bind: true
mandatory: false
}
mount {
src: "/usr"
dst: "/usr"
is_bind: true
}
mount {
src: "{DOTNET_ROOT}"
dst: "{DOTNET_ROOT}"
is_bind: true
}
mount {
src: "/etc"
dst: "/etc"
is_bind: true
}
# Container runtimes bind exactly these 3 files as separate submounts over
# /etc; nsjail's ro remount of /etc is non-recursive so they stay writable.
# Load-bearing -- do not remove as redundant with the /etc bind above.
mount {
src: "/etc/resolv.conf"
dst: "/etc/resolv.conf"
is_bind: true
mandatory: false
}
mount {
src: "/etc/hosts"
dst: "/etc/hosts"
is_bind: true
mandatory: false
}
mount {
src: "/etc/hostname"
dst: "/etc/hostname"
is_bind: true
mandatory: false
}
mount {
src: "/dev/null"
dst: "/dev/null"
is_bind: true
rw: true
}
mount {
src: "/dev/urandom"
dst: "/dev/urandom"
is_bind: true
}
mount {
src: "/dev/random"
dst: "/dev/random"
is_bind: true
}
{TMP_MOUNT_BLOCK}
# project files, nuget.config, and the published output dir are all here; cwd of dotnet
mount {
src: "{JOB_DIR}"
dst: "{JOB_DIR}"
is_bind: true
rw: true
}
# DOTNET_CLI_HOME + NUGET_PACKAGES (nuget/). Writable, like the install/run jails'
# shared caches.
mount {
src: "{CSHARP_CACHE_DIR}"
dst: "{CSHARP_CACHE_DIR}"
is_bind: true
rw: true
}
mount {
src: "{TRACING_PROXY_CA_CERT_PATH}"
dst: "{TRACING_PROXY_CA_CERT_PATH}"
is_bind: true
mandatory: false
}
envar: "HOME=/tmp"
#{DEV}
backend/windmill-worker/nsjail/lock.php.config.proto
+124
View File
@@ -0,0 +1,124 @@
name: "php composer install"
mode: ONCE
hostname: "php"
log_level: ERROR
time_limit: {TIMEOUT}
disable_rl: true
cwd: "{JOB_DIR}"
clone_newnet: false
clone_newuser: {CLONE_NEWUSER}
skip_setsid: true
keep_caps: false
keep_env: true
mount_proc: true
mount {
src: "/bin"
dst: "/bin"
is_bind: true
}
mount {
src: "/lib"
dst: "/lib"
is_bind: true
}
mount {
src: "/lib64"
dst: "/lib64"
is_bind: true
mandatory: false
}
mount {
src: "/usr"
dst: "/usr"
is_bind: true
}
mount {
src: "/etc"
dst: "/etc"
is_bind: true
}
# Container runtimes bind exactly these 3 files as separate submounts over
# /etc; nsjail's ro remount of /etc is non-recursive so they stay writable.
# Load-bearing -- do not remove as redundant with the /etc bind above.
mount {
src: "/etc/resolv.conf"
dst: "/etc/resolv.conf"
is_bind: true
mandatory: false
}
mount {
src: "/etc/hosts"
dst: "/etc/hosts"
is_bind: true
mandatory: false
}
mount {
src: "/etc/hostname"
dst: "/etc/hostname"
is_bind: true
mandatory: false
}
mount {
src: "/dev/null"
dst: "/dev/null"
is_bind: true
rw: true
}
mount {
src: "/dev/urandom"
dst: "/dev/urandom"
is_bind: true
}
mount {
src: "/dev/random"
dst: "/dev/random"
is_bind: true
}
{TMP_MOUNT_BLOCK}
# composer.json/composer.lock live here and vendor/ is written here; cwd of composer
mount {
src: "{JOB_DIR}"
dst: "{JOB_DIR}"
is_bind: true
rw: true
}
# COMPOSER_HOME: package download cache. Writable, like the install/run jails'
# shared caches.
mount {
src: "{COMPOSER_CACHE_DIR}"
dst: "{COMPOSER_CACHE_DIR}"
is_bind: true
rw: true
}
mount {
src: "{TRACING_PROXY_CA_CERT_PATH}"
dst: "{TRACING_PROXY_CA_CERT_PATH}"
is_bind: true
mandatory: false
}
{CA_MOUNT}
envar: "HOME=/tmp"
#{DEV}
backend/windmill-worker/nsjail/lock.py.config.proto
+133
View File
@@ -0,0 +1,133 @@
name: "python lock compile"
mode: ONCE
hostname: "python"
log_level: ERROR
time_limit: {TIMEOUT}
# Dependency resolution may invoke the PEP 517 build backend of an sdist whose
# metadata is not static, which runs arbitrary package code (e.g. setup.py).
# That build, and uv's bytecode compile pool, size thread/fd pools to the host's
# CPU count, so these mirror download.py.config.proto rather than the run sandbox.
rlimit_as: 8192
rlimit_cpu: 1000
rlimit_fsize: 1024
rlimit_nofile: 10000
envar: "HOME=/tmp"
envar: "LD_LIBRARY_PATH=/usr/local/lib:$LD_LIBRARY_PATH"
cwd: "{JOB_DIR}"
clone_newnet: false
clone_newuser: {CLONE_NEWUSER}
skip_setsid: true
keep_caps: true
keep_env: true
mount_proc: true
mount {
src: "/bin"
dst: "/bin"
is_bind: true
}
mount {
src: "/lib"
dst: "/lib"
is_bind: true
}
mount {
src: "/lib64"
dst: "/lib64"
is_bind: true
mandatory: false
}
mount {
src: "/usr"
dst: "/usr"
is_bind: true
}
mount {
src: "/etc"
dst: "/etc"
is_bind: true
}
# Container runtimes bind exactly these 3 files as separate submounts over
# /etc; nsjail's ro remount of /etc is non-recursive so they stay writable.
# Load-bearing -- do not remove as redundant with the /etc bind above.
mount {
src: "/etc/resolv.conf"
dst: "/etc/resolv.conf"
is_bind: true
mandatory: false
}
mount {
src: "/etc/hosts"
dst: "/etc/hosts"
is_bind: true
mandatory: false
}
mount {
src: "/etc/hostname"
dst: "/etc/hostname"
is_bind: true
mandatory: false
}
mount {
src: "/dev/null"
dst: "/dev/null"
is_bind: true
rw: true
}
mount {
src: "/dev/urandom"
dst: "/dev/urandom"
is_bind: true
}
{TMP_MOUNT_BLOCK}
# requirements.in -> requirements.txt, cwd of uv
mount {
src: "{JOB_DIR}"
dst: "{JOB_DIR}"
is_bind: true
rw: true
}
# shared uv cache: resolution metadata + where sdist build backends run
# (builds-v0/). Writable, as the install jail's target already is; a poisoned
# shared cache is the known residual, not a new one.
mount {
src: "{UV_CACHE_DIR}"
dst: "{UV_CACHE_DIR}"
is_bind: true
rw: true
}
mount {
src: "{PY_INSTALL_DIR}"
dst: "{PY_INSTALL_DIR}"
is_bind: true
}
{INDEX_CERT_MOUNT}
mount {
src: "{TRACING_PROXY_CA_CERT_PATH}"
dst: "{TRACING_PROXY_CA_CERT_PATH}"
is_bind: true
mandatory: false
}
#{DEV}
backend/windmill-worker/src/csharp_executor.rs
+91 -28
View File
@@ -33,7 +33,7 @@ use crate::{
get_proxy_envs_for_lang,
handle_child::handle_child,
is_sandboxing_enabled, read_ee_registry_with_workspace_override, CSHARP_CACHE_DIR,
DISABLE_NUSER, DOTNET_PATH, HOME_ENV, NSJAIL_PATH, NUGET_CONFIG, PATH_ENV,
DISABLE_NUSER, DOTNET_PATH, HOME_ENV, NSJAIL_PATH, NUGET_CONFIG, PATH_ENV, PROXY_ENVS,
TRACING_PROXY_CA_CERT_PATH, TZ_ENV,
};
#[cfg(feature = "csharp")]
@@ -47,6 +47,9 @@ use crate::SYSTEM_ROOT;
#[cfg(feature = "csharp")]
const NSJAIL_CONFIG_RUN_CSHARP_CONTENT: &str = include_str!("../nsjail/run.csharp.config.proto");
#[cfg(feature = "csharp")]
const NSJAIL_CONFIG_BUILD_CSHARP_CONTENT: &str =
include_str!("../nsjail/build.csharp.config.proto");
#[cfg(feature = "csharp")]
#[cfg(windows)]
@@ -124,6 +127,9 @@ pub async fn generate_nuget_lockfile(
gen_cs_proj(code, job_dir, reqs, lines_to_remove)?;
// `dotnet restore` resolves and downloads packages but does not import their
// MSBuild `build/*.targets`, so it runs no package code — only `dotnet publish`
// (jailed below) does — so it runs unsandboxed.
let mut gen_lockfile_cmd = Command::new(DOTNET_PATH.as_str());
gen_lockfile_cmd
.current_dir(job_dir)
@@ -384,31 +390,82 @@ async fn build_cs_proj(
}
}
let mut build_cs_cmd = Command::new(DOTNET_PATH.as_str());
build_cs_cmd
.current_dir(job_dir)
.env_clear()
.env("PATH", PATH_ENV.as_str())
.env("BASE_INTERNAL_URL", base_internal_url)
.env("HOME", HOME_ENV.as_str())
.env("DOTNET_CLI_HOME", &*CSHARP_CACHE_DIR)
.env("NUGET_PACKAGES", format!("{}/nuget", *CSHARP_CACHE_DIR))
.env("DOTNET_CLI_TELEMETRY_OPTOUT", "true")
.env("DOTNET_NOLOGO", "true")
.env("MSBUILDDISABLENODEREUSE", "1")
.env("DOTNET_ROOT", DOTNET_ROOT.as_str())
.args(vec![
"publish",
"--configuration",
"Release",
"-o",
&format!("{job_dir}/out"),
"--no-self-contained",
"-p:PublishSingleFile=true",
"-p:IncludeNativeLibrariesForSelfExtract=true",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let out_dir = format!("{job_dir}/out");
let publish_args = vec![
"publish",
"--configuration",
"Release",
"-o",
out_dir.as_str(),
"--no-self-contained",
"-p:PublishSingleFile=true",
"-p:IncludeNativeLibrariesForSelfExtract=true",
];
// `dotnet publish` imports and runs the MSBuild targets/tasks of restored NuGet
// packages (a package's build/*.targets execute at build, not at restore), i.e.
// arbitrary package code. Confine it under nsjail when sandboxing is on, the same
// way the run step is; HOME is set to /tmp by the config, keep_env passes the
// explicit dotnet/proxy vars, and the job dir + shared nuget cache are writable.
// Without sandboxing, publish runs directly with no isolation.
// mut is required by the #[cfg(windows)] env block below; unused on unix.
#[allow(unused_mut)]
let mut build_cs_cmd = if is_sandboxing_enabled() {
std::fs::create_dir_all(&*CSHARP_CACHE_DIR)?;
let nsjail_timeout = resolve_nsjail_timeout(conn, w_id, *job_id, None).await;
write_file(
job_dir,
"build.config.proto",
NSJAIL_CONFIG_BUILD_CSHARP_CONTENT
.replace("{TIMEOUT}", &nsjail_timeout)
.replace("{JOB_DIR}", job_dir)
.replace("{CSHARP_CACHE_DIR}", &*CSHARP_CACHE_DIR)
.replace("{DOTNET_ROOT}", DOTNET_ROOT.as_str())
.replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string())
.replace("{TRACING_PROXY_CA_CERT_PATH}", &*TRACING_PROXY_CA_CERT_PATH)
.replace("#{DEV}", DEV_CONF_NSJAIL)
.replace(
"{TMP_MOUNT_BLOCK}",
&resolve_nsjail_tmp_mount_block(job_dir).await,
)
.as_str(),
)?;
let mut nsjail_args = vec!["--config", "build.config.proto", "--", DOTNET_PATH.as_str()];
nsjail_args.extend(publish_args.iter().copied());
let mut cmd = Command::new(NSJAIL_PATH.as_str());
cmd.current_dir(job_dir)
.env_clear()
.env("PATH", PATH_ENV.as_str())
.env("BASE_INTERNAL_URL", base_internal_url)
.env("DOTNET_CLI_HOME", &*CSHARP_CACHE_DIR)
.env("NUGET_PACKAGES", format!("{}/nuget", *CSHARP_CACHE_DIR))
.env("DOTNET_CLI_TELEMETRY_OPTOUT", "true")
.env("DOTNET_NOLOGO", "true")
.env("MSBUILDDISABLENODEREUSE", "1")
.env("DOTNET_ROOT", DOTNET_ROOT.as_str())
.envs(PROXY_ENVS.clone())
.envs(&*crate::worker::WHITELIST_ENVS)
.args(nsjail_args)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
cmd
} else {
let mut cmd = Command::new(DOTNET_PATH.as_str());
cmd.current_dir(job_dir)
.env_clear()
.env("PATH", PATH_ENV.as_str())
.env("BASE_INTERNAL_URL", base_internal_url)
.env("HOME", HOME_ENV.as_str())
.env("DOTNET_CLI_HOME", &*CSHARP_CACHE_DIR)
.env("NUGET_PACKAGES", format!("{}/nuget", *CSHARP_CACHE_DIR))
.env("DOTNET_CLI_TELEMETRY_OPTOUT", "true")
.env("DOTNET_NOLOGO", "true")
.env("MSBUILDDISABLENODEREUSE", "1")
.env("DOTNET_ROOT", DOTNET_ROOT.as_str())
.args(publish_args.clone())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
cmd
};
#[cfg(windows)]
build_cs_cmd
@@ -433,14 +490,20 @@ async fn build_cs_proj(
.unwrap_or_else(|_| format!("{}\\AppData\\Local", HOME_ENV.as_str())),
);
let build_cs_process = start_child_process(build_cs_cmd, DOTNET_PATH.as_str(), true).await?;
let build_cs_executable = if is_sandboxing_enabled() {
NSJAIL_PATH.as_str()
} else {
DOTNET_PATH.as_str()
};
let build_cs_process = start_child_process(build_cs_cmd, build_cs_executable, true).await?;
handle_child(
job_id,
conn,
mem_peak,
canceled_by,
build_cs_process,
false,
// jailed when sandboxing is on, so mem sampling finds the real process
is_sandboxing_enabled(),
worker_name,
w_id,
"dotnet publish",
backend/windmill-worker/src/handle_child.rs
+117 -51
View File
@@ -210,6 +210,53 @@ pub async fn handle_child(
}
}
impl KillReason {
/// Reports a kill once the process is gone: a span event on the enclosing span and,
/// through the logs bridge, a log record carrying the job's log context.
fn emit_canceled_event(
&self,
job_id: &Uuid,
timeout_reason: &str,
signal: &'static str,
detected_at: chrono::DateTime<chrono::Utc>,
) {
// `force` is only known for a soft cancel: a job completed under the worker
// may have been force canceled or completed by anything else server-side.
let (kill_reason, canceled_by, canceled_reason, force) = match self {
KillReason::TooManyLogs => return,
KillReason::Timeout { .. } => {
("timeout", Some("timeout"), Some(timeout_reason), None)
}
KillReason::Cancelled(by) => (
"cancelled",
by.as_ref().and_then(|x| x.username.as_deref()),
by.as_ref().and_then(|x| x.reason.as_deref()),
Some(false),
),
KillReason::AlreadyCompleted => ("already_completed", None, None, None),
};
let exited_at = chrono::Utc::now();
let root_job = windmill_common::log_context::current_log_context()
.and_then(|c| c.root_job.clone());
let rfc3339 = |t: chrono::DateTime<chrono::Utc>| {
t.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
};
tracing::info!(
job_id = %job_id,
root_job,
kill_reason,
canceled_by,
canceled_reason,
force,
kill_signal = signal,
cancel_detected_at = %rfc3339(detected_at),
process_exited_at = %rfc3339(exited_at),
kill_duration_ms = (exited_at - detected_at).num_milliseconds(),
"job.canceled"
);
}
}
let (timeout_duration, timeout_warn_msg, timeout_source) =
resolve_job_timeout(&conn, w_id, job_id, custom_timeout).await;
if let Some(msg) = timeout_warn_msg {
@@ -232,6 +279,7 @@ pub async fn handle_child(
UpdateJobPollingExit::AlreadyCompleted => KillReason::AlreadyCompleted,
},
};
let detected_at = chrono::Utc::now();
tx.send(()).expect("rx should never be dropped");
drop(tx);
@@ -268,59 +316,78 @@ pub async fn handle_child(
}
};
#[allow(unused_variables)]
if let Some(id) = child.id() {
if *MAX_WAIT_FOR_SIGINT > 0 {
#[cfg(any(target_os = "linux", target_os = "macos"))]
signal::kill(Pid::from_raw(id as i32), Signal::SIGINT).unwrap();
// Resolves once the process is reaped, to the last signal it was sent, or to `None`
// when the kill could not be confirmed. SIGINT and SIGTERM are only sent where the
// sends below are compiled in: elsewhere an exit during their wait is unsignaled.
let unix_signal = |s: &'static str| {
if cfg!(any(target_os = "linux", target_os = "macos")) { s } else { "none" }
};
let kill = async {
#[allow(unused_variables)]
if let Some(id) = child.id() {
if *MAX_WAIT_FOR_SIGINT > 0 {
#[cfg(any(target_os = "linux", target_os = "macos"))]
signal::kill(Pid::from_raw(id as i32), Signal::SIGINT).unwrap();
for _ in 0..*MAX_WAIT_FOR_SIGINT {
for _ in 0..*MAX_WAIT_FOR_SIGINT {
if child.try_wait().is_ok_and(|x| x.is_some()) {
break;
}
sleep(Duration::from_secs(1)).await;
}
if child.try_wait().is_ok_and(|x| x.is_some()) {
break;
set_reason.await;
return Ok(Some(unix_signal("SIGINT")));
}
sleep(Duration::from_secs(1)).await;
}
if child.try_wait().is_ok_and(|x| x.is_some()) {
set_reason.await;
return Ok(Err(kill_reason));
if sigterm {
#[cfg(any(target_os = "linux", target_os = "macos"))]
signal::kill(Pid::from_raw(id as i32), Signal::SIGTERM).unwrap();
for _ in 0..*MAX_WAIT_FOR_SIGTERM {
if child.try_wait().is_ok_and(|x| x.is_some()) {
break;
}
sleep(Duration::from_secs(1)).await;
}
if child.try_wait().is_ok_and(|x| x.is_some()) {
set_reason.await;
return Ok(Some(unix_signal("SIGTERM")));
}
}
}
if sigterm {
#[cfg(any(target_os = "linux", target_os = "macos"))]
signal::kill(Pid::from_raw(id as i32), Signal::SIGTERM).unwrap();
for _ in 0..*MAX_WAIT_FOR_SIGTERM {
if child.try_wait().is_ok_and(|x| x.is_some()) {
break;
#[cfg(windows)]
{
let pid_to_kill = child.id();
let killed = match kill_process_tree(pid_to_kill).await {
Ok(_) => {
tracing::debug!(
"successfully killed process tree with PID: {:?}",
pid_to_kill
);
true
}
sleep(Duration::from_secs(1)).await;
}
if child.try_wait().is_ok_and(|x| x.is_some()) {
set_reason.await;
return Ok(Err(kill_reason));
}
Err(e) => {
tracing::error!("failed to kill process tree: {:?}", e);
false
}
};
set_reason.await;
return Ok(killed.then_some("taskkill"));
}
}
#[cfg(windows)]
{
let pid_to_kill = child.id();
match kill_process_tree(pid_to_kill).await {
Ok(_) => tracing::debug!(
"successfully killed process tree with PID: {:?}",
pid_to_kill
),
Err(e) => tracing::error!("failed to kill process tree: {:?}", e),
};
set_reason.await;
return Ok(Err(kill_reason));
}
#[cfg(unix)]
{
/* send SIGKILL and reap child process */
let (_, kill) = future::join(set_reason, Box::into_pin(child.kill())).await;
kill.map(|()| Err(kill_reason))
#[cfg(unix)]
{
/* send SIGKILL and reap child process */
let (_, kill) = future::join(set_reason, Box::into_pin(child.kill())).await;
kill.map(|()| Some("SIGKILL"))
}
};
let signal: io::Result<Option<&'static str>> = kill.await;
if let Some(signal) = signal? {
kill_reason.emit_canceled_event(&job_id, &timeout_reason, signal, detected_at);
}
Ok(Err(kill_reason))
};
let mut stream_result = Vec::new();
@@ -957,9 +1024,9 @@ where
let conn = conn.clone();
// No tick at t=0: a job that finishes within the first interval issues none of the
// ping/metric statements below. Cancels are picked up by the first job ping (500 ms in,
// 2 s on agent workers). Memory is still sampled at start, without SQL: it is the only
// `mem_peak` reading a short job gets, and the completed job row reports it.
// ping/metric statements below. Cancels are picked up by the next job ping. Memory is still
// sampled at start, without SQL: it is the only `mem_peak` reading a short job gets,
// and the completed job row reports it.
let mut interval = interval_at(Instant::now() + update_job_interval, update_job_interval);
interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
*mem_peak = (*mem_peak).max(get_mem.next().await.unwrap_or(0));
@@ -1023,11 +1090,10 @@ where
}
}
}
if matches!(conn, Connection::Http(_)) {
if i % 4 != 0 {
// only ping every 4th time (2s) on http agent mode
continue;
}
// An agent worker pays an HTTP round trip per ping, so it pings every 2 s for
// the first 10 s, then every 5 s.
if matches!(conn, Connection::Http(_)) && i % if i < 20 { 4 } else { 10 } != 0 {
continue;
}
let ping_job_status = ping_job_status(&conn, &job_id, Some(*mem_peak), if current_mem > 0 { Some(current_mem) } else { None }).await.unwrap_or_else(|e| {
tracing::error!("Unable to ping job status for job {job_id}. Error was: {:?}", e);
backend/windmill-worker/src/php_executor.rs
+99 -11
View File
@@ -17,6 +17,7 @@ use windmill_queue::MiniPulledJob;
use windmill_parser::{MainArgSignature, Typ};
use windmill_queue::{append_logs, CanceledBy};
use crate::common::DEV_CONF_NSJAIL;
use crate::{
common::{
build_command_with_isolation, check_executor_binary_exists, create_args_and_out_file,
@@ -24,11 +25,13 @@ use crate::{
resolve_nsjail_tmp_mount_block, start_child_process, MaybeLock, OccupancyMetrics,
},
handle_child::handle_child,
is_sandboxing_enabled, COMPOSER_CACHE_DIR, COMPOSER_PATH, DISABLE_NUSER, NSJAIL_PATH, PHP_PATH,
is_sandboxing_enabled, COMPOSER_CACHE_DIR, COMPOSER_PATH, DISABLE_NUSER, NSJAIL_PATH, PATH_ENV,
PHP_PATH, PROXY_ENVS, TRACING_PROXY_CA_CERT_PATH,
};
use windmill_common::client::AuthedClient;
const NSJAIL_CONFIG_RUN_PHP_CONTENT: &str = include_str!("../nsjail/run.php.config.proto");
const NSJAIL_CONFIG_LOCK_PHP_CONTENT: &str = include_str!("../nsjail/lock.php.config.proto");
lazy_static::lazy_static! {
static ref RE: Regex = Regex::new(r"^//\s?(\S+)\s*$").unwrap();
@@ -199,15 +202,99 @@ pub async fn composer_install(
write_file(job_dir, "composer.lock", lock)?;
}
let mut child_cmd = Command::new(&*COMPOSER_PATH);
let args = vec!["install", "--no-dev", "--no-progress"];
child_cmd
.current_dir(job_dir)
.env("COMPOSER_HOME", &*COMPOSER_CACHE_DIR)
.args(args)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let child_process = start_child_process(child_cmd, &*COMPOSER_PATH, false).await?;
// `composer install` runs the plugin and script code of installed packages, so
// when sandboxing is on it is confined under nsjail like the run step. Inside the
// jail the environment is cleared down to what composer needs: PATH, COMPOSER_HOME
// (pointed at the mounted cache dir), the proxies, the operator-controlled
// WHITELIST_ENVS, and every other COMPOSER_* var from the worker env — Windmill has
// no managed composer-registry setting, so operators supply private Packagist auth
// and composer tuning (COMPOSER_AUTH, COMPOSER_PROCESS_TIMEOUT, …) through those.
// COMPOSER_CACHE_DIR is dropped so it can't point composer outside the jail, and a
// custom COMPOSER_CAFILE/COMPOSER_CAPATH is bind-mounted so TLS to private feeds
// keeps working. COMPOSER_PATH is Windmill-internal. Without sandboxing composer
// runs directly, with no isolation.
let composer_args = ["install", "--no-dev", "--no-progress"];
let child_cmd = if is_sandboxing_enabled() {
// The cache dir is bind-mounted into the jail, so it must exist before
// nsjail builds the mount tree.
std::fs::create_dir_all(&*COMPOSER_CACHE_DIR)?;
let nsjail_timeout = resolve_nsjail_timeout(conn, w_id, *job_id, None).await;
// A custom CA bundle/dir lives outside the jail's mounts; bind it in so
// composer's TLS can read it (mandatory: false → skipped when the path is unset).
let ca_mount = ["COMPOSER_CAFILE", "COMPOSER_CAPATH"]
.iter()
.filter_map(|k| std::env::var(k).ok().filter(|v| !v.trim().is_empty()))
.map(|p| {
format!(
"mount {{\n src: \"{p}\"\n dst: \"{p}\"\n is_bind: true\n mandatory: false\n}}\n"
)
})
.collect::<String>();
write_file(
job_dir,
"lock.config.proto",
NSJAIL_CONFIG_LOCK_PHP_CONTENT
.replace("{TIMEOUT}", &nsjail_timeout)
.replace("{JOB_DIR}", job_dir)
.replace("{COMPOSER_CACHE_DIR}", &*COMPOSER_CACHE_DIR)
.replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string())
.replace("{TRACING_PROXY_CA_CERT_PATH}", &*TRACING_PROXY_CA_CERT_PATH)
.replace("{CA_MOUNT}", &ca_mount)
.replace("#{DEV}", DEV_CONF_NSJAIL)
.replace(
"{TMP_MOUNT_BLOCK}",
&resolve_nsjail_tmp_mount_block(job_dir).await,
)
.as_str(),
)?;
let composer_passthrough_envs: Vec<(String, String)> = std::env::vars()
.filter(|(k, _)| {
k.starts_with("COMPOSER_")
&& k != "COMPOSER_HOME"
&& k != "COMPOSER_CACHE_DIR"
&& k != "COMPOSER_PATH"
})
.collect();
let mut nsjail_args = vec![
"--config",
"lock.config.proto",
"--",
COMPOSER_PATH.as_str(),
];
nsjail_args.extend(composer_args);
let mut child_cmd = Command::new(NSJAIL_PATH.as_str());
child_cmd
.current_dir(job_dir)
.env_clear()
// HOME is set to /tmp by the nsjail config; keep_env passes the rest.
.env("PATH", PATH_ENV.as_str())
.envs(PROXY_ENVS.clone())
.envs(&*crate::worker::WHITELIST_ENVS)
.envs(composer_passthrough_envs)
.env("COMPOSER_HOME", &*COMPOSER_CACHE_DIR)
.args(nsjail_args)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
child_cmd
} else {
let mut child_cmd = Command::new(&*COMPOSER_PATH);
child_cmd
.current_dir(job_dir)
.env("COMPOSER_HOME", &*COMPOSER_CACHE_DIR)
.args(composer_args)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
child_cmd
};
let child_executable = if is_sandboxing_enabled() {
NSJAIL_PATH.as_str()
} else {
COMPOSER_PATH.as_str()
};
let child_process = start_child_process(child_cmd, child_executable, false).await?;
handle_child(
job_id,
@@ -215,7 +302,8 @@ pub async fn composer_install(
mem_peak,
canceled_by,
child_process,
false,
// jailed when sandboxing is on, so mem sampling finds the real process
is_sandboxing_enabled(),
worker_name,
w_id,
"composer install",
backend/windmill-worker/src/python_executor.rs
+156 -59
View File
@@ -116,6 +116,7 @@ async fn handle_piptar_uploads(mut rx: tokio::sync::mpsc::UnboundedReceiver<Pipt
const NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT: &str = include_str!("../nsjail/download.py.config.proto");
const NSJAIL_CONFIG_RUN_PYTHON3_CONTENT: &str = include_str!("../nsjail/run.python3.config.proto");
const NSJAIL_CONFIG_LOCK_PY_CONTENT: &str = include_str!("../nsjail/lock.py.config.proto");
pub const RELATIVE_PYTHON_LOADER: &str = include_str!("../loader.py");
/// Every file exchanged with a job is UTF-8 by construction, so the interpreter
@@ -156,7 +157,7 @@ use crate::{
},
get_proxy_envs_for_lang,
handle_child::handle_child,
is_sandboxing_enabled, read_ee_registry_with_workspace_override,
is_sandboxing_enabled, read_ee_registry,
worker_utils::ping_job_status,
PyV, DISABLE_NUSER, HOME_ENV, INDEX_CERT, NATIVE_CERT, NSJAIL_AVAILABLE, NSJAIL_PATH,
NSJAIL_PY_RLIMIT_AS_MB, PATH_ENV, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, PROXY_ENVS,
@@ -279,7 +280,12 @@ pub async fn uv_pip_compile(
#[cfg(feature = "enterprise")]
let requirements = replace_pip_secret(conn, w_id, &requirements, worker_name, job_id).await?;
let ws_suffix = crate::workspace_registry_cache_suffix(w_id).await;
let pip_indexes = PipIndexes::snapshot(w_id).await;
let ws_suffix = if pip_indexes.ws_scoped {
format!(":ws:{w_id}")
} else {
String::new()
};
let exclude_newer_suffix = uv_exclude_newer
.as_deref()
.map(|v| format!("-en{}", v))
@@ -343,31 +349,12 @@ pub async fn uv_pip_compile(
if no_cache {
args.extend(["--no-cache"]);
}
let pip_extra_index_url = read_ee_registry_with_workspace_override(
PIP_EXTRA_INDEX_URL.read().await.clone(),
"pip_extra_index_url",
"pip extra index url",
job_id,
w_id,
conn,
)
.await
.map(handle_ephemeral_token);
let (pip_extra_index_url, pip_index_url) = pip_indexes.urls(job_id, w_id, conn).await;
if let Some(url) = pip_extra_index_url.as_ref() {
url.split(",").for_each(|url| {
args.extend(["--extra-index-url", url]);
});
}
let pip_index_url = read_ee_registry_with_workspace_override(
PIP_INDEX_URL.read().await.clone(),
"pip_index_url",
"pip index url",
job_id,
w_id,
conn,
)
.await
.map(handle_ephemeral_token);
if let Some(url) = pip_index_url.as_ref() {
args.extend(["--index-url", url]);
}
@@ -393,18 +380,76 @@ pub async fn uv_pip_compile(
#[cfg(unix)]
let uv_cmd = UV_PATH.as_str();
let mut child_cmd = Command::new(uv_cmd);
child_cmd
.current_dir(job_dir)
.env_clear()
.env("HOME", HOME_ENV.to_string())
.env("PATH", PATH_ENV.to_string())
.env("UV_PYTHON_INSTALL_DIR", PY_INSTALL_DIR.to_string())
.env("UV_INDEX_STRATEGY", uv_index_strategy)
.envs(PROXY_ENVS.clone())
.args(&args)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
// Resolving an sdist whose metadata is not static makes uv invoke its
// PEP 517 build backend, which runs arbitrary package code (e.g.
// setup.py). When sandboxing is on, confine that under nsjail exactly
// like the install and run steps already are — otherwise it runs as the
// worker, in the worker's namespaces, with the worker's env and tokens
// readable. The jail gets network to the index, the job dir and the
// shared uv cache writable, and the managed interpreter read-only.
let mut child_cmd = if is_sandboxing_enabled() {
let index_cert_mount = INDEX_CERT
.as_ref()
.map(|p| {
format!(
"mount {{\n src: \"{p}\"\n dst: \"{p}\"\n is_bind: true\n mandatory: false\n}}\n"
)
})
.unwrap_or_default();
// The cache dir is bind-mounted into the jail, so it must exist before
// nsjail builds the mount tree (uv would otherwise create it itself).
std::fs::create_dir_all(&*UV_CACHE_DIR)?;
let nsjail_timeout = resolve_nsjail_timeout(conn, w_id, *job_id, None).await;
write_file(
job_dir,
"lock.config.proto",
NSJAIL_CONFIG_LOCK_PY_CONTENT
.replace("{TIMEOUT}", &nsjail_timeout)
.replace("{JOB_DIR}", job_dir)
.replace("{UV_CACHE_DIR}", &*UV_CACHE_DIR)
.replace("{PY_INSTALL_DIR}", &*PY_INSTALL_DIR)
.replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string())
.replace("{INDEX_CERT_MOUNT}", &index_cert_mount)
.replace("{TRACING_PROXY_CA_CERT_PATH}", &*TRACING_PROXY_CA_CERT_PATH)
.replace("#{DEV}", DEV_CONF_NSJAIL)
.replace(
"{TMP_MOUNT_BLOCK}",
&resolve_nsjail_tmp_mount_block(job_dir).await,
)
.as_str(),
)?;
let mut nsjail_args = vec!["--config", "lock.config.proto", "--", uv_cmd];
nsjail_args.extend(args.iter().copied());
let mut child_cmd = Command::new(NSJAIL_PATH.as_str());
child_cmd
.current_dir(job_dir)
.env_clear()
// HOME is set to /tmp by the nsjail config; keep_env passes the rest.
.env("PATH", PATH_ENV.to_string())
.env("UV_PYTHON_INSTALL_DIR", PY_INSTALL_DIR.to_string())
.env("UV_INDEX_STRATEGY", uv_index_strategy)
.envs(PROXY_ENVS.clone())
.args(nsjail_args)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
child_cmd
} else {
let mut child_cmd = Command::new(uv_cmd);
child_cmd
.current_dir(job_dir)
.env_clear()
.env("HOME", HOME_ENV.to_string())
.env("PATH", PATH_ENV.to_string())
.env("UV_PYTHON_INSTALL_DIR", PY_INSTALL_DIR.to_string())
.env("UV_INDEX_STRATEGY", uv_index_strategy)
.envs(PROXY_ENVS.clone())
.args(&args)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
child_cmd
};
if let Some(mirror) = UV_PYTHON_INSTALL_MIRROR.read().await.as_ref() {
child_cmd.env("UV_PYTHON_INSTALL_MIRROR", mirror);
@@ -454,7 +499,12 @@ pub async fn uv_pip_compile(
);
}
let child_process = start_child_process(child_cmd, uv_cmd, false).await?;
let child_executable = if is_sandboxing_enabled() {
NSJAIL_PATH.as_str()
} else {
uv_cmd
};
let child_process = start_child_process(child_cmd, child_executable, false).await?;
append_logs(&job_id, &w_id, logs, conn).await;
handle_child(
job_id,
@@ -462,7 +512,8 @@ pub async fn uv_pip_compile(
mem_peak,
canceled_by,
child_process,
false,
// jailed when sandboxing is on, so mem sampling finds the real process
is_sandboxing_enabled(),
worker_name,
&w_id,
// TODO: Rename to uv-pip-compile?
@@ -2068,6 +2119,62 @@ Returned from server: py_version - {:?}, py_version_v2 - {:?}
Ok((pyv, additional_python_paths))
}
/// The pip index settings a job resolves and installs from, and whether its workspace
/// overrides any registry.
///
/// Both come from one snapshot of the workspace's overrides: read separately, a settings
/// reload in between could file an install from an overridden index under a shared cache name.
struct PipIndexes {
extra_index_url: Option<String>,
index_url: Option<String>,
/// Caches filled through these URLs must not be shared with other workspaces.
ws_scoped: bool,
}
impl PipIndexes {
async fn snapshot(w_id: &str) -> Self {
let overrides = crate::workspace_registry_overrides(w_id).await;
let ws_value = |key: &str| match overrides.get(key) {
Some(serde_json::Value::String(s)) => Some(s.clone()),
_ => None,
};
// An empty value means unset, and an empty workspace override unsets the global one.
Self {
extra_index_url: ws_value("pip_extra_index_url")
.or(PIP_EXTRA_INDEX_URL.read().await.clone())
.filter(|s| !s.trim().is_empty()),
index_url: ws_value("pip_index_url")
.or(PIP_INDEX_URL.read().await.clone())
.filter(|s| !s.trim().is_empty()),
ws_scoped: !overrides.is_empty(),
}
}
/// The `(extra index, index)` URLs to hand uv. Kept apart from [`Self::snapshot`]: it runs
/// `EPHEMERAL_TOKEN_CMD`, which a caller served from a cache has no reason to pay for.
async fn urls(
self,
job_id: &Uuid,
w_id: &str,
conn: &Connection,
) -> (Option<String>, Option<String>) {
(
read_ee_registry(
self.extra_index_url,
"pip extra index url",
job_id,
w_id,
conn,
)
.await
.map(handle_ephemeral_token),
read_ee_registry(self.index_url, "pip index url", job_id, w_id, conn)
.await
.map(handle_ephemeral_token),
)
}
}
lazy_static::lazy_static! {
static ref PIP_SECRET_VARIABLE: Regex = Regex::new(r"\$\{PIP_SECRET:([^\s\}]+)\}").unwrap();
@@ -2527,28 +2634,18 @@ pub async fn handle_python_reqs(
job_id
);
let pip_indexes = (
read_ee_registry_with_workspace_override(
PIP_EXTRA_INDEX_URL.read().await.clone(),
"pip_extra_index_url",
"pip extra index url",
job_id,
w_id,
conn,
)
.await
.map(handle_ephemeral_token),
read_ee_registry_with_workspace_override(
PIP_INDEX_URL.read().await.clone(),
"pip_index_url",
"pip index url",
job_id,
w_id,
conn,
)
.await
.map(handle_ephemeral_token),
);
let pip_indexes = PipIndexes::snapshot(w_id).await;
let ws_scoped = pip_indexes.ws_scoped;
let pip_indexes = pip_indexes.urls(job_id, w_id, conn).await;
// The wheel dir is shared by every workspace on the worker, and by every worker through
// the object store, under a name that says nothing of the index it was filled from. A
// workspace with its own registries gets its own, as its resolution cache does. The id
// is hashed: it is not restricted to characters safe in a file name.
let ws_dir_suffix = if ws_scoped {
format!("-ws-{}", &calculate_hash(w_id)[..16])
} else {
String::new()
};
// Cached paths
let mut req_with_penv: Vec<(String, String)> = vec![];
@@ -2577,7 +2674,7 @@ pub async fn handle_python_reqs(
let py_prefix = &py_version.to_cache_dir(false);
let venv_p = format!(
"{py_prefix}/{}",
"{py_prefix}/{}{ws_dir_suffix}",
req.replace(' ', "").replace('/', "").replace(':', "")
);
if metadata(venv_p.clone() + "/.valid.windmill").await.is_ok() {
backend/windmill-worker/src/worker.rs
+22
View File
@@ -949,6 +949,22 @@ pub async fn workspace_registry_cache_suffix(w_id: &str) -> String {
}
}
/// The registry overrides of a workspace, empty when it has none.
///
/// One snapshot: registry settings reload under running jobs, so a caller whose cache key
/// has to agree with the registry values it used reads both from here rather than calling
/// [`workspace_registry_cache_suffix`] and a `read_ee_registry_*` helper separately.
pub async fn workspace_registry_overrides(
w_id: &str,
) -> std::collections::HashMap<String, serde_json::Value> {
let registries = WORKSPACE_REGISTRIES.read().await;
registries
.as_ref()
.and_then(|m| m.get(w_id))
.cloned()
.unwrap_or_default()
}
/// The name a build artifact is cached under, derived from `base` — the runnable's own
/// cache-key input — and its inline modules.
///
@@ -2005,6 +2021,7 @@ pub fn create_span_with_name(
root_job = field::Empty,
workspace_id = %arc_job.workspace_id,
worker = %worker_name,
worker_group = %*WORKER_GROUP,
hostname = field::Empty,
tag = %arc_job.tag,
language = field::Empty,
@@ -2156,6 +2173,7 @@ pub fn log_context_for_job(
job_id: Some(arc_job.id.to_string()),
workspace_id: Some(arc_job.workspace_id.clone()),
worker: Some(worker_name.to_string()),
worker_group: Some(WORKER_GROUP.clone()),
tag: Some(arc_job.tag.clone()),
job_kind: Some(arc_job.kind.as_str().to_string()),
created_by: Some(arc_job.created_by.clone()),
@@ -3406,6 +3424,10 @@ pub async fn run_worker(
tokio::time::sleep(Duration::from_millis(200)).await;
continue;
}
} else if WORKER_CONFIG.load().paused {
// Checked after the same-worker channel: a flow this worker already runs
// still needs it for its remaining same-worker steps.
Ok(None)
} else {
match &conn {
Connection::Sql(db) => {