improve code

.sqlx/query-03919ba8286d18980ecaa0038df76f43f5388db0d3838181fa8f15f4cbb932af.json

@@ -1,6 +1,6 @@
 {
   "db_name": "MySQL",
-  "query": "\nUPDATE generic_inference_jobs\nSET\n  status = ?,\n  on_success_result_entity_type = ?,\n  on_success_result_entity_token = ?,\n  failure_reason = NULL,\n  internal_debugging_failure_reason = NULL,\n  success_execution_millis = ?,\n  success_inference_execution_millis = ?,\n  retry_at = NULL,\n  successfully_completed_at = NOW()\nWHERE token = ?\n    ",
+  "query": "\nUPDATE generic_inference_jobs\nSET\n  status = ?,\n  on_success_result_entity_type = ?,\n  on_success_result_entity_token = ?,\n  failure_reason = NULL,\n  internal_debugging_failure_reason = NULL,\n  success_execution_millis = ?,\n  success_inference_execution_millis = ?,\n  retry_at = NULL,\n  successfully_completed_at = NOW()\nWHERE token = ?\nAND status NOT IN ('complete_success', 'complete_failure')\n    ",
   "describe": {
     "columns": [],
     "parameters": {
@@ -8,5 +8,5 @@
     },
     "nullable": []
   },
-  "hash": "28e38723c78939b26ca982952b532e8b7e0563d20a1b22f628b3944ec2cce09a"
+  "hash": "03919ba8286d18980ecaa0038df76f43f5388db0d3838181fa8f15f4cbb932af"
 }

.sqlx/query-96599b4854248d84188b092db84edb50dc5402bb7771fd8ea96a96ad43e732c0.json

@@ -1,6 +1,6 @@
 {
   "db_name": "MySQL",
-  "query": "\nUPDATE generic_inference_jobs\nSET\n  status = ?,\n  frontend_failure_category = ?,\n  failure_reason = ?,\n  internal_debugging_failure_reason = ?,\n  success_execution_millis = ?,\n  success_inference_execution_millis = ?\nWHERE token = ?\n    ",
+  "query": "\nUPDATE generic_inference_jobs\nSET\n  status = ?,\n  frontend_failure_category = ?,\n  failure_reason = ?,\n  internal_debugging_failure_reason = ?,\n  success_execution_millis = ?,\n  success_inference_execution_millis = ?\nWHERE token = ?\nAND status NOT IN ('complete_success', 'complete_failure')\n    ",
   "describe": {
     "columns": [],
     "parameters": {
@@ -8,5 +8,5 @@
     },
     "nullable": []
   },
-  "hash": "4be27401d75a1781da96e1a3243aa0a42a9fd2cb0f75c56d268f7cab794876d7"
+  "hash": "96599b4854248d84188b092db84edb50dc5402bb7771fd8ea96a96ad43e732c0"
 }

.sqlx/query-f1d41ad17cb01f5aef0ce2eea042ccb3e5e649365a0befaf0f46492b75523442.json

@@ -1,6 +1,6 @@
 {
   "db_name": "MySQL",
-  "query": "\nSELECT\n  token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n  status as `status: enums::common::job_status_plus::JobStatusPlus`,\n  job_type as `job_type: enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType`,\n  maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n  maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n  creator_ip_address,\n  maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`,\n  maybe_debug_log_event_token as `maybe_debug_log_event_token: tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken`,\n  platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`\nFROM generic_inference_jobs\nWHERE token = ?\nAND job_type IN ('artcraft_minimax_h3_turbo', 'artcraft_minimax_h3_ultra')\n    ",
+  "query": "\nSELECT\n  token as `job_token: tokens::tokens::generic_inference_jobs::InferenceJobToken`,\n  status as `status: enums::common::job_status_plus::JobStatusPlus`,\n  job_type as `job_type: enums::by_table::generic_inference_jobs::inference_job_type::InferenceJobType`,\n  maybe_creator_user_token as `maybe_creator_user_token: tokens::tokens::users::UserToken`,\n  maybe_creator_anonymous_visitor_token as `maybe_creator_anonymous_visitor_token: tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken`,\n  creator_ip_address,\n  maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`,\n  maybe_debug_log_event_token as `maybe_debug_log_event_token: tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken`,\n  platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`,\n  on_success_result_entity_token as `maybe_result_media_file_token: tokens::tokens::media_files::MediaFileToken`\nFROM generic_inference_jobs\nWHERE token = ?\nAND job_type IN ('artcraft_minimax_h3_turbo', 'artcraft_minimax_h3_ultra')\n    ",
   "describe": {
     "columns": [
       {
@@ -92,6 +92,16 @@
           "char_set": 224,
           "max_size": 64
         }
+      },
+      {
+        "ordinal": 9,
+        "name": "maybe_result_media_file_token: tokens::tokens::media_files::MediaFileToken",
+        "type_info": {
+          "type": "VarString",
+          "flags": "MULTIPLE_KEY | BINARY",
+          "char_set": 224,
+          "max_size": 128
+        }
       }
     ],
     "parameters": {
@@ -106,8 +116,9 @@
       false,
       true,
       true,
+      true,
       true
     ]
   },
-  "hash": "b27e108f07885e2677ca79d0f5dc6a81a7fd760034878d2cad3e32cd1f6a76b5"
+  "hash": "f1d41ad17cb01f5aef0ce2eea042ccb3e5e649365a0befaf0f46492b75523442"
 }

crates/schema/database/mysql_queries/src/queries/generic_inference/first_party/minimax_h3/get_first_party_minimax_h3_job_by_token.rs

@@ -12,6 +12,7 @@ use enums::common::job_status_plus::JobStatusPlus;
 use enums::common::platform_type::PlatformType;
 use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
 use tokens::tokens::generic_inference_jobs::InferenceJobToken;
+use tokens::tokens::media_files::MediaFileToken;
 use tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken;
 use tokens::tokens::prompts::PromptToken;
 use tokens::tokens::users::UserToken;
@@ -41,6 +42,9 @@ pub struct FirstPartyMinimaxH3JobDetails {
   pub maybe_debug_log_event_token: Option<DebugLogEventToken>,
 
   pub maybe_platform_type: Option<PlatformType>,
+
+  /// Set once the job has succeeded: the resulting media file.
+  pub maybe_result_media_file_token: Option<MediaFileToken>,
 }
 
 pub async fn get_first_party_minimax_h3_job_by_token<'e, 'c : 'e, E>(
@@ -59,7 +63,8 @@ SELECT
   creator_ip_address,
   maybe_prompt_token as `maybe_prompt_token: tokens::tokens::prompts::PromptToken`,
   maybe_debug_log_event_token as `maybe_debug_log_event_token: tokens::tokens::non_unique::debug_logs_event_token::DebugLogEventToken`,
-  platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`
+  platform_type as `maybe_platform_type: enums::common::platform_type::PlatformType`,
+  on_success_result_entity_token as `maybe_result_media_file_token: tokens::tokens::media_files::MediaFileToken`
 FROM generic_inference_jobs
 WHERE token = ?
 AND job_type IN ('artcraft_minimax_h3_turbo', 'artcraft_minimax_h3_ultra')
@@ -89,5 +94,6 @@ AND job_type IN ('artcraft_minimax_h3_turbo', 'artcraft_minimax_h3_ultra')
     maybe_prompt_token: row.maybe_prompt_token,
     maybe_debug_log_event_token: row.maybe_debug_log_event_token,
     maybe_platform_type: row.maybe_platform_type,
+    maybe_result_media_file_token: row.maybe_result_media_file_token,
   }))
 }

crates/schema/database/mysql_queries/src/queries/generic_inference/first_party/minimax_h3/mark_first_party_minimax_h3_job_failed.rs

@@ -1,4 +1,7 @@
 //! Permanently mark a first-party Minimax H3 job as failed.
+//!
+//! Never flips an already-terminal job: a late or duplicate failure report
+//! against a `complete_success` / `complete_failure` row is silently ignored.
 
 use sqlx::{Executor, MySql};
 use std::convert::TryFrom;
@@ -61,6 +64,7 @@ SET
   success_execution_millis = ?,
   success_inference_execution_millis = ?
 WHERE token = ?
+AND status NOT IN ('complete_success', 'complete_failure')
     "#,
     JobStatusPlus::CompleteFailure.to_str(),
     args.maybe_frontend_failure_category.map(|category| category.to_str()),

crates/schema/database/mysql_queries/src/queries/generic_inference/first_party/minimax_h3/mark_first_party_minimax_h3_job_succeeded.rs

@@ -1,5 +1,8 @@
 //! Mark a first-party Minimax H3 job as successfully completed, pointing at
 //! the resulting media file. Mirrors the Fal webhook success write.
+//!
+//! Never flips an already-terminal job: a late or duplicate success report
+//! against a `complete_success` / `complete_failure` row is silently ignored.
 
 use sqlx::{Executor, MySql};
 use std::convert::TryFrom;
@@ -48,6 +51,7 @@ SET
   retry_at = NULL,
   successfully_completed_at = NOW()
 WHERE token = ?
+AND status NOT IN ('complete_success', 'complete_failure')
     "#,
     JobStatusPlus::CompleteSuccess.to_str(),
     InferenceResultType::MediaFile.to_str(),

crates/schema/database/mysql_queries/src/queries/media_files/create/generic_insert/insert_media_file_generic.rs

@@ -1,5 +1,5 @@
 use anyhow::anyhow;
-use sqlx::MySqlPool;
+use sqlx::{Acquire, MySql};
 
 use enums::by_table::generic_synthetic_ids::id_category::IdCategory;
 use enums::by_table::media_files::media_file_class::MediaFileClass;
@@ -24,8 +24,13 @@ use crate::payloads::media_file_extra_info::media_file_extra_info::MediaFileExtr
 use crate::queries::generic_synthetic_ids::transactional_increment_generic_synthetic_id::transactional_increment_generic_synthetic_id;
 use crate::queries::media_files::create::generic_insert::insert_media_file_generic_executor::{insert_media_file_generic_executor, InsertMediaFileGenericExecutorArgs};
 
-pub struct InsertArgs<'a> {
-    pub pool: &'a MySqlPool,
+pub struct InsertArgs<'a, A>
+where
+    A: Acquire<'a, Database = MySql>,
+{
+    /// Where to get the connection: a `&MySqlPool` (acquires one) or an
+    /// already-acquired `&mut PoolConnection` (reuses it).
+    pub pool: A,
 
     // Creator info
     pub maybe_creator_user_token: Option<&'a UserToken>,
@@ -111,9 +116,12 @@ pub struct InsertArgs<'a> {
 /// Insert a media file record, allocating the creator's synthetic IDs in the
 /// same transaction. Wraps [`insert_media_file_generic_executor`], which runs
 /// the actual INSERT.
-pub async fn insert_media_file_generic(
-    args: InsertArgs<'_>
+pub async fn insert_media_file_generic<'a, A>(
+    args: InsertArgs<'a, A>
 ) -> AnyhowResult<(MediaFileToken, u64)>
+where
+    A: Acquire<'a, Database = MySql> + Send,
+    <A as Acquire<'a>>::Connection: Send,
 {
     let extra_file_modification_info = args
         .maybe_extra_media_info.map(|extra| extra.to_json_string())

crates/schema/database/mysql_queries/src/queries/media_files/create/generic_insert/insert_media_file_generic_from_job.rs

@@ -128,7 +128,7 @@ pub async fn insert_media_file_generic_from_job(
         media_type: args.media_type,
         origin_category: args.origin_category,
         origin_product_category: args.origin_product_category,
-        pool: &args.pool,
+        pool: args.pool,
         public_bucket_directory_hash: args.public_bucket_directory_hash,
     }).await
 }

crates/schema/database/mysql_queries/src/queries/media_files/create/insert_builder/media_file_insert_builder.rs

@@ -8,7 +8,8 @@ use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCat
 use enums::by_table::media_files::media_file_origin_product_category::MediaFileOriginProductCategory;
 use enums::by_table::media_files::media_file_type::MediaFileType;
 use enums::common::visibility::Visibility;
-use sqlx::MySqlPool;
+use sqlx::pool::PoolConnection;
+use sqlx::{Acquire, MySql, MySqlPool};
 use enums::common::generation_provider::GenerationProvider;
 use enums::common::platform_type::PlatformType;
 use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
@@ -293,20 +294,33 @@ impl MediaFileInsertBuilder {
     self
   }
 
-  // TODO(bt,2025-04-26): Other connector options.
   pub async fn insert_pool(self, mysql_pool: &MySqlPool) -> Result<MediaFileToken, MediaFileInsertBuilderError> {
+    self.insert_acquire(mysql_pool).await
+  }
+
+  /// Insert using an already-acquired connection, so callers holding one
+  /// don't pay for a second pool acquisition.
+  pub async fn insert_connection(self, mysql_connection: &mut PoolConnection<MySql>) -> Result<MediaFileToken, MediaFileInsertBuilderError> {
+    self.insert_acquire(mysql_connection).await
+  }
+
+  async fn insert_acquire<'a, A>(&'a self, connector: A) -> Result<MediaFileToken, MediaFileInsertBuilderError>
+  where
+    A: Acquire<'a, Database = MySql> + Send,
+    <A as Acquire<'a>>::Connection: Send,
+  {
     let media_file_type = self.media_file_type
         .ok_or_else(|| MediaFileInsertBuilderError::MissingRequiredField(
           "Media file type is required".to_string()))?;
 
-    let checksum_sha2 = self.checksum_sha2
+    let checksum_sha2 = self.checksum_sha2.as_ref()
         .ok_or_else(|| MediaFileInsertBuilderError::MissingRequiredField(
           "Checksum SHA2 is required".to_string()))?;
 
-    let bucket_path = self.public_bucket_directory_hash
+    let bucket_path = self.public_bucket_directory_hash.as_ref()
         .ok_or_else(|| MediaFileInsertBuilderError::MissingRequiredField(
           "Public bucket directory hash is required".to_string()))?;
-    
+
     let mut origin_category = self.origin_category
         .ok_or_else(|| MediaFileInsertBuilderError::MissingRequiredField(
           "Origin category is required".to_string()))?;
@@ -330,7 +344,7 @@ impl MediaFileInsertBuilder {
     }
 
     let result = insert_media_file_generic(InsertArgs {
-      pool: mysql_pool,
+      pool: connector,
       maybe_creator_user_token: self.maybe_creator_user_token.as_ref(),
       maybe_creator_anonymous_visitor_token: self.maybe_creator_anonymous_visitor_token.as_ref(),
       creator_ip_address: self.creator_ip_address.as_deref().unwrap_or("127.0.0.1"),
@@ -343,7 +357,7 @@ impl MediaFileInsertBuilder {
       origin_product_category: self.origin_product_category,
       maybe_origin_model_type: None, // TODO
       maybe_origin_model_token: None, // TODO
-      maybe_origin_filename: self.maybe_origin_filename,
+      maybe_origin_filename: self.maybe_origin_filename.clone(),
       maybe_mime_type: self.maybe_mime_type.as_deref(),
       file_size_bytes: self.file_size_bytes,
       maybe_duration_millis: self.maybe_duration_millis,
@@ -352,7 +366,7 @@ impl MediaFileInsertBuilder {
       maybe_frame_width: self.maybe_frame_width,
       maybe_frame_height: self.maybe_frame_height,
       maybe_engine_category: self.maybe_engine_category,
-      checksum_sha2: &checksum_sha2,
+      checksum_sha2: checksum_sha2.as_str(),
       maybe_title: self.maybe_title.as_deref(),
       maybe_text_transcript: None, // TODO
       maybe_scene_source_media_file_token: None, // TODO

crates/schema/database/mysql_queries/src/queries/media_files/create/insert_media_file_from_studio_scene_render.rs

@@ -50,7 +50,7 @@ pub async fn insert_media_file_from_studio_scene_render(
 ) -> AnyhowResult<MediaFileToken>
 {
   let (new_media_token, _id) = insert_media_file_generic(InsertArgs {
-    pool: &args.pool,
+    pool: args.pool,
 
     // Dynamic bits (user)
     maybe_creator_user_token: args.maybe_creator_user_token,

crates/service/web/storyteller_web/src/http_server/endpoints/internal/minimax_jobs/mark_minimax_job_success_handler.rs

@@ -8,6 +8,8 @@ use actix_web::web::Json;
 use actix_web::{web, HttpRequest};
 use log::{error, info, warn};
 use serde_derive::Deserialize;
+use sqlx::pool::PoolConnection;
+use sqlx::MySql;
 use utoipa::ToSchema;
 
 use artcraft_api_defs::internal::minimax_jobs::mark_minimax_job_success::MarkMinimaxJobSuccessResponse;
@@ -16,6 +18,7 @@ use enums::by_table::media_files::media_file_class::MediaFileClass;
 use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCategory;
 use enums::by_table::media_files::media_file_type::MediaFileType;
 use enums::common::generation_provider::GenerationProvider;
+use enums::common::job_status_plus::JobStatusPlus;
 use ffmpeg_utils::ffprobe::ffprobe_get_info::ffprobe_get_info;
 use filesys::path_to_string::path_to_string;
 use hashing::sha256::sha256_hash_bytes::sha256_hash_bytes;
@@ -134,8 +137,42 @@ pub async fn mark_minimax_job_success_handler(
     return Err(CommonWebError::NotFound);
   };
 
+  // ==================== IDEMPOTENCY: TERMINAL STATUS CHECK ==================== //
+
+  match job.status {
+    JobStatusPlus::CompleteSuccess => {
+      // A duplicate / retried success report. Don't re-upload or create a
+      // second media file — return the already-recorded result.
+      let Some(existing_media_token) = job.maybe_result_media_file_token else {
+        error!("Minimax job {} is complete_success but has no result media file token", job.job_token);
+        return Err(CommonWebError::server_error_with_message(
+          "job already complete but result media file is missing"));
+      };
+
+      info!("Minimax job {} is already complete_success; ignoring duplicate success report", job.job_token);
+
+      return Ok(Json(MarkMinimaxJobSuccessResponse {
+        success: true,
+        media_file_token: existing_media_token,
+        maybe_duration_millis: None,
+        maybe_width: None,
+        maybe_height: None,
+      }));
+    }
+    JobStatusPlus::CompleteFailure => {
+      warn!("Minimax job {} is already complete_failure; rejecting success report", job.job_token);
+      return Err(CommonWebError::BadInputWithSimpleMessage(
+        "job is already in a terminal state (complete_failure)".to_string()));
+    }
+    _ => {} // Not terminal; proceed.
+  }
+
   // ==================== READ + VALIDATE FILE ==================== //
 
+  // NB: No database connection is held anywhere in this section — the byte
+  // read, hashing, ffprobe, and bucket upload below are all slow, and
+  // holding a pool slot across them is what causes PoolTimedOut incidents.
+
   let mut file_bytes = Vec::new();
   form.file.file.read_to_end(&mut file_bytes)
     .map_err(|err| {
@@ -221,20 +258,42 @@ pub async fn mark_minimax_job_success_handler(
         CommonWebError::from_anyhow_error(err)
       })?;
 
-  let media_token = insert_media_file_record(
-    &server_state,
-    &job,
-    &file_hash,
-    file_size_bytes,
-    maybe_duration_millis,
-    maybe_frame_width,
-    maybe_frame_height,
-    media_file_type,
-    &mime_type,
-    &public_upload_path,
-  ).await?;
+  // ==================== MEDIA FILE RECORD + MARK JOB DONE ==================== //
 
-  info!("Minimax video media file uploaded with token: {}", media_token);
+  // Tight critical section: one connection for both post-upload writes, no
+  // network calls in between.
+  let media_token = {
+    let mut mysql_connection = server_state.mysql_pool.acquire().await?;
+
+    let media_token = insert_media_file_record(
+      &mut mysql_connection,
+      &job,
+      &file_hash,
+      file_size_bytes,
+      maybe_duration_millis,
+      maybe_frame_width,
+      maybe_frame_height,
+      media_file_type,
+      &mime_type,
+      &public_upload_path,
+    ).await?;
+
+    info!("Minimax video media file uploaded with token: {}", media_token);
+
+    mark_first_party_minimax_h3_job_succeeded(MarkFirstPartyMinimaxH3JobSucceededArgs {
+      job_token: &job.job_token,
+      media_file_token: &media_token,
+      maybe_execution_duration_millis: form.execution_duration_millis.map(|text| text.0),
+      maybe_inference_duration_millis: form.inference_duration_millis.map(|text| text.0),
+      mysql_executor: &mut *mysql_connection,
+      phantom: Default::default(),
+    }).await.map_err(|err| {
+      error!("Error marking minimax job {} succeeded: {:?}", job_token, err);
+      CommonWebError::from_error(err)
+    })?;
+
+    media_token
+  };
 
   // ==================== THUMBNAILS ==================== //
 
@@ -252,20 +311,6 @@ pub async fn mark_minimax_job_success_handler(
     error!("Failed to create some/all thumbnail tasks: {:?}", err);
   }
 
-  // ==================== MARK JOB DONE ==================== //
-
-  mark_first_party_minimax_h3_job_succeeded(MarkFirstPartyMinimaxH3JobSucceededArgs {
-    job_token: &job.job_token,
-    media_file_token: &media_token,
-    maybe_execution_duration_millis: form.execution_duration_millis.map(|text| text.0),
-    maybe_inference_duration_millis: form.inference_duration_millis.map(|text| text.0),
-    mysql_executor: &server_state.mysql_pool,
-    phantom: Default::default(),
-  }).await.map_err(|err| {
-    error!("Error marking minimax job {} succeeded: {:?}", job_token, err);
-    CommonWebError::from_error(err)
-  })?;
-
   info!("Minimax job {} marked as complete_success with media file {}", job.job_token, media_token);
 
   Ok(Json(MarkMinimaxJobSuccessResponse {
@@ -278,7 +323,7 @@ pub async fn mark_minimax_job_success_handler(
 }
 
 async fn insert_media_file_record(
-  server_state: &ServerState,
+  mysql_connection: &mut PoolConnection<MySql>,
   job: &FirstPartyMinimaxH3JobDetails,
   file_hash: &str,
   file_size_bytes: usize,
@@ -307,7 +352,7 @@ async fn insert_media_file_record(
       .media_file_type(media_file_type)
       .mime_type(mime_type)
       .public_bucket_directory_hash(public_upload_path)
-      .insert_pool(&server_state.mysql_pool)
+      .insert_connection(mysql_connection)
       .await
       .map_err(|err| {
         warn!("Failed to insert video media file record: {:?}", err);