fix: prevent worker not exiting if special case of same worker job

backend/windmill-worker/src/result_processor.rs

@@ -29,7 +29,7 @@ use windmill_queue::{
 
 use serde_json::{json, value::RawValue, Value};
 
-use tokio::task::JoinHandle;
+use tokio::{sync::Notify, task::JoinHandle};
 
 use windmill_queue::{add_completed_job, add_completed_job_error};
 
@@ -112,6 +112,7 @@ async fn process_jc(
 enum JobCompletedRx {
     JobCompleted(SendResult),
     Killpill,
+    WakeUp,
 }
 
 pub fn start_background_processor(
@@ -119,6 +120,7 @@ pub fn start_background_processor(
     job_completed_sender: JobCompletedSender,
     same_worker_queue_size: Arc<AtomicU16>,
     job_completed_processor_is_done: Arc<AtomicBool>,
+    wake_up_notify: Arc<Notify>,
     last_processing_duration: Arc<AtomicU16>,
     base_internal_url: String,
     db: DB,
@@ -155,7 +157,11 @@ pub fn start_background_processor(
                     }
                     result = bounded_rx.recv_async() => {
                         result.ok().map(JobCompletedRx::JobCompleted)
-                    }
+                    },
+                    _ = wake_up_notify.notified() => {
+                        tracing::info!("bg processor received wake up signal, checking if same worker queue is empty");
+                        Some(JobCompletedRx::WakeUp)
+                    },
                     _ = killpill_rx.recv() => {
                         tracing::info!("bg processor received killpill signal, queuing killpill job");
                         Some(JobCompletedRx::Killpill)
@@ -254,6 +260,8 @@ pub fn start_background_processor(
                 JobCompletedRx::Killpill => {
                     tracing::info!("killpill job received, processing only same worker jobs");
                     has_been_killed = true;
+                },  
+                JobCompletedRx::WakeUp => {
                 }
             }
         }

backend/windmill-worker/src/worker.rs

@@ -1198,12 +1198,15 @@ pub async fn run_worker(
     let job_completed_processor_is_done =
         Arc::new(AtomicBool::new(matches!(conn, Connection::Http(_))));
 
+    // This is used to wake up the background processor when main loop is done and just waiting for new same workers jobs, and that bg processor is also not processing any jobs, bg processing can exit if no more same worker jobs
+    let wake_up_notify = Arc::new(tokio::sync::Notify::new());
     let send_result = match (conn, job_completed_rx) {
         (Connection::Sql(db), Some(job_completed_receiver)) => Some(start_background_processor(
             job_completed_receiver,
             job_completed_tx.clone(),
             same_worker_queue_size.clone(),
             job_completed_processor_is_done.clone(),
+            wake_up_notify.clone(),
             last_processing_duration.clone(),
             base_internal_url.to_string(),
             db.clone(),
@@ -1489,6 +1492,7 @@ pub async fn run_worker(
                     tracing::info!(worker = %worker_name, hostname = %hostname, "all running jobs have completed and all completed jobs have been fully processed, exiting");
                     break;
                 } else {
+                    wake_up_notify.notify_one();
                     tracing::info!(worker = %worker_name, hostname = %hostname, "there may be same_worker jobs to process later, waiting for job_completed_processor to finish progressing all remaining flows before exiting");
                     tokio::time::sleep(Duration::from_millis(200)).await;
                     continue;