diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 96968605b1..85ec0edae4 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -12,7 +12,10 @@ use windmill_api_auth::{ check_scopes, maybe_refresh_folders, require_owner_of_path, ApiAuthed, }; use windmill_common::{ - utils::{BulkDeleteRequest, WithStarredInfoQuery, HTTP_CLIENT}, webhook::{WebhookMessage, WebhookShared}, workspaces::{check_user_against_rule, ProtectionRuleKind, RuleCheckResult}, DB + utils::{BulkDeleteRequest, WithStarredInfoQuery, HTTP_CLIENT}, + webhook::{WebhookMessage, WebhookShared}, + workspaces::{check_user_against_rule, ProtectionRuleKind, RuleCheckResult}, + DB, }; use windmill_queue::schedule::clear_schedule; @@ -859,10 +862,13 @@ async fn create_script_internal<'c>( } }; - let runnable_settings_handle = windmill_common::runnable_settings::insert_rs(RunnableSettings { - debouncing_settings: ns.debouncing_settings.insert_cached(&db).await?, - concurrency_settings: ns.concurrency_settings.insert_cached(&db).await?, - }, &db) + let runnable_settings_handle = windmill_common::runnable_settings::insert_rs( + RunnableSettings { + debouncing_settings: ns.debouncing_settings.insert_cached(&db).await?, + concurrency_settings: ns.concurrency_settings.insert_cached(&db).await?, + }, + &db, + ) .await?; let ( @@ -1072,7 +1078,9 @@ async fn create_script_internal<'c>( } if needs_lock_gen { let tag = if ns.dedicated_worker.is_some_and(|x| x) { - Some(format!("{}:{}", &w_id, &ns.path,)) + Some(windmill_common::worker::dedicated_worker_tag( + &w_id, &ns.path, + )) } else if ns.tag.as_ref().is_some_and(|x| x.contains("$args[")) { None } else { @@ -1839,7 +1847,9 @@ async fn get_script_by_hash( tx.commit().await?; - Ok(Json(windmill_common::scripts::prefetch_cached_script_with_starred(r, &db).await?)) + Ok(Json( + windmill_common::scripts::prefetch_cached_script_with_starred(r, &db).await?, + )) } async fn raw_script_by_hash( @@ -2042,7 +2052,9 @@ async fn archive_script_by_hash( WebhookMessage::DeleteScript { workspace: w_id, hash: hash.to_string() }, ); - Ok(Json(windmill_common::scripts::prefetch_cached_script(script, &db).await?)) + Ok(Json( + windmill_common::scripts::prefetch_cached_script(script, &db).await?, + )) } async fn delete_script_by_hash( @@ -2097,7 +2109,9 @@ async fn delete_script_by_hash( WebhookMessage::DeleteScript { workspace: w_id, hash: hash.to_string() }, ); - Ok(Json(windmill_common::scripts::prefetch_cached_script(script, &db).await?)) + Ok(Json( + windmill_common::scripts::prefetch_cached_script(script, &db).await?, + )) } #[derive(Deserialize)] diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index b2cc8623d1..fbd9a9b682 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -2135,9 +2135,7 @@ pub async fn resume_suspended_flow_as_owner( // Check approval conditions (self-approval, required groups, etc.) if let Some(ref flow_status_value) = flow.flow_status { - if let Ok(flow_status) = - serde_json::from_value::(flow_status_value.clone()) - { + if let Ok(flow_status) = serde_json::from_value::(flow_status_value.clone()) { let trigger_email = flow.email.as_deref().unwrap_or(""); conditionally_require_authed_user(Some(authed.clone()), flow_status, trigger_email)?; } @@ -5224,7 +5222,7 @@ async fn add_batch_jobs( let tag = if let Some(dedicated_worker) = dedicated_worker { if dedicated_worker && path.is_some() { - format!("{}:{}", w_id, path.clone().unwrap()) + windmill_common::worker::dedicated_worker_tag(&w_id, &path.clone().unwrap()) } else { format!("{}", language.as_str()) } diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 4a12ad00a4..6d392fa950 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -1560,6 +1560,24 @@ pub async fn update_worker_ping_main_loop_query( // occupancy_rate = $3, memory_usage = $4, wm_memory_usage = $5, vcpus = COALESCE($7, vcpus), // memory = COALESCE($8, memory), occupancy_rate_15s = $9, occupancy_rate_5m = $10, occupancy_rate_30m = $11 WHERE worker = $6", +const MAX_TAG_LEN: usize = 50; +const HASH_SUFFIX_LEN: usize = 16; + +pub fn dedicated_worker_tag(workspace_id: &str, path: &str) -> String { + let full_tag = format!("{}:{}", workspace_id, path); + if full_tag.len() <= MAX_TAG_LEN { + return full_tag; + } + let hash = ::digest(full_tag.as_bytes()); + let hex_hash = hex::encode(hash); + let prefix_len = MAX_TAG_LEN - 1 - HASH_SUFFIX_LEN; + format!( + "{}#{}", + &full_tag[..prefix_len], + &hex_hash[..HASH_SUFFIX_LEN] + ) +} + pub async fn load_worker_config( db: &DB, killpill_tx: KillpillSender, @@ -1671,7 +1689,7 @@ pub async fn load_worker_config( if let Some(ref dws) = dedicated_workers.as_ref() { let mut dedi_tags: Vec = dws .iter() - .map(|dw| format!("{}:{}", dw.workspace_id, dw.path)) + .map(|dw| dedicated_worker_tag(&dw.workspace_id, &dw.path)) .collect(); if std::env::var("ADD_FLOW_TAG").is_ok() { dedi_tags.push("flow".to_string()); @@ -1679,9 +1697,9 @@ pub async fn load_worker_config( Some(dedi_tags) } else if let Some(ref dedicated_worker) = dedicated_worker.as_ref() { // Fallback to single dedicated worker for backward compatibility - let mut dedi_tags = vec![format!( - "{}:{}", - dedicated_worker.workspace_id, dedicated_worker.path + let mut dedi_tags = vec![dedicated_worker_tag( + &dedicated_worker.workspace_id, + &dedicated_worker.path, )]; if std::env::var("ADD_FLOW_TAG").is_ok() { dedi_tags.push("flow".to_string()); @@ -2178,4 +2196,58 @@ mod tests { result.sort(); assert_eq!(result, vec!["foo", "legacy(^ws1^ws2)", "urgent(ws1+ws2)"]); } + + #[test] + fn test_dedicated_worker_tag_short() { + let tag = dedicated_worker_tag("demo", "u/alice/script"); + assert_eq!(tag, "demo:u/alice/script"); + assert!(tag.len() <= MAX_TAG_LEN); + } + + #[test] + fn test_dedicated_worker_tag_exactly_50() { + // 50 chars exactly should not be hashed + let workspace = "ws"; + let path = "a".repeat(50 - workspace.len() - 1); // -1 for ':' + let tag = dedicated_worker_tag(workspace, &path); + assert_eq!(tag.len(), 50); + assert!(!tag.contains('#')); + } + + #[test] + fn test_dedicated_worker_tag_long_is_hashed() { + let tag = dedicated_worker_tag( + "my_workspace", + "u/engineering/team/automation/critical_workflow_script_v2", + ); + assert_eq!(tag.len(), MAX_TAG_LEN); + assert_eq!(tag, "my_workspace:u/engineering/team/a#5bc26db79926d4f0"); + } + + #[test] + fn test_dedicated_worker_tag_deterministic() { + let a = dedicated_worker_tag( + "ws", + "some/very/long/path/that/exceeds/the/fifty/char/limit/easily", + ); + assert_eq!(a, "ws:some/very/long/path/that/excee#bbb038d4268a0b41"); + let b = dedicated_worker_tag( + "ws", + "some/very/long/path/that/exceeds/the/fifty/char/limit/easily", + ); + assert_eq!(a, b); + } + + #[test] + fn test_dedicated_worker_tag_different_paths_differ() { + let a = dedicated_worker_tag( + "ws", + "some/very/long/path/that/exceeds/the/fifty/char/limit/easily_a", + ); + let b = dedicated_worker_tag( + "ws", + "some/very/long/path/that/exceeds/the/fifty/char/limit/easily_b", + ); + assert_ne!(a, b); + } } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index b298fed82c..eaa75f04c1 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -5274,16 +5274,17 @@ async fn push_inner<'c, 'd>( .unwrap_or_else(|| (None, None)); let tag = if dedicated_worker.is_some_and(|x| x) { - format!( - "{}:{}{}", - workspace_id, - if job_kind == JobKind::Flow || job_kind == JobKind::FlowDependencies { - "flow/" - } else { - "" - }, + let flow_prefix = if job_kind == JobKind::Flow || job_kind == JobKind::FlowDependencies { + "flow/" + } else { + "" + }; + let full_path = format!( + "{}{}", + flow_prefix, runnable_path.clone().expect("dedicated script has a path") - ) + ); + windmill_common::worker::dedicated_worker_tag(workspace_id, &full_path) } else { if tag == Some("".to_string()) { tag = None; diff --git a/frontend/src/lib/components/DedicatedWorkersSelector.svelte b/frontend/src/lib/components/DedicatedWorkersSelector.svelte index df37b5bf17..336c2f263b 100644 --- a/frontend/src/lib/components/DedicatedWorkersSelector.svelte +++ b/frontend/src/lib/components/DedicatedWorkersSelector.svelte @@ -8,6 +8,7 @@ import { SvelteMap } from 'svelte/reactivity' import { untrack } from 'svelte' import BarsStaggered from './icons/BarsStaggered.svelte' + import { parseTag } from './dedicated_worker' // A "Runnable" is a script or flow with dedicated_worker=true interface Runnable { @@ -62,21 +63,6 @@ // Languages that support dedicated workers const DEDICATED_WORKER_LANGUAGES = ['python3', 'bun', 'deno'] - // Parse a tag to extract workspace, type (script/flow), and path - function parseTag(tag: string): { workspace: string; type: 'script' | 'flow'; path: string } | null { - const colonIndex = tag.indexOf(':') - if (colonIndex === -1) return null - - const workspace = tag.substring(0, colonIndex) - const rest = tag.substring(colonIndex + 1) - - if (rest.startsWith('flow/')) { - return { workspace, type: 'flow', path: rest.substring(5) } - } else { - return { workspace, type: 'script', path: rest } - } - } - // Resolve workspace script languages and filter to supported languages async function resolveAndFilterRunners( workspace: string, @@ -438,25 +424,27 @@ {:else}
{/if} -
- {#if info} - {#if info.type === 'flow'} - +
+
+ {#if info} + {#if info.type === 'flow'} + + {:else} + + {/if} + {info.path} + ({info.workspace}) + {#if info.type === 'flow' && info.runners} + + {info.runners.length} runner{info.runners.length !== 1 ? 's' : ''} + + {:else if info.type === 'script'} + 1 runner + {/if} {:else} - + {tag} {/if} - {info.path} - ({info.workspace}) - {#if info.type === 'flow' && info.runners} - - {info.runners.length} runner{info.runners.length !== 1 ? 's' : ''} - - {:else if info.type === 'script'} - 1 runner - {/if} - {:else} - {tag} - {/if} +
{#if !disabled}
- {runnable.displayName} + {runnable.displayName} {#if runnable.type === 'flow' && runnable.runners} - + {runnable.runners.length} {/if} @@ -614,11 +608,13 @@ {:else} {#each runnable.runners as runner (runner.stepId)}
- {runner.stepId} + {runner.stepId} {#if runner.stepSummary} - + {runner.stepSummary} {/if} diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index e2d1327eac..ca3c47ef20 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -1443,8 +1443,8 @@ {#if script.dedicated_worker}
- One worker in a worker group needs to be configured with dedicated worker - set to: 
{$workspaceStore}:{script.path}
+ A worker group needs to be configured to listen to this script. Select + it in the dedicated workers section of the worker group configuration.
{/if} diff --git a/frontend/src/lib/components/WorkerGroup.svelte b/frontend/src/lib/components/WorkerGroup.svelte index da3225df00..820f21599e 100644 --- a/frontend/src/lib/components/WorkerGroup.svelte +++ b/frontend/src/lib/components/WorkerGroup.svelte @@ -43,6 +43,7 @@ import Dropdown from './DropdownV2.svelte' import TagList from './TagList.svelte' import DedicatedWorkersSelector from './DedicatedWorkersSelector.svelte' + import { computeHashedTag } from './dedicated_worker' function computeVCpuAndMemory(workers: [string, WorkerPing[]][]) { let vcpus = 0 @@ -252,6 +253,21 @@ let openDelete = $state(false) let openClean = $state(false) + // Compute hashed tags for display (actual tags used by the worker) + let hashedDedicatedTags: Map = $state(new Map()) + $effect(() => { + const dws = config?.dedicated_workers ?? (config?.dedicated_worker ? [config.dedicated_worker] : []) + if (dws.length > 0) { + Promise.all(dws.map(async (dw) => [dw, await computeHashedTag(dw)] as const)).then( + (entries) => { + hashedDedicatedTags = new Map(entries) + } + ) + } else { + hashedDedicatedTags = new Map() + } + }) + let drawer: Drawer | undefined = $state() let vcpus_memory = $derived(computeVCpuAndMemory(workers)) let selected = $derived( @@ -1166,21 +1182,27 @@
{:else if config?.dedicated_workers && config.dedicated_workers.length > 0} -
-
Dedicated to:
-
+
+
Dedicated to:
+
{#each config.dedicated_workers as dw} -
- {dw} +
+ {hashedDedicatedTags.get(dw) ?? dw}
{/each}
{:else if config?.dedicated_worker} -
-
Dedicated to:
-
- {config.dedicated_worker} +
+
Dedicated to:
+
+ {hashedDedicatedTags.get(config.dedicated_worker) ?? config.dedicated_worker}
{/if} diff --git a/frontend/src/lib/components/dedicated_worker.ts b/frontend/src/lib/components/dedicated_worker.ts new file mode 100644 index 0000000000..f3b90fde57 --- /dev/null +++ b/frontend/src/lib/components/dedicated_worker.ts @@ -0,0 +1,41 @@ +const MAX_TAG_LEN = 50 +const HASH_SUFFIX_LEN = 16 + +// IMPORTANT: This implementation must stay in sync with the Rust backend +// `dedicated_worker_tag()` in backend/windmill-common/src/worker.rs. +// The frontend version is used for display; the backend version is authoritative. +export async function dedicatedWorkerTag(workspaceId: string, path: string): Promise { + const fullTag = `${workspaceId}:${path}` + if (fullTag.length <= MAX_TAG_LEN) return fullTag + + const encoded = new TextEncoder().encode(fullTag) + const hashBuffer = await crypto.subtle.digest('SHA-256', encoded) + const hashArray = Array.from(new Uint8Array(hashBuffer)) + const hexHash = hashArray.map((b) => b.toString(16).padStart(2, '0')).join('') + + const prefixLen = MAX_TAG_LEN - 1 - HASH_SUFFIX_LEN + return `${fullTag.substring(0, prefixLen)}#${hexHash.substring(0, HASH_SUFFIX_LEN)}` +} + +export function parseTag( + tag: string +): { workspace: string; type: 'script' | 'flow'; path: string } | null { + const colonIndex = tag.indexOf(':') + if (colonIndex === -1) return null + + const workspace = tag.substring(0, colonIndex) + const rest = tag.substring(colonIndex + 1) + + if (rest.startsWith('flow/')) { + return { workspace, type: 'flow', path: rest.substring(5) } + } else { + return { workspace, type: 'script', path: rest } + } +} + +export async function computeHashedTag(rawTag: string): Promise { + const parsed = parseTag(rawTag) + if (!parsed) return rawTag + const fullPath = parsed.type === 'flow' ? `flow/${parsed.path}` : parsed.path + return dedicatedWorkerTag(parsed.workspace, fullPath) +} diff --git a/frontend/src/lib/components/flows/content/FlowSettings.svelte b/frontend/src/lib/components/flows/content/FlowSettings.svelte index ad903f40a5..5e0e47da24 100644 --- a/frontend/src/lib/components/flows/content/FlowSettings.svelte +++ b/frontend/src/lib/components/flows/content/FlowSettings.svelte @@ -8,7 +8,7 @@ import { Alert, Button, SecondsInput } from '$lib/components/common' import { getContext } from 'svelte' import type { FlowEditorContext } from '../types' - import { enterpriseLicense, userStore, workspaceStore } from '$lib/stores' + import { enterpriseLicense, userStore } from '$lib/stores' import { isCloudHosted } from '$lib/cloud' import Tooltip from '$lib/components/Tooltip.svelte' import SimpleEditor from '$lib/components/SimpleEditor.svelte' @@ -554,9 +554,8 @@ {#if flowStore.val.dedicated_worker}
- One worker in a worker group needs to be configured with dedicated worker set to: 
{$workspaceStore}:flow/{$pathStore}
+ A worker group needs to be configured to listen to this flow. Select it in the dedicated + workers section of the worker group configuration.
{/if}