kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: replace v2 live migration with regular SQL migration (#7678)

Browse Source 
* refactor: replace v2 live migration with regular SQL migration

Remove the v2_finalize live migration that waited for workers to reach
v1.461+ and replace it with an idempotent SQL migration. Since enough
time has passed, all deployments are assumed to be on v1.461+.

Changes:
- Remove v2_finalize() function and spawn task from live_migrations.rs
- Remove MIN_VERSION_IS_AT_LEAST_1_461 constant from min_version.rs
- Add 20260125000000_v2_finalize.up.sql with all finalization steps
- All SQL operations use IF EXISTS/CASCADE for idempotency

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

* refactor: remove migration-related delays and tighten CLI tests CI trigger

- Remove 5s backend initialization delay in cargo_backend.ts that was
  needed for the v2 live migration (now replaced with SQL migration)
- Restrict CLI tests workflow to only trigger on cli/** changes
  (removed backend/**, openapi.yaml, openflow.openapi.yaml paths)

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

* fix: skip v2_finalize migration if live migration already ran

Wrap migration in DO block that checks for 'v2_finalize_job_completed'
in windmill_migrations table. If present (live migration already ran),
skip entirely to avoid unnecessary table locks on upgraded instances.

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

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-01-25 22:32:21 +00:00
committed by GitHub
parent 35e20fec93
commit a4d0b965ef
7 changed files with 104 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.github/workflows/cli-tests.yml vendored
View File

@@ -5,17 +5,11 @@ on:
branches: [main]
paths:
- 'cli/**'
- 'backend/**'
- 'openapi.yaml'
- 'openflow.openapi.yaml'
- '.github/workflows/cli-tests.yml'
pull_request:
branches: [main]
paths:
- 'cli/**'
- 'backend/**'
- 'openapi.yaml'
- 'openflow.openapi.yaml'
- '.github/workflows/cli-tests.yml'
env:

3
backend/migrations/20260125000000_v2_finalize.down.sql Normal file
View File

@@ -0,0 +1,3 @@
-- This migration cannot be reversed.
-- The v2 compatibility layer has been permanently removed.
DO $$ BEGIN RAISE EXCEPTION 'Cannot reverse v2 finalization migration'; END $$;

97
backend/migrations/20260125000000_v2_finalize.up.sql Normal file
View File

@@ -0,0 +1,97 @@
-- V2 Migration Finalization
-- This migration finalizes the v2 job table migration by removing the compatibility layer.
-- Skip if live migration already ran (indicated by 'v2_finalize_job_completed' in windmill_migrations).
DO $$
BEGIN
-- Check if live migration already completed this work
IF EXISTS (SELECT 1 FROM windmill_migrations WHERE name = 'v2_finalize_job_completed') THEN
RAISE NOTICE 'v2_finalize already done via live migration, skipping';
RETURN;
END IF;
-- 1. Disable RLS on v2 tables (no-op if already disabled)
ALTER TABLE v2_job_queue DISABLE ROW LEVEL SECURITY;
ALTER TABLE v2_job_completed DISABLE ROW LEVEL SECURITY;
-- 2. Drop sync functions (CASCADE drops associated triggers)
DROP FUNCTION IF EXISTS v2_job_after_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_after_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_before_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_completed_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_completed_before_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_runtime_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_runtime_before_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_status_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_status_before_update CASCADE;
-- 3. Drop compatibility views
DROP VIEW IF EXISTS completed_job CASCADE;
DROP VIEW IF EXISTS completed_job_view CASCADE;
DROP VIEW IF EXISTS job CASCADE;
DROP VIEW IF EXISTS queue CASCADE;
DROP VIEW IF EXISTS queue_view CASCADE;
-- 4. Drop __ columns from v2_job_queue
ALTER TABLE v2_job_queue
DROP COLUMN IF EXISTS __parent_job CASCADE,
DROP COLUMN IF EXISTS __created_by CASCADE,
DROP COLUMN IF EXISTS __script_hash CASCADE,
DROP COLUMN IF EXISTS __script_path CASCADE,
DROP COLUMN IF EXISTS __args CASCADE,
DROP COLUMN IF EXISTS __logs CASCADE,
DROP COLUMN IF EXISTS __raw_code CASCADE,
DROP COLUMN IF EXISTS __canceled CASCADE,
DROP COLUMN IF EXISTS __last_ping CASCADE,
DROP COLUMN IF EXISTS __job_kind CASCADE,
DROP COLUMN IF EXISTS __env_id CASCADE,
DROP COLUMN IF EXISTS __schedule_path CASCADE,
DROP COLUMN IF EXISTS __permissioned_as CASCADE,
DROP COLUMN IF EXISTS __flow_status CASCADE,
DROP COLUMN IF EXISTS __raw_flow CASCADE,
DROP COLUMN IF EXISTS __is_flow_step CASCADE,
DROP COLUMN IF EXISTS __language CASCADE,
DROP COLUMN IF EXISTS __same_worker CASCADE,
DROP COLUMN IF EXISTS __raw_lock CASCADE,
DROP COLUMN IF EXISTS __pre_run_error CASCADE,
DROP COLUMN IF EXISTS __email CASCADE,
DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
DROP COLUMN IF EXISTS __mem_peak CASCADE,
DROP COLUMN IF EXISTS __root_job CASCADE,
DROP COLUMN IF EXISTS __leaf_jobs CASCADE,
DROP COLUMN IF EXISTS __concurrent_limit CASCADE,
DROP COLUMN IF EXISTS __concurrency_time_window_s CASCADE,
DROP COLUMN IF EXISTS __timeout CASCADE,
DROP COLUMN IF EXISTS __flow_step_id CASCADE,
DROP COLUMN IF EXISTS __cache_ttl CASCADE;
-- 5. Drop __ columns from v2_job_completed
ALTER TABLE v2_job_completed
DROP COLUMN IF EXISTS __parent_job CASCADE,
DROP COLUMN IF EXISTS __created_by CASCADE,
DROP COLUMN IF EXISTS __created_at CASCADE,
DROP COLUMN IF EXISTS __success CASCADE,
DROP COLUMN IF EXISTS __script_hash CASCADE,
DROP COLUMN IF EXISTS __script_path CASCADE,
DROP COLUMN IF EXISTS __args CASCADE,
DROP COLUMN IF EXISTS __logs CASCADE,
DROP COLUMN IF EXISTS __raw_code CASCADE,
DROP COLUMN IF EXISTS __canceled CASCADE,
DROP COLUMN IF EXISTS __job_kind CASCADE,
DROP COLUMN IF EXISTS __env_id CASCADE,
DROP COLUMN IF EXISTS __schedule_path CASCADE,
DROP COLUMN IF EXISTS __permissioned_as CASCADE,
DROP COLUMN IF EXISTS __raw_flow CASCADE,
DROP COLUMN IF EXISTS __is_flow_step CASCADE,
DROP COLUMN IF EXISTS __language CASCADE,
DROP COLUMN IF EXISTS __is_skipped CASCADE,
DROP COLUMN IF EXISTS __raw_lock CASCADE,
DROP COLUMN IF EXISTS __email CASCADE,
DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
DROP COLUMN IF EXISTS __tag CASCADE,
DROP COLUMN IF EXISTS __priority CASCADE;
-- Mark as done in windmill_migrations for consistency
INSERT INTO windmill_migrations (name) VALUES ('v2_finalize_job_completed') ON CONFLICT DO NOTHING;
END $$;

3
backend/windmill-api/src/db.rs
View File

@@ -247,7 +247,8 @@ pub async fn migrate(
}
}
return crate::live_migrations::custom_migrations(&mut custom_migrator, db).await;
crate::live_migrations::custom_migrations(&mut custom_migrator, db).await?;
Ok(None)
}
#[derive(Clone, Debug, Default, Hash, Eq, PartialEq)]

194
backend/windmill-api/src/live_migrations.rs
View File

@@ -7,19 +7,13 @@
*/
use sqlx::Postgres;
use std::time::Duration;
use tokio::task::JoinHandle;
use windmill_common::error::Error;
use windmill_common::min_version::MIN_VERSION_IS_AT_LEAST_1_461;
use crate::db::{CustomMigrator, DB};
use sqlx::migrate::Migrate;
use sqlx::Executor;
pub async fn custom_migrations(
migrator: &mut CustomMigrator,
db: &DB,
) -> Result<Option<JoinHandle<()>>, Error> {
pub async fn custom_migrations(migrator: &mut CustomMigrator, db: &DB) -> Result<(), Error> {
if let Err(err) = fix_flow_versioning_migration(migrator, db).await {
tracing::error!("Could not apply flow versioning fix migration: {err:#}");
}
@@ -31,31 +25,7 @@ pub async fn custom_migrations(
}
});
let mut jh = None;
if !has_done_migration(db, "v2_finalize_job_completed").await {
let db2 = db.clone();
let v2jh = tokio::task::spawn(async move {
loop {
if !MIN_VERSION_IS_AT_LEAST_1_461.met().await {
tracing::info!("Waiting for all workers to be at least version 1.461 before applying v2 finalize migration, sleeping for 5s...");
tokio::time::sleep(Duration::from_secs(5)).await;
continue;
}
if let Err(err) = v2_finalize(&db2).await {
tracing::error!(
"{err:#}: Could not apply v2 finalize migration, retry in 30s.."
);
tokio::time::sleep(Duration::from_secs(30)).await;
continue;
}
tracing::info!("v2 finalization step successfully applied.");
break;
}
});
jh = Some(v2jh)
}
Ok(jh)
Ok(())
}
async fn fix_flow_versioning_migration(
@@ -179,166 +149,6 @@ macro_rules! run_windmill_migration {
};
}
async fn v2_finalize(db: &DB) -> Result<(), Error> {
run_windmill_migration!("v2_finalize_disable_sync_III", db, |tx| {
tx.execute(
r#"
LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
ALTER TABLE v2_job_queue DISABLE ROW LEVEL SECURITY;
"#,
)
.await?;
});
run_windmill_migration!("v2_finalize_disable_sync_III_2", db, |tx| {
tx.execute(
r#"
LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
ALTER TABLE v2_job_completed DISABLE ROW LEVEL SECURITY;
"#,
)
.await?;
});
run_windmill_migration!("v2_finalize_disable_sync_III_3", db, |tx| {
tx.execute(
r#"
LOCK TABLE v2_job IN ACCESS EXCLUSIVE MODE;
DROP FUNCTION IF EXISTS v2_job_after_update CASCADE;
"#,
)
.await?;
});
run_windmill_migration!("v2_finalize_disable_sync_III_4", db, |tx| {
tx.execute(
r#"
LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
DROP FUNCTION IF EXISTS v2_job_completed_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_completed_before_update CASCADE;
"#,
)
.await?;
});
run_windmill_migration!("v2_finalize_disable_sync_III_5", db, |tx| {
tx.execute(
r#"
LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
DROP FUNCTION IF EXISTS v2_job_queue_after_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_before_update CASCADE;
"#,
)
.await?;
});
run_windmill_migration!("v2_finalize_disable_sync_III_6", db, |tx| {
tx.execute(
r#"
LOCK TABLE v2_job_runtime IN ACCESS EXCLUSIVE MODE;
DROP FUNCTION IF EXISTS v2_job_runtime_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_runtime_before_update CASCADE;
"#,
)
.await?;
});
run_windmill_migration!("v2_finalize_disable_sync_III_7", db, |tx| {
tx.execute(
r#"
LOCK TABLE v2_job_status IN ACCESS EXCLUSIVE MODE;
DROP FUNCTION IF EXISTS v2_job_status_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_status_before_update CASCADE;
"#,
)
.await?;
});
run_windmill_migration!("v2_finalize_disable_sync_III_8", db, |tx| {
tx.execute(
r#"
DROP VIEW IF EXISTS completed_job, completed_job_view, job, queue, queue_view CASCADE;
"#,
)
.await?;
});
run_windmill_migration!("v2_finalize_job_queue", db, |tx| {
tx.execute(
r#"
LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
ALTER TABLE v2_job_queue
DROP COLUMN IF EXISTS __parent_job CASCADE,
DROP COLUMN IF EXISTS __created_by CASCADE,
DROP COLUMN IF EXISTS __script_hash CASCADE,
DROP COLUMN IF EXISTS __script_path CASCADE,
DROP COLUMN IF EXISTS __args CASCADE,
DROP COLUMN IF EXISTS __logs CASCADE,
DROP COLUMN IF EXISTS __raw_code CASCADE,
DROP COLUMN IF EXISTS __canceled CASCADE,
DROP COLUMN IF EXISTS __last_ping CASCADE,
DROP COLUMN IF EXISTS __job_kind CASCADE,
DROP COLUMN IF EXISTS __env_id CASCADE,
DROP COLUMN IF EXISTS __schedule_path CASCADE,
DROP COLUMN IF EXISTS __permissioned_as CASCADE,
DROP COLUMN IF EXISTS __flow_status CASCADE,
DROP COLUMN IF EXISTS __raw_flow CASCADE,
DROP COLUMN IF EXISTS __is_flow_step CASCADE,
DROP COLUMN IF EXISTS __language CASCADE,
DROP COLUMN IF EXISTS __same_worker CASCADE,
DROP COLUMN IF EXISTS __raw_lock CASCADE,
DROP COLUMN IF EXISTS __pre_run_error CASCADE,
DROP COLUMN IF EXISTS __email CASCADE,
DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
DROP COLUMN IF EXISTS __mem_peak CASCADE,
DROP COLUMN IF EXISTS __root_job CASCADE,
DROP COLUMN IF EXISTS __leaf_jobs CASCADE,
DROP COLUMN IF EXISTS __concurrent_limit CASCADE,
DROP COLUMN IF EXISTS __concurrency_time_window_s CASCADE,
DROP COLUMN IF EXISTS __timeout CASCADE,
DROP COLUMN IF EXISTS __flow_step_id CASCADE,
DROP COLUMN IF EXISTS __cache_ttl CASCADE;
"#,
)
.await?;
});
run_windmill_migration!("v2_finalize_job_completed", db, |tx| {
tx.execute(
r#"
LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
ALTER TABLE v2_job_completed
DROP COLUMN IF EXISTS __parent_job CASCADE,
DROP COLUMN IF EXISTS __created_by CASCADE,
DROP COLUMN IF EXISTS __created_at CASCADE,
DROP COLUMN IF EXISTS __success CASCADE,
DROP COLUMN IF EXISTS __script_hash CASCADE,
DROP COLUMN IF EXISTS __script_path CASCADE,
DROP COLUMN IF EXISTS __args CASCADE,
DROP COLUMN IF EXISTS __logs CASCADE,
DROP COLUMN IF EXISTS __raw_code CASCADE,
DROP COLUMN IF EXISTS __canceled CASCADE,
DROP COLUMN IF EXISTS __job_kind CASCADE,
DROP COLUMN IF EXISTS __env_id CASCADE,
DROP COLUMN IF EXISTS __schedule_path CASCADE,
DROP COLUMN IF EXISTS __permissioned_as CASCADE,
DROP COLUMN IF EXISTS __raw_flow CASCADE,
DROP COLUMN IF EXISTS __is_flow_step CASCADE,
DROP COLUMN IF EXISTS __language CASCADE,
DROP COLUMN IF EXISTS __is_skipped CASCADE,
DROP COLUMN IF EXISTS __raw_lock CASCADE,
DROP COLUMN IF EXISTS __email CASCADE,
DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
DROP COLUMN IF EXISTS __tag CASCADE,
DROP COLUMN IF EXISTS __priority CASCADE;
"#,
)
.await?;
});
Ok(())
}
async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
// let has_done_migration = sqlx::query_scalar!(
// "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index')"

1
backend/windmill-common/src/min_version.rs
View File

@@ -11,7 +11,6 @@ pub const MIN_VERSION_IS_AT_LEAST_1_595: VC = vc(1, 595, 0, "Flow status separat
pub const MIN_VERSION_SUPPORTS_RUNNABLE_SETTINGS_V0: VC = vc(1, 592, 0, "Runnable settings V0");
pub const MIN_VERSION_SUPPORTS_V0_WORKSPACE_DEPENDENCIES: VC = vc(1, 587, 0, "Workspace dependencies");
pub const MIN_VERSION_SUPPORTS_DEBOUNCING: VC = vc(1, 566, 0, "Debouncing");
pub const MIN_VERSION_IS_AT_LEAST_1_461: VC = vc(1, 461, 0, "V2 job tables");
pub const MIN_VERSION_IS_AT_LEAST_1_440: VC = vc(1, 440, 0, "Flow node value on pull");
pub const MIN_VERSION_IS_AT_LEAST_1_432: VC = vc(1, 432, 0, "Flow script job kind");
pub const MIN_VERSION_IS_AT_LEAST_1_427: VC = vc(1, 427, 0, "Flow version lite table");

5
cli/test/cargo_backend.ts
View File

@@ -160,11 +160,6 @@ export class CargoBackend {
console.log(` Server: ${this.baseUrl}`);
console.log(` Database: ${this.dbName}`);
console.log(` Workspace: ${this.config.workspace}`);
// Wait for backend to fully initialize (migrations, etc.)
console.log("⏳ Waiting 5s for backend to fully initialize...");
await new Promise(resolve => setTimeout(resolve, 5000));
console.log("✅ Ready to run tests");
}
/**