diff --git a/backend/tests/list_jobs.rs b/backend/tests/list_jobs.rs index 82a9e197cc..29a49c1610 100644 --- a/backend/tests/list_jobs.rs +++ b/backend/tests/list_jobs.rs @@ -1199,3 +1199,63 @@ async fn test_wm_labels_from_result_merged_with_static_labels( Ok(()) } + +/// `tag` lives only on `v2_job`, which count_jobs joins only when `tags` is set. +#[sqlx::test(fixtures("base"))] +async fn test_count_completed_jobs_tags_filter(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let client = windmill_api_client::create_client( + &format!("http://localhost:{port}"), + "SECRET_TOKEN".to_string(), + ); + + for (ws, tag, status) in [ + ("test-workspace", "deno", "success"), + ("test-workspace", "deno", "failure"), + ("test-workspace", "python3", "success"), + ("other-workspace", "deno", "success"), + ] { + let id = uuid::Uuid::new_v4(); + sqlx::query("INSERT INTO v2_job (id, workspace_id, tag) VALUES ($1, $2, $3)") + .bind(id) + .bind(ws) + .bind(tag) + .execute(&db) + .await?; + sqlx::query( + "INSERT INTO v2_job_completed (id, workspace_id, status, duration_ms) VALUES ($1, $2, $3::job_status, 0)", + ) + .bind(id) + .bind(ws) + .bind(status) + .execute(&db) + .await?; + } + + for (query, expected) in [ + ("", 3), + ("tags=deno", 2), + ("tags=deno&success=true", 1), + ("tags=deno,python3&completed_after_s_ago=3600", 3), + ] { + let response = client + .client() + .get(format!( + "{}/w/test-workspace/jobs/completed/count_jobs?{query}", + client.baseurl() + )) + .send() + .await?; + assert!( + response.status().is_success(), + "{query}: {}", + response.text().await? + ); + assert_eq!(response.json::().await?, expected, "{query}"); + } + + Ok(()) +} diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 4149292200..eb77cb6e36 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -4369,12 +4369,12 @@ async fn count_completed_jobs_detail( Query(query): Query, ) -> error::JsonResult { let mut sqlb = SqlBuilder::select_from("v2_job_completed"); - //FOR RLS - sqlb.join("v2_job USING (id)"); sqlb.field("COUNT(*) as count"); + // Filtering on v2_job.workspace_id instead would keep the planner off + // ix_job_workspace_id_completed_at_all and scan the whole retention window. if !(w_id == "admins" && query.all_workspaces.unwrap_or(false)) { - sqlb.and_where_eq("v2_job.workspace_id", "?".bind(&w_id)); + sqlb.and_where_eq("v2_job_completed.workspace_id", "?".bind(&w_id)); } if let Some(after_s_ago) = query.completed_after_s_ago { @@ -4393,6 +4393,7 @@ async fn count_completed_jobs_detail( } if let Some(tags) = query.tags { + sqlb.join("v2_job USING (id)"); sqlb.and_where_in( "v2_job.tag", &tags.split(",").map(|t| quote(t)).collect::>(), @@ -4400,7 +4401,19 @@ async fn count_completed_jobs_detail( } let sql = sqlb.sql()?; - let stats = sqlx::query_scalar::<_, i64>(&sql).fetch_one(&db).await?; + let mut tx = db.begin().await?; + set_list_jobs_statement_timeout(&mut tx).await?; + let stats = sqlx::query_scalar::<_, i64>(&sql) + .fetch_one(&mut *tx) + .await + .map_err(|e| { + list_jobs_timeout_error( + e, + "Counting completed jobs", + "Lower completed_after_s_ago or narrow the filters.", + ) + })?; + tx.commit().await?; Ok(Json(stats)) } @@ -4429,6 +4442,33 @@ lazy_static::lazy_static! { .unwrap_or(30); } +/// A client that gives up does not cancel its query, so without this bound every retry of a +/// slow filter stacks another scan running until the connection-wide 5min timeout. +async fn set_list_jobs_statement_timeout(tx: &mut Transaction<'_, Postgres>) -> error::Result<()> { + let timeout_secs = *LIST_JOBS_STATEMENT_TIMEOUT_SECS; + if timeout_secs > 0 { + sqlx::query(&format!("SET LOCAL statement_timeout = '{timeout_secs}s'")) + .execute(&mut **tx) + .await?; + } + Ok(()) +} + +fn list_jobs_timeout_error(e: sqlx::Error, action: &str, hint: &str) -> Error { + let timeout_secs = *LIST_JOBS_STATEMENT_TIMEOUT_SECS; + match e { + sqlx::Error::Database(ref db_err) + if timeout_secs > 0 && db_err.code().as_deref() == Some("57014") => + { + Error::Generic( + StatusCode::BAD_REQUEST, + format!("{action} took more than {timeout_secs}s and was stopped. {hint}"), + ) + } + e => e.into(), + } +} + async fn list_jobs( authed: ApiAuthed, Extension(user_db): Extension, @@ -4545,32 +4585,14 @@ async fn list_jobs( }; // tracing::info!("sql: {}", &sql); let mut tx: Transaction<'_, Postgres> = user_db.begin(&authed).await?; - - // A client that gives up does not cancel its query, so without this bound every retry of a - // slow filter stacks another scan running until the connection-wide 5min timeout. - let timeout_secs = *LIST_JOBS_STATEMENT_TIMEOUT_SECS; - if timeout_secs > 0 { - sqlx::query(&format!("SET LOCAL statement_timeout = '{timeout_secs}s'")) - .execute(&mut *tx) - .await?; - } + set_list_jobs_statement_timeout(&mut tx).await?; let jobs: Vec = sqlx::query_as(&sql) .fetch_all(&mut *tx) .warn_after_seconds_with_sql(5, format!("list_jobs: {}", sql)) .await - .map_err(|e| match e { - sqlx::Error::Database(ref db_err) - if timeout_secs > 0 && db_err.code().as_deref() == Some("57014") => - { - Error::Generic( - StatusCode::BAD_REQUEST, - format!( - "Listing jobs took more than {timeout_secs}s and was stopped. Set a start date or narrow the filters." - ), - ) - } - e => e.into(), + .map_err(|e| { + list_jobs_timeout_error(e, "Listing jobs", "Set a start date or narrow the filters.") })?; tx.commit().await?;