diff --git a/backend/windmill-api-inputs/src/lib.rs b/backend/windmill-api-inputs/src/lib.rs index 8764736e83..8e33789c3d 100644 --- a/backend/windmill-api-inputs/src/lib.rs +++ b/backend/windmill-api-inputs/src/lib.rs @@ -73,14 +73,6 @@ impl Display for RunnableType { } impl RunnableType { - fn job_kind(&self) -> JobKind { - match self { - RunnableType::ScriptHash => JobKind::Script, - RunnableType::ScriptPath => JobKind::Script, - RunnableType::FlowPath => JobKind::Flow, - } - } - fn column_name(&self) -> &'static str { match self { RunnableType::ScriptHash => "runnable_id", @@ -159,6 +151,28 @@ async fn get_input_history( "AND parent_job IS NULL" }; + // A scheduled runnable with a dynamic-skip handler (or a scheduled script with + // native retry) is wrapped in a synthetic `singlestepflow`, so its runs land under + // that kind rather than `flow`/`script` (see windmill-queue schedule.rs). Such a + // wrapper holds either a script or a flow, and the two may share a runnable_path, so + // match only the wrapped kind that belongs to the queried runnable. The wrapped type + // lives in raw_flow.modules[id in ('a','main')].value.type (mirrors the projection in + // windmill-api jobs.rs). runnable_id is NULL on these rows, so ScriptHash never matches. + let singlestepflow_filter = match r.runnable_type { + RunnableType::FlowPath => { + "AND (kind <> 'singlestepflow' OR EXISTS (\ + SELECT 1 FROM jsonb_array_elements(v2_job.raw_flow->'modules') m \ + WHERE m->>'id' IN ('a', 'main') AND m->'value'->>'type' = 'flow'))" + } + RunnableType::ScriptPath => { + "AND (kind <> 'singlestepflow' OR EXISTS (\ + SELECT 1 FROM jsonb_array_elements(v2_job.raw_flow->'modules') m \ + WHERE m->>'id' IN ('a', 'main') \ + AND COALESCE(m->'value'->>'type', 'script') <> 'flow'))" + } + RunnableType::ScriptHash => "", + }; + // Two-step approach: first fetch 2*(per_page+offset) rows using created_at ordering // (which leverages the ix_job_root_job_index_by_path_2 index on v2_job), then sort // the small result set by completed_at. This works because created_at and completed_at @@ -172,7 +186,7 @@ async fn get_input_history( kind IN ('preview', 'flowpreview') as is_preview \ FROM v2_job JOIN v2_job_completed USING (id) \ WHERE v2_job.workspace_id = $3 AND {} = $1 AND kind = any($2) \ - AND v2_job.script_entrypoint_override IS NULL \ + AND v2_job.script_entrypoint_override IS NULL {singlestepflow_filter} \ {args_query} AND v2_job_completed.status != 'skipped' {include_non_root} \ ORDER BY v2_job.created_at DESC LIMIT $4\ ) t ORDER BY completed_at DESC LIMIT $5 OFFSET $6", @@ -186,14 +200,32 @@ async fn get_input_history( _ => query.bind(&r.runnable_id), }; - let job_kinds = match r.runnable_type.job_kind() { - kind @ JobKind::Script if g.include_preview.unwrap_or(false) => { - vec![kind, JobKind::Preview] + // Include SingleStepFlow so scheduled runs surface (see `singlestepflow_filter` + // above, which restricts it to the wrapped kind matching the runnable). ScriptHash + // is omitted: those wrappers carry no runnable_id, so it can never match. + let include_preview = g.include_preview.unwrap_or(false); + let job_kinds = match r.runnable_type { + RunnableType::ScriptHash => { + let mut kinds = vec![JobKind::Script]; + if include_preview { + kinds.push(JobKind::Preview); + } + kinds } - kind @ JobKind::Flow if g.include_preview.unwrap_or(false) => { - vec![kind, JobKind::FlowPreview] + RunnableType::ScriptPath => { + let mut kinds = vec![JobKind::Script, JobKind::SingleStepFlow]; + if include_preview { + kinds.push(JobKind::Preview); + } + kinds + } + RunnableType::FlowPath => { + let mut kinds = vec![JobKind::Flow, JobKind::SingleStepFlow]; + if include_preview { + kinds.push(JobKind::FlowPreview); + } + kinds } - kind => vec![kind], }; let rows = query diff --git a/backend/windmill-api-integration-tests/tests/inputs.rs b/backend/windmill-api-integration-tests/tests/inputs.rs index 6f806efdb5..a3f0610e55 100644 --- a/backend/windmill-api-integration-tests/tests/inputs.rs +++ b/backend/windmill-api-integration-tests/tests/inputs.rs @@ -74,3 +74,94 @@ async fn test_inputs_endpoints(db: Pool) -> anyhow::Result<()> { Ok(()) } + +// A scheduled runnable with a dynamic-skip handler (or a scheduled script with native +// retry) runs as a `singlestepflow`, not `flow`/`script` (see windmill-queue schedule.rs). +// Both a flow's and a script's history must surface their own singlestepflow runs, but a +// script and flow may share a path, so each side must match only the wrapped kind that +// belongs to it — the other must not leak in. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_input_history_singlestepflow_flow_vs_script( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + + // Both rows share runnable_path 'f/test/scheduled'. + let flow_job = insert_singlestepflow(&db, "flow").await?; + let script_job = insert_singlestepflow(&db, "script").await?; + + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let base = format!("http://localhost:{port}/api/w/test-workspace/inputs"); + + let history_ids = |runnable_type: &'static str| { + let base = base.clone(); + async move { + let resp = authed(client().get(format!( + "{base}/history?runnable_id=f/test/scheduled&runnable_type={runnable_type}" + ))) + .send() + .await?; + let status = resp.status().as_u16(); + let body = resp.text().await?; + assert_2xx(status, &body, "GET /inputs/history"); + let inputs: Vec = serde_json::from_str(&body)?; + anyhow::Ok( + inputs + .iter() + .filter_map(|i| i.get("id").and_then(|v| v.as_str()).map(String::from)) + .collect::>(), + ) + } + }; + + let flow_hist = history_ids("FlowPath").await?; + assert!( + flow_hist.contains(&flow_job.to_string()), + "flow-wrapped singlestepflow missing from flow history: {flow_hist:?}", + ); + assert!( + !flow_hist.contains(&script_job.to_string()), + "script-wrapped singlestepflow leaked into flow history: {flow_hist:?}", + ); + + let script_hist = history_ids("ScriptPath").await?; + assert!( + script_hist.contains(&script_job.to_string()), + "script-wrapped singlestepflow missing from script history: {script_hist:?}", + ); + assert!( + !script_hist.contains(&flow_job.to_string()), + "flow-wrapped singlestepflow leaked into script history: {script_hist:?}", + ); + + Ok(()) +} + +// Insert a completed root singlestepflow at path f/test/scheduled wrapping `wrapped_type` +// ('flow' or 'script') as its single module — mirrors the schedule.rs wrapper shape. +async fn insert_singlestepflow( + db: &Pool, + wrapped_type: &str, +) -> anyhow::Result { + let id = uuid::Uuid::new_v4(); + let raw_flow = json!({ "modules": [{ "id": "a", "value": { "type": wrapped_type } }] }); + sqlx::query( + "INSERT INTO v2_job (id, workspace_id, tag, created_by, permissioned_as, \ + permissioned_as_email, kind, runnable_path, raw_flow, same_worker, visible_to_owner) \ + VALUES ($1, 'test-workspace', 'flow', 'test-user', 'u/test-user', \ + 'test@windmill.dev', 'singlestepflow', 'f/test/scheduled', $2, false, true)", + ) + .bind(id) + .bind(sqlx::types::Json(&raw_flow)) + .execute(db) + .await?; + sqlx::query( + "INSERT INTO v2_job_completed (id, workspace_id, duration_ms, deleted, status) \ + VALUES ($1, 'test-workspace', 1, false, 'success')", + ) + .bind(id) + .execute(db) + .await?; + Ok(id) +}