Tighten worker shutdown and poll helper cleanup

queue_jobs.py

@@ -16,8 +16,10 @@ from app_support import (
     PROJECT_ROOT,
     QUEUE_PATH,
     STATE_DB_PATH,
+    WORKER_SHUTDOWN_TIMEOUT_SECONDS,
     build_job,
     command_for_job,
+    debug_log,
     finalize_library_files,
     job_output_dir,
     job_staging_dir,
@@ -25,14 +27,12 @@ from app_support import (
     now_iso,
     printable_command,
     strip_control,
-    debug_log,
 )
 
 
 LOG_PERSIST_INTERVAL_SECONDS = 0.75
 LOG_PERSIST_LINE_BATCH = 8
 WATCHLIST_REFRESH_MAX_WORKERS = 4
-WORKER_JOIN_TIMEOUT_SECONDS = 2.5
 JOB_STRUCTURED_COLUMNS = (
     "show_id",
     "title",
@@ -97,7 +97,7 @@ class WatchlistRefreshJobs:
         self.wakeup.set()
         worker = self.worker
         if wait and worker is not None and worker.is_alive():
-            worker.join(timeout=WORKER_JOIN_TIMEOUT_SECONDS)
+            worker.join(timeout=WORKER_SHUTDOWN_TIMEOUT_SECONDS)
         return worker is None or not worker.is_alive()
 
     def _init_db(self):
@@ -241,6 +241,7 @@ class WatchlistRefreshJobs:
             self._run_job(job)
 
     def _run_job(self, job):
+        interrupted = False
         try:
             show_ids = self.store.all_show_ids()
             started_at = now_iso()
@@ -282,6 +283,10 @@ class WatchlistRefreshJobs:
                     self.current_executor = executor
                 futures = [executor.submit(refresh_one, show_id) for show_id in show_ids]
                 for index, future in enumerate(concurrent.futures.as_completed(futures), start=1):
+                    if self.stop_event.is_set():
+                        interrupted = True
+                        executor.shutdown(wait=False, cancel_futures=True)
+                        break
                     result = future.result()
                     title = result["title"]
                     with self.lock:
@@ -301,11 +306,13 @@ class WatchlistRefreshJobs:
 
             finished_at = now_iso()
             with self.lock:
-                job["status"] = "done"
+                job["status"] = "interrupted" if interrupted else "done"
                 job["finished_at"] = finished_at
                 job["updated_at"] = finished_at
                 job["current_title"] = None
-                if job["errors"]:
+                if interrupted:
+                    job["message"] = f"Watchlist refresh interrupted after {job['completed']} entries."
+                elif job["errors"]:
                     job["message"] = (
                         f"Refreshed {job['completed']} watchlist entries with {job['errors']} refresh errors."
                     )
@@ -366,10 +373,10 @@ class DownloadQueue:
         self.wakeup.set()
         worker = self.worker
         if wait and worker is not None and worker.is_alive():
-            worker.join(timeout=WORKER_JOIN_TIMEOUT_SECONDS)
+            worker.join(timeout=WORKER_SHUTDOWN_TIMEOUT_SECONDS)
             if worker.is_alive():
                 self._signal_active_process(signal.SIGKILL)
-                worker.join(timeout=WORKER_JOIN_TIMEOUT_SECONDS)
+                worker.join(timeout=WORKER_SHUTDOWN_TIMEOUT_SECONDS)
         return worker is None or not worker.is_alive()
 
     def _signal_active_process(self, signum):