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

feat: job debouncing (#6878)

Browse Source 
* feat(EE): job debouncing

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

* remove 'script' file

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

* more work

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

* properly gate it behind enterprise

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

* update ee repo ref

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

* change ee repo ref again

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

* remove unused variable

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

* feat(EE): implement TODOs and enhance tracing for job debouncing

- Add database index on script(workspace_id, debounce_key) for efficient lookups
- Update minimum version requirement to 1.564.0 throughout codebase
- Add tracing warnings when debouncing is disabled due to worker version mismatch
- Fix all documentation links from TODO placeholders to proper URLs
- Replace Gauge icon with Timer icon for debouncing UI elements
- Update placeholder text and tooltips with clear descriptions

Co-authored-by: Pyra <pyranota@users.noreply.github.com>

* create -> crate

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

* remove index

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

* some updates

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

* fix once more

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

* fix it once more

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

* Remove flow step debouncing, keep top-level flow debouncing

- Remove debounce fields from RawScript and FlowScript FlowModuleValue variants
- Remove debounce fields from JobPayload::FlowScript and RawCode
- Update raw_script_to_payload function signature
- Remove debouncing UI from flow step runtime settings
- Remove debouncing toggle handler and indicator badge
- Preserve top-level flow debouncing in FlowSettings

Co-authored-by: Pyra <pyranota@users.noreply.github.com>

* cleanup

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

* fixup claude's work

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

* cleanup: remove dbg! statements, update min version to 1.566.0, add comprehensive comments

- Removed all dbg! macro calls from production code
- Updated MIN_VERSION_SUPPORTS_DEBOUNCING from 1.564.0 to 1.566.0
- Added comprehensive documentation comments explaining:
  - Debouncing feature purpose and mechanics
  - Database schema for debounce_key and debounce_stale_data tables
  - Version check logic and guard functions
- Improved code clarity and maintainability

Co-authored-by: Pyra <pyranota@users.noreply.github.com>

* improve fallback

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

* remove comments from old migration

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

* fix pull

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

* fix once more

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

* Update frontend/src/lib/components/ScriptBuilder.svelte

Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com>

* add sqlx cache

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

* fix: frontend CI - fix debounce field names and remove leftover flow step debouncing

- Fix ScriptBuilder.svelte: change custom_debounce_key to debounce_key
- Add debounce_key and debounce_delay_s fields to NewScript schema in openapi.yaml
- Regenerate frontend types from OpenAPI spec
- Remove leftover flow step debouncing code from FlowModuleComponent.svelte
- Remove debounce fields from RawScript in openflow.openapi.yaml
- Remove unused Timer import from FlowModuleHeader.svelte

All frontend checks now passing (0 errors, 0 warnings)

Co-authored-by: Pyra <pyranota@users.noreply.github.com>

* fix ci

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

* remove unused import

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

* fix ci again

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

* udpate ee repo ref

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

* CI doesn't want to be fixed but I still try

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

* nits

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

* ci...

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

* Update ee-repo-ref.txt

* safer migration

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

* reduce noise in logs

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

* fix cli for scripts

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

* nit

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

---------

Signed-off-by: pyranota <pyra@duck.com>
Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: Pyra <pyranota@users.noreply.github.com>
Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Pyra
2025-10-23 20:18:08 +00:00
committed by GitHub
co-authored by Pyra claude[bot] claude[bot] windmill-internal-app[bot]
parent b33e3d8505
commit edece035f8
42 changed files with 2189 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d.json
Generated
+94
View File
@@ -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"
}
backend/.sqlx/query-27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32.json
Generated
+133
View File
@@ -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"
}
backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
Generated
+1 -1
View File
@@ -15,7 +15,7 @@
]
},
"nullable": [
null
true
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
backend/.sqlx/query-f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf.json
Generated
+150
View File
@@ -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"
}
backend/.sqlx/query-f1fe8508f6cd29d4c0d354473529fc46f506ddf98d45fc870e2855c4a4b424ea.json
Generated
+23
View File
@@ -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"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
c8d57987b72fd15f2e3b8d7f6501edaac8235965
e12fc17ec760fe720edc2d930457ebb0e06ede48
backend/migrations/20251017120625_job_debouncing_extra.down.sql
+4
View File
@@ -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;
backend/migrations/20251017120625_job_debouncing_extra.up.sql
+6
View File
@@ -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;
backend/src/main.rs
+2 -2
View File
@@ -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,
backend/src/monitor.rs
+6 -5
View File
@@ -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!(
backend/tests/common/mod.rs
+117 -18
View File
@@ -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<Postgres>) -> (windmill_api_client::Client, u1
(client, port, server)
}
pub async fn init_client_agent_mode(
db: Pool<Postgres>,
) -> (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<Postgres>) -> anyhow::Result<Self> {
Self::start_inner(db, false).await
}
pub async fn start_agent_mode(db: Pool<Postgres>) -> anyhow::Result<Self> {
Self::start_inner(db, true).await
}
async fn start_inner(db: Pool<Postgres>, agent_mode: bool) -> anyhow::Result<Self> {
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<String, serde_json::Value>,
pub debounce_job_id_o: Option<Uuid>,
pub scheduled_for_o: Option<chrono::DateTime<chrono::Utc>>,
}
impl From<JobPayload> 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<Uuid>) -> Self {
self.debounce_job_id_o = job_id;
self
}
pub fn push_arg_scheduled_for_o(
mut self,
scheduled_for_o: Option<chrono::DateTime<chrono::Utc>>,
) -> Self {
self.scheduled_for_o = scheduled_for_o;
self
}
pub async fn push(self, db: &Pool<Postgres>) -> 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<Postgres>, port: u16) -> CompletedJob {
pub async fn run_until_complete(
self,
db: &Pool<Postgres>,
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<F: Future<Output = ()>>(
self,
db: &Pool<Postgres>,
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<Postgres>,
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<Fut: std::future::Future>(
db: &Pool<Postgres>,
// db: &Pool<Postgres>,
// If set to http, worker will be started in agent mode.
conn: impl Into<Connection>,
inner: Fut,
port: u16,
) -> <Fut as std::future::Future>::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<Fut: std::future::Future>(
}
pub fn spawn_test_worker(
db: &Pool<Postgres>,
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!
///
backend/tests/fixtures/job_debouncing.sql
Vendored
+171
View File
@@ -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';
backend/tests/job_payload.rs
+31 -30
View File
@@ -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<Postgres>) -> 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);
backend/tests/python_jobs.rs
+15 -5
View File
@@ -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();
backend/tests/relative_imports.rs
+894 -34
View File
File diff suppressed because it is too large Load Diff
backend/tests/retry.rs
+7 -7
View File
@@ -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();
backend/tests/suspend_resume.rs
+4 -4
View File
@@ -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();
backend/tests/worker.rs
+62 -42
View File
@@ -77,7 +77,7 @@ async fn test_iteration(db: Pool<Postgres>) -> 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<Postgres>) -> anyhow::Result<()> {
let result =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("items", json!((0..257).collect::<Vec<_>>()))
.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<Postgres>) -> 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<Postgres>) -> anyhow::Result<()> {
let job =
RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None })
.arg("items", json!((0..50).collect::<Vec<_>>()))
.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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<String, String> {
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<String, String> {
// }))
// .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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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]));
backend/windmill-api/openapi.yaml
+4
View File
@@ -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:
backend/windmill-api/src/flows.rs
+22 -1
View File
@@ -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<NewFlow>,
) -> 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<String> {
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<bool>,
@@ -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": [
backend/windmill-api/src/jobs.rs
+17
View File
@@ -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,
backend/windmill-api/src/scripts.rs
+24 -4
View File
@@ -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<String>,
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(())
}
}
backend/windmill-api/src/triggers/trigger_helpers.rs
+4
View File
@@ -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!(
backend/windmill-api/src/workspaces_export.rs
+7 -4
View File
@@ -102,6 +102,10 @@ struct ScriptMetadata {
pub has_preprocessor: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_behalf_of_email: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
}
pub fn is_none_or_false(val: &Option<bool>) -> 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();
backend/windmill-common/src/agent_workers.rs
+10 -3
View File
@@ -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<String>,
base_internal_url: Option<String>,
) -> 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)]
backend/windmill-common/src/error.rs
+2
View File
@@ -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 {
backend/windmill-common/src/flows.rs
+7 -2
View File
@@ -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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub skip_expr: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -136,8 +143,6 @@ pub struct FlowValue {
// Priority at the flow level
pub priority: Option<i16>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub chat_input_enabled: Option<bool>,
}
backend/windmill-common/src/jobs.rs
+36 -8
View File
@@ -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<String>,
/// How many jobs can run at the same time
concurrent_limit: Option<i32>,
/// In seconds
concurrency_time_window_s: Option<i32>,
/// If not set, will be inferred from the hash(path + step_id + inputs)
custom_debounce_key: Option<String>,
/// Debouncing delay will be determined by the first job with the key.
/// All subsequent jobs with Some will get debounced.
/// If the job has no delay, it will execute immediately, fully ignoring pending delays.
debounce_delay_s: Option<i32>,
cache_ttl: Option<i32>,
dedicated_worker: Option<bool>,
language: ScriptLang,
@@ -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<String>,
/// How many jobs can run at the same time
concurrent_limit: Option<i32>,
/// In seconds
concurrency_time_window_s: Option<i32>,
cache_ttl: Option<i32>,
dedicated_worker: Option<bool>,
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<String>,
@@ -367,6 +378,7 @@ pub enum JobPayload {
cache_ttl: Option<i32>,
},
/// 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<usize>,
},
/// Flow Preview
RawFlow {
value: FlowValue,
path: Option<String>,
@@ -435,6 +450,8 @@ pub enum JobPayload {
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
custom_debounce_key: Option<String>,
debounce_delay_s: Option<i32>,
cache_ttl: Option<i32>,
priority: Option<i16>,
tag_override: Option<String>,
@@ -469,6 +486,8 @@ pub struct RawCode {
pub custom_concurrency_key: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub custom_debounce_key: Option<String>,
pub debounce_delay_s: Option<i32>,
pub cache_ttl: Option<i32>,
pub dedicated_worker: Option<bool>,
}
@@ -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<Option<Uuid>> {
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!(
backend/windmill-common/src/lib.rs
+17 -11
View File
@@ -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<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub debounce_key: Option<String>,
pub debounce_delay_s: Option<i32>,
pub cache_ttl: Option<i32>,
pub language: ScriptLang,
pub dedicated_worker: Option<bool>,
@@ -653,7 +655,7 @@ async fn get_script_info_for_hash_inner<'e, E: sqlx::PgExecutor<'e>>(
) -> error::Result<Option<ScriptHashInfo>> {
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<String>,
Option<i32>,
Option<i32>,
Option<String>,
Option<i32>,
Option<i32>,
ScriptLang,
Option<bool>,
@@ -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,
backend/windmill-common/src/scripts.rs
+14 -3
View File
@@ -294,10 +294,16 @@ pub struct Script {
#[serde(skip_serializing_if = "Option::is_none")]
pub envs: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub dedicated_worker: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub ws_error_handler_muted: Option<bool>,
@@ -312,8 +318,6 @@ pub struct Script {
#[serde(skip_serializing_if = "Option::is_none")]
pub restart_unless_cancelled: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub visible_to_runner_only: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub no_main_func: Option<bool>,
@@ -408,8 +412,13 @@ pub struct NewScript {
pub tag: Option<String>,
pub draft_only: Option<bool>,
pub envs: Option<Vec<String>>,
pub concurrency_key: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
pub cache_ttl: Option<i32>,
pub dedicated_worker: Option<bool>,
pub ws_error_handler_muted: Option<bool>,
@@ -419,7 +428,6 @@ pub struct NewScript {
pub restart_unless_cancelled: Option<bool>,
pub deployment_message: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
pub visible_to_runner_only: Option<bool>,
pub no_main_func: Option<bool>,
pub codebase: Option<String>,
@@ -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);
backend/windmill-common/src/worker.rs
+40 -8
View File
@@ -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<RwLock<Version>> = 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<RwLock<bool>> = Arc::new(RwLock::new(false));
pub static ref MIN_VERSION_IS_AT_LEAST_1_461: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
pub static ref MIN_VERSION_IS_AT_LEAST_1_427: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
pub static ref MIN_VERSION_IS_AT_LEAST_1_432: Arc<RwLock<bool>> = 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<String>,
}
impl Deref for HttpClient {
type Target = ClientWithMiddleware;
fn deref(&self) -> &Self::Target {
&self.client
}
}
impl HttpClient {
pub async fn post<T: Serialize, R: DeserializeOwned>(
@@ -275,10 +291,12 @@ impl HttpClient {
headers: Option<HeaderMap>,
body: &T,
) -> anyhow::Result<R> {
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<R: DeserializeOwned>(&self, url: &str) -> anyhow::Result<R> {
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<i64> {
}
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);
backend/windmill-queue/src/jobs.rs
+109 -6
View File
@@ -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<chrono::DateTime<chrono::Utc>>,
mut scheduled_for_o: Option<chrono::DateTime<chrono::Utc>>,
schedule_path: Option<String>,
parent_job: Option<Uuid>,
root_job: Option<Uuid>,
@@ -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<String>,
// 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<Uuid>,
) -> 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
backend/windmill-queue/src/schedule.rs
+15 -5
View File
@@ -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()
backend/windmill-worker/src/worker.rs
-1
View File
@@ -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();
backend/windmill-worker/src/worker_flow.rs
+6
View File
@@ -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,
cli/src/commands/script/script.ts
+4
View File
@@ -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)
) {
cli/src/guidance/flow_guidance.ts
+5 -1
View File
@@ -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.
`;
`;
frontend/src/lib/components/ScriptBuilder.svelte
+44
View File
@@ -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 @@
</Label>
</div>
</Section>
<Section label="Debouncing" eeOnly>
{#snippet header()}
<Tooltip
documentationLink="https://www.windmill.dev/docs/core_concepts/debouncing"
>
Debounce Jobs
</Tooltip>
{/snippet}
<div class="flex flex-col gap-4">
<Label label="Debounce Delay in seconds. (if not set - disabled)">
<SecondsInput
bind:seconds={script.debounce_delay_s}
/>
<Button
size="sm"
color="light"
on:click={() => {
script.debounce_delay_s = undefined
script.debounce_key = undefined
}}
variant="border">Remove Debouncing</Button
>
</Label>
<Label label="Custom debounce key (optional)">
{#snippet header()}
<Tooltip
documentationLink="https://www.windmill.dev/docs/core_concepts/debouncing#custom-debounce-key"
>
Debounce Keys are global, you can have them be workspace specific
using the variable `$workspace`. You can also use an argument's value
using `$args[name_of_arg]`</Tooltip
>
{/snippet}
<input
type="text"
autofocus
bind:value={script.debounce_key}
placeholder={`$workspace/script/${script.path}-$args[foo]`}
/>
</Label>
</div>
</Section>
<Section label="Worker group tag (queue)">
{#snippet header()}
<Tooltip
frontend/src/lib/components/custom_ui.ts
+1
View File
@@ -20,6 +20,7 @@ export type FlowBuilderWhitelabelCustomUi = {
earlyReturn?: boolean
workerGroup?: boolean
concurrency?: boolean
debouncing?: boolean
cache?: boolean
}
triggers?: boolean
frontend/src/lib/components/flows/content/FlowSettings.svelte
+57
View File
@@ -473,6 +473,63 @@
</div>
{/if}
<!-- Debouncing Section -->
{#if customUi?.settingsTabs?.debouncing != false}
<div>
<div class="flex flex-row items-center gap-2">
<Toggle
textClass="font-normal text-sm"
color="nord"
size="xs"
disabled={!$enterpriseLicense}
checked={Boolean(flowStore.val.value.debounce_delay_s)}
on:change={() => {
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}
/>
</div>
{#if flowStore.val.value.debounce_delay_s}
<div class="flex flex-col gap-4">
<Label label="Delay in seconds">
<SecondsInput
disabled={!$enterpriseLicense}
bind:seconds={flowStore.val.value.debounce_delay_s}
/>
</Label>
<Label label="Custom debounce key (optional)">
{#snippet header()}
<Tooltip>
Debounce keys are global, you can have them be workspace specific using the
variable `$workspace`. You can also use an argument's value using
`$args[name_of_arg]`</Tooltip
>
{/snippet}
<!-- svelte-ignore a11y_autofocus -->
<input
type="text"
autofocus
disabled={!$enterpriseLicense}
bind:value={flowStore.val.value.debounce_key}
placeholder={`$workspace/script/${$pathStore}-$args[foo]`}
/>
</Label>
</div>
{/if}
</div>
{/if}
<!-- Priority Section -->
<Toggle
textClass="font-medium"
frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte
+18 -1
View File
@@ -16,7 +16,8 @@
X,
Play,
Loader2,
TriangleAlert
TriangleAlert,
Timer
} from 'lucide-svelte'
import { createEventDispatcher, getContext } from 'svelte'
import { fade } from 'svelte/transition'
@@ -66,6 +67,8 @@
modType?: string | undefined
nodeState?: FlowNodeState
concurrency?: boolean
// TODO: Implement for this one. See how concurrency is implemented.
debouncing?: boolean
retries?: number | undefined
warningMessage?: string | undefined
isTrigger?: boolean
@@ -99,6 +102,7 @@
modType = undefined,
nodeState,
concurrency = false,
debouncing = false,
retries = undefined,
warningMessage = undefined,
isTrigger = false,
@@ -308,6 +312,19 @@
{/snippet}
</Popover>
{/if}
{#if debouncing}
<Popover notClickable>
<div
transition:fade|local={{ duration: 200 }}
class="center-center rounded border bg-surface border-gray-400 text-secondary px-1 py-0.5"
>
<Timer size={12} />
</div>
{#snippet text()}
Debouncing
{/snippet}
</Popover>
{/if}
{#if cache}
<Popover notClickable>
<div
openflow.openapi.yaml
+5
View File
@@ -54,6 +54,10 @@ components:
type: string
concurrency_time_window_s:
type: number
debounce_delay_s:
type: number
debounce_key:
type: string
skip_expr:
type: string
cache_ttl:
@@ -266,6 +270,7 @@ components:
- graphql
- nativets
- php
# TODO: Add missing languages
path:
type: string
lock: