INTEGRITY Документация

← Dynamic Workers / dynamic-workers / usage

Динамические Workflows

Вы можете запустить Workflow внутри Dynamic Worker, чтобы обеспечить надёжное выполнение кода, загружаемого во время выполнения. Каждый шаг Workflow переживает сбои, может приостанавливаться на часы или дни, может дожидаться внешних событий и возобновляется именно с того места, где остановился, даже если изолят между шагами был пересоздан.

Поскольку Dynamic Worker создаются по запросу, вам не нужно заранее регистрировать каждый Workflow или управлять ими по отдельности. Загружайте код тогда, когда он нужен, а движок Workflows берет на себя сохранение состояния и повторные попытки. Это одинаково хорошо работает как для разовых выполнений, так и для длительных многошаговых процессов.

Например, вы можете создавать:

@cloudflare/dynamic-workflows связывает ваш Worker Loader с движком Workflows, чтобы каждый Dynamic Worker получал все возможности устойчивых шагов (step.do(), step.sleep(), step.waitForEvent()) без необходимости самостоятельно реализовывать всю инфраструктуру для этого.

В этом руководстве вы будете использовать @cloudflare/dynamic-workflows для настройки Worker Loader, написания Dynamic Worker с устойчивыми шагами и запуска экземпляра Workflow.

Как устроена модель

Эта настройка состоит из трёх частей:

Архитектура

Вот как они работают вместе:

Библиотека предоставляет две функции, которые берут на себя связку между Worker Loader и движком Workflows, поэтому вам не нужно вручную помечать запросы, разбирать полезную нагрузку или писать собственный WorkflowEntrypoint подкласс.

Установите библиотеку

Библиотека берет на себя связку между Worker Loader и движком Workflows, поэтому вам не нужно вручную помечать запросы, разбирать полезную нагрузку или писать собственный WorkflowEntrypoint подкласс.

npm i @cloudflare/dynamic-workflows

Настройка Worker Loader

Вашему Worker Loader нужны два привязки:

{
  "$schema": "./node_modules/wrangler/config-schema.json",
  "name": "my-worker-loader",
  "main": "src/index.ts",
  // Set this to today's date
  "compatibility_date": "2026-08-28",
  "worker_loaders": [
    {
      "binding": "LOADER"
    }
  ],
  "workflows": [
    {
      "name": "dynamic-workflow",
      "binding": "WORKFLOWS",
      "class_name": "DynamicWorkflow"
    }
  ]
}
name = "my-worker-loader"
main = "src/index.ts"
# Set this to today's date
compatibility_date = "2026-08-28"

[[worker_loaders]]
binding = "LOADER"

[[workflows]]
name = "dynamic-workflow"
binding = "WORKFLOWS"
class_name = "DynamicWorkflow"

Создание Worker Loader

Worker Loader представляет собой место, где вы подключаете Dynamic Workers к движку Workflows. В этом файле вы определяете:

import {
	createDynamicWorkflowEntrypoint,
	DynamicWorkflowBinding,
	wrapWorkflowBinding,
} from "@cloudflare/dynamic-workflows";

// Required: re-exporting puts the class on cloudflare:workers exports,
// which is how wrapWorkflowBinding builds per-tenant RPC stubs.
export { DynamicWorkflowBinding };

function loadTenant(env, tenantId) {
	return env.LOADER.get(tenantId, async () => ({
		compatibilityDate: "2026-01-01",
		mainModule: "index.js",
		modules: { "index.js": await fetchTenantCode(tenantId) },
		// The Dynamic Worker uses this exactly like a real Workflow binding;
		// every create() is tagged with { tenantId } automatically.
		env: { WORKFLOWS: wrapWorkflowBinding({ tenantId }) },
	}));
}

// The entrypoint name must match `class_name` in the workflows binding of your Wrangler config file.
export const DynamicWorkflow = createDynamicWorkflowEntrypoint(
	async ({ env, metadata }) => {
		const stub = loadTenant(env, metadata.tenantId);
		return stub.getEntrypoint("TenantWorkflow");
	},
);

export default {
	fetch(request, env) {
		const tenantId = request.headers.get("x-tenant-id");
		return loadTenant(env, tenantId).getEntrypoint().fetch(request);
	},
};
import {
	createDynamicWorkflowEntrypoint,
	DynamicWorkflowBinding,
	wrapWorkflowBinding,
	type WorkflowRunner,
} from "@cloudflare/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) },
		// The Dynamic Worker uses this exactly like a real Workflow binding;
		// every create() is tagged with { tenantId } automatically.
		env: { WORKFLOWS: wrapWorkflowBinding({ tenantId }) },
	}));
}

// The entrypoint name must match `class_name` in the workflows binding of your Wrangler config file.
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);
	},
};

Вот что происходит при поступлении запроса:

  1. fetch обработчик считывает ID арендатора из заголовка запроса.
  2. loadTenant вызывает env.LOADER.get() чтобы загрузить (или переиспользовать) Dynamic Worker для этого арендатора. Dynamic Worker получает WORKFLOWS: wrapWorkflowBinding({ tenantId }) как привязку, которая выглядит и ведет себя как обычная привязка Workflow.
  3. Запрос перенаправляется обработчику Dynamic Worker fetch обработчик, который теперь может вызывать env.WORKFLOWS.create() для запуска экземпляра Workflow.

Когда этому экземпляру Workflow позже понадобится выполнить шаг, например после step.sleep() или когда его подхватывает новый isolate, движок Workflows вызывает run() на DynamicWorkflow класс. Библиотека считывает tenantId обратно из метаданных, сохраненных в экземпляре, и вызывает колбэк, который вы передали в createDynamicWorkflowEntrypoint. Этот колбэк загружает Dynamic Worker для этого арендатора и возвращает его TenantWorkflow класс, чтобы движок мог выполнить следующий шаг в исходном коде.

Написание Dynamic Worker

Dynamic Worker представляет собой код, который пишет ваш пользователь: ему не нужно ничего знать о слое маршрутизации. Это обычный Workflow, использующий step.do(), step.sleep(), а также step.waitForEvent() как обычно: с его точки зрения, env.WORKFLOWS это обычная привязка Workflow.

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 });
	},
};
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 });
	},
};

Обычное поведение Workflows по-прежнему применяется. Идентификаторы Workflow, .status(), .pause(), повторные попытки, гибернация и устойчивые шаги не зависят от этой архитектуры. Библиотека лишь добавляет маршрутизацию между Worker Loader и Dynamic Worker.

Запустить динамический workflow

Отправьте POST запрос к Worker Loader с заголовком ID арендатора и телом JSON. Worker Loader загружает соответствующий Dynamic Worker, который вызывает env.WORKFLOWS.create() и возвращает идентификатор нового экземпляра.

curl -X POST http://localhost:8787/ \
  -H "x-tenant-id: tenant-42" \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice"}'

Проверка статуса workflow

Используйте идентификатор экземпляра, полученный в ответ на предыдущий запрос, чтобы проверить статус Workflow. Дополнительную информацию об API статуса см. в Справочник Workers API.

curl "http://localhost:8787/api/status?instanceId=YOUR_INSTANCE_ID"