fix tests

.env

@@ -5,7 +5,7 @@ WM_IMAGE=ghcr.io/windmill-labs/windmill:main
 WM_LICENSE_KEY=""
 
 # For Enterprise Edition, comment the 2 lines above and uncomment below
-# WINDMILL_IMAGE=ghcr.io/windmill-labs/windmill-ee:main
+# WM_IMAGE=ghcr.io/windmill-labs/windmill-ee:main
 # WM_LICENSE_KEY="<id>.<expiry>.<signature>"
 
 

backend/tests/worker.rs

@@ -1,7 +1,10 @@
 use std::sync::Arc;
 
+#[cfg(feature = "enterprise")]
 use chrono::Timelike;
+#[cfg(feature = "enterprise")]
 use futures::StreamExt;
+
 use futures::{stream, Stream};
 use serde::Deserialize;
 use serde_json::json;
@@ -13,7 +16,7 @@ use tokio::{
 use windmill_api_client::types::{
     CreateFlowBody, EditSchedule, NewSchedule, RawScript, ScriptArgs,
 };
-use windmill_common::worker::{WORKER_CONFIG, load_worker_config};
+use windmill_common::worker::WORKER_CONFIG;
 use windmill_common::{
     flow_status::{FlowStatus, FlowStatusModule},
     flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform},
@@ -126,6 +129,7 @@ impl ApiServer {
     }
 
     async fn close(self) -> anyhow::Result<()> {
+        println!("closing api server");
         let Self { tx, task, .. } = self;
         drop(tx);
         task.await.unwrap()
@@ -892,7 +896,8 @@ impl RunJob {
         let uuid = self.push(db).await;
         let listener = listen_for_completed_jobs(db).await;
         in_test_worker(db, listener.find(&uuid), port).await;
-        completed_job(uuid, db).await
+        let r = completed_job(uuid, db).await;
+        r
     }
 }
 
@@ -933,7 +938,6 @@ async fn in_test_worker<Fut: std::future::Future>(
         .await
         .expect("worker timed out")
         .expect("worker panicked");
-
     res
 }
 
@@ -960,9 +964,10 @@ fn spawn_test_worker(
     let tx2 = tx.clone();
     let future = async move {
         let base_internal_url = format!("http://localhost:{}", port);
+        {
         let mut wc = WORKER_CONFIG.write().await;
-        *wc = load_worker_config(&db).await.unwrap();
-        drop(wc);
+        (*wc).worker_tags = windmill_common::worker::DEFAULT_TAGS.clone();
+        }
         windmill_worker::run_worker::<rsmq_async::MultiplexedRsmq>(
             &db,
             worker_instance,
@@ -1670,7 +1675,6 @@ echo "hello $msg"
     .arg("msg", json!("world"))
     .run_until_complete(&db, port)
     .await;
-
     assert_eq!(job.json_result(), Some(json!("hello world")));
 }
 
@@ -2603,6 +2607,7 @@ async fn test_rust_client(db: Pool<Postgres>) {
 }
 
 
+#[cfg(feature = "enterprise")]
 #[sqlx::test(fixtures("base"))]
 async fn test_script_schedule_handlers(db: Pool<Postgres>) {
     initialize_tracing().await;
@@ -2736,6 +2741,7 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
 }
 
 
+#[cfg(feature = "enterprise")]
 #[sqlx::test(fixtures("base"))]
 async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
     initialize_tracing().await;

backend/windmill-common/src/lib.rs

@@ -33,7 +33,7 @@ pub mod worker;
 pub mod tracing_init;
 
 pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;
-pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 5;
+pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 10;
 
 lazy_static::lazy_static! {
     pub static ref METRICS_ADDR: Option<SocketAddr> = std::env::var("METRICS_ADDR")

backend/windmill-common/src/worker.rs

@@ -119,16 +119,23 @@ fn process_custom_tags(tags: Vec<String>) -> (Vec<String>, HashMap<String, Vec<S
 }
 
 pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db: &DB) {
-    let wc = WORKER_CONFIG.read().await;
-    let tags = wc.worker_tags.as_slice();
+    let (tags, dw) = {
+        let wc = WORKER_CONFIG.read().await.clone();
+        (
+            wc.worker_tags,
+            wc.dedicated_worker
+                .as_ref()
+                .map(|x| format!("{}:{}", x.workspace_id, x.path)),
+        )
+    };
     sqlx::query!(
         "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
         worker_instance,
         worker_name,
         ip,
-        tags,
+        tags.as_slice(),
         *WORKER_GROUP,
-        wc.dedicated_worker.as_ref().map(|x| format!("{}:{}", x.workspace_id, x.path))
+        dw
     )
     .execute(db)
     .await
@@ -193,7 +200,7 @@ impl Default for WorkerConfigOpt {
     }
 }
 
-#[derive(PartialEq, Debug)]
+#[derive(PartialEq, Debug, Clone)]
 pub struct WorkerConfig {
     pub worker_tags: Vec<String>,
     pub dedicated_worker: Option<WorkspacedPath>,

backend/windmill-queue/src/jobs.rs

@@ -1164,9 +1164,9 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
          *   suspend_until is non-null
          *   and suspend = 0 when the resume messages are received
          *   or suspend_until <= now() if it has timed out */
-        let config = WORKER_CONFIG.read().await;
-        let tags = config.worker_tags.as_slice();
-
+        let config = WORKER_CONFIG.read().await.clone();
+        let tags = config.worker_tags.clone();
+        drop(config);
         let r = if suspend_first {
             sqlx::query_as::<_, QueuedJob>("UPDATE queue
             SET running = true
@@ -1188,14 +1188,12 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
         } else {
             None
         };
-        drop(config);
 
         if r.is_none() {
             // #[cfg(feature = "benchmark")]
             // let instant = Instant::now();
 
-            let config = WORKER_CONFIG.read().await;
-            let tags = config.worker_tags.as_slice();
+            let tags = WORKER_CONFIG.read().await.worker_tags.clone();
 
             let r = sqlx::query_as::<_, QueuedJob>(
                 "UPDATE queue

backend/windmill-worker/src/worker.rs

@@ -742,13 +742,12 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
         let copy_tx = _copy_to_bucket_tx.clone();
 
         if last_ping.elapsed().as_secs() > NUM_SECS_PING {
-            let wc = WORKER_CONFIG.read().await;
-            let tags = wc.worker_tags.as_slice();
+            let tags = WORKER_CONFIG.read().await.worker_tags.clone();
 
             sqlx::query!(
                 "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2 WHERE worker = $3",
                 jobs_executed,
-                tags,
+                tags.as_slice(),
                 &worker_name
             )
             .execute(db)