From 63102b036cab45e26acf255de22fcb37469f7915 Mon Sep 17 00:00:00 2001 From: Kai Jellinghaus Date: Tue, 11 Apr 2023 12:24:13 +0200 Subject: [PATCH] feat(backend): Redis based queue (#1324) * Merge? * Fix V8 breaking change * WIP * WIP * Cleanup * Move to git reference * Fix Merge conflict * update --------- Co-authored-by: Ruben Fiszel --- backend/.cargo/config.toml | 2 +- backend/Cargo.lock | 288 ++++++++++++++++-- backend/Cargo.toml | 6 + backend/src/main.rs | 49 ++- backend/tests/worker.rs | 13 +- backend/windmill-api/Cargo.toml | 1 + backend/windmill-api/src/apps.rs | 15 +- backend/windmill-api/src/flows.rs | 60 ++-- backend/windmill-api/src/jobs.rs | 87 ++++-- backend/windmill-api/src/lib.rs | 2 + backend/windmill-api/src/oauth2.rs | 10 +- backend/windmill-api/src/schedule.rs | 45 ++- backend/windmill-api/src/scripts.rs | 73 +++-- backend/windmill-audit/src/lib.rs | 4 +- backend/windmill-common/src/utils.rs | 4 +- backend/windmill-queue/Cargo.toml | 3 + backend/windmill-queue/src/jobs.rs | 99 ++++-- backend/windmill-queue/src/lib.rs | 2 + .../windmill-queue/src/queue_transaction.rs | 161 ++++++++++ backend/windmill-queue/src/schedule.rs | 28 +- backend/windmill-worker/Cargo.toml | 3 +- backend/windmill-worker/src/jobs.rs | 30 +- backend/windmill-worker/src/worker.rs | 64 ++-- backend/windmill-worker/src/worker_flow.rs | 52 +++- benchmarks/example_fake_user.json | 53 ---- benchmarks/identity.json | 7 + benchmarks/main.ts | 76 ++--- benchmarks/worker.ts | 33 +- frontend/CaddyfileDev | 4 +- 29 files changed, 920 insertions(+), 354 deletions(-) create mode 100644 backend/windmill-queue/src/queue_transaction.rs delete mode 100644 benchmarks/example_fake_user.json create mode 100644 benchmarks/identity.json diff --git a/backend/.cargo/config.toml b/backend/.cargo/config.toml index 3267e5528e..dfcffc0242 100644 --- a/backend/.cargo/config.toml +++ b/backend/.cargo/config.toml @@ -18,4 +18,4 @@ rustflags = [ rustflags = [ "-C", "link-arg=-undefined", "-C", "link-arg=dynamic_lookup", -] +] \ No newline at end of file diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 7184f082fe..c655f73d88 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -178,6 +178,55 @@ dependencies = [ "zstd-safe", ] +[[package]] +name = "async-executor" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fa3dc5f2a8564f07759c008b9109dc0d39de92a88d5588b8a5036d286383afb" +dependencies = [ + "async-lock", + "async-task", + "concurrent-queue", + "fastrand", + "futures-lite", + "slab", +] + +[[package]] +name = "async-global-executor" +version = "2.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1b6f5d7df27bd294849f8eec66ecfc63d11814df7a4f5d74168a2394467b776" +dependencies = [ + "async-channel", + "async-executor", + "async-io", + "async-lock", + "blocking", + "futures-lite", + "once_cell", +] + +[[package]] +name = "async-io" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fc5b45d93ef0529756f812ca52e44c221b35341892d3dcc34132ac02f3dd2af" +dependencies = [ + "async-lock", + "autocfg", + "cfg-if", + "concurrent-queue", + "futures-lite", + "log", + "parking", + "polling", + "rustix", + "slab", + "socket2", + "waker-fn", +] + [[package]] name = "async-lock" version = "2.7.0" @@ -217,6 +266,32 @@ dependencies = [ "syn 2.0.13", ] +[[package]] +name = "async-std" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "62565bb4402e926b29953c785397c6dc0391b7b446e45008b0049eb43cec6f5d" +dependencies = [ + "async-channel", + "async-global-executor", + "async-io", + "async-lock", + "crossbeam-utils", + "futures-channel", + "futures-core", + "futures-io", + "futures-lite", + "gloo-timers", + "kv-log-macro", + "log", + "memchr", + "once_cell", + "pin-project-lite", + "pin-utils", + "slab", + "wasm-bindgen-futures", +] + [[package]] name = "async-stripe" version = "0.14.1" @@ -240,6 +315,12 @@ dependencies = [ "uuid 0.8.2", ] +[[package]] +name = "async-task" +version = "4.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ecc7ab41815b3c653ccd2978ec3255c81349336702dfdf62ee6f7069b12a3aae" + [[package]] name = "async-timer" version = "0.7.4" @@ -286,6 +367,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "atomic-waker" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1181e1e0d1fce796a03db1ae795d67167da795f9cf4a39c37589e85ef57f26d3" + [[package]] name = "autocfg" version = "1.1.0" @@ -360,6 +447,19 @@ version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8c3c1a368f70d6cf7302d78f8f7093da241fb8e8807c05cc9e51a125895a6d5b" +[[package]] +name = "bb8" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e9f4fa9768efd269499d8fba693260cfc670891cf6de3adc935588447a77cc8" +dependencies = [ + "async-trait", + "futures-channel", + "futures-util", + "parking_lot 0.11.2", + "tokio", +] + [[package]] name = "better_scoped_tls" version = "0.1.0" @@ -434,6 +534,21 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8d696c370c750c948ada61c69a0ee2cbbb9c50b1019ddb86d9317157a99c2cae" +[[package]] +name = "blocking" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77231a1c8f801696fc0123ec6150ce92cffb8e164a02afb9c8ddee0e9b65ad65" +dependencies = [ + "async-channel", + "async-lock", + "async-task", + "atomic-waker", + "fastrand", + "futures-lite", + "log", +] + [[package]] name = "bstr" version = "0.2.17" @@ -620,6 +735,20 @@ dependencies = [ "unicode-width", ] +[[package]] +name = "combine" +version = "4.6.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35ed6e9d84f0b51a7f52daf1c7d71dd136fd7a3f41a8462b8cdb8c78d920fad4" +dependencies = [ + "bytes", + "futures-core", + "memchr", + "pin-project-lite", + "tokio", + "tokio-util", +] + [[package]] name = "concolor-override" version = "1.0.0" @@ -792,6 +921,16 @@ dependencies = [ "typenum", ] +[[package]] +name = "ctor" +version = "0.1.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d2301688392eb071b0bf1a37be05c469d3cc4dbbd95df672fe28ab021e6a096" +dependencies = [ + "quote", + "syn 1.0.109", +] + [[package]] name = "cxx" version = "1.0.94" @@ -1397,10 +1536,22 @@ dependencies = [ ] [[package]] -name = "gosyn" -version = "0.2.2" +name = "gloo-timers" +version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1584c4cbcb6f1d97481c19d1951c64f6e1256e5a03c28159ae45431c00511ee7" +checksum = "9b995a66bb87bebce9a0f4a95aed01daca4872c050bfcb21653361c03bc35e5c" +dependencies = [ + "futures-channel", + "futures-core", + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "gosyn" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "248a4b26b282ec2e90272f7e7bd4764eaa6378dee13497ef09aa0efebd80cc85" dependencies = [ "strum", "unic-ucd-category", @@ -1794,6 +1945,15 @@ version = "0.2.19" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4933f3f57a8e9d9da04db23fb153356ecaf00cbd14aee46279c33dc80925c37" +[[package]] +name = "kv-log-macro" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de8b303297635ad57c9f5059fd9cee7a47f8e8daa09df0fcd07dd39fb22977f" +dependencies = [ + "log", +] + [[package]] name = "lalrpop" version = "0.19.9" @@ -1976,6 +2136,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "abb12e687cfb44aa40f41fc3978ef76448f9b6038cad6aef4259d3c095a2382e" dependencies = [ "cfg-if", + "value-bag", ] [[package]] @@ -2254,9 +2415,9 @@ dependencies = [ [[package]] name = "openssl" -version = "0.10.49" +version = "0.10.50" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d2f106ab837a24e03672c59b1239669a0596406ff657c3c0835b6b7f0f35a33" +checksum = "7e30d8bc91859781f0a943411186324d580f2bbeb71b452fe91ae344806af3f1" dependencies = [ "bitflags", "cfg-if", @@ -2286,9 +2447,9 @@ checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" [[package]] name = "openssl-sys" -version = "0.9.84" +version = "0.9.85" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a20eace9dc2d82904039cb76dcf50fb1a0bba071cfd1629720b5d6f1ddba0fa" +checksum = "0d3d193fb1488ad46ffe3aaabc912cc931d02ee8518fe2959aea8ef52718b0c0" dependencies = [ "cc", "libc", @@ -2565,6 +2726,22 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "polling" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4be1c66a6add46bff50935c313dae30a5030cf8385c5206e8a95e9e9def974aa" +dependencies = [ + "autocfg", + "bitflags", + "cfg-if", + "concurrent-queue", + "libc", + "log", + "pin-project-lite", + "windows-sys 0.48.0", +] + [[package]] name = "ppv-lite86" version = "0.2.17" @@ -2711,6 +2888,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "radix_fmt" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce082a9940a7ace2ad4a8b7d0b1eac6aa378895f18be598230c5f2284ac05426" + [[package]] name = "rand" version = "0.7.3" @@ -2782,6 +2965,27 @@ dependencies = [ "rand_core 0.5.1", ] +[[package]] +name = "redis" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ea8c51b5dc1d8e5fd3350ec8167f464ec0995e79f2e90a075b63371500d557f" +dependencies = [ + "async-std", + "async-trait", + "bytes", + "combine", + "futures-util", + "itoa", + "percent-encoding", + "pin-project-lite", + "ryu", + "sha1_smol", + "tokio", + "tokio-util", + "url", +] + [[package]] name = "redox_syscall" version = "0.2.16" @@ -2934,6 +3138,21 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rsmq_async" +version = "5.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "086675eeb88ca55fa3b25c0075615d31dd9eb0adee219a9dcd999090cb371cab" +dependencies = [ + "async-trait", + "bb8", + "lazy_static", + "radix_fmt", + "rand 0.8.5", + "redis", + "thiserror", +] + [[package]] name = "rust-embed" version = "6.6.1" @@ -3043,7 +3262,7 @@ dependencies = [ [[package]] name = "rustpython-ast" version = "0.2.0" -source = "git+https://github.com/RustPython/RustPython#dde92d1269b5fbd09cf5dba5e7817f0e2a96c49f" +source = "git+https://github.com/RustPython/RustPython#c9546c2419f2e4db7e2987f30f0508dc29b30035" dependencies = [ "num-bigint", "rustpython-compiler-core", @@ -3052,7 +3271,7 @@ dependencies = [ [[package]] name = "rustpython-compiler-core" version = "0.2.0" -source = "git+https://github.com/RustPython/RustPython#dde92d1269b5fbd09cf5dba5e7817f0e2a96c49f" +source = "git+https://github.com/RustPython/RustPython#c9546c2419f2e4db7e2987f30f0508dc29b30035" dependencies = [ "bitflags", "bstr", @@ -3065,7 +3284,7 @@ dependencies = [ [[package]] name = "rustpython-parser" version = "0.2.0" -source = "git+https://github.com/RustPython/RustPython#dde92d1269b5fbd09cf5dba5e7817f0e2a96c49f" +source = "git+https://github.com/RustPython/RustPython#c9546c2419f2e4db7e2987f30f0508dc29b30035" dependencies = [ "ahash 0.7.6", "anyhow", @@ -3127,7 +3346,7 @@ dependencies = [ "schemars_derive", "serde", "serde_json", - "uuid 1.3.0", + "uuid 1.3.1", ] [[package]] @@ -3237,9 +3456,9 @@ dependencies = [ [[package]] name = "serde" -version = "1.0.159" +version = "1.0.160" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c04e8343c3daeec41f58990b9d77068df31209f2af111e059e9fe9646693065" +checksum = "bb2f3770c8bce3bcda7e149193a069a0f4365bda1fa5cd88e03bca26afc1216c" dependencies = [ "serde_derive", ] @@ -3266,9 +3485,9 @@ dependencies = [ [[package]] name = "serde_derive" -version = "1.0.159" +version = "1.0.160" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c614d17805b093df4b147b51339e7e44bf05ef59fba1e45d83500bcfb4d8585" +checksum = "291a097c63d8497e00160b166a967a4a79c64f3facdd01cbd7502231688d77df" dependencies = [ "proc-macro2", "quote", @@ -3392,6 +3611,12 @@ dependencies = [ "digest 0.10.6", ] +[[package]] +name = "sha1_smol" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae1a47186c03a32177042e55dbc5fd5aee900b8e0069a8d70fba96a9375cd012" + [[package]] name = "sha2" version = "0.9.9" @@ -3618,7 +3843,7 @@ dependencies = [ "thiserror", "tokio-stream", "url", - "uuid 1.3.0", + "uuid 1.3.1", "webpki-roots", "whoami", ] @@ -4429,7 +4654,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13a3aaa69b04e5b66cc27309710a569ea23593612387d67daaf102e73aa974fd" dependencies = [ "rand 0.8.5", - "uuid 1.3.0", + "uuid 1.3.1", ] [[package]] @@ -4617,9 +4842,9 @@ dependencies = [ [[package]] name = "uuid" -version = "1.3.0" +version = "1.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1674845326ee10d37ca60470760d4288a6f80f304007d92e5c53bab78c9cfd79" +checksum = "5b55a3fef2a1e3b3a00ce878640918820d3c51081576ac657d23af9fc7928fdb" dependencies = [ "getrandom 0.2.9", "serde", @@ -4643,6 +4868,16 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d" +[[package]] +name = "value-bag" +version = "1.0.0-alpha.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2209b78d1249f7e6f3293657c9779fe31ced465df091bbd433a1cf88e916ec55" +dependencies = [ + "ctor", + "version_check", +] + [[package]] name = "vcpkg" version = "0.2.15" @@ -4873,6 +5108,7 @@ dependencies = [ "rand 0.8.5", "reqwest", "rsa", + "rsmq_async", "serde", "serde_json", "sha2 0.10.6", @@ -4880,6 +5116,7 @@ dependencies = [ "tokio", "tokio-metrics", "tracing", + "url", "windmill-api", "windmill-api-client", "windmill-common", @@ -4916,6 +5153,7 @@ dependencies = [ "regex", "reqwest", "retainer", + "rsmq_async", "rust-embed", "serde", "serde_json", @@ -4956,7 +5194,7 @@ dependencies = [ "reqwest", "serde", "serde_json", - "uuid 1.3.0", + "uuid 1.3.1", ] [[package]] @@ -4994,7 +5232,7 @@ dependencies = [ "tokio", "tracing", "tracing-subscriber", - "uuid 1.3.0", + "uuid 1.3.1", ] [[package]] @@ -5070,18 +5308,21 @@ dependencies = [ "chrono", "chrono-tz", "cron", + "futures-core", "hex", "hmac", "lazy_static", "prometheus", "reqwest", + "rsmq_async", "serde", "serde_json", "sql-builder", "sqlx", + "tokio", "tracing", "ulid", - "uuid 1.3.0", + "uuid 1.3.1", "windmill-audit", "windmill-common", ] @@ -5105,12 +5346,13 @@ dependencies = [ "prometheus", "rand 0.8.5", "regex", + "rsmq_async", "serde", "serde_json", "sqlx", "tokio", "tracing", - "uuid 1.3.0", + "uuid 1.3.1", "windmill-api-client", "windmill-audit", "windmill-common", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 47f1c7970f..15d0ea66e9 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -48,6 +48,8 @@ git-version.workspace = true rsa.workspace = true base64.workspace = true sha2.workspace = true +rsmq_async.workspace = true +url.workspace = true [dev-dependencies] @@ -141,6 +143,7 @@ sqlx = { version = "^0", features = [ dotenv = "^0" ulid = { version = "^1", features = ["uuid"] } futures = "^0" +futures-core = "^0" tokio-metrics = "0.1.0" lazy_static = "1.4.0" serde_derive = "1.0.147" @@ -153,4 +156,7 @@ async-stripe = { version = "0.14", features = [ ] } async_zip = { version = "0.0.11", features = ["full"] } once_cell = "1.17.1" +rsmq_async = { version = "5.1.5" } gosyn = "0.2.2" + +[patch.crates-io] diff --git a/backend/src/main.rs b/backend/src/main.rs index 31ce62cf50..0675ba0109 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -49,8 +49,40 @@ async fn main() -> anyhow::Result<()> { .and_then(|x| x.parse::().ok()) .unwrap_or(false); + let rsmq_config = std::env::var("REDIS_URL").ok().map(|x| { + let url = x.parse::().unwrap(); + let mut config = rsmq_async::RsmqOptions { ..Default::default() }; + + config.host = url.host_str().expect("redis host required").to_owned(); + config.password = url.password().map(|s| s.to_owned()); + config.db = url + .path_segments() + .and_then(|mut segments| segments.next()) + .and_then(|segment| segment.parse().ok()) + .unwrap_or(0); + config.ns = url + .query_pairs() + .find(|s| s.0 == "rsmq_namespace") + .map(|s| s.1) + .unwrap_or(std::borrow::Cow::Borrowed("rsmq")) + .into_owned(); + config.port = url.port().unwrap_or(6379).to_string(); + + config + }); + let db = windmill_common::connect_db(server_mode).await?; + let rsmq = if let Some(config) = rsmq_config { + let mut rsmq = rsmq_async::MultiplexedRsmq::new(config).await.unwrap(); + + let _ = rsmq_async::RsmqConnection::create_queue(&mut rsmq, "main_queue", None, None, None) + .await; + Some(rsmq) + } else { + None + }; + if server_mode { windmill_api::migrate_db(&db).await?; } @@ -119,9 +151,10 @@ Windmill Community Edition {GIT_VERSION} if server_mode || num_workers > 0 { let addr = SocketAddr::from((server_bind_address, port)); + let rsmq2 = rsmq.clone(); let server_f = async { if server_mode { - windmill_api::run_server(db.clone(), addr, rx.resubscribe()).await?; + windmill_api::run_server(db.clone(), rsmq2, addr, rx.resubscribe()).await?; } Ok(()) as anyhow::Result<()> }; @@ -133,15 +166,17 @@ Windmill Community Edition {GIT_VERSION} rx.resubscribe(), num_workers, base_internal_url.clone(), + rsmq.clone(), ) .await?; } Ok(()) as anyhow::Result<()> }; + let rsmq2 = rsmq.clone(); let monitor_f = async { if server_mode { - monitor_db(&db, rx.resubscribe(), &base_internal_url); + monitor_db(&db, rx.resubscribe(), &base_internal_url, rsmq2); } Ok(()) as anyhow::Result<()> }; @@ -181,10 +216,11 @@ fn display_config(envs: Vec<&str>) { ) } -pub fn monitor_db( +pub fn monitor_db( db: &Pool, rx: tokio::sync::broadcast::Receiver<()>, base_internal_url: &str, + rsmq: Option, ) { let db1 = db.clone(); let db2 = db.clone(); @@ -192,16 +228,17 @@ pub fn monitor_db( let rx2 = rx.resubscribe(); let base_internal_url = base_internal_url.to_string(); tokio::spawn(async move { - windmill_worker::handle_zombie_jobs_periodically(&db1, rx, &base_internal_url).await + windmill_worker::handle_zombie_jobs_periodically(&db1, rx, &base_internal_url, rsmq).await }); tokio::spawn(async move { windmill_api::delete_expired_items_perdiodically(&db2, rx2).await }); } -pub async fn run_workers( +pub async fn run_workers( db: Pool, rx: tokio::sync::broadcast::Receiver<()>, num_workers: i32, base_internal_url: String, + rsmq: Option, ) -> anyhow::Result<()> { let license_key = std::env::var("LICENSE_KEY").ok(); #[cfg(feature = "enterprise")] @@ -231,6 +268,7 @@ pub async fn run_workers( let ip = ip.clone(); let rx = rx.resubscribe(); let base_internal_url = base_internal_url.clone(); + let rsmq2 = rsmq.clone(); handles.push(tokio::spawn(monitor.instrument(async move { tracing::info!(worker = %worker_name, "starting worker"); windmill_worker::run_worker( @@ -241,6 +279,7 @@ pub async fn run_workers( &ip, rx, &base_internal_url, + rsmq2, ) .await }))); diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index eef01730ca..4dfb704330 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -88,7 +88,7 @@ impl ApiServer { let addr = sock.local_addr().unwrap(); drop(sock); - let task = tokio::task::spawn(windmill_api::run_server(db.clone(), addr, rx)); + let task = tokio::task::spawn(windmill_api::run_server(db.clone(), None, addr, rx)); return Self { addr, tx, task }; } @@ -827,9 +827,8 @@ impl RunJob { async fn push(self, db: &Pool) -> Uuid { let RunJob { payload, args } = self; - let tx = db.begin().await.unwrap(); - let (uuid, tx) = windmill_queue::push( - tx, + let (uuid, tx) = windmill_queue::push::( + (None, db.begin().await.unwrap()).into(), "test-workspace", payload, args, @@ -847,8 +846,7 @@ impl RunJob { ) .await .expect("push has to succeed"); - - tx.commit().await.expect("push has to commit"); + tx.commit().await.unwrap(); uuid } @@ -918,7 +916,7 @@ fn spawn_test_worker( let ip: &str = Default::default(); let future = async move { let base_internal_url = format!("http://localhost:{}", port); - windmill_worker::run_worker( + windmill_worker::run_worker::( &db, worker_instance, worker_name, @@ -926,6 +924,7 @@ fn spawn_test_worker( ip, rx, &base_internal_url, + None, ) .await }; diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 885ca0b1f9..1f9b0b286d 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -70,4 +70,5 @@ async-stripe = { workspace = true, optional = true } lazy_static.workspace = true prometheus.workspace = true async_zip.workspace = true +rsmq_async.workspace = true regex.workspace = true diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index b3b61bef2a..8f3c36134c 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -37,7 +37,7 @@ use windmill_common::{ http_get_from_hub, list_elems_from_hub, not_found_if_none, paginate, Pagination, StripPath, }, }; -use windmill_queue::{push, JobPayload, RawCode}; +use windmill_queue::{push, JobPayload, QueueTransaction, RawCode}; pub fn workspaced_service() -> Router { Router::new() @@ -565,6 +565,7 @@ fn digest(code: &str) -> String { async fn execute_component( OptAuthed(opt_authed): OptAuthed, Extension(db): Extension, + Extension(rsmq): Extension>, Path((w_id, path)): Path<(String, StripPath)>, Json(payload): Json, ) -> Result { @@ -583,7 +584,7 @@ async fn execute_component( }; let path = path.to_path(); - let mut tx = db.begin().await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, db.begin().await?).into(); let policy = if let Some(static_fields) = payload.clone().force_viewer_static_fields { let mut hm = HashMap::new(); @@ -652,8 +653,12 @@ async fn execute_component( } ExecuteApp { args, raw_code: None, path: Some(path), .. } => { let payload = if path.starts_with("script/") { - script_path_to_payload(path.strip_prefix("script/").unwrap(), &mut tx, &w_id) - .await? + script_path_to_payload( + path.strip_prefix("script/").unwrap(), + tx.transaction_mut(), + &w_id, + ) + .await? } else if path.starts_with("flow/") { JobPayload::Flow(path.strip_prefix("flow/").unwrap().to_string()) } else { @@ -686,8 +691,8 @@ async fn execute_component( true, ) .await?; - tx.commit().await?; + Ok(uuid.to_string()) } diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index f379142bc9..dff0a32f2e 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -25,7 +25,7 @@ use windmill_common::{ http_get_from_hub, list_elems_from_hub, not_found_if_none, paginate, Pagination, StripPath, }, }; -use windmill_queue::{push, schedule::push_scheduled_job, JobPayload}; +use windmill_queue::{push, schedule::push_scheduled_job, JobPayload, QueueTransaction}; use crate::{ db::{UserDB, DB}, @@ -180,6 +180,7 @@ async fn create_flow( authed: Authed, Extension(db): Extension, Extension(user_db): Extension, + Extension(rsmq): Extension>, Extension(webhook): Extension, Path(w_id): Path, Json(nf): Json, @@ -187,10 +188,10 @@ async fn create_flow( // cron::Schedule::from_str(&ns.schedule).map_err(|e| error::Error::BadRequest(e.to_string()))?; let authed = maybe_refresh_folders(&nf.path, &w_id, authed, &db).await; - let mut tx = user_db.clone().begin(&authed).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into(); - check_path_conflict(&mut tx, &w_id, &nf.path).await?; - check_schedule_conflict(&mut tx, &w_id, &nf.path).await?; + check_path_conflict(tx.transaction_mut(), &w_id, &nf.path).await?; + check_schedule_conflict(tx.transaction_mut(), &w_id, &nf.path).await?; sqlx::query!( "INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, edited_at, \ @@ -222,13 +223,11 @@ async fn create_flow( ) .await?; - tx.commit().await?; webhook.send_message( w_id.clone(), WebhookMessage::CreateFlow { workspace: w_id.clone(), path: nf.path.clone() }, ); - let tx = user_db.begin(&authed).await?; let (dependency_job_uuid, mut tx) = push( tx, &w_id, @@ -247,6 +246,7 @@ async fn create_flow( true, ) .await?; + sqlx::query!( "UPDATE flow SET dependency_job = $1 WHERE path = $2 AND workspace_id = $3", dependency_job_uuid, @@ -286,6 +286,7 @@ async fn check_schedule_conflict<'c>( async fn update_flow( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Extension(db): Extension, Extension(webhook): Extension, Path((w_id, flow_path)): Path<(String, StripPath)>, @@ -294,9 +295,9 @@ async fn update_flow( let flow_path = flow_path.to_path(); let authed = maybe_refresh_folders(&flow_path, &w_id, authed, &db).await; - let mut tx = user_db.clone().begin(&authed).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into(); - check_schedule_conflict(&mut tx, &w_id, flow_path).await?; + check_schedule_conflict(tx.transaction_mut(), &w_id, flow_path).await?; let schema = nf.schema.map(|x| x.0); let old_dep_job = sqlx::query_scalar!( @@ -323,13 +324,13 @@ async fn update_flow( .await?; if nf.path != flow_path { - check_schedule_conflict(&mut tx, &w_id, &nf.path).await?; + check_schedule_conflict(tx.transaction_mut(), &w_id, &nf.path).await?; if !authed.is_admin { require_owner_of_path(&w_id, &authed.username, &authed.groups, &flow_path, &db).await?; } - let mut schedulables = sqlx::query_as!( + let mut schedulables: Vec = sqlx::query_as!( Schedule, "UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND path != $2 AND workspace_id = $3 AND is_flow IS true RETURNING *", nf.path, @@ -352,32 +353,32 @@ async fn update_flow( schedulables.push(schedule); } - for schedule in schedulables { - clear_schedule(&mut tx, flow_path, true).await?; + for schedule in schedulables.into_iter() { + // TODO: Why is this in the loop in the first place? Seems like it's just doing nothing after the first iteration? Should this use schedule.path? + clear_schedule(tx.transaction_mut(), flow_path, true).await?; if schedule.enabled { tx = push_scheduled_job(tx, schedule).await?; } } + + audit_log( + &mut tx, + &authed.username, + "flows.update", + ActionKind::Create, + &w_id, + Some(&nf.path.to_string()), + Some( + [Some(("flow", nf.path.as_str()))] + .into_iter() + .flatten() + .collect(), + ), + ) + .await?; } - audit_log( - &mut tx, - &authed.username, - "flows.update", - ActionKind::Create, - &w_id, - Some(&nf.path.to_string()), - Some( - [Some(("flow", nf.path.as_str()))] - .into_iter() - .flatten() - .collect(), - ), - ) - .await?; - - tx.commit().await?; webhook.send_message( w_id.clone(), WebhookMessage::UpdateFlow { @@ -387,7 +388,6 @@ async fn update_flow( }, ); - let tx = user_db.begin(&authed).await?; let (dependency_job_uuid, mut tx) = push( tx, &w_id, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 9f3df5b21a..91b9c2c8d5 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -30,7 +30,9 @@ use windmill_common::{ users::username_to_permissioned_as, utils::{not_found_if_none, now_from_db, paginate, require_admin, Pagination, StripPath}, }; -use windmill_queue::{get_queued_job, push, JobKind, JobPayload, QueuedJob, RawCode}; +use windmill_queue::{ + get_queued_job, push, JobKind, JobPayload, QueueTransaction, QueuedJob, RawCode, +}; use crate::{ db::{UserDB, DB}, @@ -64,6 +66,7 @@ pub fn workspaced_service() -> Router { .route("/run/preview_flow", post(run_preview_flow_job)) .route("/list", get(list_jobs)) .route("/queue/list", get(list_queue_jobs)) + .route("/queue/count", get(count_queue_jobs)) .route("/completed/list", get(list_completed_jobs)) .route("/completed/get/:id", get(get_completed_job)) .route("/completed/get_result/:id", get(get_completed_job_result)) @@ -114,6 +117,7 @@ async fn get_result_by_id( } async fn cancel_job_api( + Extension(rsmq): Extension>, OptAuthed(opt_authed): OptAuthed, Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, @@ -127,7 +131,7 @@ async fn cancel_job_api( }; let (mut tx, job_option) = - windmill_queue::cancel_job(&username, reason, id, &w_id, tx, false).await?; + windmill_queue::cancel_job(&username, reason, id, &w_id, tx, rsmq, false).await?; if let Some(id) = job_option { audit_log( @@ -156,6 +160,7 @@ async fn cancel_job_api( } async fn force_cancel( + Extension(rsmq): Extension>, OptAuthed(opt_authed): OptAuthed, Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, @@ -169,7 +174,7 @@ async fn force_cancel( }; let (mut tx, job_option) = - windmill_queue::cancel_job(&username, reason, id, &w_id, tx, true).await?; + windmill_queue::cancel_job(&username, reason, id, &w_id, tx, rsmq, true).await?; if let Some(id) = job_option { audit_log( @@ -496,6 +501,26 @@ async fn list_queue_jobs( Ok(Json(jobs)) } +#[derive(Serialize, Debug, FromRow)] +struct QueueStats { + database_length: i64, +} + +async fn count_queue_jobs( + Extension(db): Extension, + Path(w_id): Path, +) -> error::JsonResult { + Ok(Json( + sqlx::query_as!( + QueueStats, + "SELECT coalesce(COUNT(*), 0) as \"database_length!\" FROM queue WHERE workspace_id = $1", + w_id + ) + .fetch_one(&db) + .await?, + )) +} + async fn list_jobs( authed: Authed, Extension(user_db): Extension, @@ -793,6 +818,7 @@ async fn get_suspended_flow_info<'c>( pub async fn cancel_suspended_job( /* unauthed */ Extension(db): Extension, + Extension(rsmq): Extension>, Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>, Query(approver): Query, ) -> error::Result { @@ -815,6 +841,7 @@ pub async fn cancel_suspended_job( parent_flow, &w_id, tx, + rsmq, false, ) .await?; @@ -1208,14 +1235,15 @@ fn decode_payload(t: String) -> anyhow::Result { pub async fn run_flow_by_path( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Path((w_id, flow_path)): Path<(String, StripPath)>, Query(run_query): Query, headers: HeaderMap, Json(args): Json>>, ) -> error::Result<(StatusCode, String)> { let flow_path = flow_path.to_path(); - let mut tx = user_db.begin(&authed).await?; - let scheduled_for = run_query.get_scheduled_for(&mut tx).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); + let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let (uuid, tx) = push( @@ -1243,15 +1271,16 @@ pub async fn run_flow_by_path( pub async fn run_job_by_path( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Path((w_id, script_path)): Path<(String, StripPath)>, Query(run_query): Query, headers: HeaderMap, Json(args): Json>>, ) -> error::Result<(StatusCode, String)> { let script_path = script_path.to_path(); - let mut tx = user_db.begin(&authed).await?; - let job_payload = script_path_to_payload(script_path, &mut tx, &w_id).await?; - let scheduled_for = run_query.get_scheduled_for(&mut tx).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); + let job_payload = script_path_to_payload(script_path, tx.transaction_mut(), &w_id).await?; + let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let (uuid, tx) = push( @@ -1393,6 +1422,7 @@ lazy_static::lazy_static! { pub async fn run_wait_result_job_by_path_get( authed: Authed, + Extension(rsmq): Extension>, Extension(user_db): Extension, Extension(db): Extension, Path((w_id, script_path)): Path<(String, StripPath)>, @@ -1411,8 +1441,8 @@ pub async fn run_wait_result_job_by_path_get( check_queue_too_long(db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?; let script_path = script_path.to_path(); - let mut tx = user_db.clone().begin(&authed).await?; - let job_payload = script_path_to_payload(script_path, &mut tx, &w_id).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into(); + let job_payload = script_path_to_payload(script_path, tx.transaction_mut(), &w_id).await?; let (uuid, tx) = push( tx, @@ -1447,6 +1477,7 @@ pub async fn run_wait_result_job_by_path_get( pub async fn run_wait_result_job_by_path( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Extension(db): Extension, Path((w_id, script_path)): Path<(String, StripPath)>, Query(run_query): Query, @@ -1455,8 +1486,8 @@ pub async fn run_wait_result_job_by_path( ) -> error::JsonResult { check_queue_too_long(db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?; let script_path = script_path.to_path(); - let mut tx = user_db.clone().begin(&authed).await?; - let job_payload = script_path_to_payload(script_path, &mut tx, &w_id).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into(); + let job_payload = script_path_to_payload(script_path, tx.transaction_mut(), &w_id).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); @@ -1493,6 +1524,7 @@ pub async fn run_wait_result_job_by_path( pub async fn run_wait_result_job_by_hash( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Extension(db): Extension, Path((w_id, script_hash)): Path<(String, ScriptHash)>, Query(run_query): Query, @@ -1502,8 +1534,9 @@ pub async fn run_wait_result_job_by_hash( check_queue_too_long(db, run_query.queue_limit).await?; let hash = script_hash.0; - let mut tx = user_db.clone().begin(&authed).await?; - let path = get_path_for_hash(&mut tx, &w_id, hash).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into(); + let path = get_path_for_hash(tx.transaction_mut(), &w_id, hash).await?; + let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let (uuid, tx) = push( @@ -1539,6 +1572,7 @@ pub async fn run_wait_result_job_by_hash( pub async fn run_wait_result_flow_by_path( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Extension(db): Extension, Path((w_id, flow_path)): Path<(String, StripPath)>, Query(run_query): Query, @@ -1548,8 +1582,8 @@ pub async fn run_wait_result_flow_by_path( check_queue_too_long(db, run_query.queue_limit).await?; let flow_path = flow_path.to_path(); - let mut tx = user_db.clone().begin(&authed).await?; - let scheduled_for = run_query.get_scheduled_for(&mut tx).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into(); + let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let (uuid, tx) = push( @@ -1570,7 +1604,6 @@ pub async fn run_wait_result_flow_by_path( !run_query.invisible_to_owner.unwrap_or(false), ) .await?; - tx.commit().await?; run_wait_result( @@ -1601,13 +1634,14 @@ pub async fn script_path_to_payload<'c>( async fn run_preview_job( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Path(w_id): Path, Query(run_query): Query, headers: HeaderMap, Json(preview): Json, ) -> error::Result<(StatusCode, String)> { - let mut tx = user_db.begin(&authed).await?; - let scheduled_for = run_query.get_scheduled_for(&mut tx).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); + let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, preview.args.unwrap_or_default()); let (uuid, tx) = push( @@ -1634,19 +1668,21 @@ async fn run_preview_job( ) .await?; tx.commit().await?; + Ok((StatusCode::CREATED, uuid.to_string())) } async fn run_preview_flow_job( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Path(w_id): Path, Query(run_query): Query, headers: HeaderMap, Json(raw_flow): Json, ) -> error::Result<(StatusCode, String)> { - let mut tx = user_db.begin(&authed).await?; - let scheduled_for = run_query.get_scheduled_for(&mut tx).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); + let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, raw_flow.args.unwrap_or_default()); let (uuid, tx) = push( @@ -1668,21 +1704,23 @@ async fn run_preview_flow_job( ) .await?; tx.commit().await?; + Ok((StatusCode::CREATED, uuid.to_string())) } pub async fn run_job_by_hash( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Path((w_id, script_hash)): Path<(String, ScriptHash)>, Query(run_query): Query, headers: HeaderMap, Json(args): Json>>, ) -> error::Result<(StatusCode, String)> { let hash = script_hash.0; - let mut tx = user_db.begin(&authed).await?; - let path = get_path_for_hash(&mut tx, &w_id, hash).await?; - let scheduled_for = run_query.get_scheduled_for(&mut tx).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); + let path = get_path_for_hash(tx.transaction_mut(), &w_id, hash).await?; + let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); let (uuid, tx) = push( @@ -1704,6 +1742,7 @@ pub async fn run_job_by_hash( ) .await?; tx.commit().await?; + Ok((StatusCode::CREATED, uuid.to_string())) } diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index c17398c257..46f7ba2864 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -81,6 +81,7 @@ lazy_static::lazy_static! { pub async fn run_server( db: DB, + rsmq: Option, addr: SocketAddr, mut rx: tokio::sync::broadcast::Receiver<()>, ) -> anyhow::Result<()> { @@ -100,6 +101,7 @@ pub async fn run_server( .on_request(()), ) .layer(Extension(db.clone())) + .layer(Extension(rsmq)) .layer(Extension(user_db)) .layer(Extension(auth_cache.clone())) .layer(CookieManagerLayer::new()) diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index 861c37a8e8..99d29a93f7 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -44,7 +44,7 @@ use crate::{BASE_URL, HTTP_CLIENT, IS_SECURE, OAUTH_CLIENTS, SLACK_SIGNING_SECRE use windmill_common::error::{self, to_anyhow, Error}; use windmill_common::oauth2::*; -use windmill_queue::JobPayload; +use windmill_queue::{JobPayload, QueueTransaction}; use std::{fs, str}; @@ -729,6 +729,7 @@ where async fn slack_command( SlackSig { sig, ts }: SlackSig, Extension(db): Extension, + Extension(rsmq): Extension>, body: Bytes, ) -> error::Result { let form: SlackCommand = serde_urlencoded::from_bytes(&body) @@ -741,7 +742,7 @@ async fn slack_command( } } - let mut tx = db.begin().await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, db.begin().await?).into(); let settings = sqlx::query_as!( WorkspaceSettings, "SELECT * FROM workspace_settings WHERE slack_team_id = $1", @@ -757,7 +758,7 @@ async fn slack_command( } else { let path = path.strip_prefix("script/").unwrap_or_else(|| path); let script_hash = windmill_common::get_latest_hash_for_path( - &mut tx, + tx.transaction_mut(), &settings.workspace_id, path, ) @@ -789,14 +790,15 @@ async fn slack_command( true, ) .await?; - tx.commit().await?; let url = BASE_URL.to_owned(); + tx.commit().await?; return Ok(format!( "Job launched. See details at {url}/run/{uuid}?workspace={}", &settings.workspace_id )); } } + tx.commit().await?; return Ok(format!( "workspace not properly configured (did you set the script to trigger in the settings?)" diff --git a/backend/windmill-api/src/schedule.rs b/backend/windmill-api/src/schedule.rs index 0fb5830457..9ab73e4272 100644 --- a/backend/windmill-api/src/schedule.rs +++ b/backend/windmill-api/src/schedule.rs @@ -25,7 +25,7 @@ use windmill_common::{ schedule::Schedule, utils::{not_found_if_none, paginate, Pagination, StripPath}, }; -use windmill_queue::{self, schedule::push_scheduled_job, JobKind}; +use windmill_queue::{self, schedule::push_scheduled_job, JobKind, QueueTransaction}; pub fn workspaced_service() -> Router { Router::new() @@ -79,15 +79,23 @@ async fn create_schedule( authed: Authed, Extension(db): Extension, Extension(user_db): Extension, + Extension(rsmq): Extension>, Path(w_id): Path, Json(ns): Json, ) -> Result { let authed = maybe_refresh_folders(&ns.path, &w_id, authed, &db).await; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); - let mut tx = user_db.begin(&authed).await?; cron::Schedule::from_str(&ns.schedule).map_err(|e| Error::BadRequest(e.to_string()))?; - check_path_conflict(&mut tx, &w_id, &ns.path).await?; - check_flow_conflict(&mut tx, &w_id, &ns.path, ns.is_flow, &ns.script_path).await?; + check_path_conflict(tx.transaction_mut(), &w_id, &ns.path).await?; + check_flow_conflict( + tx.transaction_mut(), + &w_id, + &ns.path, + ns.is_flow, + &ns.script_path, + ) + .await?; let schedule = sqlx::query_as!( Schedule, @@ -139,14 +147,15 @@ async fn edit_schedule( authed: Authed, Extension(db): Extension, Extension(user_db): Extension, + Extension(rsmq): Extension>, Path((w_id, path)): Path<(String, StripPath)>, Json(es): Json, ) -> Result { let path = path.to_path(); let authed = maybe_refresh_folders(&path, &w_id, authed, &db).await; - - let mut tx = user_db.begin(&authed).await?; + let mut tx: QueueTransaction<'_, rsmq_async::MultiplexedRsmq> = + (rsmq, user_db.begin(&authed).await?).into(); cron::Schedule::from_str(&es.schedule).map_err(|e| Error::BadRequest(e.to_string()))?; @@ -158,7 +167,7 @@ async fn edit_schedule( .fetch_one(&mut tx) .await?; - clear_schedule(&mut tx, path, is_flow).await?; + clear_schedule(tx.transaction_mut(), path, is_flow).await?; let schedule = sqlx::query_as!( Schedule, "UPDATE schedule SET schedule = $1, timezone = $2, args = $3 WHERE path \ @@ -173,10 +182,6 @@ async fn edit_schedule( .await .map_err(|e| Error::InternalErr(format!("updating schedule in {w_id}: {e}")))?; - if schedule.enabled { - tx = push_scheduled_job(tx, schedule).await?; - } - audit_log( &mut tx, &authed.username, @@ -192,6 +197,10 @@ async fn edit_schedule( ), ) .await?; + + if schedule.enabled { + tx = push_scheduled_job(tx, schedule).await?; + } tx.commit().await?; Ok(path.to_string()) @@ -270,10 +279,12 @@ pub async fn preview_schedule( pub async fn set_enabled( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Path((w_id, path)): Path<(String, StripPath)>, Json(payload): Json, ) -> Result { - let mut tx = user_db.begin(&authed).await?; + let mut tx: QueueTransaction<'_, rsmq_async::MultiplexedRsmq> = + (rsmq, user_db.begin(&authed).await?).into(); let path = path.to_path(); let schedule_o = sqlx::query_as!( Schedule, @@ -288,11 +299,8 @@ pub async fn set_enabled( let schedule = not_found_if_none(schedule_o, "Schedule", path)?; - clear_schedule(&mut tx, path, schedule.is_flow).await?; + clear_schedule(tx.transaction_mut(), path, schedule.is_flow).await?; - if payload.enabled { - tx = push_scheduled_job(tx, schedule).await?; - } audit_log( &mut tx, &authed.username, @@ -303,7 +311,12 @@ pub async fn set_enabled( Some([("enabled", payload.enabled.to_string().as_ref())].into()), ) .await?; + + if payload.enabled { + tx = push_scheduled_job(tx, schedule).await?; + } tx.commit().await?; + Ok(format!( "succesfully updated schedule at path {} to status {}", path, payload.enabled diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index d7328640d7..f2cb1733e2 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -44,7 +44,7 @@ use windmill_common::{ list_elems_from_hub, not_found_if_none, paginate, require_admin, Pagination, StripPath, }, }; -use windmill_queue::{self, schedule::push_scheduled_job}; +use windmill_queue::{self, schedule::push_scheduled_job, QueueTransaction}; const MAX_HASH_HISTORY_LENGTH_STORED: usize = 20; @@ -193,6 +193,7 @@ fn hash_script(ns: &NewScript) -> i64 { async fn create_script( authed: Authed, Extension(user_db): Extension, + Extension(rsmq): Extension>, Extension(webhook): Extension, Extension(db): Extension, Path(w_id): Path, @@ -200,14 +201,14 @@ async fn create_script( ) -> Result<(StatusCode, String)> { let hash = ScriptHash(hash_script(&ns)); let authed = maybe_refresh_folders(&ns.path, &w_id, authed, &db).await; - let mut tx = user_db.begin(&authed).await?; + let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); if sqlx::query_scalar!( "SELECT 1 FROM script WHERE hash = $1 AND workspace_id = $2", hash.0, &w_id ) - .fetch_optional(&mut tx) + .fetch_optional(tx.transaction_mut()) .await? .is_some() { @@ -268,7 +269,7 @@ async fn create_script( ))); }; - let ps = get_script_by_hash_internal(&mut tx, &w_id, p_hash).await?; + let ps = get_script_by_hash_internal(tx.transaction_mut(), &w_id, p_hash).await?; if ps.path != ns.path { if !authed.is_admin { @@ -363,7 +364,7 @@ async fn create_script( .await?; for schedule in schedulables { - clear_schedule(&mut tx, &schedule.path, false).await?; + clear_schedule(tx.transaction_mut(), &schedule.path, false).await?; if schedule.enabled { tx = push_scheduled_job(tx, schedule).await?; @@ -371,36 +372,6 @@ async fn create_script( } } - let mut tx = if needs_lock_gen { - let dependencies = match ns.language { - ScriptLang::Python3 => { - windmill_parser_py::parse_python_imports(&ns.content)?.join("\n") - } - _ => ns.content, - }; - let (_, tx) = windmill_queue::push( - tx, - &w_id, - windmill_queue::JobPayload::Dependencies { hash, dependencies, language: ns.language }, - serde_json::Map::new(), - &authed.username, - &authed.email, - username_to_permissioned_as(&authed.username), - None, - None, - None, - None, - false, - false, - None, - true, - ) - .await?; - tx - } else { - tx - }; - if p_hashes.is_some() && !p_hashes.unwrap().is_empty() { audit_log( &mut tx, @@ -415,7 +386,7 @@ async fn create_script( webhook.send_message( w_id.clone(), WebhookMessage::UpdateScript { - workspace: w_id, + workspace: w_id.clone(), path: ns.path.clone(), hash: hash.to_string(), }, @@ -440,13 +411,41 @@ async fn create_script( webhook.send_message( w_id.clone(), WebhookMessage::CreateScript { - workspace: w_id, + workspace: w_id.clone(), path: ns.path.clone(), hash: hash.to_string(), }, ); } + if needs_lock_gen { + let dependencies = match ns.language { + ScriptLang::Python3 => { + windmill_parser_py::parse_python_imports(&ns.content)?.join("\n") + } + _ => ns.content, + }; + let (_, new_tx) = windmill_queue::push( + tx, + &w_id, + windmill_queue::JobPayload::Dependencies { hash, dependencies, language: ns.language }, + serde_json::Map::new(), + &authed.username, + &authed.email, + username_to_permissioned_as(&authed.username), + None, + None, + None, + None, + false, + false, + None, + true, + ) + .await?; + tx = new_tx; + } + tx.commit().await?; Ok((StatusCode::CREATED, format!("{}", hash))) diff --git a/backend/windmill-audit/src/lib.rs b/backend/windmill-audit/src/lib.rs index 4d39f853fb..9f18ad03ed 100644 --- a/backend/windmill-audit/src/lib.rs +++ b/backend/windmill-audit/src/lib.rs @@ -41,8 +41,8 @@ pub struct AuditLog { } #[tracing::instrument(level = "trace", skip_all)] -pub async fn audit_log<'c>( - db: &mut Transaction<'c, Postgres>, +pub async fn audit_log<'c, E: sqlx::Executor<'c, Database = Postgres>>( + db: E, username: &str, operation: &str, action_kind: ActionKind, diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index bf11ca909f..c1c7db3592 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -52,8 +52,8 @@ pub fn paginate(pagination: Pagination) -> (usize, usize) { } #[cfg(feature = "sqlx")] -pub async fn now_from_db<'c>( - db: &mut sqlx::Transaction<'c, sqlx::Postgres>, +pub async fn now_from_db<'c, E: sqlx::PgExecutor<'c>>( + db: E, ) -> Result> { Ok(sqlx::query_scalar!("SELECT now()") .fetch_one(db) diff --git a/backend/windmill-queue/Cargo.toml b/backend/windmill-queue/Cargo.toml index 092d663735..ca91fff2fb 100644 --- a/backend/windmill-queue/Cargo.toml +++ b/backend/windmill-queue/Cargo.toml @@ -31,3 +31,6 @@ reqwest.workspace = true lazy_static.workspace = true prometheus.workspace = true cron.workspace = true +rsmq_async.workspace = true +tokio.workspace = true +futures-core.workspace = true diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 45aaf59777..17bdb9a5b2 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -25,6 +25,8 @@ use windmill_common::{ METRICS_ENABLED, }; +use crate::QueueTransaction; + lazy_static::lazy_static! { pub static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new() .user_agent("windmill/beta") @@ -52,13 +54,15 @@ lazy_static::lazy_static! { const MAX_FREE_EXECS: i32 = 1000; const MAX_FREE_CONCURRENT_RUNS: i32 = 15; +const RSMQ_MAIN_QUEUE: &'static str = "main_queue"; -pub async fn cancel_job<'c>( +pub async fn cancel_job<'c, R: rsmq_async::RsmqConnection + Clone>( username: &str, reason: Option, id: Uuid, w_id: &str, mut tx: Transaction<'c, Postgres>, + rsmq: Option, force_rerun: bool, ) -> error::Result<(Transaction<'c, Postgres>, Option)> { let job_option = sqlx::query_scalar!( @@ -72,6 +76,12 @@ pub async fn cancel_job<'c>( ) .fetch_optional(&mut tx) .await?; + if let Some(mut rsmq) = rsmq { + rsmq.change_message_visibility(RSMQ_MAIN_QUEUE, &id.to_string(), 0) + .await + .map_err(|e| anyhow::anyhow!(e))?; + } + let mut jobs = job_option.map(|j| vec![j]).unwrap_or_default(); while !jobs.is_empty() { let p_job = jobs.pop(); @@ -91,10 +101,11 @@ pub async fn cancel_job<'c>( Ok((tx, job_option)) } -pub async fn pull( +pub async fn pull( db: &Pool, whitelist_workspaces: Option>, blacklist_workspaces: Option>, + rsmq: Option, ) -> windmill_common::error::Result> { let mut workspaces_filter = String::new(); if let Some(whitelist) = whitelist_workspaces { @@ -106,6 +117,9 @@ pub async fn pull( .collect::>() .join(",") )); + if let Some(_rsmq) = rsmq { + todo!("REDIS: Implement workspace filters for redis"); + } } if let Some(blacklist) = blacklist_workspaces { workspaces_filter.push_str(&format!( @@ -116,16 +130,50 @@ pub async fn pull( .collect::>() .join(",") )); + if let Some(_rsmq) = rsmq { + todo!("REDIS: Implement workspace filters for redis"); + } } - /* Jobs can be started if they: - * - haven't been started before, - * running = false - * - are flows with a step that needed resume, - * suspend_until is non-null - * and suspend = 0 when the resume messages are received - * or suspend_until <= now() if it has timed out */ - let job: Option = sqlx::query_as::<_, QueuedJob>(&format!( - "UPDATE queue + + let job: Option = if let Some(mut rsmq) = rsmq { + // TODO: REDIS: Race conditions / replace last_ping + let msg = rsmq + .pop_message::>(RSMQ_MAIN_QUEUE) + .await + .map_err(|e| anyhow::anyhow!(e))?; + + if let Some(msg) = msg { + let uuid = Uuid::from_bytes_le( + msg.message + .try_into() + .map_err(|_| anyhow::anyhow!("Failed to parsed Redis message"))?, + ); + + sqlx::query_as::<_, QueuedJob>( + "UPDATE queue + SET running = true + , started_at = coalesce(started_at, now()) + , last_ping = now() + , suspend_until = null + WHERE id = $1 + RETURNING *", + ) + .bind(uuid) + .fetch_optional(db) + .await? + } else { + None + } + } else { + /* Jobs can be started if they: + * - haven't been started before, + * running = false + * - are flows with a step that needed resume, + * suspend_until is non-null + * and suspend = 0 when the resume messages are received + * or suspend_until <= now() if it has timed out */ + sqlx::query_as::<_, QueuedJob>(&format!( + "UPDATE queue SET running = true , started_at = coalesce(started_at, now()) , last_ping = now() @@ -143,9 +191,10 @@ pub async fn pull( LIMIT 1 ) RETURNING *" - )) - .fetch_optional(db) - .await?; + )) + .fetch_optional(db) + .await? + }; if job.is_some() && *METRICS_ENABLED { QUEUE_PULL_COUNT.inc(); @@ -207,11 +256,11 @@ pub async fn get_result_by_id( } #[instrument(level = "trace", skip_all)] -pub async fn delete_job( - db: &Pool, +pub async fn delete_job<'c, R: rsmq_async::RsmqConnection + Clone + Send>( + mut tx: QueueTransaction<'c, R>, w_id: &str, job_id: Uuid, -) -> windmill_common::error::Result<()> { +) -> windmill_common::error::Result> { if *METRICS_ENABLED { QUEUE_DELETE_COUNT.inc(); } @@ -220,13 +269,13 @@ pub async fn delete_job( w_id, job_id ) - .fetch_one(db) + .fetch_one(&mut tx) .await .map_err(|e| Error::InternalErr(format!("Error during deletion of job {job_id}: {e}")))? .unwrap_or(0) == 1; tracing::debug!("Job {job_id} deleted: {job_removed}"); - Ok(()) + Ok(tx) } pub async fn get_queued_job<'c>( @@ -245,9 +294,9 @@ pub async fn get_queued_job<'c>( Ok(r) } -#[instrument(level = "trace", skip_all)] -pub async fn push<'c>( - mut tx: Transaction<'c, Postgres>, +// #[instrument(level = "trace", skip_all)] +pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( + mut tx: QueueTransaction<'c, R>, workspace_id: &str, job_payload: JobPayload, args: serde_json::Map, @@ -262,7 +311,7 @@ pub async fn push<'c>( mut same_worker: bool, pre_run_error: Option<&windmill_common::error::Error>, visible_to_owner: bool, -) -> Result<(Uuid, Transaction<'c, Postgres>), Error> { +) -> Result<(Uuid, QueueTransaction<'c, R>), Error> { let args_json = serde_json::Value::Object(args); let job_id: Uuid = Ulid::new().into(); @@ -581,6 +630,10 @@ pub async fn push<'c>( .instrument(tracing::info_span!("job_run", email = &email)) .await?; } + if let Some(ref mut rsmq) = tx.rsmq { + rsmq.send_message(job_id.to_bytes_le().to_vec(), scheduled_for_o); + } + Ok((uuid, tx)) } diff --git a/backend/windmill-queue/src/lib.rs b/backend/windmill-queue/src/lib.rs index b47a3c3238..bb29ee9aac 100644 --- a/backend/windmill-queue/src/lib.rs +++ b/backend/windmill-queue/src/lib.rs @@ -7,6 +7,8 @@ */ mod jobs; +mod queue_transaction; pub mod schedule; pub use jobs::*; +pub use queue_transaction::*; diff --git a/backend/windmill-queue/src/queue_transaction.rs b/backend/windmill-queue/src/queue_transaction.rs new file mode 100644 index 0000000000..edf30784e7 --- /dev/null +++ b/backend/windmill-queue/src/queue_transaction.rs @@ -0,0 +1,161 @@ +use std::fmt::Debug; + +use futures_core::{future::BoxFuture, stream::BoxStream}; +use rsmq_async::{RedisBytes, RsmqConnection}; +use sqlx::{Postgres, Transaction}; + +pub enum RedisOp { + SendMessage(RedisBytes, Option>), + DeleteMessage(String), +} + +impl RedisOp { + pub async fn apply(self, rsmq: &mut R) -> Result<(), rsmq_async::RsmqError> { + match self { + RedisOp::SendMessage(bytes, time) => { + rsmq.send_message( + "main_queue", + bytes, + time.map(|t| (t - chrono::Utc::now()).num_seconds()) + .and_then(|e| e.try_into().ok()), + ) + .await?; + } + RedisOp::DeleteMessage(id) => { + rsmq.delete_message("main_queue", &id).await?; + } + }; + + Ok(()) + } +} + +pub struct RedisTransaction { + rsmq: R, + queued_ops: Vec, +} + +impl From for RedisTransaction { + fn from(value: R) -> Self { + Self { rsmq: value, queued_ops: Vec::new() } + } +} + +impl RedisTransaction { + pub async fn commit(self) -> Result<(), rsmq_async::RsmqError> { + let mut rsmq = self.rsmq; + for op in self.queued_ops { + op.apply(&mut rsmq).await?; + } + Ok(()) + } + + pub fn send_message>( + &mut self, + bytes: E, + delay_until: Option>, + ) { + self.queued_ops + .push(RedisOp::SendMessage(bytes.into(), delay_until)) + } + + pub fn delete_message(&mut self, id: String) { + self.queued_ops.push(RedisOp::DeleteMessage(id)) + } +} + +pub struct QueueTransaction<'c, R: RsmqConnection> { + pub rsmq: Option>, + transaction: Transaction<'c, Postgres>, +} + +impl<'c, R: RsmqConnection> From<(Option, Transaction<'c, Postgres>)> + for QueueTransaction<'c, R> +{ + fn from(value: (Option, Transaction<'c, Postgres>)) -> Self { + Self { rsmq: value.0.map(|e| e.into()), transaction: value.1 } + } +} + +impl<'c, R: RsmqConnection> Debug for QueueTransaction<'c, R> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("QueueTransaction") + .field("rsmq", &self.rsmq.as_ref().map(|_| ())) // do not require R: Debug + .field("transaction", &self.transaction) + .finish() + } +} + +impl<'c, R: RsmqConnection> QueueTransaction<'c, R> { + pub async fn commit(self) -> Result<(), windmill_common::error::Error> { + self.transaction.commit().await?; + if let Some(rsmq) = self.rsmq { + rsmq.commit().await.map_err(|e| anyhow::anyhow!(e))?; + } + + Ok(()) + } + + pub fn transaction_mut<'a>(&'a mut self) -> &'a mut Transaction<'c, Postgres> { + &mut self.transaction + } +} + +impl<'c, 'b, R: RsmqConnection + Send> sqlx::Executor<'b> for &'b mut QueueTransaction<'c, R> { + type Database = Postgres; + + fn fetch_many<'e, 'q: 'e, E: 'q>( + self, + query: E, + ) -> BoxStream< + 'e, + Result< + sqlx::Either< + ::QueryResult, + ::Row, + >, + sqlx::Error, + >, + > + where + 'b: 'e, + E: sqlx::Execute<'q, Self::Database>, + { + self.transaction.fetch_many(query) + } + + fn fetch_optional<'e, 'q: 'e, E: 'q>( + self, + query: E, + ) -> BoxFuture<'e, Result::Row>, sqlx::Error>> + where + 'b: 'e, + E: sqlx::Execute<'q, Self::Database>, + { + self.transaction.fetch_optional(query) + } + + fn prepare_with<'e, 'q: 'e>( + self, + sql: &'q str, + parameters: &'e [::TypeInfo], + ) -> BoxFuture< + 'e, + Result<>::Statement, sqlx::Error>, + > + where + 'b: 'e, + { + self.transaction.prepare_with(sql, parameters) + } + + fn describe<'e, 'q: 'e>( + self, + sql: &'q str, + ) -> BoxFuture<'e, Result, sqlx::Error>> + where + 'b: 'e, + { + self.transaction.describe(sql) + } +} diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 50473c01e8..a59b1b2e86 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -15,11 +15,12 @@ use windmill_common::{ users::username_to_permissioned_as, utils::{now_from_db, StripPath}, }; +use crate::{QueueTransaction}; -pub async fn push_scheduled_job<'c>( - mut tx: Transaction<'c, Postgres>, +pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( + mut tx: QueueTransaction<'c, R>, schedule: Schedule, -) -> Result> { +) -> Result> { let sched = cron::Schedule::from_str(&schedule.schedule) .map_err(|e| error::Error::BadRequest(e.to_string()))?; @@ -70,7 +71,7 @@ pub async fn push_scheduled_job<'c>( } else { JobPayload::ScriptHash { hash: windmill_common::get_latest_hash_for_path( - &mut tx, + tx.transaction_mut(), &schedule.workspace_id, &schedule.script_path, ) @@ -79,7 +80,15 @@ pub async fn push_scheduled_job<'c>( } }; - let (_, mut tx) = push( + sqlx::query!( + "UPDATE schedule SET error = NULL WHERE workspace_id = $1 AND path = $2", + &schedule.workspace_id, + &schedule.path + ) + .execute(&mut tx) + .await?; + + let (_, tx) = push( tx, &schedule.workspace_id, payload, @@ -97,14 +106,7 @@ pub async fn push_scheduled_job<'c>( true, ) .await?; - sqlx::query!( - "UPDATE schedule SET error = NULL WHERE workspace_id = $1 AND path = $2", - &schedule.workspace_id, - &schedule.path - ) - .execute(&mut tx) - .await?; - Ok(tx) + Ok(tx) // TODO: Bubble up pushed UUID from here } pub async fn get_schedule_opt<'c>( diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 948868b275..73cfe7b1df 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -48,4 +48,5 @@ deno_core.workspace = true const_format.workspace = true git-version.workspace = true dyn-iter.workspace = true -once_cell.workspace = true \ No newline at end of file +once_cell.workspace = true +rsmq_async.workspace = true diff --git a/backend/windmill-worker/src/jobs.rs b/backend/windmill-worker/src/jobs.rs index a0d2cb377f..5008620015 100644 --- a/backend/windmill-worker/src/jobs.rs +++ b/backend/windmill-worker/src/jobs.rs @@ -6,29 +6,30 @@ * LICENSE-AGPL for a copy of the license. */ -// use std::time::SystemTime; - -use sqlx::{Pool, Postgres, Transaction}; +use sqlx::{Pool, Postgres}; use tracing::instrument; use uuid::Uuid; use windmill_common::{ error::Error, flow_status::FlowStatusModule, schedule::Schedule, METRICS_ENABLED, }; -use windmill_queue::{delete_job, schedule::get_schedule_opt, JobKind, QueuedJob, CLOUD_HOSTED}; +use windmill_queue::{ + delete_job, schedule::get_schedule_opt, JobKind, QueueTransaction, QueuedJob, CLOUD_HOSTED, +}; #[instrument(level = "trace", skip_all)] -pub async fn add_completed_job_error( +pub async fn add_completed_job_error( db: &Pool, queued_job: &QueuedJob, logs: String, e: serde_json::Value, metrics: Option, + rsmq: Option, ) -> Result { if *METRICS_ENABLED { metrics.map(|m| m.worker_execution_failed.inc()); } let result = serde_json::json!({ "error": e }); - let _ = add_completed_job(db, &queued_job, false, false, result.clone(), logs).await?; + let _ = add_completed_job(db, &queued_job, false, false, result.clone(), logs, rsmq).await?; Ok(result) } @@ -51,13 +52,14 @@ fn flatten_jobs(modules: Vec) -> Vec { } #[instrument(level = "trace", skip_all)] -pub async fn add_completed_job( +pub async fn add_completed_job( db: &Pool, queued_job: &QueuedJob, success: bool, skipped: bool, result: serde_json::Value, logs: String, + rsmq: Option, ) -> Result { let duration = if queued_job.job_kind == JobKind::Flow || queued_job.job_kind == JobKind::FlowPreview { @@ -89,7 +91,7 @@ pub async fn add_completed_job( .ok() .flatten() .flatten(); - let mut tx = db.begin().await?; + let mut tx: QueueTransaction<'_, R> = (rsmq, db.begin().await?).into(); let job_id = queued_job.id.clone(); sqlx::query!( "INSERT INTO completed_job AS cj @@ -159,9 +161,7 @@ pub async fn add_completed_job( .execute(&mut tx) .await .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e}")))?; - // println!("{:?}", SystemTime::now()); - - let _ = delete_job(db, &queued_job.workspace_id, job_id).await?; + tx = delete_job(tx, &queued_job.workspace_id, job_id).await?; if !queued_job.is_flow_step && queued_job.job_kind != JobKind::Flow && queued_job.job_kind != JobKind::FlowPreview @@ -216,14 +216,14 @@ pub async fn add_completed_job( } #[instrument(level = "trace", skip_all)] -pub async fn schedule_again_if_scheduled<'c>( - mut tx: Transaction<'c, Postgres>, +pub async fn schedule_again_if_scheduled<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c>( + mut tx: QueueTransaction<'c, R>, db: &Pool, schedule_path: &str, script_path: &str, w_id: &str, -) -> windmill_common::error::Result> { - let schedule = get_schedule_opt(&mut tx, w_id, schedule_path) +) -> windmill_common::error::Result> { + let schedule = get_schedule_opt(tx.transaction_mut(), w_id, schedule_path) .await? .ok_or_else(|| { Error::InternalErr(format!( diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 7b2df8f1c8..4b8d2e01e1 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -533,8 +533,8 @@ impl AuthedClient { } -#[tracing::instrument(level = "trace")] -pub async fn run_worker( +#[tracing::instrument(skip(rsmq), level = "trace")] +pub async fn run_worker( db: &Pool, worker_instance: &str, worker_name: String, @@ -542,8 +542,8 @@ pub async fn run_worker( ip: &str, mut rx: tokio::sync::broadcast::Receiver<()>, base_internal_url: &str, + rsmq: Option, ) { - #[cfg(not(feature = "enterprise"))] if !*DISABLE_NSJAIL { tracing::warn!( @@ -760,7 +760,8 @@ pub async fn run_worker( }, (job, timer) = { let timer = if *METRICS_ENABLED { Some(worker_pull_duration.start_timer()) } else { None }; - pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone()).map(|x| (x, timer)) } => { + pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone(), rsmq.clone()).map(|x| (x, timer)) + } => { timer.map(|timer| { let duration_pull_s = timer.stop_and_record(); worker_pull_duration_counter.inc_by(duration_pull_s); @@ -769,6 +770,7 @@ pub async fn run_worker( }, } }.instrument(trace_span!("worker_get_next_job")).await; + if do_break { return true; } @@ -848,7 +850,8 @@ pub async fn run_worker( &job_dir, metrics.clone(), same_worker_tx.clone(), - base_internal_url + base_internal_url, + rsmq.clone() ) .await .err() @@ -862,7 +865,8 @@ pub async fn run_worker( false, same_worker_tx.clone(), &worker_dir, - base_internal_url + base_internal_url, + rsmq.clone() ) .await; }; @@ -899,7 +903,7 @@ pub async fn run_worker( } } -async fn handle_job_error( +async fn handle_job_error( db: &Pool, client: &AuthedClient, job: QueuedJob, @@ -909,12 +913,14 @@ async fn handle_job_error( same_worker_tx: Sender, worker_dir: &str, base_internal_url: &str, + rsmq: Option, ) { let err = match err { Error::JsonErr(err) => err, _ => json!({"message": err.to_string(), "name": "InternalErr"}), }; + let rsmq_2 = rsmq.clone(); let update_job_future = || { add_completed_job_error( db, @@ -922,15 +928,16 @@ async fn handle_job_error( format!("Unexpected error during job execution:\n{err}"), err.clone(), metrics.clone(), + rsmq_2, ) }; - if job.is_flow_step || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::Flow { - let (flow, job_status_to_update) = if let Some(parent_job_id) = job.parent_job { + let update_job_future = if job.is_flow_step || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::Flow { + let (flow, job_status_to_update, update_job_future) = if let Some(parent_job_id) = job.parent_job { let _ = update_job_future().await; - (parent_job_id, job.id) + (parent_job_id, job.id, None) } else { - (job.id, Uuid::nil()) + (job.id, Uuid::nil(), Some(update_job_future)) }; let updated_flow = update_flow_status_after_job_completion( @@ -947,7 +954,7 @@ async fn handle_job_error( worker_dir, None, base_internal_url, - + rsmq.clone(), ) .await; @@ -963,15 +970,20 @@ async fn handle_job_error( format!("Unexpected error during flow job error handling:\n{err}"), json!({"message": err.to_string(), "name": "InternalErr"}), metrics.clone(), + rsmq, ) .await; } } } } - } - if job.parent_job.is_none() { - let _ = update_job_future().await; + + update_job_future + } else { + Some(update_job_future) + }; + if let Some(f) = update_job_future { + let _ = f().await; } tracing::error!(job_id = %job.id, "error handling job: {err:#?} {} {} {}", job.id, job.workspace_id, job.created_by); } @@ -998,7 +1010,7 @@ fn extract_error_value(log_lines: &str, i: i32) -> serde_json::Value { return json!({"message": format!("ExitCode: {i}, last log lines: {}", log_lines.to_string().trim().to_string()), "name": "ExecutionErr"}); } #[tracing::instrument(level = "trace", skip_all)] -async fn handle_queued_job( +async fn handle_queued_job( job: QueuedJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -1008,6 +1020,7 @@ async fn handle_queued_job( metrics: Option, same_worker_tx: Sender, base_internal_url: &str, + rsmq: Option, ) -> windmill_common::error::Result<()> { if job.canceled { return Err(Error::JsonErr(canceled_job_to_result(&job)))?; @@ -1026,7 +1039,8 @@ async fn handle_queued_job( args, same_worker_tx, worker_dir, - base_internal_url + base_internal_url, + rsmq, ) .await?; } @@ -1093,7 +1107,7 @@ async fn handle_queued_job( match result { Ok(r) => { // println!("bef completed job{:?}", SystemTime::now()); - add_completed_job(db, &job, true, false, r.clone(), logs).await?; + add_completed_job(db, &job, true, false, r.clone(), logs, rsmq.clone()).await?; if job.is_flow_step { if let Some(parent_job) = job.parent_job { update_flow_status_after_job_completion( @@ -1110,7 +1124,7 @@ async fn handle_queued_job( worker_dir, None, base_internal_url, - + rsmq.clone() ) .await?; } @@ -1145,7 +1159,7 @@ async fn handle_queued_job( }; let result = - add_completed_job_error(db, &job, logs, error_value, metrics.clone()) + add_completed_job_error(db, &job, logs, error_value, metrics.clone(), rsmq.clone()) .await?; if job.is_flow_step { if let Some(parent_job) = job.parent_job { @@ -1163,7 +1177,7 @@ async fn handle_queued_job( worker_dir, None, base_internal_url, - + rsmq ) .await?; } @@ -2969,13 +2983,14 @@ async fn append_logs(job_id: uuid::Uuid, logs: impl AsRef, db: impl Borrow< } } -pub async fn handle_zombie_jobs_periodically( +pub async fn handle_zombie_jobs_periodically( db: &Pool, mut rx: tokio::sync::broadcast::Receiver<()>, base_internal_url: &str, + rsmq: Option, ) { loop { - handle_zombie_jobs(db, base_internal_url).await; + handle_zombie_jobs(db, base_internal_url, rsmq.clone()).await; tokio::select! { _ = tokio::time::sleep(Duration::from_secs(30)) => (), @@ -2987,7 +3002,7 @@ pub async fn handle_zombie_jobs_periodically( } } -async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str) { +async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, rsmq: Option) { if *RESTART_ZOMBIE_JOBS { let restarted = sqlx::query!( "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND job_kind != $3 AND same_worker = false RETURNING id, workspace_id, last_ping", @@ -3065,6 +3080,7 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str) { same_worker_tx_never_used, "", base_internal_url, + rsmq.clone() ) .await; } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 1780ea05ec..837945362a 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -31,11 +31,13 @@ use windmill_common::{ type DB = sqlx::Pool; use windmill_queue::{ - canceled_job_to_result, get_queued_job, push, JobPayload, QueuedJob, RawCode, + canceled_job_to_result, get_queued_job, push, JobPayload, QueueTransaction, QueuedJob, RawCode, }; // #[instrument(level = "trace", skip_all)] -pub async fn update_flow_status_after_job_completion( +pub async fn update_flow_status_after_job_completion< + R: rsmq_async::RsmqConnection + Send + Sync + Clone, +>( db: &DB, client: &AuthedClient, flow: uuid::Uuid, @@ -49,6 +51,7 @@ pub async fn update_flow_status_after_job_completion( worker_dir: &str, stop_early_override: Option, base_internal_url: &str, + rsmq: Option, ) -> error::Result<()> { // this is manual tailrecursion because async_recursion blows up the stack let mut depth = 0; @@ -67,6 +70,7 @@ pub async fn update_flow_status_after_job_completion( stop_early_override, base_internal_url, depth, + rsmq.clone(), ) .await?; while let Some(nrec) = rec { @@ -86,6 +90,7 @@ pub async fn update_flow_status_after_job_completion( nrec.stop_early_override, base_internal_url, depth, + rsmq.clone(), ) .await?; } @@ -99,7 +104,9 @@ pub struct RecUpdateFlowStatusAfterJobCompletion { stop_early_override: Option, } // #[instrument(level = "trace", skip_all)] -pub async fn update_flow_status_after_job_completion_internal( +pub async fn update_flow_status_after_job_completion_internal< + R: rsmq_async::RsmqConnection + Send + Sync + Clone, +>( db: &DB, client: &AuthedClient, flow: uuid::Uuid, @@ -114,11 +121,12 @@ pub async fn update_flow_status_after_job_completion_internal( stop_early_override: Option, base_internal_url: &str, depth: u8, + rsmq: Option, ) -> error::Result> { let (should_continue_flow, flow_job, stop_early, skip_if_stop_early, nresult) = { tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}"); - let mut tx = db.begin().await?; + let mut tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into(); let old_status_json = sqlx::query_scalar!( "SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2", @@ -154,7 +162,7 @@ pub async fn update_flow_status_after_job_completion_internal( module_status, FlowStatusModule::InProgress { iterator: Some(_), .. } ) { - compute_skip_loop_failures(flow, old_status.step, &mut tx) + compute_skip_loop_failures(flow, old_status.step, tx.transaction_mut()) .await? .unwrap_or(false) } else { @@ -197,9 +205,11 @@ pub async fn update_flow_status_after_job_completion_internal( FlowStatusModule::InProgress { branchall: Some(BranchAllStatus { branch, .. }), .. - } => compute_skip_branchall_failure(flow, old_status.step, *branch, &mut tx) - .await? - .unwrap_or(false), + } => { + compute_skip_branchall_failure(flow, old_status.step, *branch, tx.transaction_mut()) + .await? + .unwrap_or(false) + } _ => false, }; @@ -461,7 +471,7 @@ pub async fn update_flow_status_after_job_completion_internal( .context("remove flow status retry")?; } - let flow_job = get_queued_job(flow, w_id, &mut tx) + let flow_job = get_queued_job(flow, w_id, tx.transaction_mut()) .await? .ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?; @@ -503,7 +513,9 @@ pub async fn update_flow_status_after_job_completion_internal( { true } - false if has_failure_module(flow, &mut tx).await? && !is_failure_step => true, + false if has_failure_module(flow, tx.transaction_mut()).await? && !is_failure_step => { + true + } false => false, }; @@ -548,6 +560,7 @@ pub async fn update_flow_status_after_job_completion_internal( logs, canceled_job_to_result(&flow_job), metrics.clone(), + rsmq.clone(), ) .await?; } else { @@ -558,6 +571,7 @@ pub async fn update_flow_status_after_job_completion_internal( stop_early && skip_if_stop_early, nresult.clone(), logs, + rsmq.clone(), ) .await?; } @@ -571,6 +585,7 @@ pub async fn update_flow_status_after_job_completion_internal( same_worker_tx.clone(), worker_dir, base_internal_url, + rsmq.clone(), ) .await { @@ -581,6 +596,7 @@ pub async fn update_flow_status_after_job_completion_internal( "Unexpected error during flow chaining:\n".to_string(), json!({"message": err.to_string(), "name": "InternalError"}), metrics.clone(), + rsmq.clone(), ) .await; true @@ -831,7 +847,7 @@ async fn transform_input( } #[instrument(level = "trace", skip_all)] -pub async fn handle_flow( +pub async fn handle_flow( flow_job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClient, @@ -839,6 +855,7 @@ pub async fn handle_flow( same_worker_tx: Sender, worker_dir: &str, base_internal_url: &str, + rsmq: Option, ) -> anyhow::Result<()> { let value = flow_job .raw_flow @@ -862,6 +879,7 @@ pub async fn handle_flow( same_worker_tx, worker_dir, base_internal_url, + rsmq, ) .await?; Ok(()) @@ -869,7 +887,7 @@ pub async fn handle_flow( #[async_recursion] // #[instrument(level = "trace", skip_all)] -async fn push_next_flow_job( +async fn push_next_flow_job( flow_job: &QueuedJob, mut status: FlowStatus, flow: FlowValue, @@ -879,6 +897,7 @@ async fn push_next_flow_job( same_worker_tx: Sender, worker_dir: &str, base_internal_url: &str, + rsmq: Option, ) -> error::Result<()> { let job_root = flow_job .root_job @@ -915,6 +934,7 @@ async fn push_next_flow_job( worker_dir, None, base_internal_url, + rsmq, ) .await; } @@ -1085,7 +1105,7 @@ async fn push_next_flow_job( let logs = "Timed out waiting to be resumed".to_string(); let result = json!({ "error": {"message": logs, "name": "SuspendedTimeout"}}); let _uuid = - add_completed_job(db, &flow_job, success, skipped, result, logs).await?; + add_completed_job(db, &flow_job, success, skipped, result, logs, rsmq).await?; return Ok(()); } @@ -1286,7 +1306,7 @@ async fn push_next_flow_job( flow.same_worker && module.suspend.is_none() && module.sleep.is_none(); /* Finally, push the job into the queue */ - let mut tx = db.begin().await?; + let mut tx = (rsmq.clone(), db.begin().await?).into(); let mut uuids = vec![]; let len = match &job_payloads { @@ -1452,7 +1472,7 @@ async fn push_next_flow_job( json!(i), flow_job.id ) - .execute(&mut tx) + .execute(db) .await?; } else { sqlx::query!( @@ -1468,7 +1488,7 @@ async fn push_next_flow_job( json!(i), flow_job.id ) - .execute(&mut tx) + .execute(db) .await?; }; diff --git a/benchmarks/example_fake_user.json b/benchmarks/example_fake_user.json deleted file mode 100644 index 25f7ef9c8f..0000000000 --- a/benchmarks/example_fake_user.json +++ /dev/null @@ -1,53 +0,0 @@ -{ - "type": "RANDOM", - "actions": [ - { - "weight": 1, - "action": { - "type": "RANDOM", - "actions": [ - { - "weight": 1, - "action": { - "type": "PREVIEW_SCRIPT", - "workspace": "demo", - "language": "deno", - "args": {}, - "content": "export async function main() { return \"Hello World\"; }" - } - }, - { - "weight": 1, - "action": { - "type": "PREVIEW_SCRIPT", - "workspace": "demo", - "language": "python3", - "args": {}, - "content": "def main(): return \"Hello World\";" - } - }, - { - "weight": 1, - "action": { - "type": "PREVIEW_SCRIPT", - "workspace": "demo", - "language": "go", - "args": {}, - "content": "func main() { return \"Hello World\" }" - } - } - ] - } - }, - { - "weight": 0.5, - "action": { - "type": "PREVIEW_SCRIPT", - "workspace": "demo", - "language": "deno", - "args": {}, - "content": "import { delay } from \"https://deno.land/std@0.131.0/async/delay.ts\"; export async function main() { await delay(1000); return \"Hello World\"; }" - } - } - ] -} diff --git a/benchmarks/identity.json b/benchmarks/identity.json new file mode 100644 index 0000000000..d65f35ca58 --- /dev/null +++ b/benchmarks/identity.json @@ -0,0 +1,7 @@ +{ + "type": "PREVIEW_SCRIPT", + "workspace": "demo", + "language": "deno", + "args": {}, + "content": "export async function main() { return \"Hello World\"; }" +} \ No newline at end of file diff --git a/benchmarks/main.ts b/benchmarks/main.ts index 072520fef1..f4be231647 100644 --- a/benchmarks/main.ts +++ b/benchmarks/main.ts @@ -27,60 +27,60 @@ await new Command() "The number of workers to run at once.", { default: 1, - } + }, ) .option( "-s --seconds ", "How long to run the benchmark for (in seconds).", { default: 30, - } + }, ) .option("--max ", "Maximum number of operations performed.") .option("-e --email ", "The email to use to login.") .option("-p --password ", "The password to use to login.") .env( "WM_TOKEN=", - "The token to use when talking to the API server. Preferred over manual login." + "The token to use when talking to the API server. Preferred over manual login.", ) .option( "-t --token ", - "The token to use when talking to the API server. Preferred over manual login." + "The token to use when talking to the API server. Preferred over manual login.", ) .env( "WM_WORKSPACE=", - "The workspace to spawn scripts from." + "The workspace to spawn scripts from.", ) .option( "-w --workspace ", "The workspace to spawn scripts from.", - { default: "starter" } + { default: "starter" }, ) .option("-m --metrics ", "The url to scrape metrics from.", { default: "http://localhost:8001/metrics", }) .option( "--export-json ", - "If set, exports will be into a JSON file." + "If set, exports will be into a JSON file.", ) .option( "--export-csv ", - "If set, exports will be into a csv file." + "If set, exports will be into a csv file.", ) .option( "--export-histograms [histograms...:string]", - "Mark metrics (without label) that are reported as histograms to export." + "Mark metrics (without label) that are reported as histograms to export.", ) .option( "--export-simple [simple...:string]", - "Mark metrics (without label) that are reported as simple values." + "Mark metrics (without label) that are reported as simple values.", ) .option( "--maximum-throughput ", "Maximum number of jobs/flows to start in one second.", { default: Infinity, - } + }, ) .option("--use-flows", "Run flows instead of jobs.") .option("--custom ", "Use custom actions during bench") @@ -89,11 +89,11 @@ await new Command() "The maximum time in ms to wait for jobs to complete.", { default: 90000, - } + }, ) .option( "--continous", - "Run the benchmark forever. This effectively disables metric collection & exports. No zombie jobs will be tracked." + "Run the benchmark forever. This effectively disables metric collection & exports. No zombie jobs will be tracked.", ) .option( "--histogram-buckets [buckets...:string]", @@ -114,7 +114,7 @@ await new Command() "0.01", "0.005", ], - } + }, ) .action( async ({ @@ -162,7 +162,7 @@ await new Command() new URL("./scraper.ts", import.meta.url).href, { type: "module", - } + }, ); metrics_worker.postMessage({ @@ -192,8 +192,8 @@ await new Command() zombieTimeout, }, null, - 4 - ) + 4, + ), ); const config = { @@ -245,19 +245,22 @@ await new Command() const updateState = setInterval(async () => { const elapsed = start ? Math.ceil((Date.now() - start) / 1000) : 0; const sum = jobsSent.reduce((a, b) => a + b, 0); - const queue_length = ( - await windmill.JobService.listQueue({ - workspace: config.workspace_id, - }) - ).length; + const queue_length = (await (await fetch( + host + "/api/w/" + config.workspace_id + "/jobs/queue/count", + { headers: { ["Authorization"]: "Bearer " + config.token } }, + )).json()).database_length; await Deno.stdout.write( enc( - `elapsed: ${elapsed}/${seconds} | jobs sent: ${JSON.stringify( - jobsSent - )} (sum: ${sum} thr: ${(sum / elapsed).toFixed( - 2 - )}) | queue: ${queue_length} \r` - ) + `elapsed: ${elapsed}/${seconds} | jobs sent: ${ + JSON.stringify( + jobsSent, + ) + } (sum: ${sum} thr: ${ + (sum / elapsed).toFixed( + 2, + ) + }) | queue: ${queue_length} \r`, + ), ); }, 100); @@ -284,7 +287,7 @@ await new Command() const sum = jobsSent.reduce((a, b) => a + b, 0); await Deno.stdout.write( - enc(" ".padStart(30) + `\rduration: ${seconds} | jobs sent: ${sum}\n`) + enc(" ".padStart(30) + `\rduration: ${seconds} | jobs sent: ${sum}\n`), ); const shutdown_start = Date.now(); @@ -302,7 +305,7 @@ await new Command() }; worker.addEventListener("message", l); worker.postMessage( - Number.isSafeInteger(zombieTimeout) ? zombieTimeout : 90000 + Number.isSafeInteger(zombieTimeout) ? zombieTimeout : 90000, ); }); @@ -318,11 +321,10 @@ await new Command() console.log("incorrect results: ", incorrect_results); console.log( "queue length:", - ( - await windmill.JobService.listQueue({ - workspace: config.workspace_id, - }) - ).length + (await (await fetch( + host + "/api/w/" + config.workspace_id + "/jobs/queue/count", + { headers: { ["Authorization"]: "Bearer " + config.token } }, + )).json()).database_length, ); metrics_worker!.postMessage("stop"); @@ -346,7 +348,7 @@ await new Command() const value = values[i]!; const mean = value.reduce((acc, e) => acc + e, 0) / values.length; const stdev = Math.sqrt( - value.reduce((acc, e) => acc + (e - mean) ** 2) / values.length + value.reduce((acc, e) => acc + (e - mean) ** 2) / values.length, ); obj[name] = { mean, stdev }; } @@ -374,6 +376,6 @@ await new Command() f.close(); } console.log("done"); - } + }, ) .parse(); diff --git a/benchmarks/worker.ts b/benchmarks/worker.ts index 7ec2d8c2cc..bb0810beca 100644 --- a/benchmarks/worker.ts +++ b/benchmarks/worker.ts @@ -13,6 +13,8 @@ const promise = new Promise<{ continous: boolean; max_per_worker: number; custom: Action | undefined; + server: string; + token: string; }>((resolve, _reject) => { self.onmessage = (evt) => { const sharedConfig = evt.data; @@ -24,6 +26,8 @@ const promise = new Promise<{ continous: sharedConfig.continous, max_per_worker: sharedConfig.max_per_worker, custom: sharedConfig.custom, + server: sharedConfig.server, + token: sharedConfig.token, }; self.name = "Worker " + sharedConfig.i; resolve(config); @@ -46,19 +50,20 @@ const updateStatusInterval = setInterval(() => { }, 100); while (cont) { - const queue_length = ( - await windmill.JobService.listQueue({ workspace: config.workspace_id }) - ).length; - if (queue_length > 500) { + const queue_length = (await (await fetch( + config.server + "/api/w/" + config.workspace_id + "/jobs/queue/count", + { headers: { ["Authorization"]: "Bearer " + config.token } }, + )).json()).database_length; + if (queue_length > 2500) { console.log( - `queue length: ${queue_length} > 500. waiting... ` + `queue length: ${queue_length} > 2500. waiting... `, ); await sleep(0.5); continue; } if ( (total_spawned * 1000) / (Date.now() - start_time) > - config.per_worker_throughput + config.per_worker_throughput ) { console.log("at maximum throughput. waiting..."); await sleep(0.1); @@ -138,13 +143,13 @@ while (outstanding.length > 0 && Date.now() < end_time) { await Deno.stdout.write( enc( `uuid: ${uuid}, queue length: ${ - ( - await windmill.JobService.listQueue({ - workspace: config.workspace_id, - }) - ).length - } \r` - ) + (await (await fetch( + config.server + "/api/w/" + config.workspace_id + + "/jobs/queue/count", + { headers: { ["Authorization"]: "Bearer " + config.token } }, + )).json()).database_length + } \r`, + ), ); } else if (!config.useFlows) { r = r as api.CompletedJob; @@ -156,7 +161,7 @@ while (outstanding.length > 0 && Date.now() < end_time) { " != " + uuid + "job: \n" + - JSON.stringify(r, null, 2) + JSON.stringify(r, null, 2), ); incorrect_results++; } diff --git a/frontend/CaddyfileDev b/frontend/CaddyfileDev index 516e6cb707..1ae524c0a5 100644 --- a/frontend/CaddyfileDev +++ b/frontend/CaddyfileDev @@ -4,9 +4,9 @@ :8080 { bind {$ADDRESS} - reverse_proxy /api/* http://windmill:8000 + reverse_proxy /api/* http://localhost:8000 reverse_proxy /* http://localhost:3000 reverse_proxy /ws/* http://lsp:3001 { lb_policy header "Authorization" } -} \ No newline at end of file +}