diff --git a/backend/.sqlx/query-d25c58d2722ad3dcd91101ce6f66e1d802dd5d82e1cd5f5ed3a15cbc75eb6745.json b/backend/.sqlx/query-c92cc71e6d10c41368f7aa75b0799c2e1e9ca0ed33077ade8d7560e7cc21fa06.json similarity index 60% rename from backend/.sqlx/query-d25c58d2722ad3dcd91101ce6f66e1d802dd5d82e1cd5f5ed3a15cbc75eb6745.json rename to backend/.sqlx/query-c92cc71e6d10c41368f7aa75b0799c2e1e9ca0ed33077ade8d7560e7cc21fa06.json index 2872c1655b..8a4b957c2a 100644 --- a/backend/.sqlx/query-d25c58d2722ad3dcd91101ce6f66e1d802dd5d82e1cd5f5ed3a15cbc75eb6745.json +++ b/backend/.sqlx/query-c92cc71e6d10c41368f7aa75b0799c2e1e9ca0ed33077ade8d7560e7cc21fa06.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "DELETE FROM v2_job_queue WHERE workspace_id = $1 AND id = $2 RETURNING 1", + "query": "DELETE FROM v2_job_queue WHERE id = $1 RETURNING 1", "describe": { "columns": [ { @@ -11,7 +11,6 @@ ], "parameters": { "Left": [ - "Text", "Uuid" ] }, @@ -19,5 +18,5 @@ null ] }, - "hash": "d25c58d2722ad3dcd91101ce6f66e1d802dd5d82e1cd5f5ed3a15cbc75eb6745" + "hash": "c92cc71e6d10c41368f7aa75b0799c2e1e9ca0ed33077ade8d7560e7cc21fa06" } diff --git a/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json b/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json index c2dfed73a2..5bfff47576 100644 --- a/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json +++ b/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927" diff --git a/backend/pg_log_tail b/backend/pg_log_tail new file mode 100644 index 0000000000..e69de29bb2 diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index c8996b3595..802af6196b 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -281,7 +281,7 @@ pub async fn connect_db( pub async fn connect( database_url: &str, max_connections: u32, - _worker_mode: bool, + worker_mode: bool, ) -> Result, error::Error> { use std::time::Duration; @@ -289,18 +289,18 @@ pub async fn connect( .min_connections((max_connections / 5).clamp(3, max_connections)) .max_connections(max_connections) .max_lifetime(Duration::from_secs(30 * 60)) // 30 mins - // .after_connect(move |conn, _| { - // if worker_mode { - // Box::pin(async move { - // // sqlx::query("SET enable_seqscan = OFF;") - // // .execute(conn) - // // .await?; - // Ok(()) - // }) - // } else { - // Box::pin(async move { Ok(()) }) - // } - // }) + .after_connect(move |conn, _| { + if worker_mode { + Box::pin(async move { + sqlx::query("SET enable_seqscan = OFF;") + .execute(conn) + .await?; + Ok(()) + }) + } else { + Box::pin(async move { Ok(()) }) + } + }) .connect_with( sqlx::postgres::PgConnectOptions::from_str(database_url)?.statement_cache_capacity(400), ) diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 5d360d3acb..e582fa39dc 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -566,6 +566,9 @@ pub async fn add_completed_job( let result_columns = result_columns.as_ref(); let _job_id = queued_job.id; let (opt_uuid, _duration, _skip_downstream_error_handlers) = (|| async { + + // let start = std::time::Instant::now(); + let mut tx = db.begin().await?; let job_id = queued_job.id; @@ -663,7 +666,7 @@ pub async fn add_completed_job( // tracing::error!("Added completed job {:#?}", queued_job); let mut _skip_downstream_error_handlers = false; - tx = delete_job(tx, &queued_job.workspace_id, job_id).await?; + tx = delete_job(tx, &job_id).await?; // tracing::error!("3 {:?}", start.elapsed()); if queued_job.is_flow_step { @@ -858,6 +861,7 @@ pub async fn add_completed_job( "inserted completed job: {} (success: {success})", queued_job.id ); + // tracing::info!("completed job: {:?}", start.elapsed().as_micros()); Ok((None, _duration, _skip_downstream_error_handlers)) as windmill_common::error::Result<(Option, i64, bool)> }) .retry( @@ -2597,21 +2601,17 @@ async fn extract_result_from_job_result( pub async fn delete_job<'c>( mut tx: Transaction<'c, Postgres>, - w_id: &str, - job_id: Uuid, + job_id: &Uuid, ) -> windmill_common::error::Result> { #[cfg(feature = "prometheus")] if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { QUEUE_DELETE_COUNT.inc(); } - let job_removed = sqlx::query_scalar!( - "DELETE FROM v2_job_queue WHERE workspace_id = $1 AND id = $2 RETURNING 1", - w_id, - job_id - ) - .fetch_optional(&mut *tx) - .await; + let job_removed = + sqlx::query_scalar!("DELETE FROM v2_job_queue WHERE id = $1 RETURNING 1", job_id,) + .fetch_optional(&mut *tx) + .await; if let Err(job_removed) = job_removed { tracing::error!(