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

refactor: extract windmill-dep-map crate for parallel api/worker compilation (#7846)

Browse Source 
* refactor: extract windmill-dep-map crate for parallel api/worker compilation

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: resolve WebhookShared type mismatch and missing enterprise propagation

- Make windmill-api webhook_util re-export from windmill-common instead of
  duplicating types, fixing Extension<WebhookShared> mismatch between
  windmill-store and windmill-api
- Add windmill-api-jobs/enterprise to windmill-trigger enterprise feature
  so check_license_key_valid is available when trigger subcrates enable
  enterprise on windmill-trigger

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: stop trigger features from unconditionally enabling enterprise

Move enterprise propagation for all trigger subcrates from individual
trigger feature definitions to the enterprise feature itself, so
enterprise is only enabled when explicitly requested.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: remove unused pub use re-exports and disable CI cargo cache

- Remove unused re-exports from windmill-worker/src/lib.rs:
  trigger_dependents_to_recompute_dependencies, handle_job_error,
  and unused bun/otel items
- Fix callers to use direct module paths instead
- Add windmill-dep-map as dev-dependency for tests
- Disable cargo cache in backend-check CI (faster from-scratch builds)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: restore bun re-exports used by tests

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* all

* chore: re-enable cargo cache for check_ee_full CI job

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-02-08 01:39:56 +01:00
committed by GitHub
parent 39e0265389
commit df6d081ec0
57 changed files with 1953 additions and 1334 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/backend-check.yml
+3 -3
View File
@@ -19,7 +19,7 @@ jobs:
- uses: actions-rust-lang/setup-rust-toolchain@v1
with:
cache-workspaces: backend
cache: false
toolchain: 1.90.0
- name: cargo check
working-directory: ./backend
@@ -40,7 +40,7 @@ jobs:
- uses: actions-rust-lang/setup-rust-toolchain@v1
with:
cache-workspaces: backend
cache: false
toolchain: 1.90.0
- name: cargo check
working-directory: ./backend
@@ -74,7 +74,7 @@ jobs:
- uses: actions-rust-lang/setup-rust-toolchain@v1
with:
cache-workspaces: backend
cache: false
toolchain: 1.90.0
- name: cargo check
working-directory: ./backend
backend/.sqlx/query-5bec60e207a5933aa301f87c0afaeaa8e9a3a2c64e23ab42100d48039ba422b0.json → backend/.sqlx/query-6c10c38a5d5e560d8d76159ce0ae118cf7191b06d72285e71256dd6d70c19451.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO workspace_dependencies(name, workspace_id, content, language, description)\n VALUES ($1, $2, $3, $4, $5) \n RETURNING id\n ",
"query": "\n INSERT INTO workspace_dependencies(name, workspace_id, content, language, description)\n VALUES ($1, $2, $3, $4, $5)\n RETURNING id\n ",
"describe": {
"columns": [
{
@@ -53,5 +53,5 @@
false
]
},
"hash": "5bec60e207a5933aa301f87c0afaeaa8e9a3a2c64e23ab42100d48039ba422b0"
"hash": "6c10c38a5d5e560d8d76159ce0ae118cf7191b06d72285e71256dd6d70c19451"
}
backend/.sqlx/query-70659efcf6c06a7aabd2078829c5b07b4c7d0b47ad8ecbcf0766ece62bac36f9.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM ai_agent_memory WHERE workspace_id = $1 AND conversation_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Uuid"
]
},
"nullable": []
},
"hash": "70659efcf6c06a7aabd2078829c5b07b4c7d0b47ad8ecbcf0766ece62bac36f9"
}
backend/.sqlx/query-1e285da98ac08999f0ad489f1b81885d682de0d7c6ff3e09a9fddef8bb682708.json → backend/.sqlx/query-dcc50c70ac8ecbcb0d79ea7ae0cacad6894605fd2fa391c928b35fbc37b680f9.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n importer_path,\n importer_kind::text as \"importer_kind!\", -- sqlx thinks this is nullable somehow, so enfore with !\n array_agg(importer_node_id) as importer_node_ids\n FROM dependency_map \n WHERE workspace_id = $1 AND imported_path = $2\n GROUP BY importer_path, importer_kind\n ",
"query": "\n SELECT\n importer_path,\n importer_kind::text as \"importer_kind!\",\n array_agg(importer_node_id) as importer_node_ids\n FROM dependency_map\n WHERE workspace_id = $1 AND imported_path = $2\n GROUP BY importer_path, importer_kind\n ",
"describe": {
"columns": [
{
@@ -31,5 +31,5 @@
null
]
},
"hash": "1e285da98ac08999f0ad489f1b81885d682de0d7c6ff3e09a9fddef8bb682708"
"hash": "dcc50c70ac8ecbcb0d79ea7ae0cacad6894605fd2fa391c928b35fbc37b680f9"
}
backend/.sqlx/query-37409e147cf69c39dad848b117bdb77654c167a254053bfd3682c7d9add30b6b.json → backend/.sqlx/query-f2bd875385052618533c868bee309de8863a5d10a482f012ee790eda64e88211.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE workspace_dependencies\n SET archived = true \n WHERE archived = false\n AND name IS NOT DISTINCT FROM $1\n AND workspace_id = $2\n AND language = $3\n RETURNING description\n ",
"query": "\n UPDATE workspace_dependencies\n SET archived = true\n WHERE archived = false\n AND name IS NOT DISTINCT FROM $1\n AND workspace_id = $2\n AND language = $3\n RETURNING description\n ",
"describe": {
"columns": [
{
@@ -51,5 +51,5 @@
false
]
},
"hash": "37409e147cf69c39dad848b117bdb77654c167a254053bfd3682c7d9add30b6b"
"hash": "f2bd875385052618533c868bee309de8863a5d10a482f012ee790eda64e88211"
}
backend/Cargo.lock
Generated
+22
View File
@@ -15707,6 +15707,7 @@ dependencies = [
"windmill-api-client",
"windmill-autoscaling",
"windmill-common",
"windmill-dep-map",
"windmill-git-sync",
"windmill-indexer",
"windmill-queue",
@@ -15810,6 +15811,7 @@ dependencies = [
"windmill-audit",
"windmill-autoscaling",
"windmill-common",
"windmill-dep-map",
"windmill-git-sync",
"windmill-indexer",
"windmill-mcp",
@@ -16062,6 +16064,25 @@ dependencies = [
"windmill-parser-ts",
]
[[package]]
name = "windmill-dep-map"
version = "1.628.3"
dependencies = [
"chrono",
"itertools 0.14.0",
"lazy_static",
"serde",
"serde_json",
"sqlx",
"tokio",
"tracing",
"uuid",
"windmill-common",
"windmill-parser-py-imports",
"windmill-parser-ts",
"windmill-queue",
]
[[package]]
name = "windmill-git-sync"
version = "1.628.3"
@@ -16845,6 +16866,7 @@ dependencies = [
"winapi",
"windmill-audit",
"windmill-common",
"windmill-dep-map",
"windmill-git-sync",
"windmill-macros",
"windmill-mcp",
backend/Cargo.toml
+5 -1
View File
@@ -26,6 +26,7 @@ members = [
"./windmill-store",
"./windmill-queue",
"./windmill-worker",
"./windmill-dep-map",
"./windmill-common",
"./windmill-mcp",
"./windmill-audit",
@@ -129,11 +130,12 @@ all_languages = ["python", "deno_core", "rust", "mysql", "oracledb", "duckdb", "
# For windows we have another set of languages enabled
all_languages_windows = ["python", "deno_core", "rust", "mysql", "oracledb", "duckdb", "mssql-winauth", "bigquery", "csharp", "nu", "php", "java"]
# Edition meta-features: shared groups
inline_preview = ["windmill-api/inline_preview"]
oss_core = [
"embedding", "parquet", "openidconnect", "license",
"http_trigger", "zip", "oauth2", "postgres_trigger",
"mqtt_trigger", "websocket", "smtp", "native_trigger",
"static_frontend", "mcp", "bedrock"
"static_frontend", "mcp", "bedrock", "inline_preview"
]
ce_core = ["oss_core", "private"]
ee_core = [
@@ -224,6 +226,7 @@ tikv-jemalloc-ctl = { optional = true, workspace = true }
serde_json.workspace = true
reqwest.workspace = true
windmill-queue = { workspace = true, features = ["failpoints"] }
windmill-dep-map.workspace = true
axum.workspace = true
serde.workspace = true
windmill-api-client.workspace = true
@@ -235,6 +238,7 @@ tempfile.workspace = true
windmill-api = { path = "./windmill-api", default-features = false }
windmill-queue = { path = "./windmill-queue" }
windmill-worker = { path = "./windmill-worker" }
windmill-dep-map = { path = "./windmill-dep-map" }
windmill-common = { path = "./windmill-common", default-features = false }
windmill-audit = { path = "./windmill-audit" }
windmill-git-sync = { path = "./windmill-git-sync" }
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
ca79631be6c311c5300863634c539593f805e51e
a69c11d8279401ede1ad5b54e3678c3efb3d2381
backend/src/monitor.rs
+10 -5
View File
@@ -84,10 +84,11 @@ use windmill_common::{
use windmill_common::{client::AuthedClient, global_settings::APP_WORKSPACED_ROUTE_SETTING};
use windmill_queue::{cancel_job, get_queued_job_v2, SameWorkerPayload};
use windmill_worker::{
handle_job_error, JobCompletedSender, OtelTracingProxySettings, SameWorkerSender,
BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, MAVEN_REPOS,
NO_DEFAULT_MAVEN, NPM_CONFIG_REGISTRY, NUGET_CONFIG, OTEL_TRACING_PROXY_SETTINGS,
PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, POWERSHELL_REPO_PAT, POWERSHELL_REPO_URL,
result_processor::handle_job_error, JobCompletedSender, OtelTracingProxySettings,
SameWorkerSender, BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT,
KEEP_JOB_DIR, MAVEN_REPOS, NO_DEFAULT_MAVEN, NPM_CONFIG_REGISTRY, NUGET_CONFIG,
OTEL_TRACING_PROXY_SETTINGS, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, POWERSHELL_REPO_PAT,
POWERSHELL_REPO_URL,
};
#[cfg(feature = "parquet")]
@@ -2103,7 +2104,11 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b
} else {
let wc = WORKER_CONFIG.read().await;
let config = config.unwrap();
let has_dedicated = config.dedicated_worker.is_some() || config.dedicated_workers.as_ref().is_some_and(|dws| !dws.is_empty());
let has_dedicated = config.dedicated_worker.is_some()
|| config
.dedicated_workers
.as_ref()
.is_some_and(|dws| !dws.is_empty());
if *wc != config || has_dedicated {
if kill_if_change {
if has_dedicated
backend/tests/common/mod.rs
+8 -4
View File
@@ -684,8 +684,10 @@ pub async fn run_deployed_relative_imports(
language,
priority: None,
apply_preprocessor: false,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
concurrency_settings:
windmill_common::runnable_settings::ConcurrencySettings::default(),
debouncing_settings:
windmill_common::runnable_settings::DebouncingSettings::default(),
})
.push(&db2)
.await;
@@ -734,8 +736,10 @@ pub async fn run_preview_relative_imports(
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
concurrency_settings:
windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings:
windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.push(&db2)
.await;
backend/tests/workspace_dependencies.rs
+1 -1
View File
@@ -7,7 +7,7 @@ mod workspace_dependencies {
use sqlx::{Pool, Postgres};
use tokio_stream::StreamExt;
use windmill_common::scripts::ScriptLang;
use windmill_worker::workspace_dependencies::NewWorkspaceDependencies;
use windmill_dep_map::workspace_dependencies::NewWorkspaceDependencies;
mod deps {
pub const REQUIREMENTS_IN: &'static str = "tiny==0.1.3";
// pub const GO_MOD: &'static str = r##"
backend/windmill-api/Cargo.toml
+18 -16
View File
@@ -11,36 +11,37 @@ path = "src/lib.rs"
[features]
default = []
private = ["windmill-audit/private", "windmill-common/private", "windmill-api-auth/private", "windmill-store/private", "windmill-trigger-kafka?/private", "windmill-trigger-postgres?/private", "windmill-trigger-mqtt?/private", "windmill-trigger-websocket?/private", "windmill-trigger-nats?/private", "windmill-trigger-sqs?/private", "windmill-trigger-gcp?/private", "windmill-trigger-email?/private"]
enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker/enterprise", "windmill-api-auth/enterprise", "windmill-store/enterprise", "windmill-api-jobs/enterprise"]
enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker?/enterprise", "windmill-api-auth/enterprise", "windmill-store/enterprise", "windmill-api-jobs/enterprise", "windmill-trigger/enterprise", "windmill-trigger-kafka?/enterprise", "windmill-trigger-postgres?/enterprise", "windmill-trigger-mqtt?/enterprise", "windmill-trigger-websocket?/enterprise", "windmill-trigger-email?/enterprise", "windmill-trigger-nats?/enterprise", "windmill-trigger-sqs?/enterprise", "windmill-trigger-gcp?/enterprise", "windmill-trigger-http?/enterprise", "windmill-native-triggers?/enterprise"]
stripe = []
agent_worker_server = []
inline_preview = ["dep:windmill-worker"]
agent_worker_server = ["dep:windmill-worker"]
enterprise_saml = ["dep:samael", "dep:libxml"]
benchmark = []
embedding = ["windmill-api-embeddings/embedding"]
parquet = ["dep:datafusion", "dep:object_store", "windmill-common/parquet", "windmill-worker/parquet"]
prometheus = ["windmill-common/prometheus", "windmill-queue/prometheus", "dep:prometheus", "windmill-worker/prometheus"]
parquet = ["dep:datafusion", "dep:object_store", "windmill-common/parquet", "windmill-worker?/parquet"]
prometheus = ["windmill-common/prometheus", "windmill-queue/prometheus", "dep:prometheus", "windmill-worker?/prometheus"]
openidconnect = ["dep:openidconnect", "windmill-common/openidconnect", "windmill-store/openidconnect"]
tantivy = ["dep:windmill-indexer"]
kafka = ["dep:windmill-trigger-kafka", "windmill-trigger-kafka/enterprise", "windmill-store/kafka"]
kafka = ["dep:windmill-trigger-kafka", "windmill-store/kafka"]
kafka-gssapi = ["kafka", "windmill-trigger-kafka/kafka-gssapi"]
nats = ["dep:windmill-trigger-nats", "windmill-trigger-nats/enterprise", "windmill-store/nats"]
websocket = ["dep:windmill-trigger-websocket", "windmill-trigger-websocket/enterprise"]
smtp = ["dep:mail-parser", "dep:openssl", "windmill-common/smtp", "dep:windmill-trigger-email", "windmill-trigger-email/enterprise"]
nats = ["dep:windmill-trigger-nats", "windmill-store/nats"]
websocket = ["dep:windmill-trigger-websocket"]
smtp = ["dep:mail-parser", "dep:openssl", "windmill-common/smtp", "dep:windmill-trigger-email"]
license = ["dep:rsa"]
zip = ["dep:async_zip"]
oauth2 = ["dep:windmill-oauth", "windmill-store/oauth2"]
http_trigger = ["dep:matchit", "dep:windmill-trigger-http", "windmill-trigger-http/enterprise", "windmill-store/http_trigger"]
http_trigger = ["dep:matchit", "dep:windmill-trigger-http", "windmill-store/http_trigger"]
static_frontend = ["dep:rust-embed"]
postgres_trigger = ["dep:windmill-trigger-postgres", "windmill-trigger-postgres/enterprise", "windmill-store/postgres_trigger"]
mqtt_trigger = ["dep:windmill-trigger-mqtt", "windmill-trigger-mqtt/enterprise", "windmill-store/mqtt_trigger"]
native_trigger = ["dep:windmill-native-triggers", "windmill-native-triggers/native_trigger", "windmill-native-triggers/enterprise", "dep:strum", "oauth2"]
sqs_trigger = ["dep:windmill-trigger-sqs", "windmill-trigger-sqs/enterprise", "windmill-store/sqs_trigger"]
postgres_trigger = ["dep:windmill-trigger-postgres", "windmill-store/postgres_trigger"]
mqtt_trigger = ["dep:windmill-trigger-mqtt", "windmill-store/mqtt_trigger"]
native_trigger = ["dep:windmill-native-triggers", "windmill-native-triggers/native_trigger", "dep:strum", "oauth2"]
sqs_trigger = ["dep:windmill-trigger-sqs", "windmill-store/sqs_trigger"]
deno_core = ["dep:deno_core", "dep:deno_error"]
gcp_trigger = ["dep:windmill-trigger-gcp", "windmill-trigger-gcp/enterprise", "windmill-store/gcp_trigger"]
gcp_trigger = ["dep:windmill-trigger-gcp", "windmill-store/gcp_trigger"]
cloud = ["windmill-common/cloud", "windmill-api-auth/cloud", "windmill-store/cloud"]
mcp = ["dep:windmill-mcp", "windmill-mcp/server", "windmill-mcp/auth", "windmill-api-auth/mcp", "windmill-store/mcp"]
bedrock = ["dep:aws-sdk-bedrock", "dep:aws-sdk-bedrockruntime", "windmill-common/bedrock", "dep:aws-config"]
python = []
python = ["windmill-dep-map/python"]
no_auth = ["windmill-api-auth/no_auth", "windmill-store/no_auth"]
[dependencies]
@@ -61,7 +62,8 @@ windmill-parser-py-imports.workspace = true
windmill-git-sync.workspace = true
windmill-indexer = { workspace = true, optional = true }
windmill-autoscaling.workspace = true
windmill-worker.workspace = true
windmill-worker = { workspace = true, optional = true }
windmill-dep-map.workspace = true
tokio.workspace = true
tokio-stream.workspace = true
anyhow.workspace = true
backend/windmill-api/src/db.rs
+1
View File
@@ -17,6 +17,7 @@ use tokio::task::JoinHandle;
pub use windmill_common::db::DB;
use windmill_common::{error::Error, utils::generate_lock_id};
#[allow(unused_imports)]
pub use windmill_api_auth::{ApiAuthed, OptJobAuthed};
async fn current_database(conn: &mut PgConnection) -> Result<String, MigrateError> {
backend/windmill-api/src/flow_conversations.rs
+19 -17
View File
@@ -128,24 +128,26 @@ async fn delete_conversation(
tx.commit().await?;
// Delete associated memory in background (non-blocking cleanup)
let w_id_clone = w_id.clone();
let db_clone = db.clone();
tokio::spawn(async move {
if let Err(e) = windmill_worker::memory_oss::delete_conversation_memory(
&db_clone,
&w_id_clone,
conversation_id,
)
.await
{
tracing::error!(
"Failed to delete memory for conversation {} in workspace {}: {:?}",
{
let w_id_clone = w_id.clone();
let db_clone = db.clone();
tokio::spawn(async move {
if let Err(e) = windmill_common::flow_conversations::delete_conversation_memory(
&db_clone,
&w_id_clone,
conversation_id,
w_id_clone,
e
);
}
});
)
.await
{
tracing::error!(
"Failed to delete memory for conversation {} in workspace {}: {:?}",
conversation_id,
w_id_clone,
e
);
}
});
}
Ok(format!("Conversation {} deleted", conversation_id))
}
backend/windmill-api/src/flows.rs
+1 -1
View File
@@ -49,10 +49,10 @@ use windmill_common::{
scripts::Schema,
utils::{http_get_from_hub, not_found_if_none, paginate, Pagination, RunnableKind, StripPath},
};
use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap;
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
use windmill_queue::WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT;
use windmill_queue::{push, schedule::push_scheduled_job, PushIsolationLevel};
use windmill_worker::scoped_dependency_map::ScopedDependencyMap;
pub fn workspaced_service() -> Router {
Router::new()
backend/windmill-api/src/jobs.rs
+18 -2
View File
@@ -29,17 +29,20 @@ use url::Url;
#[cfg(all(feature = "enterprise", feature = "smtp"))]
use windmill_common::auth::is_super_admin_email;
use windmill_common::auth::TOKEN_PREFIX_LEN;
#[cfg(feature = "inline_preview")]
use windmill_common::client::AuthedClient;
use windmill_common::db::UserDbWithAuthed;
use windmill_common::error::JsonResult;
use windmill_common::flow_status::{JobResult, RestartedFrom};
#[cfg(feature = "inline_preview")]
use windmill_common::jobs::RunInlinePreviewScriptFnParams;
use windmill_common::jobs::{
format_completed_job_result, format_result, DynamicInput, RunInlinePreviewScriptFnParams,
ENTRYPOINT_OVERRIDE,
format_completed_job_result, format_result, DynamicInput, ENTRYPOINT_OVERRIDE,
};
use windmill_common::runnable_settings::{
ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, RunnableSettings,
};
#[cfg(feature = "inline_preview")]
use windmill_common::runtime_assets::{register_runtime_asset, InsertRuntimeAssetParams};
use windmill_common::s3_helpers::{upload_artifact_to_store, BundleFormat};
use windmill_common::scripts::ScriptRunnableSettingsInline;
@@ -52,11 +55,14 @@ use windmill_common::workspace_dependencies::{
use windmill_common::DYNAMIC_INPUT_CACHE;
#[cfg(all(feature = "enterprise", feature = "smtp"))]
use windmill_common::{email_oss::send_email_html, server::load_smtp_config};
#[cfg(feature = "inline_preview")]
use windmill_parser::asset_parser::AssetKind;
#[cfg(feature = "inline_preview")]
use windmill_worker::get_worker_internal_server_inline_utils;
use windmill_common::variables::get_workspace_key;
#[cfg(feature = "inline_preview")]
use crate::db::OptJobAuthed;
use crate::triggers::trigger_helpers::{FlowId, ScriptId};
use crate::{
@@ -2857,6 +2863,7 @@ struct Preview {
format: Option<String>,
}
#[cfg(feature = "inline_preview")]
#[derive(Debug, Deserialize)]
struct PreviewInline {
content: String,
@@ -4591,6 +4598,7 @@ async fn run_preview_script(
Ok((StatusCode::CREATED, uuid.to_string()))
}
#[cfg(feature = "inline_preview")]
async fn run_inline_preview_script(
OptJobAuthed { authed, job_id }: OptJobAuthed,
Tokened { token }: Tokened,
@@ -4627,6 +4635,14 @@ async fn run_inline_preview_script(
Ok(Json(to_raw_value(&result)).into_response())
}
#[cfg(not(feature = "inline_preview"))]
async fn run_inline_preview_script() -> error::Result<Response> {
Err(error::Error::InternalErr(
"inline preview requires the worker feature".to_string(),
))
}
#[cfg(feature = "inline_preview")]
fn register_potential_assets_on_inline_execution(
job_id: Uuid,
w_id: &str,
backend/windmill-api/src/scripts.rs
+2 -2
View File
@@ -38,7 +38,8 @@ use sqlx::{FromRow, Postgres, Transaction};
use std::{collections::HashMap, sync::Arc};
use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_worker::{process_relative_imports, scoped_dependency_map::ScopedDependencyMap};
use windmill_dep_map::process_relative_imports;
use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap;
use windmill_common::{
assets::{
@@ -1146,7 +1147,6 @@ async fn create_script_internal<'c>(
let content = ns.content.clone();
let language = ns.language.clone();
tokio::spawn(async move {
// wait for 10 seconds to make sure the script is deployed and that the CLI sync that pushed it (f one) is complete
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
if let Err(e) = process_relative_imports(
&db2,
backend/windmill-api/src/secret_backend_ext.rs
+1 -4
View File
@@ -17,10 +17,7 @@
#[cfg(all(feature = "private", feature = "enterprise"))]
use std::sync::Arc;
use windmill_common::{
db::DB,
error::Result,
};
use windmill_common::{db::DB, error::Result};
#[cfg(all(feature = "private", feature = "enterprise"))]
use windmill_common::error::Error;
backend/windmill-api/src/webhook_util.rs
+1 -154
View File
@@ -1,154 +1 @@
use std::time::Duration;
use quick_cache::sync::Cache;
use serde::Serialize;
use tokio::{select, sync::mpsc};
#[cfg(feature = "prometheus")]
use windmill_common::METRICS_ENABLED;
use crate::db::DB;
use windmill_common::oauth2::InstanceEvent;
use windmill_common::utils::configure_client;
#[cfg(feature = "prometheus")]
lazy_static::lazy_static! {
// TODO: these aren't synced, they should be moved into the queue abstraction once/if that happens.
static ref WEBHOOK_REQUEST_COUNT: prometheus::Histogram = prometheus::register_histogram!(
"webhook_request",
"Histogram of webhook requests made"
)
.unwrap();
}
lazy_static::lazy_static! {
pub static ref INSTANCE_EVENTS_WEBHOOK: Option<String> = std::env::var("INSTANCE_EVENTS_WEBHOOK").ok();
pub static ref WEBHOOK_CACHE: Cache<String, Option<String>> = Cache::new(100);
}
pub enum WebhookPayload {
WorkspaceEvent(String, WebhookMessage),
InstanceEvent(InstanceEvent),
}
#[derive(Serialize)]
#[serde(tag = "type")]
pub enum WebhookMessage {
// See https://serde.rs/enum-representations.html#internally-tagged for how this looks in JSON
CreateApp { workspace: String, path: String },
DeleteApp { workspace: String, path: String },
UpdateApp { workspace: String, old_path: String, new_path: String },
CreateFlow { workspace: String, path: String },
UpdateFlow { workspace: String, old_path: String, new_path: String },
ArchiveFlow { workspace: String, path: String },
DeleteFlow { workspace: String, path: String },
CreateFolder { workspace: String, name: String },
UpdateFolder { workspace: String, name: String },
DeleteFolder { workspace: String, name: String },
DeleteResource { workspace: String, path: String },
CreateResource { workspace: String, path: String },
UpdateResource { workspace: String, old_path: String, new_path: String },
CreateResourceType { name: String },
DeleteResourceType { name: String },
UpdateResourceType { name: String },
CreateScript { workspace: String, path: String, hash: String },
UpdateScript { workspace: String, path: String, hash: String },
DeleteScript { workspace: String, hash: String },
DeleteScriptPath { workspace: String, path: String },
CreateVariable { workspace: String, path: String },
UpdateVariable { workspace: String, old_path: String, new_path: String },
DeleteVariable { workspace: String, path: String },
}
#[derive(Clone)]
pub struct WebhookShared {
pub channel: mpsc::UnboundedSender<WebhookPayload>,
}
impl WebhookShared {
pub fn new(mut shutdown_rx: tokio::sync::broadcast::Receiver<()>, db: DB) -> Self {
let (tx, mut rx) = mpsc::unbounded_channel::<WebhookPayload>();
let _process = tokio::spawn(async move {
let client = configure_client(
reqwest::Client::builder()
.connect_timeout(Duration::from_secs(5))
// TODO: investigate pool timeouts and such if TCP load is high
.timeout(Duration::from_secs(5)),
)
.build()
.unwrap();
loop {
select! {
biased;
_ = shutdown_rx.recv() => break,
r = rx.recv() => match r {
Some(WebhookPayload::WorkspaceEvent(workspace_id, message)) => {
let webhook_opt = match WEBHOOK_CACHE.get(&workspace_id) {
Some(guard) => {
guard
},
None => {
let Ok(mut webhook_opt) =
sqlx::query_scalar!(
"SELECT webhook FROM workspace_settings WHERE workspace_id = $1",
workspace_id
)
.fetch_one(
&db,
)
.await else {
tracing::error!("Webhook Message to send - but cannot get workspace settings! Workspace: {workspace_id}");
continue;
};
if webhook_opt.as_ref().is_some_and(|x| x.is_empty()) {
webhook_opt = None;
}
WEBHOOK_CACHE.insert(workspace_id, webhook_opt.clone());
webhook_opt
}
};
if let Some(url) = webhook_opt {
#[cfg(feature = "prometheus")]
let timer = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None };
tracing::info!("Sending webhook message to {}", url);
let _ = client.post(url).json(&message).send().await;
#[cfg(feature = "prometheus")]
timer.map(|x| x.stop_and_record());
}
},
Some(WebhookPayload::InstanceEvent(event)) => {
#[cfg(feature = "prometheus")]
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None };
let r = client.post(INSTANCE_EVENTS_WEBHOOK.as_ref().unwrap()).json(&event).send().await;
if let Err(e) = r {
tracing::error!("Error sending instance event: {}", e);
}
},
None => break,
},
}
}
});
Self { channel: tx }
}
pub fn send_message(&self, workspace_id: String, message: WebhookMessage) {
let _ = self.channel.send(WebhookPayload::WorkspaceEvent(
workspace_id.clone(),
message,
));
}
pub fn send_instance_event(&self, event: InstanceEvent) {
if INSTANCE_EVENTS_WEBHOOK.is_none() {
return;
}
let _ = self.channel.send(WebhookPayload::InstanceEvent(event));
}
}
pub use windmill_common::webhook::*;
backend/windmill-api/src/workspace_dependencies.rs
+1 -1
View File
@@ -13,7 +13,7 @@ use windmill_common::{
workspace_dependencies::WorkspaceDependencies,
DB,
};
use windmill_worker::workspace_dependencies::{
use windmill_dep_map::workspace_dependencies::{
trigger_dependents_to_recompute_dependencies_in_the_background, NewWorkspaceDependencies,
};
backend/windmill-api/src/workspaces.rs
+2 -2
View File
@@ -54,10 +54,10 @@ use windmill_common::{
oauth2::WORKSPACE_SLACK_BOT_TOKEN_PATH,
utils::{paginate, rd_string, require_admin, Pagination},
};
use windmill_git_sync::{handle_deployment_metadata, handle_fork_branch_creation, DeployedObject};
use windmill_worker::scoped_dependency_map::{
use windmill_dep_map::scoped_dependency_map::{
DependencyDependent, DependencyMap, ScopedDependencyMap,
};
use windmill_git_sync::{handle_deployment_metadata, handle_fork_branch_creation, DeployedObject};
#[cfg(feature = "enterprise")]
use windmill_common::utils::require_admin_or_devops;
backend/windmill-common/src/flow_conversations.rs
+18
View File
@@ -3,6 +3,7 @@ use serde::{Deserialize, Serialize};
use sqlx::{self, FromRow};
use uuid::Uuid;
use crate::db::DB;
use crate::error::Result;
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, sqlx::Type)]
@@ -128,3 +129,20 @@ pub async fn add_message_to_conversation_tx(
Ok(())
}
/// Delete all memory for a conversation from the database
pub async fn delete_conversation_memory(
db: &DB,
workspace_id: &str,
conversation_id: Uuid,
) -> Result<()> {
sqlx::query!(
"DELETE FROM ai_agent_memory WHERE workspace_id = $1 AND conversation_id = $2",
workspace_id,
conversation_id
)
.execute(db)
.await?;
Ok(())
}
backend/windmill-dep-map/Cargo.toml
+28
View File
@@ -0,0 +1,28 @@
[package]
name = "windmill-dep-map"
version.workspace = true
authors.workspace = true
edition.workspace = true
[lib]
name = "windmill_dep_map"
path = "src/lib.rs"
[features]
default = []
python = ["dep:windmill-parser-py-imports"]
[dependencies]
windmill-common = { workspace = true, default-features = false }
windmill-queue.workspace = true
windmill-parser-ts.workspace = true
windmill-parser-py-imports = { workspace = true, optional = true }
sqlx.workspace = true
serde.workspace = true
serde_json.workspace = true
tokio.workspace = true
tracing.workspace = true
lazy_static.workspace = true
chrono.workspace = true
itertools.workspace = true
uuid.workspace = true
backend/windmill-dep-map/src/lib.rs
+203
View File
@@ -0,0 +1,203 @@
pub mod scoped_dependency_map;
pub mod trigger_dependents;
pub mod workspace_dependencies;
use std::collections::HashMap;
use std::path::{Component, Path, PathBuf};
use serde_json::value::RawValue;
use sqlx::types::Json;
use uuid::Uuid;
use windmill_common::error;
use windmill_common::scripts::ScriptLang;
use windmill_common::utils::WarnAfterExt;
use windmill_common::workspace_dependencies::{
WorkspaceDependencies, WorkspaceDependenciesPrefetched,
};
use windmill_parser_ts::parse_expr_for_imports;
fn try_normalize(path: &Path) -> Option<PathBuf> {
let mut ret = PathBuf::new();
for component in path.components() {
match component {
Component::Prefix(..) | Component::RootDir => return None,
Component::CurDir => {}
Component::ParentDir => {
if !ret.pop() {
return None;
}
}
Component::Normal(c) => {
ret.push(c);
}
}
}
Some(ret)
}
fn parse_ts_relative_imports(
raw_code: &str,
script_path: &str,
) -> windmill_common::error::Result<Vec<String>> {
let mut relative_imports = vec![];
let r = parse_expr_for_imports(raw_code, true)?;
for import in r {
let import = import.trim_end_matches(".ts");
if import.starts_with("/") {
relative_imports.push(import.trim_start_matches("/").to_string());
} else if import.starts_with(".") {
let normalized = try_normalize(std::path::Path::new(&format!(
"{}/../{}",
script_path, import
)));
if let Some(normalized) = normalized {
let normalized = normalized.to_str().unwrap().to_string();
relative_imports.push(normalized);
} else {
tracing::error!("error canonicalizing path: {script_path} with import {import}");
}
}
}
Ok(relative_imports)
}
pub fn extract_relative_imports(
raw_code: &str,
script_path: &str,
language: &Option<ScriptLang>,
) -> Option<Vec<String>> {
match language {
#[cfg(feature = "python")]
Some(ScriptLang::Python3) => {
windmill_parser_py_imports::parse_relative_imports(&raw_code, script_path).ok()
}
Some(ScriptLang::Bun) | Some(ScriptLang::Bunnative) | Some(ScriptLang::Deno) => {
parse_ts_relative_imports(&raw_code, script_path).ok()
}
_ => None,
}
}
pub fn extract_referenced_paths(
raw_code: &str,
script_path: &str,
language: Option<ScriptLang>,
) -> Option<Vec<String>> {
let mut referenced_paths = vec![];
if let Some(wk_deps_refs) = language
.and_then(|l| l.extract_workspace_dependencies_annotated_refs(raw_code, script_path))
.map(|r| r.external)
{
let l = language.expect("should be some");
for wk_deps_ref in wk_deps_refs {
if let Some(path) = WorkspaceDependencies::to_path(&Some(wk_deps_ref), l).ok() {
referenced_paths.push(path);
};
}
} else if let (Some(l), true /* Only if it is not blacklisted */) = (
language,
WorkspaceDependenciesPrefetched::is_external_references_permitted(script_path),
) {
// we assume all runnables without annotated dependencies reference default dependencies file.
WorkspaceDependencies::to_path(&None, l)
.ok()
.inspect(|p| referenced_paths.push(p.to_owned()));
}
if let Some(relative_imports) = extract_relative_imports(raw_code, script_path, &language) {
referenced_paths.extend(relative_imports);
}
if referenced_paths.is_empty() {
None
} else {
Some(referenced_paths)
}
}
pub async fn process_relative_imports(
db: &sqlx::Pool<sqlx::Postgres>,
_job_id: Option<Uuid>,
args: Option<&Json<HashMap<String, Box<RawValue>>>>,
w_id: &str,
script_path: &str,
parent_path: Option<String>,
deployment_message: Option<String>,
code: &str,
script_lang: &Option<ScriptLang>,
permissioned_as_email: &str,
created_by: &str,
permissioned_as: &str,
) -> error::Result<()> {
use scoped_dependency_map::ScopedDependencyMap;
use trigger_dependents::trigger_dependents_to_recompute_dependencies;
// TODO: Should be moved into handle_dependency_job body to be more consistent with how flows and apps are handled
{
let mut tx = db.begin().await?;
let mut dependency_map = ScopedDependencyMap::fetch_maybe_rearranged(
&w_id,
script_path,
"script",
&parent_path,
db,
)
.await?;
tx = dependency_map
.patch(
extract_referenced_paths(&code, script_path, *script_lang),
// Ideally should be None, but due to current implementation will use empty string to represent None.
"".into(),
tx,
)
.await?;
dependency_map.dissolve(tx).await.commit().await?;
}
{
let mut already_visited = args
.map(|x| {
x.get("already_visited")
.map(|v| serde_json::from_str::<Vec<String>>(v.get()).ok())
.flatten()
})
.flatten()
.unwrap_or_default();
let importers = ScopedDependencyMap::get_dependents(script_path, w_id, db).await?;
already_visited.push(script_path.to_string());
match tokio::time::timeout(
core::time::Duration::from_secs(60),
Box::pin(trigger_dependents_to_recompute_dependencies(
w_id,
importers,
deployment_message,
parent_path,
permissioned_as_email,
created_by,
permissioned_as,
db,
already_visited,
)),
)
.warn_after_seconds(10)
.await
{
Ok(Err(e)) => {
tracing::error!(%e, "error triggering dependents to recompute dependencies")
}
Err(e) => {
tracing::error!(%e, "triggering dependents to recompute dependencies has timed out")
}
_ => {}
}
}
Ok(())
}
backend/windmill-worker/src/scoped_dependency_map.rs → backend/windmill-dep-map/src/scoped_dependency_map.rs
+12 -41
View File
@@ -11,14 +11,10 @@ use windmill_common::{
use std::collections::HashSet;
use crate::worker_lockfiles::extract_referenced_paths;
// TODO: To be removed in future versions
lazy_static::lazy_static! {
pub static ref WMDEBUG_NO_DMAP_DISSOLVE: bool = std::env::var("WMDEBUG_NO_DMAP_DISSOLVE").is_ok();
}
// TODO: Rename to DependencyRelation
#[derive(Serialize)]
pub struct DependencyMap {
pub workspace_id: Option<String>,
@@ -48,7 +44,7 @@ impl ScopedDependencyMap {
/// Calls DB, however is assumed to be called once per dependency job
/// AND is scoped to smaller subset of data
/// So it is not too expensive
pub(crate) async fn fetch_maybe_rearranged<'a>(
pub async fn fetch_maybe_rearranged<'a>(
w_id: &str,
importer_path: &str,
importer_kind: &str,
@@ -123,7 +119,7 @@ SELECT importer_node_id, imported_path
/// Add missing entries to `dependency_map`
/// Remove matching entries
pub(crate) async fn patch<'c>(
pub async fn patch<'c>(
&mut self,
referenced_paths: Option<Vec<String>>,
node_id: String, // Flow Step/Node ID
@@ -134,7 +130,7 @@ SELECT importer_node_id, imported_path
Ok(tx)
}
pub(crate) async fn patch_tx_ref<'c>(
pub async fn patch_tx_ref<'c>(
&mut self,
// NOTE: Referenced_paths should include all of the paths.
referenced_paths: Option<Vec<String>>,
@@ -150,26 +146,12 @@ SELECT importer_node_id, imported_path
return Ok(());
};
// This does:
// 1. remove all relative imports from relative_imports that ARE tracked in dependency_map
// 2. remove corresponding trackers from dependency_map
//
// After this operation `relative_imports` variable has only untracked imports.
// We will handle those in the next expression.
//
// After all `reduce`'s called ScopedDependencyMap has only extra/orphan imports
// these are going to be clean up by calling [dissolve]
// NOTE: `retain` iterates over vec and remove the ones whose closures returned false.
referenced_paths.retain(|imported_path| {
!self
.to_delete
// As dmap is HashSet, removing is O(1) operation
// thus making entire process very efficient
// NOTE: `remove` returns true if item was removed and false if wasn't.
.remove(&(node_id.to_owned(), imported_path.to_owned()))
});
// As mentioned above, usually this will always be empty.
if !referenced_paths.is_empty() {
tracing::info!("adding missing entries to dependency_map: importer_node_id - {}, importer_kind - {}, new_imported_paths - {:?}",
&node_id,
@@ -197,7 +179,7 @@ SELECT importer_node_id, imported_path
}
/// clean orphan entries from `dependency_map`
pub(crate) async fn dissolve<'a>(
pub async fn dissolve<'a>(
self,
mut tx: sqlx::Transaction<'a, sqlx::Postgres>,
) -> sqlx::Transaction<'a, sqlx::Postgres> {
@@ -210,7 +192,6 @@ SELECT importer_node_id, imported_path
tracing::info!("dissolving dependency_map: {:?}", &self);
// We _could_ shove it into single query, but this query is rarely called AND let's keep it simple for redability.
for (importer_node_id, imported_path) in self.to_delete.into_iter() {
tracing::info!("cleaning orphan entry from dependency_map: importer_kind - {}, imported_path - {}, importer_node_id - {}",
&self.importer_kind,
@@ -218,7 +199,6 @@ SELECT importer_node_id, imported_path
&importer_node_id,
);
// Dissolve MUST succeed. Error in dissolve MUST not block the execution.
if let Err(err) = sqlx::query!(
"
DELETE FROM dependency_map
@@ -265,7 +245,6 @@ SELECT importer_node_id, imported_path
"discovered orphan entry in `dependency_map`. It will be healed automatically, however please report this issue to Windmill Team. It is also advised to rebuild maps in workspace settings in troubleshooting.",
);
// MUST succeed. Error MUST not block the execution.
if let Err(err) = sqlx::query!(
"DELETE FROM dependency_map
WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND
@@ -286,7 +265,7 @@ SELECT importer_node_id, imported_path
tx
}
pub(crate) async fn rebuild_map_unchecked<'c>(
pub async fn rebuild_map_unchecked(
w_id: &str,
db: &sqlx::Pool<sqlx::Postgres>,
) -> Result<String> {
@@ -305,7 +284,7 @@ SELECT importer_node_id, imported_path
tx = dmap
.patch(
extract_referenced_paths(&sd.code, &r.path, smd.language),
crate::extract_referenced_paths(&sd.code, &r.path, smd.language),
"".into(),
tx,
)
@@ -318,18 +297,13 @@ SELECT importer_node_id, imported_path
}
// Fetch only top level versions and paths
// It is not fetching value
tracing::info!(workspace_id = w_id, "Rebuilding dependency map for flows");
for r in sqlx::query!("SELECT path, versions[array_upper(versions, 1)] as version FROM flow WHERE workspace_id = $1 AND archived = false", w_id).fetch_all(db).await? {
if let Some(version) = r.version {
// To reduce stress on db try to fetch from cache
// Since our flow versions are immutable it is safe to assume if we have cache for specific version/id it is up to date.
let flow_data = cache::flow::fetch_version(&db.clone().into(), version).await?;
// Create map for specific flow
let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "flow", db).await?;
// Traverse retrieved flow modules
let mut tx = db.begin().await?;
let mut to_process = vec![];
let mut modules_to_check = flow_data.flow.modules.iter().collect::<Vec<_>>();
@@ -342,10 +316,9 @@ SELECT importer_node_id, imported_path
FlowValue::traverse_leafs(modules_to_check, &mut |fmv, id| {
match fmv {
// Since we fetched from flow_version it is safe to assume all inline scripts are in form of RawScript.
FlowModuleValue::RawScript { content, language, .. } => {
to_process.push((
extract_referenced_paths(
crate::extract_referenced_paths(
content,
&(r.path.clone() + "/flow"),
Some(*language),
@@ -353,9 +326,7 @@ SELECT importer_node_id, imported_path
id.clone(),
));
}
// But just in case we will also handle other cases.
FlowModuleValue::FlowScript { .. } => {
// Abort will cancel transaction.
return Err(Error::internal_err("FlowScript is not supposed to be in flow."));
}
_ => {}
@@ -382,7 +353,6 @@ SELECT importer_node_id, imported_path
tracing::info!(workspace_id = w_id, "Rebuilding dependency map for apps");
for r in sqlx::query!("SELECT path, versions[array_upper(versions, 1)] as version FROM app WHERE workspace_id = $1", w_id).fetch_all(db).await? {
if let Some(version) = r.version {
// TODO: Use cache when implemented.
let value = sqlx::query_scalar!(
"SELECT value FROM app_version WHERE id = $1 LIMIT 1",
version
@@ -395,7 +365,7 @@ SELECT importer_node_id, imported_path
let mut to_process = vec![];
traverse_app_inline_scripts(&value, None, &mut |ais, id| {
to_process.push((
extract_referenced_paths(
crate::extract_referenced_paths(
&ais.content,
&(r.path.clone() + "/app"),
ais.language,
@@ -423,6 +393,7 @@ SELECT importer_node_id, imported_path
Ok("Success".into())
}
/// Run if you want to rebuild maps on specific workspace.
/// Potentially takes much time
pub async fn rebuild_map(w_id: &str, db: &sqlx::Pool<sqlx::Postgres>) -> Result<String> {
@@ -459,11 +430,11 @@ SELECT importer_node_id, imported_path
sqlx::query_as!(
DependencyDependent,
r#"
SELECT
SELECT
importer_path,
importer_kind::text as "importer_kind!", -- sqlx thinks this is nullable somehow, so enfore with !
importer_kind::text as "importer_kind!",
array_agg(importer_node_id) as importer_node_ids
FROM dependency_map
FROM dependency_map
WHERE workspace_id = $1 AND imported_path = $2
GROUP BY importer_path, importer_kind
"#,
backend/windmill-dep-map/src/trigger_dependents.rs
+220
View File
@@ -0,0 +1,220 @@
use std::collections::HashMap;
use chrono::{Duration, Utc};
use itertools::Itertools;
use serde_json::value::RawValue;
use windmill_common::error;
use windmill_common::jobs::JobPayload;
use windmill_common::runnable_settings::DebouncingSettings;
use windmill_common::scripts::ScriptHash;
use windmill_common::worker::to_raw_value;
use windmill_queue::PushIsolationLevel;
use crate::scoped_dependency_map::{DependencyDependent, ScopedDependencyMap};
lazy_static::lazy_static! {
static ref DEPENDENCY_JOB_DEBOUNCE_DELAY: usize = std::env::var("DEPENDENCY_JOB_DEBOUNCE_DELAY").ok().and_then(|flag| flag.parse().ok()).unwrap_or(
if cfg!(test) { 15 } else { 5 }
);
}
pub async fn trigger_dependents_to_recompute_dependencies(
w_id: &str,
importers: Vec<DependencyDependent>,
deployment_message: Option<String>,
parent_path: Option<String>,
email: &str,
created_by: &str,
permissioned_as: &str,
db: &sqlx::Pool<sqlx::Postgres>,
already_visited: Vec<String>,
) -> error::Result<()> {
tracing::debug!(
"Triggering dependents to recompute dependencies: {}",
importers.iter().map(|dd| &dd.importer_path).join(",")
);
for DependencyDependent { importer_path, importer_kind, importer_node_ids } in importers.iter()
{
tracing::trace!("Processing dependency: {:?}", importer_path);
if already_visited.contains(importer_path) {
tracing::trace!("Skipping already visited dependency");
continue;
}
let mut tx = db.clone().begin().await?;
let mut args: HashMap<String, Box<RawValue>> = HashMap::new();
if let Some(ref dm) = deployment_message {
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
if let Some(ref p_path) = parent_path {
args.insert("common_dependency_path".to_string(), to_raw_value(&p_path));
}
args.insert(
"already_visited".to_string(),
to_raw_value(&already_visited),
);
args.insert(
"triggered_by_relative_import".to_string(),
to_raw_value(&true),
);
let mut debouncing_settings = DebouncingSettings {
debounce_key: Some(format!("{w_id}:{importer_path}:dependency")),
debounce_delay_s: Some(5),
..Default::default()
};
let job_payload = match importer_kind.as_str() {
"script" => match sqlx::query_scalar!(
"SELECT hash FROM script WHERE path = $1 AND workspace_id = $2 AND deleted = false ORDER BY created_at DESC LIMIT 1",
importer_path,
w_id
)
.fetch_optional(&mut *tx)
.await?
{
Some(hash) => {
tracing::debug!("newest hash for {} is: {hash}", importer_path);
let info =
windmill_common::get_script_info_for_hash(None, db, w_id, hash).await?;
JobPayload::Dependencies {
path: importer_path.clone(),
hash: ScriptHash(hash),
language: info.language,
dedicated_worker: info.dedicated_worker,
debouncing_settings,
}
}
None => {
ScopedDependencyMap::clear_map_for_item(
importer_path,
w_id,
"script",
tx,
&None,
)
.await
.commit()
.await?;
continue;
}
},
"flow" => match sqlx::query_scalar!(
"SELECT id FROM flow_version WHERE path = $1 AND workspace_id = $2 ORDER BY created_at DESC LIMIT 1",
importer_path,
w_id
)
.fetch_optional(&mut *tx)
.await?
{
Some(version) => {
tracing::debug!("Handling flow dependency update for: {}", importer_path);
args.insert(
"nodes_to_relock".to_string(),
to_raw_value(&importer_node_ids),
);
debouncing_settings.debounce_args_to_accumulate = Some(vec!["nodes_to_relock".into()]);
JobPayload::FlowDependencies {
path: importer_path.clone(),
version,
dedicated_worker: None,
debouncing_settings,
}
}
None => {
ScopedDependencyMap::clear_map_for_item(importer_path, w_id, "flow", tx, &None)
.await
.commit()
.await?;
continue;
}
},
"app" => match sqlx::query_scalar!(
"SELECT id FROM app_version WHERE app_id = (SELECT id FROM app WHERE path = $1 AND workspace_id = $2) ORDER BY created_at DESC LIMIT 1",
importer_path,
w_id
)
.fetch_optional(&mut *tx)
.await?
{
Some(version) => {
tracing::debug!("Handling app dependency update for: {}", importer_path);
args.insert(
"components_to_relock".to_string(),
to_raw_value(importer_node_ids),
);
debouncing_settings.debounce_args_to_accumulate = Some(vec!["components_to_relock".into()]);
JobPayload::AppDependencies { path: importer_path.clone(), version, debouncing_settings }
}
None => {
ScopedDependencyMap::clear_map_for_item(importer_path, w_id, "app", tx, &None)
.await
.commit()
.await?;
continue;
}
},
_ => {
tracing::error!(
"unexpected importer kind: {kind:?} for path {path}",
kind = importer_kind,
path = importer_path
);
continue;
}
};
tracing::debug!("Pushing dependency job for: {}", importer_path);
let (job_uuid, new_tx) = windmill_queue::push(
db,
PushIsolationLevel::Transaction(tx),
&w_id,
job_payload,
windmill_queue::PushArgs { args: &args, extra: None },
&created_by,
email,
permissioned_as.to_string(),
Some("trigger.dependents.to.recompute.dependencies"),
Some(Utc::now() + Duration::seconds(*DEPENDENCY_JOB_DEBOUNCE_DELAY as i64)),
None,
None,
None,
None,
None,
false,
false,
None,
true,
Some("dependency".into()),
None,
None,
None,
None,
false,
None,
None,
None,
)
.await?;
tracing::info!(
"pushed dependency job due to common python path: {job_uuid} for path {path}",
path = importer_path,
);
new_tx.commit().await?;
}
Ok(())
}
backend/windmill-worker/src/workspace_dependencies.rs → backend/windmill-dep-map/src/workspace_dependencies.rs
+5 -124
View File
@@ -5,7 +5,8 @@ use windmill_common::{
};
use crate::{
scoped_dependency_map::ScopedDependencyMap, trigger_dependents_to_recompute_dependencies,
scoped_dependency_map::ScopedDependencyMap,
trigger_dependents::trigger_dependents_to_recompute_dependencies,
};
#[derive(sqlx::FromRow, Clone, Serialize, Deserialize, Hash, Debug)]
@@ -16,7 +17,6 @@ pub struct NewWorkspaceDependencies {
/// If None, will use description of previous version
/// If there is no older versions, will set to default
pub description: Option<String>,
// TODO: Make Option, or optimize it in any other way.
pub content: String,
}
@@ -31,16 +31,12 @@ impl NewWorkspaceDependencies {
metadata: (String, String, String),
db: sqlx::Pool<sqlx::Postgres>,
) -> error::Result<i64> {
// Check if all workers support workspace dependencies feature
windmill_common::workspace_dependencies::min_version_supports_v0_workspace_dependencies()
.await?;
let path = WorkspaceDependencies::to_path(&self.name, self.language)?;
// If it is unnamed then we want to rebuild dependency map. Otherwise trigger dependents to recompute locks will not work
// NOTE: We rebuild first, even before creating new w deps. We want to make sure that if rebuild failed, then no new default workspace dependencies were created.
if self.name.is_none() {
// Check if we already rebuilt the map for this workspace by checking if the setting exists
let setting_name = format!("workspace_dependencies_map_rebuilt:{}", self.workspace_id);
let already_rebuilt =
windmill_common::global_settings::load_value_from_global_settings(
@@ -57,7 +53,6 @@ impl NewWorkspaceDependencies {
);
ScopedDependencyMap::rebuild_map_unchecked(&self.workspace_id, &db).await?;
// Mark as rebuilt by creating the setting
windmill_common::global_settings::set_value_in_global_settings(
&db,
&setting_name,
@@ -80,7 +75,7 @@ impl NewWorkspaceDependencies {
let prev_description = sqlx::query_scalar!(
"
UPDATE workspace_dependencies
SET archived = true
SET archived = true
WHERE archived = false
AND name IS NOT DISTINCT FROM $1
AND workspace_id = $2
@@ -97,7 +92,7 @@ impl NewWorkspaceDependencies {
let new_id = sqlx::query_scalar!(
"
INSERT INTO workspace_dependencies(name, workspace_id, content, language, description)
VALUES ($1, $2, $3, $4, $5)
VALUES ($1, $2, $3, $4, $5)
RETURNING id
",
self.name.clone(),
@@ -141,17 +136,12 @@ pub async fn trigger_dependents_to_recompute_dependencies_in_the_background(
language = ?language,
"waiting for cache timeout after creating first unnamed workspace dependencies"
);
// Wait for cache timeout.
// For context, workers have cache on whether the unnamed workspace dependencies exists or not.
// when we trigger dependents to recompoute dependencies we want to make sure all workers are having cache timed out.
// otherwise it would result into bug, when workers skip fetch of workspace dependencies because they think they don't exist.
tokio::time::sleep(EXISTS_CACHE_TIMEOUT).await;
}
// It's ok to fail, it will return an error and user will get notified that they should redeploy workspace dependencies
if let Err(e) = trigger_dependents_to_recompute_dependencies(
&workspace_id,
match crate::scoped_dependency_map::ScopedDependencyMap::get_dependents(
match ScopedDependencyMap::get_dependents(
path.as_str(),
&workspace_id,
&db,
@@ -189,114 +179,5 @@ pub async fn trigger_dependents_to_recompute_dependencies_in_the_background(
});
}
// Type aliases for backward compatibility
pub type RawRequirements = WorkspaceDependencies;
pub type NewRawRequirements = NewWorkspaceDependencies;
#[cfg(test)]
mod workspace_dependencies_tests {
// // TODO: test all cases when it should reject.
// #[cfg(feature = "python")]
// mod new_workspace_dependencies {
// use windmill_common::scripts::ScriptLang;
// use crate::workspace_dependencies::NewWorkspaceDependencies;
// #[sqlx::test(
// fixtures("../../tests/fixtures/base.sql",),
// migrations = "../migrations"
// )]
// async fn test_create(db: sqlx::Pool<sqlx::Postgres>) -> anyhow::Result<()> {
// assert_eq!(
// NewWorkspaceDependencies {
// workspace_id: "test-workspace".into(),
// language: ScriptLang::Python3,
// name: None,
// description: None,
// content: "global:rev1".to_owned(),
// }
// .create("", "", "", &db)
// .await
// .unwrap(),
// 1
// );
// assert_eq!(
// NewWorkspaceDependencies {
// workspace_id: "test-workspace".into(),
// language: ScriptLang::Python3,
// name: Some("rrs1".to_owned()),
// description: None,
// content: "rrs1:rev1".to_owned(),
// }
// .create("", "", "", &db)
// .await
// .unwrap(),
// 2
// );
// assert!(NewWorkspaceDependencies {
// workspace_id: "test-workspace".into(),
// language: ScriptLang::DuckDb,
// description: None,
// name: None,
// content: "".to_owned(),
// }
// .create("", "", "", &db)
// .await
// .is_err());
// // Will act as redeployment
// assert_eq!(
// NewWorkspaceDependencies {
// workspace_id: "test-workspace".into(),
// language: ScriptLang::Python3,
// description: None,
// name: Some("rrs1".to_owned()),
// content: "rrs1:rev2".to_owned(),
// }
// .create("", "", "", &db)
// .await
// .unwrap(),
// // It will just increment id
// 3
// );
// Ok(())
// }
// #[sqlx::test(
// fixtures("../../tests/fixtures/base.sql",),
// migrations = "../migrations"
// )]
// async fn violate_constraints(db: sqlx::Pool<sqlx::Postgres>) -> anyhow::Result<()> {
// let db = &db;
// let create = |name| {
// sqlx::query_scalar!(
// "
// INSERT INTO workspace_dependencies(name, workspace_id, content, language)
// VALUES ($1, 'test-workspace', 'test', 'python3')
// RETURNING id
// ",
// name
// )
// .fetch_one(db)
// };
// assert_eq!(create(Some("test".to_owned())).await.unwrap(), 1);
// assert_eq!(create(None).await.unwrap(), 2);
// assert!(create(Some("test".to_owned())).await.is_err());
// assert!(create(None).await.is_err());
// assert_eq!(
// sqlx::query_scalar!("SELECT COUNT(*) FROM workspace_dependencies",)
// .fetch_one(db)
// .await
// .unwrap()
// .unwrap(),
// 2
// );
// Ok(())
// }
// }
}
backend/windmill-trigger/Cargo.toml
+1 -1
View File
@@ -10,7 +10,7 @@ path = "src/lib.rs"
[features]
default = []
enterprise = ["windmill-common/enterprise"]
enterprise = ["windmill-common/enterprise", "windmill-api-jobs/enterprise"]
cloud = ["windmill-common/cloud"]
[dependencies]
backend/windmill-worker/Cargo.toml
+2 -1
View File
@@ -32,7 +32,7 @@ dind = ["dep:bollard"]
php = ["dep:windmill-parser-php"]
mysql = ["dep:mysql_async"]
oracledb = ["dep:oracle"]
python = ["dep:windmill-parser-py", "dep:windmill-parser-py-imports"]
python = ["dep:windmill-parser-py", "dep:windmill-parser-py-imports", "windmill-dep-map/python"]
csharp = ["dep:windmill-parser-csharp"]
rust = ["dep:windmill-parser-rust"]
nu = ["dep:windmill-parser-nu"]
@@ -44,6 +44,7 @@ bedrock = ["dep:aws-sdk-bedrockruntime", "windmill-common/bedrock"]
[dependencies]
windmill-queue.workspace = true
windmill-dep-map.workspace = true
windmill-audit.workspace = true # there isn't really a reason for audit-worth actions to happen in the worker.
windmill-common = { workspace = true, default-features = false }
windmill-mcp = { workspace = true, optional = true }
backend/windmill-worker/src/ai/providers/anthropic.rs
+5 -1
View File
@@ -500,7 +500,11 @@ impl QueryBuilder for AnthropicQueryBuilder {
// For Vertex AI, the model is specified in the URL path
// Expected base_url format: https://{region}-aiplatform.googleapis.com/v1/projects/{project}/locations/{location}/publishers/anthropic/models
// We append the model and :streamRawPredict
format!("{}/{}:streamRawPredict", base_url.trim_end_matches('/'), model)
format!(
"{}/{}:streamRawPredict",
base_url.trim_end_matches('/'),
model
)
} else {
format!("{}/messages", base_url)
}
backend/windmill-worker/src/ai/providers/bedrock.rs
+4 -7
View File
@@ -17,13 +17,13 @@ use std::collections::HashMap;
use windmill_common::{client::AuthedClient, error::Error};
// Re-export from shared module for use by other parts of the worker
pub use windmill_common::ai_bedrock::{check_env_credentials, BedrockClient};
use windmill_common::ai_bedrock::{
bedrock_stream_event_is_block_stop, bedrock_stream_event_to_text,
bedrock_stream_event_to_tool_delta, bedrock_stream_event_to_tool_start, build_tool_config,
create_inference_config, format_bedrock_error, openai_messages_to_bedrock,
streaming_tool_calls_to_openai, StreamingToolCall,
};
pub use windmill_common::ai_bedrock::{check_env_credentials, BedrockClient};
// ============================================================================
// Query Builder (Worker-specific orchestration)
@@ -161,9 +161,7 @@ impl BedrockQueryBuilder {
if let Some(processor) = stream_event_processor.as_ref() {
processor
.send(
StreamingEvent::TokenDelta {
content: text_delta,
},
StreamingEvent::TokenDelta { content: text_delta },
&mut events_str,
)
.await?;
@@ -185,9 +183,8 @@ impl BedrockQueryBuilder {
}
// Extract usage from Metadata event
if let aws_sdk_bedrockruntime::types::ConverseStreamOutput::Metadata(
metadata,
) = &event
if let aws_sdk_bedrockruntime::types::ConverseStreamOutput::Metadata(metadata) =
&event
{
if let Some(token_usage) = metadata.usage() {
usage = Some(
backend/windmill-worker/src/ai/providers/openai.rs
+4 -5
View File
@@ -473,8 +473,9 @@ impl QueryBuilder for OpenAIQueryBuilder {
parser.parse_events(response).await?;
// Convert OpenAI Responses usage to TokenUsage
let usage =
parser.usage.map(|u| TokenUsage::new(u.input_tokens, u.output_tokens, u.total_tokens));
let usage = parser
.usage
.map(|u| TokenUsage::new(u.input_tokens, u.output_tokens, u.total_tokens));
Ok(ParsedResponse::Text {
content: if parser.accumulated_content.is_empty() {
@@ -518,9 +519,7 @@ impl QueryBuilder for OpenAIQueryBuilder {
"image_generation_call" => {
if output.status.as_deref() == Some("completed") {
if let Some(ref base64_image) = output.result {
return Ok(ParsedResponse::Image {
base64_data: base64_image.clone(),
});
return Ok(ParsedResponse::Image { base64_data: base64_image.clone() });
}
}
}
backend/windmill-worker/src/ai/providers/openrouter.rs
+1 -3
View File
@@ -117,9 +117,7 @@ impl QueryBuilder for OpenRouterQueryBuilder {
.and_then(|images| images.first())
{
if let Some(base64_data) = image.image_url.url.strip_prefix("data:image/png;base64,") {
return Ok(ParsedResponse::Image {
base64_data: base64_data.to_string(),
});
return Ok(ParsedResponse::Image { base64_data: base64_data.to_string() });
}
}
backend/windmill-worker/src/ai/providers/other.rs
+2 -2
View File
@@ -225,8 +225,8 @@ impl QueryBuilder for OtherQueryBuilder {
}
// Convert OpenAI Chat Completions usage to TokenUsage
let usage =
openai_usage.map(|u| TokenUsage::new(u.prompt_tokens, u.completion_tokens, u.total_tokens));
let usage = openai_usage
.map(|u| TokenUsage::new(u.prompt_tokens, u.completion_tokens, u.total_tokens));
Ok(ParsedResponse::Text {
content: if accumulated_content.is_empty() {
backend/windmill-worker/src/ai/query_builder.rs
+2 -4
View File
@@ -7,10 +7,8 @@ use windmill_queue::MiniPulledJob;
use crate::{
ai::{
providers::{
anthropic::AnthropicQueryBuilder,
google_ai::GoogleAIQueryBuilder,
openai::{OpenAIQueryBuilder},
openrouter::OpenRouterQueryBuilder,
anthropic::AnthropicQueryBuilder, google_ai::GoogleAIQueryBuilder,
openai::OpenAIQueryBuilder, openrouter::OpenRouterQueryBuilder,
other::OtherQueryBuilder,
},
types::*,
backend/windmill-worker/src/ai/tools.rs
+1 -1
View File
@@ -1,4 +1,3 @@
use windmill_common::ai_types::OpenAIToolCall;
use crate::ai::query_builder::StreamEventProcessor;
use crate::ai::types::McpToolSource;
use crate::ai::types::*;
@@ -21,6 +20,7 @@ use mappable_rc::Marc;
use serde_json::value::RawValue;
use std::{collections::HashMap, sync::Arc};
use uuid::Uuid;
use windmill_common::ai_types::OpenAIToolCall;
use windmill_common::flows::InputTransform;
use windmill_common::jobs::JobPayload;
backend/windmill-worker/src/ai/types.rs
+126 -62
View File
@@ -174,10 +174,18 @@ pub struct ProviderResource {
#[serde(default, deserialize_with = "empty_string_as_none")]
pub region: Option<String>,
#[allow(dead_code)]
#[serde(alias = "awsAccessKeyId", default, deserialize_with = "empty_string_as_none")]
#[serde(
alias = "awsAccessKeyId",
default,
deserialize_with = "empty_string_as_none"
)]
pub aws_access_key_id: Option<String>,
#[allow(dead_code)]
#[serde(alias = "awsSecretAccessKey", default, deserialize_with = "empty_string_as_none")]
#[serde(
alias = "awsSecretAccessKey",
default,
deserialize_with = "empty_string_as_none"
)]
pub aws_secret_access_key: Option<String>,
/// Platform for Anthropic API (standard or google_vertex_ai)
#[serde(default)]
@@ -202,10 +210,7 @@ impl ProviderWithResource {
pub async fn get_base_url(&self, db: &DB) -> Result<String, Error> {
self.kind
.get_base_url(
self.resource.base_url.clone(),
db,
)
.get_base_url(self.resource.base_url.clone(), db)
.await
}
@@ -295,8 +300,10 @@ impl TokenUsage {
self.total_tokens = add_option(self.total_tokens, other.total_tokens);
self.cache_read_input_tokens =
add_option(self.cache_read_input_tokens, other.cache_read_input_tokens);
self.cache_write_input_tokens =
add_option(self.cache_write_input_tokens, other.cache_write_input_tokens);
self.cache_write_input_tokens = add_option(
self.cache_write_input_tokens,
other.cache_write_input_tokens,
);
}
}
@@ -704,7 +711,9 @@ impl OpenAPISchema {
if let Some(ref other_definitions) = other.definitions {
let definitions = self.definitions.get_or_insert_with(HashMap::new);
for (key, value) in other_definitions {
definitions.entry(key.clone()).or_insert_with(|| value.clone());
definitions
.entry(key.clone())
.or_insert_with(|| value.clone());
}
}
@@ -861,7 +870,10 @@ mod tests {
schema.make_strict();
assert!(
matches!(schema.additional_properties, Some(AdditionalProperties::Bool(false))),
matches!(
schema.additional_properties,
Some(AdditionalProperties::Bool(false))
),
"Expected additionalProperties to be false"
);
}
@@ -874,33 +886,36 @@ mod tests {
schema.make_strict();
assert!(
matches!(schema.additional_properties, Some(AdditionalProperties::Bool(true))),
matches!(
schema.additional_properties,
Some(AdditionalProperties::Bool(true))
),
"Expected additionalProperties to remain true (user-specified)"
);
}
#[test]
fn test_make_strict_all_properties_required() {
let mut schema = object_schema(vec![
("name", string_schema()),
("age", integer_schema()),
]);
let mut schema = object_schema(vec![("name", string_schema()), ("age", integer_schema())]);
schema.required = Some(vec!["name".to_string()]); // Only name is required initially
schema.make_strict();
let required = schema.required.as_ref().expect("required should be set");
assert!(required.contains(&"name".to_string()), "name should be required");
assert!(required.contains(&"age".to_string()), "age should be required");
assert!(
required.contains(&"name".to_string()),
"name should be required"
);
assert!(
required.contains(&"age".to_string()),
"age should be required"
);
assert_eq!(required.len(), 2, "Should have exactly 2 required fields");
}
#[test]
fn test_make_strict_non_required_becomes_nullable() {
let mut schema = object_schema(vec![
("name", string_schema()),
("age", integer_schema()),
]);
let mut schema = object_schema(vec![("name", string_schema()), ("age", integer_schema())]);
schema.required = Some(vec!["name".to_string()]); // Only name is required
schema.make_strict();
@@ -915,7 +930,10 @@ mod tests {
match &age_prop.r#type {
Some(SchemaType::Multiple(types)) => {
assert!(types.contains(&"integer".to_string()), "Should contain integer");
assert!(
types.contains(&"integer".to_string()),
"Should contain integer"
);
assert!(types.contains(&"null".to_string()), "Should contain null");
}
_ => panic!("Expected age to have multiple types including null"),
@@ -953,12 +971,18 @@ mod tests {
.expect("nested property should exist");
assert!(
matches!(nested_prop.additional_properties, Some(AdditionalProperties::Bool(false))),
matches!(
nested_prop.additional_properties,
Some(AdditionalProperties::Bool(false))
),
"Nested object should have additionalProperties: false"
);
// Nested object should have all properties required
let nested_required = nested_prop.required.as_ref().expect("nested required should be set");
let nested_required = nested_prop
.required
.as_ref()
.expect("nested required should be set");
assert!(nested_required.contains(&"field".to_string()));
}
@@ -975,7 +999,10 @@ mod tests {
let items = schema.items.as_ref().expect("items should exist");
assert!(
matches!(items.additional_properties, Some(AdditionalProperties::Bool(false))),
matches!(
items.additional_properties,
Some(AdditionalProperties::Bool(false))
),
"Array items should have additionalProperties: false"
);
}
@@ -994,7 +1021,10 @@ mod tests {
for (i, variant) in schema.one_of.as_ref().unwrap().iter().enumerate() {
assert!(
matches!(variant.additional_properties, Some(AdditionalProperties::Bool(false))),
matches!(
variant.additional_properties,
Some(AdditionalProperties::Bool(false))
),
"oneOf variant {} should have additionalProperties: false",
i
);
@@ -1015,7 +1045,10 @@ mod tests {
for (i, variant) in schema.any_of.as_ref().unwrap().iter().enumerate() {
assert!(
matches!(variant.additional_properties, Some(AdditionalProperties::Bool(false))),
matches!(
variant.additional_properties,
Some(AdditionalProperties::Bool(false))
),
"anyOf variant {} should have additionalProperties: false",
i
);
@@ -1028,10 +1061,7 @@ mod tests {
let mut defs = HashMap::new();
defs.insert("MyType".to_string(), Box::new(def_schema));
let mut schema = OpenAPISchema {
defs: Some(defs),
..Default::default()
};
let mut schema = OpenAPISchema { defs: Some(defs), ..Default::default() };
schema.make_strict();
@@ -1043,7 +1073,10 @@ mod tests {
.expect("MyType def should exist");
assert!(
matches!(my_type.additional_properties, Some(AdditionalProperties::Bool(false))),
matches!(
my_type.additional_properties,
Some(AdditionalProperties::Bool(false))
),
"$defs schema should have additionalProperties: false"
);
}
@@ -1054,10 +1087,7 @@ mod tests {
let mut definitions = HashMap::new();
definitions.insert("MyType".to_string(), Box::new(def_schema));
let mut schema = OpenAPISchema {
definitions: Some(definitions),
..Default::default()
};
let mut schema = OpenAPISchema { definitions: Some(definitions), ..Default::default() };
schema.make_strict();
@@ -1069,7 +1099,10 @@ mod tests {
.expect("MyType definition should exist");
assert!(
matches!(my_type.additional_properties, Some(AdditionalProperties::Bool(false))),
matches!(
my_type.additional_properties,
Some(AdditionalProperties::Bool(false))
),
"definitions schema should have additionalProperties: false"
);
}
@@ -1168,7 +1201,10 @@ mod tests {
schema.make_strict();
// allOf should be removed
assert!(schema.all_of.is_none(), "allOf should be removed after flattening");
assert!(
schema.all_of.is_none(),
"allOf should be removed after flattening"
);
// Properties should be merged
let props = schema.properties.as_ref().expect("properties should exist");
@@ -1183,7 +1219,10 @@ mod tests {
// Should have additionalProperties: false (from make_strict)
assert!(
matches!(schema.additional_properties, Some(AdditionalProperties::Bool(false))),
matches!(
schema.additional_properties,
Some(AdditionalProperties::Bool(false))
),
"Should have additionalProperties: false"
);
}
@@ -1205,8 +1244,14 @@ mod tests {
// Both name and age should be in required (from merge + make_strict makes all required)
let required = schema.required.as_ref().expect("required should be set");
assert!(required.contains(&"name".to_string()), "name should be required");
assert!(required.contains(&"age".to_string()), "age should be required");
assert!(
required.contains(&"name".to_string()),
"name should be required"
);
assert!(
required.contains(&"age".to_string()),
"age should be required"
);
}
#[test]
@@ -1265,7 +1310,7 @@ mod tests {
let schema2 = OpenAPISchema {
r#type: Some(SchemaType::Single("integer".to_string())),
minimum: Some(5.0), // More restrictive
minimum: Some(5.0), // More restrictive
maximum: Some(100.0),
..Default::default()
};
@@ -1278,8 +1323,16 @@ mod tests {
schema.flatten_all_of();
// Should take the more restrictive minimum (5.0)
assert_eq!(schema.minimum, Some(5.0), "Should have more restrictive minimum");
assert_eq!(schema.maximum, Some(100.0), "Should have maximum from schema2");
assert_eq!(
schema.minimum,
Some(5.0),
"Should have more restrictive minimum"
);
assert_eq!(
schema.maximum,
Some(100.0),
"Should have maximum from schema2"
);
}
#[test]
@@ -1288,15 +1341,10 @@ mod tests {
let mut defs = HashMap::new();
defs.insert("MyType".to_string(), Box::new(def_schema));
let schema_with_defs = OpenAPISchema {
defs: Some(defs),
..Default::default()
};
let schema_with_defs = OpenAPISchema { defs: Some(defs), ..Default::default() };
let mut schema = OpenAPISchema {
all_of: Some(vec![Box::new(schema_with_defs)]),
..Default::default()
};
let mut schema =
OpenAPISchema { all_of: Some(vec![Box::new(schema_with_defs)]), ..Default::default() };
schema.flatten_all_of();
@@ -1343,9 +1391,15 @@ mod tests {
schema.sanitize_for_google();
assert!(schema.schema_url.is_none(), "Root $schema should be removed");
assert!(
schema.schema_url.is_none(),
"Root $schema should be removed"
);
let field = schema.properties.as_ref().unwrap().get("field").unwrap();
assert!(field.schema_url.is_none(), "Nested $schema should be removed");
assert!(
field.schema_url.is_none(),
"Nested $schema should be removed"
);
}
#[test]
@@ -1365,7 +1419,10 @@ mod tests {
schema.sanitize_for_google();
let items = schema.items.as_ref().unwrap();
assert!(items.schema_url.is_none(), "Array items $schema should be removed");
assert!(
items.schema_url.is_none(),
"Array items $schema should be removed"
);
}
#[test]
@@ -1376,15 +1433,16 @@ mod tests {
..Default::default()
};
let mut schema = OpenAPISchema {
one_of: Some(vec![Box::new(variant)]),
..Default::default()
};
let mut schema =
OpenAPISchema { one_of: Some(vec![Box::new(variant)]), ..Default::default() };
schema.sanitize_for_google();
let variant = &schema.one_of.as_ref().unwrap()[0];
assert!(variant.schema_url.is_none(), "oneOf variant $schema should be removed");
assert!(
variant.schema_url.is_none(),
"oneOf variant $schema should be removed"
);
}
#[test]
@@ -1405,9 +1463,15 @@ mod tests {
schema.sanitize_for_google();
assert!(schema.schema_url.is_none(), "Root $schema should be removed");
assert!(
schema.schema_url.is_none(),
"Root $schema should be removed"
);
let my_type = schema.defs.as_ref().unwrap().get("MyType").unwrap();
assert!(my_type.schema_url.is_none(), "$defs schema $schema should be removed");
assert!(
my_type.schema_url.is_none(),
"$defs schema $schema should be removed"
);
}
#[test]
backend/windmill-worker/src/ai_executor.rs
+8 -7
View File
@@ -21,7 +21,7 @@ use windmill_mcp::McpClient;
#[cfg(not(feature = "mcp"))]
use crate::ai::tools::McpClientStub as McpClient;
use windmill_common::{
ai_providers::{AIProvider},
ai_providers::AIProvider,
cache,
client::AuthedClient,
db::DB,
@@ -417,7 +417,7 @@ pub async fn run_agent(
args.provider.get_base_url(db).await?
};
let api_key = args.provider.get_api_key().unwrap_or("");
// Create the query builder for the provider
let query_builder = create_query_builder(&args.provider);
@@ -666,7 +666,10 @@ pub async fn run_agent(
let parsed = if args.provider.kind == AIProvider::AWSBedrock {
#[cfg(feature = "bedrock")]
{
let region = args.provider.get_region().unwrap_or(windmill_common::ai_providers::USE_ENV_REGION);
let region = args
.provider
.get_region()
.unwrap_or(windmill_common::ai_providers::USE_ENV_REGION);
// Use Bedrock SDK via dedicated query builder
crate::ai::providers::bedrock::BedrockQueryBuilder::default()
.execute_request(
@@ -770,10 +773,8 @@ pub async fn run_agent(
.build_request_without_usage(&build_args, client, &job.workspace_id)
.await?;
let retry_resp = build_http_request(retry_body)
.send()
.await
.map_err(|e| {
let retry_resp =
build_http_request(retry_body).send().await.map_err(|e| {
Error::internal_err(format!("Failed to call API on retry: {}", e))
})?;
backend/windmill-worker/src/common.rs
+15 -3
View File
@@ -621,9 +621,21 @@ impl OccupancyMetrics {
// long enough to have meaningful data for that window. Otherwise,
// short-lived workers would report misleadingly high occupancy rates.
(
if elapsed >= 15.0 { Some(total_occupation_15s) } else { None },
if elapsed >= 300.0 { Some(total_occupation_5m) } else { None },
if elapsed >= 1800.0 { Some(total_occupation_30m) } else { None },
if elapsed >= 15.0 {
Some(total_occupation_15s)
} else {
None
},
if elapsed >= 300.0 {
Some(total_occupation_5m)
} else {
None
},
if elapsed >= 1800.0 {
Some(total_occupation_30m)
} else {
None
},
)
} else {
(None, None, None)
backend/windmill-worker/src/csharp_executor.rs
+2 -1
View File
@@ -30,7 +30,8 @@ use crate::{
build_command_with_isolation, check_executor_binary_exists, create_args_and_out_file,
get_reserved_variables, read_result, start_child_process, DEV_CONF_NSJAIL,
},
handle_child::handle_child, get_proxy_envs_for_lang,
get_proxy_envs_for_lang,
handle_child::handle_child,
CSHARP_CACHE_DIR, DISABLE_NSJAIL, DISABLE_NUSER, DOTNET_PATH, HOME_ENV, NSJAIL_PATH,
NUGET_CONFIG, PATH_ENV, TRACING_PROXY_CA_CERT_PATH, TZ_ENV,
};
backend/windmill-worker/src/deno_executor.rs
+9 -4
View File
@@ -7,11 +7,13 @@ use windmill_queue::{append_logs, CanceledBy, MiniPulledJob};
use crate::{
common::{
build_command_with_isolation, create_args_and_out_file, get_reserved_variables, parse_npm_config, read_file, read_result,
start_child_process, OccupancyMetrics, StreamNotifier,
build_command_with_isolation, create_args_and_out_file, get_reserved_variables,
parse_npm_config, read_file, read_result, start_child_process, OccupancyMetrics,
StreamNotifier,
},
get_proxy_envs_for_lang,
handle_child::handle_child,
get_proxy_envs_for_lang, DENO_CACHE_DIR, DENO_PATH, DISABLE_NSJAIL, HOME_ENV, NPM_CONFIG_REGISTRY, PATH_ENV, TZ_ENV,
DENO_CACHE_DIR, DENO_PATH, DISABLE_NSJAIL, HOME_ENV, NPM_CONFIG_REGISTRY, PATH_ENV, TZ_ENV,
};
use windmill_common::client::AuthedClient;
@@ -94,7 +96,10 @@ async fn get_common_deno_proc_envs(
}
// Add proxy envs (including OTEL tracing proxy if enabled for deno)
for (k, v) in get_proxy_envs_for_lang(&ScriptLang::Deno).await.unwrap_or_default() {
for (k, v) in get_proxy_envs_for_lang(&ScriptLang::Deno)
.await
.unwrap_or_default()
{
deno_envs.insert(k.to_string(), v);
}
backend/windmill-worker/src/go_executor.rs
+1 -1
View File
@@ -1,6 +1,6 @@
use crate::{common::MaybeLock, get_proxy_envs_for_lang};
use windmill_common::scripts::ScriptLang;
use std::{collections::HashMap, fs::DirBuilder, process::Stdio};
use windmill_common::scripts::ScriptLang;
use itertools::Itertools;
use serde_json::value::RawValue;
backend/windmill-worker/src/java_executor.rs
+2 -2
View File
@@ -26,8 +26,8 @@ use crate::{
},
handle_child,
universal_pkg_installer::{par_install_language_dependencies_all_at_once, RequiredDependency},
COURSIER_CACHE_DIR, DISABLE_NSJAIL, DISABLE_NUSER, JAVA_CACHE_DIR,
JAVA_REPOSITORY_DIR, MAVEN_REPOS, NO_DEFAULT_MAVEN, NSJAIL_PATH, PATH_ENV, PROXY_ENVS,
COURSIER_CACHE_DIR, DISABLE_NSJAIL, DISABLE_NUSER, JAVA_CACHE_DIR, JAVA_REPOSITORY_DIR,
MAVEN_REPOS, NO_DEFAULT_MAVEN, NSJAIL_PATH, PATH_ENV, PROXY_ENVS,
};
use windmill_common::client::AuthedClient;
backend/windmill-worker/src/js_eval.rs
+14 -9
View File
@@ -329,7 +329,11 @@ async fn handle_full_regex(
// Use .ok() to match deno_core op_get_id behavior: return null for non-existent steps
// instead of throwing an error
let res = authed_client
.get_result_by_id::<Option<Box<RawValue>>>(&by_id.flow_job.to_string(), obj_key, query)
.get_result_by_id::<Option<Box<RawValue>>>(
&by_id.flow_job.to_string(),
obj_key,
query,
)
.await
.ok()
.flatten();
@@ -1393,17 +1397,18 @@ async fn eval_fetch(
// Uses job_id as trace_id so all spans are linked to the job.
// span_id is a placeholder - it gets overwritten by the OTLP handler with the real parent span_id.
#[cfg(all(feature = "private", feature = "enterprise"))]
let otel_context_inject = if crate::DENO_OTEL_INITIALIZED.load(std::sync::atomic::Ordering::SeqCst) {
let trace_id = job_id.as_simple().to_string();
format!(
r#"globalThis.__enterSpan?.({{
let otel_context_inject =
if crate::DENO_OTEL_INITIALIZED.load(std::sync::atomic::Ordering::SeqCst) {
let trace_id = job_id.as_simple().to_string();
format!(
r#"globalThis.__enterSpan?.({{
isRecording: () => true,
spanContext: () => ({{ traceId: "{trace_id}", spanId: "ffffffffffffffff", traceFlags: 1 }})
}});"#
)
} else {
String::new()
};
)
} else {
String::new()
};
#[cfg(not(all(feature = "private", feature = "enterprise")))]
let otel_context_inject = "";
backend/windmill-worker/src/js_eval_parity_tests.rs
+988 -327
View File
File diff suppressed because it is too large Load Diff
backend/windmill-worker/src/js_eval_quickjs.rs
+56 -21
View File
@@ -139,9 +139,7 @@ pub async fn eval_timeout_quickjs(
)
.await
.map_err(|_| {
anyhow::anyhow!(
"The expression evaluation `{expr}` took too long to execute (>10000ms)"
)
anyhow::anyhow!("The expression evaluation `{expr}` took too long to execute (>10000ms)")
})??
}
@@ -328,7 +326,9 @@ fn setup_async_ops<'js>(
.get_resource_value_interpolated::<serde_json::Value>(&path, None)
.await
{
Ok(value) => serde_json::to_string(&value).unwrap_or_else(|_| "null".to_string()),
Ok(value) => {
serde_json::to_string(&value).unwrap_or_else(|_| "null".to_string())
}
Err(e) => format!("{}{}", ERR_PREFIX, e),
}
}
@@ -419,27 +419,40 @@ fn setup_results_proxy<'js>(
Some(jr) => {
// Found in local cache, fetch result by job ID
match jr {
JobResult::SingleJob(job_id) => {
client
.get_completed_job_result::<serde_json::Value>(&job_id.to_string(), None)
.await
.map_err(|e| format!("Failed to fetch result for step '{}': {}", step_id_clone, e))
}
JobResult::SingleJob(job_id) => client
.get_completed_job_result::<serde_json::Value>(
&job_id.to_string(),
None,
)
.await
.map_err(|e| {
format!(
"Failed to fetch result for step '{}': {}",
step_id_clone, e
)
}),
JobResult::ListJob(job_ids) => {
let futs = job_ids.iter().map(|job_id| {
let client = client.clone();
let job_id_str = job_id.to_string();
async move {
client
.get_completed_job_result::<serde_json::Value>(&job_id_str, None)
.get_completed_job_result::<serde_json::Value>(
&job_id_str,
None,
)
.await
}
});
let results: Vec<_> = futures::future::join_all(futs).await;
let collected: Result<Vec<_>, _> = results.into_iter().collect();
collected
.map(serde_json::Value::Array)
.map_err(|e| format!("Failed to fetch results for step '{}': {}", step_id_clone, e))
let collected: Result<Vec<_>, _> =
results.into_iter().collect();
collected.map(serde_json::Value::Array).map_err(|e| {
format!(
"Failed to fetch results for step '{}': {}",
step_id_clone, e
)
})
}
}
}
@@ -449,15 +462,21 @@ fn setup_results_proxy<'js>(
// Use .ok() to match deno_core behavior: return null for non-existent steps
// instead of throwing an error
Ok(client
.get_result_by_id::<serde_json::Value>(&flow_job_id, &step_id_clone, None)
.get_result_by_id::<serde_json::Value>(
&flow_job_id,
&step_id_clone,
None,
)
.await
.ok() // Swallow errors, convert to Option
.unwrap_or(serde_json::Value::Null)) // None -> null
.ok() // Swallow errors, convert to Option
.unwrap_or(serde_json::Value::Null)) // None -> null
}
};
match result {
Ok(value) => serde_json::to_string(&value).unwrap_or_else(|_| "null".to_string()),
Ok(value) => {
serde_json::to_string(&value).unwrap_or_else(|_| "null".to_string())
}
Err(e) => format!("{}{}", ERR_PREFIX, e),
}
}
@@ -577,8 +596,24 @@ fn should_add_return_quickjs(expr: &str) -> bool {
}
let statement_prefixes = [
"const ", "let ", "var ", "if ", "if(", "for ", "for(", "while ", "while(", "switch ",
"switch(", "try ", "try{", "throw ", "function ", "class ", "async ", "await ",
"const ",
"let ",
"var ",
"if ",
"if(",
"for ",
"for(",
"while ",
"while(",
"switch ",
"switch(",
"try ",
"try{",
"throw ",
"function ",
"class ",
"async ",
"await ",
];
for prefix in &statement_prefixes {
backend/windmill-worker/src/lib.rs
+10 -19
View File
@@ -34,18 +34,15 @@ mod global_cache;
mod go_executor;
mod graphql_executor;
mod handle_child;
#[cfg(all(feature = "private", feature = "enterprise"))]
mod otel_tracing_proxy_ee;
mod otel_tracing_proxy_oss;
pub mod job_logger;
#[cfg(feature = "private")]
pub mod job_logger_ee;
mod job_logger_oss;
mod js_eval;
#[cfg(feature = "quickjs")]
pub mod js_eval_quickjs;
#[cfg(test)]
mod js_eval_parity_tests;
#[cfg(feature = "quickjs")]
pub mod js_eval_quickjs;
pub mod memory_common;
#[cfg(feature = "private")]
pub mod memory_ee;
@@ -59,9 +56,13 @@ mod oracledb_executor;
#[cfg(feature = "private")]
pub mod otel_ee;
mod otel_oss;
#[cfg(all(feature = "private", feature = "enterprise"))]
mod otel_tracing_proxy_ee;
mod otel_tracing_proxy_oss;
mod pg_executor;
#[cfg(feature = "php")]
mod php_executor;
mod prepare_deps;
#[cfg(feature = "python")]
mod python_executor;
#[cfg(feature = "python")]
@@ -71,33 +72,23 @@ pub mod result_processor;
mod rust_executor;
mod sanitized_sql_params;
mod schema;
pub mod scoped_dependency_map;
pub mod sql_utils;
mod universal_pkg_installer;
mod prepare_deps;
mod worker;
mod worker_flow;
mod worker_lockfiles;
mod worker_utils;
pub mod workspace_dependencies;
pub use worker::*;
pub use worker_lockfiles::{
process_relative_imports, trigger_dependents_to_recompute_dependencies,
};
#[cfg(all(feature = "private", feature = "enterprise"))]
pub use otel_tracing_proxy_ee::{
set_current_job_context, start_jobs_otel_tracing, TRACING_PROXY_PORT,
};
pub use otel_tracing_proxy_ee::start_jobs_otel_tracing;
#[cfg(all(feature = "private", feature = "enterprise", feature = "deno_core"))]
pub use otel_tracing_proxy_ee::{load_internal_otel_exporter, DENO_OTEL_INITIALIZED, OTLP_COLLECTOR_PORT};
pub use result_processor::handle_job_error;
pub use otel_tracing_proxy_ee::{load_internal_otel_exporter, DENO_OTEL_INITIALIZED};
pub use worker::*;
pub use bun_executor::{
build_loader, compute_bundle_local_and_remote_path, generate_dedicated_worker_wrapper,
get_common_bun_proc_envs, install_bun_lockfile, prebundle_bun_script, prepare_job_dir,
BUN_DEDICATED_WORKER_ARGS, LoaderMode, RELATIVE_BUN_BUILDER, RELATIVE_BUN_LOADER,
LoaderMode, BUN_DEDICATED_WORKER_ARGS, RELATIVE_BUN_BUILDER, RELATIVE_BUN_LOADER,
};
pub use deno_executor::generate_deno_lock;
pub use prepare_deps::run_prepare_deps_cli;
backend/windmill-worker/src/mysql_executor.rs
+4 -1
View File
@@ -396,7 +396,10 @@ fn string_date_to_mysql_date(s: &str) -> mysql_async::Value {
get_capture_by_index(&caps, 1),
get_capture_by_index(&caps, 2),
get_capture_by_index(&caps, 3),
0, 0, 0, 0,
0,
0,
0,
0,
);
}
backend/windmill-worker/src/nu_executor.rs
+2 -2
View File
@@ -16,11 +16,11 @@ use crate::{
build_command_with_isolation, create_args_and_out_file, get_reserved_variables,
read_result, start_child_process, OccupancyMetrics, DEV_CONF_NSJAIL,
},
handle_child, get_proxy_envs_for_lang, DISABLE_NSJAIL, DISABLE_NUSER, NSJAIL_PATH, PATH_ENV,
get_proxy_envs_for_lang, handle_child, DISABLE_NSJAIL, DISABLE_NUSER, NSJAIL_PATH, PATH_ENV,
TRACING_PROXY_CA_CERT_PATH,
};
use windmill_common::scripts::ScriptLang;
use windmill_common::client::AuthedClient;
use windmill_common::scripts::ScriptLang;
const NSJAIL_CONFIG_RUN_NU_CONTENT: &str = include_str!("../nsjail/run.nu.config.proto");
lazy_static::lazy_static! {
backend/windmill-worker/src/python_versions.rs
+13 -3
View File
@@ -22,7 +22,8 @@ use crate::{
common::{start_child_process, OccupancyMetrics},
handle_child::handle_child,
python_executor::{INDEX_CERT, NATIVE_CERT, PYTHON_PATH, UV_PATH},
HOME_ENV, INSTANCE_PYTHON_VERSION, PATH_ENV, PROXY_ENVS, PY_INSTALL_DIR, UV_CACHE_DIR, WIN_ENVS,
HOME_ENV, INSTANCE_PYTHON_VERSION, PATH_ENV, PROXY_ENVS, PY_INSTALL_DIR, UV_CACHE_DIR,
WIN_ENVS,
};
impl From<PyV> for PyVAlias {
@@ -527,9 +528,18 @@ impl PyV {
.env("HOME", HOME_ENV.to_string())
.env("PATH", PATH_ENV.to_string())
.envs(PROXY_ENVS.clone())
.args(["python", "install", &v, "--python-preference=only-managed", "--no-bin"])
.args([
"python",
"install",
&v,
"--python-preference=only-managed",
"--no-bin",
])
// TODO: Do we need these?
.envs([("UV_PYTHON_INSTALL_DIR", PY_INSTALL_DIR), ("UV_CACHE_DIR", UV_CACHE_DIR)])
.envs([
("UV_PYTHON_INSTALL_DIR", PY_INSTALL_DIR),
("UV_CACHE_DIR", UV_CACHE_DIR),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
backend/windmill-worker/src/ruby_executor.rs
+2 -2
View File
@@ -26,7 +26,8 @@ use crate::{
build_command_with_isolation, create_args_and_out_file, get_reserved_variables,
read_result, start_child_process, OccupancyMetrics, DEV_CONF_NSJAIL,
},
handle_child::{self}, get_proxy_envs_for_lang,
get_proxy_envs_for_lang,
handle_child::{self},
universal_pkg_installer::{par_install_language_dependencies_seq, RequiredDependency},
DISABLE_NSJAIL, DISABLE_NUSER, NSJAIL_PATH, PATH_ENV, PROXY_ENVS, RUBY_CACHE_DIR, RUBY_REPOS,
TRACING_PROXY_CA_CERT_PATH,
@@ -51,7 +52,6 @@ const NSJAIL_CONFIG_DOWNLOAD_RUBY_CONTENT: &str =
include_str!("../nsjail/download.ruby.config.proto");
const NSJAIL_CONFIG_LOCK_RUBY_CONTENT: &str = include_str!("../nsjail/lock.ruby.config.proto");
#[allow(dead_code)]
pub(crate) struct JobHandlerInput<'a> {
pub base_internal_url: &'a str,
backend/windmill-worker/src/rust_executor.rs
+7 -6
View File
@@ -19,15 +19,17 @@ use windmill_queue::{append_logs, CanceledBy};
use crate::{
common::{
build_command_with_isolation, check_executor_binary_exists, create_args_and_out_file, get_reserved_variables,
read_result, start_child_process, OccupancyMetrics, DEV_CONF_NSJAIL,
build_command_with_isolation, check_executor_binary_exists, create_args_and_out_file,
get_reserved_variables, read_result, start_child_process, OccupancyMetrics,
DEV_CONF_NSJAIL,
},
get_proxy_envs_for_lang,
handle_child::handle_child,
get_proxy_envs_for_lang, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV,
PROXY_ENVS, RUST_CACHE_DIR, TRACING_PROXY_CA_CERT_PATH, TZ_ENV,
DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV, PROXY_ENVS, RUST_CACHE_DIR,
TRACING_PROXY_CA_CERT_PATH, TZ_ENV,
};
use windmill_common::scripts::ScriptLang;
use windmill_common::client::AuthedClient;
use windmill_common::scripts::ScriptLang;
#[cfg(windows)]
use crate::SYSTEM_ROOT;
@@ -53,7 +55,6 @@ lazy_static::lazy_static! {
static ref RUSTUP_HOME_DEFAULT: String = format!("{}\\.rustup", *HOME_DIR);
}
#[cfg(not(windows))]
lazy_static::lazy_static! {
static ref CARGO_HOME_DEFAULT: String = format!("{}/.cargo", *HOME_DIR);
backend/windmill-worker/src/schema.rs
+4 -3
View File
@@ -2,7 +2,6 @@ use std::collections::HashMap;
use windmill_common::schema::{SchemaValidationRule, SchemaValidator};
use windmill_parser::{MainArgSignature, Typ};
fn make_rules_for_arg_typ(typ: &Typ) -> Vec<SchemaValidationRule> {
let mut rules = vec![];
@@ -66,7 +65,10 @@ fn make_rules_for_arg_typ(typ: &Typ) -> Vec<SchemaValidationRule> {
for prop in &variant.properties {
obj_rules.push((prop.key.to_string(), make_rules_for_arg_typ(&prop.typ)));
}
rules_map.insert(variant.label.to_string(), vec![SchemaValidationRule::IsObject(obj_rules)]);
rules_map.insert(
variant.label.to_string(),
vec![SchemaValidationRule::IsObject(obj_rules)],
);
}
rules.push(SchemaValidationRule::IsOneOf(rules_map))
@@ -94,4 +96,3 @@ pub fn schema_validator_from_main_arg_sig(sig: &MainArgSignature) -> SchemaValid
SchemaValidator { required, rules }
}
backend/windmill-worker/src/snowflake_executor.rs
+34 -22
View File
@@ -119,11 +119,16 @@ async fn poll_snowflake_async_query(
})?;
let status = response.status();
let body = response.text().await.map_err(|e| {
Error::ExecutionErr(format!("error reading poll response body: {}", e))
})?;
let body = response
.text()
.await
.map_err(|e| Error::ExecutionErr(format!("error reading poll response body: {}", e)))?;
tracing::debug!("Snowflake poll response status: {}, body: {}", status, &body[..body.len().min(500)]);
tracing::debug!(
"Snowflake poll response status: {}, body: {}",
status,
&body[..body.len().min(500)]
);
if status == reqwest::StatusCode::ACCEPTED {
// Still running, wait and poll again
@@ -243,13 +248,14 @@ fn do_snowflake_inner<'a>(
let body = raw_response.text().await.map_err(|e| {
Error::ExecutionErr(format!("error reading response body: {}", e))
})?;
let async_resp: SnowflakeAsyncResponse = serde_json::from_str(&body).map_err(|e| {
Error::ExecutionErr(format!(
"error decoding async response: {}. Body preview: {}",
e,
&body[..body.len().min(500)]
))
})?;
let async_resp: SnowflakeAsyncResponse =
serde_json::from_str(&body).map_err(|e| {
Error::ExecutionErr(format!(
"error decoding async response: {}. Body preview: {}",
e,
&body[..body.len().min(500)]
))
})?;
tracing::info!(
"Snowflake statement running asynchronously, polling for completion (handle: {})",
@@ -273,21 +279,27 @@ fn do_snowflake_inner<'a>(
// Handle both sync (200) and async (202) responses
let raw_response = handle_snowflake_result(result).await?;
let status = raw_response.status();
let body = raw_response.text().await.map_err(|e| {
Error::ExecutionErr(format!("error reading response body: {}", e))
})?;
let body = raw_response
.text()
.await
.map_err(|e| Error::ExecutionErr(format!("error reading response body: {}", e)))?;
tracing::debug!("Snowflake response status: {}, body: {}", status, &body[..body.len().min(1000)]);
tracing::debug!(
"Snowflake response status: {}, body: {}",
status,
&body[..body.len().min(1000)]
);
let response = if status == reqwest::StatusCode::ACCEPTED {
// Async execution - need to poll for results
let async_resp: SnowflakeAsyncResponse = serde_json::from_str(&body).map_err(|e| {
Error::ExecutionErr(format!(
"error decoding async response: {}. Body preview: {}",
e,
&body[..body.len().min(500)]
))
})?;
let async_resp: SnowflakeAsyncResponse =
serde_json::from_str(&body).map_err(|e| {
Error::ExecutionErr(format!(
"error decoding async response: {}. Body preview: {}",
e,
&body[..body.len().min(500)]
))
})?;
tracing::info!(
"Snowflake query running asynchronously, polling for results (handle: {})",
backend/windmill-worker/src/worker.rs
+1 -2
View File
@@ -131,12 +131,11 @@ use crate::{
go_executor::handle_go_job,
graphql_executor::do_graphql,
handle_child::SLOW_LOGS,
handle_job_error,
job_logger::NO_LOGS_AT_ALL,
js_eval::{eval_fetch_timeout, transpile_ts},
pg_executor::do_postgresql,
pwsh_executor::handle_powershell_job,
result_processor::{process_result, start_background_processor},
result_processor::{handle_job_error, process_result, start_background_processor},
schema::schema_validator_from_main_arg_sig,
worker_flow::{handle_flow, SchedulePushZombieError},
worker_lockfiles::{
backend/windmill-worker/src/worker_flow.rs
+5 -4
View File
@@ -21,6 +21,7 @@ use crate::{
use anyhow::Context;
use async_once_cell::Lazy;
use backon::{BackoffBuilder, ConstantBuilder, Retryable};
use futures::TryFutureExt;
use mappable_rc::Marc;
use serde::{Deserialize, Serialize};
@@ -30,7 +31,6 @@ use sqlx::types::Json;
use sqlx::{FromRow, Postgres, Transaction};
use tracing::instrument;
use uuid::Uuid;
use backon::{BackoffBuilder, ConstantBuilder, Retryable};
use windmill_common::auth::get_job_perms;
#[cfg(feature = "benchmark")]
use windmill_common::bench::BenchmarkIter;
@@ -68,9 +68,10 @@ use windmill_common::{
use windmill_queue::schedule::get_schedule_opt;
use windmill_queue::{
add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job,
try_schedule_next_job, insert_concurrency_key, interpolate_args,
report_error_to_workspace_handler_or_critical_side_channel, CanceledBy, FlowRunners,
MiniCompletedJob, MiniPulledJob, PushArgs, PushIsolationLevel, SameWorkerPayload, WrappedError,
insert_concurrency_key, interpolate_args,
report_error_to_workspace_handler_or_critical_side_channel, try_schedule_next_job, CanceledBy,
FlowRunners, MiniCompletedJob, MiniPulledJob, PushArgs, PushIsolationLevel, SameWorkerPayload,
WrappedError,
};
use windmill_audit::audit_oss::audit_log;
backend/windmill-worker/src/worker_lockfiles.rs
+7 -423
View File
@@ -1,20 +1,16 @@
use std::borrow::Cow;
use std::collections::HashMap;
use std::fs::{create_dir_all, remove_dir_all};
use std::path::{Component, Path, PathBuf};
#[cfg(feature = "python")]
use crate::ansible_executor::{get_git_repos_lock, AnsibleDependencyLocks};
use crate::scoped_dependency_map::{DependencyDependent, ScopedDependencyMap};
use async_recursion::async_recursion;
use chrono::{Duration, Utc};
use itertools::Itertools;
use serde::Serialize;
use serde_json::value::RawValue;
use serde_json::{from_value, json, Value};
use sha2::Digest;
use sqlx::types::Json;
use tokio::time::timeout;
use uuid::Uuid;
use windmill_common::assets::{
clear_static_asset_usage, insert_static_asset_usage, AssetUsageKind,
@@ -22,17 +18,15 @@ use windmill_common::assets::{
use windmill_common::error::Error;
use windmill_common::error::Result;
use windmill_common::flows::{FlowModule, FlowModuleValue, FlowNodeId};
use windmill_common::jobs::JobPayload;
use windmill_common::runnable_settings::DebouncingSettings;
use windmill_common::min_version::MIN_VERSION_SUPPORTS_DEBOUNCING_V2;
use windmill_common::scripts::ScriptHash;
use windmill_common::utils::WarnAfterExt;
#[cfg(feature = "python")]
use windmill_common::worker::PythonAnnotations;
use windmill_common::min_version::MIN_VERSION_SUPPORTS_DEBOUNCING_V2;
use windmill_common::worker::{to_raw_value, to_raw_value_owned, write_file, Connection};
use windmill_common::workspace_dependencies::{
RawWorkspaceDependencies, WorkspaceDependencies, WorkspaceDependenciesPrefetched,
RawWorkspaceDependencies, WorkspaceDependenciesPrefetched,
};
use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap;
#[cfg(feature = "python")]
use windmill_parser_yaml::AnsibleRequirements;
@@ -44,13 +38,12 @@ use windmill_common::{
scripts::ScriptLang,
DB,
};
pub use windmill_dep_map::{
extract_referenced_paths, extract_relative_imports, process_relative_imports,
};
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
#[cfg(feature = "python")]
use windmill_parser_py_imports::parse_relative_imports;
use windmill_parser_ts::parse_expr_for_imports;
use windmill_queue::{
append_logs, CanceledBy, MiniPulledJob, PushIsolationLevel,
WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT,
append_logs, CanceledBy, MiniPulledJob, WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT,
};
// TODO: To be removed in future versions
@@ -59,9 +52,6 @@ lazy_static::lazy_static! {
static ref WMDEBUG_NO_NEW_FLOW_VERSION_ON_DJ: bool = std::env::var("WMDEBUG_NO_NEW_FLOW_VERSION_ON_DJ").is_ok();
static ref WMDEBUG_NO_NEW_APP_VERSION_ON_DJ: bool = std::env::var("WMDEBUG_NO_NEW_APP_VERSION_ON_DJ").is_ok();
static ref WMDEBUG_NO_COMPONENTS_TO_RELOCK: bool = std::env::var("WMDEBUG_NO_COMPONENTS_TO_RELOCK").is_ok();
static ref DEPENDENCY_JOB_DEBOUNCE_DELAY: usize = std::env::var("DEPENDENCY_JOB_DEBOUNCE_DELAY").ok().and_then(|flag| flag.parse().ok()).unwrap_or(
if cfg!(test) { /* if test we want increased debouncing delay */ 15 } else { 5 }
);
}
use crate::common::{MaybeLock, OccupancyMetrics};
@@ -84,103 +74,6 @@ use crate::{
go_executor::install_go_dependencies,
};
fn try_normalize(path: &Path) -> Option<PathBuf> {
let mut ret = PathBuf::new();
for component in path.components() {
match component {
Component::Prefix(..) | Component::RootDir => return None,
Component::CurDir => {}
Component::ParentDir => {
if !ret.pop() {
return None;
}
}
Component::Normal(c) => {
ret.push(c);
}
}
}
Some(ret)
}
fn parse_ts_relative_imports(raw_code: &str, script_path: &str) -> error::Result<Vec<String>> {
let mut relative_imports = vec![];
let r = parse_expr_for_imports(raw_code, true)?;
for import in r {
let import = import.trim_end_matches(".ts");
if import.starts_with("/") {
relative_imports.push(import.trim_start_matches("/").to_string());
} else if import.starts_with(".") {
let normalized = try_normalize(std::path::Path::new(&format!(
"{}/../{}",
script_path, import
)));
if let Some(normalized) = normalized {
let normalized = normalized.to_str().unwrap().to_string();
relative_imports.push(normalized);
} else {
tracing::error!("error canonicalizing path: {script_path} with import {import}");
}
}
}
Ok(relative_imports)
}
pub fn extract_relative_imports(
raw_code: &str,
script_path: &str,
language: &Option<ScriptLang>,
) -> Option<Vec<String>> {
match language {
#[cfg(feature = "python")]
Some(ScriptLang::Python3) => parse_relative_imports(&raw_code, script_path).ok(),
Some(ScriptLang::Bun) | Some(ScriptLang::Bunnative) | Some(ScriptLang::Deno) => {
parse_ts_relative_imports(&raw_code, script_path).ok()
}
_ => None,
}
}
pub fn extract_referenced_paths(
raw_code: &str,
script_path: &str,
language: Option<ScriptLang>,
) -> Option<Vec<String>> {
let mut referenced_paths = vec![];
if let Some(wk_deps_refs) = language
.and_then(|l| l.extract_workspace_dependencies_annotated_refs(raw_code, script_path))
.map(|r| r.external)
{
let l = language.expect("should be some");
for wk_deps_ref in wk_deps_refs {
if let Some(path) = WorkspaceDependencies::to_path(&Some(wk_deps_ref), l).ok() {
referenced_paths.push(path);
};
}
} else if let (Some(l), true /* Only if it is not blacklisted */) = (
language,
WorkspaceDependenciesPrefetched::is_external_references_permitted(script_path),
) {
// we assume all runnables without annotated dependencies reference default dependencies file.
WorkspaceDependencies::to_path(&None, l)
.ok()
.inspect(|p| referenced_paths.push(p.to_owned()));
}
if let Some(relative_imports) = extract_relative_imports(raw_code, script_path, &language) {
referenced_paths.extend(relative_imports);
}
if referenced_paths.is_empty() {
None
} else {
Some(referenced_paths)
}
}
#[tracing::instrument(level = "trace", skip_all)]
pub async fn handle_dependency_job(
job: &MiniPulledJob,
@@ -364,315 +257,6 @@ fn remove_ansi_codes(s: &str) -> String {
ANSI_REGEX.replace_all(s, "").to_string()
}
pub async fn process_relative_imports(
db: &sqlx::Pool<sqlx::Postgres>,
_job_id: Option<Uuid>,
args: Option<&Json<HashMap<String, Box<RawValue>>>>,
w_id: &str,
script_path: &str,
parent_path: Option<String>,
deployment_message: Option<String>,
code: &str,
script_lang: &Option<ScriptLang>,
permissioned_as_email: &str,
created_by: &str,
permissioned_as: &str,
) -> error::Result<()> {
// TODO: Should be moved into handle_dependency_job body to be more consistent with how flows and apps are handled
{
let mut tx = db.begin().await?;
let mut dependency_map = ScopedDependencyMap::fetch_maybe_rearranged(
&w_id,
script_path,
"script",
&parent_path,
db,
)
.await?;
tx = dependency_map
.patch(
extract_referenced_paths(&code, script_path, *script_lang),
// Ideally should be None, but due to current implementation will use empty string to represent None.
"".into(),
tx,
)
.await?;
dependency_map.dissolve(tx).await.commit().await?;
}
{
let mut already_visited = args
.map(|x| {
x.get("already_visited")
.map(|v| serde_json::from_str::<Vec<String>>(v.get()).ok())
.flatten()
})
.flatten()
.unwrap_or_default();
// TODO: There is a race-condition.
// This can be old version.
// Check lines of code below, you will find that we get the latest version of the script/app/flow
// However the latest version does not necessarily mean that it is finalized.
// Instead we assume that this would be the version we would base on.
// So the script_importers might be behind. Thus some information like nodes_to_relock might be lost.
let importers = crate::scoped_dependency_map::ScopedDependencyMap::get_dependents(
script_path,
w_id,
db,
)
.await?;
already_visited.push(script_path.to_string());
// But currently we will do this extra db call for every script regardless of whether they have relative imports or not
// Script might have no relative imports but still be referenced by someone else.
match timeout(
core::time::Duration::from_secs(60),
Box::pin(trigger_dependents_to_recompute_dependencies(
w_id,
importers,
deployment_message,
parent_path,
permissioned_as_email,
created_by,
permissioned_as,
db,
already_visited,
)),
)
.warn_after_seconds(10)
.await
{
Ok(Err(e)) => {
tracing::error!(%e, "error triggering dependents to recompute dependencies")
}
Err(e) => {
tracing::error!(%e, "triggering dependents to recompute dependencies has timed out")
}
_ => {}
}
}
Ok(())
}
pub async fn trigger_dependents_to_recompute_dependencies(
w_id: &str,
importers: Vec<DependencyDependent>,
// imported_path: &str,
deployment_message: Option<String>,
parent_path: Option<String>,
email: &str,
created_by: &str,
permissioned_as: &str,
db: &sqlx::Pool<sqlx::Postgres>,
already_visited: Vec<String>,
) -> error::Result<()> {
tracing::debug!(
"Triggering dependents to recompute dependencies: {}",
importers.iter().map(|dd| &dd.importer_path).join(",")
);
for DependencyDependent { importer_path, importer_kind, importer_node_ids } in importers.iter()
{
tracing::trace!("Processing dependency: {:?}", importer_path);
if already_visited.contains(importer_path) {
tracing::trace!("Skipping already visited dependency");
continue;
}
let mut tx = db.clone().begin().await?;
let mut args: HashMap<String, Box<RawValue>> = HashMap::new();
if let Some(ref dm) = deployment_message {
args.insert("deployment_message".to_string(), to_raw_value(&dm));
}
if let Some(ref p_path) = parent_path {
// NOTE:
// it's not used but maybe one day it will be useful. allows more back-compatibility for the workers when we need it
// also very useful for debugging/observability
// it adds that information to the job args so you can see from the runs page
args.insert("common_dependency_path".to_string(), to_raw_value(&p_path));
}
args.insert(
"already_visited".to_string(),
to_raw_value(&already_visited),
);
args.insert(
"triggered_by_relative_import".to_string(),
to_raw_value(&true),
);
let mut debouncing_settings = DebouncingSettings {
debounce_key: Some(format!("{w_id}:{importer_path}:dependency")),
debounce_delay_s: Some(5),
..Default::default()
};
let job_payload = match importer_kind.as_str() {
// TODO: Make it query only non-archived
// Scripts
"script" => match sqlx::query_scalar!(
"SELECT hash FROM script WHERE path = $1 AND workspace_id = $2 AND deleted = false ORDER BY created_at DESC LIMIT 1",
importer_path,
w_id
)
.fetch_optional(&mut *tx)
.await?
{
Some(hash) => {
tracing::debug!("newest hash for {} is: {hash}", importer_path);
let info =
windmill_common::get_script_info_for_hash(None, db, w_id, hash).await?;
JobPayload::Dependencies {
path: importer_path.clone(),
hash: ScriptHash(hash),
language: info.language,
dedicated_worker: info.dedicated_worker,
debouncing_settings,
}
}
None => {
ScopedDependencyMap::clear_map_for_item(
importer_path,
w_id,
"script",
tx,
&None,
)
.await
.commit()
.await?;
continue;
}
},
// Flows
"flow" => match sqlx::query_scalar!(
"SELECT id FROM flow_version WHERE path = $1 AND workspace_id = $2 ORDER BY created_at DESC LIMIT 1",
importer_path,
w_id
)
.fetch_optional(&mut *tx)
.await?
{
Some(version) => {
tracing::debug!("Handling flow dependency update for: {}", importer_path);
args.insert(
"nodes_to_relock".to_string(),
to_raw_value(&importer_node_ids),
);
debouncing_settings.debounce_args_to_accumulate = Some(vec!["nodes_to_relock".into()]);
JobPayload::FlowDependencies {
path: importer_path.clone(),
version,
dedicated_worker: None,
debouncing_settings,
}
}
None => {
ScopedDependencyMap::clear_map_for_item(importer_path, w_id, "flow", tx, &None)
.await
.commit()
.await?;
continue;
}
},
// Apps
"app" => match sqlx::query_scalar!(
"SELECT id FROM app_version WHERE app_id = (SELECT id FROM app WHERE path = $1 AND workspace_id = $2) ORDER BY created_at DESC LIMIT 1",
importer_path,
w_id
)
.fetch_optional(&mut *tx)
.await?
{
Some(version) => {
tracing::debug!("Handling app dependency update for: {}", importer_path);
args.insert(
"components_to_relock".to_string(),
// TODO: unsafe. Importer Node Ids are not checked. They can simply be array of empty strings!
to_raw_value(importer_node_ids),
);
debouncing_settings.debounce_args_to_accumulate = Some(vec!["components_to_relock".into()]);
JobPayload::AppDependencies { path: importer_path.clone(), version, debouncing_settings }
}
None => {
ScopedDependencyMap::clear_map_for_item(importer_path, w_id, "app", tx, &None)
.await
.commit()
.await?;
continue;
}
},
_ => {
tracing::error!(
"unexpected importer kind: {kind:?} for path {path}",
kind = importer_kind,
path = importer_path
);
continue;
}
};
tracing::debug!("Pushing dependency job for: {}", importer_path);
let (job_uuid, new_tx) = windmill_queue::push(
db,
PushIsolationLevel::Transaction(tx),
&w_id,
job_payload,
windmill_queue::PushArgs { args: &args, extra: None },
&created_by,
email,
permissioned_as.to_string(),
Some("trigger.dependents.to.recompute.dependencies"),
// Schedule for future for debouncing.
Some(Utc::now() + Duration::seconds(*DEPENDENCY_JOB_DEBOUNCE_DELAY as i64)),
None,
None,
None,
None,
None,
false,
false,
None,
true,
Some("dependency".into()),
None,
None,
None,
None,
false,
None,
None,
None,
)
.await?;
tracing::info!(
"pushed dependency job due to common python path: {job_uuid} for path {path}",
path = importer_path,
);
new_tx.commit().await?;
}
Ok(())
}
pub async fn handle_flow_dependency_job(
job: MiniPulledJob,
preview_data: Option<&RawData>,