starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-09 16:05:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

nits cleanup + faster script index #6450

Browse Source
This commit is contained in:
Ruben Fiszel
2025-08-23 22:42:39 +00:00
committed by GitHub
parent 8964896c13
commit 97ed4a539b
16 changed files with 666 additions and 608 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-53e7243abd724816fb8d09c63b7ffa65f1cd622a989f5cefedbbf3c143b387c4.json → backend/.sqlx/query-2d5f58dd2aff3bd49f3891ae76df23e2aa39891931516426f65b229314a0cee1.json
Generated
+2 -2
View File
@@ -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"
}
backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json
Generated
+2 -2
View File
@@ -18,8 +18,8 @@
"Left": []
},
"nullable": [
true,
false
false,
true
]
},
"hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76"
backend/.sqlx/query-c1058d8816d139c63dd9c4a075ab63efc585942b30c9e853f2a5cff4cc9916cd.json
Generated
+12
View File
@@ -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"
}
backend/.sqlx/query-443bd83bcea1d37c79cb080095343c98104529879f991c49585cd181e34aa827.json → backend/.sqlx/query-c7cae4cf872fce0a989cf89aa35929218a9d459ee1c2b36a28b110e9741ab623.json
Generated
+2 -2
View File
@@ -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"
}
backend/.sqlx/query-61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4.json → backend/.sqlx/query-e64f7044c74e96c2338580562f6b087805dad2b6fb1aa194ac2a2026fa24ecd0.json
Generated
+4 -3
View File
@@ -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"
}
backend/parsers/windmill-parser-py-imports/src/lib.rs
+1 -2
View File
@@ -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
backend/windmill-api/src/db.rs
+3 -584
View File
@@ -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<Postgres>;
@@ -59,7 +57,7 @@ lazy_static::lazy_static! {
].into_iter().collect();
}
struct CustomMigrator {
pub struct CustomMigrator {
inner: PoolConnection<Postgres>,
}
impl Migrate for CustomMigrator {
@@ -243,586 +241,7 @@ pub async fn migrate(db: &DB) -> Result<Option<JoinHandle<()>>, 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<Postgres> = $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)]
backend/windmill-api/src/lib.rs
+2 -1
View File
@@ -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(
backend/windmill-api/src/live_migrations.rs
+613
View File
@@ -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<Option<JoinHandle<()>>, 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<Postgres> = $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(())
}
backend/windmill-api/src/users.rs
+14 -4
View File
@@ -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::<Vec<_>>(),
);
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?;
backend/windmill-common/src/cache.rs
+1
View File
@@ -169,6 +169,7 @@ impl<Key: Eq + Hash + Item + Clone, Val: Export, Root: AsRef<Path>> 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.
backend/windmill-common/src/lib.rs
+6 -4
View File
@@ -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<Tag>,
@@ -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?;
backend/windmill-queue/src/schedule.rs
+1
View File
@@ -156,6 +156,7 @@ pub async fn push_scheduled_job<'c>(
&mut *tx,
&schedule.workspace_id,
&schedule.script_path,
true,
)
.await?;
backend/windmill-worker/src/ai_executor.rs
+1 -1
View File
@@ -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
backend/windmill-worker/src/bun_executor.rs
+1 -1
View File
@@ -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<String> {
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,
backend/windmill-worker/src/dedicated_worker.rs
+1 -2
View File
@@ -622,8 +622,7 @@ async fn spawn_dedicated_worker(
} else {
sqlx::query_as::<_, (String, Option<String>, Option<ScriptLang>, Option<Vec<String>>, bool, Option<ScriptHash>)>(
"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)