perf: count completed jobs on the v2_job_completed index with a timeout (#11211)

* perf: count completed jobs on the v2_job_completed index with a timeout

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

* test: cover the count_jobs tags filter on completed jobs

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

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

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<Postgres>) -> 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::<i64>().await?, expected, "{query}");
+    }
+
+    Ok(())
+}

backend/windmill-api/src/jobs.rs

@@ -4369,12 +4369,12 @@ async fn count_completed_jobs_detail(
     Query(query): Query<CountCompletedJobsQuery>,
 ) -> error::JsonResult<i64> {
     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::<Vec<_>>(),
@@ -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<UserDB>,
@@ -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<UnifiedJob> = 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?;