diff --git a/backend/.sqlx/query-0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d.json b/backend/.sqlx/query-0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d.json new file mode 100644 index 0000000000..ecaf828737 --- /dev/null +++ b/backend/.sqlx/query-0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d.json @@ -0,0 +1,94 @@ +{ + "db_name": "PostgreSQL", + "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, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Int8", + "Varchar", + "Int8Array", + "Text", + "Text", + "Text", + "Varchar", + "Text", + "Bool", + "Jsonb", + "Text", + { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby" + ] + } + } + }, + { + "Custom": { + "name": "script_kind", + "kind": { + "Enum": [ + "script", + "trigger", + "failure", + "command", + "approval", + "preprocessor" + ] + } + } + }, + "Varchar", + "Bool", + "VarcharArray", + "Int4", + "Int4", + "Int4", + "Bool", + "Bool", + "Int2", + "Bool", + "Bool", + "Int4", + "Varchar", + "Bool", + "Bool", + "Varchar", + "Bool", + "Text", + "Bool", + "Jsonb", + "Varchar", + "Int4" + ] + }, + "nullable": [] + }, + "hash": "0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d" +} diff --git a/backend/.sqlx/query-27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32.json b/backend/.sqlx/query-27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32.json new file mode 100644 index 0000000000..c540742fe2 --- /dev/null +++ b/backend/.sqlx/query-27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32.json @@ -0,0 +1,133 @@ +{ + "db_name": "PostgreSQL", + "query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script\n WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false)\n ORDER BY created_at DESC LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "hash", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "concurrency_key", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 4, + "name": "concurrency_time_window_s", + "type_info": "Int4" + }, + { + "ordinal": 5, + "name": "debounce_key", + "type_info": "Varchar" + }, + { + "ordinal": 6, + "name": "debounce_delay_s", + "type_info": "Int4" + }, + { + "ordinal": 7, + "name": "cache_ttl", + "type_info": "Int4" + }, + { + "ordinal": 8, + "name": "language: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby" + ] + } + } + } + }, + { + "ordinal": 9, + "name": "dedicated_worker", + "type_info": "Bool" + }, + { + "ordinal": 10, + "name": "priority", + "type_info": "Int2" + }, + { + "ordinal": 11, + "name": "timeout", + "type_info": "Int4" + }, + { + "ordinal": 12, + "name": "on_behalf_of_email", + "type_info": "Text" + }, + { + "ordinal": 13, + "name": "created_by", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Bool" + ] + }, + "nullable": [ + false, + true, + true, + true, + true, + true, + true, + true, + false, + true, + true, + true, + true, + false + ] + }, + "hash": "27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32" +} diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf.json b/backend/.sqlx/query-f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf.json new file mode 100644 index 0000000000..4e7b02738e --- /dev/null +++ b/backend/.sqlx/query-f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf.json @@ -0,0 +1,150 @@ +{ + "db_name": "PostgreSQL", + "query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by, path from script where hash = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "hash", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "concurrency_key", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 4, + "name": "concurrency_time_window_s", + "type_info": "Int4" + }, + { + "ordinal": 5, + "name": "debounce_key", + "type_info": "Varchar" + }, + { + "ordinal": 6, + "name": "debounce_delay_s", + "type_info": "Int4" + }, + { + "ordinal": 7, + "name": "cache_ttl", + "type_info": "Int4" + }, + { + "ordinal": 8, + "name": "language: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby" + ] + } + } + } + }, + { + "ordinal": 9, + "name": "dedicated_worker", + "type_info": "Bool" + }, + { + "ordinal": 10, + "name": "priority", + "type_info": "Int2" + }, + { + "ordinal": 11, + "name": "delete_after_use", + "type_info": "Bool" + }, + { + "ordinal": 12, + "name": "timeout", + "type_info": "Int4" + }, + { + "ordinal": 13, + "name": "has_preprocessor", + "type_info": "Bool" + }, + { + "ordinal": 14, + "name": "on_behalf_of_email", + "type_info": "Text" + }, + { + "ordinal": 15, + "name": "created_by", + "type_info": "Varchar" + }, + { + "ordinal": 16, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + }, + "nullable": [ + false, + true, + true, + true, + true, + true, + true, + true, + false, + true, + true, + true, + true, + true, + true, + false, + false + ] + }, + "hash": "f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf" +} diff --git a/backend/.sqlx/query-f1fe8508f6cd29d4c0d354473529fc46f506ddf98d45fc870e2855c4a4b424ea.json b/backend/.sqlx/query-f1fe8508f6cd29d4c0d354473529fc46f506ddf98d45fc870e2855c4a4b424ea.json new file mode 100644 index 0000000000..effa800640 --- /dev/null +++ b/backend/.sqlx/query-f1fe8508f6cd29d4c0d354473529fc46f506ddf98d45fc870e2855c4a4b424ea.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO debounce_key (key, job_id)\n VALUES ($1, $2)\n ON CONFLICT (key)\n DO UPDATE SET job_id = debounce_key.job_id -- No actual change, just to trigger UPDATE\n RETURNING CASE WHEN xmax != 0 THEN job_id ELSE NULL END AS job_id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "job_id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Varchar", + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "f1fe8508f6cd29d4c0d354473529fc46f506ddf98d45fc870e2855c4a4b424ea" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index f9eb763340..9afb2d6c1e 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -c8d57987b72fd15f2e3b8d7f6501edaac8235965 \ No newline at end of file +e12fc17ec760fe720edc2d930457ebb0e06ede48 diff --git a/backend/migrations/20251017120625_job_debouncing_extra.down.sql b/backend/migrations/20251017120625_job_debouncing_extra.down.sql new file mode 100644 index 0000000000..b2e9df943e --- /dev/null +++ b/backend/migrations/20251017120625_job_debouncing_extra.down.sql @@ -0,0 +1,4 @@ +-- Add down migration script here +ALTER TABLE script DROP COLUMN IF EXISTS debounce_key; +ALTER TABLE script DROP COLUMN IF EXISTS debounce_delay_s; + diff --git a/backend/migrations/20251017120625_job_debouncing_extra.up.sql b/backend/migrations/20251017120625_job_debouncing_extra.up.sql new file mode 100644 index 0000000000..0d81abf65b --- /dev/null +++ b/backend/migrations/20251017120625_job_debouncing_extra.up.sql @@ -0,0 +1,6 @@ +-- Job debouncing feature: consolidate multiple job requests within a time window +-- This reduces redundant work when the same script/flow is triggered multiple times rapidly +-- debounce_key: Custom key template for grouping jobs (e.g., "$workspace/$path-$args[id]") +-- debounce_delay_s: Delay in seconds before job execution to allow consolidation window +ALTER TABLE script ADD COLUMN IF NOT EXISTS debounce_key VARCHAR(255); +ALTER TABLE script ADD COLUMN IF NOT EXISTS debounce_delay_s INTEGER; diff --git a/backend/src/main.rs b/backend/src/main.rs index f1bc4c334d..70f428d4ca 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -423,7 +423,7 @@ async fn windmill_main() -> anyhow::Result<()> { ); let suffix = create_default_worker_suffix(&hostname); ( - Connection::Http(build_agent_http_client(&suffix)), + Connection::Http(build_agent_http_client(&suffix, None, None)), Some(suffix), ) } else { @@ -774,7 +774,7 @@ Windmill Community Edition {GIT_VERSION} conn: if i == 0 || mode != Mode::Agent { conn.clone() } else { - Connection::Http(build_agent_http_client(&suffix)) + Connection::Http(build_agent_http_client(&suffix, None, None)) }, worker_name: worker_name_with_suffix( mode == Mode::Agent, diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 13cecca007..1a18588802 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -70,9 +70,9 @@ use windmill_common::{ load_env_vars, load_init_bash_from_env, load_periodic_bash_script_from_env, load_periodic_bash_script_interval_from_env, load_whitelist_env_vars_from_env, load_worker_config, reload_custom_tags_setting, store_pull_query, - store_suspended_pull_query, update_min_version, Connection, WorkerConfig, - DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, SCRIPT_TOKEN_EXPIRY, - SMTP_CONFIG, TMP_DIR, WORKER_CONFIG, WORKER_GROUP, + store_suspended_pull_query, Connection, WorkerConfig, DEFAULT_TAGS_PER_WORKSPACE, + DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, SCRIPT_TOKEN_EXPIRY, SMTP_CONFIG, TMP_DIR, + WORKER_CONFIG, WORKER_GROUP, }, KillpillSender, BASE_URL, CRITICAL_ALERTS_ON_DB_OVERSIZE, CRITICAL_ALERT_MUTE_UI_ENABLED, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS, @@ -1590,7 +1590,7 @@ pub async fn monitor_db( }; let cleanup_debounce_keys_f = async { - if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(20) { + if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(10) { if let Some(db) = conn.as_sql() { if let Err(e) = cleanup_debounce_orphaned_keys(&db).await { tracing::error!("Error cleaning up debounce keys: {:?}", e); @@ -1700,7 +1700,8 @@ pub async fn monitor_db( }; let update_min_worker_version_f = async { - update_min_version(conn).await; + #[cfg(not(feature = "test_job_debouncing"))] + windmill_common::worker::update_min_version(conn).await; }; join!( diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 73db6d1f62..475a8540b7 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -15,7 +15,7 @@ use windmill_common::{ jobs::{JobKind, JobPayload, RawCode}, jwt::JWT_SECRET, scripts::{ScriptHash, ScriptLang}, - worker::WORKER_CONFIG, + worker::{Connection, WORKER_CONFIG}, KillpillSender, }; use windmill_queue::PushIsolationLevel; @@ -31,6 +31,21 @@ pub async fn init_client(db: Pool) -> (windmill_api_client::Client, u1 (client, port, server) } +pub async fn init_client_agent_mode( + db: Pool, +) -> (windmill_api_client::Client, u16, ApiServer) { + initialize_tracing().await; + set_jwt_secret().await; + + let server = ApiServer::start_agent_mode(db).await.unwrap(); + let port = server.addr.port(); + let client = windmill_api_client::create_client( + &format!("http://localhost:{port}"), + "SECRET_TOKEN".to_string(), + ); + (client, port, server) +} + /// it's important this is unique between tests as there is one prometheus registry and /// run_worker shouldn't register the same metric with the same worker name more than once. /// @@ -66,6 +81,14 @@ pub struct ApiServer { impl ApiServer { pub async fn start(db: Pool) -> anyhow::Result { + Self::start_inner(db, false).await + } + + pub async fn start_agent_mode(db: Pool) -> anyhow::Result { + Self::start_inner(db, true).await + } + + async fn start_inner(db: Pool, agent_mode: bool) -> anyhow::Result { let (tx, rx) = tokio::sync::broadcast::channel::<()>(1); let sock = tokio::net::TcpListener::bind("127.0.0.1:0") @@ -86,7 +109,7 @@ impl ApiServer { addr, rx, port_tx, - false, + agent_mode, false, format!("http://localhost:{}", addr.port()), Some(name.clone()), @@ -116,14 +139,17 @@ impl ApiServer { } } +#[derive(Debug, Clone)] pub struct RunJob { pub payload: JobPayload, pub args: serde_json::Map, + pub debounce_job_id_o: Option, + pub scheduled_for_o: Option>, } impl From for RunJob { fn from(payload: JobPayload) -> Self { - Self { payload, args: Default::default() } + Self { payload, args: Default::default(), debounce_job_id_o: None, scheduled_for_o: None } } } @@ -133,8 +159,21 @@ impl RunJob { self } + pub fn push_arg_debounce_job_id_o(mut self, job_id: Option) -> Self { + self.debounce_job_id_o = job_id; + self + } + + pub fn push_arg_scheduled_for_o( + mut self, + scheduled_for_o: Option>, + ) -> Self { + self.scheduled_for_o = scheduled_for_o; + self + } + pub async fn push(self, db: &Pool) -> Uuid { - let RunJob { payload, args } = self; + let RunJob { payload, args, debounce_job_id_o, scheduled_for_o } = self; let mut hm_args = std::collections::HashMap::new(); for (k, v) in args { hm_args.insert(k, windmill_common::worker::to_raw_value(&v)); @@ -151,7 +190,7 @@ impl RunJob { /* email */ "test@windmill.dev", /* permissioned_as */ "u/test-user".to_string(), /* token_prefix */ None, - /* scheduled_for_o */ None, + scheduled_for_o, /* schedule_path */ None, /* parent_job */ None, /* root job */ None, @@ -168,7 +207,7 @@ impl RunJob { None, false, None, - None, + debounce_job_id_o, ) .await .expect("push has to succeed"); @@ -178,11 +217,27 @@ impl RunJob { } /// push the job, spawn a worker, wait until the job is in completed_job - pub async fn run_until_complete(self, db: &Pool, port: u16) -> CompletedJob { + pub async fn run_until_complete( + self, + db: &Pool, + agent_mode: bool, + port: u16, + ) -> CompletedJob { let uuid = self.push(db).await; let listener = listen_for_completed_jobs(db).await; - in_test_worker(db, listener.find(&uuid), port).await; - + + let conn = match agent_mode { + false => Connection::Sql(db.clone()), + #[cfg(all(feature = "private", feature = "agent_worker_server"))] + true => testing_http_connection(port).await, + #[cfg(not(all(feature = "private", feature = "agent_worker_server")))] + true => { + panic!("to use agent worker test, you need to enable 'agent_worker_server' feature") + } + }; + + in_test_worker(conn, listener.find(&uuid), port).await; + completed_job(uuid, db).await } @@ -190,36 +245,53 @@ impl RunJob { pub async fn run_until_complete_with>( self, db: &Pool, + agent_mode: bool, port: u16, test: impl Fn(Uuid) -> F, ) -> CompletedJob { let uuid = self.push(db).await; let listener = listen_for_completed_jobs(db).await; test(uuid).await; - in_test_worker(db, listener.find(&uuid), port).await; - + + let conn = match agent_mode { + false => Connection::Sql(db.clone()), + #[cfg(all(feature = "private", feature = "agent_worker_server"))] + true => testing_http_connection(port).await, + #[cfg(not(all(feature = "private", feature = "agent_worker_server")))] + true => { + panic!("to use agent worker test, you need to enable 'agent_worker_server' feature") + } + }; + + in_test_worker(conn, listener.find(&uuid), port).await; + completed_job(uuid, db).await } } pub async fn run_job_in_new_worker_until_complete( db: &Pool, + agent_mode: bool, job: JobPayload, port: u16, ) -> CompletedJob { - RunJob::from(job).run_until_complete(db, port).await + RunJob::from(job) + .run_until_complete(db, agent_mode, port) + .await } /// Start a worker with a timeout and run a future, until the worker quits or we time out. /// /// Cleans up the worker before resolving. pub async fn in_test_worker( - db: &Pool, + // db: &Pool, + // If set to http, worker will be started in agent mode. + conn: impl Into, inner: Fut, port: u16, ) -> ::Output { set_jwt_secret().await; - let (quit, worker) = spawn_test_worker(db, port); + let (quit, worker) = spawn_test_worker(&conn.into(), port); let worker = tokio::time::timeout(std::time::Duration::from_secs(60), worker); tokio::pin!(worker); @@ -244,7 +316,7 @@ pub async fn in_test_worker( } pub fn spawn_test_worker( - db: &Pool, + conn: &Connection, port: u16, ) -> (KillpillSender, tokio::task::JoinHandle<()>) { std::fs::DirBuilder::new() @@ -253,10 +325,10 @@ pub fn spawn_test_worker( .expect("could not create initial worker dir"); let (tx, rx) = KillpillSender::new(1); - let db = db.to_owned(); let worker_instance: &str = "test worker instance"; let worker_name: String = next_worker_name(); let ip: &str = Default::default(); + let conn = conn.to_owned(); let tx2 = tx.clone(); let future = async move { @@ -272,7 +344,7 @@ pub fn spawn_test_worker( windmill_common::worker::store_pull_query(&wc).await; } windmill_worker::run_worker( - &db.into(), + &conn, worker_instance, worker_name, 1, @@ -609,6 +681,8 @@ pub async fn run_deployed_relative_imports( language, priority: None, apply_preprocessor: false, + custom_debounce_key: None, + debounce_delay_s: None, }) .push(&db2) .await; @@ -646,7 +720,7 @@ pub async fn run_preview_relative_imports( let mut completed = listen_for_completed_jobs(db).await; let db2 = db.clone(); in_test_worker( - db, + db.clone(), async move { let job = RunJob::from(JobPayload::Code(RawCode { hash: None, @@ -659,6 +733,8 @@ pub async fn run_preview_relative_imports( concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .push(&db2) .await; @@ -684,6 +760,29 @@ pub async fn run_preview_relative_imports( Ok(()) } +#[cfg(all(feature = "private", feature = "agent_worker_server"))] +pub async fn testing_http_connection(port: u16) -> Connection { + let suffix = windmill_common::utils::create_default_worker_suffix("test-agent-worker"); + Connection::Http(windmill_common::agent_workers::build_agent_http_client( + &suffix, + Some(format!( + "{}{}", + windmill_common::agent_workers::AGENT_JWT_PREFIX, + windmill_common::jwt::encode_with_internal_secret( + windmill_api::agent_workers_ee::AgentAuth { + worker_group: "testing-agent".to_owned(), + suffix: Some(suffix.clone()), + tags: vec!["flow".into(), "python3".into(), "dependency".into()], + exp: Some(usize::MAX), + } + ) + .await + .expect("JWT token to be created") + )), + Some(format!("http://localhost:{port}")), + )) +} + /// IMPORTANT!: /// Do not run parallel in tests! /// diff --git a/backend/tests/fixtures/job_debouncing.sql b/backend/tests/fixtures/job_debouncing.sql new file mode 100644 index 0000000000..480d7c29a7 --- /dev/null +++ b/backend/tests/fixtures/job_debouncing.sql @@ -0,0 +1,171 @@ +-- SCRIPTS -- +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'test-user', +'def main(x: str = "hey", b: int = 1): + pass +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +'', +'', +'f/scripts/script_1', 533400, 'python3', ''); +-- Padded Hex: 0000000000082398 + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'test-user', +'def main(): + pass +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +'', +'', +'f/scripts/script_2', 533403, 'python3', ''); +-- Padded Hex: 000000000008239B + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'test-user', +' +def main(): + pass +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +'', +'', +'f/scripts/script_3', 533404, 'python3', ''); + + +-- Padded Hex: 000000000008239C +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', +'', +'', +'f/flows/flow', +'{1443253234253454}', +'{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "properties": {}, + "required": [], + "type": "object" +}', +$tag$ +{ + "modules": [ + { + "id": "a", + "value": { + "type": "rawscript", + "assets": [], + "content": "def main(x: str, y: str):\n return x", + "language": "python3", + "debounce_delay_s": 2, + "input_transforms": { + "x": { + "type": "static", + "value": "" + }, + "y": { + "type": "static", + "value": "" + } + } + }, + "continue_on_error": false + } + ], + "debounce_delay_s": 2 +}$tag$, +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) +SELECT versions[1], workspace_id, path, schema, value, edited_by FROM flow WHERE path = 'f/flows/flow'; + +-- No top level debouncing +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', +'', +'', +'f/flows/flow_full', +'{123}', +'{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "properties": {}, + "required": [], + "type": "object" +}', +$tag$ +{ + "modules": [ + { + "id": "a", + "value": { + "lock": "# py: 3.11\n", + "type": "rawscript", + "assets": [], + "content": "import time\n\ndef main(x: str, y: str):\n time.sleep(30)\n\n return x", + "language": "python3", + "concurrent_limit": 1, + "input_transforms": { + "x": { + "type": "static", + "value": "" + }, + "y": { + "type": "static", + "value": "" + } + }, + "concurrency_time_window_s": 5 + }, + "continue_on_error": false + }, + { + "id": "b", + "value": { + "type": "whileloopflow", + "modules": [ + { + "id": "c", + "value": { + "lock": "# py: 3.11\n", + "type": "rawscript", + "assets": [], + "content": "# import wmill\n\n\ndef main(x: str):\n return x", + "language": "python3", + "input_transforms": { + "x": { + "type": "static", + "value": "" + } + } + } + }, + { + "id": "d", + "value": { + "type": "rawscript", + "assets": [], + "content": "# import wmill\n\n\ndef main(x: str):\n return x", + "language": "python3", + "input_transforms": { + "x": { + "type": "static", + "value": "" + } + } + } + } + ], + "skip_failures": false + } + } + ] +}$tag$, +'system' +); + + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) +SELECT versions[1], workspace_id, path, schema, value, edited_by FROM flow WHERE path = 'f/flows/flow_full'; diff --git a/backend/tests/job_payload.rs b/backend/tests/job_payload.rs index ace3e61ef0..22e713329b 100644 --- a/backend/tests/job_payload.rs +++ b/backend/tests/job_payload.rs @@ -1,20 +1,19 @@ mod common; - mod job_payload { use serde_json::json; + use sqlx::{Pool, Postgres}; use std::sync::Arc; use tokio::sync::RwLock; - use sqlx::{Pool, Postgres}; - use windmill_common::scripts::{ScriptHash, ScriptLang}; - use windmill_common::jobs::JobPayload; - use windmill_common::flows::{FlowValue, FlowModule, FlowModuleValue}; use windmill_common::flow_status::RestartedFrom; - + use windmill_common::flows::{FlowModule, FlowModuleValue, FlowValue}; + use windmill_common::jobs::JobPayload; + use windmill_common::scripts::{ScriptHash, ScriptLang}; + + use crate::common::*; use windmill_common::worker::{ MIN_VERSION_IS_AT_LEAST_1_427, MIN_VERSION_IS_AT_LEAST_1_432, MIN_VERSION_IS_AT_LEAST_1_440, }; - use crate::common::*; pub async fn initialize_tracing() { use std::sync::Once; @@ -42,11 +41,9 @@ mod job_payload { ]; } - #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base", "hello"))] async fn test_script_hash_payload(db: Pool) -> anyhow::Result<()> { - initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); @@ -58,6 +55,8 @@ mod job_payload { custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, + custom_debounce_key: None, + debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, language: ScriptLang::Deno, @@ -65,7 +64,7 @@ mod job_payload { apply_preprocessor: false, }) .arg("world", json!("foo")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -95,8 +94,10 @@ mod job_payload { language: ScriptLang::Deno, priority: None, apply_preprocessor: true, + custom_debounce_key: None, + debounce_delay_s: None, }) - .run_until_complete_with(db, port, |id| async move { + .run_until_complete_with(db, false, port, |id| async move { let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id) .fetch_one(db) .await @@ -131,7 +132,7 @@ mod job_payload { dedicated_worker: None, version: 1443253234253454, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -174,7 +175,7 @@ mod job_payload { path: "f/system/hello/test-0".into(), }) .arg("world", json!("foo")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -194,7 +195,7 @@ mod job_payload { path: "f/system/hello/test-0".into(), }) .arg("hello", json!("You know nothing Jean Neige")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -221,7 +222,7 @@ mod job_payload { dedicated_worker: None, version: 1443253234253454, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -242,7 +243,7 @@ mod job_payload { path: "f/system/hello_with_nodes_flow/forloop-0".into(), }) .arg("iter", json!({ "value": "tests", "index": 0 })) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -264,7 +265,7 @@ mod job_payload { language: ScriptLang::Deno, dedicated_worker: None, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -308,7 +309,7 @@ mod job_payload { dedicated_worker: None, version: 1443253234253454, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -357,7 +358,7 @@ mod job_payload { .unwrap(), }) .arg("skip_flow_update", json!(true)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -371,7 +372,7 @@ mod job_payload { restarted_from: None, }) .arg("world", json!("Jean Neige")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -399,7 +400,7 @@ mod job_payload { .into(), language: ScriptLang::Deno, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -427,7 +428,7 @@ mod job_payload { apply_preprocessor: false, version: 1443253234253454, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -449,7 +450,7 @@ mod job_payload { dedicated_worker: None, version: 1443253234253454, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -475,7 +476,7 @@ mod job_payload { apply_preprocessor: true, version: 1443253234253456, }) - .run_until_complete_with(db, port, |id| async move { + .run_until_complete_with(db, false, port, |id| async move { let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id) .fetch_one(db) .await @@ -521,7 +522,7 @@ mod job_payload { dedicated_worker: None, version: 1443253234253456, }) - .run_until_complete(db, port) + .run_until_complete(db, false, port) .await .json_result() .unwrap(); @@ -544,7 +545,7 @@ mod job_payload { apply_preprocessor: true, version: 1443253234253454, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .id; @@ -554,7 +555,7 @@ mod job_payload { branch_or_iteration_n: None, }) .arg("iter", json!({ "value": "tests", "index": 0 })) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -576,7 +577,7 @@ mod job_payload { dedicated_worker: None, version: 1443253234253454, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -622,7 +623,7 @@ mod job_payload { restarted_from: None, }) .arg("world", json!("Jean Neige")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -702,7 +703,7 @@ mod job_payload { restarted_from, }) .arg("world", arg) - .run_until_complete(db, port) + .run_until_complete(db, false, port) .await; assert_eq!(job.json_result().unwrap(), result); diff --git a/backend/tests/python_jobs.rs b/backend/tests/python_jobs.rs index cc7d1cb5b9..afa662f582 100644 --- a/backend/tests/python_jobs.rs +++ b/backend/tests/python_jobs.rs @@ -181,11 +181,13 @@ def main(): custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, + custom_debounce_key: None, + debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, }); - let result = run_job_in_new_worker_until_complete(&db, job, port) + let result = run_job_in_new_worker_until_complete(&db, false, job, port) .await .json_result() .unwrap(); @@ -230,11 +232,13 @@ def main(): custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, + custom_debounce_key: None, + debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, }); - let result = run_job_in_new_worker_until_complete(&db, job, port) + let result = run_job_in_new_worker_until_complete(&db, false, job, port) .await .json_result() .unwrap(); @@ -264,11 +268,13 @@ def main(): custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, + custom_debounce_key: None, + debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, }); - let result = run_job_in_new_worker_until_complete(&db, job, port) + let result = run_job_in_new_worker_until_complete(&db, false, job, port) .await .json_result() .unwrap(); @@ -303,11 +309,13 @@ def main(): custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, + custom_debounce_key: None, + debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, }); - let result = run_job_in_new_worker_until_complete(&db, job, port) + let result = run_job_in_new_worker_until_complete(&db, false, job, port) .await .json_result() .unwrap(); @@ -340,11 +348,13 @@ def main(): custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, + custom_debounce_key: None, + debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, }); - let result = run_job_in_new_worker_until_complete(&db, job, port) + let result = run_job_in_new_worker_until_complete(&db, false, job, port) .await .json_result() .unwrap(); diff --git a/backend/tests/relative_imports.rs b/backend/tests/relative_imports.rs index b088241f5a..78484d0879 100644 --- a/backend/tests/relative_imports.rs +++ b/backend/tests/relative_imports.rs @@ -520,7 +520,7 @@ def main(): } #[cfg(feature = "test_job_debouncing")] -mod job_debouncing { +mod dependency_job_debouncing { async fn trigger_djob_for( client: &windmill_api_client::Client, path: &str, @@ -566,7 +566,7 @@ def main(): /// p.s: "LF" stands for "Leaf", "L" - "Left", "R" - "Right" mod flows { use crate::common::{in_test_worker, init_client, listen_for_completed_jobs}; - use crate::job_debouncing::trigger_djob_for; + use crate::dependency_job_debouncing::trigger_djob_for; use std::time::Duration; use tokio::time::sleep; use tokio_stream::StreamExt; @@ -876,7 +876,7 @@ def main(): dedicated_worker: None, }) // .arg("dbg_djob_sleep", serde_json::json!(10)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; RunJob::from(windmill_common::jobs::JobPayload::Dependencies { @@ -887,7 +887,7 @@ def main(): }) // So set it to this long .arg("dbg_djob_sleep", serde_json::json!(10)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; completed.next().await; // leaf_right @@ -929,6 +929,20 @@ def main(): let (_client, port, _s) = init_client(db.clone()).await; let mut completed = listen_for_completed_jobs(&db).await; + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } + + { + let mut mvsd = windmill_common::worker::MIN_VERSION_IS_AT_LEAST_1_440 + .write() + .await; + *mvsd = true; + } + // Function to create a dependency job let create_dependency_job = |delay, @@ -1005,10 +1019,6 @@ def main(): in_test_worker( &db2, async { - windmill_common::worker::update_min_version( - &windmill_common::worker::Connection::Sql(db2.clone()), - ) - .await; // Small delay to ensure the job is marked as running tokio::time::sleep(tokio::time::Duration::from_millis(200)).await; @@ -1030,30 +1040,17 @@ def main(): .unwrap(), 1 ); + + dbg!(sqlx::query_scalar!("SELECT kind::text FROM v2_job") + .fetch_one(&db) + .await + .unwrap() + .unwrap()); } - // Block all tests using this variable until we are done - // let mut min_v = windmill_common::worker::MIN_VERSION_IS_AT_LEAST_1_440 - // .write() - // .await; - - // // Save initial min_v value; - // let initi_min_v = *min_v; - - // // Make it true for this test. - // *min_v = true; - - // Now push a second dependency job while the first is being processed - // This should trigger the race condition handling code let job2 = create_dependency_job(0, vec!["b", "c"], db.clone(), 1, Some(job1)).await; - // Set it back to initial - // *min_v = initi_min_v; - - // Unblock all other tests - // drop(min_v); - // Process the first job completion, and the second job should also get debounced by this one completed.next().await; @@ -1118,7 +1115,6 @@ def main(): async fn test_3(db: sqlx::Pool) -> anyhow::Result<()> { // This tests checks if concurrency limit works correcly and there is no race conditions. - use windmill_common::worker::Connection; let (_client, port, _s) = init_client(db.clone()).await; let mut completed = listen_for_completed_jobs(&db).await; @@ -1296,6 +1292,138 @@ WHERE Ok(()) } + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "djob_debouncing"))] + async fn test_min_version_supports_debouncing( + db: sqlx::Pool, + ) -> anyhow::Result<()> { + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } + + let (_client, port, _s) = init_client(db.clone()).await; + let db = &db; + + crate::common::in_test_worker( + db, + async { + let job_template = crate::common::RunJob::from( + windmill_common::jobs::JobPayload::FlowDependencies { + path: "f/dre/flow".to_owned(), + dedicated_worker: None, + version: 1443253234253454, + }, + ) + .push_arg_scheduled_for_o(Some(chrono::Utc::now())) + .arg("triggered_by_relative_import", serde_json::json!(())); + + // This will push to the top level worker + let debounce_job_id = job_template.clone().push(db).await; + + // Will have space to run in parallel but in it's own worker + job_template + .push_arg_debounce_job_id_o(Some(debounce_job_id)) + .run_until_complete(db, false, port) + .await; + }, + port, + ) + .await; + + // Verify there is not jobs running + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job_queue") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 0 + ); + + // And there is only supposed to be one job. + assert_eq!( + sqlx::query_scalar!( + "SELECT COUNT(*) FROM v2_job_completed WHERE status = 'success'" + ) + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 1 + ); + + Ok(()) + } + // NOTE: Don't run in parallel with other tests + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "djob_debouncing"))] + #[ignore] + async fn test_min_version_does_not_support_debouncing( + db: sqlx::Pool, + ) -> anyhow::Result<()> { + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = false; + } + + let (_client, port, _s) = init_client(db.clone()).await; + let db = &db; + + crate::common::in_test_worker( + db, + async { + let job_template = crate::common::RunJob::from( + windmill_common::jobs::JobPayload::FlowDependencies { + path: "f/dre/flow".to_owned(), + dedicated_worker: None, + version: 1443253234253454, + }, + ) + .push_arg_scheduled_for_o(Some(chrono::Utc::now())) + .arg("triggered_by_relative_import", serde_json::json!(())); + + // This will push to the top level worker + let debounce_job_id = job_template.clone().push(db).await; + + // Will have space to run in parallel but in it's own worker + job_template + .push_arg_debounce_job_id_o(Some(debounce_job_id)) + .run_until_complete(db, false, port) + .await; + }, + port, + ) + .await; + + // Verify there is not jobs running + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job_queue") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 0 + ); + + // There are supposed to be two jobs, since debouncing is disabled. + assert_eq!( + sqlx::query_scalar!( + "SELECT COUNT(*) FROM v2_job_completed WHERE status = 'success'" + ) + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 2 + ); + + Ok(()) + } // TODO: // test that update or create flow that should bypass debouncing } @@ -1304,7 +1432,7 @@ WHERE /// For apps we are going to do similar tests that we did for flows mod apps { use crate::common::{in_test_worker, init_client, listen_for_completed_jobs}; - use crate::job_debouncing::trigger_djob_for; + use crate::dependency_job_debouncing::trigger_djob_for; use std::time::Duration; use tokio::time::sleep; use tokio_stream::StreamExt; @@ -1575,7 +1703,7 @@ WHERE dedicated_worker: None, }) // .arg("dbg_djob_sleep", serde_json::json!(10)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; RunJob::from(windmill_common::jobs::JobPayload::Dependencies { @@ -1586,7 +1714,7 @@ WHERE }) // So set it to this long .arg("dbg_djob_sleep", serde_json::json!(10)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; completed.next().await; // leaf_right @@ -1791,7 +1919,7 @@ WHERE /// ## Testing for Scripts mod scripts { use crate::common::{in_test_worker, init_client, listen_for_completed_jobs}; - use crate::job_debouncing::trigger_djob_for; + use crate::dependency_job_debouncing::trigger_djob_for; use std::time::Duration; use tokio::time::sleep; use tokio_stream::StreamExt; @@ -2114,7 +2242,7 @@ WHERE language: windmill_common::scripts::ScriptLang::Python3, dedicated_worker: None, }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; // This one is supposed to be started after flow djob has debounced and started but haven't finished yet. @@ -2126,7 +2254,7 @@ WHERE }) // So set it to this long .arg("dbg_djob_sleep", serde_json::json!(10)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; completed.next().await; // leaf_right @@ -2362,3 +2490,735 @@ WHERE } // TODO: Test git sync } +#[cfg(feature = "test_job_debouncing")] +mod normal_job_debouncing { + mod scripts { + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "job_debouncing"))] + async fn test_default_debounce_key(db: sqlx::Pool) -> anyhow::Result<()> { + use serde_json::json; + use windmill_common::scripts::ScriptHash; + + use crate::common::{in_test_worker, init_client, listen_for_completed_jobs, RunJob}; + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } + + let (_client, port, _s) = init_client(db.clone()).await; + let db = &db; + + in_test_worker( + db, + async { + // This job should execute and then try to start another job that will get debounced. + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + // Do not supply with custom debounce key. + // We will test if the debounce_key is created correctly. + custom_debounce_key: None, + debounce_delay_s: Some(2), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }) + .arg("x", json!("ey")) + .arg("b", json!("33")) + // Start another worker, so we have two workers at the same time. + // We don't know which will execute the job, but we do know that if the job is executed, this worker will exit. + .run_until_complete_with(db, false, port, |id| async move { + + // Verify debounce_key + assert_eq!( + sqlx::query_scalar!( + "SELECT key FROM debounce_key WHERE job_id = $1", + id.clone() + ) + .fetch_one(db) + .await + .unwrap(), + "test-workspace/script/f/scripts/script_1#args:\"33\":\"ey\"".to_owned() + ); + + // Verify it is scheduled for future and not now. + { + assert!( + dbg!( + sqlx::query_scalar!( + "SELECT (scheduled_for - created_at) FROM v2_job_queue WHERE running = false" + ) + .fetch_one(db) + .await + .unwrap() + .unwrap() + .microseconds + ) > 1_000_000 /* 1 second */ + ); + } + + // Start another job. + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + // Do not supply with custom debounce key. + // We will test if the debounce_key is created correctly. + custom_debounce_key: None, + debounce_delay_s: Some(2), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }) + .arg("x", json!("ey")) + .arg("b", json!("33")) + // But we only push it, one of the jobs should be debounced. + .push(db) + .await; + }) + .await; + }, + port, + ) + .await; + + // Verify there is not jobs running + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job_queue") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 0 + ); + + // And there is only supposed to be one job. + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 1 + ); + + // Verify debounce key clean up + assert_eq!( + 0, + sqlx::query_scalar!("SELECT COUNT(*) from debounce_key") + .fetch_one(db) + .await + .unwrap() + .unwrap() + ); + + Ok(()) + } + + #[cfg(all(feature = "python", feature = "agent_worker_server"))] + #[sqlx::test(fixtures("base", "job_debouncing"))] + async fn test_default_debounce_key_agent_wk( + db: sqlx::Pool, + ) -> anyhow::Result<()> { + use serde_json::json; + use windmill_common::scripts::ScriptHash; + + use crate::common::{init_client_agent_mode, RunJob}; + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } + + let (_client, port, _s) = init_client_agent_mode(db.clone()).await; + let db = &db; + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + // Do not supply with custom debounce key. + // We will test if the debounce_key is created correctly. + custom_debounce_key: None, + debounce_delay_s: Some(2), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }) + .arg("x", json!("ey")) + .arg("b", json!("33")) + .run_until_complete(db, true, port) + .await; + + // Verify there is not jobs running + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job_queue") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 0 + ); + + // And there is only supposed to be one job. + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 1 + ); + + // And that job execute successfully + assert_eq!( + sqlx::query_scalar!("SELECT status::text FROM v2_job_completed") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + "success" + ); + + // Verify debounce key clean up + assert_eq!( + 0, + sqlx::query_scalar!("SELECT COUNT(*) from debounce_key") + .fetch_one(db) + .await + .unwrap() + .unwrap() + ); + + Ok(()) + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "job_debouncing"))] + async fn test_custom_debounce_key(db: sqlx::Pool) -> anyhow::Result<()> { + use serde_json::json; + use windmill_common::scripts::ScriptHash; + + use crate::common::{in_test_worker, init_client, listen_for_completed_jobs, RunJob}; + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } + + let (_client, port, _s) = init_client(db.clone()).await; + let completed = listen_for_completed_jobs(&db).await; + let db = &db; + + in_test_worker( + db, + async { + // This job should execute and then try to start another job that will get debounced. + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + // Do not supply with custom debounce key. + // We will test if the debounce_key is created correctly. + custom_debounce_key: Some("$workspace:my-custom-debounce-key:$args[x]".to_owned()), + debounce_delay_s: Some(2), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }) + .arg("x", json!("ey")) + .arg("b", json!("1")) // 1 + // Start another worker, so we have two workers at the same time. + // We don't know which will execute the job, but we do know that if the job is executed, this worker will exit. + .run_until_complete_with(db, false, port, |id| async move { + + // Verify debounce_key + assert_eq!( + sqlx::query_scalar!( + "SELECT key FROM debounce_key WHERE job_id = $1", + id.clone() + ) + .fetch_one(db) + .await + .unwrap(), + "test-workspace:my-custom-debounce-key:ey".to_owned() + ); + + // Verify it is scheduled for future and not now. + { + assert!( + dbg!( + sqlx::query_scalar!( + "SELECT (scheduled_for - created_at) FROM v2_job_queue WHERE running = false" + ) + .fetch_one(db) + .await + .unwrap() + .unwrap() + .microseconds + ) > 1_000_000 /* 1 second */ + ); + } + + // Start another job. + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + custom_debounce_key: Some("$workspace:my-custom-debounce-key:$args[x]".to_owned()), + debounce_delay_s: Some(2), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }) + .arg("x", json!("ey")) + // We will pass different argument. but it should still get debounced. + .arg("b", json!("2")) // 2 + // But we only push it, one of the jobs should be debounced. + .push(db) + .await; + }) + .await; + }, + port, + ) + .await; + + // Verify there is not jobs running + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job_queue") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 0 + ); + + // And there is only supposed to be one job. + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 1 + ); + + // And that job execute successfully + assert_eq!( + sqlx::query_scalar!("SELECT status::text FROM v2_job_completed") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + "success" + ); + + // Verify debounce key clean up + assert_eq!( + 0, + sqlx::query_scalar!("SELECT COUNT(*) from debounce_key") + .fetch_one(db) + .await + .unwrap() + .unwrap() + ); + + Ok(()) + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "job_debouncing"))] + async fn test_no_debounce(db: sqlx::Pool) -> anyhow::Result<()> { + use serde_json::json; + use windmill_common::scripts::ScriptHash; + + use crate::common::{in_test_worker, init_client, listen_for_completed_jobs, RunJob}; + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } + + let (_client, port, _s) = init_client(db.clone()).await; + let completed = listen_for_completed_jobs(&db).await; + let db = &db; + + // different args + in_test_worker( + db, + async { + // This job should execute and then try to start another job that will get debounced. + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + // Do not supply with custom debounce key. + // We will test if the debounce_key is created correctly. + custom_debounce_key: None, + debounce_delay_s: Some(2), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }) + .arg("x", json!("ey")) + .arg("b", json!("33")) + // Start another worker, so we have two workers at the same time. + // We don't know which will execute the job, but we do know that if the job is executed, this worker will exit. + .run_until_complete_with(db, false, port, |_id| async move { + // Start another job. + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + // Do not supply with custom debounce key. + // We will test if the debounce_key is created correctly. + custom_debounce_key: None, + debounce_delay_s: Some(2), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }) + // Different args. + .arg("x", json!("ey")) + .arg("b", json!("34")) // Different arg + .push(db) + .await; + }) + .await; + }, + port, + ) + .await; + + // no debounce delay on second + in_test_worker( + db, + async { + // This job should execute and then try to start another job that will get debounced. + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + // Do not supply with custom debounce key. + // We will test if the debounce_key is created correctly. + custom_debounce_key: None, + debounce_delay_s: Some(2), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }) + .arg("x", json!("ey")) + .arg("b", json!("33")) + // Start another worker, so we have two workers at the same time. + // We don't know which will execute the job, but we do know that if the job is executed, this worker will exit. + .run_until_complete_with(db, false, port, |_id| async move { + // Start another job. + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + custom_debounce_key: None, + debounce_delay_s: None, // Set to none to skip debouncing + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }) + .arg("x", json!("ey")) + .arg("b", json!("33")) + .push(db) + .await; + }) + .await; + }, + port, + ) + .await; + + // Verify there is not jobs running + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job_queue") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 0 + ); + + // And there is supposed to be four jobs and no debouncing. + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 4 + ); + + // Verify debounce key clean up + assert_eq!( + 0, + sqlx::query_scalar!("SELECT COUNT(*) from debounce_key") + .fetch_one(db) + .await + .unwrap() + .unwrap() + ); + + Ok(()) + } + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "djob_debouncing"))] + #[ignore = "modifies global env variable that is used by other tests"] + async fn test_min_version_does_not_support_debouncing( + db: sqlx::Pool, + ) -> anyhow::Result<()> { + use serde_json::json; + use windmill_common::scripts::ScriptHash; + + use crate::common::{in_test_worker, init_client, listen_for_completed_jobs, RunJob}; + + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = false; + } + + let (_client, port, _s) = init_client(db.clone()).await; + let db = &db; + + crate::common::in_test_worker( + db, + async { + let job_template = + RunJob::from(windmill_common::jobs::JobPayload::ScriptHash { + hash: ScriptHash(533400), + path: "f/scripts/script_1".into(), + custom_debounce_key: None, + debounce_delay_s: None, // Set to none to skip debouncing + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + language: windmill_common::scripts::ScriptLang::Python3, + priority: None, + apply_preprocessor: false, + }); + + // This will push to the top level worker + job_template.clone().push(db).await; + + // Will have space to run in parallel but in it's own worker + job_template.run_until_complete(db, false, port).await; + }, + port, + ) + .await; + + // Verify there is not jobs running + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job_queue") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 0 + ); + + // There are supposed to be two jobs, since debouncing is disabled. + assert_eq!( + sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job") + .fetch_one(db) + .await + .unwrap() + .unwrap(), + 2 + ); + + Ok(()) + } + } + + mod flows { + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "job_debouncing"))] + async fn test_different_kinds_top_level( + db: sqlx::Pool, + ) -> anyhow::Result<()> { + use crate::common::{init_client, listen_for_completed_jobs, RunJob}; + use serde_json::json; + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } + + let (_client, port, _s) = init_client(db.clone()).await; + let db = &db; + + // We want to run this for all tables related to flow be created. + RunJob::from(windmill_common::jobs::JobPayload::FlowDependencies { + path: "f/flows/flow".into(), + dedicated_worker: None, + version: 1443253234253454, + }) + .run_until_complete(db, false, port) + .await; + + dbg!(sqlx::query!("SELECT * FROM flow_node",) + .fetch_all(db) + .await + .unwrap()); + + let (j1, j2, j3) = tokio::join!( + RunJob::from(windmill_common::jobs::JobPayload::Flow { + version: 1443253234253454, + path: "f/flows/flow".into(), + dedicated_worker: None, + apply_preprocessor: false, + }) + .push(db), + RunJob::from(windmill_common::jobs::JobPayload::SingleStepFlow { + hash: None, + path: "f/flows/flow".into(), + custom_debounce_key: None, + debounce_delay_s: Some(2), + custom_concurrency_key: None, + concurrent_limit: None, + concurrency_time_window_s: None, + flow_version: Some(1443253234253454), + args: std::collections::HashMap::new(), + retry: None, + error_handler_path: None, + error_handler_args: None, + skip_handler: None, + cache_ttl: None, + priority: None, + tag_override: None, + trigger_path: None, + apply_preprocessor: false, + }) + .push(db), + RunJob::from(windmill_common::jobs::JobPayload::RawFlow { + value: windmill_common::flows::FlowValue { + debounce_delay_s: Some(2), + modules: vec![windmill_common::flows::FlowModule { + id: "a".into(), + value: windmill_common::worker::to_raw_value(&json!({ + "lock": "# py: 3.11\n", + "type": "rawscript", + "assets": [], + "content": "def main(x: str, y: str):\n return x", + "language": "python3", + "debounce_delay_s": 15, + "input_transforms": { + "x": { + "type": "static", + "value": "" + }, + "y": { + "type": "static", + "value": "" + } + } + })), + ..Default::default() + }], + ..Default::default() + }, + path: Some("f/flows/flow".into()), + restarted_from: None, + }) + .push(db), + // RunJob::from(windmill_common::jobs::JobPayload::Code( + // windmill_common::jobs::RawCode { + // content: " + // def main(n: int): + // pass + // " + // .into(), + // path: Some("f/flows/flow".into()), + // hash: None, + // language: windmill_common::scripts::ScriptLang::Python3, + // custom_debounce_key: None, + // debounce_delay_s: Some(2), + // ..Default::default() + // }, + // )) + // .push(db) + ); + + assert_eq!(j1, j2); + assert_eq!(j1, j3); + // assert_eq!(j1, j4); + Ok(()) + } + } + + // TODO(ALL): + // - Check if all jobs were sucessfull. + // + // TODO: + // - [x] FlowNode (Script) + // - [x] FlowNode (Flow) - has no debouncing nor concurrency limits + // - [x] RawCode (Flow as code) + // - [x] RawFlow + // - [x] Flow + // - [x] FlowScript + // + // TODO(imperatively): + // - [x] Creation of flow + // - [x] Check entire flow + // - [x] Check it's inline scripts + // - [x] Creation of script + // + // TODO: [x] Agent workers. (and tests) + // TODO: [x] Backwards compat (and tests) + // TODO: [x] Concurrency limit is disabled if preprocessor is enabled. Investigate. + // TODO: [x] Last resort - monitor.rs to clean up debounce_keys + // TODO: [x] Catch debounce values by server if debouncing is disabled. (and tests) +} diff --git a/backend/tests/retry.rs b/backend/tests/retry.rs index f7162e9a45..60b81573db 100644 --- a/backend/tests/retry.rs +++ b/backend/tests/retry.rs @@ -2,13 +2,13 @@ mod common; #[cfg(feature = "deno_core")] mod retry { + use crate::common::*; use serde_json::json; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; use sqlx::{Pool, Postgres}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use windmill_common::flow_status::FlowStatusModule; use windmill_common::flows::FlowValue; use windmill_common::jobs::JobPayload; - use windmill_common::flow_status::FlowStatusModule; - use crate::common::*; pub async fn initialize_tracing() { use std::sync::Once; @@ -166,7 +166,7 @@ def main(last, port): }) .arg("items", json!(["unused", "unused", "unused"])) .arg("port", json!(server.addr.port())) - .run_until_complete(&db, server.addr.port()) + .run_until_complete(&db, false, server.addr.port()) .await .json_result() .unwrap(); @@ -201,7 +201,7 @@ def main(last, port): }) .arg("items", json!(["unused", "unused", "unused"])) .arg("port", json!(server.addr.port())) - .run_until_complete(&db, server.addr.port()) + .run_until_complete(&db, false, server.addr.port()) .await .json_result() .unwrap(); @@ -248,7 +248,7 @@ def main(last, port): }) .arg("items", json!(["unused", "unused", "unused"])) .arg("port", json!(server.addr.port())) - .run_until_complete(&db, server.addr.port()) + .run_until_complete(&db, false, server.addr.port()) .await; let result = job.json_result().unwrap(); @@ -315,7 +315,7 @@ def main(error, port): let server = Server::start(responses).await; let cjob = RunJob::from(JobPayload::RawFlow { value, path: None, restarted_from: None }) .arg("port", json!(server.addr.port())) - .run_until_complete(&db, server.addr.port()) + .run_until_complete(&db, false, server.addr.port()) .await; let result = cjob.json_result().clone().unwrap(); let failed_module = get_module(&cjob, "a").unwrap(); diff --git a/backend/tests/suspend_resume.rs b/backend/tests/suspend_resume.rs index e51aff9060..9c50488060 100644 --- a/backend/tests/suspend_resume.rs +++ b/backend/tests/suspend_resume.rs @@ -6,13 +6,13 @@ mod suspend_resume { #[cfg(feature = "deno_core")] use crate::common::*; - + #[cfg(feature = "deno_core")] - use sqlx::{Pool, Postgres}; + use futures::{Stream, StreamExt}; #[cfg(feature = "deno_core")] use sqlx::types::Uuid; #[cfg(feature = "deno_core")] - use futures::{Stream, StreamExt}; + use sqlx::{Pool, Postgres}; #[cfg(feature = "deno_core")] use windmill_common::flows::FlowValue; #[cfg(feature = "deno_core")] @@ -223,7 +223,7 @@ mod suspend_resume { .arg("n", json!(1)) .arg("op", json!("cancel")) .arg("port", json!(port)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 72d753c48c..c2516e1877 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -77,7 +77,7 @@ async fn test_iteration(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("items", json!([])) - .run_until_complete(&db, server.addr.port()) + .run_until_complete(&db, false, server.addr.port()) .await .json_result() .unwrap(); @@ -87,7 +87,7 @@ async fn test_iteration(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("items", json!((0..257).collect::>())) - .run_until_complete(&db, server.addr.port()) + .run_until_complete(&db, false, server.addr.port()) .await .json_result() .unwrap(); @@ -138,7 +138,7 @@ async fn test_iteration_parallel(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("items", json!([])) - .run_until_complete(&db, server.addr.port()) + .run_until_complete(&db, false, server.addr.port()) .await .json_result() .unwrap(); @@ -148,7 +148,7 @@ async fn test_iteration_parallel(db: Pool) -> anyhow::Result<()> { let job = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("items", json!((0..50).collect::>())) - .run_until_complete(&db, server.addr.port()) + .run_until_complete(&db, false, server.addr.port()) .await; // println!("{:#?}", job); let result = job.json_result().unwrap(); @@ -286,7 +286,7 @@ async fn test_deno_flow(db: Pool) -> anyhow::Result<()> { for i in 0..50 { println!("deno flow iteration: {}", i); - let job = run_job_in_new_worker_until_complete(&db, job.clone(), port).await; + let job = run_job_in_new_worker_until_complete(&db, false, job.clone(), port).await; // println!("job: {:#?}", job.flow_status); let result = job.json_result().unwrap(); assert_eq!(result, serde_json::json!([2, 4, 6]), "iteration: {}", i); @@ -325,7 +325,7 @@ async fn test_identity(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .run_until_complete(&db, server.addr.port()) + .run_until_complete(&db, false, server.addr.port()) .await .json_result() .unwrap(); @@ -561,7 +561,7 @@ async fn test_deno_flow_same_worker(db: Pool) -> anyhow::Result<()> { let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, job.clone(), server.addr.port()) + let result = run_job_in_new_worker_until_complete(&db, false, job.clone(), server.addr.port()) .await .json_result() .unwrap(); @@ -616,7 +616,7 @@ async fn test_flow_result_by_id(db: Pool) -> anyhow::Result<()> { .unwrap(); let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, job.clone(), port) + let result = run_job_in_new_worker_until_complete(&db, false, job.clone(), port) .await .json_result() .unwrap(); @@ -663,7 +663,7 @@ async fn test_stop_after_if(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(job.clone()) .arg("n", json!(123)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -671,7 +671,7 @@ async fn test_stop_after_if(db: Pool) -> anyhow::Result<()> { let cjob = RunJob::from(job.clone()) .arg("n", json!(-123)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; let result = cjob.json_result().unwrap(); @@ -724,7 +724,7 @@ async fn test_stop_after_if_nested(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(job.clone()) .arg("n", json!(123)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -732,7 +732,7 @@ async fn test_stop_after_if_nested(db: Pool) -> anyhow::Result<()> { let cjob = RunJob::from(job.clone()) .arg("n", json!(-123)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; let result = cjob.json_result().unwrap(); @@ -787,6 +787,7 @@ async fn test_python_flow(db: Pool) -> anyhow::Result<()> { println!("python flow iteration: {}", i); let result = run_job_in_new_worker_until_complete( &db, + false, JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }, port, ) @@ -824,6 +825,7 @@ async fn test_python_flow_2(db: Pool) -> anyhow::Result<()> { println!("python flow iteration: {}", i); let result = run_job_in_new_worker_until_complete( &db, + false, JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }, port, ) @@ -865,9 +867,11 @@ func main(derp string) (string, error) { concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .arg("derp", json!("world")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -902,9 +906,11 @@ fn main(world: String) -> Result { concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .arg("world", json!("Hyrule")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -947,7 +953,7 @@ fn main(world: String) -> Result { // })) // .arg("world", json!("Arakis")) // .arg("b", json!(3)) -// .run_until_complete(&db, port) +// .run_until_complete(&db, false, port) // .await // .json_result() // .unwrap(); @@ -978,9 +984,11 @@ echo "hello $msg" concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .arg("msg", json!("world")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; assert_eq!(job.json_result(), Some(json!("hello world"))); Ok(()) @@ -1011,9 +1019,11 @@ def main [ msg: string ] { concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .arg("msg", json!("world")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; assert_eq!(job.json_result(), Some(json!("hello world"))); Ok(()) @@ -1064,6 +1074,8 @@ def main [ concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .arg("a", json!("3")) .arg("b", json!("null")) @@ -1082,7 +1094,7 @@ def main [ ]), ) .arg("n", json!("baz")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -1126,12 +1138,14 @@ public class Main { concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .arg("a", json!(3)) .arg("b", json!(3.0)) .arg("age", json!(30)) .arg("d", json!(3.0)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; assert_eq!(job.json_result(), Some(json!("hello world"))); Ok(()) @@ -1161,9 +1175,11 @@ export async function main(a: Date) { concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .arg("a", json!("2024-09-24T10:00:00.000Z")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -1196,9 +1212,11 @@ export async function main(a: Date) { concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .arg("a", json!("2024-09-24T10:00:00.000Z")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -1232,10 +1250,12 @@ def main(a: datetime, b: bytes): concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, })) .arg("a", json!("2024-09-24T10:00:00.000Z")) .arg("b", json!("dGVzdA==")) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -1293,7 +1313,7 @@ async fn test_empty_loop_1(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1334,7 +1354,7 @@ async fn test_invalid_first_step(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let job = run_job_in_new_worker_until_complete(&db, flow, port).await; + let job = run_job_in_new_worker_until_complete(&db, false, flow, port).await; assert!( serde_json::to_string(&job.json_result().unwrap()).unwrap().contains("Expected an array value in the iterator expression, found: invalid type: map, expected a sequence at line 1 column 0") @@ -1376,7 +1396,7 @@ async fn test_empty_loop_2(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1433,7 +1453,7 @@ async fn test_step_after_loop(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1502,7 +1522,7 @@ async fn test_branchone_simple(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1540,7 +1560,7 @@ async fn test_branchone_with_cond(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1580,7 +1600,7 @@ async fn test_branchall_sequential(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1619,7 +1639,7 @@ async fn test_branchall_simple(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1667,7 +1687,7 @@ async fn test_branchall_skip_failure(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1704,7 +1724,7 @@ async fn test_branchall_skip_failure(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1770,7 +1790,7 @@ async fn test_branchone_nested(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1828,7 +1848,7 @@ async fn test_branchall_nested(db: Pool) -> anyhow::Result<()> { .unwrap(); let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; - let result = run_job_in_new_worker_until_complete(&db, flow, port) + let result = run_job_in_new_worker_until_complete(&db, false, flow, port) .await .json_result() .unwrap(); @@ -1896,7 +1916,7 @@ async fn test_failure_module(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("n", json!(0)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -1913,7 +1933,7 @@ async fn test_failure_module(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("n", json!(1)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -1930,7 +1950,7 @@ async fn test_failure_module(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("n", json!(2)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -1947,7 +1967,7 @@ async fn test_failure_module(db: Pool) -> anyhow::Result<()> { let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("n", json!(3)) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await .json_result() .unwrap(); @@ -2267,7 +2287,7 @@ async fn test_complex_flow_restart(db: Pool) -> anyhow::Result<()> { let first_run_result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; let restarted_flow_result = RunJob::from(JobPayload::RawFlow { @@ -2279,7 +2299,7 @@ async fn test_complex_flow_restart(db: Pool) -> anyhow::Result<()> { branch_or_iteration_n: None, }), }) - .run_until_complete(&db, port) + .run_until_complete(&db, false, port) .await; let first_run_result_int = @@ -2802,7 +2822,7 @@ async fn test_job_labels(db: Pool) -> anyhow::Result<()> { restarted_from: None, }) .arg("world", json!("you")) - .run_until_complete_with(db, port, |id| async move { + .run_until_complete_with(db, false, port, |id| async move { sqlx::query!( "UPDATE v2_job SET labels = $2 WHERE id = $1 AND $2::TEXT[] IS NOT NULL", id, @@ -2876,7 +2896,7 @@ async fn test_workflow_as_code(db: Pool) -> anyhow::Result<()> { ..RawCode::default() })) .arg("n", json!(3)) - .run_until_complete(db, port) + .run_until_complete(db, false, port) .await; assert_eq!(job.json_result().unwrap(), json!(["OK", 3])); diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index ad084398fb..e4c168d838 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -15414,6 +15414,10 @@ components: type: string concurrency_key: type: string + debounce_key: + type: string + debounce_delay_s: + type: integer visible_to_runner_only: type: boolean no_main_func: diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 4d1e7aecc1..a4d27f43f9 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -32,8 +32,9 @@ use sql_builder::prelude::*; use sqlx::{FromRow, Postgres, Transaction}; use windmill_audit::audit_oss::audit_log; use windmill_audit::ActionKind; +use windmill_common::flows::FlowValue; use windmill_common::utils::{query_elems_from_hub, WarnAfterExt}; -use windmill_common::worker::{to_raw_value, CLOUD_HOSTED}; +use windmill_common::worker::{to_raw_value, CLOUD_HOSTED, MIN_VERSION_SUPPORTS_DEBOUNCING}; use windmill_common::HUB_BASE_URL; use windmill_common::{ db::UserDB, @@ -385,6 +386,7 @@ async fn create_flow( Json(nf): Json, ) -> Result<(StatusCode, String)> { check_scopes(&authed, || format!("flows:write:{}", nf.path))?; + guard_flow_from_debounce_data(&nf).await?; if *CLOUD_HOSTED { let nb_flows = sqlx::query_scalar!("SELECT COUNT(*) FROM flow WHERE workspace_id = $1", &w_id) @@ -745,6 +747,7 @@ async fn update_flow( ) -> Result { let flow_path = flow_path.to_path(); check_scopes(&authed, || format!("flows:write:{}", flow_path))?; + guard_flow_from_debounce_data(&nf).await?; #[cfg(not(feature = "enterprise"))] if nf @@ -1359,6 +1362,22 @@ async fn archive_flow_by_path( Ok(format!("Flow {path} archived")) } +/// 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: FlowValue = serde_json::from_value(nf.value.clone())?; + flow_value.debounce_key.is_some() || flow_value.debounce_delay_s.is_some() + } { + tracing::warn!( + "Flow debouncing configuration rejected: workers are behind minimum required version for debouncing feature" + ); + Err(Error::WorkersAreBehind { feature: "Debouncing".into(), min_version: "1.566.0".into() }) + } else { + Ok(()) + } +} + #[derive(Deserialize)] struct DeleteFlowQuery { keep_captures: Option, @@ -1615,6 +1634,8 @@ mod tests { early_return: None, concurrency_key: None, chat_input_enabled: None, + debounce_key: None, + debounce_delay_s: None, }; let expect = serde_json::json!({ "modules": [ diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 8359c4d70e..cb6fa50b94 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -4388,6 +4388,9 @@ pub async fn run_workflow_as_code( 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, }), Some(job.tag.clone()), None, @@ -5218,6 +5221,8 @@ pub async fn run_wait_result_script_by_hash( concurrency_key, concurrent_limit, concurrency_time_window_s, + debounce_key, + debounce_delay_s, mut cache_ttl, language, dedicated_worker, @@ -5264,6 +5269,8 @@ pub async fn run_wait_result_script_by_hash( 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, cache_ttl, language, dedicated_worker, @@ -5672,6 +5679,8 @@ async fn run_preview_script( 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, cache_ttl: None, dedicated_worker: preview.dedicated_worker, }), @@ -5793,6 +5802,8 @@ async fn run_bundle_preview_script( cache_ttl: None, dedicated_worker: preview.dedicated_worker, custom_concurrency_key: None, + custom_debounce_key: None, + debounce_delay_s: None, }), PushArgs::from(&args), authed.display_username(), @@ -6521,6 +6532,8 @@ async fn run_dynamic_select( concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: None, + custom_debounce_key: None, + debounce_delay_s: None, }), PushArgs::from(&request.args.unwrap_or_default()), authed.display_username(), @@ -6596,6 +6609,8 @@ pub async fn run_job_by_hash_inner( concurrency_key, concurrent_limit, concurrency_time_window_s, + debounce_delay_s, + debounce_key, mut cache_ttl, language, dedicated_worker, @@ -6644,6 +6659,8 @@ pub async fn run_job_by_hash_inner( 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, cache_ttl, language, dedicated_worker, diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 394e5e62d2..0d915e7937 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -46,7 +46,7 @@ use windmill_common::{ s3_helpers::upload_artifact_to_store, scripts::hash_script, utils::WarnAfterExt, - worker::CLOUD_HOSTED, + worker::{CLOUD_HOSTED, MIN_VERSION_SUPPORTS_DEBOUNCING}, }; use windmill_common::{ @@ -388,6 +388,7 @@ async fn create_snapshot_script( Path(w_id): Path, mut multipart: Multipart, ) -> Result<(StatusCode, String)> { + // TODO: Check for debouncing here as well. let mut script_hash = None; let mut tx = None; let mut uploaded = false; @@ -520,6 +521,8 @@ async fn create_script_internal<'c>( )> { check_scopes(&authed, || format!("scripts:write:{}", ns.path))?; + guard_script_from_debounce_data(&ns).await?; + let codebase = ns.codebase.as_ref(); #[cfg(not(feature = "enterprise"))] if ns.ws_error_handler_muted.is_some_and(|val| val) { @@ -782,8 +785,8 @@ async fn create_script_internal<'c>( content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \ draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \ dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \ - delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34)", + delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36)", &w_id, &hash.0, ns.path, @@ -821,7 +824,9 @@ async fn create_script_internal<'c>( None }, validate_schema, - ns.assets.as_ref().and_then(|a| serde_json::to_value(a).ok()) + ns.assets.as_ref().and_then(|a| serde_json::to_value(a).ok()), + ns.debounce_key, + ns.debounce_delay_s, ) .execute(&mut *tx) .await?; @@ -2131,3 +2136,18 @@ async fn delete_scripts_bulk( Ok(Json(deleted_paths)) } + +/// Validates that script debouncing configuration is supported by all workers +/// Returns an error if debouncing is configured but workers are behind required version +async fn guard_script_from_debounce_data(ns: &NewScript) -> Result<()> { + if !*MIN_VERSION_SUPPORTS_DEBOUNCING.read().await + && (ns.debounce_key.is_some() || ns.debounce_delay_s.is_some()) + { + tracing::warn!( + "Script debouncing configuration rejected: workers are behind minimum required version for debouncing feature" + ); + Err(Error::WorkersAreBehind { feature: "Debouncing".into(), min_version: "1.566.0".into() }) + } else { + Ok(()) + } +} diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index a012b1071b..f813f6ab1b 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -804,6 +804,8 @@ async fn trigger_script_with_retry_and_error_handler( custom_concurrency_key, concurrent_limit, concurrency_time_window_s, + custom_debounce_key, + debounce_delay_s, cache_ttl, priority, apply_preprocessor, @@ -825,6 +827,8 @@ async fn trigger_script_with_retry_and_error_handler( tag_override: tag.clone(), apply_preprocessor, trigger_path: Some(trigger_path), + custom_debounce_key, + debounce_delay_s, }, _ => { return Err(windmill_common::error::Error::internal_err(format!( diff --git a/backend/windmill-api/src/workspaces_export.rs b/backend/windmill-api/src/workspaces_export.rs index 92c4f6ef50..74732c9f22 100644 --- a/backend/windmill-api/src/workspaces_export.rs +++ b/backend/windmill-api/src/workspaces_export.rs @@ -102,6 +102,10 @@ struct ScriptMetadata { pub has_preprocessor: Option, #[serde(skip_serializing_if = "Option::is_none")] pub on_behalf_of_email: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub debounce_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub debounce_delay_s: Option, } pub fn is_none_or_false(val: &Option) -> bool { @@ -426,6 +430,8 @@ pub(crate) async fn tarball_workspace( concurrency_key: script.concurrency_key, has_preprocessor: script.has_preprocessor, on_behalf_of_email: script.on_behalf_of_email, + debounce_key: script.debounce_key, + debounce_delay_s: script.debounce_delay_s, }; let metadata_str = serde_json::to_string_pretty(&metadata).unwrap(); archive @@ -509,10 +515,7 @@ pub(crate) async fn tarball_workspace( && var.is_secret { var.value = Some(decrypt(&mc, var.value.unwrap()).map_err(|e| { - Error::internal_err(format!( - "Error decrypting variable {}: {}", - var.path, e - )) + Error::internal_err(format!("Error decrypting variable {}: {}", var.path, e)) })?); } let var_str = &to_string_without_metadata(&var, false, None).unwrap(); diff --git a/backend/windmill-common/src/agent_workers.rs b/backend/windmill-common/src/agent_workers.rs index 4039554830..322ca1c4fb 100644 --- a/backend/windmill-common/src/agent_workers.rs +++ b/backend/windmill-common/src/agent_workers.rs @@ -35,7 +35,11 @@ pub struct AgentAuth { pub const AGENT_JWT_PREFIX: &str = "jwt_agent_"; -pub fn build_agent_http_client(worker_suffix: &str) -> HttpClient { +pub fn build_agent_http_client( + worker_suffix: &str, + agent_token: Option, + base_internal_url: Option, +) -> HttpClient { let client = ClientBuilder::new( configure_client(reqwest::Client::builder() .pool_max_idle_per_host(10) @@ -52,7 +56,9 @@ pub fn build_agent_http_client(worker_suffix: &str) -> HttpClient { "{}{}_{}", AGENT_JWT_PREFIX, worker_suffix, - AGENT_TOKEN.trim_start_matches(AGENT_JWT_PREFIX), + agent_token + .unwrap_or(AGENT_TOKEN.clone()) + .trim_start_matches(AGENT_JWT_PREFIX) ); headers.insert( "Authorization", @@ -67,7 +73,8 @@ pub fn build_agent_http_client(worker_suffix: &str) -> HttpClient { ExponentialBackoff::builder().build_with_max_retries(5), )) .build(); - HttpClient(client) + + HttpClient { client, base_internal_url } } #[derive(Deserialize, Serialize)] diff --git a/backend/windmill-common/src/error.rs b/backend/windmill-common/src/error.rs index d9644fbba9..8048b59586 100644 --- a/backend/windmill-common/src/error.rs +++ b/backend/windmill-common/src/error.rs @@ -84,6 +84,8 @@ pub enum Error { ArgumentErr(String), #[error("{1}")] Generic(StatusCode, String), + #[error("{feature} is unavailable due to some workers being behind. Do not use the feature or make sure all workers run at least {min_version}")] + WorkersAreBehind { feature: String, min_version: String }, } impl Error { diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 9d554db7e1..5ad543c98b 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -122,10 +122,17 @@ pub struct FlowValue { #[serde(skip_serializing_if = "is_default")] pub same_worker: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_key: Option, #[serde(skip_serializing_if = "Option::is_none")] pub concurrent_limit: Option, #[serde(skip_serializing_if = "Option::is_none")] pub concurrency_time_window_s: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub debounce_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub debounce_delay_s: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub skip_expr: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -136,8 +143,6 @@ pub struct FlowValue { // Priority at the flow level pub priority: Option, #[serde(skip_serializing_if = "Option::is_none")] - pub concurrency_key: Option, - #[serde(skip_serializing_if = "Option::is_none")] pub chat_input_enabled: Option, } diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 20107cc5ee..49af82bb64 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -323,17 +323,28 @@ impl CompletedJob { #[derive(Debug, Clone)] pub enum JobPayload { + /// Execute Hub Script ScriptHub { path: String, apply_preprocessor: bool, }, + /// Execute script ScriptHash { hash: ScriptHash, path: String, + /// Override default concurrency key custom_concurrency_key: Option, + /// How many jobs can run at the same time concurrent_limit: Option, + /// In seconds concurrency_time_window_s: Option, + /// If not set, will be inferred from the hash(path + step_id + inputs) + custom_debounce_key: Option, + /// 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, cache_ttl: Option, dedicated_worker: Option, language: ScriptLang, @@ -341,25 +352,25 @@ pub enum JobPayload { apply_preprocessor: bool, }, + /// Execute flow step (can be subflow only). + FlowNode { + id: FlowNodeId, // flow_node(id). + path: String, // flow node inner path (e.g. `outer/branchall-42`). + }, + + /// Execute flow step FlowScript { id: FlowNodeId, // flow_node(id). language: ScriptLang, - /// Override default concurrency key custom_concurrency_key: Option, - /// How many jobs can run at the same time concurrent_limit: Option, - /// In seconds concurrency_time_window_s: Option, cache_ttl: Option, dedicated_worker: Option, path: String, }, - FlowNode { - id: FlowNodeId, // flow_node(id). - path: String, // flow node inner path (e.g. `outer/branchall-42`). - }, - + /// Inline App Script AppScript { id: AppScriptId, // app_script(id). path: Option, @@ -367,6 +378,7 @@ pub enum JobPayload { cache_ttl: Option, }, + /// Script/App/FlowAsCode Preview Code(RawCode), /// Script Dependency Job @@ -411,11 +423,14 @@ pub enum JobPayload { apply_preprocessor: bool, version: i64, }, + RestartedFlow { completed_job_id: Uuid, step_id: String, branch_or_iteration_n: Option, }, + + /// Flow Preview RawFlow { value: FlowValue, path: Option, @@ -435,6 +450,8 @@ pub enum JobPayload { custom_concurrency_key: Option, concurrent_limit: Option, concurrency_time_window_s: Option, + custom_debounce_key: Option, + debounce_delay_s: Option, cache_ttl: Option, priority: Option, tag_override: Option, @@ -469,6 +486,8 @@ pub struct RawCode { pub custom_concurrency_key: Option, pub concurrent_limit: Option, pub concurrency_time_window_s: Option, + pub custom_debounce_key: Option, + pub debounce_delay_s: Option, pub cache_ttl: Option, pub dedicated_worker: Option, } @@ -540,6 +559,8 @@ pub async fn script_path_to_payload<'e>( concurrency_key, concurrent_limit, concurrency_time_window_s, + debounce_key, + debounce_delay_s, cache_ttl, language, dedicated_worker, @@ -568,6 +589,8 @@ pub async fn script_path_to_payload<'e>( custom_concurrency_key: concurrency_key, concurrent_limit, concurrency_time_window_s, + custom_debounce_key: debounce_key, + debounce_delay_s, cache_ttl, language, dedicated_worker, @@ -804,6 +827,11 @@ pub async fn lock_debounce_key<'c>( runnable_path: &str, tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, ) -> error::Result> { + if !*crate::worker::MIN_VERSION_SUPPORTS_DEBOUNCING.read().await { + tracing::warn!("Debouncing is not supported on this version of Windmill. Minimum version required for debouncing support."); + return Ok(None); + } + let key = format!("{w_id}:{runnable_path}:dependency"); tracing::debug!( diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 1f796532b4..53675c6ef4 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -499,7 +499,7 @@ pub struct ExpiringLatestVersionId { expires_at: std::time::Instant, } -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct ScriptHashInfo { pub path: String, pub hash: i64, @@ -507,6 +507,8 @@ pub struct ScriptHashInfo { pub concurrency_key: Option, pub concurrent_limit: Option, pub concurrency_time_window_s: Option, + pub debounce_key: Option, + pub debounce_delay_s: Option, pub cache_ttl: Option, pub language: ScriptLang, pub dedicated_worker: Option, @@ -653,7 +655,7 @@ async fn get_script_info_for_hash_inner<'e, E: sqlx::PgExecutor<'e>>( ) -> error::Result> { let r = sqlx::query_as!( ScriptHashInfo, - "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by, path from script where hash = $1 AND workspace_id = $2", + "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by, path from script where hash = $1 AND workspace_id = $2", hash, w_id ) @@ -853,6 +855,8 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>( Option, Option, Option, + Option, + Option, Option, ScriptLang, Option, @@ -862,15 +866,15 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>( String, )> { let r_o = sqlx::query!( - "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script - WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false) - ORDER BY created_at DESC LIMIT 1", - script_path, - w_id, - require_locked - ) - .fetch_optional(db) - .await?; + "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script + WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false) + ORDER BY created_at DESC LIMIT 1", + script_path, + w_id, + require_locked + ) + .fetch_optional(db) + .await?; let script = utils::not_found_if_none(r_o, "script", script_path)?; @@ -880,6 +884,8 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>( script.concurrency_key, script.concurrent_limit, script.concurrency_time_window_s, + script.debounce_key, + script.debounce_delay_s, script.cache_ttl, script.language, script.dedicated_worker, diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 9c1af245d8..b0e771d138 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -294,10 +294,16 @@ pub struct Script { #[serde(skip_serializing_if = "Option::is_none")] pub envs: Option>, #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub concurrent_limit: Option, #[serde(skip_serializing_if = "Option::is_none")] pub concurrency_time_window_s: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub debounce_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub debounce_delay_s: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub dedicated_worker: Option, #[serde(skip_serializing_if = "Option::is_none")] pub ws_error_handler_muted: Option, @@ -312,8 +318,6 @@ pub struct Script { #[serde(skip_serializing_if = "Option::is_none")] pub restart_unless_cancelled: Option, #[serde(skip_serializing_if = "Option::is_none")] - pub concurrency_key: Option, - #[serde(skip_serializing_if = "Option::is_none")] pub visible_to_runner_only: Option, #[serde(skip_serializing_if = "Option::is_none")] pub no_main_func: Option, @@ -408,8 +412,13 @@ pub struct NewScript { pub tag: Option, pub draft_only: Option, pub envs: Option>, + pub concurrency_key: Option, pub concurrent_limit: Option, pub concurrency_time_window_s: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub debounce_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub debounce_delay_s: Option, pub cache_ttl: Option, pub dedicated_worker: Option, pub ws_error_handler_muted: Option, @@ -419,7 +428,6 @@ pub struct NewScript { pub restart_unless_cancelled: Option, pub deployment_message: Option, #[serde(skip_serializing_if = "Option::is_none")] - pub concurrency_key: Option, pub visible_to_runner_only: Option, pub no_main_func: Option, pub codebase: Option, @@ -555,6 +563,7 @@ pub async fn get_hub_script_by_path( let hub_base_url = HUB_BASE_URL.read().await.clone(); + // let result = http_get_from_hub( http_client, &format!("{}/raw/{}.ts", hub_base_url, path), @@ -770,6 +779,8 @@ pub async fn clone_script<'c>( has_preprocessor: s.has_preprocessor, on_behalf_of_email: s.on_behalf_of_email, assets: s.assets, + debounce_delay_s: s.debounce_delay_s, + debounce_key: s.debounce_key, }; let new_hash = hash_script(&ns); diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 4568e910e4..f2e5da5057 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -14,6 +14,7 @@ use std::{ collections::{HashMap, HashSet}, fs::{self, File}, io::Write, + ops::Deref, panic::Location, path::{Component, Path, PathBuf}, str::FromStr, @@ -250,6 +251,10 @@ lazy_static::lazy_static! { .unwrap_or(false); pub static ref MIN_VERSION: Arc> = Arc::new(RwLock::new(Version::new(0, 0, 0))); + /// Global flag indicating if all workers support the debouncing feature (>= 1.566.0) + /// Debouncing consolidates multiple dependency job requests within a time window to avoid redundant work + /// This flag is updated during worker initialization by checking the minimum version across all workers + pub static ref MIN_VERSION_SUPPORTS_DEBOUNCING: Arc> = Arc::new(RwLock::new(false)); pub static ref MIN_VERSION_IS_AT_LEAST_1_461: Arc> = Arc::new(RwLock::new(false)); pub static ref MIN_VERSION_IS_AT_LEAST_1_427: Arc> = Arc::new(RwLock::new(false)); pub static ref MIN_VERSION_IS_AT_LEAST_1_432: Arc> = Arc::new(RwLock::new(false)); @@ -266,7 +271,18 @@ pub const ROOT_CACHE_NOMOUNT_DIR: &str = concatcp!(TMP_DIR, "/cache_nomount/"); pub static MIN_VERSION_IS_LATEST: AtomicBool = AtomicBool::new(false); #[derive(Clone)] -pub struct HttpClient(pub ClientWithMiddleware); +pub struct HttpClient { + pub client: ClientWithMiddleware, + pub base_internal_url: Option, +} + +impl Deref for HttpClient { + type Target = ClientWithMiddleware; + + fn deref(&self) -> &Self::Target { + &self.client + } +} impl HttpClient { pub async fn post( @@ -275,10 +291,12 @@ impl HttpClient { headers: Option, body: &T, ) -> anyhow::Result { - let response_builder = self - .0 - .post(format!("{}{}", *BASE_INTERNAL_URL, url)) - .json(body); + let base_url = self + .base_internal_url + .clone() + .unwrap_or(BASE_INTERNAL_URL.clone().to_owned()); + + let response_builder = self.client.post(format!("{}{}", base_url, url)).json(body); let response_builder = match headers { Some(headers) => response_builder.headers(headers), @@ -302,9 +320,14 @@ impl HttpClient { } pub async fn get(&self, url: &str) -> anyhow::Result { + let base_url = self + .base_internal_url + .clone() + .unwrap_or(BASE_INTERNAL_URL.clone().to_owned()); + let response = self - .0 - .get(format!("{}{}", *BASE_INTERNAL_URL, url)) + .client + .get(format!("{}{}", base_url, url)) .send() .await .map_err(|e| anyhow::anyhow!(e))?; @@ -386,6 +409,10 @@ fn format_pull_query(peek: String) -> String { raw_flow, script_entrypoint_override, preprocessed FROM v2_job WHERE id = (SELECT id FROM peek) + ), delete_debounce AS NOT MATERIALIZED ( + DELETE FROM debounce_key + USING j + WHERE j.kind::text != 'flowdependencies' AND j.kind::text != 'appdependencies' AND j.kind::text != 'dependencies' AND debounce_key.job_id = j.id ) SELECT j.id, j.workspace_id, j.parent_job, j.created_by, started_at, scheduled_for, j.runnable_id, j.runnable_path, j.args, canceled_by, canceled_reason, j.kind, j.trigger, j.trigger_kind, j.permissioned_as, @@ -399,7 +426,8 @@ fn format_pull_query(peek: String) -> String { FROM q, j LEFT JOIN v2_job_status f USING (id) LEFT JOIN job_perms p ON p.job_id = j.id - LEFT JOIN v2_job pj ON j.parent_job = pj.id", + LEFT JOIN v2_job pj ON j.parent_job = pj.id + ", peek ); // tracing::debug!("pull query: {}", r); @@ -1034,6 +1062,7 @@ pub fn get_windmill_memory_usage() -> Option { } pub async fn update_min_version(conn: &Connection) -> bool { + tracing::debug!("Updating min version"); use crate::utils::{GIT_SEM_VERSION, GIT_VERSION}; let cur_version = GIT_SEM_VERSION.clone(); @@ -1065,6 +1094,9 @@ pub async fn update_min_version(conn: &Connection) -> bool { tracing::info!("Minimal worker version: {min_version}"); } + // Debouncing feature requires minimum version 1.566.0 across all workers + // This ensures all workers can handle debounce keys and stale data accumulation + *MIN_VERSION_SUPPORTS_DEBOUNCING.write().await = min_version >= Version::new(1, 566, 0); *MIN_VERSION_IS_AT_LEAST_1_461.write().await = min_version >= Version::new(1, 461, 0); *MIN_VERSION_IS_AT_LEAST_1_427.write().await = min_version >= Version::new(1, 427, 0); *MIN_VERSION_IS_AT_LEAST_1_432.write().await = min_version >= Version::new(1, 432, 0); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 19ac7bb28c..2ee22ae088 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -36,7 +36,7 @@ use windmill_common::bench::BenchmarkIter; use windmill_common::flow_conversations::{add_message_to_conversation_tx, MessageType}; use windmill_common::jobs::{JobTriggerKind, EMAIL_ERROR_HANDLER_USER_EMAIL}; use windmill_common::utils::{configure_client, now_from_db}; -use windmill_common::worker::{Connection, SCRIPT_TOKEN_EXPIRY}; +use windmill_common::worker::{Connection, MIN_VERSION_SUPPORTS_DEBOUNCING, SCRIPT_TOKEN_EXPIRY}; use windmill_common::{ auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username}, @@ -428,6 +428,8 @@ pub async fn push_init_job<'c>( custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, + custom_debounce_key: None, + debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, }), @@ -484,6 +486,8 @@ pub async fn push_periodic_bash_job<'c>( custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, + custom_debounce_key: None, + debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, }), @@ -1379,6 +1383,9 @@ async fn restart_job_if_perpetual_inner( .unwrap_or_else(|| ScriptLang::Deno), priority: queued_job.priority, apply_preprocessor: false, + // TODO(debouncing): handle properly + custom_debounce_key: None, + debounce_delay_s: None, }, queued_job .args @@ -2681,7 +2688,7 @@ pub fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> Strin } } -fn fullpath_with_workspace( +pub fn fullpath_with_workspace( workspace_id: &str, script_path: Option<&String>, job_kind: &JobKind, @@ -3400,7 +3407,7 @@ pub async fn push<'c, 'd>( mut email: &str, mut permissioned_as: String, token_prefix: Option<&str>, - scheduled_for_o: Option>, + mut scheduled_for_o: Option>, schedule_path: Option, parent_job: Option, root_job: Option, @@ -3418,6 +3425,7 @@ pub async fn push<'c, 'd>( running: bool, // whether the job is already running: only set this to true if you don't want the job to be picked up by a worker from the queue. It will also set started_at to now. end_user_email: Option, // If we know there is already a debounce job, we can use this for debouncing. + // NOTE: Only works with dependency jobs triggered by relative imports debounce_job_id_o: Option, ) -> Result<(Uuid, Transaction<'c, Postgres>), Error> { #[cfg(feature = "cloud")] @@ -3610,6 +3618,8 @@ pub async fn push<'c, 'd>( cache_ttl, dedicated_worker, _low_level_priority, + custom_debounce_key, + debounce_delay_s, ) = match job_payload { JobPayload::ScriptHash { hash, @@ -3622,6 +3632,8 @@ pub async fn push<'c, 'd>( dedicated_worker, priority, apply_preprocessor, + custom_debounce_key, + debounce_delay_s, } => { if apply_preprocessor { preprocessed = Some(false); @@ -3641,6 +3653,8 @@ pub async fn push<'c, 'd>( cache_ttl, dedicated_worker, priority, + custom_debounce_key, + debounce_delay_s, ) } JobPayload::FlowScript { @@ -3666,6 +3680,8 @@ pub async fn push<'c, 'd>( cache_ttl, dedicated_worker, None, + None, // custom_debounce_key removed for flow steps + None, // debounce_delay_s removed for flow steps ), JobPayload::FlowNode { id, path } => { let data = cache::flow::fetch_flow(_db, id).await?; @@ -3693,6 +3709,8 @@ pub async fn push<'c, 'd>( None, None, None, + None, + None, ) } JobPayload::AppScript { @@ -3714,6 +3732,8 @@ pub async fn push<'c, 'd>( cache_ttl, None, None, + None, + None, ), JobPayload::ScriptHub { path, apply_preprocessor } => { if path == "hub/7771/slack" || path == "hub/7836/slack" || path == "hub/9084/slack" { @@ -3746,6 +3766,8 @@ pub async fn push<'c, 'd>( None, None, None, + None, + None, ) } JobPayload::Code(RawCode { @@ -3759,6 +3781,8 @@ pub async fn push<'c, 'd>( concurrency_time_window_s, cache_ttl, dedicated_worker, + custom_debounce_key, + debounce_delay_s, }) => ( hash, path, @@ -3773,6 +3797,8 @@ pub async fn push<'c, 'd>( cache_ttl, dedicated_worker, None, + custom_debounce_key, + debounce_delay_s, ), JobPayload::Dependencies { hash, language, path, dedicated_worker } => ( Some(hash.0), @@ -3788,6 +3814,8 @@ pub async fn push<'c, 'd>( None, dedicated_worker, None, + None, + None, ), // CLI usage, is not modifying db, no need for debouncing. @@ -3805,6 +3833,8 @@ pub async fn push<'c, 'd>( None, None, None, + None, + None, ), // CLI usage, is not modifying db, no need for debouncing. @@ -3822,6 +3852,8 @@ pub async fn push<'c, 'd>( None, None, None, + None, + None, ), JobPayload::FlowDependencies { path, dedicated_worker, version } => { #[cfg(test)] @@ -3858,6 +3890,8 @@ pub async fn push<'c, 'd>( None, dedicated_worker, None, + None, + None, ) } JobPayload::AppDependencies { path, version } => ( @@ -3874,6 +3908,8 @@ pub async fn push<'c, 'd>( None, None, None, + None, + None, ), JobPayload::RawFlow { mut value, path, restarted_from } => { add_virtual_items_if_necessary(&mut value.modules); @@ -3924,6 +3960,8 @@ pub async fn push<'c, 'd>( let concurrency_key = value.concurrency_key.clone(); let concurrent_limit = value.concurrent_limit; let concurrency_time_window_s = value.concurrency_time_window_s; + let debounce_key = value.debounce_key.clone(); + let debounce_delay_s = value.debounce_delay_s; let cache_ttl = value.cache_ttl.map(|x| x as i32); let priority = value.priority; ( @@ -3940,6 +3978,8 @@ pub async fn push<'c, 'd>( cache_ttl, None, priority, + debounce_key, + debounce_delay_s, ) } JobPayload::SingleStepFlow { @@ -3959,6 +3999,8 @@ pub async fn push<'c, 'd>( tag_override, trigger_path, apply_preprocessor, + custom_debounce_key, + debounce_delay_s, } => { // Determine if this is a flow or a script let is_flow = flow_version.is_some(); @@ -4094,6 +4136,8 @@ pub async fn push<'c, 'd>( failure_module, concurrency_time_window_s, concurrent_limit, + debounce_key: custom_debounce_key.clone(), + debounce_delay_s, priority, cache_ttl: cache_ttl.map(|val| val as u32), concurrency_key: custom_concurrency_key.clone(), @@ -4119,6 +4163,8 @@ pub async fn push<'c, 'd>( cache_ttl, None, priority, + custom_debounce_key, + debounce_delay_s, ) } JobPayload::Flow { path, dedicated_worker, apply_preprocessor, version } => { @@ -4146,11 +4192,16 @@ pub async fn push<'c, 'd>( let concurrency_time_window_s = value.concurrency_time_window_s; let mut concurrent_limit = value.concurrent_limit; + let custom_debounce_key = value.debounce_key.clone(); + let mut debounce_delay_s = value.debounce_delay_s; + if !apply_preprocessor { value.preprocessor_module = None; } else { tag = None; concurrent_limit = None; + // TODO: May be re-enable? + debounce_delay_s = None; preprocessed = Some(false); } @@ -4185,6 +4236,8 @@ pub async fn push<'c, 'd>( cache_ttl, dedicated_worker, priority, + custom_debounce_key, + debounce_delay_s, ) } JobPayload::RestartedFlow { completed_job_id, step_id, branch_or_iteration_n } => { @@ -4235,6 +4288,8 @@ pub async fn push<'c, 'd>( let concurrency_key = value.concurrency_key.clone(); let concurrent_limit = value.concurrent_limit; let concurrency_time_window_s = value.concurrency_time_window_s; + let debounce_key = value.debounce_key.clone(); + let debounce_delay_s = value.debounce_delay_s; let cache_ttl = value.cache_ttl.map(|x| x as i32); // Keep inserting `value` if not all workers are updated. // Starting at `v1.440`, the value is fetched on pull from the version id. @@ -4258,6 +4313,8 @@ pub async fn push<'c, 'd>( cache_ttl, None, priority, + debounce_key, + debounce_delay_s, ) } JobPayload::DeploymentCallback { path } => ( @@ -4274,6 +4331,8 @@ pub async fn push<'c, 'd>( None, None, None, + None, + None, ), JobPayload::Identity => ( None, @@ -4289,6 +4348,8 @@ pub async fn push<'c, 'd>( None, None, None, + None, + None, ), JobPayload::Noop => ( None, @@ -4304,6 +4365,8 @@ pub async fn push<'c, 'd>( None, None, None, + None, + None, ), JobPayload::AIAgent { path } => ( None, @@ -4319,6 +4382,8 @@ pub async fn push<'c, 'd>( None, None, None, + None, + None, ), }; @@ -4329,7 +4394,10 @@ pub async fn push<'c, 'd>( // This is not the case for scripts, so we can potentially have multiple djobs for scripts at the same time. if let (Some(path), true) = ( &script_path, - cfg!(feature = "private") && job_kind.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING, + cfg!(feature = "private") + && job_kind.is_dependency() + && !*WMDEBUG_NO_DJOB_DEBOUNCING + && *MIN_VERSION_SUPPORTS_DEBOUNCING.read().await, ) { custom_concurrency_key = Some(format!("dependency:{workspace_id}/{path}")); concurrent_limit = Some(1); @@ -4475,10 +4543,17 @@ pub async fn push<'c, 'd>( job_kind.is_dependency(), script_path.clone(), *WMDEBUG_NO_DJOB_DEBOUNCING, + *MIN_VERSION_SUPPORTS_DEBOUNCING.read().await, // We only do debouncing for jobs triggered by relative imports // We do not want this be the case for normal djobs, since they will always be sequential. args.args.contains_key("triggered_by_relative_import"), ) { + (_, _, _, _, false, _) => { + tracing::warn!( + "Debouncing is disabled because workers are behind the minimum required version 1.566.0. \ + Please update workers to enable debouncing feature." + ); + } // === DEPENDENCY JOB DEBOUNCING === // // Debouncing consolidates multiple dependency job requests into a single execution, @@ -4489,7 +4564,8 @@ pub async fn push<'c, 'd>( // 2. Job is a dependency job // 3. Object path is provided (script/flow/app path) // 4. Fallback mode is disabled (normal operation) - // 5. Job was created by relative imports (triggered by dependency chain) + // 5. min version supports debouncing + // 6. Job was created by relative imports (triggered by dependency chain) // // How it works: // @@ -4509,7 +4585,7 @@ pub async fn push<'c, 'd>( // - Retrieve all accumulated nodes/components from debounce_stale_data // - Process all collected dependencies in single execution // - Clean up both debounce_key and debounce_stale_data entries - (true, true, Some(obj_path), false, true) => { + (true, true, Some(obj_path), false, true, true) => { // Generate unique debounce key: "workspace_id:object_path:dependency" // This ensures each workspace+path combination has independent debounce window let debounce_key = format!("{workspace_id}:{obj_path}:dependency"); @@ -4642,6 +4718,28 @@ pub async fn push<'c, 'd>( ); } }; + #[cfg(not(all(feature = "enterprise", feature = "private")))] + { + let (_, _) = (debounce_delay_s, custom_debounce_key); + scheduled_for_o = scheduled_for_o; + } + + #[cfg(all(feature = "enterprise", feature = "private"))] + if let Some(debounced_job_id) = crate::jobs_ee::maybe_apply_debouncing( + &job_id, + debounce_delay_s, + custom_debounce_key, + workspace_id, + script_path.clone(), + &job_kind, + &args, + &mut scheduled_for_o, + &mut tx, + ) + .await? + { + return Ok((debounced_job_id, tx)); + } if concurrent_limit.is_some() { insert_concurrency_key( @@ -5430,6 +5528,11 @@ pub async fn preprocess_dependency_job(job: &mut PulledJob, db: &DB) -> error::R job.runnable_id.replace(new_id.into()); + if !*windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING.read().await { + tx.commit().await?; + tracing::warn!("Debouncing is not supported on this version of Windmill. Minimum version required for debouncing support."); + return Ok(()); + } // === RETRIEVE ACCUMULATED DEBOUNCE DATA === // // For flows and apps, retrieve all nodes/components that were accumulated diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 1aa4a11a10..33fb3e8b84 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -87,6 +87,8 @@ async fn get_schedule_metadata<'c>( _custom_concurrency_key, _concurrent_limit, _concurrency_time_window_s, + _debounce_key, + _debounce_delay_s, _cache_ttl, _language, _dedicated_worker, @@ -264,6 +266,8 @@ pub async fn push_scheduled_job<'c>( tag_override: schedule.tag.clone(), trigger_path: None, apply_preprocessor: false, + custom_debounce_key: None, + debounce_delay_s: None, }, if schedule.tag.as_ref().is_some_and(|x| x != "") { schedule.tag.clone() @@ -318,6 +322,8 @@ pub async fn push_scheduled_job<'c>( custom_concurrency_key, concurrent_limit, concurrency_time_window_s, + custom_debounce_key, + debounce_delay_s, cache_ttl, language, dedicated_worker, @@ -360,11 +366,13 @@ pub async fn push_scheduled_job<'c>( custom_concurrency_key: None, concurrent_limit: None, concurrency_time_window_s: None, - cache_ttl: cache_ttl, - priority: priority, + cache_ttl, + priority, tag_override: schedule.tag.clone(), trigger_path: None, apply_preprocessor: false, + custom_debounce_key: None, + debounce_delay_s: None, }, if schedule.tag.as_ref().is_some_and(|x| x != "") { schedule.tag.clone() @@ -381,13 +389,15 @@ pub async fn push_scheduled_job<'c>( hash, path: schedule.script_path.clone(), custom_concurrency_key, - concurrent_limit: concurrent_limit, - concurrency_time_window_s: concurrency_time_window_s, - cache_ttl: cache_ttl, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, dedicated_worker, language, priority, apply_preprocessor: false, + custom_debounce_key, + debounce_delay_s, }, if schedule.tag.as_ref().is_some_and(|x| x != "") { schedule.tag.clone() diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 16abf40c86..5fe2ffac68 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1620,7 +1620,6 @@ pub async fn run_worker( _ => {} } } - add_time!(bench, "job pulled from DB"); let duration_pull_s = pull_time.elapsed().as_secs_f64(); let err_pull = job.is_ok(); diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 9e956dbfb8..850ae46c17 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -4454,6 +4454,8 @@ pub fn raw_script_to_payload( 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, }), tag, delete_after_use, @@ -4518,6 +4520,8 @@ pub async fn script_to_payload( concurrency_key, concurrent_limit, concurrency_time_window_s, + debounce_key, + debounce_delay_s, cache_ttl, language, dedicated_worker, @@ -4544,6 +4548,8 @@ pub async fn script_to_payload( custom_concurrency_key: concurrency_key, concurrent_limit, concurrency_time_window_s, + custom_debounce_key: debounce_key, + debounce_delay_s, cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(), language, dedicated_worker, diff --git a/cli/src/commands/script/script.ts b/cli/src/commands/script/script.ts index b0e80b764b..c35e2c56f7 100644 --- a/cli/src/commands/script/script.ts +++ b/cli/src/commands/script/script.ts @@ -349,6 +349,8 @@ export async function handleFile( has_preprocessor: typed?.has_preprocessor, priority: typed?.priority, concurrency_key: typed?.concurrency_key, + debounce_key: typed?.debounce_key, + debounce_delay_s: typed?.debounce_delay_s, codebase: await codebase?.getDigest(), timeout: typed?.timeout, on_behalf_of_email: typed?.on_behalf_of_email, @@ -388,6 +390,8 @@ export async function handleFile( typed.timeout == remote.timeout && //@ts-ignore typed.concurrency_key == remote["concurrency_key"] && + typed.debounce_key == remote["debounce_key"] && + typed.debounce_delay_s == remote["debounce_delay_s"] && typed.codebase == remote.codebase && typed.on_behalf_of_email == remote.on_behalf_of_email) ) { diff --git a/cli/src/guidance/flow_guidance.ts b/cli/src/guidance/flow_guidance.ts index 4766a7725b..4b10e946b3 100644 --- a/cli/src/guidance/flow_guidance.ts +++ b/cli/src/guidance/flow_guidance.ts @@ -29,6 +29,8 @@ value: concurrent_limit: 0 # Limit concurrent executions concurrency_key: "string" # Custom concurrency grouping concurrency_time_window_s: 0 + custom_debounce_key: "key" + debounce_delay_s: 0 skip_expr: "javascript_expression" # Skip workflow condition cache_ttl: 0 # Cache results duration priority: 0 # Execution priority @@ -59,6 +61,8 @@ value: concurrent_limit: 0 concurrency_time_window_s: 0 custom_concurrency_key: "key" + custom_debounce_key: "key" + debounce_delay_s: 0 is_trigger: false assets: [] \`\`\` @@ -427,4 +431,4 @@ schema: \`\`\` When generating OpenFlow YAML, ensure proper indentation, valid YAML syntax, and logical step dependencies. Always include meaningful summaries and proper input transforms to connect workflow steps. -`; \ No newline at end of file +`; diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index 8e1a1db19d..6b84e2c295 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -520,6 +520,8 @@ dedicated_worker: script.dedicated_worker, concurrent_limit: script.concurrent_limit, concurrency_time_window_s: script.concurrency_time_window_s, + debounce_key: emptyString(script.debounce_key) ? undefined : script.debounce_key, + debounce_delay_s: script.debounce_delay_s, cache_ttl: script.cache_ttl, ws_error_handler_muted: script.ws_error_handler_muted, priority: script.priority, @@ -1261,6 +1263,48 @@ +
+ {#snippet header()} + + Debounce Jobs + + {/snippet} +
+ + +
+
{#snippet header()} {/if} + + {#if customUi?.settingsTabs?.debouncing != false} +
+
+ { + if (flowStore.val.value.debounce_delay_s) { + flowStore.val.value.debounce_delay_s = undefined + } else { + flowStore.val.value.debounce_delay_s = 1 + } + }} + options={{ + right: 'Debounce limits', + rightTooltip: 'Consolidate multiple flow executions into a single run within a time window', + rightDocumentationLink: 'https://www.windmill.dev/docs/core_concepts/debouncing' + }} + class="py-1" + eeOnly={true} + /> +
+ + {#if flowStore.val.value.debounce_delay_s} +
+ + +
+ {/if} +
+ {/if} + {/if} + {#if debouncing} + +
+ +
+ {#snippet text()} + Debouncing + {/snippet} + + {/if} {#if cache}