diff --git a/.sqlx/query-19708db99f2b3c6283d2115a511e0eb1c99fab396ad0effd9eb2dd79926e9c93.json b/.sqlx/query-19708db99f2b3c6283d2115a511e0eb1c99fab396ad0effd9eb2dd79926e9c93.json new file mode 100644 index 0000000000..c9415cb406 --- /dev/null +++ b/.sqlx/query-19708db99f2b3c6283d2115a511e0eb1c99fab396ad0effd9eb2dd79926e9c93.json @@ -0,0 +1,12 @@ +{ + "db_name": "MySQL", + "query": "\nUPDATE generic_inference_jobs\nSET\n status = 'cancelled_by_system',\n retry_at = NULL,\n internal_debugging_failure_reason = 'reap_stale_fakeyou_jobs'\nWHERE\n job_type != 'fal_queue'\n AND status IN (\n 'pending',\n 'started',\n 'attempt_failed'\n )\n AND created_at > NOW() - INTERVAL 30 MINUTE\nLIMIT 5000\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 0 + }, + "nullable": [] + }, + "hash": "19708db99f2b3c6283d2115a511e0eb1c99fab396ad0effd9eb2dd79926e9c93" +} diff --git a/crates/schema/database/mysql_queries/src/payloads/generic_inference_args/generic_inference_args.rs b/crates/schema/database/mysql_queries/src/payloads/generic_inference_args/generic_inference_args.rs index f28635a85c..a1c3c62983 100644 --- a/crates/schema/database/mysql_queries/src/payloads/generic_inference_args/generic_inference_args.rs +++ b/crates/schema/database/mysql_queries/src/payloads/generic_inference_args/generic_inference_args.rs @@ -185,6 +185,8 @@ impl GenericInferenceArgs { match json { None => Ok(None), Some("" | "null" | "NULL" | "undefined" | "UNDEFINED") => { + // NB: Something inserted a few of these records (only in local dev). + // It's better not to stall the job queue than to assert correctness. warn!("Received empty or null JSON `{json:?}` for GenericInferenceArgs, returning None."); Ok(None) }, diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/job/list_available_generic_inference_jobs.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/job/list_available_generic_inference_jobs.rs index ef53e01856..380d53b3f1 100644 --- a/crates/schema/database/mysql_queries/src/queries/generic_inference/job/list_available_generic_inference_jobs.rs +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/job/list_available_generic_inference_jobs.rs @@ -122,8 +122,6 @@ pub async fn list_available_generic_inference_jobs( .as_deref() .map(|s| AnonymousVisitorTrackingToken::new_from_str(s)); - debug!("Record: {:?}", record); - Ok(AvailableInferenceJob { id: GenericInferenceJobId(record.id), inference_job_token: InferenceJobToken::new(record.inference_job_token), diff --git a/crates/schema/database/mysql_queries/src/queries/generic_inference/job/reap_stale_fakeyou_jobs.rs b/crates/schema/database/mysql_queries/src/queries/generic_inference/job/reap_stale_fakeyou_jobs.rs index 835c518177..8d9546b143 100644 --- a/crates/schema/database/mysql_queries/src/queries/generic_inference/job/reap_stale_fakeyou_jobs.rs +++ b/crates/schema/database/mysql_queries/src/queries/generic_inference/job/reap_stale_fakeyou_jobs.rs @@ -1,4 +1,5 @@ use anyhow::anyhow; +use log::info; use sqlx; use sqlx::pool::PoolConnection; use sqlx::{MySql, MySqlPool}; @@ -39,9 +40,13 @@ LIMIT 5000 .execute(&mut **mysql_connection) .await; - if let Err(err) = query_result { - return Err(anyhow!("error with reaping stale jobs: {:?}", err)); + match query_result { + Ok(result) => { + info!("Reaped stale FakeYou jobs: {} rows affected.", result.rows_affected()); + Ok(()) + }, + Err(err) => { + Err(anyhow!("error with reaping stale jobs: {:?}", err)) + } } - - Ok(()) } diff --git a/crates/service/job/inference_job/src/http_server/run_http_server.rs b/crates/service/job/inference_job/src/http_server/run_http_server.rs index 40757da4f2..bfa8d9e705 100644 --- a/crates/service/job/inference_job/src/http_server/run_http_server.rs +++ b/crates/service/job/inference_job/src/http_server/run_http_server.rs @@ -40,6 +40,8 @@ pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult // NB: We shouldn't be logging much as the /_status endpoint is all we aim to expose. let log_format = "[%{HOSTNAME}e] IP=[%{X-Forwarded-For}i] \"%r\" %s %b \"%{Referer}i\" \"%{User-Agent}i\" %T"; + info!("HTTP server will bind to: {}", bind_address); + let handle = HttpServer::new(move || { // NB: app_data being clone()'d below should all be safe (dependencies included) App::new() diff --git a/crates/service/job/inference_job/src/main.rs b/crates/service/job/inference_job/src/main.rs index 8d8c289608..3a5f4ee0fa 100644 --- a/crates/service/job/inference_job/src/main.rs +++ b/crates/service/job/inference_job/src/main.rs @@ -349,8 +349,12 @@ async fn main() -> AnyhowResult<()> { set_up_directories(&job_dependencies)?; - std::thread::spawn(move || async { - reap_jobs_thread(mysql_pool_2).await; + std::thread::spawn(move || { + let actix_runtime = actix_web::rt::System::new(); + + actix_runtime.block_on(reap_jobs_thread(mysql_pool_2)); + + warn!("Reap jobs thread terminated"); }); std::thread::spawn(move || { diff --git a/crates/service/job/inference_job/src/threads/reap_jobs_thread.rs b/crates/service/job/inference_job/src/threads/reap_jobs_thread.rs index ab46309153..723ed918c0 100644 --- a/crates/service/job/inference_job/src/threads/reap_jobs_thread.rs +++ b/crates/service/job/inference_job/src/threads/reap_jobs_thread.rs @@ -4,12 +4,17 @@ use mysql_queries::queries::generic_inference::job::reap_stale_fakeyou_jobs::rea use sqlx::MySqlPool; use std::time::Duration; +const WAIT_BETWEEN_REAPING_SECONDS: u64 = 60 * 2; // 2 minutes + pub async fn reap_jobs_thread( mysql_pool: MySqlPool, ) -> ! { + info!("Database job reaping thread starting..."); + loop { + tokio::time::sleep(Duration::from_secs(WAIT_BETWEEN_REAPING_SECONDS)).await; + info!("Reaping jobs from database..."); - tokio::time::sleep(Duration::from_secs(6)).await; let result = do_reap_jobs(&mysql_pool).await; @@ -23,7 +28,6 @@ pub async fn reap_jobs_thread( pub async fn do_reap_jobs( mysql_pool: &MySqlPool, ) -> AnyhowResult<()> { - info!("Reaping jobs from database..."); reap_stale_fakeyou_jobs(mysql_pool).await?; Ok(()) } \ No newline at end of file