如何构建AI Agent的超时恢复与死信队列机制：FinSight AI WorkflowRecoveryScheduler完全指南-平芜编程栈

【免费下载链接】FinSight-AI

AI equity research agent with resilient workflows, evidence-grounded RAG, versioned reports, and automated quality evaluation.

项目地址：https://gitcode.com/gh_mirrors/fi/FinSight-AI

点击查看免费下载

FinSight AI 是一个 AI 股票研究 Agent，它把"财报采集 → 指标计算 → 文档索引 → 情报构建 → AI 分析"组织成一条长时运行工作流。这篇文章带你完整理解它的超时恢复（WorkflowRecoveryScheduler）与死信队列（Dead Letter Queue）机制：当某个 Agent 任务卡死、崩溃或反复失败时，系统如何自动识别、重试，最终把无法恢复的任务优雅地"停放"到死信区。

为什么 AI Agent 需要超时恢复与死信队列

AI 工作流天然昂贵且易失败：检索可能超时、模型调用可能挂起、索引可能只完成一半。如果任务卡死在RUNNING状态且无人处理，就会出现"沉默的部分失败"——用户永远等不到结果，资源也被白白占用。

FinSight 的设计目标很明确：让每个昂贵的阶段都是幂等的、可恢复的、可观测的。完整设计思路可参考 docs/design-agent-workflow.md。

任务状态机：先看懂 7 种状态

在讲恢复之前，先认识 WorkflowStatus.java 定义的状态机：

状态	含义
`CREATED`	任务已创建，等待执行
`RUNNING`	已获取租约，正在执行
`RETRYING`	失败后等待重新入队
`SUCCEEDED`	执行成功
`FAILED`	本次失败，可重试
`DEAD_LETTER`	重试次数耗尽，进入死信区
`COMPENSATING`	补偿中（预留状态）

每个任务还额外记录attempts（尝试次数）、errorMessage、leaseOwner和fencingToken（防重令牌），见 WorkflowTask.java。

核心机制一：WorkflowRecoveryScheduler 定时扫描超时任务

超时恢复的入口是 WorkflowRecoveryScheduler.java，它只做三件事：

扫描：每recovery-delay-ms（默认 60 秒）扫描一次数据库，找出所有状态为RUNNING且updatedAt早于"当前时间 − running-timeout（默认 10 分钟）"的任务。
标记：调用task.recoveredAfterTimeout(...)，根据attempts次数决定去向——尝试次数 ≥ 3 直接标记DEAD_LETTER，否则标记FAILED，并写入错误信息Workflow timed out at stage XXX。
重发布：若标记后状态为FAILED（可重试），则将任务转为RETRYING，通过WorkflowTaskPublisher重新投递到消息队列。

关键细节是第 38 行的saveIfOwned：它用状态 + fencingToken 的比较并置位（compare-and-set）来更新任务。这意味着即使"旧工人"其实还活着（比如只是 GC 停顿导致心跳变慢），它的过期写入也会被 fencingToken 挡住，不会覆盖恢复结果。

配套索引保证扫描高效：

CREATE INDEX idx_workflow_tasks_status_updated ON workflow_tasks(status, updated_at ASC);

见 V8__workflow_recovery_and_report_versions.sql。

核心机制二：RabbitMQ 侧的重试队列与死信交换器

调度器只负责"发现与标记"，真正把任务重新跑起来靠的是 RabbitMQ 的队列拓扑，定义在 RabbitWorkflowConfiguration.java。这里有三组队列配合工作：

主队列finsight.workflow.ingestion：绑定x-dead-letter-exchange，消息被拒绝（reject）后进入死信交换器。
重试队列finsight.workflow.retry：设置x-message-ttl = retry-delay-ms（默认 5 秒）。消息在这里"休眠"5 秒后过期，再由 TTL 死信机制自动路由回主队列——这是一个用TTL + DLX 实现的延时重试，无需额外的定时器线程。
死信队列finsight.workflow.dead-letter：最终无法消费的消息停在这里，方便人工排查与重放。

消费端的失败处理见 RabbitWorkflowListener.java：任务执行抛异常后，若数据库状态为FAILED，则转为RETRYING并发布到重试队列（publishRetry）；否则重新抛出异常，让消息走死信交换器。配合setDefaultRequeueRejected(false)，被拒绝的消息不会原地反复重投，避免热循环。

这样形成了完整的三层失败处理：

执行异常 → FAILED（尝试<3次）→ RETRYING → 重试队列（延时5s）→ 主队列 执行异常 → 尝试≥3次 → DEAD_LETTER（任务层死信，停止重试） 消费拒绝 → 消息层死信交换器 → finsight.workflow.dead-letter（队列层死信）

核心机制三：Redis 租约 + Fencing Token 防止重复执行

恢复机制有个隐含前提：任务必须幂等，且同一任务不能被两个执行者同时处理。FinSight 用 Redis Lua 脚本实现"单飞"租约，见 RedisBackedWorkflowLeaseService.java：

获取租约：原子地检查键是否存在 → 递增 fencing 计数器 → 写入owner:token（带 TTL）；
续租/释放：Lua 脚本校验owner:token一致后才操作，防止误删他人租约。

WorkflowOrchestrator.java 还会以TTL/3 的间隔发送租约心跳（finsight.workflow.lease.renewal.total指标记录 renewed / ownership_lost / failed 三种结果），确保长任务在 10 分钟恢复窗口内始终"活着"。

本地开发没有 Redis 时，租约服务会回退到进程内ConcurrentHashMap单飞表（allow-local-lease-fallback: true）；生产环境关闭回退、fail-closed，保证跨实例互斥不被静默绕过。

关键配置参数速查表

所有参数集中在 application.yml 的finsight.workflow段：

参数	默认值	作用
`running-timeout`	`PT10M`	RUNNING 状态超过该时长未更新即判定超时
`recovery-delay-ms`	`60000`	恢复扫描间隔（毫秒）
`retry-delay-ms`	`5000`	重试队列消息 TTL，即重试延迟
`dead-letter-queue`	`finsight.workflow.dead-letter`	队列层死信队列名
`allow-local-lease-fallback`	`true`	是否允许本地租约回退（生产建议`false`）

另外，每次恢复都会打点finsight.workflow.recovery.total（带taskType与stage标签），让"卡死任务"从隐性问题变成可监控、可告警的显性指标。

总结

FinSight AI 的可靠性设计可以浓缩为一句话：状态持久化 + 租约互斥 + 定时恢复 + 队列死信。

WorkflowRecoveryScheduler.java 负责"发现卡死、标记失败、触发重投"；
RabbitWorkflowConfiguration.java 用 TTL 队列 + DLX 实现延时重试与两层死信（任务层DEAD_LETTER+ 消息层死信队列）；
RedisBackedWorkflowLeaseService.java 用 Lua 租约与 fencingToken 保证恢复过程不会与仍在执行的旧工人冲突。

这套组合拳让 AI Agent 的长时工作流在模型超时、进程崩溃、消息积压等常见故障下，都能自动恢复或安全停下，而不是把失败藏在沉默里。