add thumbnail job sharding and logging (#1158)

crates/service/job/video_thumbnail_job/src/job/main_loop.rs

@@ -67,6 +67,14 @@ async fn run_batch_cycle(deps: &JobDependencies) -> anyhow::Result<u64> {
         break;
       }
 
+      // If sharding is enabled, skip media files that don't belong to this shard.
+      if let Some(shard_info) = &deps.shard_info {
+        let modulus = (media_file.id as u64) % (shard_info.number_of_shards as u64);
+        if modulus != shard_info.shard_index as u64 {
+          continue;
+        }
+      }
+
       match process_single_media_file(deps, media_file).await {
         Ok(()) => {
           let _ = deps.job_stats.increment_success_count();

crates/service/job/video_thumbnail_job/src/job/process_single_media_file.rs

@@ -24,6 +24,14 @@ pub async fn process_single_media_file(
   deps: &JobDependencies,
   media_file: &VideoMediaFileWithoutThumbnail,
 ) -> anyhow::Result<()> {
+
+  info!(
+    "Processing media file {:?} (id: {}, created at: {})",
+    media_file.token,
+    media_file.id,
+    media_file.created_at,
+  );
+
   let downloaded = download_video(deps, media_file).await?;
 
   info!(
@@ -86,7 +94,12 @@ pub async fn process_single_media_file(
     &deps.mysql_pool,
   ).await?;
 
-  info!("Updated thumbnail version for {}", media_file.token.as_str());
+  info!(
+    "Updated thumbnail version for media file {:?} (id: {}, created at: {})",
+    media_file.token,
+    media_file.id,
+    media_file.created_at,
+  );
 
   // `downloaded.temp_dir` is dropped here, cleaning up the temp directory and all contents.
   Ok(())

crates/service/job/video_thumbnail_job/src/job_dependencies.rs

@@ -6,6 +6,12 @@ use jobs_common::job_stats::JobStats;
 use server_environment::ServerEnvironment;
 use sqlx::MySqlPool;
 
+/// Optional sharding configuration for distributing work across parallel job instances.
+pub struct ShardInfo {
+  pub number_of_shards: u8,
+  pub shard_index: u8,
+}
+
 pub struct JobDependencies {
   pub mysql_pool: MySqlPool,
 
@@ -34,6 +40,9 @@ pub struct JobDependencies {
   /// Root directory for temporary files (video downloads, thumbnail intermediates).
   pub temp_dir: PathBuf,
 
+  /// If present, only process media files where `id % number_of_shards == shard_index`.
+  pub shard_info: Option<ShardInfo>,
+
   /// Set to `true` from another thread to trigger graceful shutdown.
   pub application_shutdown: RelaxedAtomicBool,
 }

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

@@ -27,7 +27,7 @@ use server_environment::ServerEnvironment;
 
 use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs};
 use crate::job::main_loop::main_loop;
-use crate::job_dependencies::JobDependencies;
+use crate::job_dependencies::{JobDependencies, ShardInfo};
 
 pub mod http_server;
 pub mod job;
@@ -130,6 +130,23 @@ async fn main() -> AnyhowResult<()> {
   // Ensure the temp directory exists
   tokio::fs::create_dir_all(&temp_dir).await?;
 
+  // Optional sharding configuration
+  let maybe_number_of_shards: Option<u8> = easyenv::get_env_string_optional("NUMBER_OF_SHARDS")
+      .and_then(|s| s.parse().ok());
+  let maybe_shard_index: Option<u8> = easyenv::get_env_string_optional("SHARD_INDEX")
+      .and_then(|s| s.parse().ok());
+
+  let shard_info = match (maybe_number_of_shards, maybe_shard_index) {
+    (Some(number_of_shards), Some(shard_index)) => {
+      info!("Sharding enabled: shard {shard_index} of {number_of_shards}");
+      Some(ShardInfo { number_of_shards, shard_index })
+    }
+    _ => {
+      info!("Sharding disabled: processing all media files.");
+      None
+    }
+  };
+
   let application_shutdown = RelaxedAtomicBool::new(false);
   let job_stats = JobStats::new();
 
@@ -149,6 +166,7 @@ async fn main() -> AnyhowResult<()> {
     custom_max_lookback_hours,
     custom_page_size,
     temp_dir,
+    shard_info,
     application_shutdown: application_shutdown.clone(),
   };