1import * as Cloudflare from "@/Cloudflare";
2import * as Effect from "effect/Effect";
3
4export const RollbackResults = Cloudflare.R2.Bucket("WorkflowRollbackResults", {
5 forceDestroy: true,
6});
7
8export const rollbackConfigs = {
9 undefined: { retries: undefined, timeout: undefined },
10 both: { retries: { limit: 1, delay: "1 second" }, timeout: "30 seconds" },
11 "timeout-only": { timeout: "30 seconds", retries: undefined },
12 "retries-only": {
13 retries: { limit: 1, delay: "1 second" },
14 timeout: undefined,
15 },
16} satisfies Record<string, Cloudflare.Workflows.WorkflowStepConfig>;
17
18export const failureScenarios = [
19 "timeout-zero",
20 "rollback-timeout-zero",
21 "retry-exhaustion",
22 "rollback-retry-exhaustion",
23] as const;
24
25/** Workflow fixture exercised against local workerd and live Cloudflare. */
26export default class LocalTestWorkflow extends Cloudflare.Workflow<LocalTestWorkflow>()(
27 "LocalTestWorkflow",
28 Effect.gen(function* () {
29 const bucket = yield* RollbackResults;
30 const results = yield* Cloudflare.R2.ReadWriteBucket(bucket);
31
32 return Effect.fn(function* (input: {
33 value: string;
34 ready?: boolean;
35 rollback?: boolean;
36 scenario?: (typeof failureScenarios)[number];
37 }) {
38 if (input.ready) return { ready: true };
39 const event = yield* Cloudflare.Workflows.WorkflowEvent;
40 const triggerRollback = Cloudflare.Workflows.task(
41 "fail-after-reservation",
42 Effect.fail(new Error("rollback requested")),
43 { retries: { limit: 0, delay: "1 second" }, timeout: "30 seconds" },
44 );
45
46 if (
47 input.scenario === "timeout-zero" ||
48 input.scenario === "rollback-timeout-zero"
49 ) {
50 const protectedEffect = results
51 .put(
52 `${event.instanceId}/protected`,
53 JSON.stringify({ executed: true }),
54 )
55 .pipe(Effect.asVoid, Effect.orDie);
56 yield* Cloudflare.Workflows.task(
57 "zero-timeout",
58 input.scenario === "timeout-zero" ? protectedEffect : Effect.void,
59 input.scenario === "timeout-zero"
60 ? { timeout: 0 }
61 : {
62 rollback: () => protectedEffect,
63 rollbackConfig: { timeout: 0 },
64 },
65 );
66 yield* triggerRollback;
67 }
68
69 if (input.scenario === "retry-exhaustion") {
70 yield* Cloudflare.Workflows.task(
71 "exhaust-retries",
72 Effect.gen(function* () {
73 const { attempt } = yield* Cloudflare.Workflows.WorkflowStepContext;
74 yield* results
75 .put(`${event.instanceId}/attempts`, JSON.stringify({ attempt }))
76 .pipe(Effect.orDie);
77 return yield* Effect.fail(new Error("retry budget exhausted"));
78 }),
79 { retries: { limit: 1, delay: "1 second", backoff: "constant" } },
80 );
81 }
82
83 if (input.scenario === "rollback-retry-exhaustion") {
84 yield* Cloudflare.Workflows.task(
85 "reserve-before-exhaustion",
86 Effect.succeed({ value: input.value }),
87 {
88 rollbackConfig: {
89 retries: { limit: 1, delay: "1 second", backoff: "constant" },
90 timeout: undefined,
91 },
92 rollback: () =>
93 Effect.gen(function* () {
94 const key = `${event.instanceId}/attempts`;
95 const object = yield* results.get(key);
96 const previous = object
97 ? yield* object.json<{ attempt: number }>()
98 : { attempt: 0 };
99 yield* results.put(
100 key,
101 JSON.stringify({ attempt: previous.attempt + 1 }),
102 );
103 return yield* Effect.fail(
104 new Error("rollback budget exhausted"),
105 );
106 }),
107 },
108 );
109 yield* triggerRollback;
110 }
111
112 if (input.rollback) {
113 for (const [name, rollbackConfig] of Object.entries(rollbackConfigs)) {
114 yield* Cloudflare.Workflows.task(
115 name,
116 Effect.succeed({ value: input.value, step: name }),
117 {
118 rollbackConfig,
119 rollback: ({ output, error }) =>
120 results
121 .put(
122 `${event.instanceId}/${name}`,
123 JSON.stringify({ output, error: error.message }),
124 )
125 .pipe(Effect.asVoid, Effect.orDie),
126 },
127 );
128 }
129 yield* triggerRollback;
130 }
131
132 const greeted = yield* Cloudflare.Workflows.task(
133 "greet",
134 Effect.succeed({ text: `Hello, ${input.value}!` }),
135 );
136
137 const retried = yield* Cloudflare.Workflows.task(
138 "retry-only",
139 Effect.gen(function* () {
140 const { attempt, config } =
141 yield* Cloudflare.Workflows.WorkflowStepContext;
142 if (attempt === 1) return yield* Effect.fail(new Error("retry once"));
143 return { attempt, config };
144 }),
145 { retries: { limit: 2, delay: "1 second", backoff: "constant" } },
146 );
147
148 const bounded = yield* Cloudflare.Workflows.task(
149 "timeout-only",
150 Effect.gen(function* () {
151 const { config } = yield* Cloudflare.Workflows.WorkflowStepContext;
152 return config;
153 }),
154 { timeout: "30 seconds" },
155 );
156
157 const defaults = yield* Cloudflare.Workflows.task(
158 "undefined-config",
159 Effect.succeed({ ok: true }),
160 { retries: undefined, timeout: undefined },
161 );
162
163 yield* Cloudflare.Workflows.sleep("cooldown", "1 second");
164
165 return {
166 greeting: greeted.text,
167 retryAttempt: retried.attempt,
168 retryConfig: retried.config,
169 timeoutConfig: bounded,
170 defaultsOk: defaults.ok,
171 workflowName: event.workflowName,
172 instanceId: event.instanceId,
173 };
174 });
175 }),
176) {}