From 97ed4a539b9b8fcaa5b96df3366fbd1ee4d19adc Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 23 Aug 2025 23:42:39 +0100 Subject: [PATCH] nits cleanup + faster script index #6450 --- ...23e2aa39891931516426f65b229314a0cee1.json} | 4 +- ...0cb549a34b96554ae1872355b90304f5dcb76.json | 4 +- ...b63efc585942b30c9e853f2a5cff4cc9916cd.json | 12 + ...29218a9d459ee1c2b36a28b110e9741ab623.json} | 4 +- ...087805dad2b6fb1aa194ac2a2026fa24ecd0.json} | 7 +- .../windmill-parser-py-imports/src/lib.rs | 3 +- backend/windmill-api/src/db.rs | 587 +---------------- backend/windmill-api/src/lib.rs | 3 +- backend/windmill-api/src/live_migrations.rs | 613 ++++++++++++++++++ backend/windmill-api/src/users.rs | 18 +- backend/windmill-common/src/cache.rs | 1 + backend/windmill-common/src/lib.rs | 10 +- backend/windmill-queue/src/schedule.rs | 1 + backend/windmill-worker/src/ai_executor.rs | 2 +- backend/windmill-worker/src/bun_executor.rs | 2 +- .../windmill-worker/src/dedicated_worker.rs | 3 +- 16 files changed, 666 insertions(+), 608 deletions(-) rename backend/.sqlx/{query-53e7243abd724816fb8d09c63b7ffa65f1cd622a989f5cefedbbf3c143b387c4.json => query-2d5f58dd2aff3bd49f3891ae76df23e2aa39891931516426f65b229314a0cee1.json} (72%) create mode 100644 backend/.sqlx/query-c1058d8816d139c63dd9c4a075ab63efc585942b30c9e853f2a5cff4cc9916cd.json rename backend/.sqlx/{query-443bd83bcea1d37c79cb080095343c98104529879f991c49585cd181e34aa827.json => query-c7cae4cf872fce0a989cf89aa35929218a9d459ee1c2b36a28b110e9741ab623.json} (58%) rename backend/.sqlx/{query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json => query-e64f7044c74e96c2338580562f6b087805dad2b6fb1aa194ac2a2026fa24ecd0.json} (88%) create mode 100644 backend/windmill-api/src/live_migrations.rs diff --git a/backend/.sqlx/query-53e7243abd724816fb8d09c63b7ffa65f1cd622a989f5cefedbbf3c143b387c4.json b/backend/.sqlx/query-2d5f58dd2aff3bd49f3891ae76df23e2aa39891931516426f65b229314a0cee1.json similarity index 72% rename from backend/.sqlx/query-53e7243abd724816fb8d09c63b7ffa65f1cd622a989f5cefedbbf3c143b387c4.json rename to backend/.sqlx/query-2d5f58dd2aff3bd49f3891ae76df23e2aa39891931516426f65b229314a0cee1.json index 95c51c2a44..9d2440990e 100644 --- a/backend/.sqlx/query-53e7243abd724816fb8d09c63b7ffa65f1cd622a989f5cefedbbf3c143b387c4.json +++ b/backend/.sqlx/query-2d5f58dd2aff3bd49f3891ae76df23e2aa39891931516426f65b229314a0cee1.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT workspace_id as workspace, path, summary, description, schema FROM script as o WHERE created_at = (select max(created_at) from script where o.path = path and workspace_id = $1) and workspace_id = $1", + "query": "SELECT workspace_id as workspace, path, summary, description, schema FROM script as o \n WHERE created_at = (select max(created_at) from script where o.path = path and workspace_id = $1 AND archived = false) \n AND workspace_id = $1 and archived = false", "describe": { "columns": [ { @@ -42,5 +42,5 @@ true ] }, - "hash": "53e7243abd724816fb8d09c63b7ffa65f1cd622a989f5cefedbbf3c143b387c4" + "hash": "2d5f58dd2aff3bd49f3891ae76df23e2aa39891931516426f65b229314a0cee1" } diff --git a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json index 99269c9851..54e94cfb8f 100644 --- a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json +++ b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json @@ -18,8 +18,8 @@ "Left": [] }, "nullable": [ - true, - false + false, + true ] }, "hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76" diff --git a/backend/.sqlx/query-c1058d8816d139c63dd9c4a075ab63efc585942b30c9e853f2a5cff4cc9916cd.json b/backend/.sqlx/query-c1058d8816d139c63dd9c4a075ab63efc585942b30c9e853f2a5cff4cc9916cd.json new file mode 100644 index 0000000000..450d18d524 --- /dev/null +++ b/backend/.sqlx/query-c1058d8816d139c63dd9c4a075ab63efc585942b30c9e853f2a5cff4cc9916cd.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "CREATE INDEX CONCURRENTLY IF NOT EXISTS script_not_archived ON script (workspace_id, path, created_at DESC) where archived = false;", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "c1058d8816d139c63dd9c4a075ab63efc585942b30c9e853f2a5cff4cc9916cd" +} diff --git a/backend/.sqlx/query-443bd83bcea1d37c79cb080095343c98104529879f991c49585cd181e34aa827.json b/backend/.sqlx/query-c7cae4cf872fce0a989cf89aa35929218a9d459ee1c2b36a28b110e9741ab623.json similarity index 58% rename from backend/.sqlx/query-443bd83bcea1d37c79cb080095343c98104529879f991c49585cd181e34aa827.json rename to backend/.sqlx/query-c7cae4cf872fce0a989cf89aa35929218a9d459ee1c2b36a28b110e9741ab623.json index 65b21050c0..0946fa4006 100644 --- a/backend/.sqlx/query-443bd83bcea1d37c79cb080095343c98104529879f991c49585cd181e34aa827.json +++ b/backend/.sqlx/query-c7cae4cf872fce0a989cf89aa35929218a9d459ee1c2b36a28b110e9741ab623.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT content FROM script WHERE path = $1 AND workspace_id = $2\n AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND\n workspace_id = $2)\n ", + "query": "\n SELECT content FROM script WHERE path = $1 AND workspace_id = $2\n AND archived = false ORDER BY created_at DESC LIMIT 1\n ", "describe": { "columns": [ { @@ -19,5 +19,5 @@ false ] }, - "hash": "443bd83bcea1d37c79cb080095343c98104529879f991c49585cd181e34aa827" + "hash": "c7cae4cf872fce0a989cf89aa35929218a9d459ee1c2b36a28b110e9741ab623" } diff --git a/backend/.sqlx/query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json b/backend/.sqlx/query-e64f7044c74e96c2338580562f6b087805dad2b6fb1aa194ac2a2026fa24ecd0.json similarity index 88% rename from backend/.sqlx/query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json rename to backend/.sqlx/query-e64f7044c74e96c2338580562f6b087805dad2b6fb1aa194ac2a2026fa24ecd0.json index 9560d17efa..4caab26834 100644 --- a/backend/.sqlx/query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json +++ b/backend/.sqlx/query-e64f7044c74e96c2338580562f6b087805dad2b6fb1aa194ac2a2026fa24ecd0.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)", + "query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script\n WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false)\n ORDER BY created_at DESC LIMIT 1", "describe": { "columns": [ { @@ -98,7 +98,8 @@ "parameters": { "Left": [ "Text", - "Text" + "Text", + "Bool" ] }, "nullable": [ @@ -116,5 +117,5 @@ false ] }, - "hash": "61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4" + "hash": "e64f7044c74e96c2338580562f6b087805dad2b6fb1aa194ac2a2026fa24ecd0" } diff --git a/backend/parsers/windmill-parser-py-imports/src/lib.rs b/backend/parsers/windmill-parser-py-imports/src/lib.rs index 97ac1c848f..3b181f1078 100644 --- a/backend/parsers/windmill-parser-py-imports/src/lib.rs +++ b/backend/parsers/windmill-parser-py-imports/src/lib.rs @@ -489,8 +489,7 @@ async fn parse_python_imports_inner( let code = sqlx::query_scalar!( r#" SELECT content FROM script WHERE path = $1 AND workspace_id = $2 - AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND - workspace_id = $2) + AND archived = false ORDER BY created_at DESC LIMIT 1 "#, &rpath, w_id diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 7fa7c61386..6df8633772 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -6,8 +6,6 @@ * LICENSE-AGPL for a copy of the license. */ -use std::time::Duration; - use futures::FutureExt; use sqlx::{ migrate::{Migrate, MigrateError}, @@ -17,11 +15,11 @@ use sqlx::{ use tokio::task::JoinHandle; use windmill_audit::audit_oss::{AuditAuthor, AuditAuthorable}; +use windmill_common::utils::generate_lock_id; use windmill_common::{ db::{Authable, Authed}, error::Error, }; -use windmill_common::{utils::generate_lock_id, worker::MIN_VERSION_IS_AT_LEAST_1_461}; pub type DB = Pool; @@ -59,7 +57,7 @@ lazy_static::lazy_static! { ].into_iter().collect(); } -struct CustomMigrator { +pub struct CustomMigrator { inner: PoolConnection, } impl Migrate for CustomMigrator { @@ -243,586 +241,7 @@ pub async fn migrate(db: &DB) -> Result>, Error> { Err(err) => Err(err), }?; - if let Err(err) = fix_flow_versioning_migration(&mut custom_migrator, db).await { - tracing::error!("Could not apply flow versioning fix migration: {err:#}"); - } - - let db2 = db.clone(); - let _ = tokio::task::spawn(async move { - if let Err(err) = fix_job_completed_index(&db2).await { - tracing::error!("Could not apply job completed index fix migration: {err:#}"); - } - }); - - 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.read().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) -} - -async fn fix_flow_versioning_migration( - migrator: &mut CustomMigrator, - db: &DB, -) -> Result<(), Error> { - let has_done_migration = sqlx::query_scalar!( - "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')", - ) - .fetch_one(db) - .await? - .unwrap_or(false); - - if !has_done_migration { - migrator.lock().await?; - - if migrator - .list_applied_migrations() - .await? - .iter() - .any(|x| x.version == 20240630102146) - { - let has_done_migration = sqlx::query_scalar!( - "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')", - ) - .fetch_one(db) - .await? - .unwrap_or(false); - - if !has_done_migration { - let query = include_str!("../../custom_migrations/fix_flow_versioning_2.sql"); - tracing::info!("Applying fix_flow_versioning_2.sql"); - let mut tx: sqlx::Transaction<'_, Postgres> = db.begin().await?; - tx.execute(query).await?; - tracing::info!("Applied fix_flow_versioning_2.sql"); - sqlx::query!( - "INSERT INTO windmill_migrations (name) VALUES ('fix_flow_versioning_2')" - ) - .execute(&mut *tx) - .await?; - tx.commit().await?; - } - } - - migrator.unlock().await?; - } - Ok(()) -} - -async fn has_done_migration(db: &DB, migration_job_name: &str) -> bool { - sqlx::query_scalar!( - "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = $1)", - migration_job_name - ) - .fetch_one(db) - .await - .ok() - .flatten() - .unwrap_or(false) -} - -macro_rules! run_windmill_migration { - ($migration_job_name:expr, $db:expr, |$tx:ident| $code:block) => { - { - let migration_job_name = $migration_job_name; - let db: &Pool = $db; - - let has_done = has_done_migration(db, migration_job_name).await; - if !has_done { - tracing::info!("Applying {migration_job_name} migration"); - 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) - .await - .map_err(|e| { - tracing::error!("Error acquiring {migration_job_name} lock: {e:#}"); - sqlx::migrate::MigrateError::Execute(e) - })? - .unwrap_or(false); - - if !r { - tracing::info!("PG {migration_job_name} lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)"); - drop($tx); - tokio::time::sleep(std::time::Duration::from_secs(5)).await; - $tx = db.begin().await?; - } - } - tracing::info!("acquired lock for {migration_job_name}"); - - let has_done = has_done_migration(db, migration_job_name).await; - - if !has_done { - - $code - - sqlx::query!( - "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING", - migration_job_name - ) - .execute(&mut *$tx) - .await?; - tracing::info!("Finished applying {migration_job_name} migration"); - } else { - tracing::debug!("migration {migration_job_name} already done"); - } - - let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") - .execute(&mut *$tx) - .await?; - $tx.commit().await?; - tracing::info!("released lock for {migration_job_name}"); - } else { - tracing::debug!("migration {migration_job_name} already done"); - - } - } - }; -} - -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')" - // ) - // .fetch_one(db) - // .await? - // .unwrap_or(false); - // if !has_done_migration { - // tracing::info!("Applying fix_job_completed_index migration"); - // 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) - // .await - // .map_err(|e| { - // tracing::error!("Error acquiring fix_job_completed_index lock: {e:#}"); - // sqlx::migrate::MigrateError::Execute(e) - // })? - // .unwrap_or(false); - // if !r { - // tracing::info!("PG fix_job_completed_index_migration lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)"); - // tokio::time::sleep(std::time::Duration::from_secs(5)).await; - // } - // } - // // sqlx::query( - // // "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new ON completed_job (workspace_id, job_kind, is_skipped, is_flow_step, created_at DESC, started_at DESC)" - // // ).execute(db).await?; - - // sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at") - // .execute(db) - // .await?; - - // sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index') ON CONFLICT DO NOTHING") - // .execute(&mut *tx) - // .await?; - // let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") - // .execute(&mut *tx) - // .await?; - // tx.commit().await?; - // } - - 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?; - - // sqlx::query( - // "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_started_at_new ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, started_at DESC)" - // ).execute(db).await?; - - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at") - .execute(db) - .await?; - - sqlx::query( - "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new", - ) - .execute(db) - .await?; - }); - - 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?; - - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue") - .execute(db) - .await?; - - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index") - .execute(db) - .await?; - - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created") - .execute(db) - .await?; - }); - - run_windmill_migration!("fix_job_index_1_II", &db, |tx| { - let migration_job_name = "fix_job_index_1_II"; - let mut i = 1; - tracing::info!("step {i} of {migration_job_name} migration"); - sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_3 ON v2_job (workspace_id, created_at DESC)") - .execute(db) - .await?; - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_8 ON v2_job (workspace_id, created_at DESC) where kind in ('deploymentcallback') AND parent_job IS NULL") - .execute(db) - .await?; - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_9 ON v2_job (workspace_id, created_at DESC) where kind in ('dependencies', 'flowdependencies', 'appdependencies') AND parent_job IS NULL") - .execute(db) - .await?; - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_5 ON v2_job (workspace_id, created_at DESC) where kind in ('preview', 'flowpreview') AND parent_job IS NULL") - .execute(db) - .await?; - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query!("create index concurrently if not exists ix_completed_job_workspace_id_started_at_new_2 ON v2_job_completed (workspace_id, started_at DESC)") - .execute(db) - .await?; - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query!("create index concurrently if not exists ix_job_root_job_index_by_path_2 ON v2_job (workspace_id, runnable_path, created_at desc) WHERE parent_job IS NULL") - .execute(db) - .await?; - - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path_2") - .execute(db) - .await?; - - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query!("create index concurrently if not exists ix_job_created_at ON v2_job (created_at DESC)") - .execute(db) - .await?; - - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query( - "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new_2", - ) - .execute(db) - .await?; - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query( - "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new", - ) - .execute(db) - .await?; - i += 1; - tracing::info!("step {i} of {migration_job_name} migration"); - - sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path") - .execute(db) - .await?; - }); - - 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) - .await?; - sqlx::query!( - "CREATE INDEX CONCURRENTLY labeled_jobs_on_jobs ON v2_job_completed USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_labels'" - ).execute(db).await?; - }); - - run_windmill_migration!("v2_labeled_jobs_index", &db, |tx| { - tracing::info!("Special migration to add index concurrently on job labels"); - sqlx::query!( - "CREATE INDEX CONCURRENTLY ix_v2_job_labels ON v2_job - USING GIN (labels) - WHERE labels IS NOT NULL" - ) - .execute(db) - .await?; - }); - - run_windmill_migration!("v2_jobs_rls", &db, |tx| { - sqlx::query!("ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY") - .execute(db) - .await?; - }); - - run_windmill_migration!("v2_improve_v2_job_indices_ii", &db, |tx| { - sqlx::query!("create index concurrently if not exists ix_v2_job_workspace_id_created_at ON v2_job (workspace_id, created_at DESC) where kind in ('script', 'flow', 'singlescriptflow') AND parent_job IS NULL") - .execute(db) - .await?; - - sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_6") - .execute(db) - .await?; - - sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_7") - .execute(db) - .await?; - }); - - run_windmill_migration!("v2_improve_v2_queued_jobs_indices", &db, |tx| { - sqlx::query!("CREATE INDEX CONCURRENTLY IF NOT EXISTS queue_sort_v2 ON v2_job_queue (priority DESC NULLS LAST, scheduled_for, tag) WHERE running = false") - .execute(db) - .await?; - - // sqlx::query!("CREATE INDEX CONCURRENTLY queue_sort_2_v2 ON v2_job_queue (tag, priority DESC NULLS LAST, scheduled_for) WHERE running = false") - // .execute(db) - // .await?; - - sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort") - .execute(db) - .await?; - - sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort_2") - .execute(db) - .await?; - }); - - run_windmill_migration!("audit_timestamps", db, |tx| { - sqlx::query!( - "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_audit_timestamps ON audit (timestamp DESC)" - ) - .execute(db) - .await?; - }); - - run_windmill_migration!("job_completed_completed_at", db, |tx| { - sqlx::query!( - "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_completed_completed_at ON v2_job_completed (completed_at DESC)" - ) - .execute(db) - .await?; - }); - - run_windmill_migration!("alerts_by_workspace", db, |tx| { - sqlx::query!( - "CREATE INDEX CONCURRENTLY IF NOT EXISTS alerts_by_workspace ON alerts (workspace_id);" - ) - .execute(db) - .await?; - }); - - run_windmill_migration!("remove_redundant_log_file_index", db, |tx| { - sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS log_file_hostname_log_ts_idx") - .execute(db) - .await?; - }); - - run_windmill_migration!("v2_job_queue_suspend", db, |tx| { - sqlx::query!( - "CREATE INDEX CONCURRENTLY IF NOT EXISTS v2_job_queue_suspend ON v2_job_queue (workspace_id, suspend) WHERE suspend > 0;" - ) - .execute(db) - .await?; - }); - - run_windmill_migration!("audit_recent_login_activities", db, |tx| { - sqlx::query!( - "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_audit_recent_login_activities -ON audit (timestamp, username) -WHERE operation IN ('users.login', 'oauth.login', 'users.token.refresh');" - ) - .execute(db) - .await?; - }); - Ok(()) + return crate::live_migrations::custom_migrations(&mut custom_migrator, db).await; } #[derive(Clone, Debug, Default, Hash, Eq, PartialEq)] diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index a0c0a8d8d3..842d8af40c 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -103,6 +103,7 @@ mod inkeep_ee; mod inkeep_oss; mod inputs; mod integration; +mod live_migrations; #[cfg(feature = "postgres_trigger")] mod postgres_triggers; @@ -623,7 +624,7 @@ pub async fn run_server( .nest("/mqtt_triggers", mqtt_triggers_service) .nest("/sqs_triggers", sqs_triggers_service) .nest("/gcp_triggers", gcp_triggers_service) - .nest("/postgres_triggers", postgres_triggers_service), + .nest("/postgres_triggers", postgres_triggers_service), ) .nest("/workspaces", workspaces::global_service()) .nest( diff --git a/backend/windmill-api/src/live_migrations.rs b/backend/windmill-api/src/live_migrations.rs new file mode 100644 index 0000000000..0e3fb97e02 --- /dev/null +++ b/backend/windmill-api/src/live_migrations.rs @@ -0,0 +1,613 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2022 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +use sqlx::Postgres; +use std::time::Duration; +use tokio::task::JoinHandle; +use windmill_common::error::Error; +use windmill_common::worker::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>, Error> { + if let Err(err) = fix_flow_versioning_migration(migrator, db).await { + tracing::error!("Could not apply flow versioning fix migration: {err:#}"); + } + + let db2 = db.clone(); + let _ = tokio::task::spawn(async move { + if let Err(err) = fix_job_completed_index(&db2).await { + tracing::error!("Could not apply job completed index fix migration: {err:#}"); + } + }); + + 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.read().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) +} + +async fn fix_flow_versioning_migration( + migrator: &mut CustomMigrator, + db: &DB, +) -> Result<(), Error> { + let has_done_migration = sqlx::query_scalar!( + "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')", + ) + .fetch_one(db) + .await? + .unwrap_or(false); + + if !has_done_migration { + migrator.lock().await?; + + if migrator + .list_applied_migrations() + .await? + .iter() + .any(|x| x.version == 20240630102146) + { + let has_done_migration = sqlx::query_scalar!( + "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')", + ) + .fetch_one(db) + .await? + .unwrap_or(false); + + if !has_done_migration { + let query = include_str!("../../custom_migrations/fix_flow_versioning_2.sql"); + tracing::info!("Applying fix_flow_versioning_2.sql"); + let mut tx: sqlx::Transaction<'_, Postgres> = db.begin().await?; + tx.execute(query).await?; + tracing::info!("Applied fix_flow_versioning_2.sql"); + sqlx::query!( + "INSERT INTO windmill_migrations (name) VALUES ('fix_flow_versioning_2')" + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; + } + } + + migrator.unlock().await?; + } + Ok(()) +} + +async fn has_done_migration(db: &DB, migration_job_name: &str) -> bool { + sqlx::query_scalar!( + "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = $1)", + migration_job_name + ) + .fetch_one(db) + .await + .ok() + .flatten() + .unwrap_or(false) +} + +use sqlx::Pool; + +macro_rules! run_windmill_migration { + ($migration_job_name:expr, $db:expr, |$tx:ident| $code:block) => { + { + let migration_job_name = $migration_job_name; + let db: &Pool = $db; + + let has_done = has_done_migration(db, migration_job_name).await; + if !has_done { + tracing::info!("Applying {migration_job_name} migration"); + 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) + .await + .map_err(|e| { + tracing::error!("Error acquiring {migration_job_name} lock: {e:#}"); + sqlx::migrate::MigrateError::Execute(e) + })? + .unwrap_or(false); + + if !r { + tracing::info!("PG {migration_job_name} lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)"); + drop($tx); + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + $tx = db.begin().await?; + } + } + tracing::info!("acquired lock for {migration_job_name}"); + + let has_done = has_done_migration(db, migration_job_name).await; + + if !has_done { + + $code + + sqlx::query!( + "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING", + migration_job_name + ) + .execute(&mut *$tx) + .await?; + tracing::info!("Finished applying {migration_job_name} migration"); + } else { + tracing::debug!("migration {migration_job_name} already done"); + } + + let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") + .execute(&mut *$tx) + .await?; + $tx.commit().await?; + tracing::info!("released lock for {migration_job_name}"); + } else { + tracing::debug!("migration {migration_job_name} already done"); + + } + } + }; +} + +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')" + // ) + // .fetch_one(db) + // .await? + // .unwrap_or(false); + // if !has_done_migration { + // tracing::info!("Applying fix_job_completed_index migration"); + // 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) + // .await + // .map_err(|e| { + // tracing::error!("Error acquiring fix_job_completed_index lock: {e:#}"); + // sqlx::migrate::MigrateError::Execute(e) + // })? + // .unwrap_or(false); + // if !r { + // tracing::info!("PG fix_job_completed_index_migration lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)"); + // tokio::time::sleep(std::time::Duration::from_secs(5)).await; + // } + // } + // // sqlx::query( + // // "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new ON completed_job (workspace_id, job_kind, is_skipped, is_flow_step, created_at DESC, started_at DESC)" + // // ).execute(db).await?; + + // sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at") + // .execute(db) + // .await?; + + // sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index') ON CONFLICT DO NOTHING") + // .execute(&mut *tx) + // .await?; + // let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") + // .execute(&mut *tx) + // .await?; + // tx.commit().await?; + // } + + 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?; + + // sqlx::query( + // "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_started_at_new ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, started_at DESC)" + // ).execute(db).await?; + + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at") + .execute(db) + .await?; + + sqlx::query( + "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new", + ) + .execute(db) + .await?; + }); + + 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?; + + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue") + .execute(db) + .await?; + + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index") + .execute(db) + .await?; + + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created") + .execute(db) + .await?; + }); + + run_windmill_migration!("fix_job_index_1_II", &db, |tx| { + let migration_job_name = "fix_job_index_1_II"; + let mut i = 1; + tracing::info!("step {i} of {migration_job_name} migration"); + sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_3 ON v2_job (workspace_id, created_at DESC)") + .execute(db) + .await?; + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_8 ON v2_job (workspace_id, created_at DESC) where kind in ('deploymentcallback') AND parent_job IS NULL") + .execute(db) + .await?; + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_9 ON v2_job (workspace_id, created_at DESC) where kind in ('dependencies', 'flowdependencies', 'appdependencies') AND parent_job IS NULL") + .execute(db) + .await?; + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_5 ON v2_job (workspace_id, created_at DESC) where kind in ('preview', 'flowpreview') AND parent_job IS NULL") + .execute(db) + .await?; + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query!("create index concurrently if not exists ix_completed_job_workspace_id_started_at_new_2 ON v2_job_completed (workspace_id, started_at DESC)") + .execute(db) + .await?; + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query!("create index concurrently if not exists ix_job_root_job_index_by_path_2 ON v2_job (workspace_id, runnable_path, created_at desc) WHERE parent_job IS NULL") + .execute(db) + .await?; + + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path_2") + .execute(db) + .await?; + + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query!("create index concurrently if not exists ix_job_created_at ON v2_job (created_at DESC)") + .execute(db) + .await?; + + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query( + "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new_2", + ) + .execute(db) + .await?; + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query( + "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new", + ) + .execute(db) + .await?; + i += 1; + tracing::info!("step {i} of {migration_job_name} migration"); + + sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path") + .execute(db) + .await?; + }); + + 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) + .await?; + sqlx::query!( + "CREATE INDEX CONCURRENTLY labeled_jobs_on_jobs ON v2_job_completed USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_labels'" + ).execute(db).await?; + }); + + run_windmill_migration!("v2_labeled_jobs_index", &db, |tx| { + tracing::info!("Special migration to add index concurrently on job labels"); + sqlx::query!( + "CREATE INDEX CONCURRENTLY ix_v2_job_labels ON v2_job + USING GIN (labels) + WHERE labels IS NOT NULL" + ) + .execute(db) + .await?; + }); + + run_windmill_migration!("v2_jobs_rls", &db, |tx| { + sqlx::query!("ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY") + .execute(db) + .await?; + }); + + run_windmill_migration!("v2_improve_v2_job_indices_ii", &db, |tx| { + sqlx::query!("create index concurrently if not exists ix_v2_job_workspace_id_created_at ON v2_job (workspace_id, created_at DESC) where kind in ('script', 'flow', 'singlescriptflow') AND parent_job IS NULL") + .execute(db) + .await?; + + sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_6") + .execute(db) + .await?; + + sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_7") + .execute(db) + .await?; + }); + + run_windmill_migration!("v2_improve_v2_queued_jobs_indices", &db, |tx| { + sqlx::query!("CREATE INDEX CONCURRENTLY IF NOT EXISTS queue_sort_v2 ON v2_job_queue (priority DESC NULLS LAST, scheduled_for, tag) WHERE running = false") + .execute(db) + .await?; + + // sqlx::query!("CREATE INDEX CONCURRENTLY queue_sort_2_v2 ON v2_job_queue (tag, priority DESC NULLS LAST, scheduled_for) WHERE running = false") + // .execute(db) + // .await?; + + sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort") + .execute(db) + .await?; + + sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort_2") + .execute(db) + .await?; + }); + + run_windmill_migration!("audit_timestamps", db, |tx| { + sqlx::query!( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_audit_timestamps ON audit (timestamp DESC)" + ) + .execute(db) + .await?; + }); + + run_windmill_migration!("job_completed_completed_at", db, |tx| { + sqlx::query!( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_completed_completed_at ON v2_job_completed (completed_at DESC)" + ) + .execute(db) + .await?; + }); + + run_windmill_migration!("alerts_by_workspace", db, |tx| { + sqlx::query!( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS alerts_by_workspace ON alerts (workspace_id);" + ) + .execute(db) + .await?; + }); + + run_windmill_migration!("remove_redundant_log_file_index", db, |tx| { + sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS log_file_hostname_log_ts_idx") + .execute(db) + .await?; + }); + + run_windmill_migration!("v2_job_queue_suspend", db, |tx| { + sqlx::query!( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS v2_job_queue_suspend ON v2_job_queue (workspace_id, suspend) WHERE suspend > 0;" + ) + .execute(db) + .await?; + }); + + run_windmill_migration!("audit_recent_login_activities", db, |tx| { + sqlx::query!( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_audit_recent_login_activities +ON audit (timestamp, username) +WHERE operation IN ('users.login', 'oauth.login', 'users.token.refresh');" + ) + .execute(db) + .await?; + }); + + run_windmill_migration!("v2_script_lock_index", db, |tx| { + sqlx::query!( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS script_not_archived ON script (workspace_id, path, created_at DESC) where archived = false;" + ) + .execute(db) + .await?; + }); + Ok(()) +} diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index 16d969c236..c0b7c3d665 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -1327,8 +1327,9 @@ async fn update_workspace_user( eu.operator, eu.disabled, &mut tx, - Some(&authed) - ).await?; + Some(&authed), + ) + .await?; let user_email = sqlx::query_scalar!( "SELECT email FROM usr WHERE username = $1 AND workspace_id = $2", @@ -1589,7 +1590,14 @@ async fn delete_workspace_user( let email_to_delete = not_found_if_none(email_to_delete_o, "User", &username_to_delete)?; - delete_workspace_user_internal(&w_id, &username_to_delete, &email_to_delete, &mut tx, Some(&authed)).await?; + delete_workspace_user_internal( + &w_id, + &username_to_delete, + &email_to_delete, + &mut tx, + Some(&authed), + ) + .await?; tx.commit().await?; handle_deployment_metadata( @@ -2172,7 +2180,9 @@ async fn get_all_runnables( .collect::>(), ); let scripts = sqlx::query!( - "SELECT workspace_id as workspace, path, summary, description, schema FROM script as o WHERE created_at = (select max(created_at) from script where o.path = path and workspace_id = $1) and workspace_id = $1", workspace + "SELECT workspace_id as workspace, path, summary, description, schema FROM script as o + WHERE created_at = (select max(created_at) from script where o.path = path and workspace_id = $1 AND archived = false) + AND workspace_id = $1 and archived = false", workspace ) .fetch_all(&mut *tx) .await?; diff --git a/backend/windmill-common/src/cache.rs b/backend/windmill-common/src/cache.rs index 4129948268..f43d14845b 100644 --- a/backend/windmill-common/src/cache.rs +++ b/backend/windmill-common/src/cache.rs @@ -169,6 +169,7 @@ impl> FsBackedCach ), } } + // Cache path doesn't exist or import failed, generate the content. let data = Val::resolve(with.await?)?; // Try to export data to the file-system. diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index e081cb3ceb..adf58871ba 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -631,6 +631,7 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>( db: E, w_id: &str, script_path: &str, + require_locked: bool, ) -> error::Result<( scripts::ScriptHash, Option, @@ -646,11 +647,12 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>( String, )> { let r_o = sqlx::query!( - "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script where path = $1 AND workspace_id = $2 AND - created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND - deleted = false AND archived = false)", + "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script + WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false) + ORDER BY created_at DESC LIMIT 1", script_path, - w_id + w_id, + require_locked ) .fetch_optional(db) .await?; diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 6e8f8be8cb..7f07b6f65e 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -156,6 +156,7 @@ pub async fn push_scheduled_job<'c>( &mut *tx, &schedule.workspace_id, &schedule.script_path, + true, ) .await?; diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 806737ed19..c8ee996bc7 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -926,7 +926,7 @@ pub async fn handle_ai_agent_job( .await?; Ok(Some(hub_script.schema)) } else { - let hash = get_latest_hash_for_path(db, &job.workspace_id, path) + let hash = get_latest_hash_for_path(db, &job.workspace_id, path, true) .await? .0; // update module definition to use a fixed hash so all tool calls match the same schema diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 9f81ce798e..9cad2ebba7 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -758,7 +758,7 @@ pub async fn prebundle_bun_script( pub const BUN_BUNDLE_OBJECT_STORE_PREFIX: &str = "bun_bundle/"; async fn get_script_import_updated_at(db: &DB, w_id: &str, script_path: &str) -> Result { - let script_hash = get_latest_hash_for_path(db, w_id, script_path).await?; + let script_hash = get_latest_hash_for_path(db, w_id, script_path, false).await?; let last_updated_at = sqlx::query_scalar!( "SELECT created_at FROM script WHERE workspace_id = $1 AND hash = $2", w_id, diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index 1539c33c5c..1805b22230 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -622,8 +622,7 @@ async fn spawn_dedicated_worker( } else { sqlx::query_as::<_, (String, Option, Option, Option>, bool, Option)>( "SELECT content, lock, language, envs, codebase IS NOT NULL, hash FROM script WHERE path = $1 AND workspace_id = $2 AND - created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND - deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)", + archived = false AND lock IS not NULL AND lock_error_logs IS NULL ORDER BY created_at DESC LIMIT 1", ) .bind(&path) .bind(&w_id)