From cbc0b9d8cf455e32714f8ab0078b26e4f3fefe5e Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 26 May 2026 11:47:45 +0000 Subject: [PATCH] fix(queue): duration-weighted workspace fairness signal (#9329) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(queue): bump EE ref to include worker_ping fairness signal The current ee-repo-ref.txt pointed to 31cda7c (an unrelated merge commit on the asset-graph-view-ee branch) instead of ddc9e80, which contains the workspace-fairness fix that switches the active-share signal from v2_job_queue.running=true to worker_ping. As a result cloud was still computing overload off the legacy signal, so a workspace with many in-flight/suspended flows (lancom01-prod, with 799 suspended flows × 3 v2_job_queue bookkeeping rows each = 2397 running-true rows) was flagged as 95% of cluster activity despite consuming zero worker slots. Bumping to ddc9e80 picks up the worker_ping-based signal, which naturally excludes (a) suspended jobs (no worker pinging them), (b) zombie running-rows from dead workers, and (c) flow/flownode orchestration rows that never run on a worker in the first place. * test(queue): seed v2_job rows + realistic durations for fairness helpers The new duration-weighted fairness algorithm joins v2_job_queue and v2_job_completed to v2_job for the `kind` filter (excluding flow bookkeeping) and reads `duration_ms` for the completed contribution. Update the test helpers to mirror that schema: * `insert_completed` now inserts a matching v2_job row (kind=script) and writes `duration_ms = 1000` with a 1-second [started_at, completed_at] interval, so each completed row contributes ~1 worker-second when fully inside the refresh window. * `insert_queued` likewise pre-inserts v2_job, sets `started_at` to NOW() - 1s when running=true (so running rows contribute ~1 worker-second by the time the refresh runs), and seeds v2_job_runtime.ping so the running side accrues real-time worker seconds (the algorithm bounds end-of-interval by ping). The zombie/suspended insert helpers are intentionally left without v2_job rows — the new algorithm's INNER JOIN excludes them, so they still correctly contribute zero worker-seconds. * chore(queue): bump EE ref to duration-weighted fairness algorithm Companion to windmill-ee-private#: switch the EE workspace fairness aggregation from a count-based UNION (worker_ping snapshot + v2_job_completed count) to a worker-seconds aggregation sourced directly from v2_job_queue and v2_job_completed, with kind/suspend filters mirroring handle_zombie_jobs and per-row defenses against zombie inflation on both halves. * chore(queue): bump EE ref for fairness perf fix (inline window_start) * chore(queue): bump EE ref for fairness perf rewrite (driver-side flip) * update ee ref --- backend/ee-repo-ref.txt | 2 +- backend/tests/workspace_fairness.rs | 55 +++++++++++++++++++++++------ 2 files changed, 46 insertions(+), 11 deletions(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index c21c4dbea8..761de56ebb 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -31cda7cea811108e5004842ec5b5467ca48181cd \ No newline at end of file +327d23f7438968a21bac9fd42e7f6f027c61477c \ No newline at end of file diff --git a/backend/tests/workspace_fairness.rs b/backend/tests/workspace_fairness.rs index d8529dfbcd..f5922037e4 100644 --- a/backend/tests/workspace_fairness.rs +++ b/backend/tests/workspace_fairness.rs @@ -79,15 +79,28 @@ async fn create_workspace(db: &Pool, id: &str) { .unwrap(); } +/// Insert `n` completed jobs for `workspace_id`, each ending `secs_ago` +/// seconds in the past with a 1-second wall-clock duration. The fairness +/// algorithm weights contributions by `duration_ms` (clamped to the window), +/// so each job contributes ~1 worker-second when fully inside the window. async fn insert_completed(db: &Pool, workspace_id: &str, n: usize, secs_ago: i32) { for _ in 0..n { + let id: Uuid = sqlx::query_scalar( + "INSERT INTO v2_job (id, workspace_id, kind) + VALUES (gen_random_uuid(), $1, 'script'::job_kind) RETURNING id", + ) + .bind(workspace_id) + .fetch_one(db) + .await + .unwrap(); sqlx::query( "INSERT INTO v2_job_completed (id, workspace_id, duration_ms, status, started_at, completed_at) - VALUES (gen_random_uuid(), $1, 1, 'success'::job_status, - NOW() - make_interval(secs => $2::int), - NOW() - make_interval(secs => $2::int))", + VALUES ($1, $2, 1000, 'success'::job_status, + NOW() - make_interval(secs => ($3::int + 1)), + NOW() - make_interval(secs => $3::int))", ) + .bind(id) .bind(workspace_id) .bind(secs_ago) .execute(db) @@ -106,20 +119,42 @@ async fn insert_queued( let mut ids = Vec::with_capacity(n); for _ in 0..n { let id: Uuid = sqlx::query_scalar( - "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running, tag) - VALUES (gen_random_uuid(), $1, NOW(), $2, $3) RETURNING id", + "INSERT INTO v2_job (id, workspace_id, kind, tag) + VALUES (gen_random_uuid(), $1, 'script'::job_kind, $2) RETURNING id", ) .bind(workspace_id) - .bind(running) .bind(tag) .fetch_one(db) .await .unwrap(); + // Running jobs need a `started_at` for the fairness algorithm to + // compute a positive elapsed-time contribution. Backdate by 1s so + // each running row contributes ~1 worker-second by the time the + // refresh runs, matching the `insert_completed` scale. + sqlx::query( + "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running, tag, started_at) + VALUES ($1, $2, NOW(), $3, $4, + CASE WHEN $3 THEN NOW() - interval '1 second' ELSE NULL END)", + ) + .bind(id) + .bind(workspace_id) + .bind(running) + .bind(tag) + .execute(db) + .await + .unwrap(); if running { - // The fairness algorithm reads slot occupancy from `worker_ping`, - // not from `v2_job_queue.running = true`. Mirror each running row - // with a paired live `worker_ping` so unit tests see the same - // signal a real cluster would. + // The fairness algorithm bounds the running contribution by the + // per-job `v2_job_runtime.ping`. Insert a fresh ping so each + // running row accrues real-time worker-seconds. + sqlx::query( + "INSERT INTO v2_job_runtime (id, ping) VALUES ($1, NOW()) + ON CONFLICT (id) DO UPDATE SET ping = NOW()", + ) + .bind(id) + .execute(db) + .await + .unwrap(); insert_live_worker_ping(db, workspace_id, id).await; } ids.push(id);