diff --git a/queue_jobs.py b/queue_jobs.py index 6bcff3f..6267091 100644 --- a/queue_jobs.py +++ b/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): diff --git a/template_helpers.py b/template_helpers.py index f06faf0..a1f08ec 100644 --- a/template_helpers.py +++ b/template_helpers.py @@ -37,15 +37,32 @@ BROWSER_API_HELPER = r""" BROWSER_POLL_HELPER = r""" function startSerialPoll(callback, intervalMs) { let inFlight = false; - return window.setInterval(async () => { - if (inFlight) return; + let active = true; + let timer = null; + const stop = () => { + active = false; + if (timer !== null) { + window.clearTimeout(timer); + timer = null; + } + }; + const tick = async () => { + if (!active || inFlight) { + if (active) timer = window.setTimeout(tick, intervalMs); + return; + } inFlight = true; try { await callback(); } finally { inFlight = false; + if (active) timer = window.setTimeout(tick, intervalMs); } - }, intervalMs); + }; + window.addEventListener("pagehide", stop, { once: true }); + window.addEventListener("beforeunload", stop, { once: true }); + timer = window.setTimeout(tick, intervalMs); + return { stop }; } """