diff --git a/backend/src/main.rs b/backend/src/main.rs index 08f6d081a5..4a7b23aab7 100644 --- a/backend/src/main.rs +++ b/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...") } diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 696d752bab..4c3dd60696 100644 --- a/backend/windmill-api/src/db.rs +++ b/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, 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); + } + } +} diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index d76279da44..af392b54b3 100644 --- a/backend/windmill-api/src/lib.rs +++ b/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:#}")) +}