Skip to content

refactor(tasks): converge task/work store contract - #6136

Open
runyunzhou wants to merge 1 commit into
volcengine:mainfrom
runyunzhou:refactor/task-work-store
Open

runyunzhou wants to merge 1 commit into
volcengine:mainfrom
runyunzhou:refactor/task-work-store

Conversation

@runyunzhou

Copy link
Copy Markdown
Contributor

Description

重构 Task 与 QueueFS Work 的跟踪机制,统一收敛到与存储实现无关的
TaskWorkStore 契约。

本次改动明确拆分了持久化 Task 状态、Work 生命周期协调、QueueFS delivery
处理和进程内取消句柄。File Store 持久化 Task,并在服务重启时根据 QueueFS
恢复尚未完成的 Work;同时统一 Task 终态聚合、队列状态统计和取消处理。

Human Involvement

  • A human participated in the implementation or review loop
  • This PR was generated entirely by AI agents without human participation in the loop

Related Issue

Type of Change

  • Bug fix (non-breaking change that fixes an issue)
  • New feature (non-breaking change that adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to not work as expected)
  • Documentation update
  • Refactoring (no functional changes)
  • Performance improvement
  • Test update

Changes Made

  • 新增与存储实现无关的 Task/Work 领域模型和 TaskWorkStore 语义接口,覆盖状态迁移、取消、等待、清理、统计和恢复。
  • 重构 File Store,采用 copy-on-write mutation、按记录加锁和有界 I/O;Task 持久化到文件,未完成 Work 由 QueueFS 快照恢复。
  • 将 QueueFS enqueue、process、requeue、cancel、clear 和终态聚合接入 Store 管理的 Work 状态。
  • 在队列 payload 中独立保留 Task owner,避免业务内容 owner 与 Task owner 混淆。
  • 修复写盘取消后的缓存发布、删除时锁内重新校验、clear 后 Work 悬挂以及 requeue 执行耗时累计错误。
  • 修复 ExternalTask requeue 丢失 _task_work_id 的问题,确保 replacement delivery 沿用同一个逻辑 Work,不会遗留 pending Work 导致 Task 卡在 running。
  • 按 CONTRIBUTING.md 更新并精简测试:保留生命周期、持久化、取消、恢复、隔离和锁释放等公开契约,删除缓存结构、文件写次数和旧 TaskWorkIndex 测试双身相关断言。

Testing

  • I have added tests that prove my fix is effective or that my feature works
  • New and existing unit tests pass locally with my changes
  • I have tested this on the following platforms:
    • Linux
    • macOS
    • Windows

已执行:

  • Task/Work、QueueFS、Session、Memory Compile、Semantic、Admin、Bot 和 AGFS restore 相关回归:766 passed。
  • ExternalTask 完成、取消、失败、重启和重试的参数化回归:50 passed。
  • Semantic 迁移聚焦测试:26 passed。
  • Task/Work 与 content-write 聚焦测试:54 passed。
  • Memory Compile 与 Session 测试:83 passed。
  • Ruff、Python 编译检查和 git diff --check 均通过。
  • 使用隔离配置和端口启动真实 OpenViking 服务,验证并发异步导入、事件持久化、用户隔离、取消、正常重启、SIGKILL 恢复和同步等待。

Checklist

  • My code follows the project's coding style
  • I have performed a self-review of my code
  • I have commented my code, particularly in hard-to-understand areas
  • I have made corresponding changes to the documentation
  • My changes generate no new warnings
  • Any dependent changes have been merged and published

Screenshots (if applicable)

不适用。

Additional Notes

  • File Store 中的 Work 状态是进程内状态,并通过 QueueFS 恢复,不提供跨节点一致性;分布式部署应使用共享的持久化 TaskWorkStore。

@qin-ctx qin-ctx left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

统一 Task/Work 状态归属的方向合理,但当前实现会让同步内容写入等路径等待自身的 Inline Work,导致请求无法正常完成;每次 Work 状态变化还会复制整个任务的全部 Work,引入明显的二次方开销。这两处回归需要在合并前解决。另有清理过程重复读取任务文件、终态错误明细失去上限的问题,详见 inline comments。

Comment thread openviking/service/task_tracker.py
Comment thread openviking/service/task_store.py Outdated
Comment thread openviking/service/task_store.py Outdated
Comment thread openviking/service/task_store.py Outdated
@runyunzhou
runyunzhou force-pushed the refactor/task-work-store branch from 319636f to ffde5d6 Compare October 15, 2026 16:27
@runyunzhou
runyunzhou requested a review from qin-ctx October 15, 2026 16:31
@ZaynJarvis ZaynJarvis added the kernel Core server, runtime, storage, retrieval, SDK, and CLI. label Oct 16, 2026

@qin-ctx qin-ctx left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

本轮已验证:Work 整集合深拷贝、清理重复读取及终态错误明细上限三个问题已修复。同步等待的修复仍不完整:真实内容写入会创建锁协调消息,调用方等待协调消息,而协调消息又等待调用方的 Inline Work 结束,仍会持续重新入队。另有消费阶段为状态检查反复计算全部 Work 耗时的非阻塞性能问题。

user_id=task_context.user_id,
)
siblings_pending = tracker.has_work(
siblings_pending = await tracker.has_work(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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 秒,均不包含入队阶段。这里的调用方只需要状态和身份,却重复支付面向查询展示的全量耗时计算成本。

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

kernel Core server, runtime, storage, retrieval, SDK, and CLI.

Projects

Status: Backlog

Development

Successfully merging this pull request may close these issues.

3 participants