feat(backend): monitor minimal version of living workers (#4704)

backend/.sqlx/query-ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT wm_version FROM worker_ping WHERE wm_version != $1 AND ping_at > now() - interval '5 minutes'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "wm_version",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6"
+}

backend/Cargo.lock

@@ -10811,6 +10811,7 @@ dependencies = [
  "rand 0.8.5",
  "regex",
  "reqwest 0.12.9",
+ "semver 1.0.23",
  "serde",
  "serde_json",
  "sha2 0.10.8",

backend/Cargo.toml

@@ -180,6 +180,7 @@ tokio-util = { version = "^0", features = ["io"] }
 json-pointer = "^0"
 itertools = "^0"
 regex = "^1"
+semver = "^1"
 
 deno_fetch = "0.195.0"
 deno_tls = "0.158.0"

backend/src/main.rs

@@ -7,7 +7,6 @@
  */
 
 use anyhow::Context;
-use git_version::git_version;
 use monitor::{
     reload_timeout_wait_result_setting, send_current_log_file_to_object_store,
     send_logs_to_object_store,
@@ -42,7 +41,7 @@ use windmill_common::{
     },
     scripts::ScriptLang,
     stats_ee::schedule_stats,
-    utils::{hostname, rd_string, Mode},
+    utils::{hostname, rd_string, Mode, GIT_VERSION},
     worker::{reload_custom_tags_setting, HUB_CACHE_DIR, TMP_DIR, WORKER_GROUP},
     DB, METRICS_ENABLED,
 };
@@ -84,7 +83,6 @@ use crate::monitor::{
 #[cfg(feature = "parquet")]
 use crate::monitor::reload_s3_cache_setting;
 
-const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
 const DEFAULT_NUM_WORKERS: usize = 1;
 const DEFAULT_PORT: u16 = 8000;
 const DEFAULT_SERVER_BIND_ADDR: Ipv4Addr = Ipv4Addr::new(0, 0, 0, 0);

backend/src/monitor.rs

@@ -51,8 +51,8 @@ use windmill_common::{
     utils::{now_from_db, rd_string, report_critical_error, Mode},
     worker::{
         load_worker_config, make_pull_query, make_suspended_pull_query, reload_custom_tags_setting,
-        DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, SMTP_CONFIG, WORKER_CONFIG,
-        WORKER_GROUP,
+        update_min_version, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, SMTP_CONFIG,
+        WORKER_CONFIG, WORKER_GROUP,
     },
     BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS,
     METRICS_DEBUG_ENABLED, METRICS_ENABLED, CRITICAL_ALERT_MUTE_UI_ENABLED
@@ -1080,6 +1080,10 @@ pub async fn monitor_db(
         }
     };
 
+    let update_min_worker_version_f = async {
+        update_min_version(db).await;
+    };
+
     join!(
         expired_items_f,
         zombie_jobs_f,
@@ -1088,6 +1092,7 @@ pub async fn monitor_db(
         worker_groups_alerts_f,
         jobs_waiting_alerts_f,
         apply_autoscaling_f,
+        update_min_worker_version_f,
     );
 }
 

backend/windmill-api/src/lib.rs

@@ -25,7 +25,6 @@ use argon2::Argon2;
 use axum::extract::DefaultBodyLimit;
 use axum::{middleware::from_extractor, routing::get, Extension, Router};
 use db::DB;
-use git_version::git_version;
 use http::HeaderValue;
 use reqwest::Client;
 use std::collections::HashMap;
@@ -40,7 +39,7 @@ use tower_http::{
 };
 use windmill_common::db::UserDB;
 use windmill_common::worker::{ALL_TAGS, CLOUD_HOSTED};
-use windmill_common::{BASE_URL, INSTANCE_NAME};
+use windmill_common::{BASE_URL, INSTANCE_NAME, utils::GIT_VERSION};
 
 use crate::scim_ee::has_scim_token;
 use windmill_common::error::AppError;
@@ -93,9 +92,6 @@ mod workers;
 mod workspaces;
 mod workspaces_ee;
 
-pub const GIT_VERSION: &str =
-    git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
-
 pub const DEFAULT_BODY_LIMIT: usize = 2097152 * 100; // 200MB
 
 lazy_static::lazy_static! {

backend/windmill-common/Cargo.toml

@@ -58,6 +58,7 @@ async-stream.workspace = true
 const_format.workspace = true
 crc.workspace = true
 windmill-macros.workspace = true
+semver.workspace = true
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemalloc-ctl = { optional = true, workspace = true }

backend/windmill-common/src/utils.rs

@@ -21,6 +21,7 @@ use reqwest::Client;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use sqlx::{Pool, Postgres};
+use semver::Version;
 
 pub const MAX_PER_PAGE: usize = 10000;
 pub const DEFAULT_PER_PAGE: usize = 1000;
@@ -37,6 +38,10 @@ lazy_static::lazy_static! {
         .timeout(std::time::Duration::from_secs(20))
         .connect_timeout(std::time::Duration::from_secs(10))
         .build().unwrap();
+    pub static ref GIT_SEM_VERSION: Version = Version::parse(
+        // skip first `v` character.
+        GIT_VERSION.split_at(1).1
+    ).unwrap_or(Version::new(0, 1, 0));
 }
 
 #[derive(Deserialize, Clone)]

backend/windmill-common/src/worker.rs

@@ -1,6 +1,7 @@
 use const_format::concatcp;
 use itertools::Itertools;
 use regex::Regex;
+use semver::Version;
 use serde::{Deserialize, Serialize};
 use serde_json::value::RawValue;
 use std::{
@@ -88,6 +89,7 @@ lazy_static::lazy_static! {
     .and_then(|x| x.parse::<bool>().ok())
     .unwrap_or(false);
 
+    pub static ref MIN_VERSION: Arc<RwLock<Version>> = Arc::new(RwLock::new(Version::new(0, 0, 0)));
 }
 
 pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
@@ -548,6 +550,31 @@ pub fn get_windmill_memory_usage() -> Option<i64> {
     }
 }
 
+pub async fn update_min_version<'c, E: sqlx::Executor<'c, Database = sqlx::Postgres>>(executor: E) -> bool {
+    use crate::utils::{GIT_VERSION, GIT_SEM_VERSION};
+
+    // fetch all pings with a different version than self from the last 5 minutes.
+    let pings = sqlx::query_scalar!(
+        "SELECT wm_version FROM worker_ping WHERE wm_version != $1 AND ping_at > now() - interval '5 minutes'",
+        GIT_VERSION
+    ).fetch_all(executor).await.unwrap_or_default();
+
+    let cur_version = GIT_SEM_VERSION.clone();
+    let min_version = pings
+        .iter()
+        .filter(|x| !x.is_empty())
+        .filter_map(|x| semver::Version::parse(x.split_at(1).1).ok())
+        .min()
+        .unwrap_or_else(|| cur_version.clone());
+
+    if min_version != cur_version {
+        tracing::info!("Minimal worker version: {min_version}");
+    }
+
+    *MIN_VERSION.write().await = min_version.clone();
+    min_version >= cur_version
+}
+
 pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db: &DB) {
     let (tags, dw) = {
         let wc = WORKER_CONFIG.read().await.clone();