1import { adopt } from "@/AdoptPolicy";
2import * as Cloudflare from "@/Cloudflare";
3import { WorkflowResource } from "@/Cloudflare/Workflows/Workflow";
4import { generateWorkflowName } from "@/Cloudflare/Workflows/WorkflowName";
5import { sha256 } from "@/Util/sha256";
6import * as Cause from "effect/Cause";
7import * as Clock from "effect/Clock";
8import * as Exit from "effect/Exit";
9import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
10import * as Output from "@/Output";
11import * as Test from "@/Test/Alchemy";
12import * as TestCore from "@/Test/Core";
13import * as queues from "@distilled.cloud/cloudflare/queues";
14import * as workers from "@distilled.cloud/cloudflare/workers";
15import * as workflows from "@distilled.cloud/cloudflare/workflows";
16import { expect } from "alchemy-test";
17import * as Effect from "effect/Effect";
18import { MinimumLogLevel } from "effect/References";
19import * as Schedule from "effect/Schedule";
20import * as Schema from "effect/Schema";
21import * as Stream from "effect/Stream";
22import * as HttpClient from "effect/http/HttpClient";
23import * as HttpClientRequest from "effect/http/HttpClientRequest";
24import { requestWorker } from "../Utils/WorkerRequest.ts";
25import ColdEffectWorker, {
26 COLD_EFFECT_WORKFLOW_NAME,
27} from "./fixtures/workflow-async/effect-worker.ts";
28import { AsyncWorkflowWorker } from "./fixtures/workflow-async/stack.ts";
29
30const { test } = Test.make({ providers: Cloudflare.providers() });
31
32const logLevel = Effect.provideService(
33 MinimumLogLevel,
34 process.env.DEBUG ? "Debug" : "Info",
35);
36
37const retry = Effect.retry({
38 schedule: Schedule.spaced("3 seconds"),
39 times: 10,
40});
41const hostMain = `${import.meta.dirname}/fixtures/workflow-async/worker.ts`;
42const consumerMain = `${import.meta.dirname}/fixtures/workflow-async/consumer.ts`;
43
44const WorkflowStatus = Schema.Struct({
45 status: Schema.String,
46 output: Schema.optional(
47 Schema.NullOr(
48 Schema.Struct({
49 greeting: Schema.String,
50 workflowName: Schema.optional(Schema.String),
51 }),
52 ),
53 ),
54 error: Schema.optional(
55 Schema.NullOr(Schema.Struct({ message: Schema.optional(Schema.String) })),
56 ),
57});
58
59const WorkflowEvents = Schema.Array(
60 Schema.Struct({
61 type: Schema.String,
62 source: Schema.Struct({ type: Schema.String, workflowName: Schema.String }),
63 payload: Schema.Struct({ instanceId: Schema.String }),
64 }),
65);
66
67const EnvResponse = Schema.Struct({
68 greeting: Schema.String,
69 config: Schema.String,
70 effect: Schema.String,
71 output: Schema.String,
72 asset: Schema.String,
73});
74
75const getJson = <A>(url: string, schema: Schema.Decoder<A>) =>
76 Effect.gen(function* () {
77 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
78 return yield* client.get(url).pipe(
79 Effect.flatMap((res) => res.json),
80 Effect.flatMap(Schema.decodeUnknownEffect(schema)),
81 retry,
82 );
83 });
84
85const runWorkflowToCompletion = (url: string, expectedWorkflowName?: string) =>
86 Effect.gen(function* () {
87 const http = yield* HttpClient.HttpClient;
88 const client = HttpClient.filterStatusOk(http);
89 if (expectedWorkflowName !== undefined) {
90 const identity = yield* getJson(
91 `${url}/workflow/identity`,
92 Schema.Struct({ workflowName: Schema.optional(Schema.String) }),
93 ).pipe(
94 Effect.tap((identity) =>
95 Effect.logInfo("Workflow identity readiness", {
96 expected: expectedWorkflowName,
97 observed: identity.workflowName,
98 }),
99 ),
100 Effect.repeat({
101 schedule: Schedule.spaced("2 seconds"),
102 until: (identity) => identity.workflowName === expectedWorkflowName,
103 times: 10,
104 }),
105 Effect.timeout("30 seconds"),
106 );
107 expect(identity.workflowName).toBe(expectedWorkflowName);
108 const rejected = yield* http.post(`${url}/workflow/start/world`, {
109 headers: {
110 "x-expected-workflow-name": `${expectedWorkflowName}-stale`,
111 },
112 });
113 yield* rejected.text;
114 expect(rejected.status).toBe(409);
115 }
116 const { instanceId } = yield* client
117 .post(
118 `${url}/workflow/start/world`,
119 expectedWorkflowName === undefined
120 ? undefined
121 : { headers: { "x-expected-workflow-name": expectedWorkflowName } },
122 )
123 .pipe(
124 Effect.flatMap((res) => res.json),
125 Effect.flatMap(
126 Schema.decodeUnknownEffect(
127 Schema.Struct({ instanceId: Schema.String }),
128 ),
129 ),
130 retry,
131 );
132
133 const status = yield* getJson(
134 `${url}/workflow/status/${instanceId}`,
135 WorkflowStatus,
136 ).pipe(
137 Effect.repeat({
138 schedule: Schedule.spaced("3 seconds"),
139 until: (s) => s.status === "complete" || s.status === "errored",
140 times: 10,
141 }),
142 Effect.timeout("60 seconds"),
143 );
144 if (status.status !== "complete") {
145 return yield* Effect.fail(
146 new Error(`workflow ${status.status}: ${JSON.stringify(status.error)}`),
147 );
148 }
149 expect(status.error).toBeFalsy();
150 expect(status.output?.greeting).toBe("Hello, world!");
151 return { ...status, instanceId };
152 });
153
154const readWorkflowBinding = (scriptName: string) =>
155 Effect.gen(function* () {
156 const { accountId } = yield* yield* CloudflareEnvironment;
157 const settings = yield* workers.getScriptScriptAndVersionSetting({
158 accountId,
159 scriptName,
160 });
161 const bindings = (settings.bindings ?? []).filter(
162 (binding) => binding.type === "workflow",
163 );
164 expect(bindings).toHaveLength(1);
165 return bindings[0]!;
166 });
167
168const expectWorkerGone = (accountId: string, scriptName: string) =>
169 workers.getScriptScriptAndVersionSetting({ accountId, scriptName }).pipe(
170 Effect.as(false),
171 Effect.catchTag("WorkerNotFound", () => Effect.succeed(true)),
172 Effect.catchTag("WorkerHasNoVersions", () => Effect.succeed(false)),
173 Effect.repeat({
174 schedule: Schedule.spaced("2 seconds"),
175 until: Boolean,
176 times: 10,
177 }),
178 Effect.tap((gone) => Effect.sync(() => expect(gone).toBe(true))),
179 );
180
181const expectWorkflowGone = (accountId: string, workflowName: string) =>
182 workflows.getWorkflow({ accountId, workflowName }).pipe(
183 Effect.as(false),
184 Effect.catchTag("WorkflowNotFound", () => Effect.succeed(true)),
185 Effect.repeat({
186 schedule: Schedule.spaced("2 seconds"),
187 until: Boolean,
188 times: 10,
189 }),
190 Effect.tap((gone) => Effect.sync(() => expect(gone).toBe(true))),
191 );
192
193const bindingSnapshot = (binding: Cloudflare.Workflows.WorkflowBinding) => ({
194 kind: binding.kind,
195 name: binding.name,
196 className: binding.className,
197 workflowName: binding.workflowName,
198 scriptName: binding.scriptName,
199 workflowIsOutput: Output.isOutput(binding.workflowName),
200 scriptIsOutput: Output.isOutput(binding.scriptName),
201});
202
203test.provider(
204 "async worker workflow binding is a direct first-deployment queue subscription source",
205 (stack) =>
206 Effect.gen(function* () {
207 yield* stack.destroy();
208 const { accountId } = yield* yield* CloudflareEnvironment;
209 const program = Effect.gen(function* () {
210 const worker = yield* AsyncWorkflowWorker;
211 const queue = yield* Cloudflare.Queues.Queue("WorkflowEventsQueue");
212 const subscription = yield* Cloudflare.Queues.Subscription(
213 "WorkflowEvents",
214 {
215 source: worker.env.MY_WORKFLOW,
216 events: ["instance.completed", "instance.errored"],
217 queueId: queue.queueId,
218 },
219 );
220 expect(subscription.Props.source).toEqual({
221 type: "workflows.workflow",
222 workflowName: worker.env.MY_WORKFLOW.workflowName,
223 });
224 yield* Cloudflare.Queues.Consumer("WorkflowEventsConsumer", {
225 queueId: queue.queueId,
226 scriptName: worker.workerName,
227 settings: { batchSize: 1, maxWaitTimeMs: 1000 },
228 });
229 return {
230 worker,
231 queue,
232 subscription,
233 binding: bindingSnapshot(worker.env.MY_WORKFLOW),
234 env: {
235 greeting: worker.env.GREETING,
236 config: worker.env.CONFIG,
237 effect: worker.env.EFFECT,
238 output: worker.env.OUTPUT,
239 outputIsOutput: Output.isOutput(worker.env.OUTPUT),
240 assets: worker.env.ASSETS,
241 },
242 };
243 });
244 const deployed = yield* stack.deploy(program);
245 const plan = yield* stack.plan(program);
246 expect(plan.resources.Greeting.downstream).toContain("WorkflowEvents");
247 expect(
248 Object.values(plan.resources).filter(
249 (node) => node.resource.Type === "Cloudflare.Workflow",
250 ),
251 ).toHaveLength(1);
252
253 const { binding, worker, queue, subscription } = deployed;
254 expect(binding.workflowIsOutput).toBe(true);
255 expect(binding.scriptIsOutput).toBe(true);
256 expect(binding.kind).toBe("Cloudflare.Workflow");
257 expect(binding.name).toBe("Greeting");
258 expect(binding.className).toBe("MyWorkflow");
259 expect(binding.scriptName).toBe(worker.workerName);
260 expect(deployed.env).toEqual({
261 greeting: "hello",
262 config: "configured",
263 effect: "effect",
264 output: "output",
265 outputIsOutput: true,
266 assets: { kind: "Cloudflare.Workers.Assets" },
267 });
268
269 const uploaded = yield* readWorkflowBinding(worker.workerName);
270 expect(uploaded.workflowName).toBe(binding.workflowName);
271 expect(uploaded.className).toBe(binding.className);
272 const observed = yield* workflows.getWorkflow({
273 accountId,
274 workflowName: binding.workflowName,
275 });
276 expect(observed.name).toBe(binding.workflowName);
277 expect(observed.scriptName).toBe(binding.scriptName);
278 expect(observed.className).toBe(binding.className);
279 const source = {
280 type: "workflows.workflow",
281 workflowName: binding.workflowName,
282 };
283 expect(subscription.source).toEqual(source);
284 expect(plan.resources.WorkflowEvents.state).toHaveProperty(
285 "props.source",
286 source,
287 );
288 const liveSubscription = yield* queues.getSubscription({
289 accountId,
290 subscriptionId: subscription.subscriptionId,
291 });
292 expect(liveSubscription.source).toEqual(expect.objectContaining(source));
293 expect(liveSubscription.destination.queueId).toBe(queue.queueId);
294
295 expect(yield* getJson(`${worker.url}/env`, EnvResponse)).toEqual({
296 greeting: "hello",
297 config: "configured",
298 effect: "effect",
299 output: "output",
300 asset: "workflow asset\n",
301 });
302 const { instanceId } = yield* runWorkflowToCompletion(worker.url!);
303 const events = yield* getJson(
304 `${worker.url}/events`,
305 WorkflowEvents,
306 ).pipe(
307 Effect.repeat({
308 schedule: Schedule.spaced("3 seconds"),
309 until: (events) =>
310 events.some((event) => event.payload.instanceId === instanceId),
311 times: 10,
312 }),
313 Effect.timeout("60 seconds"),
314 );
315 expect(events).toContainEqual(
316 expect.objectContaining({
317 type: "cf.workflows.workflow.instance.completed",
318 source: expect.objectContaining(source),
319 payload: expect.objectContaining({ instanceId }),
320 }),
321 );
322 yield* Effect.logInfo(
323 `Workflow queue lifecycle delivery: ${JSON.stringify(events)}`,
324 );
325
326 yield* stack.destroy();
327 yield* expectWorkerGone(accountId, worker.workerName);
328 yield* expectWorkflowGone(accountId, binding.workflowName);
329 const queueGone = yield* queues
330 .getQueue({ accountId, queueId: queue.queueId })
331 .pipe(
332 Effect.as(false),
333 Effect.catchTag("QueueNotFound", () => Effect.succeed(true)),
334 );
335 expect(queueGone).toBe(true);
336 const subscriptionGone = yield* queues
337 .getSubscription({
338 accountId,
339 subscriptionId: subscription.subscriptionId,
340 })
341 .pipe(
342 Effect.as(false),
343 Effect.catchTag("SubscriptionNotFound", () => Effect.succeed(true)),
344 );
345 expect(subscriptionGone).toBe(true);
346 }).pipe(logLevel),
347 {
348 tags: [
349 "provider:cloudflare",
350 "provider:cloudflare:queue",
351 "provider:cloudflare:worker",
352 "provider:cloudflare:workflow",
353 "live",
354 ],
355 timeout: 120_000,
356 },
357);
358
359const expectWorkflowEvent = (
360 url: string,
361 instanceId: string,
362 workflowName: string,
363) =>
364 getJson(`${url}/events`, WorkflowEvents).pipe(
365 Effect.repeat({
366 schedule: Schedule.spaced("3 seconds"),
367 until: (events) =>
368 events.some((event) => event.payload.instanceId === instanceId),
369 times: 10,
370 }),
371 Effect.timeout("60 seconds"),
372 Effect.tap((events) =>
373 Effect.sync(() =>
374 expect(events).toContainEqual(
375 expect.objectContaining({
376 type: "cf.workflows.workflow.instance.completed",
377 source: { type: "workflows.workflow", workflowName },
378 payload: expect.objectContaining({ instanceId }),
379 }),
380 ),
381 ),
382 ),
383 );
384
385const expectWorkflowSubscriptionGone = (
386 accountId: string,
387 subscriptionId: string,
388 queueId: string,
389) =>
390 Effect.gen(function* () {
391 expect(
392 yield* queues.getSubscription({ accountId, subscriptionId }).pipe(
393 Effect.as(false),
394 Effect.catchTag("SubscriptionNotFound", () => Effect.succeed(true)),
395 ),
396 ).toBe(true);
397 expect(
398 yield* queues.getQueue({ accountId, queueId }).pipe(
399 Effect.as(false),
400 Effect.catchTag("QueueNotFound", () => Effect.succeed(true)),
401 ),
402 ).toBe(true);
403 });
404
405test.provider(
406 "Workflow.ref same-stack and WorkflowResource sources deliver queue events without duplicate ownership",
407 (stack) =>
408 Effect.gen(function* () {
409 yield* stack.destroy();
410 const { accountId } = yield* yield* CloudflareEnvironment;
411 const program = (reference: boolean) =>
412 Effect.gen(function* () {
413 const worker = yield* AsyncWorkflowWorker;
414 const workflow = reference
415 ? yield* Cloudflare.Workflow.ref("Greeting")
416 : yield* WorkflowResource("Greeting", {
417 className: "MyWorkflow",
418 scriptName: worker.workerName,
419 });
420 expect(workflow.Type).toBe("Cloudflare.Workflow");
421 expect(Output.isOutput(workflow.workflowName)).toBe(true);
422 if (reference) {
423 expect(typeof workflow).toBe("function");
424 expect(Output.isOutput(workflow)).toBe(true);
425 }
426 const queue = yield* Cloudflare.Queues.Queue("RefEventsQueue");
427 const subscription = yield* Cloudflare.Queues.Subscription(
428 "RefEvents",
429 {
430 source: workflow,
431 events: ["instance.completed"],
432 queueId: queue.queueId,
433 },
434 );
435 expect(subscription.Props.source).toEqual({
436 type: "workflows.workflow",
437 workflowName: expect.anything(),
438 });
439 yield* Cloudflare.Queues.Consumer("RefEventsConsumer", {
440 queueId: queue.queueId,
441 scriptName: worker.workerName,
442 settings: { batchSize: 1, maxWaitTimeMs: 1000 },
443 });
444 return {
445 worker,
446 queue,
447 subscription,
448 workflowName: workflow.workflowName,
449 };
450 });
451 const firstPlan = yield* stack.plan(program(false));
452 expect(firstPlan.resources.Greeting.downstream).toContain("RefEvents");
453 const first = yield* stack.deploy(program(false));
454 const original = yield* workflows.getWorkflow({
455 accountId,
456 workflowName: first.workflowName,
457 });
458 // Resource refs read persisted state, so deploy the host before using one.
459 const deployed = yield* stack.deploy(program(true));
460 const plan = yield* stack.plan(program(true));
461 expect(
462 Object.values(plan.resources).filter(
463 (node) => node.resource.Type === "Cloudflare.Workflow",
464 ),
465 ).toHaveLength(1);
466 expect(deployed.workflowName).toBe(first.workflowName);
467 expect(deployed.subscription.subscriptionId).toBe(
468 first.subscription.subscriptionId,
469 );
470 const source = {
471 type: "workflows.workflow",
472 workflowName: first.workflowName,
473 };
474 expect(plan.resources.RefEvents.state).toHaveProperty(
475 "props.source",
476 source,
477 );
478 const observed = yield* queues.getSubscription({
479 accountId,
480 subscriptionId: deployed.subscription.subscriptionId,
481 });
482 expect(observed.source).toEqual(expect.objectContaining(source));
483 expect(observed.destination.queueId).toBe(deployed.queue.queueId);
484 const { instanceId } = yield* runWorkflowToCompletion(
485 deployed.worker.url!,
486 ).pipe(
487 Effect.retry({
488 schedule: Schedule.spaced("3 seconds"),
489 times: 2,
490 while: (error) =>
491 error instanceof Error &&
492 error.message ===
493 'workflow errored: {"message":"Worker not found."}',
494 }),
495 );
496 yield* expectWorkflowEvent(
497 deployed.worker.url!,
498 instanceId,
499 first.workflowName,
500 );
501
502 yield* stack.deploy(AsyncWorkflowWorker);
503 yield* expectWorkflowSubscriptionGone(
504 accountId,
505 deployed.subscription.subscriptionId,
506 deployed.queue.queueId,
507 );
508 expect(
509 (yield* workflows.getWorkflow({
510 accountId,
511 workflowName: first.workflowName,
512 })).id,
513 ).toBe(original.id);
514 expect(yield* readWorkflowName(deployed.worker.workerName)).toBe(
515 first.workflowName,
516 );
517 yield* stack.destroy();
518 yield* expectWorkerGone(accountId, deployed.worker.workerName);
519 yield* expectWorkflowGone(accountId, first.workflowName);
520 }).pipe(logLevel),
521 {
522 tags: [
523 "provider:cloudflare",
524 "provider:cloudflare:queue",
525 "provider:cloudflare:worker",
526 "provider:cloudflare:workflow",
527 "live",
528 ],
529 timeout: 120_000,
530 },
531);
532
533test.provider(
534 "Workflow.ref cross-stack and cross-stage source delivers queue events without owning the host",
535 (stack) => {
536 const host = TestCore.scratchStack(
537 { providers: Cloudflare.providers(), stage: `${stack.stage}-host` },
538 "WorkflowRefHost",
539 "test/Cloudflare/Workers/WorkflowAsync.test.ts",
540 );
541 return Effect.gen(function* () {
542 yield* stack.destroy();
543 yield* host.destroy();
544 const { accountId } = yield* yield* CloudflareEnvironment;
545 const hosted = yield* host.deploy(
546 Effect.gen(function* () {
547 const worker = yield* AsyncWorkflowWorker;
548 return { worker, workflowName: worker.env.MY_WORKFLOW.workflowName };
549 }),
550 );
551 const original = yield* workflows.getWorkflow({
552 accountId,
553 workflowName: hosted.workflowName,
554 });
555 const program = Effect.gen(function* () {
556 const worker = yield* Cloudflare.Worker("RefEventsWorker", {
557 main: hostMain,
558 env: { EVENTS: Cloudflare.DurableObject("WorkflowEvents") },
559 });
560 const queue = yield* Cloudflare.Queues.Queue("RefEventsQueue");
561 const Subscription = yield* Cloudflare.Queues.Subscription;
562 const subscription = yield* Subscription(
563 "RefEvents",
564 Effect.gen(function* () {
565 return {
566 source: yield* Cloudflare.Workflow.ref("Greeting", {
567 stack: host.name,
568 stage: host.stage,
569 }),
570 events: ["instance.completed"],
571 queueId: queue.queueId,
572 };
573 }),
574 );
575 yield* Cloudflare.Queues.Consumer("RefEventsConsumer", {
576 queueId: queue.queueId,
577 scriptName: worker.workerName,
578 settings: { batchSize: 1, maxWaitTimeMs: 1000 },
579 });
580 return { worker, queue, subscription };
581 });
582 const deployed = yield* stack.deploy(program);
583 const plan = yield* stack.plan(program);
584 expect(
585 Object.values(plan.resources).filter(
586 (node) => node.resource.Type === "Cloudflare.Workflow",
587 ),
588 ).toHaveLength(0);
589 const source = {
590 type: "workflows.workflow",
591 workflowName: hosted.workflowName,
592 };
593 expect(plan.resources.RefEvents.state).toHaveProperty(
594 "props.source",
595 source,
596 );
597 expect(deployed.subscription.source).toEqual(source);
598 const observed = yield* queues.getSubscription({
599 accountId,
600 subscriptionId: deployed.subscription.subscriptionId,
601 });
602 expect(observed.source).toEqual(expect.objectContaining(source));
603 expect(observed.destination.queueId).toBe(deployed.queue.queueId);
604 const { instanceId } = yield* runWorkflowToCompletion(hosted.worker.url!);
605 yield* expectWorkflowEvent(
606 deployed.worker.url!,
607 instanceId,
608 hosted.workflowName,
609 );
610
611 yield* stack.destroy();
612 yield* expectWorkerGone(accountId, deployed.worker.workerName);
613 yield* expectWorkflowSubscriptionGone(
614 accountId,
615 deployed.subscription.subscriptionId,
616 deployed.queue.queueId,
617 );
618 expect(
619 (yield* workflows.getWorkflow({
620 accountId,
621 workflowName: hosted.workflowName,
622 })).id,
623 ).toBe(original.id);
624 expect(yield* readWorkflowName(hosted.worker.workerName)).toBe(
625 hosted.workflowName,
626 );
627 yield* host.destroy();
628 yield* expectWorkerGone(accountId, hosted.worker.workerName);
629 yield* expectWorkflowGone(accountId, hosted.workflowName);
630 }).pipe(
631 logLevel,
632 Effect.ensuring(
633 stack.destroy().pipe(Effect.andThen(host.destroy()), Effect.ignore),
634 ),
635 );
636 },
637 {
638 tags: [
639 "provider:cloudflare",
640 "provider:cloudflare:queue",
641 "provider:cloudflare:worker",
642 "provider:cloudflare:workflow",
643 "live",
644 ],
645 timeout: 120_000,
646 },
647);
648
649test.provider(
650 "async worker workflow binding accepts scriptName and exposes cross-script identity without duplicate resources",
651 (stack) =>
652 Effect.gen(function* () {
653 yield* stack.destroy();
654 const { accountId } = yield* yield* CloudflareEnvironment;
655 const deploy = (withConsumer: boolean) =>
656 stack.deploy(
657 Effect.gen(function* () {
658 const host = yield* AsyncWorkflowWorker;
659 const consumer = withConsumer
660 ? yield* Cloudflare.Worker(
661 "WorkflowConsumer",
662 Effect.succeed({
663 main: consumerMain,
664 env: {
665 MY_WORKFLOW: Cloudflare.Workflow<{ value: string }>(
666 "RemoteGreeting",
667 {
668 className: "MyWorkflow",
669 scriptName: host.workerName,
670 },
671 ),
672 },
673 }),
674 )
675 : undefined;
676 return {
677 host,
678 consumer,
679 hostBinding: bindingSnapshot(host.env.MY_WORKFLOW),
680 consumerBinding: consumer
681 ? bindingSnapshot(consumer.env.MY_WORKFLOW)
682 : undefined,
683 };
684 }),
685 );
686 const first = yield* deploy(false);
687 const hostWorkflow = yield* workflows.getWorkflow({
688 accountId,
689 workflowName: first.hostBinding.workflowName,
690 });
691 const deployed = yield* deploy(true);
692 const { host, consumer, hostBinding, consumerBinding } = deployed;
693 expect(consumerBinding?.workflowIsOutput).toBe(true);
694 expect(consumerBinding?.scriptIsOutput).toBe(true);
695 expect(consumerBinding?.workflowName).toBe(hostBinding.workflowName);
696 expect(consumerBinding?.scriptName).toBe(host.workerName);
697 expect(consumerBinding?.name).toBe("RemoteGreeting");
698 expect(
699 (yield* readWorkflowBinding(consumer!.workerName)).workflowName,
700 ).toBe(hostBinding.workflowName);
701 const all = yield* workflows.listWorkflows
702 .items({ accountId })
703 .pipe(Stream.runCollect);
704 const owned = Array.from(all).filter(
705 (workflow) =>
706 workflow.scriptName === host.workerName ||
707 workflow.scriptName === consumer!.workerName,
708 );
709 expect(owned).toHaveLength(1);
710 expect(owned[0]?.id).toBe(hostWorkflow.id);
711 expect(owned[0]?.name).toBe(hostBinding.workflowName);
712 yield* runWorkflowToCompletion(consumer!.url!);
713
714 yield* stack.destroy();
715 yield* expectWorkerGone(accountId, consumer!.workerName);
716 yield* expectWorkerGone(accountId, host.workerName);
717 yield* expectWorkflowGone(accountId, hostBinding.workflowName);
718 }).pipe(logLevel),
719 {
720 tags: [
721 "provider:cloudflare",
722 "provider:cloudflare:worker",
723 "provider:cloudflare:workflow",
724 "live",
725 ],
726 timeout: 120_000,
727 },
728);
729
730const waitForAppliedStepLimit = (workflowName: string, expected: number) =>
731 Effect.gen(function* () {
732 const { accountId } = yield* yield* CloudflareEnvironment;
733 const versions = yield* workflows.listVersions
734 .items({ accountId, workflowName })
735 .pipe(Stream.runCollect);
736 return Array.from(versions).some(
737 (version) => version.limits?.steps === expected,
738 );
739 }).pipe(
740 Effect.repeat({
741 schedule: Schedule.spaced("3 seconds"),
742 until: Boolean,
743 times: 10,
744 }),
745 Effect.tap((applied) => Effect.sync(() => expect(applied).toBe(true))),
746 );
747
748test.provider(
749 "async worker workflow binding applies a per-workflow step limit",
750 (stack) =>
751 Effect.gen(function* () {
752 yield* stack.destroy();
753 const { accountId } = yield* yield* CloudflareEnvironment;
754 const deployed = yield* stack.deploy(
755 Effect.gen(function* () {
756 const worker = yield* Cloudflare.Worker("limits-workflow-worker", {
757 main: hostMain,
758 env: {
759 MY_WORKFLOW: Cloudflare.Workflow("MyWorkflow", {
760 limits: { steps: 100 },
761 }),
762 },
763 });
764 return { worker, workflowName: worker.env.MY_WORKFLOW.workflowName };
765 }),
766 );
767 yield* waitForAppliedStepLimit(deployed.workflowName, 100);
768 yield* stack.destroy();
769 yield* expectWorkerGone(accountId, deployed.worker.workerName);
770 yield* expectWorkflowGone(accountId, deployed.workflowName);
771 }).pipe(logLevel),
772 {
773 tags: [
774 "provider:cloudflare",
775 "provider:cloudflare:worker",
776 "provider:cloudflare:workflow",
777 "live",
778 ],
779 timeout: 120_000,
780 },
781);
782
783const scheduledWorkflowMain = `${import.meta.dirname}/fixtures/workflow-schedules/async-worker.ts`;
784const waitForAppliedSchedules = (workflowName: string, expected: string[]) =>
785 Effect.gen(function* () {
786 const { accountId } = yield* yield* CloudflareEnvironment;
787 const workflow = yield* workflows.getWorkflow({ accountId, workflowName });
788 const crons = (workflow.schedules ?? []).map((schedule) => schedule.cron);
789 return crons;
790 }).pipe(
791 Effect.flatMap((crons) =>
792 crons.length === expected.length &&
793 crons.every((cron, index) => cron === expected[index])
794 ? Effect.succeed(crons)
795 : Effect.fail(
796 new Error(`schedules not applied yet: ${JSON.stringify(crons)}`),
797 ),
798 ),
799 retry,
800 );
801
802test.provider(
803 "async worker workflow binding applies native cron schedules",
804 (stack) =>
805 Effect.gen(function* () {
806 yield* stack.destroy();
807 const { accountId } = yield* yield* CloudflareEnvironment;
808 const yearly = "0 0 1 1 *";
809 const other = "0 0 2 1 *";
810 const deployWith = (schedules: string[]) =>
811 stack.deploy(
812 Effect.gen(function* () {
813 const worker = yield* Cloudflare.Worker(
814 "scheduled-workflow-worker",
815 {
816 main: scheduledWorkflowMain,
817 env: {
818 HOURLY: Cloudflare.Workflow("HourlyWorkflow", {
819 className: "HourlyWorkflow",
820 schedules,
821 }),
822 },
823 },
824 );
825 return { worker, workflowName: worker.env.HOURLY.workflowName };
826 }),
827 );
828 const created = yield* deployWith([yearly]);
829 expect(
830 yield* waitForAppliedSchedules(created.workflowName, [yearly]),
831 ).toEqual([yearly]);
832 yield* deployWith([other]);
833 expect(
834 yield* waitForAppliedSchedules(created.workflowName, [other]),
835 ).toEqual([other]);
836 yield* deployWith([]);
837 expect(yield* waitForAppliedSchedules(created.workflowName, [])).toEqual(
838 [],
839 );
840 yield* stack.destroy();
841 yield* expectWorkerGone(accountId, created.worker.workerName);
842 yield* expectWorkflowGone(accountId, created.workflowName);
843 }).pipe(logLevel),
844 {
845 tags: [
846 "provider:cloudflare",
847 "provider:cloudflare:worker",
848 "provider:cloudflare:workflow",
849 "live",
850 ],
851 timeout: 120_000,
852 },
853);
854
855const namedWorkflowMain = `${import.meta.dirname}/fixtures/workflow-async/worker.ts`;
856const physicalName = (scratch: Test.ScratchStack) =>
857 sha256(`${scratch.name}:${scratch.stage}`).pipe(
858 Effect.map((hash) => `alchemy-workflow-${hash.slice(0, 16)}`),
859 );
860const readWorkflowName = (scriptName: string) =>
861 readWorkflowBinding(scriptName).pipe(
862 Effect.map((binding) => binding.workflowName),
863 );
864
865const namedHost = (workflowName?: string, schedules?: string[]) =>
866 Cloudflare.Worker("NamedHost", {
867 main: namedWorkflowMain,
868 env: {
869 MY_WORKFLOW: Cloudflare.Workflow("MyWorkflow", {
870 workflowName,
871 schedules,
872 }),
873 },
874 });
875
876test.provider(
877 "physical names preserve defaults, replace on rename, and retain schedules",
878 (scratch) =>
879 Effect.gen(function* () {
880 yield* scratch.destroy();
881 const { accountId } = yield* yield* CloudflareEnvironment;
882 const explicit = yield* physicalName(scratch);
883 const original = yield* scratch.deploy(namedHost());
884 const generated = yield* readWorkflowName(original.workerName);
885 expect(generated).toBe(
886 yield* generateWorkflowName(original.workerName, "MyWorkflow"),
887 );
888 const first = yield* workflows.getWorkflow({
889 accountId,
890 workflowName: generated,
891 });
892
893 yield* scratch.deploy(namedHost(generated));
894 expect(
895 (yield* workflows.getWorkflow({ accountId, workflowName: generated }))
896 .id,
897 ).toBe(first.id);
898
899 const renamed = yield* scratch.deploy(namedHost(explicit, ["0 0 1 1 *"]));
900 const replacement = yield* workflows.getWorkflow({
901 accountId,
902 workflowName: explicit,
903 });
904 expect(replacement.id).not.toBe(first.id);
905 expect(yield* readWorkflowName(renamed.workerName)).toBe(explicit);
906 yield* expectWorkflowGone(accountId, generated);
907 expect(yield* waitForAppliedSchedules(explicit, ["0 0 1 1 *"])).toEqual([
908 "0 0 1 1 *",
909 ]);
910
911 const preserved = yield* scratch.deploy(namedHost());
912 expect(yield* readWorkflowName(preserved.workerName)).toBe(explicit);
913 const observed = yield* workflows.getWorkflow({
914 accountId,
915 workflowName: explicit,
916 });
917 expect(observed.id).toBe(replacement.id);
918 expect(observed.schedules?.map((schedule) => schedule.cron)).toEqual([
919 "0 0 1 1 *",
920 ]);
921 const terminal = yield* runWorkflowToCompletion(preserved.url!).pipe(
922 Effect.retry({ schedule: Schedule.spaced("2 seconds"), times: 2 }),
923 );
924 expect(terminal.output?.workflowName).toBe(explicit);
925
926 yield* scratch.deploy(namedHost(undefined, []));
927 expect(yield* waitForAppliedSchedules(explicit, [])).toEqual([]);
928 yield* scratch.destroy();
929 yield* expectWorkflowGone(accountId, explicit);
930 }).pipe(logLevel),
931 {
932 tags: [
933 "provider:cloudflare",
934 "provider:cloudflare:worker",
935 "provider:cloudflare:workflow",
936 "live",
937 ],
938 timeout: 120_000,
939 },
940);
941
942test.provider(
943 "direct workflow sources with Effect props preserve explicit identity through omission and rename",
944 (stack) =>
945 Effect.gen(function* () {
946 yield* stack.destroy();
947 const { accountId } = yield* yield* CloudflareEnvironment;
948 const explicit = yield* physicalName(stack);
949 const program = (workflowName?: string) =>
950 Effect.gen(function* () {
951 const worker = yield* Cloudflare.Worker("IdentityHost", {
952 main: hostMain,
953 env: {
954 MY_WORKFLOW: Cloudflare.Workflow("IdentityWorkflow", {
955 className: "MyWorkflow",
956 workflowName,
957 }),
958 WORKFLOW_NAME: workflowName ?? explicit,
959 EVENTS: Cloudflare.DurableObject("WorkflowEvents"),
960 },
961 });
962 const queue = yield* Cloudflare.Queues.Queue("IdentityQueue");
963 const subscription = yield* Cloudflare.Queues.Subscription(
964 "IdentityEvents",
965 Effect.succeed({
966 source: worker.env.MY_WORKFLOW,
967 events: ["instance.completed"],
968 queueId: queue.queueId,
969 }),
970 );
971 expect(subscription.Props.source).toEqual({
972 type: "workflows.workflow",
973 workflowName: worker.env.MY_WORKFLOW.workflowName,
974 });
975 yield* Cloudflare.Queues.Consumer("IdentityConsumer", {
976 queueId: queue.queueId,
977 scriptName: worker.workerName,
978 settings: { batchSize: 1, maxWaitTimeMs: 1000 },
979 });
980 return {
981 worker,
982 queue,
983 subscription,
984 binding: bindingSnapshot(worker.env.MY_WORKFLOW),
985 };
986 });
987 const identities: string[] = [];
988 const subscriptions: string[] = [];
989 const deployments = yield* Effect.forEach(
990 [explicit, undefined, `${explicit}-renamed`],
991 (name) =>
992 Effect.gen(function* () {
993 const deployed = yield* stack.deploy(program(name));
994 const expected = name ?? explicit;
995 const { worker, binding, subscription } = deployed;
996 expect(binding.workflowIsOutput).toBe(true);
997 expect(binding.scriptIsOutput).toBe(true);
998 expect(binding.workflowName).toBe(expected);
999 expect(binding.scriptName).toBe(worker.workerName);
1000 expect(yield* readWorkflowName(worker.workerName)).toBe(expected);
1001 const observed = yield* workflows.getWorkflow({
1002 accountId,
1003 workflowName: binding.workflowName,
1004 });
1005 expect(observed.name).toBe(expected);
1006 expect(observed.scriptName).toBe(binding.scriptName);
1007 identities.push(observed.id!);
1008 subscriptions.push(subscription.subscriptionId);
1009 const source = {
1010 type: "workflows.workflow",
1011 workflowName: expected,
1012 };
1013 const liveSubscription = yield* queues.getSubscription({
1014 accountId,
1015 subscriptionId: subscription.subscriptionId,
1016 });
1017 expect(liveSubscription.source).toEqual(
1018 expect.objectContaining(source),
1019 );
1020 expect(liveSubscription.destination.queueId).toBe(
1021 deployed.queue.queueId,
1022 );
1023 const plan = yield* stack.plan(program(name));
1024 expect(plan.resources.IdentityWorkflow.downstream).toContain(
1025 "IdentityEvents",
1026 );
1027 expect(plan.resources.IdentityEvents.state).toHaveProperty(
1028 "props.source",
1029 source,
1030 );
1031 yield* Effect.logInfo("Workflow delivery phase", {
1032 phase: name ?? "omitted",
1033 workflow: observed,
1034 subscription: liveSubscription,
1035 });
1036 const terminal = yield* runWorkflowToCompletion(
1037 worker.url!,
1038 expected,
1039 );
1040 expect(terminal.output?.workflowName).toBe(expected);
1041 const terminalObservedAt = yield* Clock.currentTimeMillis;
1042 yield* Effect.logInfo("Workflow terminal", {
1043 ...terminal,
1044 observedAt: terminalObservedAt,
1045 });
1046 const client = HttpClient.filterStatusOk(
1047 yield* HttpClient.HttpClient,
1048 );
1049 const events = yield* client.get(`${worker.url}/events`).pipe(
1050 Effect.flatMap((response) =>
1051 Effect.gen(function* () {
1052 const body = yield* response.json.pipe(
1053 Effect.flatMap(Schema.decodeUnknownEffect(WorkflowEvents)),
1054 );
1055 yield* Effect.logInfo("Workflow event poll", {
1056 readAt: response.headers["x-events-read-at"],
1057 cache: response.headers["cf-cache-status"],
1058 age: response.headers.age,
1059 events: body,
1060 });
1061 return body;
1062 }),
1063 ),
1064 retry,
1065 // Live Queue dispatch has taken 55 seconds after Workflow completion.
1066 Effect.repeat({
1067 schedule: Schedule.spaced("5800 millis"),
1068 until: (events) =>
1069 events.some(
1070 (event) => event.payload.instanceId === terminal.instanceId,
1071 ),
1072 times: 10,
1073 }),
1074 Effect.timeoutOrElse({
1075 duration: "60 seconds",
1076 orElse: () => Effect.succeed([]),
1077 }),
1078 );
1079 if (
1080 !events.some(
1081 (event) => event.payload.instanceId === terminal.instanceId,
1082 )
1083 ) {
1084 const diagnostics = yield* Effect.gen(function* () {
1085 yield* Effect.logInfo(
1086 "Workflow event diagnostics",
1087 JSON.stringify({
1088 queue: yield* queues.getQueue({
1089 accountId,
1090 queueId: deployed.queue.queueId,
1091 }),
1092 metrics: yield* queues.getMetricsQueue({
1093 accountId,
1094 queueId: deployed.queue.queueId,
1095 }),
1096 subscription: yield* queues.getSubscription({
1097 accountId,
1098 subscriptionId: subscription.subscriptionId,
1099 }),
1100 storage: yield* getJson(
1101 `${worker.url}/events/diagnostics?instance=${terminal.instanceId}`,
1102 Schema.Unknown,
1103 ),
1104 }),
1105 );
1106 const probeId = `probe-${terminal.instanceId}`;
1107 yield* queues.bulkPushMessages({
1108 accountId,
1109 queueId: deployed.queue.queueId,
1110 messages: [
1111 {
1112 contentType: "json",
1113 body: { type: "diagnostic.queue.probe", probeId },
1114 },
1115 ],
1116 });
1117 yield* Effect.logInfo("Queue probe accepted", { probeId });
1118 const messages = (snapshot: unknown): unknown[] =>
1119 typeof snapshot === "object" &&
1120 snapshot !== null &&
1121 "entries" in snapshot &&
1122 Array.isArray(snapshot.entries)
1123 ? snapshot.entries.flatMap((entry: unknown) =>
1124 Array.isArray(entry) ? [entry[1]] : [],
1125 )
1126 : [];
1127 const hasProbe = (snapshot: unknown) =>
1128 messages(snapshot).some(
1129 (body) =>
1130 typeof body === "object" &&
1131 body !== null &&
1132 "type" in body &&
1133 body.type === "diagnostic.queue.probe" &&
1134 "probeId" in body &&
1135 body.probeId === probeId,
1136 );
1137 const hasOriginal = (snapshot: unknown) =>
1138 messages(snapshot).some(
1139 (body) =>
1140 typeof body === "object" &&
1141 body !== null &&
1142 "payload" in body &&
1143 typeof body.payload === "object" &&
1144 body.payload !== null &&
1145 "instanceId" in body.payload &&
1146 body.payload.instanceId === terminal.instanceId,
1147 );
1148 const probeStorage = yield* client
1149 .get(`${worker.url}/events/diagnostics?probe=${probeId}`)
1150 .pipe(
1151 Effect.flatMap((response) => response.json),
1152 Effect.tap((snapshot) =>
1153 Effect.gen(function* () {
1154 const observedAt = yield* Clock.currentTimeMillis;
1155 yield* Effect.logInfo(
1156 "Queue delivery sample",
1157 JSON.stringify({
1158 observedAt,
1159 elapsedMs: observedAt - terminalObservedAt,
1160 originalPresent: hasOriginal(snapshot),
1161 probePresent: hasProbe(snapshot),
1162 storage: snapshot,
1163 }),
1164 );
1165 }),
1166 ),
1167 Effect.repeat({
1168 schedule: Schedule.spaced("3 seconds"),
1169 until: (snapshot) =>
1170 hasOriginal(snapshot) && hasProbe(snapshot),
1171 times: 9,
1172 }),
1173 );
1174 yield* Effect.logInfo(
1175 "Queue probe delivery",
1176 JSON.stringify({
1177 probeId,
1178 observedAt: yield* Clock.currentTimeMillis,
1179 originalPresent: hasOriginal(probeStorage),
1180 probePresent: hasProbe(probeStorage),
1181 storage: probeStorage,
1182 metrics: yield* queues.getMetricsQueue({
1183 accountId,
1184 queueId: deployed.queue.queueId,
1185 }),
1186 }),
1187 );
1188 }).pipe(Effect.timeout("30 seconds"), Effect.exit);
1189 if (Exit.isFailure(diagnostics)) {
1190 yield* Effect.logWarning(
1191 "Workflow event diagnostics failed",
1192 Cause.pretty(diagnostics.cause),
1193 );
1194 }
1195 }
1196 expect(events).toContainEqual(
1197 expect.objectContaining({
1198 type: "cf.workflows.workflow.instance.completed",
1199 source: expect.objectContaining(source),
1200 payload: expect.objectContaining({
1201 instanceId: terminal.instanceId,
1202 }),
1203 }),
1204 );
1205 return deployed;
1206 }),
1207 );
1208 expect(identities[1]).toBe(identities[0]);
1209 expect(identities[2]).not.toBe(identities[0]);
1210 expect(subscriptions[1]).toBe(subscriptions[0]);
1211 expect(subscriptions[2]).not.toBe(subscriptions[0]);
1212 yield* expectWorkflowGone(accountId, explicit);
1213 const final = deployments[2]!;
1214 yield* stack.destroy();
1215 yield* expectWorkerGone(accountId, final.worker.workerName);
1216 yield* expectWorkflowGone(accountId, `${explicit}-renamed`);
1217 expect(
1218 yield* queues
1219 .getQueue({ accountId, queueId: final.queue.queueId })
1220 .pipe(
1221 Effect.as(false),
1222 Effect.catchTag("QueueNotFound", () => Effect.succeed(true)),
1223 ),
1224 ).toBe(true);
1225 for (const subscriptionId of new Set(subscriptions)) {
1226 expect(
1227 yield* queues.getSubscription({ accountId, subscriptionId }).pipe(
1228 Effect.as(false),
1229 Effect.catchTag("SubscriptionNotFound", () => Effect.succeed(true)),
1230 ),
1231 ).toBe(true);
1232 }
1233 }).pipe(logLevel),
1234 {
1235 tags: [
1236 "provider:cloudflare",
1237 "provider:cloudflare:queue",
1238 "provider:cloudflare:worker",
1239 "provider:cloudflare:workflow",
1240 "live",
1241 ],
1242 timeout: 120_000,
1243 },
1244);
1245
1246test.provider(
1247 "physical names require adoption and preserve adopted identity and schedules",
1248 (scratch) =>
1249 Effect.gen(function* () {
1250 yield* scratch.destroy();
1251 const { accountId } = yield* yield* CloudflareEnvironment;
1252 const workflowName = yield* physicalName(scratch);
1253 const host = Cloudflare.Worker("AdoptionHost", {
1254 main: namedWorkflowMain,
1255 });
1256 const deployed = yield* scratch.deploy(host);
1257 yield* expectWorkflowGone(accountId, workflowName);
1258 const foreign = yield* workflows.putWorkflow({
1259 accountId,
1260 workflowName,
1261 scriptName: deployed.workerName,
1262 className: "MyWorkflow",
1263 schedules: [{ cron: "0 0 1 1 *" }],
1264 });
1265 yield* Effect.addFinalizer(() =>
1266 workflows.deleteWorkflow({ accountId, workflowName }).pipe(
1267 Effect.catchTag("WorkflowNotFound", () => Effect.void),
1268 Effect.orDie,
1269 ),
1270 );
1271 const definition = (adopting: boolean, schedules?: string[]) =>
1272 Effect.gen(function* () {
1273 const worker = yield* host;
1274 return yield* WorkflowResource("Adopted", {
1275 workflowName,
1276 className: "MyWorkflow",
1277 scriptName: worker.workerName,
1278 schedules,
1279 }).pipe(adopt(adopting));
1280 });
1281 const denied = yield* scratch.plan(definition(false)).pipe(Effect.exit);
1282 expect(Exit.isFailure(denied)).toBe(true);
1283 if (Exit.isFailure(denied))
1284 expect(Cause.pretty(denied.cause)).toContain("OwnedBySomeoneElse");
1285 expect(
1286 (yield* workflows.getWorkflow({ accountId, workflowName })).id,
1287 ).toBe(foreign.id);
1288
1289 const adopted = yield* scratch.deploy(definition(true));
1290 expect(adopted.workflowName).toBe(workflowName);
1291 expect(adopted.workflowId).toBe(foreign.id);
1292 expect(adopted.schedules).toEqual(["0 0 1 1 *"]);
1293 const updated = yield* scratch.deploy(definition(false, []));
1294 expect(updated.workflowId).toBe(foreign.id);
1295 expect(yield* waitForAppliedSchedules(workflowName, [])).toEqual([]);
1296 yield* scratch.destroy();
1297 yield* expectWorkflowGone(accountId, workflowName);
1298 }).pipe(logLevel),
1299 {
1300 tags: [
1301 "provider:cloudflare",
1302 "provider:cloudflare:worker",
1303 "provider:cloudflare:workflow",
1304 "live",
1305 ],
1306 timeout: 120_000,
1307 },
1308);
1309
1310test.provider(
1311 "physical names refuse occupied rename targets even with adoption enabled",
1312 (scratch) =>
1313 Effect.gen(function* () {
1314 yield* scratch.destroy();
1315 const { accountId } = yield* yield* CloudflareEnvironment;
1316 const originalName = yield* physicalName(scratch);
1317 const occupiedName = `${originalName}-occupied`;
1318 const occupiedHost = Cloudflare.Worker("OccupiedHost", {
1319 main: namedWorkflowMain,
1320 env: {
1321 MY_WORKFLOW: Cloudflare.Workflow("OccupiedWorkflow", {
1322 className: "MyWorkflow",
1323 workflowName: occupiedName,
1324 schedules: ["0 0 1 1 *"],
1325 }),
1326 },
1327 });
1328 const definition = (name: string) =>
1329 Effect.gen(function* () {
1330 const host = yield* namedHost(name);
1331 const occupied = yield* occupiedHost;
1332 return { host, occupied };
1333 });
1334 yield* scratch.deploy(definition(originalName));
1335 const original = yield* workflows.getWorkflow({
1336 accountId,
1337 workflowName: originalName,
1338 });
1339 const occupied = yield* workflows.getWorkflow({
1340 accountId,
1341 workflowName: occupiedName,
1342 });
1343 const denied = yield* scratch
1344 .deploy(definition(occupiedName).pipe(adopt(true)))
1345 .pipe(Effect.exit);
1346 expect(Exit.isFailure(denied)).toBe(true);
1347 if (Exit.isFailure(denied))
1348 expect(Cause.pretty(denied.cause)).toContain("OwnedBySomeoneElse");
1349 const untouched = yield* workflows.getWorkflow({
1350 accountId,
1351 workflowName: occupiedName,
1352 });
1353 expect(untouched.id).toBe(occupied.id);
1354 expect(untouched.scriptName).toBe(occupied.scriptName);
1355 expect(untouched.schedules).toEqual(occupied.schedules);
1356 expect(
1357 (yield* workflows.getWorkflow({
1358 accountId,
1359 workflowName: originalName,
1360 })).id,
1361 ).toBe(original.id);
1362 // Removing the rejected source must not delete the occupied target.
1363 yield* scratch.deploy(
1364 Effect.gen(function* () {
1365 yield* Cloudflare.Worker("NamedHost", { main: namedWorkflowMain });
1366 yield* occupiedHost;
1367 }),
1368 );
1369 yield* expectWorkflowGone(accountId, originalName);
1370 expect(
1371 (yield* workflows.getWorkflow({
1372 accountId,
1373 workflowName: occupiedName,
1374 })).id,
1375 ).toBe(occupied.id);
1376 yield* scratch.destroy();
1377 yield* expectWorkflowGone(accountId, originalName);
1378 yield* expectWorkflowGone(accountId, occupiedName);
1379 }).pipe(logLevel),
1380 {
1381 tags: [
1382 "provider:cloudflare",
1383 "provider:cloudflare:worker",
1384 "provider:cloudflare:workflow",
1385 "live",
1386 ],
1387 timeout: 120_000,
1388 },
1389);
1390
1391test.provider(
1392 "direct foreign workflow subscription sources do not own the host Workflow",
1393 (scratch) =>
1394 Effect.gen(function* () {
1395 yield* scratch.destroy();
1396 const { accountId } = yield* yield* CloudflareEnvironment;
1397 const workflowName = yield* physicalName(scratch);
1398 const host = namedHost(workflowName);
1399 yield* scratch.deploy(host);
1400 const original = yield* workflows.getWorkflow({
1401 accountId,
1402 workflowName,
1403 });
1404 const deployed = yield* scratch.deploy(
1405 Effect.gen(function* () {
1406 const worker = yield* host;
1407 const consumer = yield* Cloudflare.Worker("NamedConsumer", {
1408 main: namedWorkflowMain,
1409 env: {
1410 MY_WORKFLOW: Cloudflare.Workflow("ForeignWorkflow", {
1411 className: "MyWorkflow",
1412 scriptName: worker.workerName,
1413 workflowName,
1414 }),
1415 },
1416 });
1417 const queue = yield* Cloudflare.Queues.Queue("ForeignEventsQueue");
1418 const Subscription = yield* Cloudflare.Queues.Subscription;
1419 const subscription = yield* Subscription("ForeignEvents", {
1420 source: consumer.env.MY_WORKFLOW,
1421 events: ["instance.completed"],
1422 queueId: queue.queueId,
1423 });
1424 expect(subscription.Props.source).toEqual({
1425 type: "workflows.workflow",
1426 workflowName: consumer.env.MY_WORKFLOW.workflowName,
1427 });
1428 return {
1429 worker,
1430 consumer,
1431 queue,
1432 subscription,
1433 hostBinding: bindingSnapshot(worker.env.MY_WORKFLOW),
1434 consumerBinding: bindingSnapshot(consumer.env.MY_WORKFLOW),
1435 };
1436 }),
1437 );
1438 expect(deployed.hostBinding.workflowName).toBe(workflowName);
1439 expect(deployed.consumerBinding.workflowName).toBe(workflowName);
1440 expect(deployed.consumerBinding.scriptName).toBe(
1441 deployed.worker.workerName,
1442 );
1443 expect(deployed.consumerBinding.workflowIsOutput).toBe(true);
1444 expect(deployed.consumerBinding.scriptIsOutput).toBe(true);
1445 const source = { type: "workflows.workflow", workflowName };
1446 expect(deployed.subscription.source).toEqual(source);
1447 const liveSubscription = yield* queues.getSubscription({
1448 accountId,
1449 subscriptionId: deployed.subscription.subscriptionId,
1450 });
1451 expect(liveSubscription.source).toEqual(expect.objectContaining(source));
1452 expect(liveSubscription.destination.queueId).toBe(deployed.queue.queueId);
1453 const owned = yield* workflows.listWorkflows.items({ accountId }).pipe(
1454 Stream.runCollect,
1455 Effect.map((all) =>
1456 Array.from(all).filter(
1457 (workflow) =>
1458 workflow.scriptName === deployed.worker.workerName ||
1459 workflow.scriptName === deployed.consumer.workerName,
1460 ),
1461 ),
1462 );
1463 expect(owned).toHaveLength(1);
1464 expect(owned[0]?.id).toBe(original.id);
1465 expect(yield* readWorkflowName(deployed.consumer.workerName)).toBe(
1466 workflowName,
1467 );
1468 const terminal = yield* runWorkflowToCompletion(
1469 deployed.consumer.url!,
1470 ).pipe(
1471 Effect.retry({ schedule: Schedule.spaced("2 seconds"), times: 2 }),
1472 );
1473 expect(terminal.output?.workflowName).toBe(workflowName);
1474 yield* scratch.deploy(host);
1475 yield* expectWorkerGone(accountId, deployed.consumer.workerName);
1476 expect(
1477 yield* queues
1478 .getSubscription({
1479 accountId,
1480 subscriptionId: deployed.subscription.subscriptionId,
1481 })
1482 .pipe(
1483 Effect.as(false),
1484 Effect.catchTag("SubscriptionNotFound", () => Effect.succeed(true)),
1485 ),
1486 ).toBe(true);
1487 expect(
1488 yield* queues
1489 .getQueue({ accountId, queueId: deployed.queue.queueId })
1490 .pipe(
1491 Effect.as(false),
1492 Effect.catchTag("QueueNotFound", () => Effect.succeed(true)),
1493 ),
1494 ).toBe(true);
1495 expect(
1496 (yield* workflows.getWorkflow({ accountId, workflowName })).id,
1497 ).toBe(original.id);
1498 yield* scratch.destroy();
1499 yield* expectWorkerGone(accountId, deployed.worker.workerName);
1500 yield* expectWorkflowGone(accountId, workflowName);
1501 }).pipe(logLevel),
1502 {
1503 tags: [
1504 "provider:cloudflare",
1505 "provider:cloudflare:queue",
1506 "provider:cloudflare:worker",
1507 "provider:cloudflare:workflow",
1508 "live",
1509 ],
1510 timeout: 120_000,
1511 },
1512);
1513
1514for (const dev of [false, true]) {
1515 const { test: hostTest } = Test.make({
1516 providers: Cloudflare.providers(),
1517 dev,
1518 });
1519
1520 hostTest.provider(
1521 `binding-only host changes update dependent scriptName in the same deploy (dev: ${dev})`,
1522 (scratch) =>
1523 Effect.gen(function* () {
1524 yield* scratch.destroy();
1525 const workflowName = yield* physicalName(scratch);
1526 const definition = (destination: "A" | "B") =>
1527 Effect.gen(function* () {
1528 const a = yield* Cloudflare.Worker("HostA", {
1529 main: namedWorkflowMain,
1530 });
1531 const b = yield* Cloudflare.Worker("HostB", {
1532 main: namedWorkflowMain,
1533 });
1534 const host = destination === "A" ? a : b;
1535 const workflow = yield* WorkflowResource("MovedWorkflow", {
1536 workflowName,
1537 className: "MyWorkflow",
1538 schedules: ["0 0 1 1 *"],
1539 });
1540 yield* workflow.bind`host`({ scriptName: host.workerName });
1541 const consumer = yield* Cloudflare.Worker("HostConsumer", {
1542 main: namedWorkflowMain,
1543 env: { WORKFLOW_SCRIPT_NAME: workflow.scriptName },
1544 });
1545 return { host, workflow, consumer };
1546 });
1547 const expectScript = (url: string, scriptName: string) =>
1548 requestWorker(
1549 HttpClientRequest.get(`${url}/workflow/script-name`).pipe(
1550 HttpClientRequest.setHeader("cache-control", "no-cache"),
1551 ),
1552 { retryDelay: "3 seconds" },
1553 ).pipe(
1554 Effect.flatMap((response) =>
1555 response.text.pipe(
1556 Effect.flatMap((body) =>
1557 response.status === 200
1558 ? Effect.succeed(body)
1559 : Effect.fail(
1560 new Error(
1561 `GET ${url}/workflow/script-name: ${response.status}: ${body}`,
1562 ),
1563 ),
1564 ),
1565 ),
1566 ),
1567 Effect.repeat({
1568 schedule: Schedule.spaced("3 seconds"),
1569 until: (name) => name === scriptName,
1570 times: 10,
1571 }),
1572 Effect.tap((name) =>
1573 Effect.sync(() => expect(name).toBe(scriptName)),
1574 ),
1575 Effect.timeout("45 seconds"),
1576 );
1577 const original = yield* scratch.deploy(definition("A"));
1578 yield* expectScript(original.consumer.url!, original.host.workerName);
1579 if (dev) {
1580 expect(original.workflow.workflowId).toMatch(/^dev:/);
1581 expect(original.consumer.url).toMatch(/^http:\/\/localhost:/);
1582 }
1583
1584 const moved = yield* scratch.deploy(definition("B"));
1585 expect(moved.host.workerName).not.toBe(original.host.workerName);
1586 expect(moved.workflow.workflowId).toBe(original.workflow.workflowId);
1587 expect(moved.workflow.scriptName).toBe(moved.host.workerName);
1588 yield* expectScript(moved.consumer.url!, moved.host.workerName);
1589 if (!dev) {
1590 const observed = yield* workflows.getWorkflow({
1591 accountId: moved.workflow.accountId,
1592 workflowName,
1593 });
1594 expect(observed.id).toBe(original.workflow.workflowId);
1595 expect(observed.scriptName).toBe(moved.host.workerName);
1596 expect(observed.schedules?.map((schedule) => schedule.cron)).toEqual([
1597 "0 0 1 1 *",
1598 ]);
1599 }
1600 yield* scratch.destroy();
1601 if (!dev) {
1602 yield* expectWorkflowGone(moved.workflow.accountId, workflowName);
1603 }
1604 }).pipe(logLevel),
1605 {
1606 tags: [
1607 "provider:cloudflare",
1608 "provider:cloudflare:worker",
1609 "provider:cloudflare:workflow",
1610 ...(dev ? ["local"] : ["live"]),
1611 ],
1612 timeout: 120_000,
1613 },
1614 );
1615}
1616
1617for (const api of ["async", "Effect"]) {
1618 test.provider(
1619 `public ${api} Workflow cold adoption with a new destination Worker`,
1620 (scratch) =>
1621 Effect.gen(function* () {
1622 yield* scratch.destroy();
1623 const { accountId } = yield* yield* CloudflareEnvironment;
1624 const workflowName =
1625 api === "async"
1626 ? yield* physicalName(scratch)
1627 : COLD_EFFECT_WORKFLOW_NAME;
1628 const source = Cloudflare.Worker("ColdAdoptionSource", {
1629 main: namedWorkflowMain,
1630 });
1631 const seededHost = yield* scratch.deploy(source);
1632 yield* expectWorkflowGone(accountId, workflowName);
1633 const existing = yield* workflows.putWorkflow({
1634 accountId,
1635 workflowName,
1636 scriptName: seededHost.workerName,
1637 className: "MyWorkflow",
1638 schedules: [{ cron: "0 0 1 1 *" }],
1639 });
1640 yield* Effect.addFinalizer(() =>
1641 workflows.deleteWorkflow({ accountId, workflowName }).pipe(
1642 Effect.catchTag("WorkflowNotFound", () => Effect.void),
1643 Effect.orDie,
1644 ),
1645 );
1646 const definition = Effect.gen(function* () {
1647 yield* source;
1648 if (api === "async") return yield* namedHost(workflowName);
1649 return yield* ColdEffectWorker;
1650 });
1651 const denied = yield* scratch
1652 .plan(definition.pipe(adopt(false)))
1653 .pipe(Effect.exit);
1654 expect(Exit.isFailure(denied)).toBe(true);
1655 if (Exit.isFailure(denied))
1656 expect(Cause.pretty(denied.cause)).toContain("OwnedBySomeoneElse");
1657 const untouched = yield* workflows.getWorkflow({
1658 accountId,
1659 workflowName,
1660 });
1661 expect(untouched.id).toBe(existing.id);
1662 expect(untouched.scriptName).toBe(seededHost.workerName);
1663 expect(untouched.schedules?.map((schedule) => schedule.cron)).toEqual([
1664 "0 0 1 1 *",
1665 ]);
1666 const destination = yield* scratch.deploy(definition.pipe(adopt(true)));
1667 const adopted = yield* workflows.getWorkflow({
1668 accountId,
1669 workflowName,
1670 });
1671 expect(adopted.id).toBe(existing.id);
1672 expect(adopted.scriptName).toBe(destination.workerName);
1673 expect(adopted.schedules?.map((schedule) => schedule.cron)).toEqual([
1674 "0 0 1 1 *",
1675 ]);
1676 const terminal = yield* runWorkflowToCompletion(destination.url!).pipe(
1677 Effect.retry({ schedule: Schedule.spaced("2 seconds"), times: 2 }),
1678 );
1679 expect(terminal.output?.greeting).toBe("Hello, world!");
1680 expect(terminal.output?.workflowName).toBe(workflowName);
1681 yield* scratch.destroy();
1682 yield* expectWorkflowGone(accountId, workflowName);
1683 }).pipe(logLevel),
1684 {
1685 tags: [
1686 "provider:cloudflare",
1687 "provider:cloudflare:worker",
1688 "provider:cloudflare:workflow",
1689 "live",
1690 ],
1691 timeout: 120_000,
1692 },
1693 );
1694}