From c90b5582fd5d5622b4de0837ee3707588166f3c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=A2=E6=96=87=E8=B0=A6?= Date: Sat, 4 Jul 2026 10:39:56 +0800 Subject: [PATCH] fix(worker): recover stale tasks from dead workers after restart Closes #2165 When a Hindsight container is restarted (OOM, docker kill, host crash), operations stuck in 'processing' status are never recovered because only matches the current worker's ID. After restart the new worker gets a different ID (hostname/container-id), so the old tasks remain stuck indefinitely, blocking all new consolidation. This PR adds which finds ALL processing tasks whose is older than 10 minutes (STALE_TASK_TIMEOUT_S=600) and resets them to 'pending'. It runs: - At startup (after recover_own_tasks) - Periodically during the polling loop (every 5 minutes) The 10-minute timeout avoids interfering with legitimately slow tasks (e.g. large consolidation batches) while still recovering promptly after a crash. --- .../hindsight_api/worker/poller.py | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/hindsight-api-slim/hindsight_api/worker/poller.py b/hindsight-api-slim/hindsight_api/worker/poller.py index 73d09161f0..e200aca550 100644 --- a/hindsight-api-slim/hindsight_api/worker/poller.py +++ b/hindsight-api-slim/hindsight_api/worker/poller.py @@ -202,6 +202,7 @@ def __init__( self._in_flight_lock = asyncio.Lock() self._last_progress_log = 0.0 self._tasks_completed_since_log = 0 + self._last_stale_recovery = 0.0 # Track active tasks locally: operation_id -> ActiveTaskInfo self._active_tasks: dict[str, ActiveTaskInfo] = {} # Track in-flight tasks by operation type @@ -821,6 +822,63 @@ async def recover_own_tasks(self) -> int: logger.info(f"Worker {self._worker_id} recovered {total_count} stale tasks from previous run") return total_count + # Maximum age (seconds) for a 'processing' task before it is considered + # stale and eligible for automatic recovery. The default of 10 minutes + # is generous enough to avoid interfering with legitimately slow tasks + # (e.g. large consolidation batches) while still recovering promptly + # after a crash / OOM-kill / docker restart. + STALE_TASK_TIMEOUT_S = 600 + + async def recover_stale_tasks(self) -> int: + """Recover tasks stuck in 'processing' from *any* dead worker. + + ``recover_own_tasks`` only handles tasks whose ``worker_id`` matches + the current process. When a container is restarted the new worker + gets a different ID (hostname / container-id), so tasks from the + previous worker remain stuck indefinitely (Issue #2165). + + This method finds *all* ``processing`` tasks whose ``claimed_at`` + is older than :pyattr:`STALE_TASK_TIMEOUT_S` and resets them to + ``pending`` so a live worker can pick them up. + + Returns: + Number of tasks recovered + """ + schemas = await self._get_schemas() + total_count = 0 + + for schema in schemas: + try: + table = fq_table("async_operations", schema) + async with self._backend.acquire() as conn: + result = await conn.execute( + f""" + UPDATE {table} + SET status = 'pending', + worker_id = NULL, + claimed_at = NULL, + updated_at = now(), + error_message = 'auto-recovered: stale processing task' + WHERE status = 'processing' + AND claimed_at < now() - interval '{self.STALE_TASK_TIMEOUT_S} seconds' + """ + ) + count = int(result.split()[-1]) if result else 0 + total_count += count + except Exception as e: + schema_display = f'"{schema}"' if schema else str(schema) + logger.warning( + f"Worker {self._worker_id} failed to recover stale tasks " + f"for schema {schema_display}: {e}" + ) + + if total_count > 0: + logger.info( + f"Worker {self._worker_id} recovered {total_count} stale tasks " + f"from dead workers (timeout={self.STALE_TASK_TIMEOUT_S}s)" + ) + return total_count + async def _recover_batch_operations(self, schema: str | None) -> int: """ Recover batch API operations that were in-flight when worker crashed. @@ -898,6 +956,7 @@ async def run(self): and immediately continues polling (up to slot limits). """ await self.recover_own_tasks() + await self.recover_stale_tasks() reservations_str = ( ", ".join(f"{k}={v}" for k, v in self._slot_reservations.items()) if self._slot_reservations else "none" @@ -953,6 +1012,8 @@ async def run(self): # Log progress stats periodically await self._log_progress_if_due() + # Periodically recover stale tasks from dead workers + await self._recover_stale_if_due() except asyncio.CancelledError: logger.info(f"Worker {self._worker_id} polling loop cancelled") @@ -1318,6 +1379,21 @@ def _maybe_dump_stuck_stack(self, op_id: str, info: ActiveTaskInfo, age_s: float # Stack capture is best-effort - never crash the polling loop over it. logger.debug(f"Failed to capture stack for {op_id}: {e}") + # How often (seconds) to sweep for stale tasks from dead workers. + # Runs in the idle polling loop so does not interfere with normal work. + STALE_RECOVERY_INTERVAL_S = 300 # 5 minutes + + async def _recover_stale_if_due(self) -> None: + """Periodically recover stale tasks from dead workers.""" + now = time.time() + if now - self._last_stale_recovery < self.STALE_RECOVERY_INTERVAL_S: + return + self._last_stale_recovery = now + try: + await self.recover_stale_tasks() + except Exception: + logger.warning("Periodic stale-task recovery failed", exc_info=True) + async def _log_db_waits(self) -> None: """Log any non-idle hindsight session that's waiting on a lock or other resource.