preview: increase timeout to 10s, add priority queue scheduling (images > video > pdf > office)

cista/onlyoffice.py

@@ -145,6 +145,8 @@ def convert_to_png(file_path: Path, timeout: float = 30.0) -> bytes:
         headers = {"Content-Type": "application/json"}
         token = _build_jwt_token(payload)
         if token:
+            # Conversion API expects JWT in request body when token checks are enabled.
+            payload["token"] = token
             headers["Authorization"] = token
 
         req = urllib.request.Request(  # noqa: S310

cista/preview.py

@@ -90,7 +90,7 @@ class PreviewCache:
 # Global preview cache instance
 _preview_cache = PreviewCache(capacity=500)
 
-PREVIEW_TIMEOUT = 3.0  # seconds until preview subprocess is killed
+PREVIEW_TIMEOUT = 10.0  # seconds until preview subprocess is killed
 PREVIEW_WORKERS = max(2, min(8, cpu_count()))
 _active_procs: set[asyncio.subprocess.Process] = set()
 _preview_pool = None
@@ -164,7 +164,12 @@ class _PreviewWorkerPool:
     def __init__(self, size: int):
         self.size = size
         self._idle: asyncio.Queue[_PreviewWorker] = asyncio.Queue()
+        self._pending: asyncio.PriorityQueue[tuple[int, int, asyncio.Future, tuple]] = (
+            asyncio.PriorityQueue()
+        )
         self._workers: set[_PreviewWorker] = set()
+        self._dispatchers: list[asyncio.Task] = []
+        self._seq = 0
         self._closed = False
 
     async def _spawn_worker(self) -> _PreviewWorker:
@@ -195,33 +200,43 @@ class _PreviewWorkerPool:
         except Exception:
             logger.exception("Failed to replace preview worker")
 
-    async def start(self) -> None:
-        for _ in range(self.size):
-            await self._add_worker()
+    async def _dispatch_loop(self) -> None:
+        while True:
+            try:
+                _priority, _seq, future, args = await self._pending.get()
+            except asyncio.CancelledError:
+                return
+
+            if future.cancelled():
+                continue
 
-    async def run(self, filepath, quality: int, maxsize: int, maxzoom: float):
-        if self._closed:
-            raise PreviewError("preview worker pool closed")
             worker = await self._idle.get()
+            filepath = args[0]
             replace = False
             try:
                 out, resp = await asyncio.wait_for(
-                worker.request(filepath, quality, maxsize, maxzoom),
+                    worker.request(*args),
                     timeout=PREVIEW_TIMEOUT,
                 )
-            return out, resp
+                if not future.done():
+                    future.set_result((out, resp))
             except TimeoutError:
                 replace = True
                 logger.warning(
                     "Preview timeout (%ds) for %s", int(PREVIEW_TIMEOUT), filepath.name
                 )
-            raise PreviewTimeoutError(filepath.name) from None
-        except WorkerChecksumError as e:
+                if not future.done():
+                    future.set_exception(PreviewTimeoutError(filepath.name))
+            except WorkerChecksumError:
                 replace = True
                 logger.error("Preview checksum mismatch for %s", filepath.name)
-            raise PreviewError(f"worker checksum mismatch for {filepath.name}") from e
-        except PreviewError:
-            raise
+                if not future.done():
+                    future.set_exception(
+                        PreviewError(f"worker checksum mismatch for {filepath.name}")
+                    )
+            except PreviewError as e:
+                if not future.done():
+                    future.set_exception(e)
             except (
                 WorkerProtocolError,
                 asyncio.IncompleteReadError,
@@ -235,9 +250,12 @@ class _PreviewWorkerPool:
                 logger.warning(
                     "Preview worker protocol failure for %s: %s", filepath.name, e
                 )
-            raise PreviewError(
+                if not future.done():
+                    future.set_exception(
+                        PreviewError(
                             f"worker protocol failure for {filepath.name}: {e}"
-            ) from e
+                        )
+                    )
             finally:
                 if replace:
                     await self._replace_worker(worker)
@@ -246,10 +264,44 @@ class _PreviewWorkerPool:
                 else:
                     await self._replace_worker(worker)
 
+    async def start(self) -> None:
+        for _ in range(self.size):
+            await self._add_worker()
+        for _ in range(self.size):
+            self._dispatchers.append(asyncio.create_task(self._dispatch_loop()))
+
+    async def run(self, filepath, quality: int, maxsize: int, maxzoom: float):
+        if self._closed:
+            raise PreviewError("preview worker pool closed")
+        loop = asyncio.get_running_loop()
+        future = loop.create_future()
+        self._seq += 1
+        await self._pending.put(
+            (
+                _preview_job_priority(filepath),
+                self._seq,
+                future,
+                (filepath, quality, maxsize, maxzoom),
+            )
+        )
+        return await future
+
     async def close(self) -> None:
         self._closed = True
+        for task in self._dispatchers:
+            task.cancel()
+        if self._dispatchers:
+            await asyncio.gather(*self._dispatchers, return_exceptions=True)
+        self._dispatchers.clear()
         workers = list(self._workers)
         self._workers.clear()
+        while not self._pending.empty():
+            try:
+                _priority, _seq, future, _args = self._pending.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+            if not future.done():
+                future.set_exception(PreviewError("preview worker pool closed"))
         while not self._idle.empty():
             try:
                 self._idle.get_nowait()
@@ -372,6 +424,24 @@ OFFICE_PREVIEW_SUFFIXES = {
 }
 
 
+def _preview_job_priority(path) -> int:
+    """Return priority for preview job (lower=higher priority).
+
+    Priority order: images (0) < video (1) < PDF (2) < office (3) < unknown (4)
+    """
+    suffix = path.suffix.lower()
+    if suffix in DOC_PREVIEW_SUFFIXES:
+        return 2
+    if suffix in OFFICE_PREVIEW_SUFFIXES:
+        return 3
+    mime_type, _ = mimetypes.guess_type(path.name)
+    if mime_type and mime_type.startswith("image/"):
+        return 0
+    if mime_type and mime_type.startswith("video/"):
+        return 1
+    return 4
+
+
 def is_previewable_path(path) -> bool:
     suffix = path.suffix.lower()
     if suffix in DOC_PREVIEW_SUFFIXES or suffix in OFFICE_PREVIEW_SUFFIXES: