fix: worker do not apply migrations anymore but wait for servers to do so

backend/src/main.rs

@@ -735,8 +735,12 @@ async fn windmill_main() -> anyhow::Result<()> {
                 .unwrap_or(false);
 
             if !skip_migration {
-                // migration code to avoid break
-                migration_handle = windmill_api::migrate_db(&db, killpill_rx.resubscribe()).await?;
+                if mode == Mode::Worker {
+                    windmill_api::wait_for_db_migrations(&db, killpill_rx.resubscribe()).await?;
+                } else {
+                    migration_handle =
+                        windmill_api::migrate_db(&db, killpill_rx.resubscribe()).await?;
+                }
             } else {
                 tracing::info!("SKIP_MIGRATION set, skipping db migration...")
             }

backend/windmill-api/src/db.rs

@@ -15,7 +15,7 @@ use sqlx::{
 
 use tokio::task::JoinHandle;
 pub use windmill_common::db::DB;
-use windmill_common::{error::Error, utils::generate_lock_id};
+use windmill_common::{error::Error, utils::{generate_lock_id, GIT_VERSION}};
 
 #[allow(unused_imports)]
 pub use windmill_api_auth::{ApiAuthed, OptJobAuthed};
@@ -289,3 +289,66 @@ pub async fn migrate(
     crate::live_migrations::custom_migrations(&mut custom_migrator, db).await?;
     Ok(None)
 }
+
+pub async fn wait_for_migrations(
+    db: &DB,
+    mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
+) -> Result<(), Error> {
+    let migrator = sqlx::migrate!("../migrations");
+    let latest_version = migrator
+        .migrations
+        .iter()
+        .map(|m| m.version)
+        .max()
+        .expect("No migrations found") as i64;
+
+    tracing::info!(
+        "This worker is on Windmill version {GIT_VERSION} (migration version {latest_version}). Only servers run migrations. Waiting for a server with version >= {GIT_VERSION} to apply the migration..."
+    );
+
+    let mut attempts = 0;
+
+    loop {
+        let is_applied: Result<Option<bool>, sqlx::Error> = sqlx::query_scalar(
+            "SELECT EXISTS(SELECT 1 FROM _sqlx_migrations WHERE version = $1)",
+        )
+        .bind(latest_version)
+        .fetch_one(db)
+        .await;
+
+        match is_applied {
+            Ok(Some(true)) => {
+                tracing::info!(
+                    "All migrations applied (version {latest_version}), continuing worker startup"
+                );
+                return Ok(());
+            }
+            Ok(_) => {
+                tracing::info!(
+                    "Database not up to date yet. This worker (Windmill {GIT_VERSION}, migration version {latest_version}) is waiting for a server with version >= {GIT_VERSION} to run the migration. Rechecking in 3s..."
+                );
+            }
+            Err(e) => {
+                tracing::info!(
+                    "Could not check migration status (migrations table may not exist yet): {e:#}. Rechecking in 3s..."
+                );
+            }
+        }
+
+        tokio::select! {
+            _ = killpill_rx.recv() => {
+                tracing::info!("Killpill received, stopping migration wait");
+                return Ok(());
+            }
+            _ = tokio::time::sleep(std::time::Duration::from_secs(3)) => {}
+        }
+
+        attempts += 1;
+        if attempts >= 10 {
+            tracing::error!(
+                "Timed out after 10 attempts waiting for migration version {latest_version}. Exiting."
+            );
+            std::process::exit(1);
+        }
+    }
+}

backend/windmill-api/src/lib.rs

@@ -962,3 +962,12 @@ pub async fn migrate_db(
         .await
         .map_err(|e| anyhow::anyhow!("Error migrating db: {e:#}"))
 }
+
+pub async fn wait_for_db_migrations(
+    db: &DB,
+    killpill_rx: tokio::sync::broadcast::Receiver<()>,
+) -> anyhow::Result<()> {
+    db::wait_for_migrations(db, killpill_rx)
+        .await
+        .map_err(|e| anyhow::anyhow!("Error waiting for db migrations: {e:#}"))
+}