audit-log workspace-fairness cap transitions (#9306)

* feat(queue): audit-log workspace-fairness cap transitions

When the cloud per-workspace fairness mechanism adds a workspace to the
capped set or releases one, write `workspace_fairness.capped` /
`workspace_fairness.uncapped` audit-log entries to the affected workspace.
The cluster admin can review the full timeline from the `admins` workspace
audit view with `all_workspaces=true`; per-workspace owners see their own
events in their normal audit list.

Only the per-cycle refresh winner emits entries (matching where the heavy
aggregation runs), so a fleet of N workers does not produce N duplicates
per transition. The diff is computed against the value already in
`background_task_state` rather than the winner's in-memory cache, so a
freshly-restarted process winning the claim does not spuriously emit
"newly capped" entries for workspaces that were already capped before it
started.

Audit writes are best-effort: failures are logged via tracing and do not
abort the refresh cycle.

Fixes WIN-1984

* feat(queue): scope fairness audit to admins workspace + queue-metrics pane

- Write `workspace_fairness.capped` / `workspace_fairness.uncapped` to the
  `admins` workspace (was: per-affected-workspace) with the affected
  workspace_id moved to the `resource` field. Cluster admins now get the
  full timeline in one place without `all_workspaces=true`.
- Add `GET /workers/workspace_fairness_events` returning the last 100
  events. Cloud-gated (returns `[]` on non-cloud) and devops-only.
- Add a `WorkspaceFairnessEvents` Section to the Queue Metrics drawer,
  rendered only when `isCloudHosted()` is true. Shows time / event
  badge / workspace / parameters with a refresh button.

Fixes WIN-1984

backend/.sqlx/query-8df7878c46d8bfd38d4df91dff578e5d2d13f3516d954610b32b0709d6768a80.json

@@ -0,0 +1,38 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT timestamp AS \"timestamp!\",\n               operation::text AS \"operation!\",\n               resource AS workspace_id,\n               parameters\n        FROM (\n            SELECT timestamp, operation, resource, parameters\n            FROM audit_partitioned\n            WHERE workspace_id = 'admins'\n              AND operation IN ('workspace_fairness.capped', 'workspace_fairness.uncapped')\n            UNION ALL\n            SELECT timestamp, operation, resource, parameters\n            FROM audit\n            WHERE workspace_id = 'admins'\n              AND operation IN ('workspace_fairness.capped', 'workspace_fairness.uncapped')\n        ) e\n        ORDER BY timestamp DESC\n        LIMIT 100\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "timestamp!",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 1,
+        "name": "operation!",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "parameters",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null,
+      null,
+      null,
+      null
+    ]
+  },
+  "hash": "8df7878c46d8bfd38d4df91dff578e5d2d13f3516d954610b32b0709d6768a80"
+}

backend/windmill-api-workers/src/lib.rs

@@ -20,7 +20,10 @@ use windmill_common::{
     error::JsonResult,
     jobs::{HIDE_WORKERS_FOR_NON_ADMINS, TAGS_ARE_SENSITIVE},
     utils::{paginate, Pagination},
-    worker::{ALL_TAGS, CUSTOM_TAGS_PER_WORKSPACE, DEFAULT_TAGS, DEFAULT_TAGS_PER_WORKSPACE},
+    worker::{
+        is_cloud_production_host, ALL_TAGS, CUSTOM_TAGS_PER_WORKSPACE, DEFAULT_TAGS,
+        DEFAULT_TAGS_PER_WORKSPACE,
+    },
     DB,
 };
 
@@ -39,6 +42,10 @@ pub fn global_service() -> Router {
         .route("/queue_metrics", get(get_queue_metrics))
         .route("/queue_counts", get(get_queue_counts))
         .route("/queue_running_counts", get(get_queue_running_counts))
+        .route(
+            "/workspace_fairness_events",
+            get(get_workspace_fairness_events),
+        )
 }
 
 pub fn workspaced_service() -> Router {
@@ -283,3 +290,58 @@ async fn get_queue_running_counts(
     let queue_running_counts = windmill_common::queue::get_queue_running_counts(&db).await;
     Ok(Json(queue_running_counts))
 }
+
+#[derive(Serialize)]
+pub struct WorkspaceFairnessEvent {
+    pub timestamp: chrono::DateTime<chrono::Utc>,
+    pub operation: String,
+    /// Affected workspace (stored in audit log `resource`). `None` only for very
+    /// old rows pre-dating the resource convention — UI should treat as "unknown".
+    pub workspace_id: Option<String>,
+    /// Snapshot of the relevant fairness settings at the time of the transition
+    /// (`max_percent`, `window_secs`, `total_overloaded`). `None` for uncap rows.
+    pub parameters: Option<serde_json::Value>,
+}
+
+/// Return the last 100 workspace-fairness cap/uncap transitions written by
+/// `workspace_fairness::emit_transition_audit`. Cloud-only: the underlying
+/// mechanism is hard-gated to `CLOUD_HOSTED=true` + `app.windmill.dev`, so on
+/// any other instance we short-circuit with an empty list rather than running
+/// a query that would always be empty.
+async fn get_workspace_fairness_events(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+) -> JsonResult<Vec<WorkspaceFairnessEvent>> {
+    require_devops_role(&db, &authed.email).await?;
+
+    if !is_cloud_production_host() {
+        return Ok(Json(vec![]));
+    }
+
+    let events = sqlx::query_as!(
+        WorkspaceFairnessEvent,
+        r#"
+        SELECT timestamp AS "timestamp!",
+               operation::text AS "operation!",
+               resource AS workspace_id,
+               parameters
+        FROM (
+            SELECT timestamp, operation, resource, parameters
+            FROM audit_partitioned
+            WHERE workspace_id = 'admins'
+              AND operation IN ('workspace_fairness.capped', 'workspace_fairness.uncapped')
+            UNION ALL
+            SELECT timestamp, operation, resource, parameters
+            FROM audit
+            WHERE workspace_id = 'admins'
+              AND operation IN ('workspace_fairness.capped', 'workspace_fairness.uncapped')
+        ) e
+        ORDER BY timestamp DESC
+        LIMIT 100
+        "#,
+    )
+    .fetch_all(&db)
+    .await?;
+
+    Ok(Json(events))
+}

backend/windmill-api/openapi.yaml

@@ -17682,6 +17682,38 @@ paths:
                 additionalProperties:
                   type: integer
 
+  /workers/workspace_fairness_events:
+    get:
+      summary: list last 100 workspace-fairness cap/uncap events (cloud-only)
+      operationId: getWorkspaceFairnessEvents
+      tags:
+        - worker
+      responses:
+        "200":
+          description: workspace fairness events (empty on non-cloud)
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  type: object
+                  properties:
+                    timestamp:
+                      type: string
+                      format: date-time
+                    operation:
+                      type: string
+                    workspace_id:
+                      type: string
+                      nullable: true
+                    parameters:
+                      type: object
+                      nullable: true
+                      additionalProperties: true
+                  required:
+                    - timestamp
+                    - operation
+
   /configs/list_worker_groups:
     get:
       summary: list worker groups

backend/windmill-queue/src/workspace_fairness.rs

@@ -37,12 +37,16 @@
 //! cloud's `global_settings` row). When either check fails, [`maybe_refresh_overloaded`]
 //! and the pull-side dispatch both treat the feature as disabled.
 
+use std::collections::HashSet;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
 
 use sqlx::{Pool, Postgres};
 
+use windmill_audit::audit_oss::audit_log;
+use windmill_audit::ActionKind;
+use windmill_common::audit::AuditAuthor;
 use windmill_common::error::Result;
 use windmill_common::worker::{
     is_cloud_production_host, WORKSPACE_FAIRNESS_DURATION_SECS, WORKSPACE_FAIRNESS_ENABLED,
@@ -190,6 +194,28 @@ async fn refresh_overloaded(db: &Pool<Postgres>) -> Result<()> {
     .await?
     .is_some();
 
+    // Winner-only: read the value the cluster currently holds, so we can emit
+    // audit-log entries for workspaces that just entered or left the capped set.
+    // We compare against the DB-stored prior list (not the per-process local
+    // cache) so a freshly-restarted worker that happens to win the claim does
+    // not emit spurious "newly capped" entries for the workspaces that were
+    // already capped before it started.
+    let prev_list_from_db: Vec<String> = if won {
+        let prev_val: Option<serde_json::Value> =
+            sqlx::query_scalar("SELECT value FROM background_task_state WHERE name = $1")
+                .bind(TASK_STATE_NAME)
+                .fetch_optional(db)
+                .await?;
+        match prev_val {
+            Some(v) => serde_json::from_value::<FairnessState>(v)
+                .map(|s| s.overloaded)
+                .unwrap_or_default(),
+            None => vec![],
+        }
+    } else {
+        vec![]
+    };
+
     // Step 2: winner-only aggregation + value write. `SET` is evaluated per
     // updated row, so issuing this statement only when `won` guarantees the
     // expensive aggregation never runs for a loser.
@@ -247,6 +273,33 @@ async fn refresh_overloaded(db: &Pool<Postgres>) -> Result<()> {
         None => vec![],
     };
 
+    // Audit-log transitions (winner only). Skipping losers prevents N duplicate
+    // entries per transition on a fleet of N worker processes.
+    if won && prev_list_from_db != new_list {
+        let prev_set: HashSet<&str> = prev_list_from_db.iter().map(String::as_str).collect();
+        let new_set: HashSet<&str> = new_list.iter().map(String::as_str).collect();
+        let newly_capped: Vec<&str> = new_list
+            .iter()
+            .map(String::as_str)
+            .filter(|w| !prev_set.contains(w))
+            .collect();
+        let newly_uncapped: Vec<&str> = prev_list_from_db
+            .iter()
+            .map(String::as_str)
+            .filter(|w| !new_set.contains(w))
+            .collect();
+
+        emit_transition_audit(
+            db,
+            &newly_capped,
+            &newly_uncapped,
+            max_percent,
+            duration_secs,
+            new_list.len(),
+        )
+        .await;
+    }
+
     let prev = WORKSPACE_FAIRNESS_OVERLOADED.load();
     if **prev != new_list {
         tracing::info!(
@@ -260,3 +313,72 @@ async fn refresh_overloaded(db: &Pool<Postgres>) -> Result<()> {
 
     Ok(())
 }
+
+/// Best-effort audit-log emission for workspaces entering or leaving the
+/// capped set. Failures are logged but never propagated — the refresh cycle
+/// must not abort just because an audit insert fails.
+///
+/// Entries are scoped to the `admins` workspace and carry the affected
+/// workspace as the `resource` field, so the queue-metrics drawer (cloud-only)
+/// can fetch a single time-ordered stream of cap transitions without merging
+/// across workspaces. Cluster admins also see them in the standard `admins`
+/// audit listing (no `all_workspaces=true` toggle required).
+async fn emit_transition_audit(
+    db: &Pool<Postgres>,
+    newly_capped: &[&str],
+    newly_uncapped: &[&str],
+    max_percent: i64,
+    duration_secs: i32,
+    total_overloaded: usize,
+) {
+    if newly_capped.is_empty() && newly_uncapped.is_empty() {
+        return;
+    }
+
+    let author = AuditAuthor {
+        username: "system".to_string(),
+        email: "system@windmill.dev".to_string(),
+        username_override: None,
+        token_prefix: None,
+    };
+
+    let max_percent_s = max_percent.to_string();
+    let window_secs_s = duration_secs.to_string();
+    let total_s = total_overloaded.to_string();
+
+    for w in newly_capped {
+        let mut params = std::collections::HashMap::<&str, &str>::new();
+        params.insert("max_percent", &max_percent_s);
+        params.insert("window_secs", &window_secs_s);
+        params.insert("total_overloaded", &total_s);
+        if let Err(e) = audit_log(
+            db,
+            &author,
+            "workspace_fairness.capped",
+            ActionKind::Update,
+            "admins",
+            Some(w),
+            Some(params),
+        )
+        .await
+        {
+            tracing::warn!("failed to write workspace_fairness.capped audit for {w}: {e:#}");
+        }
+    }
+
+    for w in newly_uncapped {
+        if let Err(e) = audit_log(
+            db,
+            &author,
+            "workspace_fairness.uncapped",
+            ActionKind::Update,
+            "admins",
+            Some(w),
+            None,
+        )
+        .await
+        {
+            tracing::warn!("failed to write workspace_fairness.uncapped audit for {w}: {e:#}");
+        }
+    }
+}

frontend/src/lib/components/QueueMetricsDrawer.svelte

@@ -2,6 +2,8 @@
 	import { Drawer, DrawerContent } from './common'
 	import QueueMetricsDrawerInner from './QueueMetricsDrawerInner.svelte'
 	import QueueAlerts from './QueueAlerts.svelte'
+	import WorkspaceFairnessEvents from './WorkspaceFairnessEvents.svelte'
+	import { isCloudHosted } from '$lib/cloud'
 
 	let drawer: Drawer | undefined = $state()
 	export function openDrawer() {
@@ -21,6 +23,11 @@
 
 		<QueueMetricsDrawerInner />
 
+		{#if isCloudHosted()}
+			<div class="py-8"></div>
+			<WorkspaceFairnessEvents />
+		{/if}
+
 		<div class="py-8"></div>
 	</DrawerContent>
 </Drawer>

frontend/src/lib/components/WorkspaceFairnessEvents.svelte

@@ -0,0 +1,106 @@
+<script lang="ts">
+	import { WorkerService } from '$lib/gen'
+	import { Section } from './common'
+	import Skeleton from './common/skeleton/Skeleton.svelte'
+	import Badge from './common/badge/Badge.svelte'
+	import Alert from './common/alert/Alert.svelte'
+	import Button from './common/button/Button.svelte'
+	import { displayDate } from '$lib/utils'
+	import { RefreshCw } from 'lucide-svelte'
+
+	type FairnessEvent = {
+		timestamp: string
+		operation: string
+		workspace_id?: string | null
+		parameters?: Record<string, unknown> | null
+	}
+
+	let events: FairnessEvent[] = $state([])
+	let loading = $state(true)
+	let error: string | undefined = $state(undefined)
+
+	async function load() {
+		loading = true
+		error = undefined
+		try {
+			const res = await WorkerService.getWorkspaceFairnessEvents()
+			events = (res as unknown as FairnessEvent[]) ?? []
+		} catch (e) {
+			error = e instanceof Error ? e.message : String(e)
+		} finally {
+			loading = false
+		}
+	}
+
+	load()
+
+	function formatParameters(p?: Record<string, unknown> | null): string {
+		if (!p) return ''
+		// Audit-log params arrive as `{ hm: { ... }, args: ... }` in EE format —
+		// surface the `hm` map when present, otherwise show the object as-is.
+		const inner = (p as { hm?: Record<string, unknown> }).hm ?? p
+		return Object.entries(inner)
+			.map(([k, v]) => `${k}=${v}`)
+			.join(', ')
+	}
+</script>
+
+<Section label="Workspace fairness events (cloud)">
+	{#snippet action()}
+		<Button
+			variant="subtle"
+			size="xs"
+			startIcon={{ icon: RefreshCw }}
+			iconOnly
+			onclick={load}
+			disabled={loading}
+		/>
+	{/snippet}
+
+	{#if loading}
+		<Skeleton layout={[[8]]} />
+	{:else if error}
+		<Alert type="error" title="Failed to load">{error}</Alert>
+	{:else if events.length === 0}
+		<p class="text-secondary text-sm">
+			No workspaces have been added to or removed from the fairness-restricted set yet. Entries
+			appear here once the cap activates.
+		</p>
+	{:else}
+		<div class="text-xs text-secondary mb-2">
+			Showing the last {events.length} workspace-fairness transitions (most recent first).
+		</div>
+		<div class="overflow-x-auto border rounded-md">
+			<table class="min-w-full text-sm">
+				<thead class="bg-surface-secondary text-secondary">
+					<tr>
+						<th class="text-left px-3 py-2 font-medium">Time</th>
+						<th class="text-left px-3 py-2 font-medium">Event</th>
+						<th class="text-left px-3 py-2 font-medium">Workspace</th>
+						<th class="text-left px-3 py-2 font-medium">Details</th>
+					</tr>
+				</thead>
+				<tbody>
+					{#each events as e}
+						<tr class="border-t">
+							<td class="px-3 py-2 whitespace-nowrap text-tertiary">
+								{displayDate(e.timestamp, true)}
+							</td>
+							<td class="px-3 py-2 whitespace-nowrap">
+								{#if e.operation === 'workspace_fairness.capped'}
+									<Badge color="red">capped</Badge>
+								{:else if e.operation === 'workspace_fairness.uncapped'}
+									<Badge color="green">uncapped</Badge>
+								{:else}
+									<Badge>{e.operation}</Badge>
+								{/if}
+							</td>
+							<td class="px-3 py-2 font-mono">{e.workspace_id ?? '—'}</td>
+							<td class="px-3 py-2 text-tertiary text-xs">{formatParameters(e.parameters)}</td>
+						</tr>
+					{/each}
+				</tbody>
+			</table>
+		</div>
+	{/if}
+</Section>