fix(dispatcher): 僵尸PROCESSING任务回收 + fam-core文件日志
- db_layer 新增 reclaim_stale_processing: PROCESSING 超过 push_timeout+120s 重置 PENDING - dispatcher 轮询前先回收僵尸任务(进程重启/Edge重启导致 in-flight 请求丢失的场景) - logger 增加 fam-core/logs/fam-core.log 文件输出(daemon 模式 stdout 不可见)
This commit is contained in:
@@ -149,6 +149,16 @@ class Dispatcher:
|
||||
|
||||
def _poll_once(self):
|
||||
"""执行一次轮询"""
|
||||
# 先回收僵尸任务:PROCESSING 超过 push_timeout+缓冲 说明推送进程已丢失
|
||||
# (如 fam-core 重启、Edge 重启掐断 in-flight 连接),重置回 PENDING 走重试
|
||||
stale_timeout = self.push_timeout + 120
|
||||
try:
|
||||
stale_ids = db_layer.reclaim_stale_processing(stale_timeout)
|
||||
for tid in stale_ids:
|
||||
logger.warning(f"[task_id={tid}] PROCESSING 超时 {stale_timeout}s,回收为 PENDING 重试")
|
||||
except Exception as e:
|
||||
logger.error(f"僵尸任务回收失败: {e}", exc_info=True)
|
||||
|
||||
tasks = db_layer.get_pending_tasks(limit=10)
|
||||
for task in tasks:
|
||||
if self._should_retry(task):
|
||||
|
||||
Reference in New Issue
Block a user