1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Alchemy from "@/index.ts";
3import * as Test from "@/Test/Alchemy.ts";
4import { describe, expect } from "alchemy-test";
5import * as Effect from "effect/Effect";
6import * as Schedule from "effect/Schedule";
7import * as HttpClientRequest from "effect/http/HttpClientRequest";
8import { requestWorker } from "../Utils/WorkerRequest.ts";
9import LifecycleWorker, {
10 type Scenario,
11} from "./fixtures/workflow-lifecycle/worker.ts";
12
13interface Status {
14 status: string;
15 output?: string[];
16 error?: { name?: string; message?: string } | null;
17 rollback?: { outcome: string } | null;
18 entries: string[];
19}
20
21const request = Effect.fn(function* (
22 url: string,
23 method: "GET" | "POST" = "GET",
24) {
25 const response = yield* requestWorker(
26 method === "GET" ? HttpClientRequest.get(url) : HttpClientRequest.post(url),
27 );
28 const body = yield* response.text;
29 if (response.status !== 200) {
30 return yield* Effect.fail(
31 new Error(`${method} ${url}: ${response.status}: ${body}`),
32 );
33 }
34 return body;
35});
36
37const waitForReady = Effect.fn(function* (url: string) {
38 const ready = yield* Effect.gen(function* () {
39 const response = yield* requestWorker(HttpClientRequest.get(url), {
40 retryDelay: "3 seconds",
41 });
42 const body = yield* response.text;
43 if (
44 response.status === 503 &&
45 response.headers["x-workflow-lifecycle-readiness"] ===
46 "journal-rpc-not-ready" &&
47 body === "LifecycleJournal.entries not ready"
48 ) {
49 yield* Effect.logInfo(
50 `Workflow readiness: native journal RPC not ready at ${url}`,
51 );
52 return false;
53 }
54 if (response.status !== 200) {
55 return yield* Effect.fail(
56 new Error(`GET ${url}: ${response.status}: ${body}`),
57 );
58 }
59 if (body === "Alchemy worker is being deployed...") return false;
60 expect(body).toBe("ready");
61 return true;
62 }).pipe(
63 Effect.repeat({
64 schedule: Schedule.spaced("1 second"),
65 times: 8,
66 until: (ready) => ready,
67 }),
68 Effect.timeout("45 seconds"),
69 );
70 expect(ready, "Journal entries method did not propagate").toBe(true);
71});
72
73const probeWorkflow = Effect.fn(function* (
74 url: string,
75 className = "LifecycleWorkflow",
76) {
77 const ready = yield* Effect.gen(function* () {
78 const started = yield* request(`${url}/probe`, "POST");
79 const { id } = yield* Effect.try(
80 () => JSON.parse(started) as { id: string },
81 );
82 const status = yield* request(`${url}/probe/${id}`).pipe(
83 Effect.flatMap((body) =>
84 Effect.try(() => JSON.parse(body) as Omit<Status, "entries">),
85 ),
86 Effect.repeat({
87 schedule: Schedule.spaced("1 second"),
88 times: 10,
89 until: (status) =>
90 status.status === "complete" || status.status === "errored",
91 }),
92 );
93 yield* Effect.logInfo(
94 `Workflow readiness probe ${id}: ${JSON.stringify(status)}`,
95 );
96 // Only a probe may be recreated when Workflow execution still sees the stub.
97 if (
98 status.status === "errored" &&
99 status.error?.name === "TypeError" &&
100 status.error.message ===
101 `The entrypoint name ${className} was not found in this worker. Ensure the worker exports an entrypoint with that name.`
102 ) {
103 return false;
104 }
105 expect(status, JSON.stringify(status)).toMatchObject({
106 status: "complete",
107 output: ["workflow-ready"],
108 });
109 return true;
110 }).pipe(
111 Effect.repeat({
112 schedule: Schedule.spaced("1 second"),
113 times: 8,
114 until: (ready) => ready,
115 }),
116 Effect.timeout("45 seconds"),
117 );
118 expect(ready, "Workflow entrypoint did not propagate").toBe(true);
119});
120
121const cases: Array<{ scenario: Scenario; entries: string[] }> = [
122 ...(
123 [
124 "unsupported-function",
125 "unsupported-symbol",
126 "unsupported-cycle",
127 "unsupported-size",
128 "unsupported-array",
129 "unsupported-accessor",
130 "unsupported-tag-accessor",
131 "unsupported-alias",
132 ] as const
133 ).map((scenario) => ({
134 scenario,
135 entries: ["open:1:captured:true", "close:1"],
136 })),
137 {
138 scenario: "success",
139 entries: ["open:1:captured:true", "close:1", "after-task"],
140 },
141 {
142 scenario: "retry",
143 entries: [
144 "open:1:captured:true",
145 "close:1",
146 "open:2:captured:true",
147 "close:2",
148 "after-task",
149 ],
150 },
151 {
152 scenario: "exhaustion",
153 entries: [
154 "open:1:captured:true",
155 "close:1",
156 "open:2:captured:true",
157 "close:2",
158 "caught:application:2:true",
159 "after-task",
160 ],
161 },
162 {
163 scenario: "uncaught",
164 entries: [
165 "open:1:captured:true",
166 "close:1",
167 "open:2:captured:true",
168 "close:2",
169 ],
170 },
171 {
172 scenario: "die",
173 entries: ["open:1:captured:true", "close:1"],
174 },
175 {
176 scenario: "orDie",
177 entries: ["open:1:captured:true", "close:1"],
178 },
179 {
180 scenario: "interrupt",
181 entries: ["open:1:captured:true", "close:1", "joined", "after-task"],
182 },
183 {
184 scenario: "interrupt-retry",
185 entries: ["open:1:captured:true", "close:1", "joined", "after-task"],
186 },
187 {
188 scenario: "rollback-retry",
189 entries: [
190 "open:1:captured:true",
191 "close:1",
192 "after-task",
193 "rollback-open:captured:true:1",
194 "rollback-body:1",
195 "rollback-close:1",
196 "rollback-open:captured:true:2",
197 "rollback-body:2",
198 "rollback-close:2",
199 ],
200 },
201 ...(["rollback-die", "rollback-orDie"] as const).map((scenario) => ({
202 scenario,
203 entries: [
204 "open:1:captured:true",
205 "close:1",
206 "after-task",
207 "rollback-open:captured:true:1",
208 "rollback-body:1",
209 "rollback-close:1",
210 ],
211 })),
212 {
213 scenario: "rollback",
214 entries: [
215 "open:1:captured:true",
216 "close:1",
217 "after-task",
218 "rollback-open:captured:true",
219 "rollback-body",
220 "rollback-close",
221 ],
222 },
223];
224
225describe.concurrent.each([
226 { dev: true, stage: "workflow-lifecycle-local" },
227 { dev: false, stage: "workflow-lifecycle-live" },
228])(
229 "Workflow lifecycle (dev: $dev)",
230 ({ dev, stage }) => {
231 const state = dev ? Alchemy.inMemoryState() : Cloudflare.state();
232 const Stack = Alchemy.Stack(
233 "WorkflowLifecycleStack",
234 { providers: Cloudflare.providers(), state },
235 Effect.gen(function* () {
236 const worker = yield* LifecycleWorker;
237 return { url: worker.url.as<string>() };
238 }),
239 );
240 const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
241 providers: Cloudflare.providers(),
242 state,
243 dev,
244 stage,
245 });
246 const stack = beforeAll(
247 Effect.gen(function* () {
248 yield* destroy(Stack);
249 const output = yield* deploy(Stack);
250 yield* waitForReady(`${output.url}/ready`);
251 yield* probeWorkflow(output.url);
252 return output;
253 }),
254 { timeout: 120_000 },
255 );
256 afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack), {
257 timeout: 30_000,
258 });
259
260 for (const [kind, message] of [
261 ["missing", 'The RPC receiver does not implement the method "entries".'],
262 ["internal", "internal error; reference = application"],
263 ]) {
264 test(
265 `does not mask an application ${kind} failure as readiness`,
266 Effect.gen(function* () {
267 const { url } = yield* stack;
268 const failure = yield* waitForReady(
269 `${url}/ready?application-error=${kind}`,
270 ).pipe(Effect.flip);
271 expect(failure.message).toContain(": 500:");
272 expect(failure.message).toContain(message);
273 expect(failure.message).not.toContain(
274 "LifecycleJournal.entries not ready",
275 );
276 }),
277 { tags: [...(dev ? ["local"] : ["live"])], timeout: 60_000 },
278 );
279 }
280
281 test(
282 "repeated starts keep the same instance and execute the scenario once",
283 Effect.gen(function* () {
284 const { url } = yield* stack;
285 const startUrl = `${url}/start/success?id=repeated-start`;
286 const first = yield* request(startUrl, "POST");
287 expect(yield* request(startUrl, "POST")).toBe(first);
288 const { id } = yield* Effect.try(
289 () => JSON.parse(first) as { id: string },
290 );
291 expect(id).toBe("repeated-start");
292 const status = yield* request(`${url}/status/${id}`).pipe(
293 Effect.flatMap((body) =>
294 Effect.try(() => JSON.parse(body) as Status),
295 ),
296 Effect.repeat({
297 schedule: Schedule.spaced("1 second"),
298 times: 10,
299 until: (status) => status.status === "complete",
300 }),
301 );
302 expect(status).toMatchObject({
303 status: "complete",
304 entries: ["open:1:captured:true", "close:1", "after-task"],
305 });
306 }),
307 { tags: [...(dev ? ["local"] : ["live"])], timeout: 60_000 },
308 );
309
310 test(
311 "preserves native validation failures when starting an instance",
312 Effect.gen(function* () {
313 const { url } = yield* stack;
314 const response = yield* requestWorker(
315 HttpClientRequest.post(`${url}/start/success?id=invalid%2Fid`),
316 );
317 expect(response.status).toBe(500);
318 const body = yield* response.text;
319 expect(body).not.toContain("WorkflowControlUnavailable");
320 expect(body.toLowerCase()).toContain("invalid");
321 }),
322 { tags: [...(dev ? ["local"] : ["live"])], timeout: 60_000 },
323 );
324
325 for (const scenario of [
326 "replay",
327 "replay-inherited",
328 "replay-terminal-die",
329 "replay-terminal-orDie",
330 ] as const) {
331 test(
332 `preserves ${scenario} error semantics after checkpoint replay`,
333 Effect.gen(function* () {
334 const { url } = yield* stack;
335 const terminal = scenario.startsWith("replay-terminal");
336 const caught = terminal ? "caught:terminal" : "caught:application:2";
337 const { id } = yield* request(
338 `${url}/start/${scenario}?stage=${encodeURIComponent(stage)}`,
339 "POST",
340 ).pipe(
341 Effect.flatMap((body) =>
342 Effect.try(() => JSON.parse(body) as { id: string }),
343 ),
344 );
345 const recovered = yield* request(`${url}/journal/${id}`).pipe(
346 Effect.flatMap((body) =>
347 Effect.try(() => JSON.parse(body) as string[]),
348 ),
349 Effect.repeat({
350 schedule: Schedule.spaced("1 second"),
351 times: 10,
352 until: (entries) => entries.includes("checkpoint"),
353 }),
354 );
355 if (!recovered.includes(`${caught}:true`)) {
356 yield* Effect.logInfo(
357 "Workflow replay recovery status",
358 yield* request(`${url}/status/${id}`),
359 );
360 }
361 expect(recovered).toContain(`${caught}:true`);
362 // A journal write does not acknowledge native run completion.
363 const completed = yield* request(`${url}/status/${id}`).pipe(
364 Effect.flatMap((body) =>
365 Effect.try(() => JSON.parse(body) as Status),
366 ),
367 Effect.repeat({
368 schedule: Schedule.spaced("1 second"),
369 times: 10,
370 until: (status) =>
371 status.status === "complete" || status.status === "errored",
372 }),
373 );
374 expect(completed, JSON.stringify(completed)).toMatchObject({
375 status: "complete",
376 output: recovered,
377 });
378 yield* request(`${url}/restart/${id}`, "POST");
379 const journal = yield* request(`${url}/journal/${id}`).pipe(
380 Effect.flatMap((body) =>
381 Effect.try(() => JSON.parse(body) as string[]),
382 ),
383 Effect.repeat({
384 schedule: Schedule.spaced("1 second"),
385 times: 10,
386 until: (entries) => entries.includes("after-task"),
387 }),
388 );
389 yield* Effect.logInfo(
390 `Replay journal ${id}: ${JSON.stringify(journal)}`,
391 );
392 if (!journal.includes("after-task")) {
393 yield* Effect.logInfo(
394 "Workflow checkpoint restart status",
395 yield* request(`${url}/status/${id}`),
396 );
397 }
398 expect(journal).toContain("after-task");
399 // Native terminal failures may rerun on an explicit checkpoint restart.
400 const reranTerminal =
401 terminal && journal.includes("open:3:captured:true");
402 expect(journal).toContain(`${caught}:${reranTerminal}`);
403 const replayed = yield* request(`${url}/status/${id}`).pipe(
404 Effect.flatMap((body) =>
405 Effect.try(() => JSON.parse(body) as Status),
406 ),
407 Effect.repeat({
408 schedule: Schedule.spaced("1 second"),
409 times: 10,
410 until: (status) =>
411 status.status === "complete" || status.status === "errored",
412 }),
413 );
414 expect(replayed, JSON.stringify(replayed)).toMatchObject({
415 status: "complete",
416 entries: [
417 "run",
418 "open:1:captured:true",
419 "close:1",
420 "open:2:captured:true",
421 "close:2",
422 `${caught}:true`,
423 "checkpoint",
424 "run",
425 ...(reranTerminal ? ["open:3:captured:true", "close:3"] : []),
426 `${caught}:${reranTerminal}`,
427 "after-task",
428 ],
429 });
430 }),
431 { tags: [...(dev ? ["local"] : ["live"])], timeout: 60_000 },
432 );
433 }
434
435 for (const { scenario, entries } of cases) {
436 test(
437 `closes and owns ${scenario} attempt resources`,
438 Effect.gen(function* () {
439 const { url } = yield* stack;
440 const started = yield* request(`${url}/start/${scenario}`, "POST");
441 const { id } = yield* Effect.try(
442 () => JSON.parse(started) as { id: string },
443 );
444 if (scenario.startsWith("rollback")) {
445 // Native status() can reject during compensation; observe cleanup first.
446 const journal = yield* request(`${url}/journal/${id}`).pipe(
447 Effect.flatMap((body) =>
448 Effect.try(() => JSON.parse(body) as string[]),
449 ),
450 Effect.repeat({
451 schedule: Schedule.spaced("2 seconds"),
452 times: 10,
453 until: (actual) => actual.includes(entries[entries.length - 1]),
454 }),
455 );
456 expect(journal).toEqual(entries);
457 }
458 const status = yield* request(`${url}/status/${id}`).pipe(
459 Effect.flatMap((body) =>
460 Effect.try(() => JSON.parse(body) as Status),
461 ),
462 Effect.repeat({
463 schedule: Schedule.spaced("2 seconds"),
464 times: 10,
465 until: (status) =>
466 status.status === "complete" || status.status === "errored",
467 }),
468 );
469 const failed =
470 scenario.startsWith("rollback") ||
471 scenario.startsWith("unsupported") ||
472 ["uncaught", "die", "orDie"].includes(scenario);
473 expect(status, JSON.stringify(status)).toMatchObject({
474 status: failed ? "errored" : "complete",
475 entries,
476 });
477 if (scenario.startsWith("rollback")) {
478 if (!dev) {
479 expect(status.rollback).toMatchObject({
480 outcome:
481 scenario === "rollback-die" || scenario === "rollback-orDie"
482 ? "failed"
483 : "complete",
484 });
485 }
486 expect(status.error).toMatchObject({
487 message: "trigger compensation",
488 });
489 } else if (scenario.startsWith("unsupported")) {
490 expect(status.error?.message).toContain(
491 "Workflow application failure is not serializable",
492 );
493 if (scenario === "unsupported-array")
494 expect(status.error?.message).toContain(
495 "sparse arrays or custom array properties",
496 );
497 if (
498 scenario === "unsupported-accessor" ||
499 scenario === "unsupported-tag-accessor"
500 ) {
501 expect(status.error?.message).toContain("accessor error data");
502 expect(status.error?.message).not.toContain("getter was invoked");
503 }
504 if (scenario === "unsupported-alias")
505 expect(status.error?.message).toContain(
506 "shared references in error data",
507 );
508 } else if (failed) {
509 expect(status.error?.message).toContain("application failure");
510 } else {
511 expect(status.output).toEqual(entries);
512 }
513 }),
514 { tags: [...(dev ? ["local"] : ["live"])], timeout: 60_000 },
515 );
516 }
517 },
518 {
519 tags: [
520 "provider:cloudflare",
521 "provider:cloudflare:worker",
522 "provider:cloudflare:workflow",
523 ],
524 },
525);