Add heartbeat_key_prefix in RQOrchestrator

Signed-off-by: Christoph Auer <cau@zurich.ibm.com>

Author: cau-git

docling_jobkit/orchestrators/rq/orchestrator.py

@@ -36,6 +36,7 @@ class RQOrchestratorConfig(BaseModel):
     results_ttl: int = 3_600 * 4
     failure_ttl: int = 3_600 * 4
     results_prefix: str = "docling:results"
+    heartbeat_key_prefix: str = "docling:job:alive"
     sub_channel: str = "docling:updates"
     scratch_dir: Optional[Path] = None
     redis_max_connections: int = 50
@@ -50,7 +51,6 @@ class _TaskUpdate(BaseModel):
     error_message: Optional[str] = None
 
 
-_HEARTBEAT_KEY_PREFIX = "docling:job:alive"
 _HEARTBEAT_TTL = 60  # seconds before an unrefreshed key expires
 _HEARTBEAT_INTERVAL = 20  # seconds between heartbeat writes
 _WATCHDOG_INTERVAL = 30.0  # seconds between watchdog scans
@@ -282,8 +282,8 @@ async def _watchdog_task(self) -> None:
 
         Runs every _WATCHDOG_INTERVAL seconds. For each task in STARTED state
         that is older than _WATCHDOG_GRACE_PERIOD, checks whether the worker's
-        liveness key (docling:job:alive:{task_id}) still exists in Redis. If the
-        key is absent the worker process has died — publishes a FAILURE update to
+        liveness key ({heartbeat_key_prefix}:{task_id}) still exists in Redis. If
+        the key is absent the worker process has died — publishes a FAILURE update to
         the pub/sub channel so that polling clients and WebSocket subscribers are
         notified within ~90 seconds of the kill instead of waiting 4 hours.
 
@@ -352,7 +352,7 @@ async def _watchdog_task(self) -> None:
                 )
 
                 for task_id in candidates:
-                    key = f"{_HEARTBEAT_KEY_PREFIX}:{task_id}"
+                    key = f"{self.config.heartbeat_key_prefix}:{task_id}"
                     alive = await self._async_redis_conn.exists(key)
                     age = (now - first_seen_started[task_id]).total_seconds()
                     _log.warning(

docling_jobkit/orchestrators/rq/worker.py

@@ -23,7 +23,6 @@
 from docling_jobkit.datamodel.task_meta import TaskStatus, TaskType
 from docling_jobkit.orchestrators.rq.orchestrator import (
     _HEARTBEAT_INTERVAL,
-    _HEARTBEAT_KEY_PREFIX,
     _HEARTBEAT_TTL,
     RQOrchestrator,
     RQOrchestratorConfig,
@@ -92,7 +91,7 @@ def _heartbeat_loop(self, job_id: str, stop_event: threading.Event) -> None:
         worker process is killed the thread dies with it and the key expires
         naturally, allowing the orchestrator watchdog to detect the dead job.
         """
-        key = f"{_HEARTBEAT_KEY_PREFIX}:{job_id}"
+        key = f"{self.orchestrator_config.heartbeat_key_prefix}:{job_id}"
         conn = None
         try:
             conn = sync_redis.Redis.from_url(self.orchestrator_config.redis_url)