feat: Add slow query warnings to database calls

backend/src/monitor.rs

@@ -1226,6 +1226,7 @@ pub async fn load_value_from_global_settings(
         setting_name
     )
     .fetch_optional(db)
+    .warn_after_seconds(1)
     .await?
     .map(|x| x.value);
     Ok(r)
@@ -1475,6 +1476,7 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
             "SELECT created_at FROM metrics WHERE id LIKE 'queue_count_%' ORDER BY created_at DESC LIMIT 1"
         )
         .fetch_optional(db)
+        .warn_after_seconds(1)
         .await
         .unwrap_or(Some(chrono::Utc::now()));
 
@@ -1516,6 +1518,7 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
                     serde_json::json!(count)
                 )
                 .execute(db)
+                .warn_after_seconds(1)
                 .await
                 .ok();
                 if count > 0 {
@@ -1531,6 +1534,7 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
                         tag
                     )
                     .execute(db)
+                    .warn_after_seconds(1)
                     .await
                     .ok();
                 }
@@ -1547,6 +1551,7 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
         "DELETE FROM metrics WHERE id LIKE 'queue_%' AND created_at < NOW() - INTERVAL '14 day'"
     )
     .execute(db)
+    .warn_after_seconds(1)
     .await
     .ok();
 }
@@ -1740,6 +1745,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
             RESTART_LIMIT
         )
         .fetch_all(db)
+        .warn_after_seconds(3)
         .await
         .ok()
         .unwrap_or_else(|| vec![]);
@@ -1799,6 +1805,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
                 restart_message
             )
             .execute(db)
+            .warn_after_seconds(1)
             .await;
             tracing::error!(critical_error_message);
             report_critical_error(
@@ -1821,6 +1828,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
     AND running = true AND ping IS NULL AND same_worker = true AND worker IS NOT NULL GROUP BY worker",
         )
         .fetch_all(db)
+        .warn_after_seconds(3)
         .await
         .ok()
         .unwrap_or_else(|| vec![]);
@@ -1838,6 +1846,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
             &worker_ids[..]
         )
         .fetch_all(db)
+        .warn_after_seconds(3)
         .await
         .ok()
         .unwrap_or_else(|| vec![])
@@ -1867,6 +1876,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
         let jobs = sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE id = ANY($1)")
             .bind(&timeouts[..])
             .fetch_all(db)
+            .warn_after_seconds(3)
             .await
             .map_err(|e| tracing::error!("Error fetching same worker jobs: {:?}", e))
             .unwrap_or_default();
@@ -1881,6 +1891,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
     AND running = true  AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false")
         .bind(ZOMBIE_JOB_TIMEOUT.as_str())
         .fetch_all(db)
+        .warn_after_seconds(3)
         .await
         .ok()
             .unwrap_or_else(|| vec![])
@@ -1906,6 +1917,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
         sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE id = ANY($1)")
             .bind(&zombie_jobs_uuid_restart_limit_reached[..])
             .fetch_all(db)
+            .warn_after_seconds(3)
             .await
             .ok()
             .unwrap_or_else(|| vec![]);
@@ -2009,6 +2021,7 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
         FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str()
     )
     .fetch_all(db)
+    .warn_after_seconds(3)
     .await?;
 
     for flow in flows {
@@ -2035,6 +2048,7 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
             let concurrency_key =
                 sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", flow.id)
                     .fetch_optional(&mut *tx)
+                    .warn_after_seconds(1)
                     .await?;
 
             if let Some(key) = concurrency_key {
@@ -2047,6 +2061,7 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
                         flow.id.hyphenated().to_string()
                     )
                     .execute(&mut *tx)
+                    .warn_after_seconds(1)
                     .await?;
                 }
             }
@@ -2057,6 +2072,7 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
                 flow.id
             )
             .execute(&mut *tx)
+            .warn_after_seconds(1)
             .await?;
 
             tx.commit().await?;
@@ -2100,6 +2116,7 @@ Please check your worker logs for more details and feel free to report it to the
         FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str()
     )
     .fetch_all(db)
+    .warn_after_seconds(3)
     .await?;
 
     for flow in flows2 {

backend/windmill-api/src/capture.rs

@@ -275,6 +275,7 @@ async fn get_configs(
         matches!(runnable_kind, RunnableKind::Flow),
     )
     .fetch_all(&mut *tx)
+    .warn_after_seconds(3)
     .await?;
 
     tx.commit().await?;
@@ -314,7 +315,7 @@ async fn set_postgres_trigger_config(
 
     let query = drop_publication_query(&publication_name);
 
-    sqlx::query(&query).execute(&mut connection).await?;
+    sqlx::query(&query).execute(&mut connection).warn_after_seconds(3).await?;
 
     let query = create_publication_query(
         &publication_name,
@@ -447,6 +448,7 @@ async fn set_config(
         &authed.email,
     )
     .execute(&mut *tx)
+    .warn_after_seconds(1)
     .await?;
 
     tx.commit().await?;
@@ -484,6 +486,7 @@ async fn ping_config(
         trigger_kind as TriggerKind,
     )
     .execute(&mut *tx)
+    .warn_after_seconds(3)
     .await?;
 
     tx.commit().await?;
@@ -551,6 +554,7 @@ async fn list_captures(
         per_page as i64,
     )
     .fetch_all(&mut *tx)
+    .warn_after_seconds(3)
     .await?;
 
     tx.commit().await?;
@@ -584,6 +588,7 @@ async fn get_capture(
         &w_id,
     )
     .fetch_one(&mut *tx)
+    .warn_after_seconds(1)
     .await?;
 
     tx.commit().await?;
@@ -606,6 +611,7 @@ async fn delete_capture(
         id
     )
     .execute(&mut *tx)
+    .warn_after_seconds(1)
     .await?;
     tx.commit().await?;
     Ok(())
@@ -642,6 +648,7 @@ async fn move_captures_and_configs(
         matches!(runnable_kind, RunnableKind::Flow),
     )
     .execute(&mut *tx)
+    .warn_after_seconds(3)
     .await?;
 
     sqlx::query!(
@@ -701,6 +708,7 @@ pub async fn get_active_capture_owner_and_email(
         kind as &TriggerKind,
     )
     .fetch_optional(db)
+    .warn_after_seconds(1)
     .await?;
 
     let capture_config = not_found_if_none(
@@ -759,6 +767,7 @@ async fn get_capture_trigger_config_and_owner<T: DeserializeOwned>(
         matches!(kind, TriggerKind::Gcp)
     )
     .fetch_optional(db)
+    .warn_after_seconds(1)
     .await?;
 
     let capture_config = not_found_if_none(
@@ -811,6 +820,7 @@ async fn clear_captures_history(db: &DB, w_id: &str) -> Result<()> {
             KEEP_LAST,
         )
         .execute(db)
+        .warn_after_seconds(3)
         .await?;
     }
     Ok(())
@@ -846,6 +856,7 @@ pub async fn insert_capture_payload(
         owner,
     )
     .execute(db)
+    .warn_after_seconds(1)
     .await?;
 
     clear_captures_history(db, &w_id).await?;

backend/windmill-api/src/configs.rs

@@ -48,6 +48,7 @@ async fn list_worker_groups(
     let mut configs_raw =
         sqlx::query_as!(Config, "SELECT * FROM config WHERE name LIKE 'worker__%'")
             .fetch_all(&db)
+            .warn_after_seconds(1)
             .await?;
     // Remove the 'worker__' prefix from all config names
     for config in configs_raw.iter_mut() {
@@ -103,6 +104,7 @@ async fn get_config(
 
     let config = sqlx::query_as!(Config, "SELECT * FROM config WHERE name = $1", name)
         .fetch_optional(&db)
+        .warn_after_seconds(1)
         .await?
         .map(|c| c.config);
 
@@ -132,6 +134,7 @@ async fn update_config(
         config
     )
     .execute(&mut *tx)
+    .warn_after_seconds(3)
     .await?;
 
     audit_log(
@@ -159,6 +162,7 @@ async fn delete_config(
 
     let deleted = sqlx::query!("DELETE FROM config WHERE name = $1 RETURNING name", name)
         .fetch_all(&db)
+        .warn_after_seconds(3)
         .await?;
 
     audit_log(
@@ -201,6 +205,7 @@ async fn list_autoscaling_events(
         worker_group
     )
     .fetch_all(&db)
+    .warn_after_seconds(1)
     .await?;
     Ok(Json(events))
 }
@@ -213,6 +218,7 @@ async fn list_configs(
     require_super_admin(&db, &authed.email).await?;
     let configs = sqlx::query_as!(Config, "SELECT name, config FROM config")
         .fetch_all(&db)
+        .warn_after_seconds(1)
         .await?;
     Ok(Json(configs))
 }

backend/windmill-api/src/db.rs

@@ -29,6 +29,7 @@ async fn current_database(conn: &mut PgConnection) -> Result<String, MigrateErro
     // language=SQL
     Ok(sqlx::query_scalar("SELECT current_database()")
         .fetch_one(conn)
+        .warn_after_seconds(1)
         .await?)
 }
 
@@ -96,6 +97,7 @@ impl Migrate for CustomMigrator {
 
             let pid = sqlx::query_scalar!("SELECT pg_backend_pid()")
                 .fetch_one(&mut *self.inner)
+                .warn_after_seconds(1)
                 .await?;
             tracing::info!("Acquiring global PG lock for potential migration with pid: {pid:?}");
             let database_name = current_database(&mut *self.inner).await?;
@@ -106,6 +108,7 @@ impl Migrate for CustomMigrator {
             while !r {
                 r = sqlx::query_scalar!("SELECT pg_try_advisory_lock($1)", lock_id)
                     .fetch_one(&mut *self.inner)
+                    .warn_after_seconds(1)
                     .await
                     .map_err(|e| {
                         tracing::error!("Error acquiring lock: {e:#}");
@@ -138,6 +141,7 @@ impl Migrate for CustomMigrator {
             let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
                 .bind(lock_id)
                 .execute(&mut *self.inner)
+                .warn_after_seconds(1)
                 .await?;
 
             tracing::info!("Released PG lock");
@@ -166,6 +170,7 @@ impl Migrate for CustomMigrator {
 
                 self.inner
                     .execute(&**migration_sql)
+                    .warn_after_seconds(10)
                     .await?;
                 let _ = sqlx::query(
                     r#"
@@ -177,6 +182,7 @@ impl Migrate for CustomMigrator {
                 .bind(&*migration.description)
                 .bind(&*migration.checksum)
                 .execute(&mut *self.inner)
+                .warn_after_seconds(10)
                 .await?;
                 return Ok(std::time::Duration::from_secs(0));
             } else {
@@ -205,6 +211,7 @@ pub async fn migrate(db: &DB) -> Result<Option<JoinHandle<()>>, Error> {
 
     if let Err(err) = sqlx::query!("DELETE FROM _sqlx_migrations WHERE version=20250131115248")
         .execute(db)
+        .warn_after_seconds(10)
         .await
     {
         tracing::info!("Could not remove sqlx migration with version=20250131115248: {err:#}");
@@ -213,6 +220,7 @@ pub async fn migrate(db: &DB) -> Result<Option<JoinHandle<()>>, Error> {
     // Remove the migration `v2_fix_no_runtime` in favor of `v2_fix_no_runtime_2`.
     if let Err(err) = sqlx::query!("DELETE FROM _sqlx_migrations WHERE version=20250201145632")
         .execute(db)
+        .warn_after_seconds(10)
         .await
     {
         tracing::info!("Could not remove sqlx migration with version=20250201145632: {err:#}");
@@ -223,6 +231,7 @@ pub async fn migrate(db: &DB) -> Result<Option<JoinHandle<()>>, Error> {
         "DELETE FROM _sqlx_migrations WHERE version=20250201145630 OR version=20250201145631"
     )
     .execute(db)
+    .warn_after_seconds(10)
     .await
     {
         tracing::info!("Could not remove sqlx migration with version=[20250201145630, 20250201145631] : {err:#}");
@@ -289,6 +298,7 @@ async fn fix_flow_versioning_migration(
         "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')",
     )
     .fetch_one(db)
+    .warn_after_seconds(1)
     .await?
     .unwrap_or(false);
 
@@ -305,6 +315,7 @@ async fn fix_flow_versioning_migration(
                 "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')",
             )
             .fetch_one(db)
+            .warn_after_seconds(1)
             .await?
             .unwrap_or(false);
 
@@ -312,12 +323,13 @@ async fn fix_flow_versioning_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?;
+                tx.execute(query).warn_after_seconds(10).await?;
                 tracing::info!("Applied fix_flow_versioning_2.sql");
                 sqlx::query!(
                     "INSERT INTO windmill_migrations (name) VALUES ('fix_flow_versioning_2')"
                 )
                 .execute(&mut *tx)
+                .warn_after_seconds(10)
                 .await?;
                 tx.commit().await?;
             }
@@ -334,6 +346,7 @@ async fn has_done_migration(db: &DB, migration_job_name: &str) -> bool {
         migration_job_name
     )
     .fetch_one(db)
+    .warn_after_seconds(1)
     .await
     .ok()
     .flatten()
@@ -354,6 +367,7 @@ macro_rules! run_windmill_migration {
                 while !r {
                     r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)")
                         .fetch_one(&mut *$tx)
+                        .warn_after_seconds(1)
                         .await
                         .map_err(|e| {
                             tracing::error!("Error acquiring {migration_job_name} lock: {e:#}");
@@ -381,6 +395,7 @@ macro_rules! run_windmill_migration {
                         migration_job_name
                     )
                     .execute(&mut *$tx)
+                    .warn_after_seconds(10)
                     .await?;
                     tracing::info!("Finished applying {migration_job_name} migration");
                 } else {
@@ -389,6 +404,7 @@ macro_rules! run_windmill_migration {
 
                 let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
                     .execute(&mut *$tx)
+                    .warn_after_seconds(1)
                     .await?;
                 $tx.commit().await?;
                 tracing::info!("released lock for {migration_job_name}");
@@ -408,6 +424,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
             ALTER TABLE v2_job_queue DISABLE ROW LEVEL SECURITY;
             "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
 
@@ -418,6 +435,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
             ALTER TABLE v2_job_completed DISABLE ROW LEVEL SECURITY;
             "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
 
@@ -428,6 +446,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
             DROP FUNCTION IF EXISTS v2_job_after_update CASCADE;
         "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
 
@@ -439,6 +458,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
             DROP FUNCTION IF EXISTS v2_job_completed_before_update CASCADE;
             "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
 
@@ -451,6 +471,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
             DROP FUNCTION IF EXISTS v2_job_queue_before_update CASCADE;
             "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
 
@@ -462,6 +483,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
             DROP FUNCTION IF EXISTS v2_job_runtime_before_update CASCADE;
             "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
 
@@ -473,6 +495,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
             DROP FUNCTION IF EXISTS v2_job_status_before_update CASCADE;
             "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
 
@@ -482,6 +505,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
             DROP VIEW IF EXISTS completed_job, completed_job_view, job, queue, queue_view CASCADE;
             "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
 
@@ -522,6 +546,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
                 DROP COLUMN IF EXISTS __cache_ttl CASCADE;
             "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
     run_windmill_migration!("v2_finalize_job_completed", db, |tx| {
@@ -554,6 +579,7 @@ async fn v2_finalize(db: &DB) -> Result<(), Error> {
                 DROP COLUMN IF EXISTS __priority CASCADE;
             "#,
         )
+        .warn_after_seconds(10)
         .await?;
     });
 
@@ -609,34 +635,40 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
 
         //     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?;
+        // ).execute(db).warn_after_seconds(10).await?;
 
         sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at")
             .execute(db)
+            .warn_after_seconds(10)
             .await?;
 
         sqlx::query(
             "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new",
         )
         .execute(db)
+        .warn_after_seconds(10)
         .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)
+            .warn_after_seconds(10)
             .await?;
 
         sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue")
             .execute(db)
+            .warn_after_seconds(10)
             .await?;
 
         sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index")
             .execute(db)
+            .warn_after_seconds(10)
             .await?;
 
         sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created")
             .execute(db)
+            .warn_after_seconds(10)
             .await?;
     });
 
@@ -646,36 +678,42 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
         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)
+                .warn_after_seconds(10)
                 .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)
+            .warn_after_seconds(10)
             .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)
+            .warn_after_seconds(10)
             .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)
+                .warn_after_seconds(10)
                 .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)
+                .warn_after_seconds(10)
                 .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)
+                .warn_after_seconds(10)
                 .await?;
 
         i += 1;
@@ -683,6 +721,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
 
         sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path_2")
             .execute(db)
+            .warn_after_seconds(10)
             .await?;
 
         i += 1;
@@ -690,6 +729,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
 
         sqlx::query!("create index concurrently if not exists ix_job_created_at ON v2_job  (created_at DESC)")
                     .execute(db)
+                    .warn_after_seconds(10)
                     .await?;
 
         i += 1;
@@ -699,6 +739,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
             "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new_2",
         )
         .execute(db)
+        .warn_after_seconds(10)
         .await?;
         i += 1;
         tracing::info!("step {i} of {migration_job_name} migration");
@@ -707,12 +748,14 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
             "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new",
         )
         .execute(db)
+        .warn_after_seconds(10)
         .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)
+            .warn_after_seconds(10)
             .await?;
     });
 
@@ -720,10 +763,11 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
         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)
+            .warn_after_seconds(10)
             .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?;
+        ).execute(db).warn_after_seconds(10).await?;
     });
 
     run_windmill_migration!("v2_labeled_jobs_index", &db, |tx| {
@@ -734,44 +778,53 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
                 WHERE labels IS NOT NULL"
         )
         .execute(db)
+        .warn_after_seconds(10)
         .await?;
     });
 
     run_windmill_migration!("v2_jobs_rls", &db, |tx| {
         sqlx::query!("ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY")
             .execute(db)
+            .warn_after_seconds(10)
             .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)
+                .warn_after_seconds(10)
                 .await?;
 
         sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_6")
             .execute(db)
+            .warn_after_seconds(10)
             .await?;
 
         sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_7")
             .execute(db)
+            .warn_after_seconds(10)
             .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)
+            .warn_after_seconds(10)
             .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)
+        //     .warn_after_seconds(10)
         //     .await?;
 
         sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort")
             .execute(db)
+            .warn_after_seconds(10)
             .await?;
 
         sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort_2")
             .execute(db)
+            .warn_after_seconds(10)
             .await?;
     });
 
@@ -780,6 +833,7 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
             "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_audit_timestamps ON audit (timestamp DESC)"
         )
         .execute(db)
+        .warn_after_seconds(10)
         .await?;
     });
     Ok(())

backend/windmill-api/src/postgres_triggers/trigger.rs

@@ -225,6 +225,7 @@ impl PostgresTrigger {
             self.path,
         )
         .fetch_optional(&db)
+        .warn_after_seconds(1)
         .await;
         match postgres_trigger {
             Ok(has_lock) => {
@@ -273,6 +274,7 @@ impl PostgresTrigger {
             *INSTANCE_NAME
         )
         .fetch_optional(db)
+        .warn_after_seconds(1)
         .await;
 
         match updated {
@@ -481,6 +483,7 @@ impl PostgresConfig {
                 "SELECT pubname FROM pg_publication WHERE pubname = {}",
                 quote_literal(&publication_name)
             ))
+            .warn_after_seconds(3)
             .await?;
 
         if !publication.row_exist() {
@@ -494,6 +497,7 @@ impl PostgresConfig {
                 "SELECT slot_name FROM pg_replication_slots WHERE slot_name = {}",
                 quote_literal(&replication_slot_name)
             ))
+            .warn_after_seconds(3)
             .await?;
 
         if !replication_slot.row_exist() {
@@ -504,6 +508,7 @@ impl PostgresConfig {
 
         let (logical_replication_stream, logical_replication_settings) = client
             .get_logical_replication_stream(&publication_name, &replication_slot_name)
+            .warn_after_seconds(3)
             .await?;
 
         Ok((logical_replication_stream, logical_replication_settings))
@@ -550,11 +555,11 @@ impl PostgresConfig {
 
                 let query = drop_logical_replication_slot_query(replication_slot_name);
 
-                let _ = sqlx::query(&query).execute(&mut connection).await;
+                let _ = sqlx::query(&query).execute(&mut connection).warn_after_seconds(3).await;
 
                 let query = drop_publication_query(publication_name);
 
-                let _ = sqlx::query(&query).execute(&mut connection).await;
+                let _ = sqlx::query(&query).execute(&mut connection).warn_after_seconds(3).await;
 
                 Ok(())
             }
@@ -782,6 +787,7 @@ impl CaptureConfigForPostgresTrigger {
             self.is_flow,
         )
         .fetch_optional(&db)
+        .warn_after_seconds(1)
         .await
         {
             Ok(has_lock) => {
@@ -829,6 +835,7 @@ impl CaptureConfigForPostgresTrigger {
             *INSTANCE_NAME
         )
         .fetch_optional(db)
+        .warn_after_seconds(1)
         .await
         {
             Ok(updated) => {
@@ -971,6 +978,7 @@ async fn listen_to_unlistened_database_events(
             "#
     )
     .fetch_all(db)
+    .warn_after_seconds(3)
     .await;
 
     match postgres_triggers {
@@ -1007,6 +1015,7 @@ async fn listen_to_unlistened_database_events(
             "#
     )
     .fetch_all(db)
+    .warn_after_seconds(3)
     .await;
 
     match postgres_triggers_capture {

backend/windmill-common/src/db.rs

@@ -92,6 +92,7 @@ impl UserDB {
         if let Some(schema) = PG_SCHEMA.as_ref() {
             sqlx::query(&format!("SET LOCAL search_path TO {}", schema))
                 .execute(&mut *tx)
+                .warn_after_seconds(1)
                 .await?;
         }
 
@@ -118,6 +119,7 @@ impl UserDB {
                 .join(",")
         )
         .execute(&mut *tx)
+        .warn_after_seconds(1)
         .await?;
 
         // set_session_context(

backend/windmill-common/src/global_settings.rs

@@ -120,6 +120,7 @@ pub async fn load_value_from_global_settings(
         setting_name
     )
     .fetch_optional(db)
+    .warn_after_seconds(1)
     .await?
     .map(|x| x.value);
     Ok(r)

backend/windmill-common/src/lib.rs

@@ -399,6 +399,7 @@ pub fn get_latest_deployed_hash_for_path<
                     w_id
                 )
                 .fetch_optional(&mut *conn)
+                .warn_after_seconds(1)
                 .await?;
 
                 let hash = utils::not_found_if_none(hash, "script", script_path)?;
@@ -440,6 +441,7 @@ pub async fn get_script_info_for_hash<'e, E: sqlx::PgExecutor<'e>>(
                     w_id
                 )
                 .fetch_optional(db)
+                .warn_after_seconds(1)
                 .await?;
 
             let info = utils::not_found_if_none(info, "script", &hash.to_string())?;
@@ -499,6 +501,7 @@ pub fn get_latest_flow_version_info_for_path<
                     w_id
                 )
                 .fetch_optional(&mut *conn)
+                .warn_after_seconds(1)
                 .await?;
 
                 let version = utils::not_found_if_none(version, "flow", path)?;
@@ -536,6 +539,7 @@ pub fn get_latest_flow_version_info_for_path<
                 version
             )
                 .fetch_optional(&mut *conn)
+                .warn_after_seconds(1)
                 .await?;
 
                 let info = utils::not_found_if_none(info, "flow", path)?;
@@ -574,6 +578,7 @@ pub async fn get_latest_hash_for_path<'c>(
         w_id
     )
     .fetch_optional(&mut **db)
+    .warn_after_seconds(1)
     .await?;
 
     let script = utils::not_found_if_none(r_o, "script", script_path)?;