1import * as Cloudflare from "@/Cloudflare";
2import * as Test from "@/Test/Alchemy";
3import { expect } from "alchemy-test";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import { MinimumLogLevel } from "effect/References";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import PauseWorkflowWorker from "./fixtures/workflow-pause/worker.ts";
10
11const { test } = Test.make({
12 providers: Cloudflare.providers(),
13 dev: true,
14});
15
16const logLevel = Effect.provideService(
17 MinimumLogLevel,
18 process.env.DEBUG ? "Debug" : "Info",
19);
20
21interface WorkflowStatus {
22 status: string;
23 error?: { message?: string } | null;
24}
25
26class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
27 status: number;
28}> {}
29
30const isSettledAfterPause = (status: WorkflowStatus) =>
31 status.status === "paused" ||
32 status.status === "errored" ||
33 status.status === "complete" ||
34 status.status === "terminated";
35
36/**
37 * End-to-end regression for #930: under local `alchemy dev`, pausing during an
38 * active `Cloudflare.Workflows.task` must leave the instance `paused` once that
39 * task finishes — not `errored` with `UnknownError` from `Effect.tryPromise`.
40 */
41test.provider(
42 "pausing during an active task settles as paused under local workerd",
43 (stack) =>
44 Effect.gen(function* () {
45 yield* stack.destroy();
46
47 const { worker } = yield* stack.deploy(
48 Effect.gen(function* () {
49 const w = yield* PauseWorkflowWorker;
50 return { worker: w };
51 }),
52 );
53
54 const client = yield* HttpClient.HttpClient;
55 const url = worker.url!;
56 expect(url).toBeTypeOf("string");
57
58 const { instanceId } = yield* client
59 .post(`${url}/workflow/start/world`)
60 .pipe(
61 Effect.flatMap((res) =>
62 res.status === 200
63 ? res.json.pipe(
64 Effect.flatMap((body) => {
65 const instanceId = (body as { instanceId?: unknown })
66 .instanceId;
67 return typeof instanceId === "string"
68 ? Effect.succeed({ instanceId })
69 : Effect.fail(
70 new Error("Worker returned no workflow id"),
71 );
72 }),
73 )
74 : Effect.fail(new WorkerNotReady({ status: res.status })),
75 ),
76 Effect.retry({
77 while: (e): e is WorkerNotReady => e instanceof WorkerNotReady,
78 schedule: Schedule.max([
79 Schedule.exponential("500 millis"),
80 Schedule.recurs(15),
81 ]),
82 }),
83 );
84
85 const pauseRes = yield* client.post(
86 `${url}/workflow/pause/${instanceId}`,
87 );
88 expect(pauseRes.status).toBe(200);
89
90 // The active task sleeps 8s; give the engine time to finish the step and
91 // apply the pending pause (or incorrectly mark the run errored).
92 const lastStatus = yield* client
93 .get(`${url}/workflow/status/${instanceId}`)
94 .pipe(
95 Effect.flatMap((res) =>
96 res.status === 200
97 ? res.json.pipe(
98 Effect.map((json) => json as unknown as WorkflowStatus),
99 )
100 : Effect.succeed({ status: "pending" } as WorkflowStatus),
101 ),
102 Effect.repeat({
103 schedule: Schedule.spaced("500 millis"),
104 until: isSettledAfterPause,
105 times: 40,
106 }),
107 );
108
109 expect(lastStatus.status).toBe("paused");
110 expect(lastStatus.error).toBeFalsy();
111
112 yield* stack.destroy();
113 }).pipe(logLevel),
114 {
115 tags: [
116 "provider:cloudflare",
117 "provider:cloudflare:worker",
118 "provider:cloudflare:workflow",
119 "local",
120 ],
121 timeout: 180_000,
122 },
123);