make excutor process one cv at a time

main.py

@@ -27,7 +27,7 @@ from services.workers import CvWorker, QueueManager, CvTask
 converter = CVConverter()
 matcher = JobMatcher()  # SentenceTransformer loaded once at startup
 cv_worker = CvWorker(converter)
-queue_manager = QueueManager(cv_worker, concurrency=2)
+queue_manager = QueueManager(cv_worker, concurrency=1)
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):

services/cv_converter.py

@@ -316,22 +316,52 @@ class CVConverter:
             return True, 0, []
 
     def _load_marker_models(self):
-        """Lazy-load and cache Marker model dict (once per process)."""
+        """Lazy-load and cache Marker model dict (once per process).
+
+        PyTorch lazy / meta-device initialization can cause:
+          'Cannot copy out of meta tensor; no data!'
+        when Marker calls model.to(device) on a model that was built on the
+        meta device.  We patch torch.nn.Module.to to fall back to to_empty()
+        in that case, then restore the original after loading completes.
+        """
         if self._marker_models is None:
             logger.info(
                 "Loading Marker models for the first time "
                 "(this may take ~10–30 s)…"
             )
             try:
-                from marker.models import create_model_dict          # v1.x API
-                self._marker_models = create_model_dict()
-                logger.info("Marker models loaded and cached.")
+                import torch
+
+                _original_to = torch.nn.Module.to
+
+                def _safe_to(module, *args, **kwargs):
+                    try:
+                        return _original_to(module, *args, **kwargs)
+                    except RuntimeError as exc:
+                        if "Cannot copy out of meta tensor" in str(exc):
+                            device = args[0] if args else kwargs.get("device", "cpu")
+                            logger.debug(
+                                "Meta-tensor detected — using to_empty(device=%s)", device
+                            )
+                            return module.to_empty(device=device)
+                        raise
+
+                torch.nn.Module.to = _safe_to
+                try:
+                    from marker.models import create_model_dict
+                    self._marker_models = create_model_dict()
+                    logger.info("Marker models loaded and cached.")
+                finally:
+                    # Always restore the original .to even if loading fails
+                    torch.nn.Module.to = _original_to
+
             except ImportError as exc:
                 raise ImportError(
                     "Marker is not installed. Fix: pip install marker-pdf"
                 ) from exc
         return self._marker_models
 
+
     def _run_marker(self, path: Path, force_ocr: bool = False) -> str:
         """
         Execute Marker and return the raw Markdown string.

services/job_store.py

@@ -6,9 +6,10 @@ an in-memory dict is sufficient. Jobs are keyed by job_id (string UUID).
 
 Each entry shape:
 {
-    "status":  "PENDING" | "PROCESSING" | "COMPLETED" | "FAILED",
-    "result":  <ProcessorResponse payload dict> | None,
-    "error":   <str> | None,
+    "status":         "PENDING" | "QUEUED" | "PROCESSING" | "COMPLETED" | "FAILED",
+    "queue_position": int | None,  # position in queue (1 = next to process)
+    "result":         <ProcessorResponse payload dict> | None,
+    "error":          <str> | None,
 }
 """
 
@@ -19,17 +20,29 @@ from typing import Optional
 
 _lock = threading.Lock()
 _store: dict[str, dict] = {}
+_queue_counter = 0  # monotonic counter to assign queue positions
 
 
 def create_job(job_id: str) -> None:
     with _lock:
-        _store[job_id] = {"status": "PENDING", "result": None, "error": None}
+        _store[job_id] = {"status": "PENDING", "queue_position": None, "result": None, "error": None}
+
+
+def set_queued(job_id: str) -> None:
+    """Mark job as waiting in the Python processing queue."""
+    global _queue_counter
+    with _lock:
+        _queue_counter += 1
+        if job_id in _store:
+            _store[job_id]["status"] = "QUEUED"
+            _store[job_id]["queue_position"] = _queue_counter
 
 
 def set_processing(job_id: str) -> None:
     with _lock:
         if job_id in _store:
             _store[job_id]["status"] = "PROCESSING"
+            _store[job_id]["queue_position"] = None  # no longer in queue
 
 
 def set_completed(job_id: str, result: dict) -> None:

services/workers.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import tempfile
 import httpx
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Optional
@@ -12,6 +13,26 @@ from services import job_store
 from services.cv_chunker import chunk_cv
 from services.cv_converter import CVConverter
 
+
+# ---------------------------------------------------------------------------
+# Dedicated single-thread executor for ML work.
+#
+# Why not asyncio's default thread pool?
+#   asyncio.to_thread() uses the loop's default ThreadPoolExecutor which is
+#   shared with ALL other coroutines in the process (file I/O, HTTP clients,
+#   etc.).  When two heavy ML tasks run simultaneously they can saturate that
+#   shared pool, starving incoming HTTP requests of threads and making the
+#   server appear frozen even though the event loop is technically free.
+#
+#   A dedicated pool with max_workers=1 means:
+#   • ML work is 100% isolated — never competes with HTTP request threads.
+#   • Only one Marker/chunk_cv call runs at a time (model is not thread-safe).
+#   • The asyncio default pool stays free for file reads, httpx, etc.
+# ---------------------------------------------------------------------------
+
+_ML_EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="ml-worker")
+
+
 class CvWorker:
     """Handles background tasks for CV processing."""
 
@@ -26,11 +47,10 @@ class CvWorker:
         callback_url: str,
         callback_secret: str = None
     ) -> None:
-        """Background task: process CV then POST result to Java callback endpoint."""
+        """Process a CV and POST the result back to Java callback endpoint."""
+        loop = asyncio.get_running_loop()
         tmp_path: Optional[Path] = None
         try:
-            job_store.set_processing(job_id)
-
             start_time = datetime.now(timezone.utc).isoformat()
 
             suffix = Path(filename).suffix.lower() if filename else ".pdf"
@@ -38,7 +58,12 @@ class CvWorker:
                 tmp.write(file_bytes)
                 tmp_path = Path(tmp.name)
 
-            conversion = await asyncio.to_thread(self.converter.convert, tmp_path)
+            # Run blocking ML work in the dedicated single-thread executor.
+            # This keeps the asyncio event loop free to handle all other
+            # incoming requests (job-status polls, new CV submissions, etc.)
+            conversion = await loop.run_in_executor(
+                _ML_EXECUTOR, self.converter.convert, tmp_path
+            )
 
             if not conversion.success:
                 error_msg = conversion.error or "Conversion failed"
@@ -50,7 +75,9 @@ class CvWorker:
                 }, callback_secret)
                 return
 
-            chunks = await asyncio.to_thread(chunk_cv, conversion.markdown)
+            chunks = await loop.run_in_executor(
+                _ML_EXECUTOR, chunk_cv, conversion.markdown
+            )
             end_time = datetime.now(timezone.utc).isoformat()
 
             payload = {
@@ -105,14 +132,14 @@ class CvWorker:
             headers = {}
             if callback_secret:
                 headers["X-Callback-Secret"] = callback_secret
-                
+
             async with httpx.AsyncClient(timeout=30.0) as client:
                 response = await client.post(callback_url, json=body, headers=headers)
                 response.raise_for_status()
         except Exception as exc:
-            # Log but don't crash — Java will see job stay in PROCESSING and can time out
             print(f"[CV-ASYNC] Failed to POST callback to {callback_url}: {exc}")
 
+
 @dataclass
 class CvTask:
     job_id: str
@@ -121,47 +148,63 @@ class CvTask:
     callback_url: str
     callback_secret: str = None
 
+
 class QueueManager:
-    """Manages an asyncio Queue with bounded concurrent workers."""
-    
-    def __init__(self, worker: CvWorker, concurrency: int = 2):
+    """
+    Manages an asyncio Queue with a single background worker.
+
+    concurrency=1 is intentional — the Marker ML model and the
+    sentence-transformer are NOT thread-safe and must not run in
+    parallel on the same process.  Jobs queue up and are processed
+    one at a time.  The dedicated _ML_EXECUTOR above ensures ML work
+    never blocks the asyncio event loop — HTTP endpoints (job-status,
+    new submissions) stay responsive even while a CV is being processed.
+    """
+
+    def __init__(self, worker: CvWorker, concurrency: int = 1):
         self.worker = worker
         self.concurrency = concurrency
-        self.queue = asyncio.Queue()
-        self.tasks = []
+        self.queue: asyncio.Queue = asyncio.Queue()
+        self.tasks: list = []
 
-    async def start(self):
-        """Starts the background worker tasks."""
+    async def start(self) -> None:
+        """Spawn background worker coroutines."""
         for _ in range(self.concurrency):
             task = asyncio.create_task(self._worker_loop())
             self.tasks.append(task)
-        print(f"[QUEUE] Started {self.concurrency} concurrent workers.")
+        print(f"[QUEUE] Started {self.concurrency} worker(s) — one CV processed at a time.")
 
-    async def stop(self):
-        """Cancels all running workers."""
+    async def stop(self) -> None:
+        """Cancel all running workers cleanly."""
         for task in self.tasks:
             task.cancel()
         await asyncio.gather(*self.tasks, return_exceptions=True)
+        _ML_EXECUTOR.shutdown(wait=False)
         print("[QUEUE] Stopped all workers.")
 
-    async def _worker_loop(self):
-        """Continuously pulls tasks from the queue and processes them."""
+    async def _worker_loop(self) -> None:
+        """Continuously pull tasks from the queue and process them one by one."""
         while True:
             try:
-                task = await self.queue.get()
+                task: CvTask = await self.queue.get()  # suspends here — event loop stays free
+                job_store.set_processing(task.job_id)
+                print(f"[QUEUE] Worker picked up job {task.job_id}. "
+                      f"Remaining in queue: {self.queue.qsize()}")
                 await self.worker.run_cv_processing(
                     task.job_id,
                     task.file_bytes,
                     task.filename,
                     task.callback_url,
-                    task.callback_secret
+                    task.callback_secret,
                 )
                 self.queue.task_done()
             except asyncio.CancelledError:
                 break
-            except Exception as e:
-                print(f"[QUEUE] Unhandled error in worker loop: {e}")
-
-    async def enqueue(self, task: CvTask):
-        """Adds a new task to the queue."""
-        await self.queue.put(task)
+            except Exception as exc:
+                print(f"[QUEUE] Unhandled error in worker loop: {exc}")
+
+    async def enqueue(self, task: CvTask) -> None:
+        """Add a task to the queue and mark it QUEUED in the store immediately."""
+        job_store.set_queued(task.job_id)
+        await self.queue.put(task)  # asyncio.Queue with no maxsize — never blocks
+        print(f"[QUEUE] Job {task.job_id} enqueued. Queue depth: {self.queue.qsize()}")