feat: Front end for script concurrency limit (#1863)

* Front end for script concurrency limit

* Fix frontend

* sqlx prepare

* fix compile

* add flow inline scripts

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
Co-authored-by: Ruben Fiszel <ruben@rubenfiszel.com>

backend/sqlx-data.json

@@ -3914,63 +3914,6 @@
     },
     "query": "SELECT EXISTS(SELECT 1 FROM variable WHERE path = $1 AND workspace_id = $2)"
   },
-  "88a5f7d43e2775cdbc121e4f39f7d881af75c24740446ca86f26ba8b2e9e5405": {
-    "describe": {
-      "columns": [],
-      "nullable": [],
-      "parameters": {
-        "Left": [
-          "Varchar",
-          "Int8",
-          "Varchar",
-          "Int8Array",
-          "Text",
-          "Text",
-          "Text",
-          "Varchar",
-          "Text",
-          "Bool",
-          "Jsonb",
-          "Text",
-          {
-            "Custom": {
-              "kind": {
-                "Enum": [
-                  "python3",
-                  "deno",
-                  "go",
-                  "bash",
-                  "postgresql",
-                  "nativets",
-                  "bun",
-                  "mysql"
-                ]
-              },
-              "name": "script_lang"
-            }
-          },
-          {
-            "Custom": {
-              "kind": {
-                "Enum": [
-                  "script",
-                  "trigger",
-                  "failure",
-                  "command",
-                  "approval"
-                ]
-              },
-              "name": "script_kind"
-            }
-          },
-          "Varchar",
-          "Bool",
-          "VarcharArray"
-        ]
-      }
-    },
-    "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17)"
-  },
   "88b7589a6416da8be4b26af3bf30fcfcd6aeae7bc5a37e9a735cabbe2691c570": {
     "describe": {
       "columns": [
@@ -4806,6 +4749,65 @@
     },
     "query": "UPDATE flow SET path = $1, summary = $2, description = $3, value = $4, edited_by = $5, edited_at = now(), schema = $6::text::json, dependency_job = NULL, draft_only = NULL WHERE path = $7 AND workspace_id = $8"
   },
+  "9ee1423945740c17dedc2fc7ea7e7183380c43d0feeb615c651aa486eef3bd53": {
+    "describe": {
+      "columns": [],
+      "nullable": [],
+      "parameters": {
+        "Left": [
+          "Varchar",
+          "Int8",
+          "Varchar",
+          "Int8Array",
+          "Text",
+          "Text",
+          "Text",
+          "Varchar",
+          "Text",
+          "Bool",
+          "Jsonb",
+          "Text",
+          {
+            "Custom": {
+              "kind": {
+                "Enum": [
+                  "python3",
+                  "deno",
+                  "go",
+                  "bash",
+                  "postgresql",
+                  "nativets",
+                  "bun",
+                  "mysql"
+                ]
+              },
+              "name": "script_lang"
+            }
+          },
+          {
+            "Custom": {
+              "kind": {
+                "Enum": [
+                  "script",
+                  "trigger",
+                  "failure",
+                  "command",
+                  "approval"
+                ]
+              },
+              "name": "script_kind"
+            }
+          },
+          "Varchar",
+          "Bool",
+          "VarcharArray",
+          "Int4",
+          "Int4"
+        ]
+      }
+    },
+    "query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)"
+  },
   "a17260a1f1ee02e786690994d98c84ddf81e2eeb883f895c9cfc47e144d422cb": {
     "describe": {
       "columns": [

backend/windmill-api/openapi.yaml

@@ -5649,6 +5649,15 @@ components:
           type: array
           items:
             type: string
+        concurrent_limit:
+          type: array
+          items:
+            type: integer
+        concurrency_time_window_s:
+          type: array
+          items:
+            type: integer
+        
       required:
         - hash
         - path
@@ -5700,6 +5709,14 @@ components:
           type: array
           items:
             type: string
+        concurrent_limit:
+          type: array
+          items:
+            type: integer
+        concurrency_time_window_s:
+          type: array
+          items:
+            type: integer
       required:
         - path
         - summary

backend/windmill-api/src/scripts.rs

@@ -71,6 +71,10 @@ pub struct ScriptWDraft {
     pub draft_only: Option<bool>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub envs: Option<Vec<String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub concurrent_limit: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub concurrency_time_window_s: Option<i32>,
 }
 
 pub fn global_service() -> Router {
@@ -367,8 +371,9 @@ async fn create_script(
     //::text::json is to ensure we use serde_json with preserve order
     sqlx::query!(
         "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \
-         content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs) \
-         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17)",
+         content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \
+         draft_only, envs, concurrent_limit, concurrency_time_window_s) \
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)",
         &w_id,
         &hash.0,
         ns.path,
@@ -385,7 +390,9 @@ async fn create_script(
         ns.kind.unwrap_or(ScriptKind::Script): ScriptKind,
         ns.tag,
         ns.draft_only,
-        envs
+        envs,
+        ns.concurrent_limit,
+        ns.concurrency_time_window_s
     )
     .execute(&mut tx)
     .await?;
@@ -565,7 +572,7 @@ async fn get_script_by_path_w_draft(
     let mut tx = user_db.begin(&authed).await?;
 
     let script_o = sqlx::query_as::<_, ScriptWDraft>(
-        "SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, draft.value as draft FROM script LEFT JOIN draft ON 
+        "SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, draft.value as draft FROM script LEFT JOIN draft ON 
          script.path = draft.path AND script.workspace_id = draft.workspace_id AND draft.typ = 'script'
          WHERE script.path = $1 AND script.workspace_id = $2 \
          AND script.created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND \

backend/windmill-common/src/scripts.rs

@@ -149,6 +149,10 @@ pub struct Script {
     pub draft_only: Option<bool>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub envs: Option<Vec<String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub concurrent_limit: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub concurrency_time_window_s: Option<i32>,
 }
 
 #[derive(Serialize)]
@@ -200,6 +204,8 @@ pub struct NewScript {
     pub tag: Option<String>,
     pub draft_only: Option<bool>,
     pub envs: Option<Vec<String>>,
+    pub concurrent_limit: Option<i32>,
+    pub concurrency_time_window_s: Option<i32>,
 }
 
 #[derive(Deserialize)]

backend/windmill-worker/src/worker_flow.rs

@@ -1364,7 +1364,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                             failure_module: fm.clone(),
                             same_worker: flow.same_worker,
                         },
-                        path: Some(format!("{}/loop-{}", flow_job.script_path(), i)),
+                        path: Some(format!("{}/forloop", flow_job.script_path())),
                     },
                     tag: None,
                 }
@@ -1708,9 +1708,15 @@ async fn compute_next_flow_transform(
             } else {
                 let hash = script_hash.clone().unwrap();
                 let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?;
-                let (tag, concurrent_limit, concurrency_time_window_s) = script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?;
+                let (tag, concurrent_limit, concurrency_time_window_s) =
+                    script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?;
                 (
-                    JobPayload::ScriptHash { hash, path: script_path.to_owned(), concurrent_limit, concurrency_time_window_s },
+                    JobPayload::ScriptHash {
+                        hash,
+                        path: script_path.to_owned(),
+                        concurrent_limit,
+                        concurrency_time_window_s,
+                    },
                     tag,
                 )
             };
@@ -1719,7 +1725,16 @@ async fn compute_next_flow_transform(
                 NextStatus::NextStep,
             ))
         }
-        FlowModuleValue::RawScript { path, content, language, lock, tag, concurrent_limit, concurrency_time_window_s, ..} => {
+        FlowModuleValue::RawScript {
+            path,
+            content,
+            language,
+            lock,
+            tag,
+            concurrent_limit,
+            concurrency_time_window_s,
+            ..
+        } => {
             let path = path
                 .clone()
                 .or_else(|| Some(format!("{}/step-{}", flow_job.script_path(), status.step)));

frontend/src/lib/components/ScriptBuilder.svelte

@@ -158,7 +158,9 @@
 					language: script.language,
 					kind: script.kind,
 					tag: script.tag,
-					envs: script.envs
+					envs: script.envs,
+					concurrent_limit: script.concurrent_limit,
+					concurrency_time_window_s: script.concurrency_time_window_s
 				}
 			})
 			history.replaceState(history.state, '', `/scripts/edit/${script.path}`)
@@ -198,7 +200,9 @@
 						kind: script.kind,
 						tag: script.tag,
 						draft_only: true,
-						envs: script.envs
+						envs: script.envs,
+						concurrent_limit: script.concurrent_limit,
+						concurrency_time_window_s: script.concurrency_time_window_s
 					}
 				})
 			}
@@ -256,7 +260,6 @@
 </script>
 
 <svelte:window on:keydown={onKeyDown} />
-
 <UnsavedConfirmationModal />
 {#if !$userStore?.operator}
 	<Drawer placement="right" bind:open={metadataOpen} size="800px">
@@ -370,6 +373,23 @@
 				class="text-sm"
 			/>
 
+			<h2 class="border-b pb-1 mt-10 mb-4">Concurrency limits</h2>
+			<div class="flex gap-x-4 shrink">
+				<label class="block shrink min-w-0">
+					<span class="text-gray-700 text-sm">Maximum number of runs</span>
+					<input class="!w-55" type="number" bind:value={script.concurrent_limit} placeholder="5" />
+				</label>
+				<label class="block shrink min-w-0">
+					<span class="text-gray-700 text-sm">Per time window (seconds)</span>
+					<input
+						class="!w-18"
+						type="number"
+						bind:value={script.concurrency_time_window_s}
+						placeholder="60"
+					/>
+				</label>
+			</div>
+
 			<h2 class="border-b pb-1 mt-10 mb-4"
 				>Worker group tag <Tooltip
 					documentationLink="https://www.windmill.dev/docs/core_concepts/worker_groups"

frontend/src/lib/components/details/DetailPageHeader.svelte

@@ -27,8 +27,9 @@
 <div class="border-b p-2 shadow-md">
 	<div class="mx-auto">
 		<div class="flex w-full flex-wrap md:flex-nowrap justify-end gap-x-2 gap-y-4 h-8 items-center">
-			<div class="grow text-lg font-bold truncate px-2 w-24 sm:w-full">
-				{title}
+			<div class="grow px-2 sm:w-full inline-flex items-center gap-4">
+				<div class="text-lg min-w-24 font-bold truncate">{title}</div>
+				<slot />
 			</div>
 			<div class="flex gap-1 md:gap-2 items-center">
 				<Menu>

frontend/src/lib/components/flows/content/FlowModuleComponent.svelte

@@ -30,6 +30,8 @@
 	import InputTransformSchemaForm from '$lib/components/InputTransformSchemaForm.svelte'
 	import { schemaToObject } from '$lib/schema'
 	import FlowModuleMock from './FlowModuleMock.svelte'
+	import Tooltip from '$lib/components/Tooltip.svelte'
+	import { SecondsInput } from '$lib/components/common'
 
 	const { selectedId, previewArgs, flowStateStore, flowStore, saveDraft } =
 		getContext<FlowEditorContext>('FlowEditorContext')
@@ -136,6 +138,7 @@
 					on:toggleSleep={() => selectAdvanced('sleep')}
 					on:toggleMock={() => selectAdvanced('mock')}
 					on:toggleRetry={() => selectAdvanced('retries')}
+					on:toggleConcurrency={() => selectAdvanced('concurrency')}
 					on:toggleCache={() => selectAdvanced('cache')}
 					on:toggleStopAfterIf={() => selectAdvanced('early-stop')}
 					on:fork={async () => {
@@ -264,6 +267,9 @@
 									<Tab value="retries">Retries</Tab>
 									{#if !$selectedId.includes('failure')}
 										<Tab value="cache">Cache</Tab>
+										{#if flowModule.value.type == 'rawscript'}
+											<Tab value="concurrency">Concurrency</Tab>
+										{/if}
 										<Tab value="early-stop">Early Stop/Break</Tab>
 										<Tab value="suspend">Suspend/Approval</Tab>
 										<Tab value="sleep">Sleep</Tab>
@@ -292,6 +298,35 @@
 										<div>
 											<FlowModuleMock bind:flowModule />
 										</div>
+									{:else if advancedSelected === 'concurrency'}
+										{#if flowModule.value.type == 'rawscript'}
+											<div>
+												<h2 class="pb-4">
+													Concurrency Limits
+													<Tooltip>Allowed concurrency within a given timeframe</Tooltip>
+												</h2>
+											</div>
+											<div>
+												<div class="text-xs font-bold !mt-2"
+													>Max number of executions within the time window</div
+												>
+												<div class="flex flex-row gap-2 max-w-sm"
+													><input bind:value={flowModule.value.concurrent_limit} type="number" />
+													<Button
+														size="sm"
+														color="light"
+														on:click={() => {
+															if (flowModule.value.type == 'rawscript') {
+																flowModule.value.concurrent_limit = undefined
+															}
+														}}
+														variant="border">Remove Limits</Button
+													></div
+												>
+												<div class="text-xs font-bold !mt-2">Time window in seconds</div>
+												<SecondsInput bind:seconds={flowModule.value.concurrency_time_window_s} />
+											</div>
+										{/if}
 									{:else if advancedSelected === 'same_worker'}
 										<div>
 											<Alert type="info" title="Share a directory between steps">

frontend/src/lib/components/flows/content/FlowModuleConcurrencyLimit.svelte

@@ -0,0 +1,57 @@
+<script lang="ts">
+	import Toggle from '$lib/components/Toggle.svelte'
+	import Tooltip from '$lib/components/Tooltip.svelte'
+	import type { FlowModule } from '$lib/gen'
+
+	import JsonEditor from '$lib/components/apps/editor/settingsPanel/inputEditor/JsonEditor.svelte'
+
+	export let flowModule: FlowModule
+
+	let code: string | undefined = flowModule.mock?.return_value
+		? JSON.stringify(flowModule.mock?.return_value, null, 2)
+		: undefined
+
+	$: isMockEnabled = Boolean(flowModule.cache_ttl)
+</script>
+
+<h2 class="pb-4">
+	Mock
+	<Tooltip>
+		If defined and enabled, the step will immediately return the mock value instead of being
+		executed.
+	</Tooltip>
+</h2>
+<Toggle
+	checked={isMockEnabled}
+	on:change={() => {
+		if (isMockEnabled) {
+			flowModule.mock = {
+				enabled: false,
+				return_value: flowModule.mock?.return_value
+			}
+		} else {
+			flowModule.mock = {
+				enabled: true,
+				return_value: flowModule.mock?.return_value ?? { example: 'value' }
+			}
+			code = JSON.stringify(flowModule.mock?.return_value, null, 2)
+		}
+		console.log(isMockEnabled, flowModule.mock)
+	}}
+	options={{
+		right: 'Enable step mocking'
+	}}
+/>
+<div>
+	<span class="text-xs font-bold">Mocked Return value</span>
+	{#if flowModule?.mock?.return_value != undefined}
+		<JsonEditor {code} bind:value={flowModule.mock.return_value} />
+	{:else}
+		<input
+			type="text"
+			disabled
+			value={flowModule.mock?.return_value ? code : ''}
+			class="w-full p-2 border rounded-md"
+		/>
+	{/if}
+</div>

frontend/src/lib/components/flows/content/FlowModuleHeader.svelte

@@ -3,7 +3,7 @@
 	import { WorkerService, type FlowModule } from '$lib/gen'
 	import { faCodeBranch, faPen, faSave } from '@fortawesome/free-solid-svg-icons'
 	import { createEventDispatcher, getContext } from 'svelte'
-	import { Bed, Database, PhoneIncoming, Repeat, Square, Voicemail } from 'lucide-svelte'
+	import { Bed, Database, Gauge, PhoneIncoming, Repeat, Square, Voicemail } from 'lucide-svelte'
 	import Popover from '../../Popover.svelte'
 	import type { FlowEditorContext } from '../types'
 	import { sendUserToast } from '$lib/utils'
@@ -39,6 +39,17 @@
 			<Repeat size={14} />
 			<svelte:fragment slot="text">Retries</svelte:fragment>
 		</Popover>
+		<Popover
+			placement="bottom"
+			class="center-center rounded p-2 
+		{module?.value?.['concurrency_limit'] != undefined
+				? 'bg-blue-100 text-blue-800 border border-blue-300 hover:bg-blue-200'
+				: 'bg-white text-gray-800 border-gray-300 hover:bg-gray-100'}"
+			on:click={() => dispatch('toggleConcurrency')}
+		>
+			<Gauge size={14} />
+			<svelte:fragment slot="text">Concurrency Limits</svelte:fragment>
+		</Popover>
 		<Popover
 			placement="bottom"
 			class="center-center rounded p-2 

frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte

@@ -2,7 +2,17 @@
 	import Badge from '$lib/components/common/badge/Badge.svelte'
 	import Popover from '$lib/components/Popover.svelte'
 	import { classNames } from '$lib/utils'
-	import { Bed, Database, Move, PhoneIncoming, Repeat, Square, Voicemail, X } from 'lucide-svelte'
+	import {
+		Bed,
+		Database,
+		Gauge,
+		Move,
+		PhoneIncoming,
+		Repeat,
+		Square,
+		Voicemail,
+		X
+	} from 'lucide-svelte'
 	import { createEventDispatcher } from 'svelte'
 	import { fade } from 'svelte/transition'
 
@@ -19,6 +29,7 @@
 	export let label: string
 	export let modType: string | undefined = undefined
 	export let bgColor: string = ''
+	export let concurrency: boolean = false
 
 	const dispatch = createEventDispatcher()
 </script>
@@ -45,6 +56,17 @@
 				<svelte:fragment slot="text">Retries</svelte:fragment>
 			</Popover>
 		{/if}
+		{#if concurrency}
+			<Popover notClickable>
+				<div
+					transition:fade|local={{ duration: 200 }}
+					class="center-center rounded border bg-white border-gray-400 text-gray-700 px-1 py-0.5"
+				>
+					<Gauge size={14} />
+				</div>
+				<svelte:fragment slot="text">Concurrency Limits</svelte:fragment>
+			</Popover>
+		{/if}
 		{#if cache}
 			<Popover notClickable>
 				<div

frontend/src/lib/components/flows/map/MapItem.svelte

@@ -43,7 +43,8 @@
 		suspend: Boolean(mod.suspend),
 		sleep: Boolean(mod.sleep),
 		cache: Boolean(mod.cache_ttl),
-		mock: Boolean(mod.mock?.enabled)
+		mock: Boolean(mod.mock?.enabled),
+		concurrency: Boolean(mod?.value?.['concurrent_limit'])
 	}
 
 	function onDelete(event: CustomEvent<MouseEvent>) {

frontend/src/lib/scripts.ts

@@ -86,6 +86,8 @@ export async function getScriptByPath(path: string): Promise<{
 	schema: any
 	description: string
 	tag: string | undefined
+	concurrent_limit: number[] | undefined
+	concurrency_time_window_s: number[] | undefined
 }> {
 	if (path.startsWith('hub/')) {
 		const { content, language, schema } = await ScriptService.getHubScriptByPath({ path })
@@ -95,7 +97,9 @@ export async function getScriptByPath(path: string): Promise<{
 			language: language as SupportedLanguage,
 			schema,
 			description: '',
-			tag: undefined
+			tag: undefined,
+			concurrent_limit: undefined,
+			concurrency_time_window_s: undefined,
 		}
 	} else {
 		const script = await ScriptService.getScriptByPath({
@@ -107,7 +111,9 @@ export async function getScriptByPath(path: string): Promise<{
 			language: script.language,
 			schema: script.schema,
 			description: script.description,
-			tag: script.tag
+			tag: script.tag,
+			concurrent_limit: script.concurrent_limit,
+			concurrency_time_window_s: script.concurrency_time_window_s,
 		}
 	}
 }

frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte

@@ -376,7 +376,15 @@
 				{mainButtons}
 				menuItems={getMenuItems(script)}
 				title={defaultIfEmptyString(script.summary, script.path)}
-			/>
+			>
+				{#if script?.concurrent_limit != undefined && script.concurrency_time_window_s != undefined}
+					<div class="hidden md:block">
+						<Badge color="gray" variant="outlined" size="xs">
+							{`Concurrency limit: ${script.concurrent_limit} runs every ${script.concurrency_time_window_s}s`}
+						</Badge>
+					</div>
+				{/if}
+			</DetailPageHeader>
 		</svelte:fragment>
 		<svelte:fragment slot="form">
 			<div class="p-8 w-full max-w-3xl mx-auto">

openflow.openapi.yaml

@@ -206,6 +206,10 @@ components:
             - rawscript
         tag:
           type: string
+        concurrent_limit:
+          type: number
+        concurrency_time_window_s:
+          type: number
       required:
         - type
         - content