diff --git a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json index e7ed0aee65..d29a18c691 100644 --- a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json +++ b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json @@ -46,11 +46,11 @@ ] }, "nullable": [ - false, - false, - false, - false, - false, + true, + true, + true, + true, + true, true, true ] diff --git a/backend/.sqlx/query-12d37d75a429c0ddf2b2c190ab28bea5aefd27d0ed8a1bb2c8b3c1b0ece4efb7.json b/backend/.sqlx/query-12d37d75a429c0ddf2b2c190ab28bea5aefd27d0ed8a1bb2c8b3c1b0ece4efb7.json new file mode 100644 index 0000000000..684f4a6cb4 --- /dev/null +++ b/backend/.sqlx/query-12d37d75a429c0ddf2b2c190ab28bea5aefd27d0ed8a1bb2c8b3c1b0ece4efb7.json @@ -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" +} diff --git a/backend/.sqlx/query-145b364bcd45b6a8b3b80fd67a5ae17212785bd7206fbd3901a3b516eb77dc55.json b/backend/.sqlx/query-145b364bcd45b6a8b3b80fd67a5ae17212785bd7206fbd3901a3b516eb77dc55.json new file mode 100644 index 0000000000..574f239135 --- /dev/null +++ b/backend/.sqlx/query-145b364bcd45b6a8b3b80fd67a5ae17212785bd7206fbd3901a3b516eb77dc55.json @@ -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" +} diff --git a/backend/.sqlx/query-16e4b1bead9fc77fd98658b8cb8cc6d6bf1df758b30e99bd661da866062ef14f.json b/backend/.sqlx/query-16e4b1bead9fc77fd98658b8cb8cc6d6bf1df758b30e99bd661da866062ef14f.json new file mode 100644 index 0000000000..1af42ff529 --- /dev/null +++ b/backend/.sqlx/query-16e4b1bead9fc77fd98658b8cb8cc6d6bf1df758b30e99bd661da866062ef14f.json @@ -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" +} diff --git a/backend/.sqlx/query-1a54356c1e1353950bf6ab1d25ab21270131e9e93ca10d195664e7e5a774fe9e.json b/backend/.sqlx/query-1a54356c1e1353950bf6ab1d25ab21270131e9e93ca10d195664e7e5a774fe9e.json index 9462f9b48b..66453d2659 100644 --- a/backend/.sqlx/query-1a54356c1e1353950bf6ab1d25ab21270131e9e93ca10d195664e7e5a774fe9e.json +++ b/backend/.sqlx/query-1a54356c1e1353950bf6ab1d25ab21270131e9e93ca10d195664e7e5a774fe9e.json @@ -59,7 +59,9 @@ "failure", "command", "approval", - "preprocessor" + "preprocessor", + "schedule_handler_old", + "dynamic_skip" ] } } diff --git a/backend/.sqlx/query-a0c20436e0506bf9e0e50bd3dcabdd35131cc374edac0a08c5492954f08c9d98.json b/backend/.sqlx/query-23e4c6e3dc6a48f702c2b26a6b1f94668e086caaa0093a3b685f87483513b0d2.json similarity index 74% rename from backend/.sqlx/query-a0c20436e0506bf9e0e50bd3dcabdd35131cc374edac0a08c5492954f08c9d98.json rename to backend/.sqlx/query-23e4c6e3dc6a48f702c2b26a6b1f94668e086caaa0093a3b685f87483513b0d2.json index 1724d969cf..fe2b1e3427 100644 --- a/backend/.sqlx/query-a0c20436e0506bf9e0e50bd3dcabdd35131cc374edac0a08c5492954f08c9d98.json +++ b/backend/.sqlx/query-23e4c6e3dc6a48f702c2b26a6b1f94668e086caaa0093a3b685f87483513b0d2.json @@ -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" } diff --git a/backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json b/backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json index 6007094518..fa013ba585 100644 --- a/backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json +++ b/backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json @@ -65,7 +65,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json b/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json index 5c5d9b68ba..9cda5cc93e 100644 --- a/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json +++ b/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json @@ -25,7 +25,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-348f73fa8222ec195bb5b2260d1595e6e9fa9d73779cf352e94039fd47de4294.json b/backend/.sqlx/query-4144c87c25a939aafb2f57da189d94d038bcad7a36fbf87e0403c89a979c5b3f.json similarity index 78% rename from backend/.sqlx/query-348f73fa8222ec195bb5b2260d1595e6e9fa9d73779cf352e94039fd47de4294.json rename to backend/.sqlx/query-4144c87c25a939aafb2f57da189d94d038bcad7a36fbf87e0403c89a979c5b3f.json index 4f9e060ead..5ea52ccd39 100644 --- a/backend/.sqlx/query-348f73fa8222ec195bb5b2260d1595e6e9fa9d73779cf352e94039fd47de4294.json +++ b/backend/.sqlx/query-4144c87c25a939aafb2f57da189d94d038bcad7a36fbf87e0403c89a979c5b3f.json @@ -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" } diff --git a/backend/.sqlx/query-488dd591096b2b47787afdc3a1d73917ed13269f2ee20b86df79fca2c8efe672.json b/backend/.sqlx/query-488dd591096b2b47787afdc3a1d73917ed13269f2ee20b86df79fca2c8efe672.json index af477b4db6..3cfb853bc2 100644 --- a/backend/.sqlx/query-488dd591096b2b47787afdc3a1d73917ed13269f2ee20b86df79fca2c8efe672.json +++ b/backend/.sqlx/query-488dd591096b2b47787afdc3a1d73917ed13269f2ee20b86df79fca2c8efe672.json @@ -37,7 +37,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json b/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json index 63363e74ff..15dba71f3e 100644 --- a/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json +++ b/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json @@ -95,7 +95,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-7cd070aaca3b4f95bb6e669ac07f288bd1cc8a85625255d86e45f63e8267d34c.json b/backend/.sqlx/query-7927b80ce75d99b2a30f6b29196af000578a3c166509f032d14452cc637d884f.json similarity index 94% rename from backend/.sqlx/query-7cd070aaca3b4f95bb6e669ac07f288bd1cc8a85625255d86e45f63e8267d34c.json rename to backend/.sqlx/query-7927b80ce75d99b2a30f6b29196af000578a3c166509f032d14452cc637d884f.json index 99095e3fcc..84585a66a9 100644 --- a/backend/.sqlx/query-7cd070aaca3b4f95bb6e669ac07f288bd1cc8a85625255d86e45f63e8267d34c.json +++ b/backend/.sqlx/query-7927b80ce75d99b2a30f6b29196af000578a3c166509f032d14452cc637d884f.json @@ -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" } diff --git a/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json b/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json index e4a0fec1b2..45a36b4fe3 100644 --- a/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json +++ b/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json @@ -35,7 +35,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-804fc11e35f4afc0db194b6fe2594f91df7e588d4d2431bc85f4d8734920c8bf.json b/backend/.sqlx/query-804fc11e35f4afc0db194b6fe2594f91df7e588d4d2431bc85f4d8734920c8bf.json new file mode 100644 index 0000000000..6f08d98113 --- /dev/null +++ b/backend/.sqlx/query-804fc11e35f4afc0db194b6fe2594f91df7e588d4d2431bc85f4d8734920c8bf.json @@ -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" +} diff --git a/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json b/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json index 5500f83561..c18d338929 100644 --- a/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json +++ b/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json @@ -35,7 +35,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-85705fc3d7f8ba5f1b12d5fb222c38fc64deb1226aab9dc3bc4465324fce37d1.json b/backend/.sqlx/query-85705fc3d7f8ba5f1b12d5fb222c38fc64deb1226aab9dc3bc4465324fce37d1.json new file mode 100644 index 0000000000..bce7324fb6 --- /dev/null +++ b/backend/.sqlx/query-85705fc3d7f8ba5f1b12d5fb222c38fc64deb1226aab9dc3bc4465324fce37d1.json @@ -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" +} diff --git a/backend/.sqlx/query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json b/backend/.sqlx/query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json index a5a077eb73..2c70bdd6b6 100644 --- a/backend/.sqlx/query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json +++ b/backend/.sqlx/query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json @@ -36,7 +36,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json b/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json index bcc10fcf4a..32358798b9 100644 --- a/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json +++ b/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json @@ -25,7 +25,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-a0b3e10e077d30c1da135dff9feca3761d400391f1f46a8294da3e6c9af63887.json b/backend/.sqlx/query-a0b3e10e077d30c1da135dff9feca3761d400391f1f46a8294da3e6c9af63887.json new file mode 100644 index 0000000000..1afc61978e --- /dev/null +++ b/backend/.sqlx/query-a0b3e10e077d30c1da135dff9feca3761d400391f1f46a8294da3e6c9af63887.json @@ -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" +} diff --git a/backend/.sqlx/query-a6608d47b96d851eb7b04d2e4b472889ff257db0f0e3e9252adda2e4ef2039d6.json b/backend/.sqlx/query-a6608d47b96d851eb7b04d2e4b472889ff257db0f0e3e9252adda2e4ef2039d6.json index 8fea4c40ec..e2baca86e0 100644 --- a/backend/.sqlx/query-a6608d47b96d851eb7b04d2e4b472889ff257db0f0e3e9252adda2e4ef2039d6.json +++ b/backend/.sqlx/query-a6608d47b96d851eb7b04d2e4b472889ff257db0f0e3e9252adda2e4ef2039d6.json @@ -95,7 +95,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-ab04cda71f8e2be9acbecabe1ee5ef756b8e5c1955fbe111df9ee171dc262338.json b/backend/.sqlx/query-ab04cda71f8e2be9acbecabe1ee5ef756b8e5c1955fbe111df9ee171dc262338.json new file mode 100644 index 0000000000..fe1a67b0d4 --- /dev/null +++ b/backend/.sqlx/query-ab04cda71f8e2be9acbecabe1ee5ef756b8e5c1955fbe111df9ee171dc262338.json @@ -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" +} diff --git a/backend/.sqlx/query-b3c02fd225a6aa78785d466e7f033b38deb9c7fa17bd3836c9ad8884f27be84a.json b/backend/.sqlx/query-b3c02fd225a6aa78785d466e7f033b38deb9c7fa17bd3836c9ad8884f27be84a.json index 5a6aa7dd69..313180bd10 100644 --- a/backend/.sqlx/query-b3c02fd225a6aa78785d466e7f033b38deb9c7fa17bd3836c9ad8884f27be84a.json +++ b/backend/.sqlx/query-b3c02fd225a6aa78785d466e7f033b38deb9c7fa17bd3836c9ad8884f27be84a.json @@ -30,7 +30,7 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow", + "singlestepflow", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json index 54e94cfb8f..99269c9851 100644 --- a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json +++ b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json @@ -18,8 +18,8 @@ "Left": [] }, "nullable": [ - false, - true + true, + false ] }, "hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76" diff --git a/backend/.sqlx/query-b4a9abcb38997587b28655b0f4a212a5bd4039b57fab20b163617e33a4c9dd46.json b/backend/.sqlx/query-b4a9abcb38997587b28655b0f4a212a5bd4039b57fab20b163617e33a4c9dd46.json new file mode 100644 index 0000000000..a49baeefaf --- /dev/null +++ b/backend/.sqlx/query-b4a9abcb38997587b28655b0f4a212a5bd4039b57fab20b163617e33a4c9dd46.json @@ -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" +} diff --git a/backend/.sqlx/query-bc9a17567cc71f51a8fa0a6fe12c0aa3a52e2ba194d40c66993d02ae2c129327.json b/backend/.sqlx/query-bc9a17567cc71f51a8fa0a6fe12c0aa3a52e2ba194d40c66993d02ae2c129327.json new file mode 100644 index 0000000000..6a5c49f43a --- /dev/null +++ b/backend/.sqlx/query-bc9a17567cc71f51a8fa0a6fe12c0aa3a52e2ba194d40c66993d02ae2c129327.json @@ -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" +} diff --git a/backend/.sqlx/query-daf9674838fb3e3653a356c7434c719616a614d77e726433737e5f5d9bd60134.json b/backend/.sqlx/query-daf9674838fb3e3653a356c7434c719616a614d77e726433737e5f5d9bd60134.json deleted file mode 100644 index 2e2a9ba027..0000000000 --- a/backend/.sqlx/query-daf9674838fb3e3653a356c7434c719616a614d77e726433737e5f5d9bd60134.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-e2f4eeb896c22e3215370ed30ed5cc890503e4197b0fdb3a906504eba166aa97.json b/backend/.sqlx/query-e2f4eeb896c22e3215370ed30ed5cc890503e4197b0fdb3a906504eba166aa97.json deleted file mode 100644 index 46da760227..0000000000 --- a/backend/.sqlx/query-e2f4eeb896c22e3215370ed30ed5cc890503e4197b0fdb3a906504eba166aa97.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-ff0403790674cdb07022af71c2377afbd8b3a660b3be27514b517c077c63c238.json b/backend/.sqlx/query-ff0403790674cdb07022af71c2377afbd8b3a660b3be27514b517c077c63c238.json new file mode 100644 index 0000000000..00662e2fd2 --- /dev/null +++ b/backend/.sqlx/query-ff0403790674cdb07022af71c2377afbd8b3a660b3be27514b517c077c63c238.json @@ -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" +} diff --git a/backend/migrations/20250930145445_add_dynamic_skip.down.sql b/backend/migrations/20250930145445_add_dynamic_skip.down.sql new file mode 100644 index 0000000000..c9eeecd8a5 --- /dev/null +++ b/backend/migrations/20250930145445_add_dynamic_skip.down.sql @@ -0,0 +1,2 @@ +-- Remove dynamic_skip column from schedule table +ALTER TABLE schedule DROP COLUMN dynamic_skip; diff --git a/backend/migrations/20250930145445_add_dynamic_skip.up.sql b/backend/migrations/20250930145445_add_dynamic_skip.up.sql new file mode 100644 index 0000000000..4879aeead9 --- /dev/null +++ b/backend/migrations/20250930145445_add_dynamic_skip.up.sql @@ -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; diff --git a/backend/migrations/20251002151319_rename_singlescriptflow_to_singlestepflow.down.sql b/backend/migrations/20251002151319_rename_singlescriptflow_to_singlestepflow.down.sql new file mode 100644 index 0000000000..45ffaabec5 --- /dev/null +++ b/backend/migrations/20251002151319_rename_singlescriptflow_to_singlestepflow.down.sql @@ -0,0 +1,2 @@ +-- Revert singlestepflow back to singlescriptflow in job_kind enum +ALTER TYPE job_kind RENAME VALUE 'singlestepflow' TO 'singlescriptflow'; diff --git a/backend/migrations/20251002151319_rename_singlescriptflow_to_singlestepflow.up.sql b/backend/migrations/20251002151319_rename_singlescriptflow_to_singlestepflow.up.sql new file mode 100644 index 0000000000..eb97db090d --- /dev/null +++ b/backend/migrations/20251002151319_rename_singlescriptflow_to_singlestepflow.up.sql @@ -0,0 +1,2 @@ +-- Rename singlescriptflow to singlestepflow in job_kind enum +ALTER TYPE job_kind RENAME VALUE 'singlescriptflow' TO 'singlestepflow'; diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index e012887d70..b8a1d455e1 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -2035,7 +2035,7 @@ async fn handle_zombie_jobs(db: &Pool, 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, 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 diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index f9a68ed567..362df943a0 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -208,6 +208,7 @@ async fn test_deno_flow(db: Pool) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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) -> anyhow::Result<()> { continue_on_error: None, skip_if: None, apply_preprocessor: None, + pass_flow_input_directly: None, }, ], same_worker: true, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index a4dd3d2f2e..6f6ac26d6e 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -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 diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index caf65a473e..cd21e5b3c8 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -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, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 813e912d6e..27c4ed7b9e 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -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 { diff --git a/backend/windmill-api/src/schedule.rs b/backend/windmill-api/src/schedule.rs index 9eaa897341..3b7413781b 100644 --- a/backend/windmill-api/src/schedule.rs +++ b/backend/windmill-api/src/schedule.rs @@ -78,6 +78,7 @@ pub struct NewSchedule { pub tag: Option, pub paused_until: Option>, pub cron_version: Option, + pub dynamic_skip: Option, } #[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, @@ -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, pub paused_until: Option>, pub cron_version: Option, + pub dynamic_skip: Option, } pub async fn clear_schedule<'c>( diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 7791104e11..6aaf77366b 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -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?; diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index 16e4140534..5da92541bb 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -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, diff --git a/backend/windmill-common/src/cache.rs b/backend/windmill-common/src/cache.rs index 4890b9715b..a6305b784f 100644 --- a/backend/windmill-common/src/cache.rs +++ b/backend/windmill-common/src/cache.rs @@ -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, }, diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 16accc7d0f..810368f901 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -393,6 +393,8 @@ pub struct FlowModule { pub skip_if: Option, #[serde(skip_serializing_if = "Option::is_none")] pub apply_preprocessor: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub pass_flow_input_directly: Option, } #[derive(Deserialize, Serialize, Debug, Clone)] @@ -617,6 +619,8 @@ pub enum FlowModuleValue { tag_override: Option, #[serde(skip_serializing_if = "Option::is_none")] is_trigger: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pass_flow_input_directly: Option, }, /// Reference to another flow on the workspace @@ -625,6 +629,8 @@ pub enum FlowModuleValue { #[serde(alias = "input_transform")] input_transforms: HashMap, path: String, + #[serde(skip_serializing_if = "Option::is_none")] + pass_flow_input_directly: Option, }, /// For loop node @@ -757,6 +763,7 @@ struct UntaggedFlowModuleValue { modules_node: Option, assets: Option>, tools: Option>, + pass_flow_input_directly: Option, } 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) { continue_on_error: None, skip_if: None, apply_preprocessor: None, + pass_flow_input_directly: None, }); } } diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index e8c663988b..4608fe72b3 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -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, restarted_from: Option, }, - SingleScriptFlow { + SingleStepFlow { path: String, - hash: ScriptHash, + hash: Option, + flow_version: Option, args: HashMap>, retry: Option, error_handler_path: Option, error_handler_args: Option>>, + skip_handler: Option, custom_concurrency_key: Option, concurrent_limit: Option, concurrency_time_window_s: Option, @@ -426,6 +428,14 @@ pub enum JobPayload { }, } +#[derive(Clone, Serialize, Deserialize, Debug)] +pub struct SkipHandler { + pub path: String, + pub args: HashMap>, + pub stop_condition: String, + pub stop_message: String, +} + #[derive(Clone, Serialize, Deserialize, Debug, Default)] pub struct RawCode { pub content: String, diff --git a/backend/windmill-common/src/schedule.rs b/backend/windmill-common/src/schedule.rs index e78d3a211c..cc7c9af758 100644 --- a/backend/windmill-common/src/schedule.rs +++ b/backend/windmill-common/src/schedule.rs @@ -60,6 +60,8 @@ pub struct Schedule { pub paused_until: Option>, #[serde(skip_serializing_if = "Option::is_none")] pub cron_version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub dynamic_skip: Option, } impl Schedule { diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index dabedefc00..66369111f2 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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::::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::::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::::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::::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", diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 4ae3621b73..c35a94a2dc 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -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, // tag + Option, // timeout + Option, // on_behalf_of_email + String, // created_by + Option, // hash (for scripts) + Option, // flow_version (for flows) + Option, // retry +)> { + let parsed_retry = schedule + .retry + .clone() + .and_then(|r| serde_json::from_value::(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::>::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, diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 06d265e435..b4ff9cf5bf 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -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 diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 50b1f8a967..64b350079a 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -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, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 6889f487cc..42fbd2e5f6 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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, diff --git a/frontend/src/lib/components/flows/scheduleUtils.ts b/frontend/src/lib/components/flows/scheduleUtils.ts index 77642ecc40..ad35e5f61f 100644 --- a/frontend/src/lib/components/flows/scheduleUtils.ts +++ b/frontend/src/lib/components/flows/scheduleUtils.ts @@ -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) { diff --git a/frontend/src/lib/components/runs/JobsLoader.svelte b/frontend/src/lib/components/runs/JobsLoader.svelte index 920609665a..86df1963d0 100644 --- a/frontend/src/lib/components/runs/JobsLoader.svelte +++ b/frontend/src/lib/components/runs/JobsLoader.svelte @@ -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, diff --git a/frontend/src/lib/components/runs/RunRow.svelte b/frontend/src/lib/components/runs/RunRow.svelte index b6858a0467..1c25bcb578 100644 --- a/frontend/src/lib/components/runs/RunRow.svelte +++ b/frontend/src/lib/components/runs/RunRow.svelte @@ -85,7 +85,7 @@ jobKind === 'script' || isScriptPreview(jobKind) || jobKind === 'script_hub' || - jobKind === 'singlescriptflow' + jobKind === 'singlestepflow' ) { return Code } else if (jobKind === 'aiagent') { diff --git a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte index e3309f99e0..37bb11b409 100644 --- a/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte +++ b/frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte @@ -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 @@ Recovery Handler Success Handler Retries + Dynamic skip {#if itemKind === 'script'} Custom tag {/if} @@ -1186,6 +1190,43 @@ /> {/if} + {:else if optionTabSelected === 'dynamic_skip'} +
+ {#snippet header()} + + Optional script to filter scheduled dates. Receives the proposed datetime and returns + boolean. True = run on this date, False = skip to next occurrence. + + {/snippet} +
+ + + Handler must return a boolean value. Return true to execute the scheduled job, false to skip. + +
+
{:else if optionTabSelected === 'tag'}