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

feat: add dynamic skip for schedules (#6739)

Browse Source 
* feat: add dynamic skip for schedules

* npm check

* accidental rename

* fixing tests

* Update SQLx metadata

* simplify api / queries

---------

Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Alexander Petric
2025-10-08 09:18:15 +00:00
committed by GitHub
co-authored by windmill-internal-app[bot]
parent 0af13254f8
commit 82c83ff658
54 changed files with 838 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json
Generated
+5 -5
View File
@@ -46,11 +46,11 @@
]
},
"nullable": [
false,
false,
false,
false,
false,
true,
true,
true,
true,
true,
true,
true
]
backend/.sqlx/query-12d37d75a429c0ddf2b2c190ab28bea5aefd27d0ed8a1bb2c8b3c1b0ece4efb7.json
Generated
+41
View File
@@ -0,0 +1,41 @@
{
"db_name": "PostgreSQL",
"query": "WITH to_update AS (\n SELECT q.id, q.workspace_id, r.ping, COALESCE(zjc.counter, 0) as counter\n FROM v2_job_queue q\n JOIN v2_job j ON j.id = q.id\n JOIN v2_job_runtime r ON r.id = j.id\n LEFT JOIN zombie_job_counter zjc ON zjc.job_id = q.id\n WHERE ping < now() - ($1 || ' seconds')::interval\n AND running = true\n AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow')\n AND same_worker = false\n AND (zjc.counter IS NULL OR zjc.counter <= $2)\n FOR UPDATE of q SKIP LOCKED\n ),\n zombie_jobs AS (\n UPDATE v2_job_queue q\n SET running = false, started_at = null\n FROM to_update tu\n WHERE q.id = tu.id AND (tu.counter IS NULL OR tu.counter < $2)\n RETURNING q.id, q.workspace_id, ping, tu.counter\n ),\n update_ping AS (\n UPDATE v2_job_runtime r\n SET ping = null\n FROM zombie_jobs zj\n WHERE r.id = zj.id\n ),\n increment_counter AS (\n INSERT INTO zombie_job_counter (job_id, counter)\n SELECT id, 1 FROM to_update WHERE counter < $2\n ON CONFLICT (job_id) DO UPDATE\n SET counter = zombie_job_counter.counter + 1\n ),\n update_concurrency AS (\n UPDATE concurrency_counter cc\n SET job_uuids = job_uuids - zj.id::text\n FROM zombie_jobs zj\n INNER JOIN concurrency_key ck ON ck.job_id = zj.id\n WHERE cc.concurrency_id = ck.key\n )\n SELECT id AS \"id!\", workspace_id AS \"workspace_id!\", ping, counter + 1 AS counter FROM to_update",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "workspace_id!",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "ping",
"type_info": "Timestamptz"
},
{
"ordinal": 3,
"name": "counter",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Int4"
]
},
"nullable": [
false,
false,
true,
null
]
},
"hash": "12d37d75a429c0ddf2b2c190ab28bea5aefd27d0ed8a1bb2c8b3c1b0ece4efb7"
}
backend/.sqlx/query-145b364bcd45b6a8b3b80fd67a5ae17212785bd7206fbd3901a3b516eb77dc55.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS(\n SELECT 1 FROM script\n WHERE workspace_id = $1 AND path = $2 AND archived = false AND deleted = false\n )",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "145b364bcd45b6a8b3b80fd67a5ae17212785bd7206fbd3901a3b516eb77dc55"
}
backend/.sqlx/query-16e4b1bead9fc77fd98658b8cb8cc6d6bf1df758b30e99bd661da866062ef14f.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT hash FROM script WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "hash",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "16e4b1bead9fc77fd98658b8cb8cc6d6bf1df758b30e99bd661da866062ef14f"
}
backend/.sqlx/query-1a54356c1e1353950bf6ab1d25ab21270131e9e93ca10d195664e7e5a774fe9e.json
Generated
+3 -1
View File
@@ -59,7 +59,9 @@
"failure",
"command",
"approval",
"preprocessor"
"preprocessor",
"schedule_handler_old",
"dynamic_skip"
]
}
}
backend/.sqlx/query-a0c20436e0506bf9e0e50bd3dcabdd35131cc374edac0a08c5492954f08c9d98.json → backend/.sqlx/query-23e4c6e3dc6a48f702c2b26a6b1f94668e086caaa0093a3b685f87483513b0d2.json
Generated
+10 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO schedule (\n workspace_id, path, schedule, timezone, edited_by, script_path,\n is_flow, args, enabled, email,\n on_failure, on_failure_times, on_failure_exact, on_failure_extra_args,\n on_recovery, on_recovery_times, on_recovery_extra_args,\n on_success, on_success_extra_args,\n ws_error_handler_muted, retry, summary, no_flow_overlap,\n tag, paused_until, cron_version, description\n ) VALUES (\n $1, $2, $3, $4, $5, $6,\n $7, $8, $9, $10,\n $11, $12, $13, $14,\n $15, $16, $17,\n $18, $19,\n $20, $21, $22, $23,\n $24, $25, $26, $27\n )\n RETURNING\n workspace_id,\n path,\n edited_by,\n edited_at,\n schedule,\n timezone,\n enabled,\n script_path,\n is_flow,\n args AS \"args: _\",\n extra_perms,\n email,\n error,\n on_failure,\n on_failure_times,\n on_failure_exact,\n on_failure_extra_args AS \"on_failure_extra_args: _\",\n on_recovery,\n on_recovery_times,\n on_recovery_extra_args AS \"on_recovery_extra_args: _\",\n on_success,\n on_success_extra_args AS \"on_success_extra_args: _\",\n ws_error_handler_muted,\n retry,\n no_flow_overlap,\n summary,\n description,\n tag,\n paused_until,\n cron_version\n ",
"query": "\n INSERT INTO schedule (\n workspace_id, path, schedule, timezone, edited_by, script_path,\n is_flow, args, enabled, email,\n on_failure, on_failure_times, on_failure_exact, on_failure_extra_args,\n on_recovery, on_recovery_times, on_recovery_extra_args,\n on_success, on_success_extra_args,\n ws_error_handler_muted, retry, summary, no_flow_overlap,\n tag, paused_until, cron_version, description, dynamic_skip\n ) VALUES (\n $1, $2, $3, $4, $5, $6,\n $7, $8, $9, $10,\n $11, $12, $13, $14,\n $15, $16, $17,\n $18, $19,\n $20, $21, $22, $23,\n $24, $25, $26, $27, $28\n )\n RETURNING\n workspace_id,\n path,\n edited_by,\n edited_at,\n schedule,\n timezone,\n enabled,\n script_path,\n is_flow,\n args AS \"args: _\",\n extra_perms,\n email,\n error,\n on_failure,\n on_failure_times,\n on_failure_exact,\n on_failure_extra_args AS \"on_failure_extra_args: _\",\n on_recovery,\n on_recovery_times,\n on_recovery_extra_args AS \"on_recovery_extra_args: _\",\n on_success,\n on_success_extra_args AS \"on_success_extra_args: _\",\n ws_error_handler_muted,\n retry,\n no_flow_overlap,\n summary,\n description,\n tag,\n paused_until,\n cron_version,\n dynamic_skip\n ",
"describe": {
"columns": [
{
@@ -152,6 +152,11 @@
"ordinal": 29,
"name": "cron_version",
"type_info": "Text"
},
{
"ordinal": 30,
"name": "dynamic_skip",
"type_info": "Varchar"
}
],
"parameters": {
@@ -182,7 +187,8 @@
"Varchar",
"Timestamptz",
"Text",
"Text"
"Text",
"Varchar"
]
},
"nullable": [
@@ -215,8 +221,9 @@
true,
true,
true,
true,
true
]
},
"hash": "a0c20436e0506bf9e0e50bd3dcabdd35131cc374edac0a08c5492954f08c9d98"
"hash": "23e4c6e3dc6a48f702c2b26a6b1f94668e086caaa0093a3b685f87483513b0d2"
}
backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json
Generated
+1 -1
View File
@@ -65,7 +65,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json
Generated
+1 -1
View File
@@ -25,7 +25,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-348f73fa8222ec195bb5b2260d1595e6e9fa9d73779cf352e94039fd47de4294.json → backend/.sqlx/query-4144c87c25a939aafb2f57da189d94d038bcad7a36fbf87e0403c89a979c5b3f.json
Generated
+10 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE schedule SET\n schedule = $1,\n timezone = $2,\n args = $3,\n on_failure = $4,\n on_failure_times = $5,\n on_failure_exact = $6,\n on_failure_extra_args = $7,\n on_recovery = $8,\n on_recovery_times = $9,\n on_recovery_extra_args = $10,\n on_success = $11,\n on_success_extra_args = $12,\n ws_error_handler_muted = $13,\n retry = $14,\n summary = $15,\n no_flow_overlap = $16,\n tag = $17,\n paused_until = $18,\n path = $19,\n workspace_id = $20,\n cron_version = COALESCE($21, cron_version),\n description = $22\n WHERE path = $19 AND workspace_id = $20\n RETURNING\n workspace_id,\n path,\n edited_by,\n edited_at,\n schedule,\n timezone,\n enabled,\n script_path,\n is_flow,\n args AS \"args: _\",\n extra_perms,\n email,\n error,\n on_failure,\n on_failure_times,\n on_failure_exact,\n on_failure_extra_args AS \"on_failure_extra_args: _\",\n on_recovery,\n on_recovery_times,\n on_recovery_extra_args AS \"on_recovery_extra_args: _\",\n on_success,\n on_success_extra_args AS \"on_success_extra_args: _\",\n ws_error_handler_muted,\n retry,\n no_flow_overlap,\n summary,\n description,\n tag,\n paused_until,\n cron_version\n ",
"query": "\n UPDATE schedule SET\n schedule = $1,\n timezone = $2,\n args = $3,\n on_failure = $4,\n on_failure_times = $5,\n on_failure_exact = $6,\n on_failure_extra_args = $7,\n on_recovery = $8,\n on_recovery_times = $9,\n on_recovery_extra_args = $10,\n on_success = $11,\n on_success_extra_args = $12,\n ws_error_handler_muted = $13,\n retry = $14,\n summary = $15,\n no_flow_overlap = $16,\n tag = $17,\n paused_until = $18,\n path = $19,\n workspace_id = $20,\n cron_version = COALESCE($21, cron_version),\n description = $22,\n dynamic_skip = $23\n WHERE path = $19 AND workspace_id = $20\n RETURNING\n workspace_id,\n path,\n edited_by,\n edited_at,\n schedule,\n timezone,\n enabled,\n script_path,\n is_flow,\n args AS \"args: _\",\n extra_perms,\n email,\n error,\n on_failure,\n on_failure_times,\n on_failure_exact,\n on_failure_extra_args AS \"on_failure_extra_args: _\",\n on_recovery,\n on_recovery_times,\n on_recovery_extra_args AS \"on_recovery_extra_args: _\",\n on_success,\n on_success_extra_args AS \"on_success_extra_args: _\",\n ws_error_handler_muted,\n retry,\n no_flow_overlap,\n summary,\n description,\n tag,\n paused_until,\n cron_version,\n dynamic_skip\n ",
"describe": {
"columns": [
{
@@ -152,6 +152,11 @@
"ordinal": 29,
"name": "cron_version",
"type_info": "Text"
},
{
"ordinal": 30,
"name": "dynamic_skip",
"type_info": "Varchar"
}
],
"parameters": {
@@ -177,7 +182,8 @@
"Text",
"Text",
"Text",
"Text"
"Text",
"Varchar"
]
},
"nullable": [
@@ -210,8 +216,9 @@
true,
true,
true,
true,
true
]
},
"hash": "348f73fa8222ec195bb5b2260d1595e6e9fa9d73779cf352e94039fd47de4294"
"hash": "4144c87c25a939aafb2f57da189d94d038bcad7a36fbf87e0403c89a979c5b3f"
}
backend/.sqlx/query-488dd591096b2b47787afdc3a1d73917ed13269f2ee20b86df79fca2c8efe672.json
Generated
+1 -1
View File
@@ -37,7 +37,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json
Generated
+1 -1
View File
@@ -95,7 +95,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-7cd070aaca3b4f95bb6e669ac07f288bd1cc8a85625255d86e45f63e8267d34c.json → backend/.sqlx/query-7927b80ce75d99b2a30f6b29196af000578a3c166509f032d14452cc637d884f.json
Generated
+8 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE schedule SET\n enabled = $1,\n email = $2\n WHERE path = $3 AND workspace_id = $4\n RETURNING\n workspace_id,\n path,\n edited_by,\n edited_at,\n schedule,\n timezone,\n enabled,\n script_path,\n is_flow,\n args AS \"args: _\",\n extra_perms,\n email,\n error,\n on_failure,\n on_failure_times,\n on_failure_exact,\n on_failure_extra_args AS \"on_failure_extra_args: _\",\n on_recovery,\n on_recovery_times,\n on_recovery_extra_args AS \"on_recovery_extra_args: _\",\n on_success,\n on_success_extra_args AS \"on_success_extra_args: _\",\n ws_error_handler_muted,\n retry,\n no_flow_overlap,\n summary,\n description,\n tag,\n paused_until,\n cron_version\n ",
"query": "\n UPDATE schedule SET\n enabled = $1,\n email = $2\n WHERE path = $3 AND workspace_id = $4\n RETURNING\n workspace_id,\n path,\n edited_by,\n edited_at,\n schedule,\n timezone,\n enabled,\n script_path,\n is_flow,\n args AS \"args: _\",\n extra_perms,\n email,\n error,\n on_failure,\n on_failure_times,\n on_failure_exact,\n on_failure_extra_args AS \"on_failure_extra_args: _\",\n on_recovery,\n on_recovery_times,\n on_recovery_extra_args AS \"on_recovery_extra_args: _\",\n on_success,\n on_success_extra_args AS \"on_success_extra_args: _\",\n ws_error_handler_muted,\n retry,\n no_flow_overlap,\n summary,\n description,\n tag,\n paused_until,\n cron_version,\n dynamic_skip\n ",
"describe": {
"columns": [
{
@@ -152,6 +152,11 @@
"ordinal": 29,
"name": "cron_version",
"type_info": "Text"
},
{
"ordinal": 30,
"name": "dynamic_skip",
"type_info": "Varchar"
}
],
"parameters": {
@@ -192,8 +197,9 @@
true,
true,
true,
true,
true
]
},
"hash": "7cd070aaca3b4f95bb6e669ac07f288bd1cc8a85625255d86e45f63e8267d34c"
"hash": "7927b80ce75d99b2a30f6b29196af000578a3c166509f032d14452cc637d884f"
}
backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json
Generated
+1 -1
View File
@@ -35,7 +35,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-804fc11e35f4afc0db194b6fe2594f91df7e588d4d2431bc85f4d8734920c8bf.json
Generated
+51
View File
@@ -0,0 +1,51 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO script (summary, description, dedicated_worker, content, workspace_id, path, hash, language, tag, created_by, lock) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '') ON CONFLICT (workspace_id, hash) DO NOTHING",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Varchar",
"Varchar",
"Int8",
{
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb",
"nu",
"java",
"duckdb",
"ruby"
]
}
}
},
"Varchar",
"Varchar"
]
},
"nullable": []
},
"hash": "804fc11e35f4afc0db194b6fe2594f91df7e588d4d2431bc85f4d8734920c8bf"
}
backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json
Generated
+1 -1
View File
@@ -35,7 +35,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-85705fc3d7f8ba5f1b12d5fb222c38fc64deb1226aab9dc3bc4465324fce37d1.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"UuidArray",
"Varchar",
"Varchar"
]
},
"nullable": []
},
"hash": "85705fc3d7f8ba5f1b12d5fb222c38fc64deb1226aab9dc3bc4465324fce37d1"
}
backend/.sqlx/query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json
Generated
+1 -1
View File
@@ -36,7 +36,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json
Generated
+1 -1
View File
@@ -25,7 +25,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-a0b3e10e077d30c1da135dff9feca3761d400391f1f46a8294da3e6c9af63887.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job_status (id, flow_status) SELECT unnest($1::uuid[]), $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"UuidArray",
"Jsonb"
]
},
"nullable": []
},
"hash": "a0b3e10e077d30c1da135dff9feca3761d400391f1f46a8294da3e6c9af63887"
}
backend/.sqlx/query-a6608d47b96d851eb7b04d2e4b472889ff257db0f0e3e9252adda2e4ef2039d6.json
Generated
+1 -1
View File
@@ -95,7 +95,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-ab04cda71f8e2be9acbecabe1ee5ef756b8e5c1955fbe111df9ee171dc262338.json
Generated
+89
View File
@@ -0,0 +1,89 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_flow) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, 1)) RETURNING id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Int8",
"Varchar",
{
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
"aiagent"
]
}
}
},
{
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb",
"nu",
"java",
"duckdb",
"ruby"
]
}
}
},
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Jsonb"
]
},
"nullable": [
false
]
},
"hash": "ab04cda71f8e2be9acbecabe1ee5ef756b8e5c1955fbe111df9ee171dc262338"
}
backend/.sqlx/query-b3c02fd225a6aa78785d466e7f033b38deb9c7fa17bd3836c9ad8884f27be84a.json
Generated
+1 -1
View File
@@ -30,7 +30,7 @@
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json
Generated
+2 -2
View File
@@ -18,8 +18,8 @@
"Left": []
},
"nullable": [
false,
true
true,
false
]
},
"hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76"
backend/.sqlx/query-b4a9abcb38997587b28655b0f4a212a5bd4039b57fab20b163617e33a4c9dd46.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
"describe": {
"columns": [],
"parameters": {
"Left": [
"UuidArray"
]
},
"nullable": []
},
"hash": "b4a9abcb38997587b28655b0f4a212a5bd4039b57fab20b163617e33a4c9dd46"
}
backend/.sqlx/query-bc9a17567cc71f51a8fa0a6fe12c0aa3a52e2ba194d40c66993d02ae2c129327.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE schedule SET dynamic_skip = $1 WHERE dynamic_skip = $2 AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "bc9a17567cc71f51a8fa0a6fe12c0aa3a52e2ba194d40c66993d02ae2c129327"
}
backend/.sqlx/query-daf9674838fb3e3653a356c7434c719616a614d77e726433737e5f5d9bd60134.json
Generated
-41
View File
@@ -1,41 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH to_update AS (\n SELECT q.id, q.workspace_id, r.ping, COALESCE(zjc.counter, 0) as counter\n FROM v2_job_queue q\n JOIN v2_job j ON j.id = q.id\n JOIN v2_job_runtime r ON r.id = j.id\n LEFT JOIN zombie_job_counter zjc ON zjc.job_id = q.id\n WHERE ping < now() - ($1 || ' seconds')::interval\n AND running = true\n AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow')\n AND same_worker = false\n AND (zjc.counter IS NULL OR zjc.counter <= $2)\n FOR UPDATE of q SKIP LOCKED\n ),\n zombie_jobs AS (\n UPDATE v2_job_queue q\n SET running = false, started_at = null\n FROM to_update tu\n WHERE q.id = tu.id AND (tu.counter IS NULL OR tu.counter < $2)\n RETURNING q.id, q.workspace_id, ping, tu.counter\n ),\n update_ping AS (\n UPDATE v2_job_runtime r\n SET ping = null\n FROM zombie_jobs zj\n WHERE r.id = zj.id\n ),\n increment_counter AS (\n INSERT INTO zombie_job_counter (job_id, counter)\n SELECT id, 1 FROM to_update WHERE counter < $2\n ON CONFLICT (job_id) DO UPDATE\n SET counter = zombie_job_counter.counter + 1\n ),\n update_concurrency AS (\n UPDATE concurrency_counter cc\n SET job_uuids = job_uuids - zj.id::text\n FROM zombie_jobs zj\n INNER JOIN concurrency_key ck ON ck.job_id = zj.id\n WHERE cc.concurrency_id = ck.key\n )\n SELECT id AS \"id!\", workspace_id AS \"workspace_id!\", ping, counter + 1 AS counter FROM to_update",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "workspace_id!",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "ping",
"type_info": "Timestamptz"
},
{
"ordinal": 3,
"name": "counter",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Int4"
]
},
"nullable": [
false,
false,
true,
null
]
},
"hash": "daf9674838fb3e3653a356c7434c719616a614d77e726433737e5f5d9bd60134"
}
backend/.sqlx/query-e2f4eeb896c22e3215370ed30ed5cc890503e4197b0fdb3a906504eba166aa97.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS(SELECT 1 FROM flow_conversation_message WHERE job_id = $1 AND message_type = 'assistant')",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "e2f4eeb896c22e3215370ed30ed5cc890503e4197b0fdb3a906504eba166aa97"
}
backend/.sqlx/query-ff0403790674cdb07022af71c2377afbd8b3a660b3be27514b517c077c63c238.json
Generated
+89
View File
@@ -0,0 +1,89 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Int8",
"Varchar",
{
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
"aiagent"
]
}
}
},
{
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb",
"nu",
"java",
"duckdb",
"ruby"
]
}
}
},
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Int4"
]
},
"nullable": [
false
]
},
"hash": "ff0403790674cdb07022af71c2377afbd8b3a660b3be27514b517c077c63c238"
}
backend/migrations/20250930145445_add_dynamic_skip.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Remove dynamic_skip column from schedule table
ALTER TABLE schedule DROP COLUMN dynamic_skip;
backend/migrations/20250930145445_add_dynamic_skip.up.sql
+4
View File
@@ -0,0 +1,4 @@
-- Add dynamic_skip column to schedule table
-- This column stores the path to a script that validates scheduled datetimes
-- The handler receives the scheduled_for datetime and returns a boolean
ALTER TABLE schedule ADD COLUMN dynamic_skip VARCHAR(1000) DEFAULT NULL;
backend/migrations/20251002151319_rename_singlescriptflow_to_singlestepflow.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Revert singlestepflow back to singlescriptflow in job_kind enum
ALTER TYPE job_kind RENAME VALUE 'singlestepflow' TO 'singlescriptflow';
backend/migrations/20251002151319_rename_singlescriptflow_to_singlestepflow.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Rename singlescriptflow to singlestepflow in job_kind enum
ALTER TYPE job_kind RENAME VALUE 'singlescriptflow' TO 'singlestepflow';
backend/src/monitor.rs
+2 -2
View File
@@ -2035,7 +2035,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
LEFT JOIN zombie_job_counter zjc ON zjc.job_id = q.id
WHERE ping < now() - ($1 || ' seconds')::interval
AND running = true
AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow')
AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow')
AND same_worker = false
AND (zjc.counter IS NULL OR zjc.counter <= $2)
FOR UPDATE of q SKIP LOCKED
@@ -2211,7 +2211,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
vec![]
} else {
sqlx::query_as::<_, QueuedJob>("SELECT *, null as workflow_as_code_status FROM v2_as_queue WHERE last_ping < now() - ($1 || ' seconds')::interval
AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false")
AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow') AND same_worker = false")
.bind(ZOMBIE_JOB_TIMEOUT.as_str())
.fetch_all(db)
.await
backend/tests/worker.rs
+8
View File
@@ -208,6 +208,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
FlowModule {
id: "b".to_string(),
@@ -252,6 +253,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
}],
modules_node: None,
}
@@ -270,6 +272,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
],
same_worker: false,
@@ -387,6 +390,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
FlowModule {
id: "b".to_string(),
@@ -441,6 +445,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
FlowModule {
id: "e".to_string(),
@@ -482,6 +487,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
],
modules_node: None,
@@ -500,6 +506,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
FlowModule {
id: "c".to_string(),
@@ -547,6 +554,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
],
same_worker: true,
backend/windmill-api/openapi.yaml
+11 -2
View File
@@ -15299,7 +15299,7 @@ components:
"script_hub",
"identity",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
@@ -15409,7 +15409,7 @@ components:
"script_hub",
"identity",
"deploymentcallback",
"singlescriptflow",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
@@ -16278,6 +16278,9 @@ components:
format: date-time
cron_version:
type: string
dynamic_skip:
type: string
description: Path to a script that validates scheduled datetimes. Receives scheduled_for datetime and returns boolean.
required:
- path
- edited_by
@@ -16396,6 +16399,9 @@ components:
cron_version:
type: string
description: The version of the cron schedule to use (last is v2)
dynamic_skip:
type: string
description: Path to a script that validates scheduled datetimes. Receives scheduled_for datetime and returns boolean.
required:
- path
- schedule
@@ -16469,6 +16475,9 @@ components:
cron_version:
type: string
description: The version of the cron schedule to use (last is v2)
dynamic_skip:
type: string
description: Path to a script that validates scheduled datetimes. Receives scheduled_for datetime and returns boolean.
required:
- schedule
- timezone
backend/windmill-api/src/flows.rs
+6
View File
@@ -1478,6 +1478,7 @@ mod tests {
hash: None,
tag_override: None,
is_trigger: None,
pass_flow_input_directly: None,
}),
stop_after_if: None,
stop_after_all_iters_if: None,
@@ -1493,6 +1494,7 @@ mod tests {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
FlowModule {
id: "b".to_string(),
@@ -1526,6 +1528,7 @@ mod tests {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
FlowModule {
id: "c".to_string(),
@@ -1556,6 +1559,7 @@ mod tests {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
},
],
failure_module: Some(Box::new(FlowModule {
@@ -1566,6 +1570,7 @@ mod tests {
hash: None,
tag_override: None,
is_trigger: None,
pass_flow_input_directly: None,
}
.into(),
stop_after_if: Some(StopAfterIf {
@@ -1585,6 +1590,7 @@ mod tests {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
})),
preprocessor_module: None,
same_worker: false,
backend/windmill-api/src/jobs.rs
+6 -2
View File
@@ -7558,8 +7558,12 @@ pub fn filter_list_completed_query(
sqlb.and_where_le("started_at", "?".bind(&dt.to_rfc3339()));
}
if let Some(dt) = &lq.created_or_started_after {
sqlb.and_where_ge("created_at", "?".bind(&dt.to_rfc3339()));
sqlb.and_where_ge("started_at", "?".bind(&dt.to_rfc3339()));
let ts = dt.to_rfc3339();
sqlb.and_where(format!(
"(created_at >= '{}' OR started_at >= '{}')",
ts.replace("'", "''"),
ts.replace("'", "''")
));
}
if let Some(dt) = &lq.created_before {
backend/windmill-api/src/schedule.rs
+55 -8
View File
@@ -78,6 +78,7 @@ pub struct NewSchedule {
pub tag: Option<String>,
pub paused_until: Option<DateTime<Utc>>,
pub cron_version: Option<String>,
pub dynamic_skip: Option<String>,
}
#[derive(Serialize, Deserialize)]
@@ -128,6 +129,35 @@ fn to_json_raw_opt(
value.map(|v| sqlx::types::Json(to_raw_value(&v)))
}
/// Validate that a dynamic skip handler (script or flow) exists
async fn validate_dynamic_skip<'c>(
tx: &mut Transaction<'c, Postgres>,
w_id: &str,
handler_path: &str,
) -> Result<()> {
// Check for script only (flows are not supported in the UI)
let exists = sqlx::query_scalar!(
"SELECT EXISTS(
SELECT 1 FROM script
WHERE workspace_id = $1 AND path = $2 AND archived = false AND deleted = false
)",
w_id,
handler_path
)
.fetch_one(&mut **tx)
.await?
.unwrap_or(false);
if exists {
Ok(())
} else {
Err(Error::BadRequest(format!(
"Dynamic skip handler '{}' not found. The handler must be an existing, non-archived script at schedule creation time.",
handler_path
)))
}
}
async fn create_schedule(
authed: ApiAuthed,
Extension(db): Extension<DB>,
@@ -169,6 +199,11 @@ async fn create_schedule(
check_path_conflict(&mut tx, &w_id, &ns.path).await?;
check_flow_conflict(&mut tx, &w_id, &ns.path, ns.is_flow, &ns.script_path).await?;
// Validate dynamic_skip if provided
if let Some(handler_path) = &ns.dynamic_skip {
validate_dynamic_skip(&mut tx, &w_id, handler_path).await?;
}
let schedule = sqlx::query_as!(
Schedule,
r#"
@@ -179,7 +214,7 @@ async fn create_schedule(
on_recovery, on_recovery_times, on_recovery_extra_args,
on_success, on_success_extra_args,
ws_error_handler_muted, retry, summary, no_flow_overlap,
tag, paused_until, cron_version, description
tag, paused_until, cron_version, description, dynamic_skip
) VALUES (
$1, $2, $3, $4, $5, $6,
$7, $8, $9, $10,
@@ -187,7 +222,7 @@ async fn create_schedule(
$15, $16, $17,
$18, $19,
$20, $21, $22, $23,
$24, $25, $26, $27
$24, $25, $26, $27, $28
)
RETURNING
workspace_id,
@@ -219,7 +254,8 @@ async fn create_schedule(
description,
tag,
paused_until,
cron_version
cron_version,
dynamic_skip
"#,
w_id,
ns.path,
@@ -251,7 +287,8 @@ async fn create_schedule(
ns.tag,
ns.paused_until,
ns.cron_version.clone().unwrap_or_else(|| "v2".to_string()),
ns.description
ns.description,
ns.dynamic_skip
)
.fetch_one(&mut *tx)
.await
@@ -311,6 +348,11 @@ async fn edit_schedule(
// Check schedule for error
ScheduleType::from_str(&es.schedule, es.cron_version.as_deref(), true)?;
// Validate dynamic_skip if provided
if let Some(handler_path) = &es.dynamic_skip {
validate_dynamic_skip(&mut tx, &w_id, handler_path).await?;
}
clear_schedule(&mut tx, path, &w_id).await?;
let schedule = sqlx::query_as!(
Schedule,
@@ -337,7 +379,8 @@ async fn edit_schedule(
path = $19,
workspace_id = $20,
cron_version = COALESCE($21, cron_version),
description = $22
description = $22,
dynamic_skip = $23
WHERE path = $19 AND workspace_id = $20
RETURNING
workspace_id,
@@ -369,7 +412,8 @@ async fn edit_schedule(
description,
tag,
paused_until,
cron_version
cron_version,
dynamic_skip
"#,
es.schedule,
es.timezone,
@@ -396,7 +440,8 @@ async fn edit_schedule(
path,
w_id,
es.cron_version,
es.description
es.description,
es.dynamic_skip
)
.fetch_one(&mut *tx)
.await
@@ -657,7 +702,8 @@ pub async fn set_enabled(
description,
tag,
paused_until,
cron_version
cron_version,
dynamic_skip
"#,
payload.enabled,
authed.email,
@@ -1006,6 +1052,7 @@ pub struct EditSchedule {
pub tag: Option<String>,
pub paused_until: Option<DateTime<Utc>>,
pub cron_version: Option<String>,
pub dynamic_skip: Option<String>,
}
pub async fn clear_schedule<'c>(
backend/windmill-api/src/scripts.rs
+10
View File
@@ -865,6 +865,16 @@ async fn create_script_internal<'c>(
schedulables.push(schedule);
}
// Update dynamic_skip references when script is renamed
sqlx::query!(
"UPDATE schedule SET dynamic_skip = $1 WHERE dynamic_skip = $2 AND workspace_id = $3",
&ns.path,
&p_path,
&w_id
)
.execute(&mut *tx)
.await?;
for schedule in schedulables {
clear_schedule(&mut tx, &schedule.path, &w_id).await?;
backend/windmill-api/src/triggers/trigger_helpers.rs
+4 -2
View File
@@ -808,13 +808,15 @@ async fn trigger_script_with_retry_and_error_handler(
priority,
apply_preprocessor,
..
} => JobPayload::SingleScriptFlow {
} => JobPayload::SingleStepFlow {
path,
hash,
hash: Some(hash),
flow_version: None,
args: HashMap::from(&push_args),
retry,
error_handler_path,
error_handler_args,
skip_handler: None,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
backend/windmill-common/src/cache.rs
+1 -1
View File
@@ -837,7 +837,7 @@ pub mod job {
match (kind, hash.map(|ScriptHash(id)| id)) {
(FlowDependencies, Some(id)) => flow::fetch_version(db, id).await,
(FlowNode, Some(id)) => flow::fetch_flow(db, FlowNodeId(id)).await,
(Flow, Some(id)) => match flow::fetch_version_lite(db, id).await {
(Flow, Some(id)) | (SingleStepFlow, Some(id)) => match flow::fetch_version_lite(db, id).await {
Ok(raw_flow) => Ok(raw_flow),
Err(_) => flow::fetch_version(db, id).await,
},
backend/windmill-common/src/flows.rs
+10
View File
@@ -393,6 +393,8 @@ pub struct FlowModule {
pub skip_if: Option<SkipIf>,
#[serde(skip_serializing_if = "Option::is_none")]
pub apply_preprocessor: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub pass_flow_input_directly: Option<bool>,
}
#[derive(Deserialize, Serialize, Debug, Clone)]
@@ -617,6 +619,8 @@ pub enum FlowModuleValue {
tag_override: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
is_trigger: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pass_flow_input_directly: Option<bool>,
},
/// Reference to another flow on the workspace
@@ -625,6 +629,8 @@ pub enum FlowModuleValue {
#[serde(alias = "input_transform")]
input_transforms: HashMap<String, InputTransform>,
path: String,
#[serde(skip_serializing_if = "Option::is_none")]
pass_flow_input_directly: Option<bool>,
},
/// For loop node
@@ -757,6 +763,7 @@ struct UntaggedFlowModuleValue {
modules_node: Option<FlowNodeId>,
assets: Option<Vec<AssetWithAltAccessType>>,
tools: Option<Vec<FlowModule>>,
pass_flow_input_directly: Option<bool>,
}
impl<'de> Deserialize<'de> for FlowModuleValue {
@@ -775,12 +782,14 @@ impl<'de> Deserialize<'de> for FlowModuleValue {
hash: untagged.hash,
tag_override: untagged.tag_override,
is_trigger: untagged.is_trigger,
pass_flow_input_directly: untagged.pass_flow_input_directly,
}),
"flow" => Ok(FlowModuleValue::Flow {
input_transforms: untagged.input_transforms.unwrap_or_default(),
path: untagged
.path
.ok_or_else(|| serde::de::Error::missing_field("path"))?,
pass_flow_input_directly: untagged.pass_flow_input_directly,
}),
"forloopflow" => Ok(FlowModuleValue::ForloopFlow {
iterator: untagged
@@ -922,6 +931,7 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
continue_on_error: None,
skip_if: None,
apply_preprocessor: None,
pass_flow_input_directly: None,
});
}
}
backend/windmill-common/src/jobs.rs
+14 -4
View File
@@ -84,7 +84,7 @@ pub enum JobKind {
Dependencies,
Flow,
FlowPreview,
SingleScriptFlow,
SingleStepFlow,
Identity,
FlowDependencies,
AppDependencies,
@@ -100,7 +100,7 @@ impl JobKind {
pub fn is_flow(&self) -> bool {
matches!(
self,
JobKind::Flow | JobKind::FlowPreview | JobKind::SingleScriptFlow | JobKind::FlowNode
JobKind::Flow | JobKind::FlowPreview | JobKind::SingleStepFlow | JobKind::FlowNode
)
}
@@ -400,13 +400,15 @@ pub enum JobPayload {
path: Option<String>,
restarted_from: Option<RestartedFrom>,
},
SingleScriptFlow {
SingleStepFlow {
path: String,
hash: ScriptHash,
hash: Option<ScriptHash>,
flow_version: Option<i64>,
args: HashMap<String, Box<serde_json::value::RawValue>>,
retry: Option<Retry>,
error_handler_path: Option<String>,
error_handler_args: Option<HashMap<String, Box<RawValue>>>,
skip_handler: Option<SkipHandler>,
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
@@ -426,6 +428,14 @@ pub enum JobPayload {
},
}
#[derive(Clone, Serialize, Deserialize, Debug)]
pub struct SkipHandler {
pub path: String,
pub args: HashMap<String, Box<RawValue>>,
pub stop_condition: String,
pub stop_message: String,
}
#[derive(Clone, Serialize, Deserialize, Debug, Default)]
pub struct RawCode {
pub content: String,
backend/windmill-common/src/schedule.rs
+2
View File
@@ -60,6 +60,8 @@ pub struct Schedule {
pub paused_until: Option<DateTime<chrono::Utc>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cron_version: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub dynamic_skip: Option<String>,
}
impl Schedule {
backend/windmill-queue/src/jobs.rs
+78 -21
View File
@@ -49,6 +49,7 @@ use windmill_common::{
},
flows::{
add_virtual_items_if_necessary, FlowModule, FlowModuleValue, FlowValue, InputTransform,
StopAfterIf,
},
jobs::{get_payload_tag_from_prefixed_path, JobKind, JobPayload, QueuedJob, RawCode},
schedule::Schedule,
@@ -3672,12 +3673,14 @@ pub async fn push<'c, 'd>(
priority,
)
}
JobPayload::SingleScriptFlow {
JobPayload::SingleStepFlow {
path,
hash,
flow_version,
retry,
error_handler_path,
error_handler_args,
skip_handler,
args,
custom_concurrency_key,
concurrent_limit,
@@ -3688,10 +3691,75 @@ pub async fn push<'c, 'd>(
trigger_path,
apply_preprocessor,
} => {
let mut input_transforms = HashMap::<String, InputTransform>::new();
for (arg_name, arg_value) in args {
input_transforms.insert(arg_name, InputTransform::Static { value: arg_value });
// Determine if this is a flow or a script
let is_flow = flow_version.is_some();
// Build modules list
let mut modules = vec![];
// Add skip validation module if provided
if let Some(skip_handler) = skip_handler {
let mut skip_input_transforms = HashMap::<String, InputTransform>::new();
for (arg_name, arg_value) in skip_handler.args {
skip_input_transforms.insert(arg_name, InputTransform::Static { value: arg_value });
}
modules.push(FlowModule {
id: "skip_validation".to_string(),
value: to_raw_value(&FlowModuleValue::Script {
input_transforms: skip_input_transforms,
path: skip_handler.path,
hash: None,
tag_override: None,
is_trigger: None,
pass_flow_input_directly: None,
}),
stop_after_if: Some(StopAfterIf {
expr: skip_handler.stop_condition,
skip_if_stopped: true,
error_message: Some(skip_handler.stop_message),
}),
..Default::default()
});
}
// Add main module (script or flow)
let mut main_input_transforms = HashMap::<String, InputTransform>::new();
for (arg_name, arg_value) in args {
main_input_transforms.insert(arg_name, InputTransform::Static { value: arg_value });
}
let main_module = if is_flow {
FlowModule {
id: "a".to_string(),
value: to_raw_value(&FlowModuleValue::Flow {
path: path.clone(),
input_transforms: main_input_transforms,
pass_flow_input_directly: None,
}),
retry,
pass_flow_input_directly: Some(true),
..Default::default()
}
} else {
FlowModule {
id: "a".to_string(),
value: to_raw_value(&FlowModuleValue::Script {
input_transforms: main_input_transforms,
path: path.clone(),
hash,
tag_override,
is_trigger: None,
pass_flow_input_directly: None,
}),
retry,
apply_preprocessor: Some(apply_preprocessor),
..Default::default()
}
};
modules.push(main_module);
// Build failure module if error handler is provided
let failure_module = if let Some(error_handler_path) = error_handler_path {
let mut input_transforms = HashMap::<String, InputTransform>::new();
input_transforms.insert(
@@ -3704,7 +3772,7 @@ pub async fn push<'c, 'd>(
);
input_transforms.insert(
"is_flow".to_string(),
InputTransform::Static { value: to_raw_value(&false) },
InputTransform::Static { value: to_raw_value(&is_flow) },
);
input_transforms.insert(
"trigger_path".to_string(),
@@ -3743,6 +3811,7 @@ pub async fn push<'c, 'd>(
hash: None,
tag_override: None,
is_trigger: None,
pass_flow_input_directly: None,
}),
..Default::default()
}))
@@ -3751,19 +3820,7 @@ pub async fn push<'c, 'd>(
};
let flow_value = FlowValue {
modules: vec![FlowModule {
id: "a".to_string(),
value: to_raw_value(&FlowModuleValue::Script {
input_transforms,
path: path.clone(),
hash: Some(hash),
tag_override,
is_trigger: None,
}),
retry,
apply_preprocessor: Some(apply_preprocessor),
..Default::default()
}],
modules,
failure_module,
concurrency_time_window_s,
concurrent_limit,
@@ -3779,10 +3836,10 @@ pub async fn push<'c, 'd>(
// this is a new flow being pushed, flow_status is set to flow_value:
let flow_status: FlowStatus = FlowStatus::new(&flow_value);
(
None,
None, // No version needed - flow is stored in raw_flow like FlowPreview
Some(path),
None,
JobKind::Flow,
JobKind::SingleStepFlow,
Some(flow_value),
Some(flow_status),
None,
@@ -4344,7 +4401,7 @@ pub async fn push<'c, 'd>(
}
JobKind::Flow => "jobs.run.flow",
JobKind::FlowPreview => "jobs.run.flow_preview",
JobKind::SingleScriptFlow => "jobs.run.single_script_flow",
JobKind::SingleStepFlow => "jobs.run.single_step_flow",
JobKind::Script_Hub => "jobs.run.script_hub",
JobKind::Dependencies => "jobs.run.dependencies",
JobKind::Identity => "jobs.run.identity",
backend/windmill-queue/src/schedule.rs
+130 -3
View File
@@ -22,6 +22,8 @@ use windmill_common::get_latest_flow_version_info_for_path_from_version;
use windmill_common::jobs::check_tag_available_for_workspace_internal;
use windmill_common::jobs::JobPayload;
use windmill_common::schedule::schedule_to_user;
use windmill_common::scripts::ScriptHash;
use windmill_common::worker::to_raw_value;
use windmill_common::utils::WarnAfterExt;
use windmill_common::FlowVersionInfo;
use windmill_common::DB;
@@ -32,6 +34,74 @@ use windmill_common::{
utils::{now_from_db, ScheduleType, StripPath},
};
/// Helper to fetch metadata for a schedule's script or flow
async fn get_schedule_metadata<'c>(
tx: &mut sqlx::Transaction<'c, sqlx::Postgres>,
schedule: &Schedule,
) -> Result<(
Option<String>, // tag
Option<i32>, // timeout
Option<String>, // on_behalf_of_email
String, // created_by
Option<ScriptHash>, // hash (for scripts)
Option<i64>, // flow_version (for flows)
Option<Retry>, // retry
)> {
let parsed_retry = schedule
.retry
.clone()
.and_then(|r| serde_json::from_value::<Retry>(r).ok());
if schedule.is_flow {
let version = get_latest_flow_version_id_for_path(
None,
&mut **tx,
&schedule.workspace_id,
&schedule.script_path,
false,
)
.await?;
let FlowVersionInfo {
tag,
on_behalf_of_email,
edited_by,
..
} = get_latest_flow_version_info_for_path_from_version(
&mut **tx,
version,
&schedule.workspace_id,
&schedule.script_path,
)
.await?;
Ok((tag, None, on_behalf_of_email, edited_by, None, Some(version), parsed_retry))
} else {
let (
hash,
tag,
_custom_concurrency_key,
_concurrent_limit,
_concurrency_time_window_s,
_cache_ttl,
_language,
_dedicated_worker,
_priority,
timeout,
on_behalf_of_email,
created_by,
) = windmill_common::get_latest_hash_for_path(
&mut **tx,
&schedule.workspace_id,
&schedule.script_path,
false,
)
.await?;
Ok((tag, timeout, on_behalf_of_email, created_by, Some(hash), None, parsed_retry))
}
}
pub async fn push_scheduled_job<'c>(
db: &DB,
mut tx: Transaction<'c, Postgres>,
@@ -136,7 +206,61 @@ pub async fn push_scheduled_job<'c>(
}
}
let (payload, tag, timeout, on_behalf_of_email, created_by) = if schedule.is_flow {
// If schedule handler is defined, wrap the scheduled job in a synthetic flow
// with the handler as the first step (with stop_after_if to skip if handler returns false)
let (payload, tag, timeout, on_behalf_of_email, created_by) = if let Some(handler_path) = &schedule.dynamic_skip {
// Build skip handler args
let mut skip_handler_args = HashMap::<String, Box<serde_json::value::RawValue>>::new();
skip_handler_args.insert(
"scheduled_for".to_string(),
to_raw_value(&next.to_rfc3339()),
);
let stop_condition = "result !== true".to_string();
let stop_message = format!(
"Schedule handler {} did not return true for datetime {}. Handler must return boolean true to execute scheduled job.",
handler_path,
next.to_rfc3339()
);
// Get metadata from the scheduled script/flow for tag, timeout, etc.
let (tag, timeout, on_behalf_of_email, created_by, hash, flow_version, retry) =
get_schedule_metadata(&mut tx, schedule).await?;
(
JobPayload::SingleStepFlow {
path: schedule.script_path.clone(),
hash,
flow_version,
args: args.clone(),
retry,
error_handler_path: None,
error_handler_args: None,
skip_handler: Some(windmill_common::jobs::SkipHandler {
path: handler_path.clone(),
args: skip_handler_args,
stop_condition,
stop_message,
}),
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
priority: None,
tag_override: schedule.tag.clone(),
trigger_path: None,
apply_preprocessor: false,
},
if schedule.tag.as_ref().is_some_and(|x| x != "") {
schedule.tag.clone()
} else {
tag
},
timeout,
on_behalf_of_email,
created_by,
)
} else if schedule.is_flow {
let version = get_latest_flow_version_id_for_path(
None,
&mut *tx,
@@ -160,6 +284,7 @@ pub async fn push_scheduled_job<'c>(
"get_latest_flow_version_info_for_path_from_version".to_string(),
)
.await?;
(
JobPayload::Flow {
path: schedule.script_path.clone(),
@@ -209,12 +334,14 @@ pub async fn push_scheduled_job<'c>(
}
// if retry is set, we wrap the script into a one step flow with a retry on the module
(
JobPayload::SingleScriptFlow {
JobPayload::SingleStepFlow {
path: schedule.script_path.clone(),
hash: hash,
hash: Some(hash),
flow_version: None,
retry: Some(parsed_retry),
error_handler_path: None,
error_handler_args: None,
skip_handler: None,
args: static_args,
custom_concurrency_key: None,
concurrent_limit: None,
backend/windmill-worker/src/ai_executor.rs
+2
View File
@@ -201,6 +201,7 @@ pub async fn handle_ai_agent_job(
tag_override,
input_transforms,
is_trigger,
pass_flow_input_directly,
}) => match hash {
Some(hash) => {
let (_, metadata) = cache::script::fetch(conn, hash.clone()).await?;
@@ -232,6 +233,7 @@ pub async fn handle_ai_agent_job(
tag_override: tag_override.clone(),
input_transforms: input_transforms.clone(),
is_trigger: *is_trigger,
pass_flow_input_directly: *pass_flow_input_directly,
});
let (_, metadata) = cache::script::fetch(conn, hash).await?;
Ok(metadata
backend/windmill-worker/src/worker.rs
+3 -2
View File
@@ -2393,7 +2393,8 @@ pub async fn handle_queued_job(
| JobKind::Dependencies
| JobKind::FlowPreview
| JobKind::Flow
| JobKind::FlowDependencies,
| JobKind::FlowDependencies
| JobKind::SingleStepFlow,
x,
) => {
if x.map(|x| x.0).is_none_or(|x| is_special_codebase_hash(x)) {
@@ -2813,7 +2814,7 @@ async fn try_validate_schema(
JobKind::Script_Hub => 3,
JobKind::Preview => 4,
JobKind::DeploymentCallback => 5,
JobKind::SingleScriptFlow => 6,
JobKind::SingleStepFlow => 6,
JobKind::Dependencies => 7,
JobKind::Flow => 8,
JobKind::FlowPreview => 9,
backend/windmill-worker/src/worker_flow.rs
+11 -7
View File
@@ -734,7 +734,7 @@ pub async fn update_flow_status_after_job_completion_internal(
}
let new_status = if
!(stop_early && stop_early_err_msg.is_some()) // if stop_early with error message, we want to set the job as failure and trigger the error handler if it exists
!(stop_early && stop_early_err_msg.is_some() && !skip_if_stop_early) // if stop_early with error and NOT skip_if_stopped, mark as failure
&& (
skip_loop_failures
|| sqlx::query_scalar!(
@@ -762,7 +762,7 @@ pub async fn update_flow_status_after_job_completion_internal(
branch_chosen: None,
approvers: vec![],
failed_retries: vec![],
skipped: false,
skipped: stop_early && skip_if_stop_early,
agent_actions: None,
agent_actions_success: None,
}
@@ -956,10 +956,11 @@ pub async fn update_flow_status_after_job_completion_internal(
flow_jobs_duration.set(position, &flow_job_duration);
}
// if stop_early with error message, we want to set the job as failure and trigger the error handler if it exists
// if stop_early with error message and NOT skip_if_stopped, mark as failure
// if skip_if_stopped=true, we want to mark as success (skipped), not failure
if (success
|| (flow_jobs.is_some() && (skip_loop_failures || skip_seq_branch_failure)))
&& !(stop_early && stop_early_err_msg.is_some())
&& !(stop_early && stop_early_err_msg.is_some() && !skip_if_stop_early)
{
let is_skipped = if current_module.as_ref().is_some_and(|m| m.skip_if.is_some())
{
@@ -974,7 +975,7 @@ pub async fn update_flow_status_after_job_completion_internal(
})?
.unwrap_or(false)
} else {
false
stop_early && skip_if_stop_early // Mark as skipped when stop_after_if with skip_if_stopped=true
};
success = true;
(
@@ -2746,6 +2747,9 @@ async fn push_next_flow_job(
to_raw_value(&"preprocessor"),
);
Ok(Marc::new(hm))
} else if module.pass_flow_input_directly.unwrap_or(false) {
// If pass_flow_input_directly is set, use flow args directly
Ok(arc_flow_job_args.clone())
} else {
let value = module.get_value();
match &value {
@@ -4528,9 +4532,9 @@ pub async fn script_to_payload(
};
(
// We only apply the preprocessor if it's explicitly set to true in the module,
// which can only happen if the the flow is a SingleScriptFlow triggered by a trigger with retries or error handling.
// which can only happen if the the flow is a SingleStepFlow triggered by a trigger with retries or error handling.
// In that case, apply_preprocessor is still only set to true if the script has a preprocesor.
// We only check for script hash because SingleScriptFlow triggers specifies the script hash
// We only check for script hash because SingleStepFlow triggers specifies the script hash
JobPayload::ScriptHash {
hash,
path: script_path,
frontend/src/lib/components/flows/scheduleUtils.ts
+2 -1
View File
@@ -169,7 +169,8 @@ export async function saveScheduleFromCfg(
no_flow_overlap: scheduleCfg.no_flow_overlap,
tag: scheduleCfg.tag,
paused_until: scheduleCfg.paused_until,
cron_version: scheduleCfg.cron_version
cron_version: scheduleCfg.cron_version,
dynamic_skip: scheduleCfg.dynamic_skip
}
try {
if (edit) {
frontend/src/lib/components/runs/JobsLoader.svelte
+3 -3
View File
@@ -118,7 +118,7 @@
let kinds: CompletedJob['job_kind'][] = ['deploymentcallback']
return kinds.join(',')
} else if (jobKindsCat == 'runs') {
let kinds: CompletedJob['job_kind'][] = ['script', 'flow', 'singlescriptflow']
let kinds: CompletedJob['job_kind'][] = ['script', 'flow', 'singlestepflow']
return kinds.join(',')
} else {
let kinds: CompletedJob['job_kind'][] = [
@@ -176,7 +176,7 @@
: success == 'waiting'
? false
: undefined,
isSkipped: isSkipped ? undefined : false,
isSkipped: isSkipped ? true : undefined,
// isFlowStep: jobKindsCat != 'all' ? false : undefined,
hasNullParent: jobKindsCat != 'all' ? true : undefined,
label: label === null || label === '' ? undefined : label,
@@ -231,7 +231,7 @@
jobKinds: jobKindsCat == 'all' || jobKinds == '' ? undefined : jobKinds,
success: success == 'success' ? true : success == 'failure' ? false : undefined,
running: success == 'running' ? true : undefined,
isSkipped: isSkipped ? undefined : false,
isSkipped: isSkipped ? true : undefined,
isFlowStep: jobKindsCat != 'all' ? false : undefined,
label: label === null || label === '' ? undefined : label,
tag: tag === null || tag === '' ? undefined : tag,
frontend/src/lib/components/runs/RunRow.svelte
+1 -1
View File
@@ -85,7 +85,7 @@
jobKind === 'script' ||
isScriptPreview(jobKind) ||
jobKind === 'script_hub' ||
jobKind === 'singlescriptflow'
jobKind === 'singlestepflow'
) {
return Code
} else if (jobKind === 'aiagent') {
frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte
+43 -2
View File
@@ -51,7 +51,7 @@
trigger = undefined
} = $props()
let optionTabSelected: 'error_handler' | 'recovery_handler' | 'success_handler' | 'retries' =
let optionTabSelected: 'error_handler' | 'recovery_handler' | 'success_handler' | 'retries' | 'dynamic_skip' =
$state('error_handler')
let initialPath = $state('')
let edit = $state(true)
@@ -81,6 +81,7 @@
let failedExact = $state(false)
let recoveredTimes = $state(1)
let retry: Retry | undefined = $state(undefined)
let dynamicSkipPath: string | undefined = $state(undefined)
let script_path = $state('')
let initialScriptPath = $state('')
let runnable: Script | Flow | undefined = $state()
@@ -494,6 +495,7 @@
successHandlerSelected = 'slack'
successHandlerExtraArgs = {}
}
dynamicSkipPath = cfg.dynamic_skip
args = cfg.args ?? {}
extraPerms = cfg.extra_perms ?? {}
can_write = canWrite(cfg.path, cfg.extra_perms, $userStore)
@@ -595,7 +597,8 @@
tag: tag,
paused_until: paused_until,
cron_version: cronVersion,
extra_perms: extraPerms
extra_perms: extraPerms,
dynamic_skip: dynamicSkipPath
}
}
@@ -901,6 +904,7 @@
<Tab value="recovery_handler">Recovery Handler</Tab>
<Tab value="success_handler">Success Handler</Tab>
<Tab value="retries">Retries</Tab>
<Tab value="dynamic_skip">Dynamic skip</Tab>
{#if itemKind === 'script'}
<Tab value="tag">Custom tag</Tab>
{/if}
@@ -1186,6 +1190,43 @@
/>
{/if}
</Section>
{:else if optionTabSelected === 'dynamic_skip'}
<Section label="Dynamic skip">
{#snippet header()}
<Tooltip>
Optional script to filter scheduled dates. Receives the proposed datetime and returns
boolean. True = run on this date, False = skip to next occurrence.
</Tooltip>
{/snippet}
<div class="flex flex-col gap-2">
<Label label="Dynamic skip script">
<div class="flex flex-row">
<ScriptPicker
disabled={!can_write}
bind:scriptPath={dynamicSkipPath}
kinds={['script']}
allowRefresh={can_write}
clearable
/>
{#if !dynamicSkipPath}
<Button
btnClasses="ml-4 mt-2"
color="dark"
size="xs"
href="/scripts/add?hub=hub%2F19822%2Fwindmill%2Fdynamic_skip_template"
disabled={!can_write}
target="_blank"
>
Create from template
</Button>
{/if}
</div>
</Label>
<Alert type="info" size="xs" title="Handler requirements">
Handler must return a boolean value. Return true to execute the scheduled job, false to skip.
</Alert>
</div>
</Section>
{:else if optionTabSelected === 'tag'}
<Section
label="Custom script tag"
frontend/src/lib/utils.ts
+1 -1
View File
@@ -1316,7 +1316,7 @@ export function isFlowPreview(job_kind: Job['job_kind'] | undefined) {
}
export function isNotFlow(job_kind: Job['job_kind'] | undefined) {
return job_kind !== 'flow' && job_kind !== 'singlescriptflow' && !isFlowPreview(job_kind)
return job_kind !== 'flow' && job_kind !== 'singlestepflow' && !isFlowPreview(job_kind)
}
export function isScriptPreview(job_kind: Job['job_kind'] | undefined) {