Pipeline Backend Lifecycle Management#
The pipeline backend has three distinct lifecycle coordination challenges, all stemming from its callback-driven, concurrent streaming design — which is fundamentally different from the synchronous, sequential approach used by VLM and hybrid backends:
- PIL image race condition (hybrid backend): images passed by reference to async predictor threads are closed before those threads finish, triggering
EngineDeadError. - Output task submission during shutdown: the
on_doc_readycallback submits background futures that may not be joined before the server exits. - Health check gap: fatal engine exceptions inside individual task processing don't propagate to
last_worker_error, so/healthstays green andmineru-routernever restarts the dead worker.
Callback-Driven Streaming Architecture#
_process_pipeline() in common.py orchestrates inference and output as two concurrent streams:
- A
ThreadPoolExecutor(max_workers=1)is created for output writing . - An
on_doc_readycallback, when called, submitsrun_output_task()to the executor and appends theFuturetooutput_futures. doc_analyze_streaming()runs the windowed inference loop, firingon_doc_readywhen a document's last page window completes .- After the loop,
_process_pipeline()blocks on each future viafuture.result()to surface errors .
doc_analyze_streaming() _process_pipeline()
[window 1 inference] ──────► on_doc_ready(doc0)
[window 2 inference] └─ executor.submit(run_output_task) [parallel]
[window 3 inference] ──────► on_doc_ready(doc1)
[done] for future in output_futures: future.result()
The inference loop continues immediately without waiting for output to finish, overlapping file I/O with GPU work. A context['closed'] guard in _finalize_processing_window_context() ensures each document's callback fires exactly once .
In contrast, VLM and hybrid backends call _process_output() synchronously after each vlm_doc_analyze() / aio_vlm_doc_analyze() call — no background executor, no futures. The pipeline backend also has no true async code path: even when called from aio_do_parse(), it falls back to synchronous _process_pipeline() via asyncio.to_thread .
Race Condition 1 — PIL Image Lifetime vs. Async Predictor Threads (Hybrid Backend)#
Scope: hybrid-auto-engine backend, not pure pipeline. Documented as the most common production trigger for EngineDeadError .
Root cause: In the hybrid backend's doc_analyze() / aio_doc_analyze(), PIL images are passed by reference to async predictor threads running in a ThreadPoolExecutor. The window loop's finally block then calls _close_images(), closing those same image objects while in-flight threads may still be calling .resize() on them. Under concurrent load (≥2 PDFs), this raises ValueError: Operation on closed image, which propagates into the vLLM event loop and kills EngineCore with EngineDeadError .
Fix — PR #4979 (merged May 19, 2026): copy PIL images before passing them to the predictor in both the sync and async paths of hybrid_analyze.py:
# Before (race condition)
images_pil_list = [image_dict["img_pil"] for image_dict in images_list]
# After (safe copy)
images_pil_list = [image_dict["img_pil"].copy() for image_dict in images_list]
This decouples the image objects owned by the predictor thread from the ones closed by _close_images(), eliminating the most common EngineDeadError trigger. It does not address the general health-reporting gap (see below).
Race Condition 2 — Output Task Submission During Server Shutdown#
Scope: Pipeline backend, tracked in issue #5311 .
Root cause: The on_doc_ready callback fires and submits to output_executor while doc_analyze_streaming() is still running inference. If the server begins shutting down (e.g., due to a signal or stdin EOF) before _process_pipeline() reaches its for future in output_futures: future.result() join , the background output threads are abruptly interrupted. The task fails with an empty error message and no output files are written, even though inference completed successfully .
The ThreadPoolExecutor context manager (with ThreadPoolExecutor(...) as output_executor) does call executor.shutdown(wait=True) on exit, which should join pending futures. The vulnerability is at a higher level: if asyncio.to_thread(do_parse, ...) is cancelled (e.g., when AsyncTaskManager.shutdown() cancels active_tasks), the entire _process_pipeline() call — including the executor's __exit__ — may be interrupted from the asyncio side before shutdown can complete .
Workaround: Switch to the vlm-engine backend, which processes documents sequentially with no background futures .
Health Check Gap — Fatal Task Exceptions Not Propagated#
Tracked in: issue #5171. Fixed by: PR #5188.
How the health signal is supposed to work#
GET /health returns HTTP 503 only when AsyncTaskManager.is_healthy() returns False. is_healthy() returns False when:
dispatcher_taskisNoneor done (unexpectedly)cleanup_taskisNoneor done (when retention is enabled)last_worker_erroris non-None
mineru-router polls /health every 2 seconds (WORKER_REFRESH_INTERVAL_SECONDS) and restarts a local worker after 5 consecutive 503 responses (WORKER_HEALTH_FAILURE_RESTART_THRESHOLD = 5), giving a ~10-second recovery window once the signal is correct .
The gap#
_process_task() catches all exceptions in a bare except Exception block, marks the task failed, and calls _signal_task_event(task_id) — but does not set self.last_worker_error. By contrast, if the dispatcher loop itself crashes, it does write last_worker_error . Also, _on_processor_done (the task's done-callback) does write last_worker_error when the asyncio task raises an unhandled exception — but _process_task swallows all exceptions internally, so _on_processor_done never sees them .
Result: After EngineCore dies inside _process_task, the worker enters a "zombie-healthy" state: /health returns 200, the router keeps routing tasks to it, and every subsequent task fails immediately .
The fix (PR #5188)#
PR #5188 adds FATAL_WORKER_ERROR_MARKERS / FATAL_WORKER_ERROR_NORMALIZED_MARKERS constants and a _is_fatal_worker_error(exc) helper that inspects the exception's module, class hierarchy, and message string for patterns like "engine dead", "enginecore", "async llm engine has failed", and "unrecoverable" + "engine"/"vllm" .
Modified exception handler in _process_task:
except Exception as exc:
error = str(exc) or exc.__class__.__name__
task.status = TASK_FAILED
task.error = error
task.completed_at = utc_now_iso()
if _is_fatal_worker_error(exc):
self.last_worker_error = error
self._wake_waiters() # unblocks all waiters + marks unhealthy
else:
self._signal_task_event(task_id) # task-local signal only
logger.exception(f"Async task failed: {task_id}")
Document-specific failures (e.g., a corrupt PDF) continue to be task-local. Fatal engine failures trigger _wake_waiters(), which sets last_worker_error, causing /health to return 503 and the router to restart the worker within ~10 seconds .
Key Files and Entry Points#
| File | Relevance |
|---|---|
mineru/cli/fast_api.py | AsyncTaskManager — _process_task, is_healthy(), shutdown(), /health endpoint |
mineru/cli/common.py | _process_pipeline(), on_doc_ready callback, output future join |
mineru/backend/pipeline/pipeline_analyze.py | doc_analyze_streaming(), _finalize_processing_window_context() |
mineru/backend/hybrid/hybrid_analyze.py | PIL .copy() fix (PR #4979) |
mineru/cli/router.py | WORKER_HEALTH_FAILURE_RESTART_THRESHOLD, health monitor loop |
Related issues and PRs:
- Issue #5171 — health check gap for
EngineDeadError - PR #5188 — fatal error classification fix
- PR #4979 — PIL image copy fix for race condition
- Issue #5311 — output task submission during shutdown