diff --git a/crates/service/job/seedance2_pro_job/src/jobs/main_loop.rs b/crates/service/job/seedance2_pro_job/src/jobs/main_loop.rs index e2516668fe..53cc7ce886 100644 --- a/crates/service/job/seedance2_pro_job/src/jobs/main_loop.rs +++ b/crates/service/job/seedance2_pro_job/src/jobs/main_loop.rs @@ -83,6 +83,11 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> { let mut pages_in_current_batch: u32 = 0; loop { + if deps.application_shutdown.get() { + info!("Shutdown requested during pagination. Stopping early."); + break; + } + total_page_number += 1; pages_in_current_batch += 1; diff --git a/crates/service/job/seedance2_pro_job/src/main.rs b/crates/service/job/seedance2_pro_job/src/main.rs index beb273156a..5bc074b6ce 100644 --- a/crates/service/job/seedance2_pro_job/src/main.rs +++ b/crates/service/job/seedance2_pro_job/src/main.rs @@ -43,7 +43,7 @@ const ENV_PUBLIC_BUCKET_NAME: &str = "PUBLIC_BUCKET_NAME"; const ENV_S3_ENDPOINT: &str = "S3_COMPATIBLE_ENDPOINT_URL"; const ENV_SEEDANCE2PRO_COOKIES : &str = "SEEDANCE2PRO_COOKIES"; -#[actix_web::main] +#[tokio::main] async fn main() -> AnyhowResult<()> { let container_environment = bootstrap(BootstrapArgs { @@ -117,15 +117,21 @@ async fn main() -> AnyhowResult<()> { let (pager, pager_worker) = build_pager(server_environment, &container_environment.hostname); - info!("Spawning pager worker thread."); + info!("Spawning pager worker."); - tokio::spawn(async move { - pager_worker.run().await; + // NB: The pager worker uses Condvar::wait() which is a blocking syscall. + // It must run on a dedicated OS thread, not a tokio task, to avoid blocking + // the tokio runtime. + std::thread::spawn(move || { + let rt = tokio::runtime::Runtime::new().expect("pager worker tokio runtime"); + rt.block_on(pager_worker.run()); }); let application_shutdown = RelaxedAtomicBool::new(false); let job_stats = JobStats::new(); + let pager_for_shutdown = pager.clone(); + let create_server_args = CreateServerArgs { container_environment: container_environment.clone(), job_stats: job_stats.clone(), @@ -144,6 +150,7 @@ async fn main() -> AnyhowResult<()> { pager, }; + // HTTP server runs on a separate OS thread with its own actix System. std::thread::spawn(move || { let actix_runtime = actix_web::rt::System::new(); let http_server_handle = launch_http_server(create_server_args); @@ -151,11 +158,30 @@ async fn main() -> AnyhowResult<()> { actix_runtime.block_on(http_server_handle) .expect("HTTP server should not exit."); - warn!("Server thread is shut down."); - application_shutdown.set(true); + warn!("HTTP server thread is shut down."); + }); + + // Listen for SIGTERM / Ctrl-C to trigger graceful shutdown. + let application_shutdown_for_signal = application_shutdown.clone(); + + tokio::spawn(async move { + match tokio::signal::ctrl_c().await { + Ok(()) => { + info!("Received shutdown signal. Shutting down..."); + application_shutdown_for_signal.set(true); + } + Err(err) => { + warn!("Error listening for shutdown signal: {:?}", err); + } + } }); main_loop(job_dependencies).await; + info!("Shutting down pager worker..."); + pager_for_shutdown.shutdown_worker(); + + info!("Seedance2Pro job exiting."); + Ok(()) }