INTEGRITY Dokumentace

← Dynamic Workers / dynamic-workers / usage

Dynamic Workflows

Uvnitř Dynamic Workeru můžete spustit Workflow, díky čemuž kód načtený za běhu získá odolné provádění (durable execution). Každý krok Workflow přežije selhání, může být na hodiny nebo dny uspán, může čekat na externí události a poté pokračuje přesně tam, kde skončil, a to i v případě, že je isolate mezi jednotlivými kroky recyklován.

Protože se Dynamic Workery vytvářejí podle potřeby, nemusíte každý Workflow předem registrovat ani je spravovat jednotlivě. Kód stačí načíst, až je potřeba, a o perzistenci a opakování se na pozadí postará Workflows engine. Funguje to stejně dobře pro jednorázová spuštění jako pro dlouho běžící, vícekrokové procesy.

Můžete například vytvářet:

@cloudflare/dynamic-workflows knihovna propojuje váš Worker Loader s enginem Workflows, takže každý Dynamic Worker získá plnou sílu trvalých kroků (step.do(), step.sleep(), step.waitForEvent()) aniž byste museli sami budovat propojení.

V tomto průvodci použijete @cloudflare/dynamic-workflows knihovnu k nastavení Worker Loaderu, napsání Dynamic Workeru s trvalými kroky a spuštění instance Workflow.

Pochopte model

Toto nastavení má tři části:

Architektura

Takto spolu fungují:

Knihovna poskytuje dvě funkce, které se starají o propojení mezi Worker Loaderem a enginem Workflows, takže nemusíte požadavky ručně tagovat, parsovat payloady ani psát vlastní WorkflowEntrypoint podtřída.

Instalace knihovny

Knihovna se stará o propojení mezi Worker Loaderem a enginem Workflows, takže nemusíte požadavky ručně tagovat, parsovat payloady ani psát vlastní WorkflowEntrypoint podtřída.

npm i @cloudflare/dynamic-workflows

Konfigurace vašeho Worker Loaderu

Váš Worker Loader potřebuje dva vazby:

{
  "$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"

Vytvoření Worker Loaderu

Worker Loader je místo, kde propojíte Dynamic Workers s enginem Workflows. V tomto souboru definujete:

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

Takto to probíhá, když dorazí požadavek:

  1. fetch handler čte ID tenanta z hlavičky požadavku.
  2. loadTenant volání env.LOADER.get() k načtení (nebo opětovnému použití) Dynamic Workeru pro daného tenanta. Dynamic Worker obdrží WORKFLOWS: wrapWorkflowBinding({ tenantId }) jako binding, který vypadá a chová se jako běžný binding pro Workflow.
  3. Požadavek je přeposlán Dynamic Workeru, konkrétně jeho fetch handler, který teď může volat env.WORKFLOWS.create() ke spuštění instance Workflow.

Když daná instance Workflow později potřebuje spustit krok, například po step.sleep() nebo když si ho převezme nový izolát, engine Workflows zavolá run() na DynamicWorkflow třídy. Knihovna čte tenantId zpět z metadat uložených u instance a zavolá callback, který jste předali createDynamicWorkflowEntrypoint. Tento callback načte Dynamic Worker pro daného tenanta a vrátí jeho TenantWorkflow třídy, aby engine mohl provést další krok v původním kódu.

Napište Dynamic Worker

Dynamic Worker je kód, který píše váš uživatel, a nemusí vůbec nic vědět o vrstvě routování. Jde o standardní Workflow, které využívá step.do(), step.sleep(), a step.waitForEvent() jako obvykle, z jeho pohledu env.WORKFLOWS je běžný binding pro 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 });
	},
};

Nadále platí běžné chování Workflows. ID Workflow, .status(), .pause(), opakování, hibernace a durable kroky touto architekturou nejsou ovlivněny. Knihovna pouze doplňuje směrování mezi Worker Loaderem a Dynamic Workerem.

Spusťte dynamický workflow

Odešlete POST požadavek na Worker Loader s hlavičkou obsahující ID tenanta a JSON payloadem. Worker Loader načte odpovídající Dynamic Worker, který zavolá env.WORKFLOWS.create() a vrátí ID nové instance.

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

Kontrola stavu workflow

Pomocí ID instance vráceného z předchozího požadavku zkontrolujte stav Workflow. Další informace o API pro zjišťování stavu najdete v Referenční dokumentace Workers API.

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