From 232f66eb69b93504894c331aa2553fbcc6fa1a63 Mon Sep 17 00:00:00 2001
From: Brandon Thomas <echelon@gmail.com>
Date: Sat, 22 May 2021 03:57:07 -0400
Subject: [PATCH] runs jobs to exhaustion

---
 src/job/job_queries.rs  | 26 +++++++++++++++++++++++++-
 src/tts_download_job.rs | 19 +++++++++++++++----
 2 files changed, 40 insertions(+), 5 deletions(-)

diff --git a/src/job/job_queries.rs b/src/job/job_queries.rs
index 28fc23dc05..d6c5581799 100644
--- a/src/job/job_queries.rs
+++ b/src/job/job_queries.rs
@@ -81,4 +81,28 @@ WHERE id = ?
     .await?;
 
   Ok(())
-}
\ No newline at end of file
+}
+pub async fn mark_tts_upload_job_done(pool: &MySqlPool,
+                                      job: &TtsUploadJobRecord,
+                                      success: bool)
+                                      -> AnyhowResult<()>
+{
+  let status = if success { "complete_success" } else { "complete_failure" };
+
+  let query_result = sqlx::query!(
+        r#"
+UPDATE tts_model_upload_jobs
+SET
+  status = ?,
+  failure_reason = NULL,
+  retry_at = NULL
+WHERE id = ?
+        "#,
+        status,
+        job.id,
+    )
+    .execute(pool)
+    .await?;
+
+  Ok(())
+}
diff --git a/src/tts_download_job.rs b/src/tts_download_job.rs
index 9376f5f7b2..566648c87e 100644
--- a/src/tts_download_job.rs
+++ b/src/tts_download_job.rs
@@ -1,3 +1,4 @@
+#![deny(unused_must_use)]
 #![allow(dead_code)]
 #![allow(unused_mut)]
 #![allow(unused_imports)]
@@ -9,8 +10,9 @@ pub mod util;
 
 use chrono::Utc;
 use crate::job::job_queries::TtsUploadJobRecord;
-use crate::job::job_queries::query_tts_upload_job_records;
+use crate::job::job_queries::mark_tts_upload_job_done;
 use crate::job::job_queries::mark_tts_upload_job_failure;
+use crate::job::job_queries::query_tts_upload_job_records;
 use crate::util::anyhow_result::AnyhowResult;
 use log::{warn, info};
 use sqlx::MySqlPool;
@@ -44,7 +46,7 @@ async fn main() -> AnyhowResult<()> {
     .connect(&db_connection_string)
     .await?;
 
-  main_loop(pool);
+  main_loop(pool).await;
 
   Ok(())
 }
@@ -69,6 +71,12 @@ async fn main_loop(pool: MySqlPool) {
       }
     };
 
+    if jobs.is_empty() {
+      info!("No jobs!");
+      std::thread::sleep(Duration::from_millis(1500));
+      continue;
+    }
+
     let result = process_jobs(&pool, jobs).await;
 
     match result {
@@ -89,7 +97,7 @@ async fn main_loop(pool: MySqlPool) {
 
 async fn process_jobs(pool: &MySqlPool, jobs: Vec<TtsUploadJobRecord>) -> AnyhowResult<()> {
   for job in jobs.into_iter() {
-    let result = process_job(&job).await;
+    let result = process_job(pool, &job).await;
     match result {
       Ok(_) => {},
       Err(e) => {
@@ -103,13 +111,16 @@ async fn process_jobs(pool: &MySqlPool, jobs: Vec<TtsUploadJobRecord>) -> Anyhow
   Ok(())
 }
 
-async fn process_job(job: &TtsUploadJobRecord) -> AnyhowResult<()> {
+async fn process_job(pool: &MySqlPool, job: &TtsUploadJobRecord) -> AnyhowResult<()> {
   // TODO: 1. Mark processing.
   // TODO: 2. Download.
   // TODO: 3. Upload.
   // TODO: 4. Save record.
   // TODO: 5. Mark job done.
 
+  info!("Job done: {}", job.id);
+  mark_tts_upload_job_done(pool, job, true).await?;
+
   Ok(())
 }