Documentation
¶
Overview ¶
Package run 提供 worker agent run 命令的 cmd.run lane handler。
Handler 实现了基于持久化 inbox 的异步分发，提供 at-least-once 的崩溃恢复语义。 消息经过校验、持久化后，在后台 goroutine 中分发给 RunCoordinator， 使得 NATS Ack 可以立即返回，不阻塞消息确认。
确认决策：
- 永久错误（payload、route、model 校验失败）→ Term
- inbox 写入失败 → NakWithDelay(5s)，等待重试
- 持久化成功并启动后台 goroutine → Ack
- 重启后通过 RecoverNonTerminal 恢复未完成的任务
Index ¶
- func RequestFromWorkerTask(task runTask) *agentrundomain.RunRequest
- type Config
- type Handler
- func (h *Handler) Close() error
- func (h *Handler) Drain(timeout time.Duration) bool
- func (h *Handler) HandleControlCommand(ctx context.Context, cmd messaging.WorkerCommand) error
- func (h *Handler) HandleRunCommand(ctx context.Context, cmd messaging.WorkerCommand, ...) error
- func (h *Handler) RecoverNonTerminal(ctx context.Context) error
- func (h *Handler) RunInbox() inbox.RunInbox
- func (h *Handler) RunSubject() string
- func (h *Handler) StopAdmission()
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func RequestFromWorkerTask ¶
func RequestFromWorkerTask(task runTask) *agentrundomain.RunRequest
RequestFromWorkerTask converts the internal runTask into the agent runtime boundary.
Types ¶
type Config ¶
type Config struct {
OrgID uint
WorkerID uint
Env string
MaxConcurrency int
DebounceWindow time.Duration
InboxDBPath string // required
}
Config controls a worker run handler.
type Handler ¶
type Handler struct {
// contains filtered or unexported fields
}
Handler receives run commands and dispatches them asynchronously to the RunCoordinator.
func New ¶
New creates a worker run handler backed by the agentrun.Service through a Coordinator. InboxDBPath is required — the handler must not operate without a durable inbox.
func (*Handler) Drain ¶ added in v0.1.19
Drain 等待所有正在执行的后台分发任务（包括恢复 feeder）完成。 返回 true 表示在超时前全部完成，false 表示超时。 必须在 StopAdmission 之后调用，确保没有新的任务再加入。
func (*Handler) HandleControlCommand ¶
HandleControlCommand 处理 control lane 的控制命令（如 cancel）。 控制命令不需要手动确认，handler 同步完成后由 dispatcher 的自动 Ack 机制确认。
func (*Handler) HandleRunCommand ¶
func (h *Handler) HandleRunCommand(ctx context.Context, cmd messaging.WorkerCommand, delivery eventbus.ManualDelivery) error
HandleRunCommand 处理 run 命令，使用 ManualDelivery 手动控制确认时机。
确认决策：
- 永久错误（payload、route、model 校验失败）→ Term，不再重试
- inbox 写入失败 → NakWithDelay(5s)，请求 NATS 延迟重投
- 持久化成功 + 注册 inflight + 启动后台 goroutine → Ack，异步执行
func (*Handler) RecoverNonTerminal ¶ added in v0.1.19
RecoverNonTerminal 加载当前 topic 下所有非终态的 inbox 记录， 将它们注册为 owned（防止重复投递），然后启动后台 feeder goroutine 通过 semaphore 逐个恢复执行。
必须在 NATS 订阅开始前调用，确保崩溃恢复的消息不会被实时消息插队。
func (*Handler) RunSubject ¶
RunSubject returns the NATS subject for this handler's cmd.run lane.
func (*Handler) StopAdmission ¶ added in v0.1.19
func (h *Handler) StopAdmission()
StopAdmission 停止接受新的消息准入。 调用后不会有新的 WaitGroup.Add 发生，用于优雅关闭的第一步。
Source Files
Directories