Harden queue worker crashes and POST validation

2026-05-16 11:14:52 +02:00
parent 92762b07d5
commit d5f81d15b7
7 changed files with 150 additions and 59 deletions
@@ -768,6 +768,29 @@ class DownloadQueue:
            self._save_job_locked(job)
        self._cleanup_staging_dir(job)

+    def _fail_job_runtime(self, job, exc, process=None):
+        if process is not None:
+            try:
+                os.killpg(os.getpgid(process.pid), signal.SIGTERM)
+            except OSError:
+                try:
+                    process.terminate()
+                except OSError:
+                    pass
+        message = f"Download worker failed unexpectedly: {exc}"
+        with self.lock:
+            job["status"] = "failed"
+            job["exit_code"] = 1
+            job["pid"] = None
+            job["finished_at"] = now_iso()
+            job["updated_at"] = job["finished_at"]
+            job.setdefault("log", []).append(message)
+            self.current_process = None
+            self.current_job_id = None
+            self.current_job = None
+            self._save_job_locked(job)
+        self._cleanup_staging_dir(job)
+
    def _run(self):
        while not self.stop_event.is_set():
            job = self._next_pending()
@@ -830,56 +853,59 @@ class DownloadQueue:
            job["pid"] = process.pid
            self._save_job_locked(job)

-        assert process.stdout is not None
-        for line in process.stdout:
-            self._append_log(job, line)
-        exit_code = process.wait()
-        finalize_error = None
-        moved_files = []
-        watchlist_sync_error = None
-        cleanup_staging = False
-        if exit_code == 0 and not job.get("cancel_requested"):
-            try:
-                moved_files = finalize_library_files(job)
-            except Exception as exc:
-                finalize_error = exc
-            else:
-                if self.watchlist_sync_fn is not None:
-                    try:
-                        self.watchlist_sync_fn(job)
-                    except Exception as exc:
-                        watchlist_sync_error = exc
+        try:
+            assert process.stdout is not None
+            for line in process.stdout:
+                self._append_log(job, line)
+            exit_code = process.wait()
+            finalize_error = None
+            moved_files = []
+            watchlist_sync_error = None
+            cleanup_staging = False
+            if exit_code == 0 and not job.get("cancel_requested"):
+                try:
+                    moved_files = finalize_library_files(job)
+                except Exception as exc:
+                    finalize_error = exc
+                else:
+                    if self.watchlist_sync_fn is not None:
+                        try:
+                            self.watchlist_sync_fn(job)
+                        except Exception as exc:
+                            watchlist_sync_error = exc

-        with self.lock:
-            canceled = job.get("cancel_requested")
-            job["exit_code"] = 1 if finalize_error else exit_code
-            job["pid"] = None
-            job["finished_at"] = now_iso()
-            job["updated_at"] = job["finished_at"]
-            if canceled:
-                job["status"] = "canceled"
-                job.setdefault("log", []).append("Canceled.")
-            elif finalize_error:
-                job["status"] = "failed"
-                job.setdefault("log", []).append(f"Library rename failed: {finalize_error}")
-            elif exit_code == 0:
-                job["status"] = "done"
-                for path in moved_files:
-                    job.setdefault("log", []).append(f"Saved: {path}")
-                job.setdefault("log", []).append("Download completed.")
-                if watchlist_sync_error:
-                    job.setdefault("log", []).append(f"Watchlist sync failed: {watchlist_sync_error}")
-                elif self.watchlist_sync_fn is not None:
-                    job.setdefault("log", []).append("Watchlist download state synced.")
-            else:
-                job["status"] = "failed"
-                job.setdefault("log", []).append(f"Download failed with exit code {exit_code}.")
-                cleanup_staging = True
-            if canceled:
-                cleanup_staging = True
-            self.current_process = None
-            self.current_job_id = None
-            self.current_job = None
-            self._save_job_locked(job)
-        if cleanup_staging:
-            self._cleanup_staging_dir(job)
+            with self.lock:
+                canceled = job.get("cancel_requested")
+                job["exit_code"] = 1 if finalize_error else exit_code
+                job["pid"] = None
+                job["finished_at"] = now_iso()
+                job["updated_at"] = job["finished_at"]
+                if canceled:
+                    job["status"] = "canceled"
+                    job.setdefault("log", []).append("Canceled.")
+                elif finalize_error:
+                    job["status"] = "failed"
+                    job.setdefault("log", []).append(f"Library rename failed: {finalize_error}")
+                elif exit_code == 0:
+                    job["status"] = "done"
+                    for path in moved_files:
+                        job.setdefault("log", []).append(f"Saved: {path}")
+                    job.setdefault("log", []).append("Download completed.")
+                    if watchlist_sync_error:
+                        job.setdefault("log", []).append(f"Watchlist sync failed: {watchlist_sync_error}")
+                    elif self.watchlist_sync_fn is not None:
+                        job.setdefault("log", []).append("Watchlist download state synced.")
+                else:
+                    job["status"] = "failed"
+                    job.setdefault("log", []).append(f"Download failed with exit code {exit_code}.")
+                    cleanup_staging = True
+                if canceled:
+                    cleanup_staging = True
+                self.current_process = None
+                self.current_job_id = None
+                self.current_job = None
+                self._save_job_locked(job)
+            if cleanup_staging:
+                self._cleanup_staging_dir(job)
+        except Exception as exc:
+            self._fail_job_runtime(job, exc, process=process)