From 9e89bc95e3bb73354a96fd6f539b09d553974e15 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@rubenfiszel.com>
Date: Fri, 8 Sep 2023 17:32:20 +0200
Subject: [PATCH] improve benchmarks

---
 backend/windmill-api/openapi.yaml  | 21 +++++++++++++++++++++
 backend/windmill-api/src/jobs.rs   | 16 ++++++++++++++++
 backend/windmill-queue/src/jobs.rs | 21 +++++++++++++++------
 benchmarks/main.ts                 | 28 +++++++++++++++++++---------
 4 files changed, 71 insertions(+), 15 deletions(-)

diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml
index db49e5c731..2480c13d9a 100644
--- a/backend/windmill-api/openapi.yaml
+++ b/backend/windmill-api/openapi.yaml
@@ -3954,6 +3954,27 @@ paths:
                 required:
                   - database_length
 
+  /w/{workspace}/jobs/completed/count:
+    get:
+      summary: get completed count
+      operationId: getCompletedCount
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      responses:
+        "200":
+          description: completed count
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  database_length:
+                    type: integer
+                required:
+                  - database_length
+
   /w/{workspace}/jobs/queue/cancel_all:
     post:
       summary: cancel all jobs
diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs
index 12fd818837..7134c021d7 100644
--- a/backend/windmill-api/src/jobs.rs
+++ b/backend/windmill-api/src/jobs.rs
@@ -110,6 +110,7 @@ pub fn workspaced_service() -> Router {
         .route("/queue/list", get(list_queue_jobs))
         .route("/queue/count", get(count_queue_jobs))
         .route("/queue/cancel_all", post(cancel_all))
+        .route("/completed/count", get(count_completed_jobs))
         .route(
             "/completed/list",
             get(list_completed_jobs).layer(cors.clone()),
@@ -706,6 +707,21 @@ async fn count_queue_jobs(
     ))
 }
 
+async fn count_completed_jobs(
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+) -> error::JsonResult<QueueStats> {
+    Ok(Json(
+        sqlx::query_as!(
+            QueueStats,
+            "SELECT coalesce(COUNT(*), 0) as \"database_length!\" FROM completed_job WHERE workspace_id = $1",
+            w_id
+        )
+        .fetch_one(&db)
+        .await?,
+    ))
+}
+
 async fn list_jobs(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs
index 1b2b1f3c98..ac1918042b 100644
--- a/backend/windmill-queue/src/jobs.rs
+++ b/backend/windmill-queue/src/jobs.rs
@@ -1353,12 +1353,21 @@ pub async fn delete_job<'c, R: rsmq_async::RsmqConnection + Clone + Send>(
         w_id,
         job_id
     )
-    .fetch_one(&mut tx)
-    .await
-    .map_err(|e| Error::InternalErr(format!("Error during deletion of job {job_id}: {e}")))?
-    .unwrap_or(0)
-        == 1;
-    tracing::debug!("Job {job_id} deleted: {job_removed}");
+    .fetch_optional(&mut tx)
+    .await;
+
+    if let Err(job_removed) = job_removed {
+        tracing::error!(
+            "Job {job_id} could not be deleted: {job_removed}. This is not necessarily an error, as the job might have been deleted by another process such as in the case of cancelling"
+        );
+    } else {
+        let job_removed = job_removed.unwrap().flatten().unwrap_or(0);
+        if job_removed != 1 {
+            tracing::error!("Job {job_id} could not be deleted, returned not 1: {job_removed}. This is not necessarily an error, as the job might have been deleted by another process such as in the case of cancelling");
+        }
+    }
+
+    tracing::debug!("Job {job_id} deleted");
     Ok(tx)
 }
 
diff --git a/benchmarks/main.ts b/benchmarks/main.ts
index 55e10083e8..9e0d5b359d 100644
--- a/benchmarks/main.ts
+++ b/benchmarks/main.ts
@@ -242,20 +242,28 @@ export async function main({
   const jobsSent = Array(num_workers).fill(0);
   const enc = (s: string) => new TextEncoder().encode(s);
 
+  async function getQueueCount() {
+    return (
+      await (
+        await fetch(
+          config.server + "/api/w/" + config.workspace_id + "/jobs/queue/count",
+          { headers: { ["Authorization"]: "Bearer " + config.token } }
+        )
+      ).json()
+    ).database_length;
+  }
+
+  const initial_queue_length = await getQueueCount();
+
+  console.log("Initial queue length:", initial_queue_length);
+
   const updateState = setInterval(async () => {
     const elapsed = start ? Math.ceil((Date.now() - start) / 1000) : 0;
     const sum = jobsSent.reduce((a, b) => a + b, 0);
     let queue_length = -1;
     while (queue_length === -1) {
       try {
-        queue_length = (
-          await (
-            await fetch(
-              host + "/api/w/" + config.workspace_id + "/jobs/queue/count",
-              { headers: { ["Authorization"]: "Bearer " + config.token } }
-            )
-          ).json()
-        ).database_length;
+        queue_length = await getQueueCount();
       } catch (e) {
         console.log(
           `queue count not reachable. waiting...                                                           `
@@ -268,7 +276,9 @@ export async function main({
       enc(
         `elapsed: ${elapsed}/${seconds} | jobs sent: ${JSON.stringify(
           jobsSent
-        )} (sum: ${sum} thr: ${(sum / elapsed).toFixed(
+        )} (sum: ${sum} thr: ${(sum / elapsed).toFixed(2)}) - processed (sum: ${
+          sum - queue_length
+        } thr: ${((sum - queue_length) / elapsed).toFixed(
           2
         )}) | queue: ${queue_length}                          \r`
       )