diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs
index 6268f603e6..3498825f6b 100644
--- a/backend/windmill-api/src/db.rs
+++ b/backend/windmill-api/src/db.rs
@@ -6,14 +6,16 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use futures::FutureExt;
-use sqlx::Executor;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
 
+use futures::FutureExt;
 use sqlx::{
     migrate::{Migrate, MigrateError},
     pool::PoolConnection,
-    PgConnection, Pool, Postgres,
+    Executor, PgConnection, Pool, Postgres,
 };
+
 use windmill_audit::audit_ee::{AuditAuthor, AuditAuthorable};
 use windmill_common::utils::generate_lock_id;
 use windmill_common::{
@@ -199,6 +201,24 @@ pub async fn migrate(db: &DB) -> Result<(), Error> {
         }
     });
 
+    let db2 = db.clone();
+    let _ = tokio::task::spawn(async move {
+        use windmill_common::worker::MIN_VERSION_IS_LATEST;
+        loop {
+            if !MIN_VERSION_IS_LATEST.load(Ordering::Relaxed) {
+                tokio::time::sleep(Duration::from_secs(30)).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;
+        }
+    });
+
     Ok(())
 }
 
@@ -250,7 +270,7 @@ async fn fix_flow_versioning_migration(
 }
 
 macro_rules! run_windmill_migration {
-    ($migration_job_name:expr, $db:expr, $code:block) => {
+    ($migration_job_name:expr, $db:expr, |$tx:ident| $code:block) => {
         {
             let migration_job_name = $migration_job_name;
             let db: &Pool<Postgres> = $db;
@@ -264,11 +284,11 @@ macro_rules! run_windmill_migration {
             .unwrap_or(false);
             if !has_done_migration {
                 tracing::info!("Applying {migration_job_name} migration");
-                let mut tx = db.begin().await?;
+                let mut $tx = db.begin().await?;
                 let mut r = false;
                 while !r {
                     r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)")
-                        .fetch_one(&mut *tx)
+                        .fetch_one(&mut *$tx)
                         .await
                         .map_err(|e| {
                             tracing::error!("Error acquiring {migration_job_name} lock: {e:#}");
@@ -298,7 +318,7 @@ macro_rules! run_windmill_migration {
                         "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING",
                         migration_job_name
                     )
-                    .execute(&mut *tx)
+                    .execute(&mut *$tx)
                     .await?;
                     tracing::info!("Finished applying {migration_job_name} migration");
                 } else {
@@ -306,9 +326,9 @@ macro_rules! run_windmill_migration {
                 }
 
                 let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
-                    .execute(&mut *tx)
+                    .execute(&mut *$tx)
                     .await?;
-                tx.commit().await?;
+                $tx.commit().await?;
                 tracing::info!("released lock for {migration_job_name}");
             } else {
                 tracing::debug!("migration {migration_job_name} already done");
@@ -318,6 +338,108 @@ macro_rules! run_windmill_migration {
     };
 }
 
+async fn v2_finalize(db: &DB) -> Result<(), Error> {
+    run_windmill_migration!("v2_finalize_disable_sync", db, |tx| {
+        tx.execute(
+            r#"
+            CREATE OR REPLACE TRIGGER v2_queue_instead_of_update_trigger
+                INSTEAD OF UPDATE ON v2_queue
+                FOR EACH ROW
+            EXECUTE PROCEDURE v2_queue_instead_of_update();
+
+            CREATE OR REPLACE TRIGGER v2_completed_job_instead_of_update_trigger
+                INSTEAD OF UPDATE ON v2_completed_job
+                FOR EACH ROW
+            EXECUTE PROCEDURE v2_completed_job_instead_of_update();
+
+            DROP FUNCTION v2_queue_instead_of_update_overlay() CASCADE;
+            DROP FUNCTION v2_completed_job_instead_of_update_overlay() CASCADE;
+            DROP FUNCTION v2_job_completed_before_insert() CASCADE;
+            DROP FUNCTION v2_job_flow_runtime_before_insert() CASCADE;
+            DROP FUNCTION v2_job_flow_runtime_before_update() CASCADE;
+            DROP FUNCTION v2_job_queue_after_insert() CASCADE;
+            DROP FUNCTION v2_job_queue_before_insert() CASCADE;
+            DROP FUNCTION v2_job_queue_before_update() CASCADE;
+            DROP FUNCTION v2_job_runtime_before_insert() CASCADE;
+            DROP FUNCTION v2_job_runtime_before_update() CASCADE;
+
+            DROP VIEW completed_job, completed_job_view, job, queue, queue_view CASCADE;
+            "#,
+        )
+        .await?;
+    });
+    run_windmill_migration!("v2_finalize_job_queue", db, |tx| {
+        tx.execute(
+            r#"
+            ALTER TABLE v2_job_queue
+                DROP COLUMN __parent_job CASCADE,
+                DROP COLUMN __created_by CASCADE,
+                DROP COLUMN __script_hash CASCADE,
+                DROP COLUMN __script_path CASCADE,
+                DROP COLUMN __args CASCADE,
+                DROP COLUMN __logs CASCADE,
+                DROP COLUMN __raw_code CASCADE,
+                DROP COLUMN __canceled CASCADE,
+                DROP COLUMN __last_ping CASCADE,
+                DROP COLUMN __job_kind CASCADE,
+                DROP COLUMN __env_id CASCADE,
+                DROP COLUMN __schedule_path CASCADE,
+                DROP COLUMN __permissioned_as CASCADE,
+                DROP COLUMN __flow_status CASCADE,
+                DROP COLUMN __raw_flow CASCADE,
+                DROP COLUMN __is_flow_step CASCADE,
+                DROP COLUMN __language CASCADE,
+                DROP COLUMN __same_worker CASCADE,
+                DROP COLUMN __raw_lock CASCADE,
+                DROP COLUMN __pre_run_error CASCADE,
+                DROP COLUMN __email CASCADE,
+                DROP COLUMN __visible_to_owner CASCADE,
+                DROP COLUMN __mem_peak CASCADE,
+                DROP COLUMN __root_job CASCADE,
+                DROP COLUMN __leaf_jobs CASCADE,
+                DROP COLUMN __concurrent_limit CASCADE,
+                DROP COLUMN __concurrency_time_window_s CASCADE,
+                DROP COLUMN __timeout CASCADE,
+                DROP COLUMN __flow_step_id CASCADE,
+                DROP COLUMN __cache_ttl CASCADE;
+            "#,
+        )
+        .await?;
+    });
+    run_windmill_migration!("v2_finalize_job_completed", db, |tx| {
+        tx.execute(
+            r#"
+            ALTER TABLE v2_job_completed
+                DROP COLUMN __parent_job CASCADE,
+                DROP COLUMN __created_by CASCADE,
+                DROP COLUMN __created_at CASCADE,
+                DROP COLUMN __success CASCADE,
+                DROP COLUMN __script_hash CASCADE,
+                DROP COLUMN __script_path CASCADE,
+                DROP COLUMN __args CASCADE,
+                DROP COLUMN __logs CASCADE,
+                DROP COLUMN __raw_code CASCADE,
+                DROP COLUMN __canceled CASCADE,
+                DROP COLUMN __job_kind CASCADE,
+                DROP COLUMN __env_id CASCADE,
+                DROP COLUMN __schedule_path CASCADE,
+                DROP COLUMN __permissioned_as CASCADE,
+                DROP COLUMN __raw_flow CASCADE,
+                DROP COLUMN __is_flow_step CASCADE,
+                DROP COLUMN __language CASCADE,
+                DROP COLUMN __is_skipped CASCADE,
+                DROP COLUMN __raw_lock CASCADE,
+                DROP COLUMN __email CASCADE,
+                DROP COLUMN __visible_to_owner CASCADE,
+                DROP COLUMN __tag CASCADE,
+                DROP COLUMN __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')"
@@ -360,7 +482,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
     //     tx.commit().await?;
     // }
 
-    run_windmill_migration!("fix_job_completed_index_2", &db, {
+    run_windmill_migration!("fix_job_completed_index_2", &db, |tx| {
         //     sqlx::query(
         //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new_2 ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, created_at DESC)"
         // ).execute(db).await?;
@@ -380,7 +502,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
         .await?;
     });
 
-    run_windmill_migration!("fix_job_completed_index_3", &db, {
+    run_windmill_migration!("fix_job_completed_index_3", &db, |tx| {
         sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_job_on_schedule_path")
             .execute(db)
             .await?;
@@ -398,7 +520,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
             .await?;
     });
 
-    run_windmill_migration!("fix_job_index_1", &db, {
+    run_windmill_migration!("fix_job_index_1", &db, |tx| {
         let migration_job_name = "fix_job_completed_index_4";
         let mut i = 1;
         tracing::info!("step {i} of {migration_job_name} migration");
@@ -479,7 +601,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
             .await?;
     });
 
-    run_windmill_migration!("fix_labeled_jobs_index", &db, {
+    run_windmill_migration!("fix_labeled_jobs_index", &db, |tx| {
         tracing::info!("Special migration to add index concurrently on job labels 2");
         sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS labeled_jobs_on_jobs")
             .execute(db)
diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs
index a6c3257bf5..d647032167 100644
--- a/backend/windmill-common/src/worker.rs
+++ b/backend/windmill-common/src/worker.rs
@@ -12,7 +12,10 @@ use std::{
     io::Write,
     path::{Component, Path, PathBuf},
     str::FromStr,
-    sync::{atomic::AtomicBool, Arc},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
 };
 use tokio::sync::RwLock;
 use windmill_macros::annotations;
@@ -103,6 +106,8 @@ lazy_static::lazy_static! {
     pub static ref DISABLE_FLOW_SCRIPT: bool = std::env::var("DISABLE_FLOW_SCRIPT").ok().is_some_and(|x| x == "1" || x == "true");
 }
 
+pub static MIN_VERSION_IS_LATEST: AtomicBool = AtomicBool::new(false);
+
 pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
     if wc.worker_tags.len() == 0 {
         tracing::error!("Empty tags in worker tags, skipping");
@@ -609,6 +614,8 @@ pub async fn update_min_version<'c, E: sqlx::Executor<'c, Database = sqlx::Postg
         tracing::info!("Minimal worker version: {min_version}");
     }
 
+    MIN_VERSION_IS_LATEST.store(min_version == cur_version, Ordering::Relaxed);
+
     *MIN_VERSION_IS_AT_LEAST_1_427.write().await = min_version >= Version::new(1, 427, 0);
     *MIN_VERSION_IS_AT_LEAST_1_432.write().await = min_version >= Version::new(1, 432, 0);
     *MIN_VERSION_IS_AT_LEAST_1_440.write().await = min_version >= Version::new(1, 440, 0);