launch reaping thread successfully, codegen query

.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"
+}

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)
       },

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),

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(())
 }

crates/service/job/inference_job/src/http_server/run_http_server.rs

@@ -40,6 +40,8 @@ pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server>
   // 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()

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 || {

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(())
 }