starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-05 00:03:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: move concurrency and debouncing settings to their own structs (#7286)

Browse Source 
* move concurrency and debouncing settings to it's own structs

Signed-off-by: pyranota <pyra@duck.com>

* cleanup queue::jobs::pull

Signed-off-by: pyranota <pyra@duck.com>

* done

Signed-off-by: pyranota <pyra@duck.com>

* migrated the rest

Signed-off-by: pyranota <pyra@duck.com>

* implement todo!

* impmenet todo! v2

* more cleanup

* fix all

Signed-off-by: pyranota <pyra@duck.com>

* remove unrelated code

Signed-off-by: pyranota <pyra@duck.com>

* nits

Signed-off-by: pyranota <pyra@duck.com>

* do not serialize rawcode

Signed-off-by: pyranota <pyra@duck.com>

* cleanup

Signed-off-by: pyranota <pyra@duck.com>

* fixes

Signed-off-by: pyranota <pyra@duck.com>

* update ee repo ref

Signed-off-by: pyranota <pyra@duck.com>

* update ee repo

Signed-off-by: pyranota <pyra@duck.com>

* fix tests

Signed-off-by: pyranota <pyra@duck.com>

* Update ee-repo-ref.txt

---------

Signed-off-by: pyranota <pyra@duck.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Pyra
2025-12-04 10:11:54 +00:00
committed by GitHub
co-authored by windmill-internal-app[bot]
parent 1fea9bf943
commit fcbfd99ba9
17 changed files with 677 additions and 844 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
d54c99a1e4a08ff30c35737bd282d6d8d2b49e1f
937820a9e572c10c0ce050e1bd5eddb3180fe921
backend/tests/common/mod.rs
+4 -10
View File
@@ -684,16 +684,13 @@ pub async fn run_deployed_relative_imports(
let job = RunJob::from(JobPayload::ScriptHash {
path: "f/system/test_import".to_string(),
hash: ScriptHash(script.hash),
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
language,
priority: None,
apply_preprocessor: false,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
})
.push(&db2)
.await;
@@ -739,13 +736,10 @@ pub async fn run_preview_relative_imports(
path: Some("f/system/test_import".to_string()),
language,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
}))
.push(&db2)
.await;
backend/tests/job_payload.rs
+6 -16
View File
@@ -52,11 +52,8 @@ mod job_payload {
let result = RunJob::from(JobPayload::ScriptHash {
hash: ScriptHash(123412),
path: "f/system/hello".to_string(),
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
dedicated_worker: None,
language: ScriptLang::Deno,
@@ -86,16 +83,13 @@ mod job_payload {
let job = RunJob::from(JobPayload::ScriptHash {
hash: ScriptHash(123413),
path: "f/system/hello_with_preprocessor".to_string(),
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
language: ScriptLang::Deno,
priority: None,
apply_preprocessor: true,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
})
.run_until_complete_with(db, false, port, |id| async move {
let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id)
@@ -167,9 +161,7 @@ mod job_payload {
let result = RunJob::from(JobPayload::FlowScript {
id: flow_scripts[0],
language: ScriptLang::Deno,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
cache_ttl: None,
dedicated_worker: None,
path: "f/system/hello/test-0".into(),
@@ -187,9 +179,7 @@ mod job_payload {
let result = RunJob::from(JobPayload::FlowScript {
id: flow_scripts[1],
language: ScriptLang::Deno,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
cache_ttl: None,
dedicated_worker: None,
path: "f/system/hello/test-0".into(),
backend/tests/python_jobs.rs
+14 -26
View File
@@ -25,7 +25,10 @@ def main():
&db,
content,
ScriptLang::Python3,
vec!["# workspace-dependencies-mode: manual\n# py: 3.11.11","tiny==0.1.3"],
vec![
"# workspace-dependencies-mode: manual\n# py: 3.11.11",
"tiny==0.1.3",
],
)
.await?;
Ok(())
@@ -186,11 +189,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
dedicated_worker: None,
});
@@ -237,11 +237,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
dedicated_worker: None,
});
@@ -273,11 +270,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
dedicated_worker: None,
});
@@ -314,11 +308,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
dedicated_worker: None,
});
@@ -353,11 +344,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
dedicated_worker: None,
});
backend/tests/worker.rs
+26 -63
View File
@@ -188,9 +188,8 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default()
.into(),
is_trigger: None,
assets: None,
}
@@ -234,9 +233,8 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
concurrency_settings:
windmill_common::jobs::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}
@@ -368,9 +366,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
@@ -425,9 +421,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}.into(),
@@ -466,9 +460,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
@@ -534,9 +526,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}.into(),
@@ -864,13 +854,10 @@ func main(derp string) (string, error) {
path: None,
lock: None,
language: ScriptLang::Go,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
}))
.arg("derp", json!("world"))
.run_until_complete(&db, false, port)
@@ -903,13 +890,10 @@ fn main(world: String) -> Result<String, String> {
path: None,
lock: None,
language: ScriptLang::Rust,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
}))
.arg("world", json!("Hyrule"))
.run_until_complete(&db, false, port)
@@ -981,13 +965,10 @@ echo "hello $msg"
path: None,
lock: None,
language: ScriptLang::Bash,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
}))
.arg("msg", json!("world"))
.run_until_complete(&db, false, port)
@@ -1016,13 +997,10 @@ def main [ msg: string ] {
path: None,
lock: None,
language: ScriptLang::Nu,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
}))
.arg("msg", json!("world"))
.run_until_complete(&db, false, port)
@@ -1071,13 +1049,10 @@ def main [
path: None,
lock: None,
language: ScriptLang::Nu,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
}))
.arg("a", json!("3"))
.arg("b", json!("null"))
@@ -1135,13 +1110,10 @@ public class Main {
path: None,
lock: None,
language: ScriptLang::Java,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
}))
.arg("a", json!(3))
.arg("b", json!(3.0))
@@ -1172,13 +1144,10 @@ export async function main(a: Date) {
path: None,
lock: None,
language: ScriptLang::Bun,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.run_until_complete(&db, false, port)
@@ -1209,13 +1178,10 @@ export async function main(a: Date) {
path: None,
lock: None,
language: ScriptLang::Deno,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.run_until_complete(&db, false, port)
@@ -1247,13 +1213,10 @@ def main(a: datetime, b: bytes):
path: None,
lock: None,
language: ScriptLang::Python3,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.arg("b", json!("dGVzdA=="))
backend/windmill-api/src/flows.rs
+19 -26
View File
@@ -729,13 +729,12 @@ async fn get_flow_version_by_id(
let mut tx = user_db.begin(&authed).await?;
// First, fetch the path to perform authorization check early
let path: Option<String> = sqlx::query_scalar(
"SELECT path FROM flow_version WHERE id = $1 AND workspace_id = $2",
)
.bind(version)
.bind(&w_id)
.fetch_optional(&mut *tx)
.await?;
let path: Option<String> =
sqlx::query_scalar("SELECT path FROM flow_version WHERE id = $1 AND workspace_id = $2")
.bind(version)
.bind(&w_id)
.fetch_optional(&mut *tx)
.await?;
let path = not_found_if_none(
path,
@@ -802,13 +801,12 @@ async fn update_flow_history(
let mut tx = user_db.begin(&authed).await?;
// Fetch path and perform authorization check early
let path: Option<String> = sqlx::query_scalar(
"SELECT path FROM flow_version WHERE workspace_id = $1 AND id = $2",
)
.bind(&w_id)
.bind(version)
.fetch_optional(&mut *tx)
.await?;
let path: Option<String> =
sqlx::query_scalar("SELECT path FROM flow_version WHERE workspace_id = $1 AND id = $2")
.bind(&w_id)
.bind(version)
.fetch_optional(&mut *tx)
.await?;
let path = not_found_if_none(
path,
@@ -1482,10 +1480,9 @@ async fn archive_flow_by_path(
/// Validates that flow debouncing configuration is supported by all workers
/// Returns an error if debouncing is configured but workers are behind required version
async fn guard_flow_from_debounce_data(nf: &NewFlow) -> Result<()> {
if !*MIN_VERSION_SUPPORTS_DEBOUNCING.read().await && {
let flow_value = nf.parse_flow_value()?;
flow_value.debounce_key.is_some() || flow_value.debounce_delay_s.is_some()
} {
if !*MIN_VERSION_SUPPORTS_DEBOUNCING.read().await
&& !nf.parse_flow_value()?.debouncing_settings.is_default()
{
tracing::warn!(
"Flow debouncing configuration rejected: workers are behind minimum required version for debouncing feature"
);
@@ -1604,6 +1601,7 @@ mod tests {
ConstantDelay, ExponentialDelay, FlowModule, FlowModuleValue, FlowValue,
InputTransform, Retry, StopAfterIf,
},
jobs::{ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings},
scripts,
};
@@ -1654,11 +1652,9 @@ mod tests {
path: None,
lock: None,
tag: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
is_trigger: None,
assets: None,
concurrency_settings: ConcurrencySettingsWithCustom::default(),
}),
stop_after_if: Some(StopAfterIf {
expr: "foo = 'bar'".to_string(),
@@ -1744,17 +1740,14 @@ mod tests {
})),
preprocessor_module: None,
same_worker: false,
concurrent_limit: None,
concurrency_time_window_s: None,
skip_expr: None,
cache_ttl: None,
priority: None,
early_return: None,
concurrency_key: None,
chat_input_enabled: None,
flow_env: None,
debounce_key: None,
debounce_delay_s: None,
concurrency_settings: ConcurrencySettings::default(),
debouncing_settings: DebouncingSettings::default(),
};
let expect = serde_json::json!({
"modules": [
backend/windmill-api/src/jobs.rs
+51 -50
View File
@@ -35,7 +35,8 @@ use windmill_common::flow_conversations::add_message_to_conversation_tx;
use windmill_common::flow_status::{JobResult, RestartedFrom};
use windmill_common::jobs::{
check_tag_available_for_workspace_internal, format_completed_job_result, format_result,
DynamicInput, JobTriggerKind, ENTRYPOINT_OVERRIDE,
ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, DynamicInput,
JobTriggerKind, ENTRYPOINT_OVERRIDE,
};
use windmill_common::s3_helpers::{upload_artifact_to_store, BundleFormat};
use windmill_common::utils::{RunnableKind, WarnAfterExt};
@@ -284,10 +285,7 @@ pub fn workspaced_service() -> Router {
"/completed/import",
post(crate::jobs_export::import_completed_jobs).layer(cors.clone()),
)
.route(
"/delete",
post(crate::jobs_export::delete_jobs),
)
.route("/delete", post(crate::jobs_export::delete_jobs))
.route(
"/completed/get/:id",
get(get_completed_job).layer(cors.clone()),
@@ -4588,16 +4586,17 @@ pub async fn run_workflow_as_code(
path: job.script_path,
language: job.language.unwrap_or_else(|| ScriptLang::Deno),
lock: raw_lock,
custom_concurrency_key: windmill_queue::custom_concurrency_key(&db, &job.id)
.await
.map_err(to_anyhow)?,
concurrent_limit: job.concurrent_limit,
concurrency_time_window_s: job.concurrency_time_window_s,
concurrency_settings: windmill_common::jobs::ConcurrencySettingsWithCustom {
custom_concurrency_key: windmill_queue::custom_concurrency_key(&db, &job.id)
.await
.map_err(to_anyhow)?,
concurrent_limit: job.concurrent_limit,
concurrency_time_window_s: job.concurrency_time_window_s,
},
cache_ttl: job.cache_ttl,
dedicated_worker: None,
// TODO(debouncing): enable for this mode
custom_debounce_key: None,
debounce_delay_s: None,
debouncing_settings: DebouncingSettings::default(),
}),
Some(job.tag.clone()),
None,
@@ -5458,11 +5457,17 @@ pub async fn run_wait_result_script_by_hash(
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path: path,
custom_concurrency_key: concurrency_key,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
custom_debounce_key: debounce_key,
debounce_delay_s,
concurrency_settings: windmill_common::jobs::ConcurrencySettingsWithCustom {
custom_concurrency_key: concurrency_key,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
}
.into(),
debouncing_settings: DebouncingSettings {
custom_key: debounce_key,
delay_s: debounce_delay_s,
..Default::default() // TODO
},
cache_ttl,
language,
dedicated_worker,
@@ -5934,11 +5939,8 @@ async fn run_preview_script(
path: preview.path,
language: preview.language.unwrap_or(ScriptLang::Deno),
lock: preview.lock,
custom_concurrency_key: None,
concurrent_limit: None, // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here
concurrency_time_window_s: None, // TODO(gbouv): same as above
custom_debounce_key: None, // TODO(pyra): same as for concurrency limits.
debounce_delay_s: None,
concurrency_settings: ConcurrencySettingsWithCustom::default(), // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here
debouncing_settings: DebouncingSettings::default(), // TODO(pyra): same as for concurrency limits.
cache_ttl: None,
dedicated_worker: preview.dedicated_worker,
}),
@@ -6056,13 +6058,10 @@ async fn run_bundle_preview_script(
path: preview.path,
language: preview.language.unwrap_or(ScriptLang::Deno),
lock: preview.lock,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: preview.dedicated_worker,
custom_concurrency_key: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: ConcurrencySettingsWithCustom::default(),
debouncing_settings: DebouncingSettings::default(),
}),
PushArgs::from(&args),
authed.display_username(),
@@ -6466,19 +6465,19 @@ async fn add_batch_jobs(
add_virtual_items_if_necessary(&mut value.modules);
let flow_status = FlowStatus::new(&value);
(
None, // script_hash
path, // script_path
job_kind, // job_kind
None, // language
None, // dedicated_worker
value.concurrency_key.clone(), // custom_concurrency_key
value.concurrent_limit.clone(), // concurrent_limit
value.concurrency_time_window_s, // concurrency_time_window_s
None, // timeout
None, // raw_code
None, // raw_lock
Some(value), // raw_flow
Some(flow_status), // flow_status
None, // script_hash
path, // script_path
job_kind, // job_kind
None, // language
None, // dedicated_worker
value.concurrency_settings.concurrency_key.clone(), // custom_concurrency_key
value.concurrency_settings.concurrent_limit.clone(), // concurrent_limit
value.concurrency_settings.concurrency_time_window_s, // concurrency_time_window_s
None, // timeout
None, // raw_code
None, // raw_lock
Some(value), // raw_flow
Some(flow_status), // flow_status
)
}
"noop" => (
@@ -6847,13 +6846,10 @@ async fn run_dynamic_select(
path: None,
language: dynamic_input.x_windmill_dyn_select_lang,
lock: None,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: ConcurrencySettings::default().into(),
debouncing_settings: DebouncingSettings::default(),
}),
PushArgs::from(&request.args.unwrap_or_default()),
authed.display_username(),
@@ -6977,11 +6973,16 @@ pub async fn run_job_by_hash_inner(
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path: path,
custom_concurrency_key: concurrency_key,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
custom_debounce_key: debounce_key,
debounce_delay_s,
concurrency_settings: ConcurrencySettings {
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
},
debouncing_settings: DebouncingSettings {
custom_key: debounce_key,
delay_s: debounce_delay_s,
..Default::default()
},
cache_ttl,
language,
dedicated_worker,
backend/windmill-api/src/triggers/trigger_helpers.rs
+4 -10
View File
@@ -844,11 +844,8 @@ async fn trigger_script_with_retry_and_error_handler(
JobPayload::ScriptHash {
hash,
path,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
custom_debounce_key,
debounce_delay_s,
concurrency_settings,
debouncing_settings,
cache_ttl,
priority,
apply_preprocessor,
@@ -862,16 +859,13 @@ async fn trigger_script_with_retry_and_error_handler(
error_handler_path,
error_handler_args,
skip_handler: None,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
priority,
tag_override: tag.clone(),
apply_preprocessor,
trigger_path: Some(trigger_path),
custom_debounce_key,
debounce_delay_s,
concurrency_settings,
debouncing_settings,
},
_ => {
return Err(windmill_common::error::Error::internal_err(format!(
backend/windmill-common/src/flows.rs
+15 -39
View File
@@ -24,6 +24,7 @@ use crate::{
cache,
db::DB,
error::{Error, Result as WindmillResult},
jobs::{ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings},
more_serde::{default_empty_string, default_id, default_null, default_true, is_default},
scripts::{Schema, ScriptHash, ScriptLang},
worker::{to_raw_value, Connection},
@@ -171,18 +172,10 @@ pub struct FlowValue {
#[serde(default)]
#[serde(skip_serializing_if = "is_default")]
pub same_worker: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<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>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
#[serde(flatten)]
pub concurrency_settings: ConcurrencySettings,
#[serde(flatten)]
pub debouncing_settings: DebouncingSettings,
#[serde(skip_serializing_if = "Option::is_none")]
pub skip_expr: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -919,12 +912,8 @@ pub enum FlowModuleValue {
#[serde(skip_serializing_if = "is_none_or_empty")]
tag: Option<String>,
language: ScriptLang,
#[serde(skip_serializing_if = "Option::is_none")]
custom_concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
concurrency_time_window_s: Option<i32>,
#[serde(flatten)]
concurrency_settings: ConcurrencySettingsWithCustom,
#[serde(skip_serializing_if = "Option::is_none")]
is_trigger: Option<bool>,
#[serde(skip_serializing_if = "is_none_or_empty_vec")]
@@ -945,12 +934,8 @@ pub enum FlowModuleValue {
#[serde(skip_serializing_if = "is_none_or_empty")]
tag: Option<String>,
language: ScriptLang,
#[serde(skip_serializing_if = "Option::is_none")]
custom_concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
concurrency_time_window_s: Option<i32>,
#[serde(flatten)]
concurrency_settings: ConcurrencySettingsWithCustom,
#[serde(skip_serializing_if = "Option::is_none")]
is_trigger: Option<bool>,
#[serde(skip_serializing_if = "is_none_or_empty_vec")]
@@ -989,9 +974,6 @@ struct UntaggedFlowModuleValue {
lock: Option<String>,
tag: Option<String>,
language: Option<ScriptLang>,
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
is_trigger: Option<bool>,
id: Option<FlowNodeId>,
default_node: Option<FlowNodeId>,
@@ -1000,6 +982,8 @@ struct UntaggedFlowModuleValue {
tools: Option<Vec<AgentTool>>,
pass_flow_input_directly: Option<bool>,
squash: Option<bool>,
#[serde(flatten)]
concurrency_settings: ConcurrencySettingsWithCustom,
}
impl<'de> Deserialize<'de> for FlowModuleValue {
@@ -1074,9 +1058,7 @@ impl<'de> Deserialize<'de> for FlowModuleValue {
language: untagged
.language
.ok_or_else(|| serde::de::Error::missing_field("language"))?,
custom_concurrency_key: untagged.custom_concurrency_key,
concurrent_limit: untagged.concurrent_limit,
concurrency_time_window_s: untagged.concurrency_time_window_s,
concurrency_settings: untagged.concurrency_settings,
is_trigger: untagged.is_trigger,
assets: untagged.assets,
}),
@@ -1089,9 +1071,7 @@ impl<'de> Deserialize<'de> for FlowModuleValue {
language: untagged
.language
.ok_or_else(|| serde::de::Error::missing_field("language"))?,
custom_concurrency_key: untagged.custom_concurrency_key,
concurrent_limit: untagged.concurrent_limit,
concurrency_time_window_s: untagged.concurrency_time_window_s,
concurrency_settings: untagged.concurrency_settings,
is_trigger: untagged.is_trigger,
assets: untagged.assets,
}),
@@ -1234,11 +1214,9 @@ pub async fn resolve_module(
id,
tag,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
is_trigger,
assets,
concurrency_settings,
} = std::mem::replace(&mut val, Identity)
else {
unreachable!()
@@ -1258,11 +1236,9 @@ pub async fn resolve_module(
path: None,
tag,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
is_trigger,
assets,
concurrency_settings,
};
}
ForloopFlow { modules, modules_node, .. } | WhileloopFlow { modules, modules_node, .. } => {
backend/windmill-common/src/jobs.rs
+157 -33
View File
@@ -73,7 +73,7 @@ impl std::fmt::Display for JobTriggerKind {
}
}
#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Copy, Clone)]
#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Copy, Clone, Default)]
#[sqlx(type_name = "JOB_KIND", rename_all = "lowercase")]
#[serde(rename_all(serialize = "lowercase", deserialize = "lowercase"))]
pub enum JobKind {
@@ -88,6 +88,7 @@ pub enum JobKind {
Identity,
FlowDependencies,
AppDependencies,
#[default]
Noop,
DeploymentCallback,
FlowScript,
@@ -344,23 +345,13 @@ pub enum JobPayload {
ScriptHash {
hash: ScriptHash,
path: String,
/// Override default concurrency key
custom_concurrency_key: Option<String>,
/// How many jobs can run at the same time
concurrent_limit: Option<i32>,
/// In seconds
concurrency_time_window_s: Option<i32>,
/// If not set, will be inferred from the hash(path + step_id + inputs)
custom_debounce_key: Option<String>,
/// Debouncing delay will be determined by the first job with the key.
/// All subsequent jobs with Some will get debounced.
/// If the job has no delay, it will execute immediately, fully ignoring pending delays.
debounce_delay_s: Option<i32>,
cache_ttl: Option<i32>,
dedicated_worker: Option<bool>,
language: ScriptLang,
priority: Option<i16>,
apply_preprocessor: bool,
concurrency_settings: ConcurrencySettings,
debouncing_settings: DebouncingSettings,
},
/// Execute flow step (can be subflow only).
@@ -372,13 +363,11 @@ pub enum JobPayload {
/// Execute flow step
FlowScript {
id: FlowNodeId, // flow_node(id).
path: String,
language: ScriptLang,
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
cache_ttl: Option<i32>,
dedicated_worker: Option<bool>,
path: String,
concurrency_settings: ConcurrencySettings,
},
/// Inline App Script
@@ -458,19 +447,17 @@ pub enum JobPayload {
error_handler_path: Option<String>,
error_handler_args: Option<HashMap<String, Box<RawValue>>>,
skip_handler: Option<SkipHandler>,
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
custom_debounce_key: Option<String>,
debounce_delay_s: Option<i32>,
cache_ttl: Option<i32>,
priority: Option<i16>,
tag_override: Option<String>,
trigger_path: Option<String>,
apply_preprocessor: bool,
concurrency_settings: ConcurrencySettings,
debouncing_settings: DebouncingSettings,
},
DeploymentCallback {
path: String,
// debouncing_settings: Option<DebouncingSettings>,
},
Identity,
Noop,
@@ -479,6 +466,108 @@ pub enum JobPayload {
},
}
// TODO: Add validation logic.
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
pub struct DebouncingSettings {
#[serde(
skip_serializing_if = "Option::is_none",
rename = "debounce_key",
alias = "custom_debounce_key"
)]
/// debounce key is usually stored in the db
/// including when:
///
/// 1. User have created custom debounce key from ui or cli
/// 2. User used default one
///
/// in either cases this argument serves as reactive way of overwriting debounce key from the backend.
/// Default: hash(path + step_id + inputs)
pub custom_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none", rename = "debounce_delay_s")]
/// Debouncing delay will be determined by the first job with the key.
/// All subsequent jobs with Some will get debounced.
/// If the job has no delay, it will execute immediately, fully ignoring pending delays.
pub delay_s: Option<i32>,
#[serde(
skip_serializing_if = "Option::is_none",
rename = "max_total_debouncing_time"
)]
pub max_total_time: Option<i32>,
#[serde(
skip_serializing_if = "Option::is_none",
rename = "max_total_debounces_amount"
)]
pub max_total_amount: Option<i32>,
#[serde(
skip_serializing_if = "Option::is_none",
rename = "debounce_args_to_accumulate"
)]
/// top level arguments to preserve
/// For every debounce selected arguments will be saved
/// in the end (when job finally starts) arguments will be appended and passed to runnable
///
/// NOTE: selected args should be the lists.
pub args_to_accumulate: Option<Vec<String>>,
}
impl DebouncingSettings {
pub fn is_default(&self) -> bool {
self == &Self::default()
}
}
#[derive(Debug, Default, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct ConcurrencySettings {
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<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(Debug, Clone, Serialize, Deserialize, sqlx::FromRow, Default)]
pub struct ConcurrencySettingsWithCustom {
#[serde(skip_serializing_if = "Option::is_none")]
pub custom_concurrency_key: Option<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>,
}
impl From<ConcurrencySettings> for ConcurrencySettingsWithCustom {
fn from(
ConcurrencySettings { concurrency_key, concurrent_limit, concurrency_time_window_s }: ConcurrencySettings,
) -> Self {
ConcurrencySettingsWithCustom {
custom_concurrency_key: concurrency_key,
concurrency_time_window_s,
concurrent_limit,
}
}
}
impl From<ConcurrencySettingsWithCustom> for ConcurrencySettings {
fn from(
ConcurrencySettingsWithCustom {
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
}: ConcurrencySettingsWithCustom,
) -> Self {
ConcurrencySettings {
concurrency_key: custom_concurrency_key,
concurrency_time_window_s,
concurrent_limit,
}
}
}
#[derive(Clone, Serialize, Deserialize, Debug)]
pub struct SkipHandler {
pub path: String,
@@ -487,20 +576,49 @@ pub struct SkipHandler {
pub stop_message: String,
}
#[derive(Clone, Serialize, Deserialize, Debug, Default)]
#[derive(Clone, Deserialize, Debug, Default)]
pub struct RawCode {
pub content: String,
pub path: Option<String>,
pub hash: Option<i64>,
pub language: ScriptLang,
pub lock: Option<String>,
pub custom_concurrency_key: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub custom_debounce_key: Option<String>,
pub debounce_delay_s: Option<i32>,
pub cache_ttl: Option<i32>,
pub dedicated_worker: Option<bool>,
#[serde(flatten)]
pub concurrency_settings: ConcurrencySettingsWithCustom,
#[serde(flatten)]
// NOTE: Since we can only deserialize the struct,
// even though the older versions pass `custom_debounce_key` to RawCode,
// we can still have `debounce_key` in DebouncingSettings
// we just add alias `custom_debounce_key`
// however, serializing this settings will produce `debounce_key`
pub debouncing_settings: DebouncingSettings,
}
impl JobPayload {
pub fn job_kind(&self) -> JobKind {
match self {
JobPayload::Noop => JobKind::Noop,
JobPayload::Identity => JobKind::Identity,
JobPayload::Code { .. } => JobKind::Preview,
JobPayload::AIAgent { .. } => JobKind::AIAgent,
JobPayload::FlowNode { .. } => JobKind::FlowNode,
JobPayload::ScriptHash { .. } => JobKind::Script,
JobPayload::AppScript { .. } => JobKind::AppScript,
JobPayload::RawFlow { .. } => JobKind::FlowPreview,
JobPayload::ScriptHub { .. } => JobKind::Script_Hub,
JobPayload::FlowScript { .. } => JobKind::FlowScript,
JobPayload::Dependencies { .. } => JobKind::Dependencies,
JobPayload::SingleStepFlow { .. } => JobKind::SingleStepFlow,
JobPayload::AppDependencies { .. } => JobKind::AppDependencies,
JobPayload::FlowDependencies { .. } => JobKind::FlowDependencies,
JobPayload::RawScriptDependencies { .. } => JobKind::Dependencies,
JobPayload::RawFlowDependencies { .. } => JobKind::FlowDependencies,
JobPayload::DeploymentCallback { .. } => JobKind::DeploymentCallback,
JobPayload::Flow { .. } | JobPayload::RestartedFlow { .. } => JobKind::Flow,
}
}
}
type Tag = String;
@@ -597,17 +715,23 @@ pub async fn script_path_to_payload<'e>(
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path: script_path.to_owned(),
custom_concurrency_key: concurrency_key,
concurrent_limit,
concurrency_time_window_s,
custom_debounce_key: debounce_key,
debounce_delay_s,
cache_ttl,
language,
dedicated_worker,
priority,
apply_preprocessor: !skip_preprocessor.unwrap_or(false)
&& has_preprocessor.unwrap_or(false),
concurrency_settings: ConcurrencySettingsWithCustom {
custom_concurrency_key: concurrency_key,
concurrent_limit,
concurrency_time_window_s,
}
.into(),
debouncing_settings: DebouncingSettings {
custom_key: debounce_key,
delay_s: debounce_delay_s,
..Default::default()
},
},
tag,
delete_after_use,
backend/windmill-common/src/scripts.rs
+4
View File
@@ -495,6 +495,10 @@ pub struct NewScript {
pub tag: Option<String>,
pub draft_only: Option<bool>,
pub envs: Option<Vec<String>>,
// NOTE: concurrency and debounce data is inline,
// bc it was this before refactor
// and rust seems to hash it differently
// for backwards compat we keep them inline
pub concurrency_key: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
backend/windmill-queue/src/jobs.rs
+314 -477
View File
File diff suppressed because it is too large Load Diff
backend/windmill-queue/src/schedule.rs
+16 -15
View File
@@ -20,6 +20,8 @@ use windmill_common::flows::Retry;
use windmill_common::get_flow_version_info_from_version;
use windmill_common::get_latest_flow_version_id_for_path;
use windmill_common::jobs::check_tag_available_for_workspace_internal;
use windmill_common::jobs::ConcurrencySettings;
use windmill_common::jobs::DebouncingSettings;
use windmill_common::jobs::JobPayload;
use windmill_common::schedule::schedule_to_user;
use windmill_common::scripts::ScriptHash;
@@ -258,16 +260,13 @@ pub async fn push_scheduled_job<'c>(
stop_condition,
stop_message,
}),
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
priority: None,
tag_override: schedule.tag.clone(),
trigger_path: None,
apply_preprocessor: false,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: ConcurrencySettings::default(),
debouncing_settings: DebouncingSettings::default(),
},
if schedule.tag.as_ref().is_some_and(|x| x != "") {
schedule.tag.clone()
@@ -360,16 +359,13 @@ pub async fn push_scheduled_job<'c>(
error_handler_args: None,
skip_handler: None,
args: static_args,
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl,
priority,
tag_override: schedule.tag.clone(),
trigger_path: None,
apply_preprocessor: false,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings: ConcurrencySettings::default(),
debouncing_settings: DebouncingSettings::default(),
},
if schedule.tag.as_ref().is_some_and(|x| x != "") {
schedule.tag.clone()
@@ -385,16 +381,21 @@ pub async fn push_scheduled_job<'c>(
JobPayload::ScriptHash {
hash,
path: schedule.script_path.clone(),
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
dedicated_worker,
language,
priority,
apply_preprocessor: false,
custom_debounce_key,
debounce_delay_s,
debouncing_settings: DebouncingSettings {
custom_key: custom_debounce_key,
delay_s: debounce_delay_s,
..Default::default()
},
concurrency_settings: ConcurrencySettings {
concurrency_key: custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
},
},
if schedule.tag.as_ref().is_some_and(|x| x != "") {
schedule.tag.clone()
backend/windmill-worker/src/ai/tools.rs
+4 -18
View File
@@ -366,9 +366,7 @@ async fn execute_windmill_tool(
language,
lock,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
concurrency_settings,
..
} => {
let path = path
@@ -379,32 +377,20 @@ async fn execute_windmill_tool(
content,
language,
lock,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
concurrency_settings,
tool_module,
tag,
tool_module.delete_after_use.unwrap_or(false),
)
}
FlowModuleValue::FlowScript {
id,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
tag,
..
} => {
FlowModuleValue::FlowScript { id, language, concurrency_settings, tag, .. } => {
let path = format!("{}/tools/{}", ctx.job.runnable_path(), tool_module.id);
let payload = JobPayloadWithTag {
payload: JobPayload::FlowScript {
id,
language,
custom_concurrency_key: custom_concurrency_key.clone(),
concurrent_limit,
concurrency_time_window_s,
concurrency_settings: concurrency_settings.into(),
cache_ttl: tool_module.cache_ttl.map(|x| x as i32),
dedicated_worker: None,
path,
backend/windmill-worker/src/worker_flow.rs
+29 -43
View File
@@ -43,7 +43,8 @@ use windmill_common::flow_status::{
};
use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId, StopAfterIf};
use windmill_common::jobs::{
script_path_to_payload, JobKind, JobPayload, OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE,
script_path_to_payload, ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings,
JobKind, JobPayload, OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE,
};
use windmill_common::scripts::ScriptHash;
use windmill_common::users::username_to_permissioned_as;
@@ -3956,9 +3957,7 @@ async fn compute_next_flow_transform(
language,
lock,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
concurrency_settings,
..
} => {
let path = path.unwrap_or_else(|| get_path(flow_job, status, module));
@@ -3968,9 +3967,7 @@ async fn compute_next_flow_transform(
content,
language,
lock,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
concurrency_settings,
module,
tag,
delete_after_use,
@@ -3984,9 +3981,7 @@ async fn compute_next_flow_transform(
id, // flow_node(id).
tag,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
concurrency_settings,
..
} => {
let path = get_path(flow_job, status, module);
@@ -3995,9 +3990,7 @@ async fn compute_next_flow_transform(
payload: JobPayload::FlowScript {
id,
language,
custom_concurrency_key: custom_concurrency_key.clone(),
concurrent_limit,
concurrency_time_window_s,
concurrency_settings: concurrency_settings.into(),
cache_ttl: module.cache_ttl.map(|x| x as i32),
dedicated_worker: None,
path,
@@ -4625,18 +4618,14 @@ async fn payload_from_simple_module(
language,
lock,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
concurrency_settings,
..
} => raw_script_to_payload(
path.unwrap_or_else(|| inner_path),
content,
language,
lock,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
concurrency_settings,
module,
tag,
delete_after_use,
@@ -4645,20 +4634,16 @@ async fn payload_from_simple_module(
id, // flow_node(id).
tag,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
concurrency_settings,
..
} => JobPayloadWithTag {
payload: JobPayload::FlowScript {
id,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl: module.cache_ttl.map(|x| x as i32),
dedicated_worker: None,
path: inner_path,
concurrency_settings: concurrency_settings.into(),
},
tag,
delete_after_use,
@@ -4674,9 +4659,7 @@ pub fn raw_script_to_payload(
content: String,
language: windmill_common::scripts::ScriptLang,
lock: Option<String>,
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
concurrency_settings: ConcurrencySettingsWithCustom,
module: &FlowModule,
tag: Option<String>,
delete_after_use: bool,
@@ -4688,13 +4671,11 @@ pub fn raw_script_to_payload(
content,
language,
lock,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
cache_ttl: module.cache_ttl.map(|x| x as i32),
dedicated_worker: None,
custom_debounce_key: None,
debounce_delay_s: None,
concurrency_settings,
// TODO: Should this have debouncing?
debouncing_settings: DebouncingSettings::default(),
}),
tag,
delete_after_use,
@@ -4756,11 +4737,6 @@ pub async fn script_to_payload(
let ScriptHashInfo {
tag,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
debounce_key,
debounce_delay_s,
cache_ttl,
language,
dedicated_worker,
@@ -4769,6 +4745,11 @@ pub async fn script_to_payload(
timeout,
on_behalf_of_email,
created_by,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
debounce_key,
debounce_delay_s,
..
} = get_script_info_for_hash(None, db, &flow_job.workspace_id, hash.0).await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
@@ -4784,11 +4765,16 @@ pub async fn script_to_payload(
JobPayload::ScriptHash {
hash,
path: script_path,
custom_concurrency_key: concurrency_key,
concurrent_limit,
concurrency_time_window_s,
custom_debounce_key: debounce_key,
debounce_delay_s,
debouncing_settings: DebouncingSettings {
custom_key: debounce_key,
delay_s: debounce_delay_s,
..Default::default()
},
concurrency_settings: ConcurrencySettings {
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
},
cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(),
language,
dedicated_worker,
backend/windmill-worker/src/worker_lockfiles.rs
+4 -12
View File
@@ -1171,11 +1171,9 @@ async fn lock_modules<'c>(
mut language,
input_transforms,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
is_trigger,
assets,
concurrency_settings,
} = e.get_value()?
else {
let mut nmodified_ids = Vec::new();
@@ -1541,11 +1539,9 @@ async fn lock_modules<'c>(
content,
language,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
is_trigger,
assets,
concurrency_settings,
});
new_flow_modules.push(e);
@@ -1737,11 +1733,9 @@ async fn reduce_flow<'c>(
language,
input_transforms,
tag,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
is_trigger,
assets,
concurrency_settings,
..
} = std::mem::replace(&mut val, Identity)
else {
@@ -1764,11 +1758,9 @@ async fn reduce_flow<'c>(
id,
tag,
language,
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
is_trigger,
assets,
concurrency_settings,
};
}
ForloopFlow { modules, modules_node, .. }
backend/windmill-worker/src/worker_utils.rs
+9 -5
View File
@@ -395,15 +395,19 @@ pub async fn get_tag_and_concurrency(job_id: &Uuid, db: &DB) -> Option<TagAndCon
Err(_) => cache::flow::fetch_version(db, version).await,
};
let flow_value = flow.map(|f| f.value().clone()).ok();
let concurrency_key = flow_value
.as_ref()
.map(|fv| fv.concurrency_key.clone())
.flatten();
let concurrent_limit = flow_value.as_ref().map(|fv| fv.concurrent_limit).flatten();
.and_then(|fv| fv.concurrency_settings.concurrency_key.to_owned());
let concurrent_limit = flow_value
.as_ref()
.and_then(|fv| fv.concurrency_settings.concurrent_limit);
let concurrent_time_window_s = flow_value
.as_ref()
.map(|fv| fv.concurrency_time_window_s)
.flatten();
.and_then(|fv| fv.concurrency_settings.concurrency_time_window_s);
Some(TagAndConcurrencyKey {
tag: tag_and_concurrency_key.tag,
concurrency_key,