From 36a509ef7afcd5a9ed98708533d79e397a2417c9 Mon Sep 17 00:00:00 2001 From: "github-actions[bot] (aider)" Date: Wed, 14 May 2025 09:18:18 +0000 Subject: [PATCH] feat: Add slow query warnings to database calls --- backend/src/monitor.rs | 17 ++++++ backend/windmill-api/src/capture.rs | 13 +++- backend/windmill-api/src/configs.rs | 6 ++ backend/windmill-api/src/db.rs | 60 ++++++++++++++++++- .../src/postgres_triggers/trigger.rs | 13 +++- backend/windmill-common/src/db.rs | 2 + .../windmill-common/src/global_settings.rs | 1 + backend/windmill-common/src/lib.rs | 5 ++ 8 files changed, 111 insertions(+), 6 deletions(-) diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index cf04ff8c32..b69f60a5ea 100644 --- a/backend/src/monitor.rs +++ b/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) { "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) { 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) { tag ) .execute(db) + .warn_after_seconds(1) .await .ok(); } @@ -1547,6 +1551,7 @@ pub async fn expose_queue_metrics(db: &Pool) { "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, 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, 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, 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, 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, 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, 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, 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 { diff --git a/backend/windmill-api/src/capture.rs b/backend/windmill-api/src/capture.rs index 94b74ef9b9..9c015615e3 100644 --- a/backend/windmill-api/src/capture.rs +++ b/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( 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?; diff --git a/backend/windmill-api/src/configs.rs b/backend/windmill-api/src/configs.rs index 81b65216e0..addc494660 100644 --- a/backend/windmill-api/src/configs.rs +++ b/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)) } diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 557981d214..ae7f6cacb7 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -29,6 +29,7 @@ async fn current_database(conn: &mut PgConnection) -> Result Result>, 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>, 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>, 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(()) diff --git a/backend/windmill-api/src/postgres_triggers/trigger.rs b/backend/windmill-api/src/postgres_triggers/trigger.rs index f340666977..16186a8d5e 100644 --- a/backend/windmill-api/src/postgres_triggers/trigger.rs +++ b/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 { diff --git a/backend/windmill-common/src/db.rs b/backend/windmill-common/src/db.rs index 47c698b05c..5f9a4fbb19 100644 --- a/backend/windmill-common/src/db.rs +++ b/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( diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index 61184b627c..baeeba20dd 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/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) diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index e1c2f4aa15..ce2cc87671 100644 --- a/backend/windmill-common/src/lib.rs +++ b/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)?;