Open History to see the last commit
1import * as Cloudflare from "@alchemy.run/cloudflare";
2import { describe } from "@alchemy.run/test";
3import * as Test from "@alchemy.run/test/Alchemy";
4import { inMemoryState } from "alchemy/State";
5import type { Status } from "alchemy/Workflows";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as HttpClient from "effect/http/HttpClient";
9import * as Schedule from "effect/Schedule";
10import { conformance } from "../../Workflows/conformance/conformance.ts";
11import makeStack from "./fixtures/durable-workflow/stack.ts";
12
13class NotReady extends Data.TaggedError("NotReady")<{ readonly message: string }> {}
14
15/** A fixture route's body, failing with `NotReady` until it answers 200. */
16const fetchText = (url: string, method: "GET" | "POST" = "GET") =>
17 Effect.gen(function* () {
18 const client = yield* HttpClient.HttpClient;
19 const response = yield* method === "POST" ? client.post(url) : client.get(url);
20 const body = yield* response.text;
21 if (response.status !== 200) {
22 return yield* new NotReady({ message: `${method} ${url}: HTTP ${response.status}: ${body}` });
23 }
24 return body;
25 });
26
27describe.concurrent.each([
28 { engine: "durable-object", dev: true, stage: `${Test.defaultStage()}-durable-workflow-local` },
29 { engine: "durable-object", dev: false, stage: `${Test.defaultStage()}-durable-workflow-live` },
30 {
31 engine: "durable-object-eu",
32 dev: true,
33 stage: `${Test.defaultStage()}-durable-workflow-eu-local`,
34 },
35 {
36 engine: "durable-object-eu",
37 dev: false,
38 stage: `${Test.defaultStage()}-durable-workflow-eu-live`,
39 },
40 { engine: "native", dev: true, stage: `${Test.defaultStage()}-native-engine-local` },
41 { engine: "native", dev: false, stage: `${Test.defaultStage()}-native-engine-live` },
42] as const)("Workflows on the $engine engine (dev: $dev)", ({ engine, dev, stage }) => {
43 const state = dev ? inMemoryState() : Cloudflare.state();
44 const Stack = makeStack(state, engine);
45 const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
46 providers: Cloudflare.providers(),
47 state,
48 dev,
49 stage,
50 });
51
52 const stack = beforeAll(
53 Effect.gen(function* () {
54 yield* destroy(Stack);
55 const output = yield* deploy(Stack);
56 // A fresh workers.dev URL can take a while to start serving.
57 yield* fetchText(`${output.url}/ready`).pipe(
58 Effect.retry({ schedule: Schedule.spaced("2 seconds"), times: 30 }),
59 );
60 // A Workflow class can take a while to reach every Worker version after
61 // a deploy: wait until a run actually completes.
62 yield* Effect.gen(function* () {
63 const id = `probe-${crypto.randomUUID().slice(0, 8)}`;
64 const { executionId } = JSON.parse(
65 yield* fetchText(`${output.url}/start/Greet/${id}`, "POST"),
66 ) as { executionId: string };
67 yield* Effect.gen(function* () {
68 const status = JSON.parse(
69 yield* fetchText(`${output.url}/status/Greet/${id}?executionId=${executionId}`),
70 ) as Status<unknown, unknown>;
71 if (status._tag !== "Completed") {
72 return yield* new NotReady({ message: JSON.stringify(status) });
73 }
74 }).pipe(Effect.retry({ schedule: Schedule.spaced("1 second"), times: 10 }));
75 }).pipe(Effect.retry({ schedule: Schedule.spaced("3 seconds"), times: 10 }));
76 return output;
77 }),
78 { timeout: 240_000 },
79 );
80 afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack), { timeout: 60_000 });
81
82 conformance({
83 test,
84 url: Effect.map(stack, ({ url }) => url.replace(/\/+$/, "")),
85 napsWithinMillis: 20_000,
86 // The Durable Object engine retries a run with its alarm backoff: 5 then
87 // 10 seconds, then the one-second sleep. Cloudflare Workflows retries a
88 // step with its own policy.
89 recoveryWithinMillis: engine === "native" ? undefined : 30_000,
90 timeout: 90_000,
91 unsupported: engine === "native" ? { raceAll: "aren't supported on Cloudflare Workflows" } : {},
92 });
93});