diff --git a/CHANGELOG.md b/CHANGELOG.md index 88e1649..afa33d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) ยท versioned wi - **Voice list rows no longer show the "Source unknown" badge** (it crowded the play buttons). The badge stays in the voice inspector, next to the Source & consent panel. ### Fixed +- **Background jobs could be garbage-collected mid-run.** Fire-and-forget `asyncio.create_task` calls (magazine, conversation, tts, stt, library, sources) now go through `core.tasks.spawn`, which keeps a strong reference until the task ends and logs failures by exception class only. - **Voice archetype gallery broke `npm run check:frontend`**: its `say` helper collided with `magazine.js`; renamed to `archSay`. - **Export failed with HTTP 500 when provenance tagging failed** (EPUB/ZIP/article MP3 with audio ffmpeg cannot tag); it now ships the untagged file and logs a warning. - **Archetype `language` filter accepted only full names.** `GET /api/archetypes?language=de` returned 0 items; it now accepts ISO codes and names, case-insensitive (`de`, `DE`, `German`); unknown values return an empty list. diff --git a/core/tasks.py b/core/tasks.py new file mode 100644 index 0000000..49e49c0 --- /dev/null +++ b/core/tasks.py @@ -0,0 +1,23 @@ +"""Fire-and-forget asyncio tasks that are kept alive until they finish.""" +from __future__ import annotations + +import asyncio +import logging + +log = logging.getLogger(__name__) +_BACKGROUND: set[asyncio.Task] = set() + + +def _done(task: asyncio.Task) -> None: + _BACKGROUND.discard(task) + if not task.cancelled() and task.exception() is not None: + # class name only: exception text may carry paths/URLs/keys + log.error("background task failed: %s", type(task.exception()).__name__) + + +def spawn(coro) -> asyncio.Task: + """create_task + strong reference (the loop only keeps weak ones).""" + task = asyncio.create_task(coro) + _BACKGROUND.add(task) + task.add_done_callback(_done) + return task diff --git a/routes/conversation.py b/routes/conversation.py index b37778a..838f0d9 100644 --- a/routes/conversation.py +++ b/routes/conversation.py @@ -23,6 +23,7 @@ from PIL import Image, ImageDraw, ImageOps from fastapi import APIRouter, File, Form, Header, HTTPException, Request, UploadFile from fastapi.responses import Response, StreamingResponse +from core.tasks import spawn from core.config import _load_settings, _save_settings, _clean_preview_backend from core.constants import _VOICES_DIR_DEFAULT, _MAX_UPLOAD_BYTES from core.registry import _registry_get, TEMP_DIR @@ -930,7 +931,7 @@ async def match_characters_voices(request: Request): oldest = min(_CAST_JOB_IDEMPOTENCY, key=lambda key: _CAST_JOB_IDEMPOTENCY[key][1]) _CAST_JOB_IDEMPOTENCY.pop(oldest, None) _cast_job_trim() - asyncio.create_task(_run_cast_job(job_id, job_data, {"idempotency-key": idempotency_key})) + spawn(_run_cast_job(job_id, job_data, {"idempotency-key": idempotency_key})) return {"job_id": job_id, "stage": "queued"} request_fingerprint = hashlib.sha256(json.dumps(data, sort_keys=True, default=str).encode()).hexdigest() if idempotency_key: diff --git a/routes/library.py b/routes/library.py index 12710ae..844fb79 100644 --- a/routes/library.py +++ b/routes/library.py @@ -11,6 +11,7 @@ from pathlib import Path from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile from fastapi.responses import FileResponse, Response, StreamingResponse +from core.tasks import spawn from core.config import _load_settings from core.constants import _VOICES_DIR_DEFAULT, _OUTPUT_DIR_DEFAULT, _VOICE_TARGET_DBFS, _VOICE_PEAK_DBFS, _BENCHMARK_SENTENCES, _MAX_TTS_OUTPUT_SECONDS, _MAX_UPLOAD_BYTES, _MAX_PICTURE_BYTES, _MAX_SOUND_BYTES, _ALLOW_PRIVATE_DOWNLOADS from core.registry import _registry_put, _registry_get, TEMP_DIR @@ -739,7 +740,7 @@ async def prepare_missing_references(request: Request): "errors": [], "started_at": datetime.now().isoformat(), } - asyncio.create_task(_run_prepare_missing_job(job_id, candidates, settings)) + spawn(_run_prepare_missing_job(job_id, candidates, settings)) return {"job_id": job_id, "queued": len(candidates)} diff --git a/routes/magazine.py b/routes/magazine.py index c2d7add..4994f5c 100644 --- a/routes/magazine.py +++ b/routes/magazine.py @@ -38,6 +38,7 @@ from fastapi.responses import FileResponse, Response from core import magazine as mag from core import magazine_untitled as untitled from core import word_alignment +from core.tasks import spawn from core.config import _load_settings from core.constants import CONFIG_DIR from core.database import ( @@ -511,7 +512,7 @@ async def magazine_ingest(doc_id: str, request: Request): meta["ingest"] = {"status": "running", "job_id": job_id} _write_meta(d, meta) - asyncio.create_task( + spawn( asyncio.to_thread(_ingest_sync, doc_id, job_id, ocr_lang, render_dpi, vision_ocr_url, vision_ocr_model, vision_ocr_fallback_url, vision_ocr_fallback_model, @@ -987,7 +988,7 @@ async def magazine_suggest_titles(doc_id: str): _require_magazine(doc_id) job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-suggest-titles", fingerprint=doc_id) - asyncio.create_task(asyncio.to_thread(_suggest_titles_sync, doc_id, job_id)) + spawn(asyncio.to_thread(_suggest_titles_sync, doc_id, job_id)) return {"job_id": job_id} @@ -1382,7 +1383,7 @@ async def magazine_reocr_page(doc_id: str, page: int, request: Request): raise HTTPException(409, f"Page {page} is still being read; wait for it to finish") job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-reocr-page", fingerprint=f"{doc_id}:{page}") - asyncio.create_task( + spawn( asyncio.to_thread(_reocr_page_sync, doc_id, job_id, page, vision_ocr_url, vision_ocr_model, vision_ocr_fallback_url, vision_ocr_fallback_model) @@ -1420,7 +1421,7 @@ async def magazine_rotate_page(doc_id: str, page: int, request: Request): raise HTTPException(409, f"Page {page} is still being turned or read; wait for it to finish before turning it again") job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-rotate-page", fingerprint=f"{doc_id}:{page}") - asyncio.create_task(asyncio.to_thread(_reocr_page_sync, doc_id, job_id, page, url, model, fb_url, fb_model, degrees)) + spawn(asyncio.to_thread(_reocr_page_sync, doc_id, job_id, page, url, model, fb_url, fb_model, degrees)) return {"job_id": job_id, "degrees": degrees} @@ -1471,7 +1472,7 @@ def _start_quality_page_job(doc_id: str, page: int, mode: str, job_id = uuid.uuid4().hex background_job_create(job_id, f"magazine-quality-{mode}", fingerprint=f"{doc_id}:{page}") - asyncio.create_task(asyncio.to_thread( + spawn(asyncio.to_thread( _quality_page_sync, doc_id, job_id, page, mode, str(settings.get("vision_ocr_url") or "").strip() or None, str(settings.get("vision_ocr_model") or "").strip() or None, @@ -1702,7 +1703,7 @@ async def magazine_quality_run(doc_id: str): _require_magazine(doc_id) job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-quality-run", fingerprint=doc_id) - asyncio.create_task(asyncio.to_thread(_quality_run_sync, doc_id, job_id, _load_settings())) + spawn(asyncio.to_thread(_quality_run_sync, doc_id, job_id, _load_settings())) return {"job_id": job_id} @@ -1770,7 +1771,7 @@ async def magazine_autofix_page(doc_id: str, page: int, request: Request): mode = "boxes" if isinstance(body, dict) and body.get("mode") == "boxes" else "all" job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-autofix", fingerprint=f"{doc_id}:{page}") - asyncio.create_task(asyncio.to_thread(_autofix_page_sync, doc_id, job_id, page, mode, _load_settings())) + spawn(asyncio.to_thread(_autofix_page_sync, doc_id, job_id, page, mode, _load_settings())) return {"job_id": job_id} @@ -2253,7 +2254,7 @@ async def magazine_refine_figures(doc_id: str): _require_magazine(doc_id) job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-refine-figures", fingerprint=doc_id) - asyncio.create_task(asyncio.to_thread(_refine_figures_sync, doc_id, job_id)) + spawn(asyncio.to_thread(_refine_figures_sync, doc_id, job_id)) return {"job_id": job_id} @@ -2267,7 +2268,7 @@ async def magazine_refine_page_figures(doc_id: str, page: int): job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-refine-page-figures", fingerprint=f"{doc_id}:{page}") - asyncio.create_task(asyncio.to_thread(_refine_figures_sync, doc_id, job_id, page)) + spawn(asyncio.to_thread(_refine_figures_sync, doc_id, job_id, page)) return {"job_id": job_id} @@ -2328,7 +2329,7 @@ async def magazine_add_region(doc_id: str, page: int, request: Request): job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-add-region", fingerprint=f"{doc_id}:{page}") - asyncio.create_task( + spawn( asyncio.to_thread(_add_region_sync, doc_id, job_id, page, [float(v) for v in bbox], kind, text, vision_ocr_url, vision_ocr_model, vision_ocr_fallback_url, vision_ocr_fallback_model) @@ -2548,7 +2549,7 @@ async def magazine_describe_image(doc_id: str, image_id: str, request: Request): job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-describe-image", fingerprint=f"{doc_id}:{image_id}") - asyncio.create_task( + spawn( asyncio.to_thread(_describe_image_sync, doc_id, job_id, image_id, vision_ocr_url, vision_ocr_model, vision_ocr_fallback_url, vision_ocr_fallback_model) @@ -3015,7 +3016,7 @@ async def magazine_spelling_agent(doc_id: str, request: Request): raise HTTPException(400, "No LLM model configured") job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-spelling", fingerprint=f"{doc_id}:{article or page or 'all'}") - asyncio.create_task(asyncio.to_thread(_spelling_agent_job, doc_id, job_id, article, page, settings)) + spawn(asyncio.to_thread(_spelling_agent_job, doc_id, job_id, article, page, settings)) return {"job_id": job_id} @@ -3221,7 +3222,7 @@ async def magazine_generate_article_audio(doc_id: str, article_id: str, request: raise HTTPException(400, "No narrator voice given") job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-article-audio", fingerprint=f"{doc_id}:{article_id}") - asyncio.create_task(asyncio.to_thread( + spawn(asyncio.to_thread( _audio_generate_one_sync, doc_id, job_id, article_id, voice, interview_voice)) return {"job_id": job_id} @@ -3397,7 +3398,7 @@ async def magazine_generate_batch_audio(doc_id: str, request: Request): raise HTTPException(400, "No approved articles to render") job_id = uuid.uuid4().hex background_job_create(job_id, "magazine-audio-batch", fingerprint=doc_id) - asyncio.create_task(asyncio.to_thread( + spawn(asyncio.to_thread( _audio_generate_batch_sync, doc_id, job_id, article_ids, voice, interview_voice)) return {"job_id": job_id, "count": len(article_ids)} diff --git a/routes/magazine_sources.py b/routes/magazine_sources.py index 5bdfb11..6146d0d 100644 --- a/routes/magazine_sources.py +++ b/routes/magazine_sources.py @@ -15,6 +15,7 @@ import uuid from fastapi import APIRouter, HTTPException, Request from core import magazine_sources as ms +from core.tasks import spawn from core.config import _load_settings, _save_settings from core.constants import CONFIG_DIR from core.database import background_job_create, background_job_update @@ -162,5 +163,5 @@ async def magazine_sources_import(request: Request): async with _import_lock: await asyncio.to_thread(_download_sync, doc_id, job_id, key, item_id) - asyncio.create_task(run()) + spawn(run()) return {"id": doc_id, "job_id": job_id} diff --git a/routes/sources.py b/routes/sources.py index 282f185..c0ba300 100644 --- a/routes/sources.py +++ b/routes/sources.py @@ -17,6 +17,7 @@ import requests from fastapi import APIRouter, HTTPException, Request from fastapi.responses import StreamingResponse +from core.tasks import spawn from core.config import _load_settings from core.constants import _VOICES_DIR_DEFAULT, _MAX_UPLOAD_BYTES, _ALLOW_PRIVATE_DOWNLOADS, CONFIG_DIR, _BENCHMARK_SENTENCES from core.registry import _registry_put, _registry_get, TEMP_DIR @@ -705,7 +706,7 @@ async def quick_import_voice(request: Request): _save_meta(wav_dest, meta) await asyncio.to_thread(upsert_voice_in_index, settings, wav_dest) if auto_prepare and not transcript: - asyncio.create_task(_prepare_imported_voice(wav_dest, settings)) + spawn(_prepare_imported_voice(wav_dest, settings)) return { "voice_id": final_id, "loudness": loudness, diff --git a/routes/stt.py b/routes/stt.py index 050aadd..db19f20 100644 --- a/routes/stt.py +++ b/routes/stt.py @@ -12,6 +12,7 @@ from pathlib import Path import requests from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile +from core.tasks import spawn from core.config import _load_settings from core.constants import ( _WHISPER_DEFAULT, _FASTER_WHISPER_DEFAULT, _WHISPER_CPP_DEFAULT, @@ -579,7 +580,7 @@ async def transcribe(request: Request): } background_job_create(job_id, "transcription", created=now) _stt_jobs_trim() - asyncio.create_task(_run_stt_job(job_id, src, settings, backend)) + spawn(_run_stt_job(job_id, src, settings, backend)) return {"job_id": job_id} try: text, used_backend = await _bounded_stt_call(_transcribe_audio, src, settings, backend) diff --git a/routes/tts.py b/routes/tts.py index d368203..12e3fb8 100644 --- a/routes/tts.py +++ b/routes/tts.py @@ -19,6 +19,7 @@ import requests from fastapi import APIRouter, HTTPException, Request from fastapi.responses import Response, StreamingResponse +from core.tasks import spawn from core.config import _load_settings, _clean_preview_backend, _preview_backend_base_url from core.constants import ( _VOICES_DIR_DEFAULT, _TTS_CONTAINER, _TTS_CONTAINERS_RAW, @@ -958,7 +959,7 @@ async def voice_design(request: Request): } background_job_create(job_id, "voice_design", created=_VOICE_DESIGN_JOBS[job_id]["created"]) _voice_design_job_trim() - asyncio.create_task(_run_voice_design_job(job_id, job_data, idempotency_key, request_fingerprint)) + spawn(_run_voice_design_job(job_id, job_data, idempotency_key, request_fingerprint)) return {"job_id": job_id, "stage": "queued"} return await _voice_design_execute(data, idempotency_key, request_fingerprint) diff --git a/tests/test_magazine_rotation.py b/tests/test_magazine_rotation.py index cfa3b5e..05971a8 100644 --- a/tests/test_magazine_rotation.py +++ b/tests/test_magazine_rotation.py @@ -408,7 +408,7 @@ class EndpointTests(unittest.TestCase): async def fake_task(coro): created.append(coro) with mock.patch.object(self.routes, "_load_settings", return_value=settings), \ - mock.patch.object(self.routes.asyncio, "create_task", side_effect=lambda c: (c.close(), created.append(1))[1]): + mock.patch.object(self.routes.asyncio, "create_task", side_effect=lambda c: (c.close(), created.append(1), mock.MagicMock())[2]): return self.client.post(f"/api/magazine/{self.doc}/pages/{page}/rotate", json={"degrees": degrees}), created def test_a_valid_request_starts_a_job(self): diff --git a/tests/test_tasks.py b/tests/test_tasks.py new file mode 100644 index 0000000..a6ac502 --- /dev/null +++ b/tests/test_tasks.py @@ -0,0 +1,37 @@ +import asyncio +import gc +import logging +import unittest + +from core import tasks + + +class SpawnTests(unittest.IsolatedAsyncioTestCase): + async def test_survives_gc_and_set_emptied(self): + done = asyncio.Event() + + async def work(): + await asyncio.sleep(0.01) + done.set() + + tasks.spawn(work()) + gc.collect() + self.assertEqual(len(tasks._BACKGROUND), 1) + await asyncio.wait_for(done.wait(), 1) + await asyncio.sleep(0) + self.assertEqual(len(tasks._BACKGROUND), 0) + + async def test_exception_logged_without_message(self): + async def boom(): + raise ValueError("secret-token") + + with self.assertLogs("core.tasks", "ERROR") as cm: + tasks.spawn(boom()) + await asyncio.sleep(0.02) + self.assertIn("ValueError", cm.output[0]) + self.assertNotIn("secret-token", cm.output[0]) + self.assertEqual(len(tasks._BACKGROUND), 0) + + +if __name__ == "__main__": + unittest.main()