动态工作流程
运行不同的 工作流 由单一调度工人提供的每名租户执行.
工作流通常与一个静态相接 class_name 在部署时。 如果你在顶上建造一个平台 动态工人 每一个客户都用自己的代码 问题就在于你只得到一个 WorkflowEntrypoint但你需要它 引导进入N租户工人.
这个图书馆就是胶水 调度员给每个租户一个 包起来 Workflow 绑定,悄悄标记每个 create() 有租户身份; 调度员的单人 WorkflowEntrypoint 在运行时读取标记并向前 run(event, step) 进入正确的房客.
┌──────────── Dispatcher Worker ────────────┐ ┌──── Tenant's dynamic worker ────┐
│ │ │ │
│ env.WORKFLOWS (real Workflow binding) │ │ env.WORKFLOWS (wrapped!) │
│ │ │ │
│ │ │ ── env.WORKFLOWS.create({ │
│ │ │ params: { ... } │
│ │ │ }) │
│ │ │ │ │
│ ┌──── wrapWorkflowBinding ──────────┐ │ tags │ │ │
│ │ injects { __dispatcherMetadata, │◀───┼──────────┼─────────┘ │
│ │ params } into create(...) │ │ │ │
│ └────────────────┬──────────────────┘ │ └─────────────────────────────────┘
│ │ │
│ ▼ │
│ Workflows engine │
│ │ │
│ │ run(event, step) │
│ ▼ │
│ ┌── createDynamicWorkflowEntrypoint ──┐ │ ┌──── Tenant's dynamic worker ────┐
│ │ pulls metadata off event.payload │ │ │ │
│ │ loadRunner({metadata, env, ctx}) │──┼─────────▶│ class TenantWorkflow │
│ │ forwards run(innerEvent, step) │ │ │ extends WorkflowEntrypoint { │
│ └─────────────────────────────────────┘ │ │ run(event, step) { … } │
└───────────────────────────────────────────┘ │ } │
└─────────────────────────────────┘
你负责 怎么样 租户代码被装入了——工人装入器,服务装入器,无论什么——这个图书馆只携带了两个半个舞会之间的租户代号.
安装
npm install dynamic-workflows
用这个
调度员需要三件事:再出口 DynamicWorkflowBinding,将每个租户一份包裹装订,并登记 WorkflowEntrypoint 那会装入租户代码.
// dispatcher/src/index.ts
import {
createDynamicWorkflowEntrypoint,
DynamicWorkflowBinding,
wrapWorkflowBinding,
type WorkflowRunner,
} from 'dynamic-workflows';
// Required: re-exporting puts the class on `cloudflare:workers` exports,
// which is how `wrapWorkflowBinding` builds per-tenant RPC stubs.
export { DynamicWorkflowBinding };
interface Env {
WORKFLOWS: Workflow;
LOADER: WorkerLoader;
}
function loadTenant(env: Env, tenantId: string) {
return env.LOADER.get(tenantId, async () => ({
compatibilityDate: '2026-01-01',
mainModule: 'index.js',
modules: { 'index.js': await fetchTenantCode(tenantId) },
env: {
// The tenant uses this exactly like a real Workflow binding;
// every create() is tagged with { tenantId } automatically.
WORKFLOWS: wrapWorkflowBinding({ tenantId }),
},
globalOutbound: null,
}));
}
// Register this as `class_name` in wrangler.jsonc.
export const DynamicWorkflow = createDynamicWorkflowEntrypoint<Env>(
async ({ env, metadata }) => {
const stub = loadTenant(env, metadata.tenantId as string);
return stub.getEntrypoint('TenantWorkflow') as unknown as WorkflowRunner;
}
);
export default {
fetch(request: Request, env: Env) {
const tenantId = request.headers.get('x-tenant-id')!;
return loadTenant(env, tenantId).getEntrypoint().fetch(request);
},
};
// dispatcher/wrangler.jsonc
{
"name": "my-dispatcher",
"main": "src/index.ts",
"compatibility_date": "2026-01-01",
"worker_loaders": [{ "binding": "LOADER" }],
"workflows": [
{
"name": "dynamic-workflow",
"binding": "WORKFLOWS",
"class_name": "DynamicWorkflow"
}
]
}
租户的代码是普通的工作流——他们不知道自己被派遣:
// tenant code, loaded at runtime by the dispatcher
import { WorkflowEntrypoint } from 'cloudflare:workers';
export class TenantWorkflow extends WorkflowEntrypoint {
async run(event, step) {
return step.do('greet', async () => `Hello, ${event.payload.name}!`);
}
}
export default {
async fetch(request, env) {
const instance = await env.WORKFLOWS.create({ params: await request.json() });
// `instance` is an RPC stub — `.id` is an RpcPromise, so await it.
return Response.json({ id: await instance.id });
},
};
就是这样。 工作流的ID,状态,回溯,休眠——其他一切按正常情况运作.
API 密码
wrapWorkflowBinding(metadata, options?): Workflow
回返a Workflow- 形状为RPC的支架。 每个 create() 编号 : createBatch() 通过它调用标记实例的参数 metadata将它传递给动态工人作为约束.
metadata 是一个 JSON 可串行对象—— 通常 { tenantId },但路由提示,区域,计划层次等全部工作. options.bindingName 默认为 'WORKFLOWS'; 如果您的调度器绑定有不同的名称, 请覆盖它 .
如果你忘记再出口就扔出去 DynamicWorkflowBinding。 。 。 .
createDynamicWorkflowEntrypoint<Env>(loadRunner): typeof WorkflowEntrypoint
回返a WorkflowEntrypoint 子类。 登记为 class_name 您的电话 [[workflows]] 装订.
loadRunner({ metadata, env, ctx }) 每运行一次 run() 打电话,必须归还的东西 run(event, step) 方法。 在实际中你回来 stub.getEntrypoint('YourWorkflowClass') 从工人装货员的支架.
DynamicWorkflowBinding
这个 WorkerEntrypoint 课后 wrapWorkflowBindings. 从调度员主模块中再导出 ——云花花相接物 cloudflare:workers 您的顶级出口,以及 wrapWorkflowBinding 看那边的课.
dispatchWorkflow(context, event, step, loadRunner)
低级原始人 createDynamicWorkflowEntrypoint 继续发展。 需要子类时直接使用 WorkflowEntrypoint 自己(例如包装) run 在自定义日志中).
MissingDispatcherMetadataError
丢弃自 run() 当事件有效载荷不是一个调度器信封——即工作流程是针对原始装订而不是被包裹的装订而创建的.
事情要知道这一点
- 捆绑会穿过RPC,所以它们必须是RPC的活塞. 这就是为什么
DynamicWorkflowBinding是一个WorkerEntrypoint: 一个平原{ create, get }对象不是结构化可克隆的,而是真实的Workflow绑定也不能被序列化. - 持续的工作流量
event.payload。 。 。 . 这个有效载荷是发送器的信封 包括元数据 不要在元数据中隐藏秘密,并且把它当作路由提示,而不是授权——租户代码可以通过instance.status()。 。 。 . WorkflowInstance是租户一方的RPC根茎. 财产读取返回RpcPromise这样await instance.id方法调用status(), (中文(简体) ),pause(),.)正常工作.- 信封外形为内部.
wrapWorkflowBinding财务报告和审定财务报表createDynamicWorkflowEntrypoint保证彼此相容,不要自己分析持续有效载荷.
许可证
麻省理工学院