refactor job so we can use multiple threads

Cargo.lock

@@ -473,6 +473,7 @@ dependencies = [
  "easyenv",
  "enums",
  "errors",
+ "futures 0.3.29",
  "hostname 0.4.0",
  "log",
  "my-workspace-hack",

crates/service/job/analytics_job/Cargo.toml

@@ -42,6 +42,7 @@ sqlx = { version = "0.7.2", features = [ "mysql", "runtime-tokio", "tls-rustls",
 # Runtime
 # NB: We switched to Tokio from `async-std` for broader ecosystem support.
 tokio = { version = "1.38.0", features = ["macros"] }
+futures = "0.3.29"
 
 # Serialization
 serde = "1.0.125"

crates/service/job/analytics_job/src/job_state.rs

@@ -1,5 +1,6 @@
 use sqlx::MySqlPool;
 
+#[derive(Clone)]
 pub struct JobState {
   pub mysql_pool: MySqlPool,
 
@@ -7,6 +8,7 @@ pub struct JobState {
 }
 
 /// Use sleep to not overload the database.
+#[derive(Clone)]
 pub struct SleepConfigs {
   // How long to wait between individual "jobs".
   pub between_job_wait_millis: u64,

crates/service/job/analytics_job/src/main.rs

@@ -17,6 +17,7 @@
 //#![forbid(warnings)]
 
 use log::info;
+use sqlx::{MySql, Pool};
 use sqlx::mysql::MySqlPoolOptions;
 
 use config::shared_constants::DEFAULT_MYSQL_CONNECTION_STRING;
@@ -24,11 +25,10 @@ use config::shared_constants::DEFAULT_RUST_LOG;
 use errors::AnyhowResult;
 
 use crate::job_state::{JobState, SleepConfigs};
-use crate::main_loop::main_loop;
+use crate::tasks::calculate_old_model_analytics::calculate_old_model_analytics_loop::calculate_old_model_analytics_loop;
 
 pub mod job_state;
 pub mod tasks;
-pub mod main_loop;
 
 #[tokio::main]
 async fn main() -> AnyhowResult<()> {
@@ -46,17 +46,7 @@ async fn main() -> AnyhowResult<()> {
 
   info!("Hostname: {}", &server_hostname);
 
-  let db_connection_string =
-      easyenv::get_env_string_or_default(
-        "MYSQL_URL",
-        DEFAULT_MYSQL_CONNECTION_STRING);
-
-  info!("Connecting to database...");
-
-  let mysql_pool = MySqlPoolOptions::new()
-      .max_connections(5)
-      .connect(&db_connection_string)
-      .await?;
+  let mysql_pool = get_mysql_pool().await?;
 
   let job_state = JobState {
     mysql_pool,
@@ -68,7 +58,32 @@ async fn main() -> AnyhowResult<()> {
     },
   };
 
-  let _r = main_loop(job_state).await;
+  let job_state_1 = job_state.clone();
+
+  let handle_1 = tokio::task::spawn(async move {
+    let _r = calculate_old_model_analytics_loop(job_state_1).await;
+  });
+
+  futures::future::join_all([
+    handle_1,
+  ]).await;
+
 
   Ok(())
 }
+
+async fn get_mysql_pool() -> AnyhowResult<Pool<MySql>> {
+  info!("Connecting to MySQL database...");
+
+  let db_connection_string =
+      easyenv::get_env_string_or_default(
+        "MYSQL_URL",
+        DEFAULT_MYSQL_CONNECTION_STRING);
+
+  let mysql_pool = MySqlPoolOptions::new()
+      .max_connections(5)
+      .connect(&db_connection_string)
+      .await?;
+
+  Ok(mysql_pool)
+}

crates/service/job/analytics_job/src/tasks/calculate_old_model_analytics/calculate_old_model_analytics_loop.rs

@@ -7,7 +7,7 @@ use errors::AnyhowResult;
 use crate::job_state::JobState;
 use crate::tasks::calculate_old_model_analytics::calculate_old_model_analytics::calculate_old_model_analytics;
 
-pub async fn main_loop(job_state: JobState) -> AnyhowResult<()> {
+pub async fn calculate_old_model_analytics_loop(job_state: JobState) -> AnyhowResult<()> {
   loop {
     info!("Begin analytics job batch.");
 

crates/service/job/analytics_job/src/tasks/calculate_old_model_analytics/mod.rs

@@ -1 +1,2 @@
 pub mod calculate_old_model_analytics;
+pub mod calculate_old_model_analytics_loop;

crates/service/job/es-update-job/Cargo.toml

@@ -49,9 +49,9 @@ elasticsearch = "8.5.0-alpha.1"
 # Runtime
 # NB: We switched to Tokio from `async-std` for broader ecosystem support.
 tokio = { version = "1.38.0", features = ["macros"] }
+futures = "0.3.29"
 
 # Serialization
 serde = "1.0.125"
 serde_derive = "1.0.125"
 serde_json = "1.0.108"
-futures = "0.3.29"