starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-07 08:02:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: handle singlestepflow zombies and stop filtering them from runs page (#9055)

Browse Source 
* fix: handle singlestepflow zombies and stop filtering them from runs page

* fix: support singlestepflow in batch_rerun_jobs

Previous PR added singlestepflow to list_selected_job_groups so the BatchReRun
pane shows them, but batch_rerun_jobs_inner still joined on kind = 'script' /
'flow' with j.runnable_id (which is NULL for SingleStepFlow), so the rows were
silently filtered out — user sees the option, click Re-run, gets zero successes.

Mirror the norm_kind CTE projection from list_selected_job_groups inside
batch_rerun_jobs_inner: pull the wrapped runnable type and pinned script hash
from raw_flow.modules[id='a'], cast back to JOB_KIND so the existing handler
dispatch works unchanged. Path-based schema fallback so input_transforms still
resolve at rerun time.

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

* fix: project singlestepflow in batch-rerun schema lookups

Codex review pointed out two follow-on regressions from the previous fix:

(1) list_selected_job_groups returned schemas with script_hash=null and
schema=null for singlestepflow rows because the inner schemas subquery still
joined runnable metadata via j.runnable_id (NULL for SingleStepFlow). The
BatchReRun pane consumes every selected.schemas entry through
mergeSchemasForBatchReruns / buildExtraLibForBatchReruns, both of which
assume real schema objects.

(2) When use_latest_version=true, batch_rerun_handle_job re-fetched
latest_schema from v2_job filtering jb.kind='script' or 'flow' — neither
matched singlestepflow, so schema came back NULL and every input_transforms
entry silently no-op'd.

Both queries now project singlestepflow rows via raw_flow.modules[id='a'] —
norm_kind for dispatch and effective_hash for the schemas join, plus a
path-based latest-schema fallback so flow-wrapped SSF (no version pinning)
and any SSF whose pinned hash has been deleted still resolve.

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

* test: add batch_rerun integration tests, fix SSF hash hex parsing

Adds 11 integration tests against /jobs/run/batch_rerun_jobs and
/jobs/list_selected_job_groups (both endpoints had zero CI coverage).
Tests cover the full 4-kind × 3-mode matrix: regular Script and Flow
(baseline regression for the SQL refactor), script-wrapped and flow-
wrapped SingleStepFlow (regression for the bugs this PR fixes), and a
mixed-kind batch.

Writing the tests caught a real bug in the previous commit: ScriptHash
serializes as a 16-char hex string in raw_flow.modules[a].value.hash
(per the custom Serialize impl in windmill-types/scripts.rs), not as
an integer. The earlier `(m->'value'->>'hash')::bigint` cast worked
on the hand-inserted SQL fixture I'd used for live testing (which
embedded the hash as a raw integer) but failed in production where
all SSF jobs are pushed via JobPayload::SingleStepFlow's serialized
form. Replaced with `('x' || lpad(hex, 16, '0'))::bit(64)::bigint` —
preserves the twos-complement bit pattern so both positive and
negative i64 hashes round-trip correctly.

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

* Update SQLx metadata

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
hugocasa
2026-05-07 08:00:38 +00:00
committed by GitHub
co-authored by Claude Opus 4.7 windmill-internal-app[bot]
parent 5ecb644dd7
commit e74f06cb56
24 changed files with 1284 additions and 328 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-bd000b7ab6438826093ba556af819741b493a91babaeb103852d3912f520719f.json → backend/.sqlx/query-074dc85ffb596585eb99336ffa34fa29fc0a1aff4d10d6fe4aba4b5357afb0f3.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n j.id AS \"id!\", j.workspace_id AS \"workspace_id!\", j.parent_job, j.flow_step_id IS NOT NULL AS \"is_flow_step?\",\n COALESCE(s.flow_status, s.workflow_as_code_status) AS \"flow_status: Box<str>\", r.ping AS last_ping, j.same_worker AS \"same_worker?\",\n q.worker AS \"worker?\",\n wp.ping_at AS \"worker_last_ping?\",\n wp.memory_usage AS \"worker_memory_usage?\",\n wp.wm_memory_usage AS \"worker_wm_memory_usage?\",\n wp.memory AS \"worker_memory_total?\",\n wp.worker_group AS \"worker_group?\",\n wp.wm_version AS \"worker_version?\",\n wp.current_job_id AS \"worker_current_job_id?\",\n wp.worker_instance AS \"worker_instance?\"\n FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n LEFT JOIN worker_ping wp ON wp.worker = q.worker\n WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()\n AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode')\n AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval\n AND q.canceled_by IS NULL\n\n ",
"query": "\n SELECT\n j.id AS \"id!\", j.workspace_id AS \"workspace_id!\", j.parent_job, j.flow_step_id IS NOT NULL AS \"is_flow_step?\",\n COALESCE(s.flow_status, s.workflow_as_code_status) AS \"flow_status: Box<str>\", r.ping AS last_ping, j.same_worker AS \"same_worker?\",\n q.worker AS \"worker?\",\n wp.ping_at AS \"worker_last_ping?\",\n wp.memory_usage AS \"worker_memory_usage?\",\n wp.wm_memory_usage AS \"worker_wm_memory_usage?\",\n wp.memory AS \"worker_memory_total?\",\n wp.worker_group AS \"worker_group?\",\n wp.wm_version AS \"worker_version?\",\n wp.current_job_id AS \"worker_current_job_id?\",\n wp.worker_instance AS \"worker_instance?\"\n FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n LEFT JOIN worker_ping wp ON wp.worker = q.worker\n WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()\n AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode' OR j.kind = 'singlestepflow')\n AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval\n AND q.canceled_by IS NULL\n\n ",
"describe": {
"columns": [
{
@@ -108,5 +108,5 @@
false
]
},
"hash": "bd000b7ab6438826093ba556af819741b493a91babaeb103852d3912f520719f"
"hash": "074dc85ffb596585eb99336ffa34fa29fc0a1aff4d10d6fe4aba4b5357afb0f3"
}
backend/.sqlx/query-0c89ef278782f5a72b0b07ab3ba0edc487f03edd61936fcf77dee93fb22839ea.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT jsonb_build_object(\n 'kind', jb.kind,\n 'script_path', jb.runnable_path,\n 'latest_schema', COALESCE(\n (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.workspace_id = $1 AND s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),\n (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.workspace_id = $1 AND flow.path = jb.runnable_path AND jb.kind = 'flow')\n ),\n 'schemas', ARRAY(\n SELECT jsonb_build_object(\n 'script_hash', LPAD(TO_HEX(COALESCE(s.hash, f.id)), 16, '0'),\n 'job_ids', ARRAY_AGG(DISTINCT j.id),\n 'schema', (ARRAY_AGG(COALESCE(s.schema, f.schema)))[1]\n ) FROM v2_job j\n LEFT JOIN script s ON s.hash = j.runnable_id AND j.kind = 'script'\n LEFT JOIN flow_version f ON f.id = j.runnable_id AND j.kind = 'flow'\n WHERE j.id = ANY(ARRAY_AGG(jb.id))\n GROUP BY COALESCE(s.hash, f.id)\n )\n ) FROM v2_job jb\n WHERE (jb.kind = 'flow' OR jb.kind = 'script')\n AND jb.workspace_id = $1 AND jb.id = ANY($2)\n GROUP BY jb.kind, jb.runnable_path",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "jsonb_build_object",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"UuidArray"
]
},
"nullable": [
null
]
},
"hash": "0c89ef278782f5a72b0b07ab3ba0edc487f03edd61936fcf77dee93fb22839ea"
}
backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json
Generated
-63
View File
@@ -1,63 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n id,\n args as \"args: _\",\n created_at\n FROM v2_job\n WHERE workspace_id = $1\n AND (\n kind = 'unassigned_script'::JOB_KIND OR\n kind = 'unassigned_flow'::JOB_KIND OR\n kind = 'unassigned_singlestepflow'::JOB_KIND\n )\n AND trigger_kind = $2\n AND trigger = $3\n AND id = ANY($4)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "args: _",
"type_info": "Jsonb"
},
{
"ordinal": 2,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "job_trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats",
"schedule",
"app",
"ui",
"postgres",
"sqs",
"gcp",
"mqtt",
"nextcloud",
"google",
"ci_test",
"github",
"azure"
]
}
}
},
"Text",
"UuidArray"
]
},
"nullable": [
false,
true,
false
]
},
"hash": "19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2"
}
backend/.sqlx/query-dbc7e74e259b502e700491ee0248e0c9c8c61e1bf609be60ac5dc5d438189353.json → backend/.sqlx/query-3e68b73a1c2fdcb2ef538de1addbb096921e26fed398d5c0966b60503ca97fdb.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n WITH job_info AS (\n SELECT id, kind::text AS kind, parent_job\n FROM v2_job\n WHERE id = $1\n )\n SELECT\n q.id AS \"id!\",\n s.flow_status,\n q.suspend AS \"suspend!\",\n j.runnable_path AS script_path,\n j.permissioned_as_email AS email,\n (ji.kind IN ('flow', 'flowpreview')) AS \"is_flow_level!\",\n (ji.kind NOT IN ('flow', 'flowpreview') AND q.id = ji.id) AS \"is_wac!\"\n FROM job_info ji\n JOIN v2_job_queue q ON q.id = CASE\n WHEN ji.kind IN ('flow', 'flowpreview') THEN ji.id\n ELSE COALESCE(ji.parent_job, ji.id)\n END\n JOIN v2_job j ON j.id = q.id\n LEFT JOIN v2_job_status s ON s.id = q.id\n FOR UPDATE OF q\n ",
"query": "\n WITH job_info AS (\n SELECT id, kind::text AS kind, parent_job\n FROM v2_job\n WHERE id = $1\n )\n SELECT\n q.id AS \"id!\",\n s.flow_status,\n q.suspend AS \"suspend!\",\n j.runnable_path AS script_path,\n j.permissioned_as_email AS email,\n (ji.kind IN ('flow', 'flowpreview', 'singlestepflow')) AS \"is_flow_level!\",\n (ji.kind NOT IN ('flow', 'flowpreview', 'singlestepflow') AND q.id = ji.id) AS \"is_wac!\"\n FROM job_info ji\n JOIN v2_job_queue q ON q.id = CASE\n WHEN ji.kind IN ('flow', 'flowpreview', 'singlestepflow') THEN ji.id\n ELSE COALESCE(ji.parent_job, ji.id)\n END\n JOIN v2_job j ON j.id = q.id\n LEFT JOIN v2_job_status s ON s.id = q.id\n FOR UPDATE OF q\n ",
"describe": {
"columns": [
{
@@ -54,5 +54,5 @@
null
]
},
"hash": "dbc7e74e259b502e700491ee0248e0c9c8c61e1bf609be60ac5dc5d438189353"
"hash": "3e68b73a1c2fdcb2ef538de1addbb096921e26fed398d5c0966b60503ca97fdb"
}
backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json
Generated
+99
View File
@@ -0,0 +1,99 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n id,\n args as \"args: _\",\n created_at,\n kind AS \"kind: _\"\n FROM v2_job\n WHERE workspace_id = $1\n AND (\n kind = 'unassigned_script'::JOB_KIND OR\n kind = 'unassigned_flow'::JOB_KIND OR\n kind = 'unassigned_singlestepflow'::JOB_KIND\n )\n AND trigger_kind = $2\n AND trigger = $3\n AND id = ANY($4)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "args: _",
"type_info": "Jsonb"
},
{
"ordinal": 2,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 3,
"name": "kind: _",
"type_info": {
"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",
"unassigned_script",
"unassigned_flow",
"unassigned_singlestepflow"
]
}
}
}
}
],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "job_trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats",
"schedule",
"app",
"ui",
"postgres",
"sqs",
"gcp",
"mqtt",
"nextcloud",
"google",
"ci_test",
"github",
"azure"
]
}
}
},
"Text",
"UuidArray"
]
},
"nullable": [
false,
true,
false,
false
]
},
"hash": "4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d"
}
backend/.sqlx/query-6cda14bc33e3144b78b2147ac1153c516b5962b5648ae07c247fbd57b1b16ada.json
Generated
+89
View File
@@ -0,0 +1,89 @@
{
"db_name": "PostgreSQL",
"query": "WITH norm AS (\n SELECT\n j.id, j.workspace_id, j.runnable_path, j.runnable_id, j.kind, j.args,\n -- Project effective kind for dispatch: pass script/flow through;\n -- for singlestepflow, read the wrapped runnable's type from\n -- raw_flow.modules[id='a'].value.type (always 'script' or 'flow').\n CASE\n WHEN j.kind IN ('script', 'flow') THEN j.kind::text\n WHEN j.kind = 'singlestepflow' THEN\n COALESCE(\n (SELECT m->'value'->>'type'\n FROM jsonb_array_elements(j.raw_flow->'modules') m\n WHERE m->>'id' = 'a'\n LIMIT 1),\n 'script'\n )\n END AS norm_kind,\n -- Pinned script hash for script-wrapped singlestepflow lives in\n -- raw_flow.modules[id='a'].value.hash. Flow-wrapped doesn't pin a\n -- version, so this is NULL there (Flow rerun pushes by path).\n (CASE WHEN j.kind = 'singlestepflow' THEN\n (SELECT ('x' || lpad(m->'value'->>'hash', 16, '0'))::bit(64)::bigint\n FROM jsonb_array_elements(j.raw_flow->'modules') m\n WHERE m->>'id' = 'a'\n AND m->'value'->>'hash' IS NOT NULL\n LIMIT 1)\n END) AS ssf_hash\n FROM v2_job j\n WHERE j.id = ANY($1)\n AND j.workspace_id = $2\n AND j.kind IN ('script', 'flow', 'singlestepflow')\n )\n SELECT\n n.id,\n n.norm_kind::JOB_KIND AS \"kind!: _\",\n COALESCE(s.path, f.path, n.runnable_path) AS \"script_path!\",\n -- script_hash is unused on the Flow rerun path (path-based push), so\n -- 0 is a safe placeholder when no version is pinned.\n COALESCE(s.hash, f.id, n.ssf_hash, 0::bigint) AS \"script_hash!: _\",\n COALESCE(jc.started_at, jq.scheduled_for, make_date(1970, 1, 1)) AS \"scheduled_for!: _\",\n n.args AS input,\n -- Pinned schema for script/flow; latest-by-path fallback for\n -- singlestepflow so input_transforms still resolve at rerun time.\n COALESCE(\n s.schema,\n f.schema,\n (CASE WHEN n.kind = 'singlestepflow' AND n.norm_kind = 'script' THEN\n (SELECT s2.schema FROM script s2\n WHERE s2.workspace_id = $2 AND s2.path = n.runnable_path\n ORDER BY s2.created_at DESC LIMIT 1)\n END),\n (CASE WHEN n.kind = 'singlestepflow' AND n.norm_kind = 'flow' THEN\n (SELECT fv.schema FROM flow\n LEFT JOIN flow_version fv ON fv.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.workspace_id = $2 AND flow.path = n.runnable_path)\n END)\n ) AS \"schema: _\"\n FROM norm n\n LEFT JOIN script s ON s.hash = n.runnable_id AND n.kind = 'script'\n LEFT JOIN flow_version f ON f.id = n.runnable_id AND f.path = n.runnable_path AND n.kind = 'flow'\n LEFT JOIN v2_job_completed jc ON jc.id = n.id\n LEFT JOIN v2_job_queue jq ON jq.id = n.id\n WHERE n.norm_kind IS NOT NULL\n AND COALESCE(s.path, f.path, n.runnable_path) IS NOT NULL\n AND (\n n.kind = 'singlestepflow'\n OR (COALESCE(s.hash, f.id) IS NOT NULL AND COALESCE(s.path, f.path) IS NOT NULL)\n )",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "kind!: _",
"type_info": {
"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",
"unassigned_script",
"unassigned_flow",
"unassigned_singlestepflow"
]
}
}
}
},
{
"ordinal": 2,
"name": "script_path!",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "script_hash!: _",
"type_info": "Int8"
},
{
"ordinal": 4,
"name": "scheduled_for!: _",
"type_info": "Timestamptz"
},
{
"ordinal": 5,
"name": "input",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "schema: _",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"UuidArray",
"Text"
]
},
"nullable": [
false,
null,
null,
null,
null,
true,
null
]
},
"hash": "6cda14bc33e3144b78b2147ac1153c516b5962b5648ae07c247fbd57b1b16ada"
}
backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json
Generated
-62
View File
@@ -1,62 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n id,\n args as \"args: _\",\n created_at\n FROM v2_job\n WHERE workspace_id = $1\n AND (\n kind = 'unassigned_script'::JOB_KIND OR\n kind = 'unassigned_flow'::JOB_KIND OR\n kind = 'unassigned_singlestepflow'::JOB_KIND\n )\n AND trigger_kind = $2\n AND trigger = $3\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "args: _",
"type_info": "Jsonb"
},
{
"ordinal": 2,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "job_trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats",
"schedule",
"app",
"ui",
"postgres",
"sqs",
"gcp",
"mqtt",
"nextcloud",
"google",
"ci_test",
"github",
"azure"
]
}
}
},
"Text"
]
},
"nullable": [
false,
true,
false
]
},
"hash": "757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c"
}
backend/.sqlx/query-9f978c08b3a0150bb6f6ef71a6c4a1d8b126495a514ea4c0e791216aef55dc38.json
Generated
+59
View File
@@ -0,0 +1,59 @@
{
"db_name": "PostgreSQL",
"query": "SELECT kind AS \"kind: JobKind\", count(*) AS \"count!\"\n FROM v2_job\n WHERE workspace_id = $1\n AND id <> ALL($2)\n AND parent_job IS NULL\n GROUP BY kind\n ORDER BY kind::text",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "kind: JobKind",
"type_info": {
"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",
"unassigned_script",
"unassigned_flow",
"unassigned_singlestepflow"
]
}
}
}
},
{
"ordinal": 1,
"name": "count!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text",
"UuidArray"
]
},
"nullable": [
false,
null
]
},
"hash": "9f978c08b3a0150bb6f6ef71a6c4a1d8b126495a514ea4c0e791216aef55dc38"
}
backend/.sqlx/query-a1ab1f23f49496f745d89d6c33a91c6b693afc4477a634aa84f81db91f9e03a4.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE(\n (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),\n (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = jb.runnable_path AND jb.kind = 'flow')\n ) FROM v2_job jb\n WHERE jb.id = $1 AND jb.workspace_id = $2\n GROUP BY jb.kind, jb.runnable_path",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "coalesce",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
null
]
},
"hash": "a1ab1f23f49496f745d89d6c33a91c6b693afc4477a634aa84f81db91f9e03a4"
}
backend/.sqlx/query-edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd.json → backend/.sqlx/query-a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT usr.email, usage.executions\n FROM usr, LATERAL (\n SELECT COALESCE(SUM(c.duration_ms + 1000)/1000 , 0)::BIGINT executions\n FROM v2_job_completed c JOIN v2_job j USING (id)\n WHERE j.workspace_id = $1\n AND j.kind NOT IN ('flow', 'flowpreview', 'flownode')\n AND j.permissioned_as_email = usr.email\n AND now() - '1 week'::interval < j.created_at\n ) usage\n WHERE workspace_id = $1\n ",
"query": "\n SELECT usr.email, usage.executions\n FROM usr, LATERAL (\n SELECT COALESCE(SUM(c.duration_ms + 1000)/1000 , 0)::BIGINT executions\n FROM v2_job_completed c JOIN v2_job j USING (id)\n WHERE j.workspace_id = $1\n AND j.kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow')\n AND j.permissioned_as_email = usr.email\n AND now() - '1 week'::interval < j.created_at\n ) usage\n WHERE workspace_id = $1\n ",
"describe": {
"columns": [
{
@@ -24,5 +24,5 @@
null
]
},
"hash": "edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd"
"hash": "a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665"
}
backend/.sqlx/query-aba492cb21cbbd514959a16ca02ab3b62efb138edd968d90f10f9043ae9a7c62.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "WITH normalized AS (\n SELECT\n jb.id,\n jb.workspace_id,\n jb.runnable_path,\n CASE\n WHEN jb.kind IN ('flow', 'script') THEN jb.kind::text\n WHEN jb.kind = 'singlestepflow' THEN\n COALESCE(\n (SELECT m->'value'->>'type'\n FROM jsonb_array_elements(jb.raw_flow->'modules') m\n WHERE m->>'id' IN ('a', 'main')\n LIMIT 1),\n 'script'\n )\n ELSE NULL\n END AS norm_kind,\n COALESCE(\n jb.runnable_id,\n CASE WHEN jb.kind = 'singlestepflow' THEN\n (SELECT ('x' || lpad(m->'value'->>'hash', 16, '0'))::bit(64)::bigint\n FROM jsonb_array_elements(jb.raw_flow->'modules') m\n WHERE m->>'id' IN ('a', 'main')\n AND m->'value'->>'hash' IS NOT NULL\n LIMIT 1)\n END\n ) AS effective_hash\n FROM v2_job jb\n WHERE jb.kind IN ('flow', 'script', 'singlestepflow')\n AND jb.workspace_id = $1 AND jb.id = ANY($2)\n )\n SELECT jsonb_build_object(\n 'kind', n.norm_kind,\n 'script_path', n.runnable_path,\n 'latest_schema', COALESCE(\n (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.workspace_id = $1 AND s.path = n.runnable_path AND n.norm_kind = 'script' ORDER BY s.path, s.created_at DESC),\n (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.workspace_id = $1 AND flow.path = n.runnable_path AND n.norm_kind = 'flow')\n ),\n 'schemas', ARRAY(\n SELECT jsonb_build_object(\n 'script_hash', CASE WHEN COALESCE(s.hash, f.id) IS NULL THEN NULL ELSE LPAD(TO_HEX(COALESCE(s.hash, f.id)), 16, '0') END,\n 'job_ids', ARRAY_AGG(DISTINCT n2.id),\n 'schema', COALESCE(\n (ARRAY_AGG(COALESCE(s.schema, f.schema)))[1],\n CASE WHEN n.norm_kind = 'script' THEN\n (SELECT DISTINCT ON (s2.path) s2.schema FROM script s2 WHERE s2.workspace_id = $1 AND s2.path = n.runnable_path ORDER BY s2.path, s2.created_at DESC)\n END,\n CASE WHEN n.norm_kind = 'flow' THEN\n (SELECT fv.schema FROM flow LEFT JOIN flow_version fv ON fv.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.workspace_id = $1 AND flow.path = n.runnable_path)\n END\n )\n ) FROM normalized n2\n LEFT JOIN script s ON s.hash = n2.effective_hash AND n2.norm_kind = 'script'\n LEFT JOIN flow_version f ON f.id = n2.effective_hash AND n2.norm_kind = 'flow'\n WHERE n2.id = ANY(ARRAY_AGG(n.id))\n GROUP BY COALESCE(s.hash, f.id)\n )\n ) FROM normalized n\n WHERE n.norm_kind IS NOT NULL\n GROUP BY n.norm_kind, n.runnable_path",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "jsonb_build_object",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"UuidArray"
]
},
"nullable": [
null
]
},
"hash": "aba492cb21cbbd514959a16ca02ab3b62efb138edd968d90f10f9043ae9a7c62"
}
backend/.sqlx/query-b32d8b364001f5d54c1c4e564aac6b49d514771191a004c21c2e0d042f9d6f4d.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE(\n (SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = norm.path AND s.workspace_id = $2 AND norm.kind = 'script' ORDER BY s.path, s.created_at DESC),\n (SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = norm.path AND flow.workspace_id = $2 AND norm.kind = 'flow')\n ) FROM (\n SELECT\n jb.runnable_path AS path,\n CASE\n WHEN jb.kind IN ('script', 'flow') THEN jb.kind::text\n WHEN jb.kind = 'singlestepflow' THEN COALESCE(\n (SELECT m->'value'->>'type' FROM jsonb_array_elements(jb.raw_flow->'modules') m WHERE m->>'id' IN ('a', 'main') LIMIT 1),\n 'script'\n )\n END AS kind\n FROM v2_job jb\n WHERE jb.id = $1 AND jb.workspace_id = $2\n ) norm\n GROUP BY norm.kind, norm.path",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "coalesce",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
null
]
},
"hash": "b32d8b364001f5d54c1c4e564aac6b49d514771191a004c21c2e0d042f9d6f4d"
}
backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json
Generated
+98
View File
@@ -0,0 +1,98 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n id,\n args as \"args: _\",\n created_at,\n kind AS \"kind: _\"\n FROM v2_job\n WHERE workspace_id = $1\n AND (\n kind = 'unassigned_script'::JOB_KIND OR\n kind = 'unassigned_flow'::JOB_KIND OR\n kind = 'unassigned_singlestepflow'::JOB_KIND\n )\n AND trigger_kind = $2\n AND trigger = $3\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "args: _",
"type_info": "Jsonb"
},
{
"ordinal": 2,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 3,
"name": "kind: _",
"type_info": {
"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",
"unassigned_script",
"unassigned_flow",
"unassigned_singlestepflow"
]
}
}
}
}
],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "job_trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats",
"schedule",
"app",
"ui",
"postgres",
"sqs",
"gcp",
"mqtt",
"nextcloud",
"google",
"ci_test",
"github",
"azure"
]
}
}
},
"Text"
]
},
"nullable": [
false,
true,
false,
false
]
},
"hash": "bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77"
}
backend/.sqlx/query-e2905bca184696a80357d8e4126832a902b3088d91fbbb858f6c0aa9de8a5ff7.json
Generated
-89
View File
@@ -1,89 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n j.id,\n j.kind AS \"kind: _\",\n COALESCE(s.path, f.path) AS \"script_path!\",\n COALESCE(s.hash, f.id) AS \"script_hash!: _\",\n COALESCE(jc.started_at, jq.scheduled_for, make_date(1970, 1, 1)) AS \"scheduled_for!: _\",\n args AS input,\n COALESCE(s.schema, f.schema) AS \"schema: _\"\n FROM v2_job j\n LEFT JOIN script s ON j.runnable_id = s.hash AND j.kind = 'script'\n LEFT JOIN flow_version f ON j.runnable_id = f.id AND j.runnable_path = f.path AND j.kind = 'flow'\n LEFT JOIN v2_job_completed jc ON jc.id = j.id\n LEFT JOIN v2_job_queue jq ON jq.id = j.id\n WHERE j.id = ANY($1)\n AND j.workspace_id = $2\n AND COALESCE(s.hash, f.id) IS NOT NULL\n AND COALESCE(s.path, f.path) IS NOT NULL",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "kind: _",
"type_info": {
"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",
"unassigned_script",
"unassigned_flow",
"unassigned_singlestepflow"
]
}
}
}
},
{
"ordinal": 2,
"name": "script_path!",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "script_hash!: _",
"type_info": "Int8"
},
{
"ordinal": 4,
"name": "scheduled_for!: _",
"type_info": "Timestamptz"
},
{
"ordinal": 5,
"name": "input",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "schema: _",
"type_info": "Json"
}
],
"parameters": {
"Left": [
"UuidArray",
"Text"
]
},
"nullable": [
false,
false,
null,
null,
null,
true,
null
]
},
"hash": "e2905bca184696a80357d8e4126832a902b3088d91fbbb858f6c0aa9de8a5ff7"
}
backend/.sqlx/query-e8597d72fc73446d6ac1e76dffcbc6a1c77e754825b59ca4bc79c4e675bed8e5.json
Generated
+58
View File
@@ -0,0 +1,58 @@
{
"db_name": "PostgreSQL",
"query": "SELECT kind AS \"kind: JobKind\", args::text AS \"args!\"\n FROM v2_job WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "kind: JobKind",
"type_info": {
"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",
"unassigned_script",
"unassigned_flow",
"unassigned_singlestepflow"
]
}
}
}
},
{
"ordinal": 1,
"name": "args!",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
null
]
},
"hash": "e8597d72fc73446d6ac1e76dffcbc6a1c77e754825b59ca4bc79c4e675bed8e5"
}
backend/.sqlx/query-750f09238caf114d01f752a145fd4289f04f3961d5035d42f3fd0bd2f87eed1a.json → backend/.sqlx/query-f6eff53e00b33310bd9626b44b72af91a2da474ddf622f15f4f618dfcc7f1c39.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE v2_job_status f SET flow_status = JSONB_SET(flow_status, ARRAY['user_states'], JSONB_SET(COALESCE(flow_status->'user_states', '{}'::jsonb), ARRAY[$1], $2))\n FROM v2_job j\n WHERE f.id = $3 AND f.id = j.id AND j.workspace_id = $4 AND kind IN ('flow', 'flowpreview', 'flownode') RETURNING 1\n ",
"query": "\n UPDATE v2_job_status f SET flow_status = JSONB_SET(flow_status, ARRAY['user_states'], JSONB_SET(COALESCE(flow_status->'user_states', '{}'::jsonb), ARRAY[$1], $2))\n FROM v2_job j\n WHERE f.id = $3 AND f.id = j.id AND j.workspace_id = $4 AND kind IN ('flow', 'flowpreview', 'flownode', 'singlestepflow') RETURNING 1\n ",
"describe": {
"columns": [
{
@@ -21,5 +21,5 @@
null
]
},
"hash": "750f09238caf114d01f752a145fd4289f04f3961d5035d42f3fd0bd2f87eed1a"
"hash": "f6eff53e00b33310bd9626b44b72af91a2da474ddf622f15f4f618dfcc7f1c39"
}
backend/src/monitor.rs
+1 -1
View File
@@ -3404,7 +3404,7 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)
LEFT JOIN worker_ping wp ON wp.worker = q.worker
WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()
AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode')
AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode' OR j.kind = 'singlestepflow')
AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval
AND q.canceled_by IS NULL
backend/tests/batch_rerun.rs
+615
View File
@@ -0,0 +1,615 @@
//! Tests for `/jobs/run/batch_rerun_jobs` and `/jobs/list_selected_job_groups`.
//!
//! These cover both the long-existing baseline behavior (regular Script and Flow
//! jobs reran with default args, `use_latest_version=true`, and `input_transforms`)
//! and the SingleStepFlow projection paths added in the PR that introduced this
//! file. The endpoint had zero coverage before — every change here exists because
//! a code reviewer or reader caught it; the tests exist so the next reader doesn't
//! have to.
use serde_json::json;
use sqlx::{Pool, Postgres};
use uuid::Uuid;
use windmill_common::{
flows::Retry,
jobs::{JobKind, JobPayload},
runnable_settings::{ConcurrencySettings, DebouncingSettings},
scripts::{ScriptHash, ScriptLang},
};
use windmill_test_utils::*;
const WORKSPACE: &str = "test-workspace";
const SCRIPT_PATH: &str = "u/test-user/rerun_script";
const SCRIPT_HASH: i64 = 1111111111;
const FLOW_PATH: &str = "u/test-user/rerun_flow";
const FLOW_VERSION: i64 = 2222222222;
/// Mark a queued job as completed with success — needed so it's eligible for
/// `list_selected_job_groups` / `batch_rerun_jobs` (both join `v2_job_completed`).
async fn complete_job(db: &Pool<Postgres>, job_id: Uuid) -> anyhow::Result<()> {
sqlx::query(
"INSERT INTO v2_job_completed (workspace_id, id, result, status, duration_ms, started_at, completed_at)
VALUES ($1, $2, '{}'::jsonb, 'success', 0, now(), now())",
)
.bind(WORKSPACE)
.bind(job_id)
.execute(db)
.await?;
Ok(())
}
fn script_payload() -> JobPayload {
JobPayload::ScriptHash {
hash: ScriptHash(SCRIPT_HASH),
path: SCRIPT_PATH.to_string(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
language: ScriptLang::Deno,
priority: None,
apply_preprocessor: false,
concurrency_settings: ConcurrencySettings::default().into(),
debouncing_settings: DebouncingSettings::default(),
labels: None,
}
}
fn flow_payload() -> JobPayload {
JobPayload::Flow {
path: FLOW_PATH.to_string(),
dedicated_worker: None,
apply_preprocessor: false,
version: FLOW_VERSION,
labels: None,
}
}
fn ssf_script_payload(hash: Option<ScriptHash>, retry: Option<Retry>) -> JobPayload {
JobPayload::SingleStepFlow {
path: SCRIPT_PATH.to_string(),
hash,
flow_version: None,
args: Default::default(),
retry,
error_handler_path: None,
error_handler_args: None,
skip_handler: None,
cache_ttl: None,
cache_ignore_s3_path: None,
priority: None,
tag_override: None,
trigger_path: None,
apply_preprocessor: false,
concurrency_settings: ConcurrencySettings::default(),
debouncing_settings: DebouncingSettings::default(),
}
}
fn ssf_flow_payload() -> JobPayload {
JobPayload::SingleStepFlow {
path: FLOW_PATH.to_string(),
hash: None,
flow_version: Some(FLOW_VERSION),
args: Default::default(),
retry: None,
error_handler_path: None,
error_handler_args: None,
skip_handler: None,
cache_ttl: None,
cache_ignore_s3_path: None,
priority: None,
tag_override: None,
trigger_path: None,
apply_preprocessor: false,
concurrency_settings: ConcurrencySettings::default(),
debouncing_settings: DebouncingSettings::default(),
}
}
/// Push + complete a job so it's eligible for batch rerun.
async fn push_completed(
db: &Pool<Postgres>,
payload: JobPayload,
args: Vec<(&str, serde_json::Value)>,
) -> anyhow::Result<Uuid> {
let mut runner = RunJob::from(payload);
for (k, v) in args {
runner = runner.arg(k.to_string(), v);
}
let id = runner.push(db).await;
complete_job(db, id).await?;
Ok(id)
}
/// POST `/jobs/run/batch_rerun_jobs` and parse the SSE-style line-per-result
/// stream into a list of (uuid, error?) pairs.
async fn batch_rerun(
client: &windmill_api_client::Client,
body: serde_json::Value,
) -> anyhow::Result<Vec<Result<Uuid, String>>> {
let response = client
.client()
.post(format!(
"{}/w/{}/jobs/run/batch_rerun_jobs",
client.baseurl(),
WORKSPACE
))
.json(&body)
.send()
.await?;
assert!(
response.status().is_success(),
"batch_rerun_jobs returned {}",
response.status()
);
let body = response.text().await?;
Ok(body
.lines()
.filter(|l| !l.is_empty())
.map(|l| {
if let Some(err) = l.strip_prefix("Error: ") {
Err(err.to_string())
} else {
Ok(Uuid::parse_str(l.trim()).expect("rerun line should be a UUID"))
}
})
.collect())
}
/// Read kind + args of a freshly-rerun job.
async fn rerun_job(db: &Pool<Postgres>, id: Uuid) -> anyhow::Result<(JobKind, serde_json::Value)> {
let row = sqlx::query!(
r#"SELECT kind AS "kind: JobKind", args::text AS "args!"
FROM v2_job WHERE id = $1"#,
id
)
.fetch_one(db)
.await?;
Ok((row.kind, serde_json::from_str(&row.args)?))
}
// ---------------------------------------------------------------------------
// Baseline regression — regular Script and Flow rerun.
/// Default-mode rerun on a regular Script: new Script job inherits original args.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn batch_rerun_script_default(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let original = push_completed(&db, script_payload(), vec![("name", json!("orig"))]).await?;
let results = batch_rerun(
&client,
json!({
"job_ids": [original],
"script_options_by_path": {},
"flow_options_by_path": {},
}),
)
.await?;
assert_eq!(results.len(), 1, "expected one rerun");
let new_id = results[0].as_ref().expect("rerun should succeed").clone();
let (kind, args) = rerun_job(&db, new_id).await?;
assert_eq!(kind, JobKind::Script);
assert_eq!(args, json!({"name": "orig"}));
Ok(())
}
/// `use_latest_version=true` reruns regular Script via path-based push.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn batch_rerun_script_latest_version(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let original = push_completed(&db, script_payload(), vec![("name", json!("orig"))]).await?;
let results = batch_rerun(
&client,
json!({
"job_ids": [original],
"script_options_by_path": {
SCRIPT_PATH: { "use_latest_version": true }
},
"flow_options_by_path": {},
}),
)
.await?;
let new_id = results[0].as_ref().expect("rerun should succeed").clone();
let (kind, args) = rerun_job(&db, new_id).await?;
assert_eq!(kind, JobKind::Script);
assert_eq!(args, json!({"name": "orig"}));
Ok(())
}
/// `input_transforms` static value overrides original args on Script rerun.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn batch_rerun_script_static_transform(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let original = push_completed(&db, script_payload(), vec![("name", json!("orig"))]).await?;
let results = batch_rerun(
&client,
json!({
"job_ids": [original],
"script_options_by_path": {
SCRIPT_PATH: {
"input_transforms": { "name": { "type": "static", "value": "\"X\"" } }
}
},
"flow_options_by_path": {},
}),
)
.await?;
let new_id = results[0].as_ref().expect("rerun should succeed").clone();
let (_, args) = rerun_job(&db, new_id).await?;
assert_eq!(
args,
json!({"name": "\"X\""}),
"static transform should override original args"
);
Ok(())
}
/// Regular Flow rerun: new Flow job inherits original args.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn batch_rerun_flow_default(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let original = push_completed(&db, flow_payload(), vec![("name", json!("orig-flow"))]).await?;
let results = batch_rerun(
&client,
json!({
"job_ids": [original],
"script_options_by_path": {},
"flow_options_by_path": {},
}),
)
.await?;
let new_id = results[0].as_ref().expect("rerun should succeed").clone();
let (kind, args) = rerun_job(&db, new_id).await?;
assert_eq!(kind, JobKind::Flow);
assert_eq!(args, json!({"name": "orig-flow"}));
Ok(())
}
/// `input_transforms` override original args on Flow rerun (path-based,
/// frontend forces use_latest_version=true for flow).
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn batch_rerun_flow_static_transform(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let original = push_completed(&db, flow_payload(), vec![("name", json!("orig"))]).await?;
let results = batch_rerun(
&client,
json!({
"job_ids": [original],
"script_options_by_path": {},
"flow_options_by_path": {
FLOW_PATH: {
"use_latest_version": true,
"input_transforms": { "name": { "type": "static", "value": "\"FX\"" } }
}
},
}),
)
.await?;
let new_id = results[0].as_ref().expect("rerun should succeed").clone();
let (_, args) = rerun_job(&db, new_id).await?;
assert_eq!(args, json!({"name": "\"FX\""}));
Ok(())
}
// ---------------------------------------------------------------------------
// SingleStepFlow projection — the bug class this PR fixes.
/// Default rerun of a script-wrapped SingleStepFlow lands as a plain Script.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn batch_rerun_ssf_script_default(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let original = push_completed(
&db,
ssf_script_payload(Some(ScriptHash(SCRIPT_HASH)), None),
vec![("name", json!("orig-ssf"))],
)
.await?;
let results = batch_rerun(
&client,
json!({
"job_ids": [original],
"script_options_by_path": {},
"flow_options_by_path": {},
}),
)
.await?;
let new_id = results[0].as_ref().expect("SSF should be reruable").clone();
let (kind, args) = rerun_job(&db, new_id).await?;
// Wrapper unwraps to plain Script (retry policy belongs to the trigger, not to "rerun").
assert_eq!(kind, JobKind::Script);
assert_eq!(args, json!({"name": "orig-ssf"}));
Ok(())
}
/// `use_latest_version=true` + `input_transforms` on SSF — exercises the
/// `latest_schema` projection in `batch_rerun_handle_job`. Regression for the
/// fix in this PR's third commit; without it the transform silently no-ops.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn batch_rerun_ssf_script_latest_with_transform(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let original = push_completed(
&db,
ssf_script_payload(Some(ScriptHash(SCRIPT_HASH)), None),
vec![("name", json!("orig"))],
)
.await?;
let results = batch_rerun(
&client,
json!({
"job_ids": [original],
"script_options_by_path": {
SCRIPT_PATH: {
"use_latest_version": true,
"input_transforms": { "name": { "type": "static", "value": "\"S\"" } }
}
},
"flow_options_by_path": {},
}),
)
.await?;
let new_id = results[0]
.as_ref()
.expect("SSF rerun should succeed")
.clone();
let (_, args) = rerun_job(&db, new_id).await?;
assert_eq!(args, json!({"name": "\"S\""}));
Ok(())
}
/// Flow-wrapped SingleStepFlow projects to Flow and reruns by path.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn batch_rerun_ssf_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let original = push_completed(
&db,
ssf_flow_payload(),
vec![("name", json!("orig-ssf-flow"))],
)
.await?;
let results = batch_rerun(
&client,
json!({
"job_ids": [original],
"script_options_by_path": {},
"flow_options_by_path": {
FLOW_PATH: {
"use_latest_version": true,
"input_transforms": { "name": { "type": "static", "value": "\"FF\"" } }
}
},
}),
)
.await?;
let new_id = results[0]
.as_ref()
.expect("SSF flow should be reruable")
.clone();
let (kind, args) = rerun_job(&db, new_id).await?;
assert_eq!(kind, JobKind::Flow);
assert_eq!(args, json!({"name": "\"FF\""}));
Ok(())
}
/// Mixed batch (script + flow + SSF-script + SSF-flow) — exercises the
/// dispatch arms across all four kinds in a single request.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn batch_rerun_mixed_kinds(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let s = push_completed(&db, script_payload(), vec![("name", json!("S"))]).await?;
let f = push_completed(&db, flow_payload(), vec![("name", json!("F"))]).await?;
let ss = push_completed(
&db,
ssf_script_payload(Some(ScriptHash(SCRIPT_HASH)), None),
vec![("name", json!("SS"))],
)
.await?;
let sf = push_completed(&db, ssf_flow_payload(), vec![("name", json!("SF"))]).await?;
let results = batch_rerun(
&client,
json!({
"job_ids": [s, f, ss, sf],
"script_options_by_path": {},
"flow_options_by_path": {},
}),
)
.await?;
assert_eq!(
results.len(),
4,
"all 4 kinds should rerun, got: {results:?}"
);
for r in &results {
assert!(r.is_ok(), "expected all reruns to succeed, got {r:?}");
}
// Two new Scripts (regular + SSF-script projected) and two new Flows
// (regular + SSF-flow projected).
let counts = sqlx::query!(
r#"SELECT kind AS "kind: JobKind", count(*) AS "count!"
FROM v2_job
WHERE workspace_id = $1
AND id <> ALL($2)
AND parent_job IS NULL
GROUP BY kind
ORDER BY kind::text"#,
WORKSPACE,
&[s, f, ss, sf][..]
)
.fetch_all(&db)
.await?;
let mut script = 0;
let mut flow = 0;
for c in counts {
match c.kind {
JobKind::Script => script = c.count,
JobKind::Flow => flow = c.count,
_ => {}
}
}
assert_eq!(
script, 2,
"expected 2 new Script jobs (script + SSF-script)"
);
assert_eq!(flow, 2, "expected 2 new Flow jobs (flow + SSF-flow)");
Ok(())
}
// ---------------------------------------------------------------------------
// list_selected_job_groups — the front door for the BatchReRun pane. Crashes
// here would stop the user before they could even click Re-run.
#[derive(serde::Deserialize)]
struct GroupResp {
kind: String,
script_path: String,
schemas: Vec<SchemaEntry>,
latest_schema: Option<serde_json::Value>,
}
#[derive(serde::Deserialize)]
struct SchemaEntry {
script_hash: Option<String>,
schema: Option<serde_json::Value>,
job_ids: Vec<Uuid>,
}
async fn list_groups(
client: &windmill_api_client::Client,
job_ids: &[Uuid],
) -> anyhow::Result<Vec<GroupResp>> {
let response = client
.client()
.post(format!(
"{}/w/{}/jobs/list_selected_job_groups",
client.baseurl(),
WORKSPACE
))
.json(&job_ids)
.send()
.await?;
assert!(response.status().is_success());
Ok(response.json().await?)
}
/// SSF script-wrapped: pinned hash projected, schema non-null. Without this,
/// the BatchReRun pane crashes on `mergeSchemasForBatchReruns`.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn list_groups_ssf_script_has_schema(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let id = push_completed(
&db,
ssf_script_payload(Some(ScriptHash(SCRIPT_HASH)), None),
vec![("name", json!("x"))],
)
.await?;
let groups = list_groups(&client, &[id]).await?;
assert_eq!(groups.len(), 1);
let g = &groups[0];
assert_eq!(g.kind, "script", "SSF wrapping a script projects to script");
assert_eq!(g.script_path, SCRIPT_PATH);
assert!(g.latest_schema.is_some(), "latest_schema must resolve");
assert_eq!(g.schemas.len(), 1, "expected one schema entry");
let s = &g.schemas[0];
assert!(s.schema.is_some(), "per-version schema must be non-null");
let expected_hash = format!("{:0>16x}", SCRIPT_HASH as u64);
assert_eq!(s.script_hash.as_deref(), Some(expected_hash.as_str()));
assert_eq!(s.job_ids, vec![id]);
Ok(())
}
/// SSF flow-wrapped: no hash to pin, but path-based fallback fills in schema.
#[sqlx::test(fixtures("base", "batch_rerun"))]
async fn list_groups_ssf_flow_has_schema_via_path(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let client = windmill_api_client::create_client(
&format!("http://localhost:{}", server.addr.port()),
"SECRET_TOKEN".to_string(),
);
let id = push_completed(&db, ssf_flow_payload(), vec![("name", json!("x"))]).await?;
let groups = list_groups(&client, &[id]).await?;
let g = groups.first().expect("expected one group");
assert_eq!(g.kind, "flow", "SSF wrapping a flow projects to flow");
assert!(g.latest_schema.is_some());
assert_eq!(g.schemas.len(), 1);
let s = &g.schemas[0];
assert!(
s.schema.is_some(),
"flow-wrapped SSF schema falls back to latest by path"
);
Ok(())
}
backend/tests/fixtures/batch_rerun.sql
Vendored
+32
View File
@@ -0,0 +1,32 @@
-- Add-on fixture for batch_rerun.rs (combine with `base.sql` via
-- `#[sqlx::test(fixtures("base", "batch_rerun"))]`). Provides a deployed
-- script and a deployed flow so SingleStepFlow wrappers and rerun queries
-- find real runnable rows to join against.
INSERT INTO script (workspace_id, hash, path, content, language, kind, created_by, schema, summary, description, lock)
VALUES (
'test-workspace', 1111111111, 'u/test-user/rerun_script',
'export function main(name = "world") { return "hi " + name; }',
'deno', 'script', 'test-user',
'{"type":"object","properties":{"name":{"type":"string"}},"order":["name"],"required":[]}',
'', '', ''
);
INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, edited_at, schema, extra_perms, versions)
VALUES (
'test-workspace', 'u/test-user/rerun_flow',
'', '',
'{"modules":[{"id":"a","value":{"type":"rawscript","language":"deno","content":"export function main(name = \"world\"){ return \"flow:\" + name; }","input_transforms":{"name":{"type":"javascript","expr":"flow_input.name"}}}}]}',
'test-user', NOW(),
'{"type":"object","properties":{"name":{"type":"string"}},"order":["name"],"required":[]}',
'{}',
ARRAY[2222222222::bigint]
);
INSERT INTO flow_version (id, workspace_id, path, value, schema, created_by, created_at)
VALUES (
2222222222, 'test-workspace', 'u/test-user/rerun_flow',
'{"modules":[{"id":"a","value":{"type":"rawscript","language":"deno","content":"export function main(name = \"world\"){ return \"flow:\" + name; }","input_transforms":{"name":{"type":"javascript","expr":"flow_input.name"}}}}]}',
'{"type":"object","properties":{"name":{"type":"string"}},"order":["name"],"required":[]}',
'test-user', NOW()
);
backend/windmill-api-users/src/users.rs
+1 -1
View File
@@ -416,7 +416,7 @@ async fn list_user_usage(
SELECT COALESCE(SUM(c.duration_ms + 1000)/1000 , 0)::BIGINT executions
FROM v2_job_completed c JOIN v2_job j USING (id)
WHERE j.workspace_id = $1
AND j.kind NOT IN ('flow', 'flowpreview', 'flownode')
AND j.kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow')
AND j.permissioned_as_email = usr.email
AND now() - '1 week'::interval < j.created_at
) usage
backend/windmill-api/src/jobs.rs
+152 -41
View File
@@ -769,29 +769,75 @@ async fn list_selected_job_groups(
) -> error::Result<Response> {
let mut tx = user_db.begin(&authed).await?;
// Single-step-flows wrap either a script or a flow. The wrapped runnable type
// sits in raw_flow.modules under id='a' (id='main' is also tolerated), and the
// wrapped script's hash (when pinned) sits in the same module's value.hash.
// We project singlestepflow rows onto the script/flow grouping the rest of the
// query uses, and use the wrapped hash (when present) so the per-version
// schemas subquery resolves real schemas instead of returning nulls. A
// path-based schema fallback covers flow-wrapped singlestepflow (no version
// pinning) and any singlestepflow whose pinned hash has since been deleted.
let results = sqlx::query_scalar!(
r#"SELECT jsonb_build_object(
'kind', jb.kind,
'script_path', jb.runnable_path,
r#"WITH normalized AS (
SELECT
jb.id,
jb.workspace_id,
jb.runnable_path,
CASE
WHEN jb.kind IN ('flow', 'script') THEN jb.kind::text
WHEN jb.kind = 'singlestepflow' THEN
COALESCE(
(SELECT m->'value'->>'type'
FROM jsonb_array_elements(jb.raw_flow->'modules') m
WHERE m->>'id' IN ('a', 'main')
LIMIT 1),
'script'
)
ELSE NULL
END AS norm_kind,
COALESCE(
jb.runnable_id,
CASE WHEN jb.kind = 'singlestepflow' THEN
(SELECT ('x' || lpad(m->'value'->>'hash', 16, '0'))::bit(64)::bigint
FROM jsonb_array_elements(jb.raw_flow->'modules') m
WHERE m->>'id' IN ('a', 'main')
AND m->'value'->>'hash' IS NOT NULL
LIMIT 1)
END
) AS effective_hash
FROM v2_job jb
WHERE jb.kind IN ('flow', 'script', 'singlestepflow')
AND jb.workspace_id = $1 AND jb.id = ANY($2)
)
SELECT jsonb_build_object(
'kind', n.norm_kind,
'script_path', n.runnable_path,
'latest_schema', COALESCE(
(SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.workspace_id = $1 AND s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),
(SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.workspace_id = $1 AND flow.path = jb.runnable_path AND jb.kind = 'flow')
(SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.workspace_id = $1 AND s.path = n.runnable_path AND n.norm_kind = 'script' ORDER BY s.path, s.created_at DESC),
(SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.workspace_id = $1 AND flow.path = n.runnable_path AND n.norm_kind = 'flow')
),
'schemas', ARRAY(
SELECT jsonb_build_object(
'script_hash', LPAD(TO_HEX(COALESCE(s.hash, f.id)), 16, '0'),
'job_ids', ARRAY_AGG(DISTINCT j.id),
'schema', (ARRAY_AGG(COALESCE(s.schema, f.schema)))[1]
) FROM v2_job j
LEFT JOIN script s ON s.hash = j.runnable_id AND j.kind = 'script'
LEFT JOIN flow_version f ON f.id = j.runnable_id AND j.kind = 'flow'
WHERE j.id = ANY(ARRAY_AGG(jb.id))
'script_hash', CASE WHEN COALESCE(s.hash, f.id) IS NULL THEN NULL ELSE LPAD(TO_HEX(COALESCE(s.hash, f.id)), 16, '0') END,
'job_ids', ARRAY_AGG(DISTINCT n2.id),
'schema', COALESCE(
(ARRAY_AGG(COALESCE(s.schema, f.schema)))[1],
CASE WHEN n.norm_kind = 'script' THEN
(SELECT DISTINCT ON (s2.path) s2.schema FROM script s2 WHERE s2.workspace_id = $1 AND s2.path = n.runnable_path ORDER BY s2.path, s2.created_at DESC)
END,
CASE WHEN n.norm_kind = 'flow' THEN
(SELECT fv.schema FROM flow LEFT JOIN flow_version fv ON fv.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.workspace_id = $1 AND flow.path = n.runnable_path)
END
)
) FROM normalized n2
LEFT JOIN script s ON s.hash = n2.effective_hash AND n2.norm_kind = 'script'
LEFT JOIN flow_version f ON f.id = n2.effective_hash AND n2.norm_kind = 'flow'
WHERE n2.id = ANY(ARRAY_AGG(n.id))
GROUP BY COALESCE(s.hash, f.id)
)
) FROM v2_job jb
WHERE (jb.kind = 'flow' OR jb.kind = 'script')
AND jb.workspace_id = $1 AND jb.id = ANY($2)
GROUP BY jb.kind, jb.runnable_path"#,
) FROM normalized n
WHERE n.norm_kind IS NOT NULL
GROUP BY n.norm_kind, n.runnable_path"#,
&w_id,
&uuids
)
@@ -3106,11 +3152,11 @@ async fn get_flow_info_for_resume(job_id: Uuid, db: &DB) -> error::Result<(FlowI
q.suspend AS "suspend!",
j.runnable_path AS script_path,
j.permissioned_as_email AS email,
(ji.kind IN ('flow', 'flowpreview')) AS "is_flow_level!",
(ji.kind NOT IN ('flow', 'flowpreview') AND q.id = ji.id) AS "is_wac!"
(ji.kind IN ('flow', 'flowpreview', 'singlestepflow')) AS "is_flow_level!",
(ji.kind NOT IN ('flow', 'flowpreview', 'singlestepflow') AND q.id = ji.id) AS "is_wac!"
FROM job_info ji
JOIN v2_job_queue q ON q.id = CASE
WHEN ji.kind IN ('flow', 'flowpreview') THEN ji.id
WHEN ji.kind IN ('flow', 'flowpreview', 'singlestepflow') THEN ji.id
ELSE COALESCE(ji.parent_job, ji.id)
END
JOIN v2_job j ON j.id = q.id
@@ -3408,7 +3454,7 @@ pub async fn set_flow_user_state(
r#"
UPDATE v2_job_status f SET flow_status = JSONB_SET(flow_status, ARRAY['user_states'], JSONB_SET(COALESCE(flow_status->'user_states', '{}'::jsonb), ARRAY[$1], $2))
FROM v2_job j
WHERE f.id = $3 AND f.id = j.id AND j.workspace_id = $4 AND kind IN ('flow', 'flowpreview', 'flownode') RETURNING 1
WHERE f.id = $3 AND f.id = j.id AND j.workspace_id = $4 AND kind IN ('flow', 'flowpreview', 'flownode', 'singlestepflow') RETURNING 1
"#,
key,
value,
@@ -3699,23 +3745,74 @@ fn batch_rerun_jobs_inner(
tokio::spawn(async move {
let mut job_stream = sqlx::query_as!(
BatchReRunQueryReturnType,
r#"SELECT
j.id,
j.kind AS "kind: _",
COALESCE(s.path, f.path) AS "script_path!",
COALESCE(s.hash, f.id) AS "script_hash!: _",
r#"WITH norm AS (
SELECT
j.id, j.workspace_id, j.runnable_path, j.runnable_id, j.kind, j.args,
-- Project effective kind for dispatch: pass script/flow through;
-- for singlestepflow, read the wrapped runnable's type from
-- raw_flow.modules[id='a'].value.type (always 'script' or 'flow').
CASE
WHEN j.kind IN ('script', 'flow') THEN j.kind::text
WHEN j.kind = 'singlestepflow' THEN
COALESCE(
(SELECT m->'value'->>'type'
FROM jsonb_array_elements(j.raw_flow->'modules') m
WHERE m->>'id' = 'a'
LIMIT 1),
'script'
)
END AS norm_kind,
-- Pinned script hash for script-wrapped singlestepflow lives in
-- raw_flow.modules[id='a'].value.hash. Flow-wrapped doesn't pin a
-- version, so this is NULL there (Flow rerun pushes by path).
(CASE WHEN j.kind = 'singlestepflow' THEN
(SELECT ('x' || lpad(m->'value'->>'hash', 16, '0'))::bit(64)::bigint
FROM jsonb_array_elements(j.raw_flow->'modules') m
WHERE m->>'id' = 'a'
AND m->'value'->>'hash' IS NOT NULL
LIMIT 1)
END) AS ssf_hash
FROM v2_job j
WHERE j.id = ANY($1)
AND j.workspace_id = $2
AND j.kind IN ('script', 'flow', 'singlestepflow')
)
SELECT
n.id,
n.norm_kind::JOB_KIND AS "kind!: _",
COALESCE(s.path, f.path, n.runnable_path) AS "script_path!",
-- script_hash is unused on the Flow rerun path (path-based push), so
-- 0 is a safe placeholder when no version is pinned.
COALESCE(s.hash, f.id, n.ssf_hash, 0::bigint) AS "script_hash!: _",
COALESCE(jc.started_at, jq.scheduled_for, make_date(1970, 1, 1)) AS "scheduled_for!: _",
args AS input,
COALESCE(s.schema, f.schema) AS "schema: _"
FROM v2_job j
LEFT JOIN script s ON j.runnable_id = s.hash AND j.kind = 'script'
LEFT JOIN flow_version f ON j.runnable_id = f.id AND j.runnable_path = f.path AND j.kind = 'flow'
LEFT JOIN v2_job_completed jc ON jc.id = j.id
LEFT JOIN v2_job_queue jq ON jq.id = j.id
WHERE j.id = ANY($1)
AND j.workspace_id = $2
AND COALESCE(s.hash, f.id) IS NOT NULL
AND COALESCE(s.path, f.path) IS NOT NULL"#,
n.args AS input,
-- Pinned schema for script/flow; latest-by-path fallback for
-- singlestepflow so input_transforms still resolve at rerun time.
COALESCE(
s.schema,
f.schema,
(CASE WHEN n.kind = 'singlestepflow' AND n.norm_kind = 'script' THEN
(SELECT s2.schema FROM script s2
WHERE s2.workspace_id = $2 AND s2.path = n.runnable_path
ORDER BY s2.created_at DESC LIMIT 1)
END),
(CASE WHEN n.kind = 'singlestepflow' AND n.norm_kind = 'flow' THEN
(SELECT fv.schema FROM flow
LEFT JOIN flow_version fv ON fv.id = flow.versions[array_upper(flow.versions, 1)]
WHERE flow.workspace_id = $2 AND flow.path = n.runnable_path)
END)
) AS "schema: _"
FROM norm n
LEFT JOIN script s ON s.hash = n.runnable_id AND n.kind = 'script'
LEFT JOIN flow_version f ON f.id = n.runnable_id AND f.path = n.runnable_path AND n.kind = 'flow'
LEFT JOIN v2_job_completed jc ON jc.id = n.id
LEFT JOIN v2_job_queue jq ON jq.id = n.id
WHERE n.norm_kind IS NOT NULL
AND COALESCE(s.path, f.path, n.runnable_path) IS NOT NULL
AND (
n.kind = 'singlestepflow'
OR (COALESCE(s.hash, f.id) IS NOT NULL AND COALESCE(s.path, f.path) IS NOT NULL)
)"#,
&body.job_ids,
w_id
).fetch(&db);
@@ -3762,13 +3859,27 @@ async fn batch_rerun_handle_job(
let latest_schema;
let schema = if use_latest_version {
// Project singlestepflow's wrapped runnable type so the path-based schema
// lookup resolves it to the underlying script/flow's latest schema —
// without this, transforms silently no-op for singlestepflow reruns.
latest_schema = sqlx::query_scalar!(
r#"SELECT COALESCE(
(SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = jb.runnable_path AND jb.kind = 'script' ORDER BY s.path, s.created_at DESC),
(SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = jb.runnable_path AND jb.kind = 'flow')
) FROM v2_job jb
WHERE jb.id = $1 AND jb.workspace_id = $2
GROUP BY jb.kind, jb.runnable_path"#,
(SELECT DISTINCT ON (s.path) s.schema FROM script s WHERE s.path = norm.path AND s.workspace_id = $2 AND norm.kind = 'script' ORDER BY s.path, s.created_at DESC),
(SELECT flow_version.schema FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] WHERE flow.path = norm.path AND flow.workspace_id = $2 AND norm.kind = 'flow')
) FROM (
SELECT
jb.runnable_path AS path,
CASE
WHEN jb.kind IN ('script', 'flow') THEN jb.kind::text
WHEN jb.kind = 'singlestepflow' THEN COALESCE(
(SELECT m->'value'->>'type' FROM jsonb_array_elements(jb.raw_flow->'modules') m WHERE m->>'id' IN ('a', 'main') LIMIT 1),
'script'
)
END AS kind
FROM v2_job jb
WHERE jb.id = $1 AND jb.workspace_id = $2
) norm
GROUP BY norm.kind, norm.path"#,
&job.id,
&w_id
).fetch_optional(db).await?.flatten();
backend/windmill-queue/tests/schedule_push.rs
+1 -1
View File
@@ -1445,7 +1445,7 @@ mod schedule_push {
LEFT JOIN v2_job_status s USING (id)
WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null
AND q.scheduled_for <= now()
AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode')
AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode' OR j.kind = 'singlestepflow')
AND r.ping IS NOT NULL
AND r.ping < NOW() - ('60' || ' seconds')::interval
AND q.canceled_by IS NULL
backend/windmill-trigger/src/global_handler.rs
+20 -6
View File
@@ -84,6 +84,7 @@ struct JobWithArgs {
id: Uuid,
args: Option<sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
created_at: chrono::DateTime<chrono::Utc>,
kind: windmill_common::jobs::JobKind,
}
#[derive(Deserialize, Serialize, Default)]
@@ -113,7 +114,8 @@ pub async fn resume_suspended_trigger_jobs(
SELECT
id,
args as "args: _",
created_at
created_at,
kind AS "kind: _"
FROM v2_job
WHERE workspace_id = $1
AND (
@@ -140,7 +142,8 @@ pub async fn resume_suspended_trigger_jobs(
SELECT
id,
args as "args: _",
created_at
created_at,
kind AS "kind: _"
FROM v2_job
WHERE workspace_id = $1
AND (
@@ -167,10 +170,21 @@ pub async fn resume_suspended_trigger_jobs(
// If job was created before trigger was edited, simply update it to unsuspend
// instead of deleting and repushing
if job.created_at > trigger.edited_at {
let job_kind = if trigger.is_flow {
windmill_common::jobs::JobKind::Flow
} else {
windmill_common::jobs::JobKind::Script
// Map the placeholder unassigned kind back to its assigned counterpart so
// singlestepflow wrappers (retry/error_handler/skip_handler) keep their
// flow-orchestrator identity.
let job_kind = match job.kind {
windmill_common::jobs::JobKind::UnassignedSinglestepFlow => {
windmill_common::jobs::JobKind::SingleStepFlow
}
windmill_common::jobs::JobKind::UnassignedFlow => {
windmill_common::jobs::JobKind::Flow
}
windmill_common::jobs::JobKind::UnassignedScript => {
windmill_common::jobs::JobKind::Script
}
_ if trigger.is_flow => windmill_common::jobs::JobKind::Flow,
_ => windmill_common::jobs::JobKind::Script,
};
sqlx::query!(
frontend/src/lib/components/runs/useJobsLoader.svelte.ts
+5 -10
View File
@@ -30,17 +30,12 @@ export function computeJobKinds(jobKindsCat: string | null): string {
} else if (jobKindsCat == 'deploymentcallbacks') {
let kinds: CompletedJob['job_kind'][] = ['deploymentcallback']
return kinds.join(',')
} else if (jobKindsCat == 'runs') {
let kinds: CompletedJob['job_kind'][] = ['script', 'flow', 'singlestepflow']
return kinds.join(',')
} else {
let kinds: CompletedJob['job_kind'][] = [
'script',
'flow',
'flowscript',
'flownode',
'appscript'
]
// Default mirrors the explicit 'runs' category — top-level scripts, flows,
// and single-step flows. flowscript/flownode/appscript are intermediate
// flow children with non-null parent_job, and the loader pairs this with
// hasNullParent: true, so they would never match here anyway.
let kinds: CompletedJob['job_kind'][] = ['script', 'flow', 'singlestepflow']
return kinds.join(',')
}
}