拉取模式 Worker

每个 Dex Worker 都是拉取模式 Worker。它主动向 Dex Server 建立任务 stream，从不监听入站调用。 Worker 只需要能出站连接 Dex Server，因此可以位于 NAT 或防火墙后，无需开放入站端口。 如果 Worker 在内存中保存了后续调用会用到的状态，也可以利用黏性路由。

启动 Worker

设置 Worker 的 Dex Server 地址。pull 选项是可选的，用于调整任务容量、注册和关闭行为。 Client 只需要 Dex Server 地址。Dex 将所有 Flow、SubFlow、Step 和 RPC 调用发送给拉取模式 Worker，因此无需配置 Worker 地址或目标。

以下示例使用拉取模式 Worker 处理一个 Greeting Flow。Step 写入问候语，问候语 RPC 读取它。 启动细节见 Go、 Python、 Java、 TypeScript 和 Rust SDK 的 README。

Dex Web v2 通过拉取模式 Worker 启动 Flow。Start Flow 不再要求填写 Worker address， 启动请求也可以发送 workerTargetPull: true。Dex Web 不会在启动前检查 pull Worker 是否已注册； Dex 在分配工作时确定其可用性。见 Dex Web。

async def start_pull_greeting(
    cache: BlobCache, server_address: str
) -> tuple[GreetingFlow, AsyncWorker, AsyncClient, asyncio.Task[None]]:
    flow = GreetingFlow()
    registry = Registry((flow,), allow_async_handlers=True)
    worker = AsyncWorker(
        registry,
        cache,
        WorkerOptions(
            server_address=server_address,
            pull=PullWorkerOptions(
                max_concurrent_tasks=8,
                # Keep one slot for RPCs so Flows stay responsive.
                reserved_rpc_tasks=1,
            ),
        ),
    )
    worker_task = asyncio.create_task(worker.start())
    client = AsyncClient(
        registry,
        cache,
        ClientOptions(server_address=server_address),
    )
    try:
        await worker.wait_pull_registered(timeout=30)
    except BaseException:
        await stop_pull_greeting(worker, client, worker_task)
        raise
    return flow, worker, client, worker_task


async def stop_pull_greeting(
    worker: AsyncWorker, client: AsyncClient, worker_task: asyncio.Task[None]
) -> None:
    await client.close()
    await worker.close()
    await worker_task

例子: examples/python/dex_examples/production/pull_workers.py

就绪检查

从 Go SDK 1.13.0 开始，WaitPullRegistered 会等待所有 Flow type 队列和 app Worker ID 队列收到注册确认后再允许启动流程继续。Java 提供 waitPullRegistered(Duration) 和 getPullRegistrationState()。 Python 的 Worker 和 AsyncWorker 均提供 wait_pull_registered(timeout) 和 pull_registration_state()；AsyncWorker 的等待方法需要 await。 Java 的 stopOnPermanentStreamRejection 和 Python 的 stop_on_permanent_stream_rejection 提供与 Go 相同的可选停止行为。 在独立 goroutine 中调用 Start，然后设置启动截止时间并等待注册完成。 滚动部署时，先等待新 Worker 就绪，再停止旧 Worker。注册成功不代表处理器 仍有空闲容量。

TypeScript 提供 waitPullRegistered、pullRegistrationState 和 stopOnPermanentStreamRejection。Rust 提供 wait_pull_registered、 pull_registration_state 和 stop_on_permanent_stream_rejection，就绪语义相同。 TypeScript 等待支持 AbortSignal 或毫秒超时；Rust 可通过丢弃 future 取消等待， 或使用 Tokio 超时。启用停止选项后，TypeScript 的 run 和 Rust 的 start 会在排空后返回永久拒绝错误。

WaitPullRegistered 支持并发调用，也可以在 Start 之前调用。 如果就绪前发生永久拒绝，且该队列没有其他已注册的 stream，它会返回包含 队列名称和底层 gRPC 错误的 PullRegistrationError。只要同一队列仍有其他 已注册的 stream，多余 stream 的拒绝就只会被记录并重试。Start 的启动错误也会传递给等待方。 取消等待会返回 context 错误；就绪前停止 Worker 会返回错误。 所有队列都注册成功过一次后，后续等待始终成功。

这些可运行示例使用 30 秒启动期限等待注册，启动失败时会停止 Worker。 Go SDK 从 1.13.0 开始提供就绪 API；Java、Python、TypeScript 和 Rust SDK 从 1.14.0 开始提供。 将 Go HTTP handler 挂载到就绪探针端点。其他语言可在探针端点调用布尔值 helper： 返回 true 时响应 200，返回 false 时响应 503。

def is_pull_greeting_ready(worker: AsyncWorker) -> bool:
    return worker.pull_registration_state().is_ready

例子: examples/python/dex_examples/production/pull_workers.py

Queues 提供每个队列的 HasRegisteredStream 和 LastPermanentError。首次就绪后，重连、所有权迁移、临时断连和退避 都保持 IsReady 为 true。只有某个队列既没有已注册的 stream， 又存在永久错误时，它才变为 false。重新注册成功会清除该队列的永久错误； 其他队列均正常时，就绪状态恢复。单独调用 Stop 不会清除就绪状态； 关闭时应另行将 Worker 从服务中移除。

StopOnPermanentStreamRejection 默认为 false，保留自动重试行为。 设为 true 后，首次导致其队列没有任何已注册 stream 的永久拒绝会让 Start 返回 PullRegistrationError，即使 Worker 已经就绪。 同一队列仍有其他已注册 stream 时，拒绝不会停止 Worker。Start 排空其他队列时，这些队列仍可能持有 stream。 排空时间由 StopDrainTimeout 限制，超时后请求取消并等待处理器返回。 处理器必须响应取消。退出前始终调用 Stop；Start 返回后仍可安全调用。

路由和可用性

每个拉取模式 Worker 都轮询自己的 app Worker ID 队列，以及注册的每个 Flow 类型队列。 Dex 首先尝试此前服务该 Flow 的 app Worker ID。该 ID 没有任何已连接的 stream 时，例如 Worker 重启或部署后，Dex 会立即改用 Flow 类型队列。否则，Dex 在黏性等待结束后改用 Flow 类型队列，让其他 Worker 接手。黏性路由是尽力而为；持久状态应保存在 Dex 中，不应依赖某个 Worker 的内存。同一个 Dex Server 内的 Flow 类型名称必须唯一。app Worker ID 必须 唯一标识一个 Worker 进程，并在该进程重启后变化。未显式设置时，SDK 会生成 ID。

Dex 在 matching dispatch 等待时间内寻找可用 Worker。如果没有 Worker，Step 会收到 可重试的 unavailable 错误，并消耗一次 Step 重试。RPC 调用方则直接收到 Unavailable 错误。dispatch 默认等待 10 秒，其中有已连接 stream 的黏性队列最多等待 5 秒。等待时间不会超过这次 attempt 的 method deadline。 等待设置见 Pull Worker 分发。

如果拉取模式 Worker 注册了该 Flow 类型，却没有请求的 Step 类型或 RPC 名称， 它会在运行 handler 前拒绝任务。Dex 会立即改用 Flow 类型队列，并在本次分配中 排除该 Worker，即使 Flow 原本黏附于它。重新分配不会消耗 Step attempt。 如果在 dispatch 等待时间内没有可处理任务的 Worker，Step 会按其失败和重试策略 收到注明 Step 或 RPC 名称的错误；RPC 调用方会在最多等待 dispatch 等待时间 （10 秒）后直接收到该错误。此重新分配需要支持它的 Dex Server 版本。面对旧版 Server，Worker 与之前一样返回未注册错误。

任务至少投递一次。如果 Worker 在 Dex 收到任何输出前断开连接，同一个调用可能再次分配。 外部副作用应具备幂等性，例如使用 Flow ID 和 Step execution ID 作为幂等键。

容量和关闭

并发任务上限限制同时运行的调用数。每队列的空闲 stream 数量使 stream 随时可用。 默认值分别为 CPU 核心数和每队列一个空闲 stream。每条 stream 一次只处理 一个调用。达到容量上限时，Worker 仍保留空闲 stream，并拒绝额外分配，让 Dex 尝试 其他 Worker。

SDK	并发任务数	每队列空闲 stream 数	停止时的排空期限	关闭方法
Go	MaxConcurrentTasks	IdleStreamsPerQueue	StopDrainTimeout	Stop
Python	max_concurrent_tasks	idle_streams_per_queue	stop_drain_timeout	stop
Java	maxConcurrentTasks	idleStreamsPerQueue	stopDrainTimeout	stop
TypeScript	maxConcurrentTasks	idleStreamsPerQueue	stopDrainTimeoutMs	close
Rust	max_concurrent_tasks	idle_streams_per_queue	stop_drain_timeout	stop

表中 Java 和 Rust 的选项名称是 builder method。下方公式使用 TypeScript 的两个容量参数名称； 同一上限适用于所有 SDK。

任务完成后，Worker 会正常关闭多余的空闲 stream。每个队列保留至少 idleStreamsPerQueue 条已注册的空闲 stream，清理期间就绪状态保持为 true。 对于注册了 F 个 Flow 类型的 Worker，若近期没有预留容量拒绝，负载下降后的稳态上限为 maxConcurrentTasks + (F + 1) × idleStreamsPerQueue 条 stream。 额外的队列是该 Worker 的 app Worker ID 队列。最近发生预留容量拒绝的队列， 在最后一次拒绝后的 30 秒内可能再保留一条空闲 stream。配置 matching.maxConcurrentTasksCap 时，最多还需为 F + 1 条 stream 留出容量； 默认每个 app Worker ID 最多 256 条 stream。 matching.maxStreams 默认限制节点上最多注册 100,000 条 stream。 突发负载期间的瞬时峰值仍受 matching 容量上限约束。 matching.maxQueueLength、matching.maxFlowTypes 和 matching.maxFlowTypeLength 也限制注册内容，默认值均为 256。

部署期间应正常停止 Worker。关闭方法会停止接收新调用，并在配置的排空期限内等待正在运行的 handler，默认期限为 30 秒。超过期限后，它请求取消，并等待 handler 返回。长时间运行的 handler 应响应取消。Server 要求重新连接时，Worker 也会先完成 正在运行的调用。

为 RPC 预留容量

自 SDK 1.15.0 起，Worker 预留 RPC 容量，使 Step 长时间运行时，同步 RPC 仍能及时响应。 Go 使用 ReservedRPCTasks，Java 和 TypeScript 使用 reservedRpcTasks， Python 和 Rust 使用 reserved_rpc_tasks。未设置时，如果解析后的任务上限至少为二， 默认预留一个槽位；上限为一时则预留零个。使用 CPU 数量默认值时，Step 比以前少一个槽位。 显式设置为零可以关闭预留。上限为一且未显式设置预留值时，Worker 启动会记录一次警告： RPC 可能等待长时间运行的 Step，并建议将上限设为至少二，或显式设置预留值。 Step 占用其余容量时，预留槽位可让 Flow 保持响应。 预留值必须非负，并且小于解析 CPU 数量默认值后的并发任务上限。 无效值会导致 Worker 构造失败。

RPC 任务包括 API 直接调用和事务型 InvokeWorkerRPC 任务。 RPC 和 Step 的处理函数共同计入任务上限。当忙碌处理函数数量达到任务上限减去预留值时， Worker 会以 NO_CAPACITY 拒绝所有 Step 任务，包括 WaitFor 和 Execute。RPC 任务可以使用全部容量。 例如，上限为八、预留值为二时，只有忙碌处理函数少于六个，才能启动新的 Step 任务。

Dex 会尝试其他 Worker，不消耗 Step 尝试次数。如果所有 Worker 都持续达到 Step 容量上限， Step 最多等待到分发等待时间结束，然后收到 NO_WORKER_AVAILABLE、消耗一次尝试并遵循其重试策略， 与所有 Worker 都已满载时相同。 每个队列仍保留可用的空闲流。队列因预留 RPC 容量拒绝 Step 后， 会多保留一条空闲 stream，直到最后一次拒绝后的 30 秒。其他队列维持配置的空闲数量。 所有任务槽位都忙碌时，Worker 不会新建这条额外 stream。Step 积压较多的 Worker 应将 idleStreamsPerQueue 至少设为 2。预留容量导致的拒绝不会关闭流， 也不会改变注册就绪状态。 每次此类拒绝都会生成包含 Flow 类型的调试日志，并增加 Worker 的累计预留拒绝计数： Go 为 PullReservedRPCDeclines()，Java 为 getPullReservedRpcDeclines()， TypeScript 为 pullReservedRpcDeclines，Python 为 pull_reserved_rpc_declines， Rust 为 pull_reserved_rpc_declines()。满载拒绝和未知名称不会增加此计数。 该计数统计拒绝事件；Dex 重试同一次 Step 分发时，可能多次拒绝它，因此应观察计数的变化速率。 计数从不重置，关闭后仍可读取。日志配置请参阅各 SDK 的 README。

上面的启动示例将任务上限设为八，并显式为 RPC 预留一个槽位，与默认值一致。 下面的诊断函数返回注册就绪状态和累计预留拒绝计数，供监控使用。 计数增加表示 Step 容量紧张，不会使 Worker 变为未就绪。

@dataclass(frozen=True)
class PullGreetingDiagnostics:
    is_ready: bool
    reserved_rpc_declines: int


def get_pull_greeting_diagnostics(worker: AsyncWorker) -> PullGreetingDiagnostics:
    return PullGreetingDiagnostics(
        is_ready=worker.pull_registration_state().is_ready,
        reserved_rpc_declines=worker.pull_reserved_rpc_declines,
    )

例子: examples/python/dex_examples/production/pull_workers.py

缩容到零

server-zd 没有 Worker 唤醒 hook。Worker 可能缩容到零时，应按 Worker 启动时间设置 dispatch 等待和 Step 重试，让 Step 重试可以等待 Worker 启动。

Server 和网络配置

Worker 连接到 server 的 FlowService 端口，即 api.listen。经过负载均衡器或代理时请保持 双向 HTTP/2 streaming。任务运行期间，SDK 每隔 20 秒向上游发送 Pong frame。升级 SDK 即可获得此行为。Dex 按 matching.pingInterval 向空闲 stream 发送 Ping frame。将负载均衡器 和代理的 idle timeout 设为同时大于 20 秒和 matching.pingInterval，并为网络延迟留出余量。旧版 SDK 需要将 idle timeout 设为大于最长任务时长。matching 限制和等待设置见 Pull Worker 分发。

Go、Python、Java、TypeScript 和 Rust SDK，以及 dexcli 和 Dex Web，均可发送所需的 bearer credential。 Worker 和 Client 各有 credential 选项。 参阅身份认证设置。