diff --git a/backend/.sqlx/query-bd000b7ab6438826093ba556af819741b493a91babaeb103852d3912f520719f.json b/backend/.sqlx/query-074dc85ffb596585eb99336ffa34fa29fc0a1aff4d10d6fe4aba4b5357afb0f3.json similarity index 92% rename from backend/.sqlx/query-bd000b7ab6438826093ba556af819741b493a91babaeb103852d3912f520719f.json rename to backend/.sqlx/query-074dc85ffb596585eb99336ffa34fa29fc0a1aff4d10d6fe4aba4b5357afb0f3.json index 42b90ef229..a932976a2f 100644 --- a/backend/.sqlx/query-bd000b7ab6438826093ba556af819741b493a91babaeb103852d3912f520719f.json +++ b/backend/.sqlx/query-074dc85ffb596585eb99336ffa34fa29fc0a1aff4d10d6fe4aba4b5357afb0f3.json @@ -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\", 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\", 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" } diff --git a/backend/.sqlx/query-0c89ef278782f5a72b0b07ab3ba0edc487f03edd61936fcf77dee93fb22839ea.json b/backend/.sqlx/query-0c89ef278782f5a72b0b07ab3ba0edc487f03edd61936fcf77dee93fb22839ea.json deleted file mode 100644 index e74e12f501..0000000000 --- a/backend/.sqlx/query-0c89ef278782f5a72b0b07ab3ba0edc487f03edd61936fcf77dee93fb22839ea.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json b/backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json deleted file mode 100644 index 41520bfc88..0000000000 --- a/backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-dbc7e74e259b502e700491ee0248e0c9c8c61e1bf609be60ac5dc5d438189353.json b/backend/.sqlx/query-3e68b73a1c2fdcb2ef538de1addbb096921e26fed398d5c0966b60503ca97fdb.json similarity index 68% rename from backend/.sqlx/query-dbc7e74e259b502e700491ee0248e0c9c8c61e1bf609be60ac5dc5d438189353.json rename to backend/.sqlx/query-3e68b73a1c2fdcb2ef538de1addbb096921e26fed398d5c0966b60503ca97fdb.json index 5791090fc1..a40cc12148 100644 --- a/backend/.sqlx/query-dbc7e74e259b502e700491ee0248e0c9c8c61e1bf609be60ac5dc5d438189353.json +++ b/backend/.sqlx/query-3e68b73a1c2fdcb2ef538de1addbb096921e26fed398d5c0966b60503ca97fdb.json @@ -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" } diff --git a/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json b/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json new file mode 100644 index 0000000000..2a6930755a --- /dev/null +++ b/backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json @@ -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" +} diff --git a/backend/.sqlx/query-6cda14bc33e3144b78b2147ac1153c516b5962b5648ae07c247fbd57b1b16ada.json b/backend/.sqlx/query-6cda14bc33e3144b78b2147ac1153c516b5962b5648ae07c247fbd57b1b16ada.json new file mode 100644 index 0000000000..e8f746ded0 --- /dev/null +++ b/backend/.sqlx/query-6cda14bc33e3144b78b2147ac1153c516b5962b5648ae07c247fbd57b1b16ada.json @@ -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" +} diff --git a/backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json b/backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json deleted file mode 100644 index b705130ffe..0000000000 --- a/backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-9f978c08b3a0150bb6f6ef71a6c4a1d8b126495a514ea4c0e791216aef55dc38.json b/backend/.sqlx/query-9f978c08b3a0150bb6f6ef71a6c4a1d8b126495a514ea4c0e791216aef55dc38.json new file mode 100644 index 0000000000..6819848a06 --- /dev/null +++ b/backend/.sqlx/query-9f978c08b3a0150bb6f6ef71a6c4a1d8b126495a514ea4c0e791216aef55dc38.json @@ -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" +} diff --git a/backend/.sqlx/query-a1ab1f23f49496f745d89d6c33a91c6b693afc4477a634aa84f81db91f9e03a4.json b/backend/.sqlx/query-a1ab1f23f49496f745d89d6c33a91c6b693afc4477a634aa84f81db91f9e03a4.json deleted file mode 100644 index 0c8209f0d0..0000000000 --- a/backend/.sqlx/query-a1ab1f23f49496f745d89d6c33a91c6b693afc4477a634aa84f81db91f9e03a4.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd.json b/backend/.sqlx/query-a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665.json similarity index 69% rename from backend/.sqlx/query-edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd.json rename to backend/.sqlx/query-a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665.json index f67e52ab6b..2358af807f 100644 --- a/backend/.sqlx/query-edd57b3d59ddc21b99212b5fabd4a8b793c4ae618a04220b3609c8c3c168f8fd.json +++ b/backend/.sqlx/query-a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665.json @@ -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" } diff --git a/backend/.sqlx/query-aba492cb21cbbd514959a16ca02ab3b62efb138edd968d90f10f9043ae9a7c62.json b/backend/.sqlx/query-aba492cb21cbbd514959a16ca02ab3b62efb138edd968d90f10f9043ae9a7c62.json new file mode 100644 index 0000000000..291e8279fa --- /dev/null +++ b/backend/.sqlx/query-aba492cb21cbbd514959a16ca02ab3b62efb138edd968d90f10f9043ae9a7c62.json @@ -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" +} diff --git a/backend/.sqlx/query-b32d8b364001f5d54c1c4e564aac6b49d514771191a004c21c2e0d042f9d6f4d.json b/backend/.sqlx/query-b32d8b364001f5d54c1c4e564aac6b49d514771191a004c21c2e0d042f9d6f4d.json new file mode 100644 index 0000000000..9704af007f --- /dev/null +++ b/backend/.sqlx/query-b32d8b364001f5d54c1c4e564aac6b49d514771191a004c21c2e0d042f9d6f4d.json @@ -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" +} diff --git a/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json b/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json new file mode 100644 index 0000000000..045d470de5 --- /dev/null +++ b/backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json @@ -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" +} diff --git a/backend/.sqlx/query-e2905bca184696a80357d8e4126832a902b3088d91fbbb858f6c0aa9de8a5ff7.json b/backend/.sqlx/query-e2905bca184696a80357d8e4126832a902b3088d91fbbb858f6c0aa9de8a5ff7.json deleted file mode 100644 index cab697d254..0000000000 --- a/backend/.sqlx/query-e2905bca184696a80357d8e4126832a902b3088d91fbbb858f6c0aa9de8a5ff7.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-e8597d72fc73446d6ac1e76dffcbc6a1c77e754825b59ca4bc79c4e675bed8e5.json b/backend/.sqlx/query-e8597d72fc73446d6ac1e76dffcbc6a1c77e754825b59ca4bc79c4e675bed8e5.json new file mode 100644 index 0000000000..b67b639a79 --- /dev/null +++ b/backend/.sqlx/query-e8597d72fc73446d6ac1e76dffcbc6a1c77e754825b59ca4bc79c4e675bed8e5.json @@ -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" +} diff --git a/backend/.sqlx/query-750f09238caf114d01f752a145fd4289f04f3961d5035d42f3fd0bd2f87eed1a.json b/backend/.sqlx/query-f6eff53e00b33310bd9626b44b72af91a2da474ddf622f15f4f618dfcc7f1c39.json similarity index 82% rename from backend/.sqlx/query-750f09238caf114d01f752a145fd4289f04f3961d5035d42f3fd0bd2f87eed1a.json rename to backend/.sqlx/query-f6eff53e00b33310bd9626b44b72af91a2da474ddf622f15f4f618dfcc7f1c39.json index c72f20178c..6829ce231d 100644 --- a/backend/.sqlx/query-750f09238caf114d01f752a145fd4289f04f3961d5035d42f3fd0bd2f87eed1a.json +++ b/backend/.sqlx/query-f6eff53e00b33310bd9626b44b72af91a2da474ddf622f15f4f618dfcc7f1c39.json @@ -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" } diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index ef3fb99696..59a2f47ed0 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -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 diff --git a/backend/tests/batch_rerun.rs b/backend/tests/batch_rerun.rs new file mode 100644 index 0000000000..b750379834 --- /dev/null +++ b/backend/tests/batch_rerun.rs @@ -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, 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, retry: Option) -> 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, + payload: JobPayload, + args: Vec<(&str, serde_json::Value)>, +) -> anyhow::Result { + 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>> { + 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, 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) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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, + latest_schema: Option, +} + +#[derive(serde::Deserialize)] +struct SchemaEntry { + script_hash: Option, + schema: Option, + job_ids: Vec, +} + +async fn list_groups( + client: &windmill_api_client::Client, + job_ids: &[Uuid], +) -> anyhow::Result> { + 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) -> 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) -> 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(()) +} diff --git a/backend/tests/fixtures/batch_rerun.sql b/backend/tests/fixtures/batch_rerun.sql new file mode 100644 index 0000000000..85ba377f5d --- /dev/null +++ b/backend/tests/fixtures/batch_rerun.sql @@ -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() +); diff --git a/backend/windmill-api-users/src/users.rs b/backend/windmill-api-users/src/users.rs index 382a69201b..e4540ca903 100644 --- a/backend/windmill-api-users/src/users.rs +++ b/backend/windmill-api-users/src/users.rs @@ -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 diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 1ed5c7c9e4..efe03f73dd 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -769,29 +769,75 @@ async fn list_selected_job_groups( ) -> error::Result { 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(); diff --git a/backend/windmill-queue/tests/schedule_push.rs b/backend/windmill-queue/tests/schedule_push.rs index 8b5394b3cb..c0e09003f0 100644 --- a/backend/windmill-queue/tests/schedule_push.rs +++ b/backend/windmill-queue/tests/schedule_push.rs @@ -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 diff --git a/backend/windmill-trigger/src/global_handler.rs b/backend/windmill-trigger/src/global_handler.rs index 84058f5baa..def5a5457d 100644 --- a/backend/windmill-trigger/src/global_handler.rs +++ b/backend/windmill-trigger/src/global_handler.rs @@ -84,6 +84,7 @@ struct JobWithArgs { id: Uuid, args: Option>>>, created_at: chrono::DateTime, + 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!( diff --git a/frontend/src/lib/components/runs/useJobsLoader.svelte.ts b/frontend/src/lib/components/runs/useJobsLoader.svelte.ts index 74578a037c..b75355a6cb 100644 --- a/frontend/src/lib/components/runs/useJobsLoader.svelte.ts +++ b/frontend/src/lib/components/runs/useJobsLoader.svelte.ts @@ -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(',') } }