Repository navigation
refactor(tasks): converge task/work store contract - #6136
runyunzhou wants to merge 1 commit into
Conversation
qin-ctx
left a comment
There was a problem hiding this comment.
统一 Task/Work 状态归属的方向合理,但当前实现会让同步内容写入等路径等待自身的 Inline Work,导致请求无法正常完成;每次 Work 状态变化还会复制整个任务的全部 Work,引入明显的二次方开销。这两处回归需要在合并前解决。另有清理过程重复读取任务文件、终态错误明细失去上限的问题,详见 inline comments。
319636f to
ffde5d6
Compare
qin-ctx
left a comment
There was a problem hiding this comment.
本轮已验证:Work 整集合深拷贝、清理重复读取及终态错误明细上限三个问题已修复。同步等待的修复仍不完整:真实内容写入会创建锁协调消息,调用方等待协调消息,而协调消息又等待调用方的 Inline Work 结束,仍会持续重新入队。另有消费阶段为状态检查反复计算全部 Work 耗时的非阻塞性能问题。
| user_id=task_context.user_id, | ||
| ) | ||
| siblings_pending = tracker.has_work( | ||
| siblings_pending = await tracker.has_work( |
There was a problem hiding this comment.
[Bug] (blocking)
wait=true 的资源写入和批量写入仍会卡住:请求在等锁协调消息,锁协调消息却在等这个请求结束。
本轮把同步执行的 Inline Work ID 传入上下文,已经修好了“没有子消息时等待自身”的复现。但实际写入会通过 _enqueue_lock_coordinator() 产生 wait_for_request_completion=True 的 Semantic 消息。调用方在 _wait_for_request() 中等待这条消息;这里检查其他未完成工作时只排除了协调消息自身,仍把调用方的 Inline Work 算进去。即使子工作已经全部提交并完成,协调消息也会持续走重新入队分支;Inline Work 又只能在调用方返回后的 unregister_running_task() 中结束。
我用实际的任务包装、协调消息生产、入队中间件、SemanticProcessor 和等待函数做了相同输入的隔离对比,只替换存储和锁 I/O:原始 base 的协调消息能成功结束,请求返回;当前 Head 在 0.4 秒内连续重新入队 6 次,最终触发外层超时。这个新增验证覆盖了上轮简单复现缺少的真实协调消息,因此线程虽然标记为已解决,实际同步写入路径仍需要在合并前消除循环等待。
| account_id: Optional[str] = None, | ||
| user_id: Optional[str] = None, | ||
| ) -> bool: | ||
| task = await self.get_task(task_id, account_id=account_id, user_id=user_id) |
There was a problem hiding this comment.
[Design] (non-blocking)
消费大量向量消息时,仅检查任务是否取消,就会反复遍历和排序这个 Task 的全部 Work。上轮的整集合深拷贝已经修复,这里是消费阶段另一处可测量的开销。
每条消息开始执行时,中间件检查取消、注册运行句柄、再次检查取消,都会读取 Task 快照。这里的 get_task() 不只读取状态,还通过 _copy_task() 重新计算完整执行耗时;该计算遍历所有 Work,并排序已经开始执行的区间,而且运行在 Store 共用的线程锁内。
用相同的队列中间件消费入口、立即成功的 handler 和存储替身对比原始 base 与当前 Head:1600 条消息约为 0.007 秒 / 0.50 秒,3200 条约为 0.014 秒 / 1.76 秒,均不包含入队阶段。这里的调用方只需要状态和身份,却重复支付面向查询展示的全量耗时计算成本。
Description
重构 Task 与 QueueFS Work 的跟踪机制,统一收敛到与存储实现无关的
TaskWorkStore契约。本次改动明确拆分了持久化 Task 状态、Work 生命周期协调、QueueFS delivery
处理和进程内取消句柄。File Store 持久化 Task,并在服务重启时根据 QueueFS
恢复尚未完成的 Work;同时统一 Task 终态聚合、队列状态统计和取消处理。
Human Involvement
Related Issue
Type of Change
Changes Made
TaskWorkStore语义接口,覆盖状态迁移、取消、等待、清理、统计和恢复。_task_work_id的问题,确保 replacement delivery 沿用同一个逻辑 Work,不会遗留 pending Work 导致 Task 卡在running。CONTRIBUTING.md更新并精简测试:保留生命周期、持久化、取消、恢复、隔离和锁释放等公开契约,删除缓存结构、文件写次数和旧TaskWorkIndex测试双身相关断言。Testing
已执行:
766 passed。50 passed。26 passed。54 passed。83 passed。git diff --check均通过。Checklist
Screenshots (if applicable)
不适用。
Additional Notes
TaskWorkStore。