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.
This commit is contained in:
@@ -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
|
queue, and skips the chapters that it stored. If Bindery stops during the same job three times, the
|
||||||
job fails.
|
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
|
### Backups
|
||||||
|
|
||||||
Once a day, Bindery copies its database and `personal.ini` to `/config/backups`, and keeps the
|
Once a day, Bindery copies its database and `personal.ini` to `/config/backups`, and keeps the
|
||||||
|
|||||||
+37
-14
@@ -1290,25 +1290,38 @@ def _run_job(job: Job) -> str:
|
|||||||
return _run_site_job(job)
|
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:
|
def download_loop() -> None:
|
||||||
"""Background thread for processing the job queue."""
|
"""Background thread for processing the job queue."""
|
||||||
logger.info("Starting download loop")
|
logger.info("Starting download loop")
|
||||||
while True:
|
while True:
|
||||||
job = job_queue.get_next()
|
|
||||||
if not job:
|
|
||||||
time.sleep(2)
|
|
||||||
continue
|
|
||||||
job_queue.update(job, status=QueueStatus.RUNNING, message="Starting")
|
|
||||||
try:
|
try:
|
||||||
message = _run_job(job)
|
job = _run_next_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:
|
except Exception as e:
|
||||||
logger.exception(f"Job {job.id} failed: {e}")
|
# The queue or the store failed, for example on a full disk. The loop goes on, because no job runs
|
||||||
_fail(job, f"{type(e).__name__}: {e}")
|
# after it stops.
|
||||||
if job.mode in SITE_MODES:
|
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)
|
time.sleep(config.SITE_DELAY)
|
||||||
|
|
||||||
|
|
||||||
@@ -1354,10 +1367,20 @@ def scheduler_loop() -> None:
|
|||||||
time.sleep(60)
|
time.sleep(60)
|
||||||
|
|
||||||
|
|
||||||
|
_threads: list[threading.Thread] = []
|
||||||
|
|
||||||
|
|
||||||
def start_threads() -> None:
|
def start_threads() -> None:
|
||||||
try:
|
try:
|
||||||
adopt_build_keys()
|
adopt_build_keys()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.exception(f"Cannot check the built books: {e}")
|
logger.exception(f"Cannot check the built books: {e}")
|
||||||
for target in (download_loop, scheduler_loop):
|
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()]
|
||||||
|
|||||||
@@ -201,6 +201,9 @@ def server_error(e: Exception) -> Any:
|
|||||||
|
|
||||||
@app.get("/api/health")
|
@app.get("/api/health")
|
||||||
def health() -> Any:
|
def health() -> Any:
|
||||||
|
stopped = orchestrator.stopped_threads()
|
||||||
|
if stopped:
|
||||||
|
return {"status": "error", "error": f"Stopped: {', '.join(stopped)}. Restart Bindery."}, 503
|
||||||
return {"status": "ok"}
|
return {"status": "ok"}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ from unittest.mock import Mock
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from bindery import config, fff
|
from bindery import config, fff, web
|
||||||
from bindery import orchestrator as o
|
from bindery import orchestrator as o
|
||||||
from bindery.diffing import text_hash
|
from bindery.diffing import text_hash
|
||||||
from bindery.jobs import DownloadMode, Job, JobQueue, QueueStatus
|
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]
|
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() == ["<lambda>", "<lambda>"]
|
||||||
|
|
||||||
|
|
||||||
def test_scheduler_skips_disabled_busy_and_recent_stories(state, monkeypatch):
|
def test_scheduler_skips_disabled_busy_and_recent_stories(state, monkeypatch):
|
||||||
store, queue = state
|
store, queue = state
|
||||||
now = datetime(2026, 6, 1, 12)
|
now = datetime(2026, 6, 1, 12)
|
||||||
|
|||||||
Reference in New Issue
Block a user