Merge branch 'main' into migration/queue-params

CHANGELOG.md

@@ -1,9 +1,5 @@
 # Changelog
 
-## 1.417.0 (2024-11-03)
-
-*  feat(cli): instance yaml files inside the instance prefix folders (#4627)
-
 ## [1.416.2](https://github.com/windmill-labs/windmill/compare/v1.416.1...v1.416.2) (2024-11-02)
 
 

backend/Cargo.lock

@@ -441,7 +441,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "swc_macros_common",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -588,7 +588,7 @@ checksum = "3b43422f69d8ff38f95f1b2bb76517c91589a924d1559a0e935d7c8ce0274c11"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -636,7 +636,7 @@ checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -678,7 +678,7 @@ checksum = "721cae7de5c34fbb2acd27e21e6d2cf7b886dce0c27388d46c4e6c47ea4318dd"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -1218,7 +1218,7 @@ dependencies = [
  "regex",
  "rustc-hash 1.1.0",
  "shlex",
- "syn 2.0.87",
+ "syn 2.0.86",
  "which 4.4.2",
 ]
 
@@ -1237,7 +1237,7 @@ dependencies = [
  "regex",
  "rustc-hash 1.1.0",
  "shlex",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -1381,7 +1381,7 @@ dependencies = [
  "proc-macro-crate",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
  "syn_derive",
 ]
 
@@ -1484,7 +1484,7 @@ checksum = "bcfcc3cd946cb52f0bbfdbbcfa2f4e24f75ebb6c0e1002f7c25904fada18b9ec"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -1599,9 +1599,9 @@ dependencies = [
 
 [[package]]
 name = "cc"
-version = "1.1.34"
+version = "1.1.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "67b9470d453346108f93a59222a9a1a5724db32d0a4727b7ab7ace4b4d822dc9"
+checksum = "e3788d6ac30243803df38a3e9991cf37e41210232916d41a8222ae378f912624"
 dependencies = [
  "jobserver",
  "libc",
@@ -1744,7 +1744,7 @@ dependencies = [
  "heck 0.5.0",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -2124,7 +2124,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -2172,7 +2172,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "strsim 0.11.1",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -2194,7 +2194,7 @@ checksum = "d336a2a514f6ccccaa3e09b02d41d35330c07ddf03a62165fcec10bb561c7806"
 dependencies = [
  "darling_core 0.20.10",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -2715,7 +2715,7 @@ dependencies = [
  "quote",
  "strum 0.25.0",
  "strum_macros 0.25.3",
- "syn 2.0.87",
+ "syn 2.0.86",
  "thiserror",
 ]
 
@@ -2907,7 +2907,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustc_version 0.4.1",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -3004,7 +3004,7 @@ checksum = "f2b99bf03862d7f545ebc28ddd33a665b50865f4dfd84031a393823879bd4c54"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -3160,7 +3160,7 @@ dependencies = [
  "heck 0.5.0",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -3180,7 +3180,7 @@ checksum = "de0d48a183585823424a4ce1aa132d174a6a81bd540895822eb4c8373a8e49e8"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -3395,7 +3395,7 @@ checksum = "32016f1242eb82af5474752d00fd8ebcd9004bd69b462b1c91de833972d08ed4"
 dependencies = [
  "proc-macro2",
  "swc_macros_common",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -3427,7 +3427,7 @@ checksum = "e99b8b3c28ae0e84b604c75f721c21dc77afb3706076af5e8216d15fd1deaae3"
 dependencies = [
  "frunk_proc_macro_helpers",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -3439,7 +3439,7 @@ dependencies = [
  "frunk_core",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -3451,7 +3451,7 @@ dependencies = [
  "frunk_core",
  "frunk_proc_macro_helpers",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -3575,7 +3575,7 @@ checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -3828,7 +3828,7 @@ checksum = "53010ccb100b96a67bc32c0175f0ed1426b31b655d562898e57325f81c023ac0"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -4463,7 +4463,7 @@ dependencies = [
  "Inflector",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -5113,7 +5113,7 @@ checksum = "a7ce64b975ed4f123575d11afd9491f2e37bbd5813fbfbc0f09ae1fbddea74e0"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -5152,7 +5152,7 @@ dependencies = [
  "proc-macro-error",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
  "termcolor",
  "thiserror",
 ]
@@ -5562,7 +5562,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -5839,7 +5839,7 @@ dependencies = [
  "phf_shared",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -5880,7 +5880,7 @@ checksum = "3c0f5fad0874fc7abcd4d750e76917eaebbecaa2c20bde22e1dbeeba8beb758c"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -6062,7 +6062,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "64d1ec885c64d0457d564db4ec299b2dae3f9c02808b8ad9c3a089c591b18033"
 dependencies = [
  "proc-macro2",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -6115,7 +6115,7 @@ checksum = "07c277e4e643ef00c1233393c673f655e3672cf7eb3ba08a00bdd0ea59139b5f"
 dependencies = [
  "proc-macro-rules-macros",
  "proc-macro2",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -6127,7 +6127,7 @@ dependencies = [
  "once_cell",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -6188,7 +6188,7 @@ dependencies = [
  "schemars",
  "serde",
  "serde_json",
- "syn 2.0.87",
+ "syn 2.0.86",
  "thiserror",
  "typify",
  "unicode-ident",
@@ -6208,7 +6208,7 @@ dependencies = [
  "serde_json",
  "serde_tokenstream",
  "serde_yaml",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -6251,7 +6251,7 @@ dependencies = [
  "itertools 0.12.1",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -6935,7 +6935,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rust-embed-utils",
- "syn 2.0.87",
+ "syn 2.0.86",
  "walkdir",
 ]
 
@@ -7315,7 +7315,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "serde_derive_internals",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -7462,7 +7462,7 @@ checksum = "de523f781f095e28fa605cdce0f8307e451cc0fd14e2eb4cd2e98a355b147766"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -7473,7 +7473,7 @@ checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -7548,7 +7548,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "serde",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -7603,7 +7603,7 @@ dependencies = [
  "darling 0.20.10",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -7972,7 +7972,7 @@ checksum = "01b2e185515564f15375f593fb966b5718bc624ba77fe49fa4616ad619690554"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8043,7 +8043,7 @@ dependencies = [
  "quote",
  "sqlx-core",
  "sqlx-macros-core",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8066,7 +8066,7 @@ dependencies = [
  "sqlx-mysql",
  "sqlx-postgres",
  "sqlx-sqlite",
- "syn 2.0.87",
+ "syn 2.0.86",
  "tempfile",
  "tokio",
  "url",
@@ -8218,7 +8218,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "swc_macros_common",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8272,7 +8272,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustversion",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8285,7 +8285,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustversion",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8393,7 +8393,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "swc_macros_common",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8442,7 +8442,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "swc_macros_common",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8527,7 +8527,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "swc_macros_common",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8634,7 +8634,7 @@ checksum = "63db0adcff29d220c3d151c5b25c0eabe7e32dd936212b84cdaa1392e3130497"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8645,7 +8645,7 @@ checksum = "f486687bfb7b5c560868f69ed2d458b880cebc9babebcb67e49f31b55c5bf847"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8668,7 +8668,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "swc_macros_common",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8684,9 +8684,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "2.0.87"
+version = "2.0.86"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "25aa4ce346d03a6dcd68dd8b4010bcb74e54e62c90c573f394c46eae99aba32d"
+checksum = "e89275301d38033efb81a6e60e3497e734dfcc62571f2854bf4b16690398824c"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -8702,7 +8702,7 @@ dependencies = [
  "proc-macro-error",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8728,7 +8728,7 @@ checksum = "c8af7666ab7b6390ab78131fb5b0fce11d6b7a6951602017c35fa82800708971"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -8993,7 +8993,7 @@ checksum = "b08be0f17bd307950653ce45db00cd31200d82b624b36e181337d9c7d92765b5"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -9218,7 +9218,7 @@ checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -9545,7 +9545,7 @@ checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -9782,7 +9782,7 @@ dependencies = [
  "regress",
  "schemars",
  "serde_json",
- "syn 2.0.87",
+ "syn 2.0.86",
  "thiserror",
  "unicode-ident",
 ]
@@ -9799,7 +9799,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_tokenstream",
- "syn 2.0.87",
+ "syn 2.0.86",
  "typify-impl",
 ]
 
@@ -10194,7 +10194,7 @@ dependencies = [
  "once_cell",
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
  "wasm-bindgen-shared",
 ]
 
@@ -10228,7 +10228,7 @@ checksum = "e94f17b526d0a461a191c78ea52bbce64071ed5c04c9ffe424dcb38f74171bb7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
  "wasm-bindgen-backend",
  "wasm-bindgen-shared",
 ]
@@ -10261,7 +10261,7 @@ checksum = "b7f89739351a2e03cb94beb799d47fb2cac01759b40ec441f7de39b00cbf7ef0"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -10395,11 +10395,11 @@ checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
 
 [[package]]
 name = "windmill"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "axum",
- "base64 0.22.1",
+ "base64 0.21.7",
  "chrono",
  "deno_core",
  "dotenv",
@@ -10437,7 +10437,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-api"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "argon2",
@@ -10448,7 +10448,7 @@ dependencies = [
  "async_zip",
  "axum",
  "base32",
- "base64 0.22.1",
+ "base64 0.21.7",
  "bytes",
  "candle-core",
  "candle-nn",
@@ -10522,9 +10522,9 @@ dependencies = [
 
 [[package]]
 name = "windmill-api-client"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
- "base64 0.22.1",
+ "base64 0.21.7",
  "chrono",
  "openapiv3",
  "prettyplease 0.1.25",
@@ -10540,7 +10540,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-audit"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "chrono",
  "serde",
@@ -10553,7 +10553,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-autoscaling"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "rsmq_async",
@@ -10568,7 +10568,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-common"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "async-stream",
@@ -10614,7 +10614,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-git-sync"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "regex",
  "rsmq_async",
@@ -10629,7 +10629,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-indexer"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "bytes",
@@ -10651,19 +10651,19 @@ dependencies = [
 
 [[package]]
 name = "windmill-macros"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "itertools 0.13.0",
  "lazy_static",
  "proc-macro2",
  "quote",
  "regex",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
 name = "windmill-parser"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "convert_case 0.6.0",
  "serde",
@@ -10672,7 +10672,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-bash"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -10684,7 +10684,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-go"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "gosyn",
@@ -10696,7 +10696,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-graphql"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -10708,7 +10708,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-php"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "itertools 0.13.0",
@@ -10719,7 +10719,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-py"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "itertools 0.13.0",
@@ -10730,7 +10730,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-py-imports"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "async-recursion",
@@ -10748,7 +10748,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-rust"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "convert_case 0.6.0",
@@ -10758,14 +10758,14 @@ dependencies = [
  "quote",
  "regex",
  "serde_json",
- "syn 2.0.87",
+ "syn 2.0.86",
  "toml 0.7.8",
  "windmill-parser",
 ]
 
 [[package]]
 name = "windmill-parser-sql"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -10777,7 +10777,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-ts"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -10795,7 +10795,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-wasm"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "getrandom 0.2.15",
@@ -10816,7 +10816,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-yaml"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "serde_json",
@@ -10826,7 +10826,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-queue"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "async-recursion",
@@ -10834,6 +10834,7 @@ dependencies = [
  "bigdecimal",
  "chrono",
  "chrono-tz 0.10.0",
+ "const_format",
  "cron",
  "futures-core",
  "hex",
@@ -10859,7 +10860,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-sql-datatype-parser-wasm"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "wasm-bindgen",
  "wasm-bindgen-test",
@@ -10869,11 +10870,11 @@ dependencies = [
 
 [[package]]
 name = "windmill-worker"
-version = "1.417.0"
+version = "1.416.2"
 dependencies = [
  "anyhow",
  "async-recursion",
- "base64 0.22.1",
+ "base64 0.21.7",
  "bit-vec",
  "bytes",
  "chrono",
@@ -11235,7 +11236,7 @@ checksum = "28cc31741b18cb6f1d5ff12f5b7523e3d6eb0852bbbad19d73905511d9849b95"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
  "synstructure",
 ]
 
@@ -11257,7 +11258,7 @@ checksum = "fa4f8080344d4671fb4e831a13ad1e68092748387dfc4f55e356242fae12ce3e"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
 ]
 
 [[package]]
@@ -11277,7 +11278,7 @@ checksum = "0ea7b4a3637ea8669cedf0f1fd5c286a17f3de97b8dd5a70a6c167a1730e63a5"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.87",
+ "syn 2.0.86",
  "synstructure",
 ]
 

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "windmill"
-version = "1.417.0"
+version = "1.416.2"
 authors.workspace = true
 edition.workspace = true
 
@@ -29,7 +29,7 @@ members = [
 ]
 
 [workspace.package]
-version = "1.417.0"
+version = "1.416.2"
 authors = ["Ruben Fiszel <ruben@windmill.dev>"]
 edition = "2021"
 

backend/migrations/20241004112544_create_params_args_result_tables.down.sql

@@ -0,0 +1,3 @@
+DROP TABLE job_params;
+DROP TABLE job_args;
+DROP TABLE completed_jobs_result;

backend/migrations/20241004112544_create_params_args_result_tables.up.sql

@@ -0,0 +1,25 @@
+-- Add up migration script here
+-- Add down migration script here
+-- Add up migration script here
+CREATE TABLE job_params (
+  id UUID PRIMARY KEY,
+  raw_code TEXT,
+  raw_flow jsonb NULL,
+  tag VARCHAR(50),
+  workspace_id VARCHAR(50)
+);
+
+-- Add up migration script here
+CREATE TABLE job_args (
+  id UUID PRIMARY KEY,
+  args JSONB,
+  tag VARCHAR(50),
+  workspace_id VARCHAR(50)
+);
+
+CREATE TABLE completed_jobs_result (
+  id UUID PRIMARY KEY,
+  result JSONB,
+  tag VARCHAR(50),
+  workspace_id VARCHAR(50)
+);

backend/src/monitor.rs

@@ -19,6 +19,7 @@ use tokio::{
     sync::{mpsc, RwLock},
 };
 
+use uuid::Uuid;
 #[cfg(feature = "embedding")]
 use windmill_api::embeddings::update_embeddings_db;
 use windmill_api::{
@@ -32,7 +33,7 @@ use windmill_common::{
     auth::JWT_SECRET,
     ee::CriticalErrorChannel,
     error,
-    flow_status::FlowStatusModule,
+    flow_status::{FlowStatusModule, ParsedFlowStatusGetter as _},
     global_settings::{
         BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
         DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING,
@@ -1420,7 +1421,7 @@ async fn handle_zombie_flows(
                 }
             );
             report_critical_error(reason.clone(), db.clone()).await;
-            cancel_zombie_flow_job(db, flow, &rsmq, reason).await?;
+            cancel_zombie_flow_job(db, &flow.id, &flow.workspace_id, &rsmq, reason).await?;
         }
     }
 
@@ -1436,11 +1437,17 @@ async fn handle_zombie_flows(
     .fetch_all(db)
     .await?;
 
+    #[derive(sqlx::FromRow, Debug)]
+    struct InQueueJobResult {
+        id: uuid::Uuid,
+        workspace_id: String,
+    }
+
     for flow in flows2 {
-        let in_queue = sqlx::query_as::<_, QueuedJob>(
-            "SELECT * FROM queue WHERE id = $1 AND running = true AND canceled = false",
+        let in_queue = sqlx::query_as!(InQueueJobResult,
+            "SELECT id, workspace_id FROM queue WHERE id = $1 AND running = true AND canceled = false",
+            flow.parent_flow_id
         )
-        .bind(flow.parent_flow_id)
         .fetch_optional(db)
         .await?;
         if let Some(job) = in_queue {
@@ -1450,7 +1457,7 @@ async fn handle_zombie_flows(
                 job.workspace_id,
                 flow.last_ping
             );
-            cancel_zombie_flow_job(db, job, &rsmq,
+            cancel_zombie_flow_job(db, &job.id, &job.workspace_id, &rsmq,
                 format!("Flow {} cancelled as one of the parallel branch {} was unable to make the last transition ", flow.parent_flow_id, flow.job_id))
                 .await?;
         } else {
@@ -1462,21 +1469,22 @@ async fn handle_zombie_flows(
 
 async fn cancel_zombie_flow_job(
     db: &Pool<Postgres>,
-    flow: QueuedJob,
+    job_id: &Uuid,
+    workspace_id: &str,
     rsmq: &Option<MultiplexedRsmq>,
     message: String,
 ) -> Result<(), error::Error> {
     let tx = db.begin().await.unwrap();
     tracing::error!(
         "zombie flow detected: {} in workspace {}. Cancelling it.",
-        flow.id,
-        flow.workspace_id
+        job_id,
+        workspace_id
     );
     let (ntx, _) = cancel_job(
         "monitor",
         Some(message),
-        flow.id,
-        flow.workspace_id.as_str(),
+        *job_id,
+        workspace_id,
         tx,
         db,
         rsmq.clone(),

backend/windmill-api/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: "3.0.3"
 
 info:
-  version: 1.417.0
+  version: 1.416.2
   title: Windmill API
 
   contact:

backend/windmill-api/src/inputs.rs

@@ -21,11 +21,7 @@ use std::{
     vec,
 };
 use windmill_common::{
-    db::UserDB,
-    error::JsonResult,
-    jobs::JobKind,
-    scripts::to_i64,
-    utils::{not_found_if_none, paginate, Pagination},
+    db::UserDB, error::JsonResult, jobs::JobKind, query_scalar_with_fallback, scripts::to_i64, utils::{not_found_if_none, paginate, Pagination}
 };
 pub fn workspaced_service() -> Router {
     Router::new()
@@ -178,6 +174,7 @@ struct GetArgs {
     input: Option<bool>,
     allow_large: Option<bool>,
 }
+
 async fn get_args_from_history_or_saved_input(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
@@ -185,6 +182,7 @@ async fn get_args_from_history_or_saved_input(
     Path((w_id, job_or_input_id)): Path<(String, Uuid)>,
 ) -> JsonResult<Option<Value>> {
     let mut tx = user_db.begin(&authed).await?;
+
     let result_o = if let Some(input) = g.input {
         if input {
             sqlx::query_scalar!(
@@ -196,24 +194,20 @@ async fn get_args_from_history_or_saved_input(
                 .fetch_optional(&mut *tx)
                 .await?
         } else {
-            sqlx::query_scalar!(
+            query_scalar_with_fallback!(tx, 
+                "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM job_args WHERE id = $1 AND workspace_id = $2", 
                 "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM completed_job WHERE id = $1 AND workspace_id = $2",
-                job_or_input_id,
+                job_or_input_id, 
                 w_id,
-                g.allow_large.unwrap_or(true)
-            )
-            .fetch_optional(&mut *tx)
-            .await?
+                g.allow_large.unwrap_or(true))?
         }
     } else {
-        sqlx::query_scalar!(
-        "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM input WHERE id = $1 AND workspace_id = $2",
-        job_or_input_id,
-        w_id,
-        g.allow_large.unwrap_or(true)
-    )
-    .fetch_optional(&mut *tx)
-    .await?
+        query_scalar_with_fallback!(tx, 
+                 "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM job_args WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM input WHERE id = $1 AND workspace_id = $2", 
+                "SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT CASE WHEN pg_column_size(args) < 40000 OR $3 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args FROM input WHERE id = $1 AND workspace_id = $2",
+                job_or_input_id,
+                w_id,
+                g.allow_large.unwrap_or(true))?
     };
 
     tx.commit().await?;

backend/windmill-api/src/jobs.rs

@@ -314,9 +314,14 @@ async fn get_result_by_id(
     Path((w_id, flow_id, node_id)): Path<(String, Uuid, String)>,
     Query(JsonPath { json_path, .. }): Query<JsonPath>,
 ) -> windmill_common::error::JsonResult<Box<JsonRawValue>> {
-    let res =
-        windmill_queue::get_result_by_id(db.clone(), w_id.clone(), flow_id, node_id, json_path)
-            .await?;
+    let res = windmill_queue::get_result_by_id(
+        db.clone(),
+        &w_id,
+        flow_id,
+        &node_id,
+        json_path.as_deref(),
+    )
+    .await?;
 
     log_job_view(&db, Some(&authed), &w_id, &flow_id).await?;
 
@@ -1349,6 +1354,8 @@ async fn cancel_jobs(
 ) -> error::JsonResult<Vec<Uuid>> {
     let mut uuids = vec![];
     let mut tx = db.begin().await?;
+
+    let result = serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}});
     let trivial_jobs =  sqlx::query!("INSERT INTO completed_job AS cj
                    ( workspace_id
                    , id
@@ -1412,10 +1419,19 @@ async fn cancel_jobs(
                    , tag
                    , priority FROM queue 
         WHERE id = any($2) AND running = false AND parent_job IS NULL AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED
-        ON CONFLICT (id) DO NOTHING RETURNING id", username, &jobs, w_id, serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}}))
+        ON CONFLICT (id) DO NOTHING RETURNING id", username, &jobs, w_id, &result)
         .fetch_all(&mut *tx)
         .await?.into_iter().map(|x| x.id).collect::<Vec<Uuid>>();
 
+    sqlx::query!(
+        "INSERT INTO completed_jobs_result(id, result, tag, workspace_id) SELECT id, $1, tag, $2 FROM completed_job  WHERE id = any($3) ON CONFLICT (id) DO NOTHING",
+        result,
+        w_id,
+        &trivial_jobs,
+    )
+    .execute(&mut *tx)
+    .await?;
+
     sqlx::query!(
         "DELETE FROM queue WHERE id = any($1) AND workspace_id = $2",
         &trivial_jobs,
@@ -1423,6 +1439,7 @@ async fn cancel_jobs(
     )
     .execute(&mut *tx)
     .await?;
+
     tx.commit().await?;
 
     // sqlx::query!(
@@ -1436,18 +1453,10 @@ async fn cancel_jobs(
             continue;
         }
         let rsmq = rsmq.clone();
-        match tokio::time::timeout(tokio::time::Duration::from_secs(5), async move {
+        if let Ok(result) = tokio::time::timeout(tokio::time::Duration::from_secs(5), async move {
             let tx = db.begin().await?;
             let (tx, _) = windmill_queue::cancel_job(
-                username,
-                None,
-                job_id.clone(),
-                w_id,
-                tx,
-                db,
-                rsmq,
-                false,
-                false,
+                username, None, job_id, w_id, tx, db, rsmq, false, false,
             )
             .await?;
             tx.commit().await?;
@@ -1455,20 +1464,19 @@ async fn cancel_jobs(
         })
         .await
         {
-            Ok(result) => match result {
+            match result {
                 Ok(_) => {
                     uuids.push(job_id);
                 }
                 Err(e) => {
                     tracing::error!("Failed to cancel job {:?}: {:?}", job_id, e);
                 }
-            },
-            Err(_) => {
-                tracing::error!(
-                    "Timeout while trying to cancel job {:?} after 5 seconds",
-                    job_id
-                );
             }
+        } else {
+            tracing::error!(
+                "Timeout while trying to cancel job {:?} after 5 seconds",
+                job_id
+            );
         }
     }
 
@@ -3283,7 +3291,7 @@ async fn run_wait_result(
     };
 
     let fast_poll_duration = *WAIT_RESULT_FAST_POLL_DURATION_SECS as u64 * 1000;
-    let mut accumulated_delay = 0 as u64;
+    let mut accumulated_delay = 0_u64;
 
     loop {
         if let Some(node_id_for_empty_return) = node_id_for_empty_return.as_ref() {
@@ -3400,6 +3408,16 @@ async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Er
     )
     .execute(db)
     .await?;
+
+    sqlx::query!(
+        "UPDATE job_args
+        SET args = '{}'::jsonb
+        WHERE id = $1",
+        job_uuid,
+    )
+    .execute(db)
+    .await?;
+
     sqlx::query!(
         "UPDATE job_logs
         SET logs = '##DELETED##'
@@ -4384,11 +4402,8 @@ async fn add_batch_jobs(
                 tx = PushIsolationLevel::Transaction(ntx);
                 uuids.push(uuid);
             }
-            match tx {
-                PushIsolationLevel::Transaction(tx) => {
-                    tx.commit().await?;
-                }
-                _ => (),
+            if let PushIsolationLevel::Transaction(tx) = tx {
+                tx.commit().await?;
             }
             return Ok(Json(uuids));
         }

backend/windmill-api/src/resources.rs

@@ -31,7 +31,6 @@ use windmill_audit::ActionKind;
 use windmill_common::{
     db::UserDB,
     error::{Error, JsonResult, Result},
-    jobs::QueuedJob,
     utils::{not_found_if_none, paginate, require_admin, Pagination, StripPath},
     variables,
 };
@@ -536,11 +535,29 @@ pub async fn transform_json_value<'c>(
         }
         Value::String(y) if y.starts_with("$") && job_id.is_some() => {
             let mut tx = authed_transaction_or_default(authed, user_db.clone(), db).await?;
-            let job = sqlx::query_as::<_, QueuedJob>(
-                "SELECT * FROM queue WHERE id = $1 AND workspace_id = $2",
+
+            #[derive(sqlx::FromRow, Debug)]
+            struct QueuedJobLite {
+                pub id: Uuid,
+                pub workspace_id: String,
+                pub parent_job: Option<Uuid>,
+                pub created_by: String,
+                pub email: String,
+                pub permissioned_as: String,
+                pub script_path: Option<String>,
+                pub schedule_path: Option<String>,
+                pub root_job: Option<Uuid>,
+                pub flow_step_id: Option<String>,
+                pub scheduled_for: chrono::DateTime<chrono::Utc>,
+            }
+
+            let job = sqlx::query_as!(
+                QueuedJobLite,
+                "SELECT id, workspace_id,parent_job, created_by, email, permissioned_as, script_path, schedule_path, root_job, flow_step_id, scheduled_for
+                 FROM queue WHERE id = $1 AND workspace_id = $2",
+                job_id.unwrap(),
+                workspace
             )
-            .bind(job_id.unwrap())
-            .bind(workspace)
             .fetch_optional(&mut *tx)
             .await?;
             tx.commit().await?;

backend/windmill-common/src/flow_status.rs

@@ -45,6 +45,33 @@ pub struct FlowStatus {
     pub restarted_from: Option<RestartedFrom>,
 }
 
+pub trait FlowStatusGetter {
+    fn get_raw_flow_status(&self) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>>;
+}
+
+#[macro_export]
+macro_rules! impl_flow_status_getter {
+    ($struct_name:ident) => {
+        impl FlowStatusGetter for $struct_name {
+            fn get_raw_flow_status(
+                &self,
+            ) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>> {
+                self.flow_status.as_ref()
+            }
+        }
+    };
+}
+
+pub trait ParsedFlowStatusGetter {
+    fn parse_flow_status(&self) -> Option<FlowStatus>;
+}
+impl<I: FlowStatusGetter> ParsedFlowStatusGetter for I {
+    fn parse_flow_status(&self) -> Option<FlowStatus> {
+        self.get_raw_flow_status()
+            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
 #[serde(default)]
 pub struct RetryStatus {

backend/windmill-common/src/flows.rs

@@ -125,6 +125,34 @@ pub struct FlowValue {
     pub concurrency_key: Option<String>,
 }
 
+pub trait FlowValueGetter {
+    fn get_raw_flow_value(&self) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>>;
+}
+
+#[macro_export]
+macro_rules! impl_flow_value_getter {
+    ($struct_name:ident) => {
+        impl FlowValueGetter for $struct_name {
+            fn get_raw_flow_value(
+                &self,
+            ) -> Option<&sqlx::types::Json<Box<serde_json::value::RawValue>>> {
+                self.raw_flow.as_ref()
+            }
+        }
+    };
+}
+
+pub trait ParsedFlowValueGetter {
+    fn parse_raw_flow(&self) -> Option<FlowValue>;
+}
+
+impl<I: FlowValueGetter> ParsedFlowValueGetter for I {
+    fn parse_raw_flow(&self) -> Option<FlowValue> {
+        self.get_raw_flow_value()
+            .and_then(|v| serde_json::from_str::<FlowValue>((**v).get()).ok())
+    }
+}
+
 #[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct StopAfterIf {
     pub expr: String,

backend/windmill-common/src/jobs.rs

@@ -15,9 +15,9 @@ pub const PREPROCESSOR_FAKE_ENTRYPOINT: &str = "__WM_PREPROCESSOR";
 
 use crate::{
     error::{self, to_anyhow, Error},
-    flow_status::{FlowStatus, RestartedFrom},
-    flows::{FlowValue, Retry},
-    get_latest_deployed_hash_for_path,
+    flow_status::{FlowStatusGetter, RestartedFrom},
+    flows::{FlowValue, FlowValueGetter, Retry},
+    get_latest_deployed_hash_for_path, impl_flow_status_getter, impl_flow_value_getter,
     scripts::{ScriptHash, ScriptLang},
     worker::{to_raw_value, TMP_DIR},
 };
@@ -119,10 +119,7 @@ pub struct QueuedJob {
 
 impl QueuedJob {
     pub fn script_path(&self) -> &str {
-        self.script_path
-            .as_ref()
-            .map(String::as_str)
-            .unwrap_or("tmp/main")
+        self.script_path.as_deref().unwrap_or("tmp/main")
     }
     pub fn is_flow(&self) -> bool {
         matches!(
@@ -139,22 +136,11 @@ impl QueuedJob {
             self.script_path()
         )
     }
-
-    pub fn parse_raw_flow(&self) -> Option<FlowValue> {
-        self.raw_flow.as_ref().and_then(|v| {
-            let str = (**v).get();
-            // tracing::error!("raw_flow: {}", str);
-            return serde_json::from_str::<FlowValue>(str).ok();
-        })
-    }
-
-    pub fn parse_flow_status(&self) -> Option<FlowStatus> {
-        self.flow_status
-            .as_ref()
-            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
-    }
 }
 
+impl_flow_status_getter!(QueuedJob);
+impl_flow_value_getter!(QueuedJob);
+
 impl Default for QueuedJob {
     fn default() -> Self {
         Self {
@@ -266,23 +252,13 @@ impl CompletedJob {
     pub fn json_result(&self) -> Option<serde_json::Value> {
         self.result
             .as_ref()
-            .map(|r| serde_json::from_str(r.get()).ok())
-            .flatten()
-    }
-
-    pub fn parse_raw_flow(&self) -> Option<FlowValue> {
-        self.raw_flow
-            .as_ref()
-            .and_then(|v| serde_json::from_str::<FlowValue>((**v).get()).ok())
-    }
-
-    pub fn parse_flow_status(&self) -> Option<FlowStatus> {
-        self.flow_status
-            .as_ref()
-            .and_then(|v| serde_json::from_str::<FlowStatus>((**v).get()).ok())
+            .and_then(|r| serde_json::from_str(r.get()).ok())
     }
 }
 
+impl_flow_status_getter!(CompletedJob);
+impl_flow_value_getter!(CompletedJob);
+
 #[derive(sqlx::FromRow)]
 pub struct BranchResults {
     pub result: sqlx::types::Json<Box<RawValue>>,

backend/windmill-common/src/lib.rs

@@ -6,6 +6,7 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
+pub mod macros;
 use std::{
     net::SocketAddr,
     sync::{atomic::AtomicBool, Arc},

backend/windmill-common/src/macros.rs

@@ -0,0 +1,57 @@
+#[macro_export]
+macro_rules! fetch_one_with_fallback {
+    ($db:expr, $query_method:ident, $row_type:ty, $query:literal, $table:literal || $fallback_table:literal, $( $param:expr ),* ) => {{
+        let primary_query = sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $table))
+            $(.bind($param))*
+            .fetch_one($db)
+            .await;
+
+        if let Err(sqlx::Error::RowNotFound) = primary_query {
+            tracing::info!("Data not found in job_params, falling back to fetching from $fallback_table");
+            sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $fallback_table))
+                $(.bind($param))*
+                .fetch_one($db)
+                .await
+        } else {
+            primary_query
+        }
+    }};
+}
+
+#[macro_export]
+macro_rules! fetch_optional_with_fallback {
+    ($db:expr, $query_method:ident, $row_type:ty, $query:literal, $table:literal || $fallback_table:literal, $( $param:expr ),* ) => {{
+        let primary_query = sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $table))
+            $(.bind($param))*
+            .fetch_optional($db)
+            .await;
+
+        if let Ok(None) = primary_query {
+            tracing::info!("Data not found in job_params, falling back to fetching from $fallback_table");
+            sqlx::$query_method::<_, $row_type>(const_format::formatcp!($query, $fallback_table))
+                $(.bind($param))*
+                .fetch_optional($db)
+                .await
+        } else {
+            primary_query
+        }
+    }};
+}
+
+#[macro_export]
+macro_rules! query_scalar_with_fallback {
+    ($tx:expr, $query:literal, $fallback_query:literal,$( $param:expr ),* ) => {{
+        let primary_query = sqlx::query_scalar!($query,$($param),*)
+            .fetch_optional(&mut *$tx)
+            .await;
+
+        if let Ok(None) = primary_query {
+            tracing::info!("Data not found in job_args, falling back to fetching from queue");
+            sqlx::query_scalar!($fallback_query, $($param),*)
+                .fetch_optional(&mut *$tx)
+                .await
+        } else {
+            primary_query
+        }
+    }};
+}

backend/windmill-common/src/worker.rs

@@ -91,7 +91,7 @@ lazy_static::lazy_static! {
 }
 
 pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
-    if wc.worker_tags.len() == 0 {
+    if wc.worker_tags.is_empty() {
         tracing::error!("Empty tags in worker tags, skipping");
         return;
     }
@@ -123,7 +123,7 @@ pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
 pub async fn make_pull_query(wc: &WorkerConfig) {
     let mut queries = vec![];
     for tags in wc.priority_tags_sorted.iter() {
-        if tags.tags.len() == 0 {
+        if tags.tags.is_empty() {
             tracing::error!("Empty tags in priority tags, skipping");
             continue;
         }

backend/windmill-queue/Cargo.toml

@@ -42,4 +42,5 @@ async-recursion.workspace = true
 bigdecimal.workspace = true
 axum.workspace = true
 serde_urlencoded.workspace = true
-regex.workspace = true
+regex.workspace = true
+const_format.workspace = true

backend/windmill-queue/src/jobs.rs

@@ -6,8 +6,6 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec};
-
 use anyhow::Context;
 use async_recursion::async_recursion;
 use axum::{
@@ -31,12 +29,14 @@ use serde_json::{json, value::RawValue};
 use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction};
 #[cfg(feature = "benchmark")]
 use std::time::Instant;
+use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec};
 use tokio::{sync::RwLock, time::sleep};
 use tracing::{instrument, Instrument};
 use ulid::Ulid;
 use uuid::Uuid;
 use windmill_audit::audit_ee::{audit_log, AuditAuthor};
 use windmill_audit::ActionKind;
+use windmill_common::{fetch_optional_with_fallback, flows::FlowValueGetter};
 
 use windmill_common::{
     add_time,
@@ -44,12 +44,15 @@ use windmill_common::{
     db::{Authed, UserDB},
     error::{self, to_anyhow, Error},
     flow_status::{
-        BranchAllStatus, FlowCleanupModule, FlowStatus, FlowStatusModule, FlowStatusModuleWParent,
-        Iterator, JobResult, RestartedFrom, RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
+        BranchAllStatus, FlowCleanupModule, FlowStatus, FlowStatusGetter, FlowStatusModule,
+        FlowStatusModuleWParent, Iterator, JobResult, ParsedFlowStatusGetter, RestartedFrom,
+        RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
     },
     flows::{
         add_virtual_items_if_necessary, FlowModule, FlowModuleValue, FlowValue, InputTransform,
+        ParsedFlowValueGetter,
     },
+    impl_flow_status_getter, impl_flow_value_getter,
     jobs::{
         get_payload_tag_from_prefixed_path, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode,
         ENTRYPOINT_OVERRIDE, PREPROCESSOR_FAKE_ENTRYPOINT,
@@ -142,6 +145,14 @@ pub struct CanceledBy {
     pub reason: Option<String>,
 }
 
+#[derive(Debug, sqlx::FromRow)]
+pub struct CompletedSubFlow {
+    pub id: Uuid,
+    pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
+}
+
+impl_flow_status_getter!(CompletedSubFlow);
+
 pub async fn cancel_single_job<'c>(
     username: &str,
     reason: Option<String>,
@@ -222,26 +233,23 @@ pub async fn cancel_job<'c>(
     force_cancel: bool,
     require_anonymous: bool,
 ) -> error::Result<(Transaction<'c, Postgres>, Option<Uuid>)> {
-    let job = get_queued_job_tx(id, &w_id, &mut tx).await?;
-
-    if job.is_none() {
+    let Some(mut job) = get_queued_job_tx(id, w_id, &mut tx).await? else {
         return Ok((tx, None));
-    }
+    };
 
-    if require_anonymous && job.as_ref().unwrap().created_by != "anonymous" {
+    if require_anonymous && job.created_by != "anonymous" {
         return Err(Error::BadRequest(
             "You are not logged in and this job was not created by an anonymous user like you so you cannot cancel it".to_string(),
         ));
     }
 
-    let mut job = job.unwrap();
     if force_cancel {
         // if force canceling a flow step, make sure we force cancel from the highest parent
         loop {
             if job.parent_job.is_none() {
                 break;
             }
-            match get_queued_job_tx(job.parent_job.unwrap(), &w_id, &mut tx).await? {
+            match get_queued_job_tx(job.parent_job.unwrap(), w_id, &mut tx).await? {
                 Some(j) => {
                     job = j;
                 }
@@ -587,6 +595,8 @@ pub async fn add_completed_job<
 
     let mem_peak = mem_peak.max(queued_job.mem_peak.unwrap_or(0));
     add_time!(bench, "add_completed_job query START");
+
+    // On conflict (when id already exists), update the success and result fields.
     let _duration: i64 = sqlx::query_scalar!(
         "INSERT INTO completed_job AS cj
                    ( workspace_id
@@ -623,8 +633,8 @@ pub async fn add_completed_job<
             VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000, $7, $8, $9,\
                     $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)
          ON CONFLICT (id) DO UPDATE SET success = $7, result = $11 RETURNING duration_ms",
-        queued_job.workspace_id,
-        queued_job.id,
+        &queued_job.workspace_id,
+        &queued_job.id,
         queued_job.parent_job,
         queued_job.created_by,
         queued_job.created_at,
@@ -633,7 +643,7 @@ pub async fn add_completed_job<
         queued_job.script_hash.map(|x| x.0),
         queued_job.script_path,
         &queued_job.args as &Option<Json<HashMap<String, Box<RawValue>>>>,
-        result as Json<&T>,
+        &result as &Json<&T>,
         queued_job.raw_code,
         queued_job.raw_lock,
         canceled_by.is_some(),
@@ -650,16 +660,32 @@ pub async fn add_completed_job<
         queued_job.email,
         queued_job.visible_to_owner,
         if mem_peak > 0 { Some(mem_peak) } else { None },
-        queued_job.tag,
+        &queued_job.tag,
         queued_job.priority,
     )
     .fetch_one(&mut tx)
     .await
     .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e:#}")))?;
-    // tracing::error!("2 {:?}", start.elapsed());
-
     add_time!(bench, "add_completed_job query END");
 
+    add_time!(bench, "completed_jobs_result query START");
+    sqlx::query!(
+        "INSERT INTO completed_jobs_result(id, result, tag, workspace_id) VALUES($1, $2, $3, $4) ON CONFLICT (id) DO UPDATE SET result = $2",
+        queued_job.id,
+        &result as &Json<&T>,
+        queued_job.tag,
+        queued_job.workspace_id,
+    )
+    .execute(&mut tx)
+    .await
+    .map_err(|e| {
+        Error::InternalErr(format!(
+            "Could not add completed job result {job_id}: {e:#}"
+        ))
+    })?;
+
+    add_time!(bench, "completed_jobs_result query END");
+
     if !queued_job.is_flow_step {
         if _duration > 500
             && (queued_job.job_kind == JobKind::Script || queued_job.job_kind == JobKind::Preview)
@@ -864,10 +890,10 @@ pub async fn add_completed_job<
     tx.commit().await?;
     tracing::info!(
         %job_id,
-        root_job = ?queued_job.root_job.map(|x| x.to_string()).unwrap_or_else(|| String::new()),
+        root_job = ?queued_job.root_job.map(|x| x.to_string()).unwrap_or_default(),
         path = &queued_job.script_path(),
         job_kind = ?queued_job.job_kind,
-        started_at = ?queued_job.started_at.map(|x| x.to_string()).unwrap_or_else(|| String::new()),
+        started_at = ?queued_job.started_at.map(|x| x.to_string()).unwrap_or_default(),
         duration = ?_duration,
         permissioned_as = ?queued_job.permissioned_as,
         email = ?queued_job.email,
@@ -2267,55 +2293,48 @@ pub struct ResultWithId {
 
 pub async fn get_result_by_id(
     db: Pool<Postgres>,
-    w_id: String,
+    w_id: &str,
     flow_id: Uuid,
-    node_id: String,
-    json_path: Option<String>,
+    node_id: &str,
+    json_path: Option<&str>,
 ) -> error::Result<Box<RawValue>> {
-    match get_result_by_id_from_running_flow(
-        &db,
-        w_id.as_str(),
-        &flow_id,
-        node_id.as_str(),
-        json_path.clone(),
-    )
-    .await
-    {
+    #[derive(sqlx::FromRow, Debug)]
+    struct RunningFlowJobResult {
+        pub id: Uuid,
+        pub flow_status: Option<Json<Box<RawValue>>>,
+    }
+    impl_flow_status_getter!(RunningFlowJobResult);
+
+    match get_result_by_id_from_running_flow(&db, w_id, &flow_id, node_id, json_path).await {
         Ok(res) => Ok(res),
         Err(_) => {
-            let running_flow_job =sqlx::query_as::<_, QueuedJob>(
-                "SELECT * FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $1), $1) = id AND workspace_id = $2"
+            let running_flow_job =sqlx::query_as::<_, RunningFlowJobResult>(
+                "SELECT id, flow_status FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $1), $1) = id AND workspace_id = $2"
             ).bind(flow_id)
             .bind(&w_id)
             .fetch_optional(&db).await?;
+
             match running_flow_job {
                 Some(job) => {
                     let restarted_from = windmill_common::utils::not_found_if_none(
                         job.parse_flow_status()
-                            .map(|status| status.restarted_from)
-                            .flatten(),
+                            .and_then(|status| status.restarted_from),
                         "Id not found in the result's mapping of the root job and root job had no restarted from information",
                         format!("parent: {}, root: {}, id: {}", flow_id, job.id, node_id),
                     )?;
 
                     get_result_by_id_from_original_flow(
                         &db,
-                        w_id.as_str(),
+                        w_id,
                         &restarted_from.flow_job_id,
-                        node_id.as_str(),
-                        json_path.clone(),
+                        node_id,
+                        json_path,
                     )
                     .await
                 }
                 None => {
-                    get_result_by_id_from_original_flow(
-                        &db,
-                        w_id.as_str(),
-                        &flow_id,
-                        node_id.as_str(),
-                        json_path.clone(),
-                    )
-                    .await
+                    get_result_by_id_from_original_flow(&db, w_id, &flow_id, node_id, json_path)
+                        .await
                 }
             }
         }
@@ -2334,7 +2353,7 @@ pub async fn get_result_by_id_from_running_flow(
     w_id: &str,
     flow_id: &Uuid,
     node_id: &str,
-    json_path: Option<String>,
+    json_path: Option<&str>,
 ) -> error::Result<Box<RawValue>> {
     let flow_job_result = sqlx::query_as::<_, FlowJobResult>(
         "SELECT leaf_jobs->$1::text as leaf_jobs, parent_job FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3")
@@ -2352,8 +2371,7 @@ pub async fn get_result_by_id_from_running_flow(
 
     let job_result = flow_job_result
         .leaf_jobs
-        .map(|x| serde_json::from_str(x.get()).ok())
-        .flatten();
+        .and_then(|x| serde_json::from_str(x.get()).ok());
 
     if job_result.is_none() && flow_job_result.parent_job.is_some() {
         let parent_job = flow_job_result.parent_job.unwrap();
@@ -2378,9 +2396,9 @@ pub async fn get_result_by_id_from_running_flow(
 async fn get_completed_flow_node_result_rec(
     db: &Pool<Postgres>,
     w_id: &str,
-    subflows: Vec<CompletedJob>,
+    subflows: &[CompletedSubFlow],
     node_id: &str,
-    json_path: Option<String>,
+    json_path: Option<&str>,
 ) -> error::Result<Option<Box<RawValue>>> {
     for subflow in subflows {
         let flow_status = subflow.parse_flow_status().ok_or_else(|| {
@@ -2397,7 +2415,7 @@ async fn get_completed_flow_node_result_rec(
                     db,
                     w_id,
                     JobResult::SingleJob(leaf_job_uuid),
-                    json_path.clone(),
+                    json_path,
                 )
                 .await
                 .map(Some),
@@ -2405,7 +2423,7 @@ async fn get_completed_flow_node_result_rec(
                     db,
                     w_id,
                     JobResult::ListJob(jobs),
-                    json_path.clone(),
+                    json_path,
                 )
                 .await
                 .map(Some),
@@ -2416,10 +2434,11 @@ async fn get_completed_flow_node_result_rec(
                 ))),
             };
         } else {
-            let subflows = sqlx::query_as::<_, CompletedJob>(
-                "SELECT *, null as labels FROM completed_job WHERE parent_job = $1 AND workspace_id = $2 AND flow_status IS NOT NULL",
+            let subflows = sqlx::query_as::<_, CompletedSubFlow>(
+                "SELECT id, flow_status as labels FROM completed_job WHERE parent_job = $1 AND workspace_id = $2 AND flow_status IS NOT NULL",
             ).bind(subflow.id).bind(w_id).fetch_all(db).await?;
-            match get_completed_flow_node_result_rec(db, w_id, subflows, node_id, json_path.clone())
+
+            match get_completed_flow_node_result_rec(db, w_id, &subflows, node_id, json_path)
                 .await?
             {
                 Some(res) => return Ok(Some(res)),
@@ -2436,10 +2455,10 @@ async fn get_result_by_id_from_original_flow(
     w_id: &str,
     completed_flow_id: &Uuid,
     node_id: &str,
-    json_path: Option<String>,
+    json_path: Option<&str>,
 ) -> error::Result<Box<RawValue>> {
-    let flow_job = sqlx::query_as::<_, CompletedJob>(
-        "SELECT *, null as labels FROM completed_job WHERE id = $1 AND workspace_id = $2",
+    let flow_job = sqlx::query_as::<_, CompletedSubFlow>(
+        "SELECT id, flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2",
     )
     .bind(completed_flow_id)
     .bind(w_id)
@@ -2452,7 +2471,7 @@ async fn get_result_by_id_from_original_flow(
         format!("root: {}, id: {}", completed_flow_id, node_id),
     )?;
 
-    match get_completed_flow_node_result_rec(db, w_id, vec![flow_job], node_id, json_path).await? {
+    match get_completed_flow_node_result_rec(db, w_id, &[flow_job], node_id, json_path).await? {
         Some(res) => Ok(res),
         None => Err(error::Error::NotFound(format!(
             "Flow result by id not found going top-down from {}, (id: {})",
@@ -2465,31 +2484,33 @@ async fn extract_result_from_job_result(
     db: &Pool<Postgres>,
     w_id: &str,
     job_result: JobResult,
-    json_path: Option<String>,
+    json_path: Option<&str>,
 ) -> error::Result<Box<RawValue>> {
     match job_result {
         JobResult::ListJob(job_ids) => match json_path {
             Some(json_path) => {
-                let mut parts = json_path.split(".");
+                let mut parts = json_path.split('.');
 
-                let Some(idx) = parts.next().map(|x| x.parse::<usize>().ok()).flatten() else {
+                let Some(ref idx) = parts.next().and_then(|x| x.parse::<usize>().ok()) else {
                     return Ok(to_raw_value(&serde_json::Value::Null));
                 };
-                let Some(job_id) = job_ids.get(idx).cloned() else {
+                let Some(job_id) = job_ids.get(*idx) else {
                     return Ok(to_raw_value(&serde_json::Value::Null));
                 };
-                Ok(sqlx::query_as::<_, ResultR>(
-                    "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2",
-                )
-                .bind(job_id)
-                .bind(w_id)
-                .bind(
-                    parts.map(|x| x.to_string()).collect::<Vec<_>>()
-                )
-                .fetch_optional(db)
-                .await?
-                .map(|r| r.result.map(|x| x.0))
-                .flatten()
+
+                let parts = parts.map(|x| x.to_string()).collect_vec();
+
+                Ok(fetch_optional_with_fallback!(
+                    db,
+                    query_as,
+                    ResultR,
+                    "SELECT result #> $3 as result FROM {} WHERE id = $1 AND workspace_id = $2",
+                    "completed_jobs_result" || "completed_job",
+                    *job_id,
+                    w_id,
+                    &parts
+                )?
+                .and_then(|r| r.result.map(|x| x.0))
                 .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)))
             }
             None => {
@@ -2503,6 +2524,7 @@ async fn extract_result_from_job_result(
                 .into_iter()
                 .filter_map(|x| x.result.map(|y| (x.id, y)))
                 .collect::<HashMap<Uuid, Json<Box<RawValue>>>>();
+
                 let result = job_ids
                     .into_iter()
                     .map(|id| {
@@ -2510,25 +2532,31 @@ async fn extract_result_from_job_result(
                             .map(|x| x.0.clone())
                             .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))
                     })
-                    .collect::<Vec<_>>();
+                    .collect_vec();
                 Ok(to_raw_value(&result))
             }
         },
-        JobResult::SingleJob(x) => Ok(sqlx::query_as::<_, ResultR>(
-            "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2",
-        )
-        .bind(x)
-        .bind(w_id)
-        .bind(
-            json_path
+        // ici
+        JobResult::SingleJob(x) => {
+            let path = json_path
                 .map(|x| x.split(".").map(|x| x.to_string()).collect::<Vec<_>>())
-                .unwrap_or_default(),
-        )
-        .fetch_optional(db)
-        .await?
-        .map(|r| r.result.map(|x| x.0))
-        .flatten()
-        .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))),
+                .unwrap_or_default();
+
+            let res = fetch_optional_with_fallback!(
+                db,
+                query_as,
+                ResultR,
+                "SELECT result #> $3 as result FROM {} WHERE id = $1 AND workspace_id = $2",
+                "completed_jobs_result" || "completed_job",
+                x,
+                w_id,
+                &path
+            )?
+            .and_then(|r| r.result.map(|x| x.0))
+            .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
+
+            Ok(res)
+        }
     }
 }
 
@@ -2643,7 +2671,7 @@ pub struct PushArgsOwned {
     pub args: HashMap<String, Box<RawValue>>,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct PushArgs<'c> {
     pub extra: Option<HashMap<String, Box<RawValue>>>,
     pub args: &'c HashMap<String, Box<RawValue>>,
@@ -2754,7 +2782,7 @@ fn restructure_cloudevents_metadata(
         .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
     let str = data.to_string();
 
-    let wrap_body = str.len() > 0 && str.chars().next().unwrap() != '{';
+    let wrap_body = !str.is_empty() && !str.starts_with('{');
 
     if wrap_body {
         let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
@@ -2784,7 +2812,7 @@ impl PushArgsOwned {
             extra.insert("raw_string".to_string(), to_raw_value(&str));
         }
 
-        let wrap_body = force_wrap_body || str.len() > 0 && str.chars().next().unwrap() != '{';
+        let wrap_body = force_wrap_body || !str.is_empty() && !str.starts_with('{');
 
         if wrap_body {
             let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
@@ -3895,6 +3923,10 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
     };
 
     tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}");
+
+    let raw_flow = raw_flow.map(Json);
+    let args = Json(args);
+
     let uuid = sqlx::query_scalar!(
         "INSERT INTO queue
             (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, 
@@ -3913,12 +3945,12 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
         scheduled_for_o,
         script_hash,
         script_path.clone(),
-        raw_code,
+        raw_code.clone(),
         raw_lock,
-        Json(args) as Json<PushArgs>,
+        args.clone() as Json<PushArgs>,
         job_kind.clone() as JobKind,
         schedule_path,
-        raw_flow.map(Json) as Option<Json<FlowValue>>,
+        raw_flow.clone() as Option<Json<FlowValue>>,
         flow_status.map(Json) as Option<Json<FlowStatus>>,
         is_flow_step,
         language as Option<ScriptLang>,
@@ -3927,7 +3959,7 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
         email,
         visible_to_owner,
         root_job,
-        tag,
+        tag.clone(),
         concurrent_limit,
         if concurrent_limit.is_some() {  concurrency_time_window_s } else { None },
         custom_timeout,
@@ -3939,6 +3971,34 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
     .await
     .map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?;
 
+    // insert into args queue
+    sqlx::query!(
+        r#"
+        INSERT INTO job_args (id, workspace_id, args, tag)
+        VALUES ($1, $2, $3, $4)
+        "#,
+        uuid,
+        workspace_id,
+        args as Json<PushArgs>,
+        &tag
+    )
+    .execute(&mut tx)
+    .await?;
+
+    sqlx::query!(
+        r#"
+        INSERT INTO job_params (id, workspace_id, raw_code, raw_flow, tag)
+        VALUES ($1, $2, $3, $4, $5)
+        "#,
+        uuid,
+        workspace_id,
+        raw_code,
+        raw_flow as Option<Json<FlowValue>>,
+        tag
+    )
+    .execute(&mut tx)
+    .await?;
+
     tracing::debug!("Pushed {job_id}");
     // TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction.
     #[cfg(feature = "prometheus")]
@@ -4059,11 +4119,8 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>(
 }
 
 pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value {
-    let reason = job
-        .canceled_reason
-        .as_deref()
-        .unwrap_or_else(|| "no reason given");
-    let canceler = job.canceled_by.as_deref().unwrap_or_else(|| "unknown");
+    let reason = job.canceled_reason.as_deref().unwrap_or("no reason given");
+    let canceler = job.canceled_by.as_deref().unwrap_or("unknown");
     serde_json::json!({"message": format!("Job canceled: {reason} by {canceler}"), "name": "Canceled", "reason": reason, "canceler": canceler})
 }
 
@@ -4086,8 +4143,19 @@ async fn restarted_flows_resolution(
     ),
     Error,
 > {
-    let completed_job = sqlx::query_as::<_, CompletedJob>(
-        "SELECT *, null as labels FROM completed_job WHERE id = $1 and workspace_id = $2",
+    #[derive(Debug, sqlx::FromRow)]
+    struct QueryResults {
+        pub script_path: Option<String>,
+        pub priority: Option<i16>,
+        pub raw_flow: Option<sqlx::types::Json<Box<RawValue>>>,
+        pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
+    }
+
+    impl_flow_status_getter!(QueryResults);
+    impl_flow_value_getter!(QueryResults);
+
+    let completed_job = sqlx::query_as::<_, QueryResults>(
+        "SELECT script_path, priority, raw_flow, flow_status FROM completed_job WHERE id = $1 and workspace_id = $2",
     )
     .bind(completed_flow_id)
     .bind(workspace_id)
@@ -4120,10 +4188,9 @@ async fn restarted_flows_resolution(
         if flow_value_if_any
             .clone()
             .map(|fv| {
-                fv.modules
+                !fv.modules
                     .iter()
-                    .find(|flow_value_module| flow_value_module.id == module.id())
-                    .is_none()
+                    .any(|flow_value_module| flow_value_module.id == module.id())
             })
             .unwrap_or(false)
         {
@@ -4239,7 +4306,7 @@ async fn restarted_flows_resolution(
             truncated_modules.push(FlowStatusModule::WaitingForPriorSteps { id: module.id() });
         } else {
             // else we simply "transfer" the module from the completed flow to the new one if it's a success
-            step_n = step_n + 1;
+            step_n += 1;
             match module.clone() {
                 FlowStatusModule::Success { .. } => Ok(truncated_modules.push(module)),
                 _ => Err(Error::InternalErr(format!(

backend/windmill-worker/src/worker_flow.rs

@@ -29,20 +29,20 @@ use sqlx::FromRow;
 use tokio::sync::mpsc::Sender;
 use tracing::instrument;
 use uuid::Uuid;
-use windmill_common::add_time;
 use windmill_common::auth::JobPerms;
 #[cfg(feature = "benchmark")]
 use windmill_common::bench::BenchmarkIter;
 use windmill_common::db::Authed;
 use windmill_common::flow_status::{
-    ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult,
+    ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult, ParsedFlowStatusGetter,
 };
-use windmill_common::flows::add_virtual_items_if_necessary;
+use windmill_common::flows::{add_virtual_items_if_necessary, ParsedFlowValueGetter};
 use windmill_common::jobs::{
     script_hash_to_tag_and_limits, script_path_to_payload, BranchResults, JobPayload, QueuedJob,
     RawCode, ENTRYPOINT_OVERRIDE,
 };
 use windmill_common::worker::to_raw_value;
+use windmill_common::{add_time, fetch_one_with_fallback, fetch_optional_with_fallback};
 use windmill_common::{
     error::{self, to_anyhow, Error},
     flow_status::{
@@ -61,6 +61,19 @@ type DB = sqlx::Pool<sqlx::Postgres>;
 
 use windmill_queue::{canceled_job_to_result, get_queued_job_tx, push, QueueTransaction};
 
+async fn get_args_from_job_id(db: &DB, id: &Uuid) -> Result<RowArgs, Error> {
+    let args = fetch_one_with_fallback!(
+        db,
+        query_as,
+        RowArgs,
+        "SELECT args FROM {} WHERE id = $1",
+        "job_args" || "queue",
+        id
+    );
+
+    args.map_err(|e| Error::InternalErr(format!("retrieval of args from state: {e:#}")))
+}
+
 // #[instrument(level = "trace", skip_all)]
 pub async fn update_flow_status_after_job_completion<
     R: rsmq_async::RsmqConnection + Send + Sync + Clone,
@@ -229,10 +242,10 @@ pub async fn update_flow_status_after_job_completion_internal<
         })?;
 
         let old_status = serde_json::from_str::<FlowStatus>(old_status_json.flow_status.get())
-            .or_else(|e| {
-                Err(Error::InternalErr(format!(
+            .map_err(|e| {
+                Error::InternalErr(format!(
                     "requiring status to be parsable as FlowStatus: {e:?}"
-                )))
+                ))
             })?;
 
         let current_module = if let Some(x) = old_status_json.current_module {
@@ -289,7 +302,7 @@ pub async fn update_flow_status_after_job_completion_internal<
 
         // 0 length flows are not failure steps
         let is_failure_step =
-            old_status.step >= old_status.modules.len() as i32 && old_status.modules.len() > 0;
+            old_status.step >= old_status.modules.len() as i32 && !old_status.modules.is_empty();
 
         let (mut stop_early, mut skip_if_stop_early, continue_on_error) = if let Some(se) =
             stop_early_override
@@ -302,14 +315,13 @@ pub async fn update_flow_status_after_job_completion_internal<
             };
 
             let is_flow = if let Some(step) = step {
-                sqlx::query_scalar!(
-                    "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM queue WHERE id = $2",
+                fetch_one_with_fallback!(db,
+                    query_scalar,
+                    Option<bool>,
+                    "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM {} WHERE id = $2",
+                    "job_params" || "queue",
                     step as i32,
-                    &flow
-                )
-                    .fetch_one(db)
-                    .await
-                    .map_err(|e| {
+                    &flow).map_err(|e| {
                         Error::InternalErr(format!("error during retrieval of step's type: {e:#}"))
                     })?
                     .unwrap_or(false)
@@ -342,19 +354,9 @@ pub async fn update_flow_status_after_job_completion_internal<
                         }
                         _ => None,
                     };
-                    let args = sqlx::query_as::<_, RowArgs>(
-                        "SELECT
-                                        args
-                                    FROM queue
-                                    WHERE id = $2",
-                    )
-                    .bind(old_status.step)
-                    .bind(flow)
-                    .fetch_one(db)
-                    .await
-                    .map_err(|e| {
-                        Error::InternalErr(format!("retrieval of args from state: {e:#}"))
-                    })?;
+
+                    let args = get_args_from_job_id(db, &flow).await?;
+
                     compute_bool_from_expr(
                         expr.to_string(),
                         Marc::new(args.args.unwrap_or_default().0),
@@ -442,7 +444,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                             None
                         };
 
-                        let nindex = if let Some(position) = position { 
+                        let nindex = if let Some(position) = position {
                             sqlx::query_scalar!(
                             "UPDATE queue
                             SET flow_status = JSONB_SET(
@@ -489,7 +491,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                             None
                         };
 
-                        let nindex = if let Some(position) = position { 
+                        let nindex = if let Some(position) = position {
                             sqlx::query_scalar!(
                                 "UPDATE queue
                                 SET flow_status = JSONB_SET(
@@ -658,7 +660,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                 flow_jobs_success,
                 flow_jobs,
                 ..
-            } if branch.to_owned() < len - 1 && (success || skip_branch_failure) => {
+            } if *branch < len - 1 && (success || skip_branch_failure) => {
                 if let Some(jobs) = flow_jobs {
                     set_success_in_flow_job_success(
                         flow_jobs_success,
@@ -870,19 +872,7 @@ pub async fn update_flow_status_after_job_completion_internal<
                     .as_ref()
                     .and_then(|m| m.stop_after_all_iters_if.as_ref().map(|x| x.expr.clone()))
                 {
-                    let args = sqlx::query_as::<_, RowArgs>(
-                        "SELECT
-                            args
-                        FROM queue
-                        WHERE id = $2",
-                    )
-                    .bind(old_status.step)
-                    .bind(flow)
-                    .fetch_one(db)
-                    .await
-                    .map_err(|e| {
-                        Error::InternalErr(format!("retrieval of args from state: {e:#}"))
-                    })?;
+                    let args = get_args_from_job_id(db, &flow).await?;
 
                     let should_stop = compute_bool_from_expr(
                         expr.to_string(),
@@ -1244,7 +1234,7 @@ fn get_module(flow_job: &QueuedJob, module_step: &Step) -> Option<FlowModule> {
     if let Some(raw_flow) = raw_flow {
         match module_step {
             Step::PreprocessorStep => raw_flow.preprocessor_module.map(|x| *x.clone()),
-            Step::Step(i) => raw_flow.modules.get(*i).map(|x| x.clone()),
+            Step::Step(i) => raw_flow.modules.get(*i).cloned(),
             Step::FailureStep => raw_flow.failure_module.map(|x| *x.clone()),
         }
     } else {
@@ -1269,8 +1259,7 @@ async fn compute_skip_branchall_failure<'c>(
             .map(|p| {
                 BRANCHALL_INDEX_RE
                     .captures(&p)
-                    .map(|x| x.get(1).unwrap().as_str().parse::<i32>().ok())
-                    .flatten()
+                    .and_then(|x| x.get(1).unwrap().as_str().parse::<i32>().ok())
                     .ok_or(Error::InternalErr(format!(
                         "could not parse branchall index from path: {p}"
                     )))
@@ -1291,14 +1280,14 @@ async fn compute_skip_branchall_failure<'c>(
 }
 
 async fn has_failure_module<'c>(flow: Uuid, db: &DB) -> Result<bool, Error> {
-    sqlx::query_scalar::<_, Option<bool>>(
-        "SELECT raw_flow->'failure_module' != 'null'::jsonb
-        FROM queue
-        WHERE id = $1",
+    fetch_one_with_fallback!(
+        db,
+        query_scalar,
+        Option<bool>,
+        "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM {} WHERE id = $1",
+        "job_params" || "queue",
+        flow
     )
-    .bind(flow)
-    .fetch_one(db)
-    .await
     .map_err(|e| {
         Error::InternalErr(format!(
             "error during retrieval of has_failure_module: {e:#}"
@@ -1610,6 +1599,7 @@ pub struct ResumeRow {
 
 #[derive(FromRow)]
 pub struct RawArgs {
+    #[allow(dead_code)]
     pub args: Option<Json<HashMap<String, Box<RawValue>>>>,
 }
 
@@ -1638,6 +1628,7 @@ fn potentially_crash_for_testing() {
 lazy_static::lazy_static! {
     pub static ref EHM: HashMap<String, Box<RawValue>> = HashMap::new();
 }
+
 // #[async_recursion]
 // #[instrument(level = "trace", skip_all)]
 async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
@@ -1725,7 +1716,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
                     flow_job.workspace_id.as_str(),
                     flow_job.id
                 ).fetch_all(db).await?;
-                if overlapping.len() > 0 {
+                if !overlapping.is_empty() {
                     let overlapping_str = overlapping
                         .iter()
                         .map(|x| x.to_string())
@@ -1804,8 +1795,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         // else pass the last job result. Either from the function arg if it's set, or manually fetch it from the previous job
         // having last_job_result empty can happen either when the job was suspended and is being restarted, or if it's a
         // flow restart from a specific step
-        if last_job_result.is_some() {
-            last_job_result.unwrap()
+        if let Some(last_job_result) = last_job_result {
+            last_job_result
         } else {
             match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status).await? {
                 None => Arc::new(to_raw_value(&json!("{}"))),
@@ -2260,17 +2251,18 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             );
             Ok(Marc::new(hm))
         } else if let Some(id) = get_args_from_id {
-            let row = sqlx::query_as::<_, RawArgs>(
-                "SELECT args FROM completed_job WHERE id = $1 AND workspace_id = $2",
-            )
-            .bind(id)
-            .bind(&flow_job.workspace_id)
-            .fetch_optional(db)
-            .await?;
+            let row = fetch_optional_with_fallback!(
+                db,
+                query_as,
+                RowArgs,
+                "SELECT args FROM {} WHERE id = $1 AND workspace_id = $2",
+                "job_args" || "completed_job",
+                id,
+                &flow_job.workspace_id
+            )?;
+
             if let Some(raw_args) = row {
-                Ok(Marc::new(
-                    raw_args.args.map(|x| x.0).unwrap_or_else(HashMap::new),
-                ))
+                Ok(Marc::new(raw_args.args.map(|x| x.0).unwrap_or_default()))
             } else {
                 Ok(Marc::new(HashMap::new()))
             }
@@ -2389,7 +2381,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
 
     let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
     let nargs = args.as_ref();
-    for i in (0..len).into_iter() {
+    for i in 0..len {
         if i % 100 == 0 && i != 0 {
             tracing::info!(id = %flow_job.id, root_id = %job_root, "pushed (non-commited yet) first {i} subflows of {len}");
             sqlx::query!(
@@ -3902,13 +3894,15 @@ async fn get_previous_job_result(
             Ok(Some(retrieve_flow_jobs_results(db, w_id, flow_jobs).await?))
         }
         Some(FlowStatusModule::Success { job, .. }) => Ok(Some(
-            sqlx::query_scalar::<_, Json<Box<RawValue>>>(
-                "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2",
-            )
-            .bind(job)
-            .bind(w_id)
-            .fetch_one(db)
-            .await?
+            fetch_one_with_fallback!(
+                db,
+                query_scalar,
+                Json<Box<RawValue>>,
+                "SELECT result FROM {} WHERE id = $1 AND workspace_id = $2",
+                "completed_jobs_result" || "completed_job",
+                job,
+                w_id
+            )?
             .0,
         )),
         _ => Ok(None),

benchmarks/lib.ts

@@ -2,7 +2,7 @@ import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts";
 import * as windmill from "https://deno.land/x/windmill@v1.174.0/mod.ts";
 import * as api from "https://deno.land/x/windmill@v1.174.0/windmill-api/index.ts";
 
-export const VERSION = "v1.417.0";
+export const VERSION = "v1.416.2";
 
 export async function login(email: string, password: string): Promise<string> {
   return await windmill.UserService.login({

cli/instance.ts

@@ -184,7 +184,6 @@ export type InstanceSyncOptions = {
   folderPerInstance?: boolean;
   yes?: boolean;
   prefix?: string;
-  prefixSettings?: boolean;
 };
 
 export async function pickInstance(
@@ -266,22 +265,21 @@ async function instancePull(opts: GlobalOptions & InstanceSyncOptions) {
   log.info("Pulling instance-level changes");
   log.info(`remote (${instance.name}) -> local`);
 
-  opts.prefix = opts.prefix ?? instance.prefix;
   let uChanges = 0;
   if (!opts.skipUsers) {
-    uChanges = (await pullInstanceUsers(opts, true)) as number;
+    uChanges = (await pullInstanceUsers(true)) as number;
   }
   let sChanges = 0;
   if (!opts.skipSettings) {
-    sChanges = (await pullInstanceSettings(opts, true)) as number;
+    sChanges = (await pullInstanceSettings(true)) as number;
   }
   let cChanges = 0;
   if (!opts.skipConfigs) {
-    cChanges = (await pullInstanceConfigs(opts, true)) as number;
+    cChanges = (await pullInstanceConfigs(true)) as number;
   }
   let gChanges = 0;
   if (!opts.skipGroups) {
-    gChanges = (await pullInstanceGroups(opts, true)) as number;
+    gChanges = (await pullInstanceGroups(true)) as number;
   }
 
   const totalChanges = uChanges + sChanges + cChanges + gChanges;
@@ -297,16 +295,16 @@ async function instancePull(opts: GlobalOptions & InstanceSyncOptions) {
 
     if (confirm) {
       if (!opts.skipUsers && uChanges > 0) {
-        await pullInstanceUsers(opts);
+        await pullInstanceUsers();
       }
       if (!opts.skipSettings && sChanges > 0) {
-        await pullInstanceSettings(opts);
+        await pullInstanceSettings();
       }
       if (!opts.skipConfigs && cChanges > 0) {
-        await pullInstanceConfigs(opts);
+        await pullInstanceConfigs();
       }
       if (!opts.skipGroups && gChanges > 0) {
-        await pullInstanceGroups(opts);
+        await pullInstanceGroups();
       }
     }
   } else {
@@ -402,19 +400,19 @@ async function instancePush(opts: GlobalOptions & InstanceSyncOptions) {
 
   let uChanges = 0;
   if (!opts.skipUsers) {
-    uChanges = (await pushInstanceUsers(opts, true)) as number;
+    uChanges = (await pushInstanceUsers(true)) as number;
   }
   let sChanges = 0;
   if (!opts.skipSettings) {
-    sChanges = (await pushInstanceSettings(opts, true)) as number;
+    sChanges = (await pushInstanceSettings(true, opts.baseUrl)) as number;
   }
   let cChanges = 0;
   if (!opts.skipConfigs) {
-    cChanges = (await pushInstanceConfigs(opts, true)) as number;
+    cChanges = (await pushInstanceConfigs(true)) as number;
   }
   let gChanges = 0;
   if (!opts.skipGroups) {
-    gChanges = (await pushInstanceGroups(opts, true)) as number;
+    gChanges = (await pushInstanceGroups(true)) as number;
   }
 
   const totalChanges = uChanges + sChanges + cChanges + gChanges;
@@ -430,16 +428,16 @@ async function instancePush(opts: GlobalOptions & InstanceSyncOptions) {
 
     if (confirm) {
       if (!opts.skipUsers && uChanges > 0) {
-        await pushInstanceUsers(opts);
+        await pushInstanceUsers();
       }
       if (!opts.skipSettings && sChanges > 0) {
-        await pushInstanceSettings(opts);
+        await pushInstanceSettings(false, opts.baseUrl);
       }
       if (!opts.skipConfigs && cChanges > 0) {
-        await pushInstanceConfigs(opts);
+        await pushInstanceConfigs();
       }
       if (!opts.skipGroups && gChanges > 0) {
-        await pushInstanceGroups(opts);
+        await pushInstanceGroups();
       }
     }
   } else {
@@ -711,10 +709,7 @@ const command = new Command()
     "--prefix <prefix:string>",
     "Prefix of the local workspaces to pull, used to create the folders when using --include-workspaces"
   )
-  .option(
-    "--prefix-settings",
-    "Store instance yamls inside prefixed folders when using --prefix and --folder-per-instance"
-  )
+
   .action(instancePull as any)
   .command("push")
   .description(
@@ -735,10 +730,6 @@ const command = new Command()
     "--prefix <prefix:string>",
     "Prefix of the local workspaces folders to push"
   )
-  .option(
-    "--prefix-settings",
-    "Store instance yamls inside prefixed folders when using --prefix and --folder-per-instance"
-  )
   .action(instancePush as any)
   .command("whoami")
   .description("Display information about the currently logged-in user")

cli/main.ts

@@ -60,7 +60,7 @@ export {
 //   }
 // });
 
-export const VERSION = "1.417.0";
+export const VERSION = "1.416.2";
 
 const command = new Command()
   .name("wmill")

cli/settings.ts

@@ -3,8 +3,8 @@ import { Confirm } from "./deps.ts";
 import { colors } from "./deps.ts";
 import { yamlParseFile } from "./deps.ts";
 import { log } from "./deps.ts";
-import { compareInstanceObjects, InstanceSyncOptions } from "./instance.ts";
-import { isSuperset, GlobalOptions } from "./types.ts";
+import { compareInstanceObjects } from "./instance.ts";
+import { isSuperset } from "./types.ts";
 import { deepEqual } from "./utils.ts";
 import * as wmill from "./gen/services.gen.ts";
 import { Config, GlobalSetting } from "./gen/types.gen.ts";
@@ -113,10 +113,10 @@ export async function pushWorkspaceSettings(
         workspace,
         requestBody: localSettings.auto_invite_enabled
           ? {
-              operator: localSettings.auto_invite_as === "operator",
-              invite_all: true,
-              auto_add: localSettings.auto_invite_mode === "add",
-            }
+            operator: localSettings.auto_invite_as === "operator",
+            invite_all: true,
+            auto_add: localSettings.auto_invite_mode === "add",
+          }
           : {},
       });
     } catch (_) {
@@ -128,10 +128,10 @@ export async function pushWorkspaceSettings(
         workspace,
         requestBody: localSettings.auto_invite_enabled
           ? {
-              operator: localSettings.auto_invite_as === "operator",
-              invite_all: false,
-              auto_add: localSettings.auto_invite_mode === "add",
-            }
+            operator: localSettings.auto_invite_as === "operator",
+            invite_all: false,
+            auto_add: localSettings.auto_invite_mode === "add",
+          }
           : {},
       });
     }
@@ -156,7 +156,7 @@ export async function pushWorkspaceSettings(
       settings.error_handler_extra_args
     ) ||
     localSettings.error_handler_muted_on_cancel !==
-      settings.error_handler_muted_on_cancel
+    settings.error_handler_muted_on_cancel
   ) {
     log.debug(`Updating error handler...`);
     await wmill.editErrorHandler({
@@ -262,41 +262,30 @@ export async function pushWorkspaceKey(
 
 const INSTANCE_SETTINGS_PATH = "instance_settings.yaml";
 
-export async function readInstanceSettings(
-  opts: GlobalOptions & InstanceSyncOptions
-) {
+export async function readInstanceSettings() {
   let localSettings: GlobalSetting[] = [];
 
-  let path = INSTANCE_SETTINGS_PATH;
-  if (opts.prefix && opts.folderPerInstance && opts.prefixSettings) {
-    path = `${opts.prefix}/${INSTANCE_SETTINGS_PATH}`;
-  }
-
   try {
-    localSettings = (await yamlParseFile(path)) as GlobalSetting[];
+    localSettings = (await yamlParseFile(INSTANCE_SETTINGS_PATH)) as GlobalSetting[];
   } catch {
-    log.warn(`No ${path} found`);
+    log.warn(`No ${INSTANCE_SETTINGS_PATH} found`);
   }
   return localSettings;
 }
 
+
 import { decrypt, encrypt } from "./local_encryption.ts";
 
-const SENSITIVE_FIELD: string[] = ["license_key", "jwt_secret"];
+const SENSITIVE_FIELD: string[] = ["license_key", "jwt_secret"]
 
-async function processInstanceSettings(
-  settings: GlobalSetting[],
-  mode: "encode" | "decode"
-): Promise<GlobalSetting[]> {
+async function processInstanceSettings(settings: GlobalSetting[], mode: "encode" | "decode"): Promise<GlobalSetting[]> {
   const encKey = process.env.WMILL_INSTANCE_LOCAL_ENCRYPTION_KEY;
   if (encKey) {
-    const res: GlobalSetting[] = [];
+    const res: GlobalSetting[] = []
 
     for (const s of settings) {
       if (SENSITIVE_FIELD.includes(s.name) && typeof s.value === "string") {
-        res.push(
-          (await processField(s, "value", encKey, mode)) as GlobalSetting
-        );
+        res.push(await processField(s, "value", encKey, mode) as GlobalSetting);
       } else if (s.name == "oauths") {
         if (typeof s.value === "object") {
           const oauths = s.value as { [key: string]: any };
@@ -314,40 +303,24 @@ async function processInstanceSettings(
     }
     return res;
   } else {
-    log.warn(
-      "No encryption key found, skipping encryption. Recommend setting WMILL_INSTANCE_LOCAL_ENCRYPTION_KEY"
-    );
+    log.warn("No encryption key found, skipping encryption. Recommend setting WMILL_INSTANCE_LOCAL_ENCRYPTION_KEY");
   }
   return settings;
 }
 
-async function processField(
-  obj: { [key: string]: any },
-  field: string,
-  encKey: string,
-  mode: "encode" | "decode"
-): Promise<{ [key: string]: any }> {
+async function processField(obj: { [key: string]: any }, field: string, encKey: string, mode: "encode" | "decode"): Promise<{ [key: string]: any }> {
   return {
     ...obj,
-    [field]:
-      mode === "encode"
-        ? await encrypt(obj[field], encKey)
-        : ((await decrypt(obj[field], encKey)) as any),
-  };
+    [field]: mode === "encode" ? await encrypt(obj[field], encKey) : await decrypt(obj[field], encKey) as any,
+  }
 }
 
-export async function pullInstanceSettings(
-  opts: GlobalOptions & InstanceSyncOptions,
-  preview = false
-) {
+export async function pullInstanceSettings(preview = false) {
   const remoteSettings = await wmill.listGlobalSettings();
 
   if (preview) {
-    const localSettings: GlobalSetting[] = await readInstanceSettings(opts);
-    const processedSettings = await processInstanceSettings(
-      remoteSettings,
-      "encode"
-    );
+    const localSettings: GlobalSetting[] = await readInstanceSettings();
+    const processedSettings = await processInstanceSettings(remoteSettings, "encode");
     return compareInstanceObjects(
       processedSettings,
       localSettings,
@@ -357,10 +330,7 @@ export async function pullInstanceSettings(
   } else {
     log.info("Pulling settings from instance");
 
-    const processedSettings = await processInstanceSettings(
-      remoteSettings,
-      "encode"
-    );
+    const processedSettings = await processInstanceSettings(remoteSettings, "encode");
     await Deno.writeTextFile(
       INSTANCE_SETTINGS_PATH,
       yamlStringify(processedSettings)
@@ -371,19 +341,19 @@ export async function pullInstanceSettings(
 }
 
 export async function pushInstanceSettings(
-  opts: GlobalOptions & InstanceSyncOptions,
-  preview: boolean = false
+  preview: boolean = false,
+  baseUrl?: string
 ) {
   const remoteSettings = await wmill.listGlobalSettings();
-  let localSettings: GlobalSetting[] = await readInstanceSettings(opts);
+  let localSettings: GlobalSetting[] = await readInstanceSettings();
   localSettings = await processInstanceSettings(localSettings, "decode");
 
-  if (opts.baseUrl) {
+  if (baseUrl) {
     localSettings = localSettings.filter((s) => s.name !== "base_url");
     localSettings.push({
       name: "base_url",
       //@ts-ignore
-      value: opts.baseUrl,
+      value: baseUrl,
     });
   }
 
@@ -434,30 +404,18 @@ export async function pushInstanceSettings(
   }
 }
 
-const INSTANCE_CONFIGS_PATH = "instance_configs.yaml";
-
-export async function readLocalConfigs(
-  opts: GlobalOptions & InstanceSyncOptions
-) {
+export async function readLocalConfigs() {
   let localConfigs: Config[] = [];
 
-  let path = INSTANCE_CONFIGS_PATH;
-  if (opts.prefix && opts.folderPerInstance && opts.prefixSettings) {
-    path = `${opts.prefix}/${INSTANCE_CONFIGS_PATH}`;
-  }
-
   try {
-    localConfigs = (await yamlParseFile(path)) as Config[];
+    localConfigs = (await yamlParseFile("instance_configs.yaml")) as Config[];
   } catch {
-    log.warn(`No ${path} found`);
+    log.warn("No instance_configs.yaml found");
   }
   return localConfigs;
 }
 
-export async function pullInstanceConfigs(
-  opts: GlobalOptions & InstanceSyncOptions,
-  preview = false
-) {
+export async function pullInstanceConfigs(preview = false) {
   const remoteConfigs = (await wmill.listConfigs()).map((x) => {
     return {
       ...x,
@@ -466,7 +424,7 @@ export async function pullInstanceConfigs(
   });
 
   if (preview) {
-    const localConfigs: Config[] = await readLocalConfigs(opts);
+    const localConfigs: Config[] = await readLocalConfigs();
 
     return compareInstanceObjects(
       remoteConfigs,
@@ -478,25 +436,22 @@ export async function pullInstanceConfigs(
     log.info("Pulling configs from instance");
 
     await Deno.writeTextFile(
-      INSTANCE_CONFIGS_PATH,
+      "instance_configs.yaml",
       yamlStringify(remoteConfigs as any)
     );
 
-    log.info(colors.green(`Configs written to ${INSTANCE_CONFIGS_PATH}`));
+    log.info(colors.green("Configs written to instance_configs.yaml"));
   }
 }
 
-export async function pushInstanceConfigs(
-  opts: GlobalOptions & InstanceSyncOptions,
-  preview: boolean = false
-) {
+export async function pushInstanceConfigs(preview: boolean = false) {
   const remoteConfigs = (await wmill.listConfigs()).map((x) => {
     return {
       ...x,
       name: removeWorkerPrefix(x.name),
     };
   });
-  const localConfigs = await readLocalConfigs(opts);
+  const localConfigs = await readLocalConfigs();
 
   if (preview) {
     return compareInstanceObjects(

cli/user.ts

@@ -6,7 +6,7 @@ import {
   removeType,
   removePathPrefix,
 } from "./types.ts";
-import { compareInstanceObjects, InstanceSyncOptions } from "./instance.ts";
+import { compareInstanceObjects } from "./instance.ts";
 import {
   colors,
   Command,
@@ -386,71 +386,47 @@ export async function pushGroup(
   }
 }
 
-const INSTANCE_USERS_PATH = "instance_users.yaml";
-
-export async function pullInstanceUsers(
-  opts: GlobalOptions & InstanceSyncOptions,
-  preview: boolean = false
-) {
+export async function pullInstanceUsers(preview: boolean = false) {
   const remoteUsers = await wmill.globalUsersExport();
 
   if (preview) {
-    const localUsers: ExportedUser[] = await readInstanceUsers(opts);
+    const localUsers: ExportedUser[] = await readInstanceUsers();
     return compareInstanceObjects(remoteUsers, localUsers, "email", "user");
   } else {
     log.info("Pulling users from instance...");
     await Deno.writeTextFile(
-      INSTANCE_USERS_PATH,
+      "instance_users.yaml",
       yamlStringify(remoteUsers as any)
     );
-    log.info(colors.green(`Users written to ${INSTANCE_USERS_PATH}`));
+    log.info(colors.green("Users written to instance_users.yaml"));
   }
 }
 
-export async function readInstanceUsers(
-  opts: GlobalOptions & InstanceSyncOptions
-) {
+export async function readInstanceUsers() {
   let localUsers: ExportedUser[] = [];
-
-  let path = INSTANCE_USERS_PATH;
-  if (opts.prefix && opts.folderPerInstance && opts.prefixSettings) {
-    path = `${opts.prefix}/${INSTANCE_USERS_PATH}`;
-  }
-
   try {
-    localUsers = (await yamlParseFile(path)) as ExportedUser[];
+    localUsers = (await yamlParseFile("instance_users.yaml")) as ExportedUser[];
   } catch {
-    log.warn(`No ${path} file found`);
+    log.warn("No instance_users.yaml file found");
   }
   return localUsers;
 }
 
-const INSTANCE_GROUPS_PATH = "instance_groups.yaml";
-
-export async function readInstanceGroups(
-  opts: GlobalOptions & InstanceSyncOptions
-) {
+export async function readInstanceGroups() {
   let localGroups: InstanceGroup[] = [];
-
-  let path = INSTANCE_GROUPS_PATH;
-  if (opts.prefix && opts.folderPerInstance && opts.prefixSettings) {
-    path = `${opts.prefix}/${INSTANCE_GROUPS_PATH}`;
-  }
-
   try {
-    localGroups = (await yamlParseFile(path)) as ExportedInstanceGroup[];
+    localGroups = (await yamlParseFile(
+      "instance_groups.yaml"
+    )) as ExportedInstanceGroup[];
   } catch {
-    log.warn(`No ${path} file found`);
+    log.warn("No instance_groups.yaml file found");
   }
   return localGroups;
 }
 
-export async function pushInstanceUsers(
-  opts: GlobalOptions & InstanceSyncOptions,
-  preview: boolean = false
-) {
+export async function pushInstanceUsers(preview: boolean = false) {
   const remoteUsers = await wmill.globalUsersExport();
-  const localUsers: ExportedUser[] = await readInstanceUsers(opts);
+  const localUsers: ExportedUser[] = await readInstanceUsers();
 
   if (preview) {
     return compareInstanceObjects(localUsers, remoteUsers, "email", "user");
@@ -464,33 +440,27 @@ export async function pushInstanceUsers(
   }
 }
 
-export async function pullInstanceGroups(
-  opts: GlobalOptions & InstanceSyncOptions,
-  preview = false
-) {
+export async function pullInstanceGroups(preview = false) {
   const remoteGroups = await wmill.exportInstanceGroups();
 
   if (preview) {
-    const localGroups = await readInstanceGroups(opts);
+    const localGroups = await readInstanceGroups();
     return compareInstanceObjects(remoteGroups, localGroups, "name", "group");
   } else {
     log.info("Pulling groups from instance...");
 
     await Deno.writeTextFile(
-      INSTANCE_GROUPS_PATH,
+      "instance_groups.yaml",
       yamlStringify(remoteGroups as any)
     );
 
-    log.info(colors.green(`Groups written to ${INSTANCE_GROUPS_PATH}`));
+    log.info(colors.green("Groups written to instance_groups.yaml"));
   }
 }
 
-export async function pushInstanceGroups(
-  opts: GlobalOptions & InstanceSyncOptions,
-  preview: boolean = false
-) {
+export async function pushInstanceGroups(preview: boolean = false) {
   const remoteGroups = await wmill.exportInstanceGroups();
-  const localGroups = await readInstanceGroups(opts);
+  const localGroups = await readInstanceGroups();
 
   if (preview) {
     return compareInstanceObjects(localGroups, remoteGroups, "name", "group");

frontend/.gitignore

@@ -9,4 +9,5 @@ tests-out/
 storageState.json
 .env.production
 dist/
-static/tsdocs/
+static/tsdocs/
+!build/.gitkeep

frontend/package-lock.json

@@ -1,12 +1,12 @@
 {
 	"name": "windmill-components",
-	"version": "1.417.0",
+	"version": "1.416.2",
 	"lockfileVersion": 3,
 	"requires": true,
 	"packages": {
 		"": {
 			"name": "windmill-components",
-			"version": "1.417.0",
+			"version": "1.416.2",
 			"license": "AGPL-3.0",
 			"dependencies": {
 				"@aws-crypto/sha256-js": "^4.0.0",

frontend/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "windmill-components",
-	"version": "1.417.0",
+	"version": "1.416.2",
 	"scripts": {
 		"dev": "vite dev",
 		"build": "vite build",

frontend/src/lib/components/ApiConnectForm.svelte

@@ -80,8 +80,9 @@
 	let connectionString = ''
 	let validConnectionString = true
 	function parseConnectionString(close: (_: any) => void) {
+		// parse postgres connection string
 		const regex =
-			/postgres(?:ql)?:\/\/(?<user>[^:@]+)(?::(?<password>[^@]+))?@(?<host>[^:\/?]+)(?::(?<port>\d+))?\/(?<dbname>[^\?]+)?(?:\?.*sslmode=(?<sslmode>[^&]+))?/
+			/postgres:\/\/(?<user>[^:@]+)(?::(?<password>[^@]+))?@(?<host>[^:\/?]+)(?::(?<port>\d+))?\/(?<dbname>[^\?]+)?(?:\?.*sslmode=(?<sslmode>[^&]+))?/
 		const match = connectionString.match(regex)
 		if (match) {
 			validConnectionString = true

frontend/src/lib/components/DeployWorkspace.svelte

@@ -533,7 +533,7 @@
 	>
 {:else if notSet == true}
 	<Alert type="error" title="Staging/Prod deploy not set up"
-		>As an admin, go to Settings {'->'} Workspace {'->'} Deployment UI</Alert
+		>As an admin, go to "Workspace {'->'} Dev/Staging/Prod"</Alert
 	>
 {:else}
 	<Alert type="info" title="Shareable page"

frontend/src/lib/components/common/table/AppRow.svelte

@@ -16,9 +16,9 @@
 		Eye,
 		File,
 		FileJson,
-		FolderOpen,
+		FileUp,
 		GitFork,
-		ChevronUpSquare,
+		Globe,
 		History,
 		Pen,
 		Share,
@@ -162,7 +162,7 @@
 					},
 					{
 						displayName: 'Move/Rename',
-						icon: FolderOpen,
+						icon: FileUp,
 						action: () => {
 							moveDrawer.openDrawer(path, summary, 'app')
 						},
@@ -173,7 +173,7 @@
 						? [
 								{
 									displayName: 'Deploy to staging/prod',
-									icon: ChevronUpSquare,
+									icon: Globe,
 									action: () => {
 										deploymentDrawer.openDrawer(path, 'app')
 									},

frontend/src/lib/components/common/table/FlowRow.svelte

@@ -23,8 +23,8 @@
 		GitFork,
 		Trash,
 		List,
-		FolderOpen,
-		ChevronUpSquare,
+		FileUp,
+		Globe,
 		Calendar,
 		Share,
 		Archive,
@@ -179,7 +179,7 @@
 					},
 					{
 						displayName: 'Move/Rename',
-						icon: FolderOpen,
+						icon: FileUp,
 						action: () => {
 							moveDrawer.openDrawer(path, flow.summary, 'flow')
 						},
@@ -197,7 +197,7 @@
 						? [
 								{
 									displayName: 'Deploy to staging/prod',
-									icon: ChevronUpSquare,
+									icon: Globe,
 									action: () => {
 										deploymentDrawer.openDrawer(path, 'flow')
 									},

frontend/src/lib/components/common/table/RawAppRow.svelte

@@ -14,7 +14,7 @@
 	import FileInput from '../fileInput/FileInput.svelte'
 	import { goto } from '$lib/navigation'
 	import type DeployWorkspaceDrawer from '$lib/components/DeployWorkspaceDrawer.svelte'
-	import { FolderOpen, Globe, Pen, Share, Trash } from 'lucide-svelte'
+	import { FileUp, Globe, Pen, Share, Trash } from 'lucide-svelte'
 	import { isDeployable } from '$lib/utils_deployable'
 	import { getDeployUiSettings } from '$lib/components/home/deploy_ui'
 
@@ -94,7 +94,7 @@
 				return [
 					{
 						displayName: 'Move/Rename',
-						icon: FolderOpen,
+						icon: FileUp,
 						action: () => {
 							moveDrawer.openDrawer(path, summary, 'raw_app')
 						},

frontend/src/lib/components/common/table/ScriptRow.svelte

@@ -27,8 +27,7 @@
 		Code,
 		Copy,
 		Eye,
-		FolderOpen,
-		ChevronUpSquare,
+		FileUp,
 		GitFork,
 		List,
 		Pen,
@@ -208,7 +207,7 @@
 					},
 					{
 						displayName: 'Move/Rename',
-						icon: FolderOpen,
+						icon: FileUp,
 						action: () => {
 							moveDrawer.openDrawer(script.path, script.summary, 'script')
 						},
@@ -219,7 +218,7 @@
 						? [
 								{
 									displayName: 'Deploy to staging/prod',
-									icon: ChevronUpSquare,
+									icon: FileUp,
 									action: () => {
 										deploymentDrawer.openDrawer(script.path, 'script')
 									},

frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte

@@ -27,7 +27,7 @@
 		FolderOpen,
 		Archive,
 		Trash,
-		ChevronUpSquare,
+		Server,
 		Share,
 		Badge,
 		Loader2,
@@ -294,7 +294,7 @@
 			menuItems.push({
 				label: 'Deploy to staging/prod',
 				onclick: () => deploymentDrawer.openDrawer(flow?.path ?? '', 'flow'),
-				Icon: ChevronUpSquare
+				Icon: Server
 			})
 		}
 

frontend/src/routes/(root)/(logged)/scripts/get/[...hash]/+page.svelte

@@ -57,7 +57,7 @@
 		History,
 		Loader2,
 		Pen,
-		ChevronUpSquare,
+		Server,
 		Share,
 		Table2,
 		Trash,
@@ -407,7 +407,7 @@
 		if (isDeployable('script', script?.path ?? '', deployUiSettings)) {
 			menuItems.push({
 				label: 'Deploy to staging/prod',
-				Icon: ChevronUpSquare,
+				Icon: Server,
 				onclick: () => {
 					deploymentDrawer.openDrawer(script?.path ?? '', 'script')
 				}

lsp/Pipfile

@@ -4,8 +4,8 @@ verify_ssl = true
 name = "pypi"
 
 [packages]
-wmill = ">=1.417.0"
-wmill_pg = ">=1.417.0"
+wmill = ">=1.416.2"
+wmill_pg = ">=1.416.2"
 sendgrid = "*"
 mysql-connector-python = "*"
 pymongo = "*"

openflow.openapi.yaml

@@ -1,7 +1,7 @@
 openapi: "3.0.3"
 
 info:
-  version: 1.417.0
+  version: 1.416.2
   title: OpenFlow Spec
   contact:
     name: Ruben Fiszel

powershell-client/WindmillClient/WindmillClient.psd1

@@ -12,7 +12,7 @@
 RootModule = 'WindmillClient.psm1'
 
 # Version number of this module.
-ModuleVersion = '1.417.0'
+ModuleVersion = '1.416.2'
 
 # Supported PSEditions
 # CompatiblePSEditions = @()

python-client/wmill/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "wmill"
-version = "1.417.0"
+version = "1.416.2"
 description = "A client library for accessing Windmill server wrapping the Windmill client API"
 license = "Apache-2.0"
 homepage = "https://windmill.dev"

python-client/wmill_pg/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "wmill-pg"
-version = "1.417.0"
+version = "1.416.2"
 description = "An extension client for the wmill client library focused on pg"
 license = "Apache-2.0"
 homepage = "https://windmill.dev"

typescript-client/jsr.json

@@ -1,6 +1,6 @@
 {
   "name": "@windmill/windmill",
-  "version": "1.417.0",
+  "version": "1.416.2",
   "exports": "./src/index.ts",
   "publish": {
     "exclude": ["!src", "./s3Types.ts", "./client.ts"]

typescript-client/package.json

@@ -1,7 +1,7 @@
 {
   "name": "windmill-client",
   "description": "Windmill SDK client for browsers and Node.js",
-  "version": "1.417.0",
+  "version": "1.416.2",
   "author": "Ruben Fiszel",
   "license": "Apache 2.0",
   "devDependencies": {

version.txt

@@ -1 +1 @@
-1.417.0
+1.416.2