Worker · mcpd

worker 是什么

worker 持有那些可领取步骤的步骤主体，并向守护进程要活干。它跑在你的代码跑的地方，用的是你的依赖、你的连接池和你的凭据，而守护进程一样也看不到。守护进程发给它一份租约，里面是渲染好的输入和先前的那些步骤；worker 跑完步骤主体，再报告跑得怎么样。整个关系就是这些。

每个 SDK 都带了一个。它注册自己持有的工作流，用长轮询要租约，在步骤主体运行期间发心跳，并按原因逐一应对每一种拒绝。下面这个例子和工作流那一页上的是同一个程序；这里是把它服务起来的那一部分。

服务步骤主体

// Serve step bodies for as long as this process lives. The worker registers
// the definition, long-polls for leases, heartbeats while a body runs, and
// answers every 409 by its reason.
const worker = new Worker(mcpd, { concurrency: 8 }).register(support)
void worker.serve()

# Serve step bodies for as long as this process lives. serve() blocks, so it
# runs on a thread of its own when the process does other work too.
worker = Worker(mcpd, concurrency=8).register(support)
threading.Thread(target=worker.serve, daemon=True).start()

// Serve step bodies for as long as this process lives: register, long-poll
// for leases, heartbeat while a body runs, answer every 409 by its reason.
let worker = Worker::builder(&mcpd).concurrency(8).register(support).build();
let serving = {
    let worker = worker.clone();
    tokio::spawn(async move { worker.serve().await })
};

// Serve step bodies for as long as this process lives: register, long-poll
// for leases, heartbeat while a body runs, answer every 409 by its reason.
worker := mcpd.NewWorker(m, mcpd.WorkerConcurrency(8)).Register(support)
go func() { _ = worker.Serve(ctx) }()

// Serve step bodies for as long as this process lives: register, long-poll
// for leases, heartbeat while a body runs, answer every 409 by its reason.
val worker = Worker(mcpd, concurrency = 8).register(support)
launch { worker.serve() }

# Serve step bodies for as long as this node lives: register, long-poll for
# leases, heartbeat while a body runs, answer every 409 by its reason.
{:ok, worker} = StepWorker.start_link(mcpd, workflows: [support], concurrency: 8)
:ok = StepWorker.serve(worker)

协议

一共五个调用；会讲这五个调用的 worker，可以不用 SDK，直接对着 HTTP API 写出来。每个 SDK 的 worker 都只是它们的一个客户端，此外没有别的。

One lease

The daemon learns that a body is running from started, that the worker is alive from the heartbeats, and how the step ended from complete or fail. Miss the heartbeats and the lease is reclaimed; a later call about it is refused with a reason, never lost.

调用	它做什么
`POST /v1/api/workers/register`	宣告这个 worker，并存下它持有的定义。响应里带着 `lease_ttl_ms` 和 `heartbeat_ms`，协议其余的部分就靠这两个数字运转。
`POST /v1/api/steps/claim`	最多派出 `max` 个可运行的步骤；一个都没有时，把请求最多开着 `wait_ms`。守护进程挂住一次领取的时间，最多是它请求超时的五分之四，默认是 24 秒，并在它停下时立即应答，所以一次领取可能在 `wait_ms` 之前就空着回来：再领一次。一次领取可以限定在一次运行上，脚本就是这样把它自己的那次运行推到底的。
`POST /v1/api/steps/{lease}/started`	声明步骤主体已经在跑了。永远在步骤主体之前调用：守护进程就是靠它来区分“已派出”和“正在运行”的。
`POST /v1/api/steps/{lease}/heartbeat`	延长租约。这是守护进程唯一能拿到的、说明一个 worker 还活着的证据。
`POST /v1/api/steps/{lease}/complete`、`/fail`	结束这个步骤：输出、一份对运行状态的补丁，以及守护进程要定价的用量；或者一个错误，附带决定它是否重试的类别。

租约

一份租约就是一次尝试，由一个 worker 在一段有限的时间内持有。它带着渲染好的输入、按 id 索引的先前步骤、这次运行的状态和关联信息、一个在这个步骤的多次重试之间保持稳定的键，以及守护进程开的那个步骤 span 的 trace id 和 span id。如果步骤主体通过步骤上下文记录自己的 span，那些 span 会挂在这个 span 下面，于是一个步骤里的模型调用，会落在这次运行的 trace 里正确的深度上。

worker 不再发心跳的租约会过期。守护进程把这次尝试标记为孤儿，过了一个回收窗口之后，在 at_least_once 下把它重新入队，在 at_most_once 下把它判为失败。一个从暂停中缓过来、又对着已被回收的租约提交的 worker，不会被无视：它会被告知。

四种拒绝

守护进程接受不了的 worker 调用，会以一个 409 被拒绝，并带上四种原因之一。这组原因是封闭的，每个 SDK 的 worker 都按名字逐一处理，因为把它们塞进同一个分支的 worker，会丢掉自己已经做完的步骤，或者把一次成功报成失败。

原因	含义	worker 该怎么做
`already_applied`	这份租约自己之前的 complete 或 fail 已经生效，只是它的响应丢了。	当作成功，不再提交任何东西。
`fenced`	已经存在更晚的一次尝试，或者另一份租约结束了这一份。	放弃这份活，不再为这份租约提交任何东西。被栅栏挡下的心跳还会通过上下文取消正在运行的步骤主体。
`run_terminal`	这次运行已经结束了：完成、失败、超时，或者被取消。	停下来，并且不再为那次运行领取步骤。
`not_started`	一次 `at_most_once` 的尝试从来没有发过 `started`，却送来了 complete 或 fail。	把它当作 worker 里的 bug 大声报出来。SDK 自带的 worker 产生不了它；手写的可以。

步骤主体

步骤主体收到一个上下文、渲染好的输入，以及先前的那些步骤。上下文是它上报守护进程该保留什么的地方：一份对运行状态的补丁，在 complete 时合并，顶层的键直接替换；它花掉的 token，守护进程用给 span 定价的同一套价目表给它定价；还有一个取消信号，在租约丢失或运行结束时触发，所以把这个信号交给自己 I/O 的步骤主体，会在该停的时候停下来。token 会自己累加：步骤主体通过上下文记录的 span，以及它下面的每一个 span，结束时都会把自己的用量加到这个步骤上，所以把模型调用记成 span 的步骤主体从来不用手动求和；而 set_usage 是给一个 span 都不记的步骤主体准备的显式覆盖。

步骤主体以返回输出结束，或者以抛出结束。抛出 SDK 的可重试错误，就会在这个步骤的预算之内被重试；抛出一个有类型的步骤失败，可以指明其他任何类别；除此之外的都是致命的。错误消息会送到守护进程，并显示在事件流水里。调用栈不会。

如果一份租约对应的定义、版本或步骤是这个 worker 并不持有的，它会以 unservable 判为失败，步骤主体根本不会运行。这次尝试会回到队列里，等一个确实持有它的 worker，而且不消耗重试预算，这也正是一次发布过程中两个版本的 worker 可以并排运行的原因。

横向扩容

对着一个守护进程，你想跑多少个 worker 都可以。每个 worker 领取到自己的并发上限为止，守护进程把每一次尝试恰好交给其中一个；万一有两个 worker 同时持有同一份租约，栅栏会让它归结出唯一的赢家。worker 的 id 记录在它接过的每一次尝试上，所以事件流水能说出是哪个进程跑了某个步骤。给每个进程起个名字，或者让 SDK 从主机名推出一个。

领取是一次长轮询，所以一批闲着的 worker 只让守护进程为每个 worker 挂着一个请求，不产生流量。等待期间，一次领取会在一次运行启动、一个步骤结束、一个截止时间触发或一个事件恢复一次等待时去找活，其余时候每秒最多找两次。忙起来的时候，上限由 worker 自己的并发度决定；一个要跑一小时的步骤主体，代价是每十秒一次心跳，除此之外没有别的。

密钥

worker 的这些调用需要 worker 作用域；启动运行、给运行发信号和取消运行需要 execute。两者在未设置时都回退到写入密钥，而一把 admin 密钥同时具备这两者。一个只负责服务步骤的 worker，不需要能启动运行的密钥；一个只负责发布事件的 webhook，也不需要能领取步骤的密钥。相关的参数见配置。

worker 的调用不会记进审计日志，尽管一份租约带着这个步骤的输入和先前那些步骤的输出。它们是一个每分钟领取很多次的机器循环，每次领取记一行，会把人做的读取淹没。通过运行端点把一次运行读回来，是会被记录的；见安全。

接下来

启动一次运行、跟随它，以及答复一次等待：运行。
worker 服务的那份定义：工作流。
worker 协议的每一个端点，都由规范生成： HTTP API，Workers。