From 81efc03b75460c4c8a3887c3f5d3e4b74b32728a Mon Sep 17 00:00:00 2001 From: Timothy Allen Date: Fri, 2 Oct 2026 21:24:21 +0200 Subject: [PATCH] Keep the job thread alive after an error of the queue An error outside a job, such as a full disk, stopped the thread, and no job ran until a restart. The health check now answers 503 when a job thread stopped. --- README.md | 3 +++ bindery/orchestrator.py | 51 +++++++++++++++++++++++++++----------- bindery/web.py | 3 +++ tests/test_worker_flows.py | 38 +++++++++++++++++++++++++++- 4 files changed, 80 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 0487042..6b7bc97 100644 --- a/README.md +++ b/README.md @@ -426,6 +426,9 @@ After a restart, the waiting jobs run. A job that was running starts again at th queue, and skips the chapters that it stored. If Bindery stops during the same job three times, the job fails. +If a job thread stops, `/api/health` answers 503, so the container shows as unhealthy. Restart +Bindery to start the thread again. + ### Backups Once a day, Bindery copies its database and `personal.ini` to `/config/backups`, and keeps the diff --git a/bindery/orchestrator.py b/bindery/orchestrator.py index 416e39a..3e8f219 100644 --- a/bindery/orchestrator.py +++ b/bindery/orchestrator.py @@ -1290,25 +1290,38 @@ def _run_job(job: Job) -> str: return _run_site_job(job) +def _run_next_job() -> Job | None: + """Run the job at the front of the queue. Returns the job, or None if no job waits.""" + job = job_queue.get_next() + if not job: + return None + job_queue.update(job, status=QueueStatus.RUNNING, message="Starting") + try: + message = _run_job(job) + job_queue.update(job, status=QueueStatus.DONE, message=message, progress=1.0) + logger.info(f"Job {job.id} ({job.mode.value}) for {job.title}: {message}") + except (StoryError, CwaError, EpubError, RuleError, KeyError, ValueError) as e: + _fail(job, str(e)) + except Exception as e: + logger.exception(f"Job {job.id} failed: {e}") + _fail(job, f"{type(e).__name__}: {e}") + return job + + def download_loop() -> None: """Background thread for processing the job queue.""" logger.info("Starting download loop") while True: - job = job_queue.get_next() - if not job: - time.sleep(2) - continue - job_queue.update(job, status=QueueStatus.RUNNING, message="Starting") try: - message = _run_job(job) - job_queue.update(job, status=QueueStatus.DONE, message=message, progress=1.0) - logger.info(f"Job {job.id} ({job.mode.value}) for {job.title}: {message}") - except (StoryError, CwaError, EpubError, RuleError, KeyError, ValueError) as e: - _fail(job, str(e)) + job = _run_next_job() except Exception as e: - logger.exception(f"Job {job.id} failed: {e}") - _fail(job, f"{type(e).__name__}: {e}") - if job.mode in SITE_MODES: + # The queue or the store failed, for example on a full disk. The loop goes on, because no job runs + # after it stops. + logger.exception(f"Job loop error: {e}") + job = None + if job is None: + time.sleep(2) + elif job.mode in SITE_MODES: time.sleep(config.SITE_DELAY) @@ -1354,10 +1367,20 @@ def scheduler_loop() -> None: time.sleep(60) +_threads: list[threading.Thread] = [] + + def start_threads() -> None: try: adopt_build_keys() except Exception as e: logger.exception(f"Cannot check the built books: {e}") for target in (download_loop, scheduler_loop): - threading.Thread(target=target, daemon=True, name=target.__name__).start() + thread = threading.Thread(target=target, daemon=True, name=target.__name__) + thread.start() + _threads.append(thread) + + +def stopped_threads() -> list[str]: + """The names of the job threads that stopped. Bindery must restart to start them again.""" + return [t.name for t in _threads if not t.is_alive()] diff --git a/bindery/web.py b/bindery/web.py index 1cd33ee..8f54a36 100644 --- a/bindery/web.py +++ b/bindery/web.py @@ -201,6 +201,9 @@ def server_error(e: Exception) -> Any: @app.get("/api/health") def health() -> Any: + stopped = orchestrator.stopped_threads() + if stopped: + return {"status": "error", "error": f"Stopped: {', '.join(stopped)}. Restart Bindery."}, 503 return {"status": "ok"} diff --git a/tests/test_worker_flows.py b/tests/test_worker_flows.py index 8ff84b5..22d517d 100644 --- a/tests/test_worker_flows.py +++ b/tests/test_worker_flows.py @@ -6,7 +6,7 @@ from unittest.mock import Mock import pytest -from bindery import config, fff +from bindery import config, fff, web from bindery import orchestrator as o from bindery.diffing import text_hash from bindery.jobs import DownloadMode, Job, JobQueue, QueueStatus @@ -87,6 +87,42 @@ def test_worker_persists_outcomes_and_continues_after_failure(state, monkeypatch assert [c.args[0] for c in sleeps.call_args_list] == [2, 7] +def test_worker_goes_on_after_an_error_of_the_queue(state, monkeypatch, caplog): + store, queue = state + job = queue.add(Job(mode=DownloadMode.REBUILD)) + monkeypatch.setattr(queue, "get_next", Mock(side_effect=[OSError("disk full"), job, StopLoop()])) + monkeypatch.setattr(queue, "update", Mock(side_effect=[OSError("disk full"), None, None])) + sleeps = Mock() + monkeypatch.setattr(o.time, "sleep", sleeps) + with pytest.raises(StopLoop): + o.download_loop() + assert caplog.text.count("Job loop error: disk full") == 2 + assert [c.args[0] for c in sleeps.call_args_list] == [2, 2] + + +def test_health_reports_a_stopped_job_thread(state, monkeypatch): + client = web.app.test_client() + assert o.stopped_threads() == [] + alive, dead = Mock(), Mock() + alive.is_alive.return_value, dead.is_alive.return_value, dead.name = True, False, "download_loop" + monkeypatch.setattr(o, "_threads", [alive, dead]) + response = client.get("/api/health") + assert response.status_code == 503 + assert response.json == {"status": "error", "error": "Stopped: download_loop. Restart Bindery."} + monkeypatch.setattr(o, "_threads", [alive]) + assert client.get("/api/health").json == {"status": "ok"} + + +def test_start_threads_keeps_the_threads_for_the_health_check(state, monkeypatch): + monkeypatch.setattr(o, "_threads", []) + monkeypatch.setattr(o, "download_loop", lambda: None) + monkeypatch.setattr(o, "scheduler_loop", lambda: None) + o.start_threads() + for thread in o._threads: + thread.join() + assert o.stopped_threads() == ["", ""] + + def test_scheduler_skips_disabled_busy_and_recent_stories(state, monkeypatch): store, queue = state now = datetime(2026, 6, 1, 12)