1// Alchemy modifications are licensed under Apache-2.0.
2// This file includes third-party code; see /THIRD_PARTY_LICENSES.md.
3// Alchemy modifications: uses Array<T> syntax and expects current workerd error serialization, which preserves the message without prefixing the custom class name.
4import { runInDurableObject } from "cloudflare:test";
5import { env } from "cloudflare:workers";
6import { NonRetryableError } from "cloudflare:workflows";
7import { afterEach, describe, it, vi } from "vitest";
8import workerdUnsafe from "workerd:unsafe";
9import { DEFAULT_STEP_LIMIT, InstanceEvent, InstanceStatus } from "../index.ts";
10import { ABORT_REASONS, isAbortError } from "../lib/errors.ts";
11import { setTestWorkflowCallback } from "./test-entry.ts";
12import { runWorkflow, runWorkflowAndAwait } from "./utils.ts";
13import type {
14 DatabaseInstance,
15 DatabaseVersion,
16 DatabaseWorkflow,
17 EngineLogs,
18} from "../engine.ts";
19import type {
20 RollbackContext,
21 RollbackFn,
22 WorkflowStepRollbackOptions,
23} from "../lib/rollback.ts";
24import type {
25 WorkflowDelayFunction,
26 WorkflowStep,
27 WorkflowStepConfig,
28 WorkflowStepContext,
29} from "cloudflare:workers";
30
31afterEach(async () => {
32 await workerdUnsafe.abortAllDurableObjects();
33});
34
35describe("Engine", () => {
36 it("should not retry after NonRetryableError is thrown", async ({
37 expect,
38 }) => {
39 const instanceId = "NON-RETRYABLE-ERROR";
40 const engineId = env.ENGINE.idFromName(instanceId);
41
42 await runWorkflow(instanceId, async (_event, step) => {
43 await step.do("should only have one retry", async () => {
44 throw new NonRetryableError("Should only retry once");
45 });
46 });
47
48 await vi.waitUntil(
49 async () => {
50 try {
51 const logs = (await env.ENGINE.get(
52 engineId,
53 ).readLogs()) as EngineLogs;
54 return logs.logs.some(
55 (val) => val.event === InstanceEvent.WORKFLOW_FAILURE,
56 );
57 } catch (e) {
58 // DO may still be aborting — retry
59 if (isAbortError(e)) {
60 return false;
61 }
62 throw e;
63 }
64 },
65 { timeout: 5000 },
66 );
67
68 const logs = (await env.ENGINE.get(engineId).readLogs()) as EngineLogs;
69
70 expect(
71 logs.logs.filter((val) => val.event === InstanceEvent.ATTEMPT_START),
72 ).toHaveLength(1);
73 });
74
75 it("should preserve NonRetryableError message when compat flag is enabled", async ({
76 expect,
77 }) => {
78 const instanceId = "NON-RETRYABLE-PRESERVE-MSG";
79 const engineId = env.ENGINE.idFromName(instanceId);
80 const engineStub = env.ENGINE.get(engineId);
81
82 vi.stubGlobal("Cloudflare", {
83 compatibilityFlags: {
84 workflows_preserve_non_retryable_error_message: true,
85 },
86 });
87
88 try {
89 setTestWorkflowCallback(async (_event, step) => {
90 await step.do("failing-step", async () => {
91 throw new NonRetryableError("my custom error message");
92 });
93 });
94
95 await engineStub
96 .init(
97 12346,
98 {} as DatabaseWorkflow,
99 {} as DatabaseVersion,
100 { id: instanceId } as DatabaseInstance,
101 { payload: {}, timestamp: new Date(), instanceId, workflowName: "" },
102 )
103 .catch(() => {});
104
105 await vi.waitUntil(
106 async () => {
107 try {
108 const logs = (await env.ENGINE.get(
109 engineId,
110 ).readLogs()) as EngineLogs;
111 return logs.logs.some(
112 (val) => val.event === InstanceEvent.WORKFLOW_FAILURE,
113 );
114 } catch (e) {
115 if (isAbortError(e)) {
116 return false;
117 }
118 throw e;
119 }
120 },
121 { timeout: 3000 },
122 );
123
124 const logs = (await env.ENGINE.get(engineId).readLogs()) as EngineLogs;
125
126 const workflowFailure = logs.logs.find(
127 (val) => val.event === InstanceEvent.WORKFLOW_FAILURE,
128 );
129 expect(workflowFailure?.metadata.error).toEqual({
130 name: "NonRetryableError",
131 message: "my custom error message",
132 });
133
134 const attemptFailure = logs.logs.find(
135 (val) => val.event === InstanceEvent.ATTEMPT_FAILURE,
136 );
137 expect(attemptFailure?.metadata.error).toEqual({
138 name: "NonRetryableError",
139 message: "my custom error message",
140 });
141 } finally {
142 vi.unstubAllGlobals();
143 }
144 });
145
146 it("should not error out if step fails but is try-catched", async ({
147 expect,
148 }) => {
149 const engineStub = await runWorkflow(
150 "MOCK-INSTANCE-ID-TRY-CATCH",
151 async (_event, step) => {
152 try {
153 await step.do(
154 "always errors out",
155 {
156 retries: {
157 limit: 0,
158 delay: 1000,
159 },
160 },
161 async () => {
162 throw new Error("Step errors out");
163 },
164 );
165 } catch {}
166 return "finished";
167 },
168 );
169
170 await vi.waitUntil(
171 async () => {
172 const logs = (await engineStub.readLogs()) as EngineLogs;
173 return logs.logs.some(
174 (val) => val.event == InstanceEvent.WORKFLOW_SUCCESS,
175 );
176 },
177 { timeout: 5000 },
178 );
179
180 const logs = (await engineStub.readLogs()) as EngineLogs;
181
182 expect(
183 logs.logs.some((val) => val.event == InstanceEvent.WORKFLOW_SUCCESS),
184 ).toBe(true);
185
186 expect(
187 logs.logs.filter((val) => val.event == InstanceEvent.ATTEMPT_FAILURE),
188 ).toHaveLength(1);
189 });
190
191 it("applies a dynamic delay function and succeeds after retries", async ({
192 expect,
193 }) => {
194 const instanceId = "DYNAMIC-DELAY-APPLIED";
195 const dynamicDelay: WorkflowDelayFunction = () => 50;
196 let attempts = 0;
197
198 const engineStub = await runWorkflow(instanceId, async (_event, step) => {
199 await step.do(
200 "dynamic-delay-step",
201 {
202 retries: { limit: 3, delay: dynamicDelay, backoff: "constant" },
203 },
204 async () => {
205 attempts++;
206 if (attempts < 2) {
207 throw new Error("transient");
208 }
209 return "ok";
210 },
211 );
212 return "done";
213 });
214
215 await vi.waitUntil(
216 async () => {
217 const logs = (await engineStub.readLogs()) as EngineLogs;
218 return logs.logs.some(
219 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
220 );
221 },
222 { timeout: 5000 },
223 );
224
225 const logs = (await engineStub.readLogs()) as EngineLogs;
226 const attemptFailures = logs.logs.filter(
227 (val) => val.event === InstanceEvent.ATTEMPT_FAILURE,
228 );
229
230 expect(attemptFailures.length).toBeGreaterThanOrEqual(1);
231 expect(attemptFailures[0]?.metadata).toMatchObject({ retryDelayMs: 50 });
232 });
233
234 it("fails non-retryably when the delay function throws", async ({
235 expect,
236 }) => {
237 const instanceId = "DYNAMIC-DELAY-BROKEN";
238 const throwingDelay: WorkflowDelayFunction = () => {
239 throw new Error("delay boom");
240 };
241
242 const engineStub = await runWorkflow(instanceId, async (_event, step) => {
243 await step.do(
244 "broken-delay-step",
245 {
246 retries: { limit: 5, delay: throwingDelay, backoff: "constant" },
247 },
248 async () => {
249 throw new Error("step failed");
250 },
251 );
252 return "done";
253 });
254
255 await vi.waitUntil(
256 async () => {
257 const logs = (await engineStub.readLogs()) as EngineLogs;
258 return logs.logs.some(
259 (val) => val.event === InstanceEvent.WORKFLOW_FAILURE,
260 );
261 },
262 { timeout: 5000 },
263 );
264
265 const logs = (await engineStub.readLogs()) as EngineLogs;
266
267 // The broken delay function turns the retry into a non-retryable failure,
268 // so the step is only attempted once.
269 expect(
270 logs.logs.filter((val) => val.event === InstanceEvent.ATTEMPT_START),
271 ).toHaveLength(1);
272
273 // Current workerd preserves the error message across the RPC boundary
274 // without folding the custom class name into it.
275 const workflowFailure = logs.logs.find(
276 (val) => val.event === InstanceEvent.WORKFLOW_FAILURE,
277 );
278 expect(workflowFailure?.metadata.error.message).toContain("delay function");
279 expect(workflowFailure?.metadata.error.message).toContain("delay boom");
280 });
281
282 it("omits delay from ctx.config for a dynamic delay but includes it for a static delay", async ({
283 expect,
284 }) => {
285 const dynamicDelay: WorkflowDelayFunction = () => 10;
286 let dynamicConfig:
287 | WorkflowStepContext<WorkflowDelayFunction>["config"]
288 | undefined;
289 let staticConfig: WorkflowStepContext<number>["config"] | undefined;
290
291 const dynamicStub = await runWorkflow(
292 "DYNAMIC-DELAY-CTX-HIDDEN",
293 async (_event, step) => {
294 await step.do(
295 "dynamic-config-step",
296 {
297 retries: { limit: 1, delay: dynamicDelay, backoff: "constant" },
298 },
299 async (ctx) => {
300 dynamicConfig = ctx.config;
301 return "ok";
302 },
303 );
304 return "done";
305 },
306 );
307
308 await vi.waitUntil(
309 async () => {
310 const logs = (await dynamicStub.readLogs()) as EngineLogs;
311 return logs.logs.some(
312 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
313 );
314 },
315 { timeout: 5000 },
316 );
317
318 const staticStub = await runWorkflow(
319 "STATIC-DELAY-CTX-VISIBLE",
320 async (_event, step) => {
321 await step.do(
322 "static-config-step",
323 {
324 retries: { limit: 1, delay: 1234, backoff: "constant" },
325 },
326 async (ctx) => {
327 staticConfig = ctx.config;
328 return "ok";
329 },
330 );
331 return "done";
332 },
333 );
334
335 await vi.waitUntil(
336 async () => {
337 const logs = (await staticStub.readLogs()) as EngineLogs;
338 return logs.logs.some(
339 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
340 );
341 },
342 { timeout: 5000 },
343 );
344
345 const dynamicRetries = dynamicConfig?.retries;
346 expect(dynamicRetries).toBeDefined();
347 expect(dynamicRetries && "delay" in dynamicRetries).toBe(false);
348 expect(staticConfig?.retries?.delay).toBe(1234);
349 });
350
351 it("waitForEvent should receive events while active", async ({ expect }) => {
352 const engineStub = await runWorkflow(
353 "MOCK-INSTANCE-ID-WAIT-FOR-EVENT",
354 async (_, step) => {
355 return await step.waitForEvent("i'm a event!", {
356 type: "event-type-1",
357 timeout: "10 seconds",
358 });
359 },
360 );
361
362 await vi.waitUntil(
363 async () => {
364 const logs = (await engineStub.readLogs()) as EngineLogs;
365 return logs.logs.some((val) => val.event === InstanceEvent.WAIT_START);
366 },
367 { timeout: 5000 },
368 );
369
370 await engineStub.receiveEvent({
371 type: "event-type-1",
372 timestamp: new Date(),
373 payload: {},
374 });
375
376 await vi.waitUntil(
377 async () => {
378 const logs = (await engineStub.readLogs()) as EngineLogs;
379 return logs.logs.some(
380 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
381 );
382 },
383 { timeout: 5000 },
384 );
385
386 const logs = (await engineStub.readLogs()) as EngineLogs;
387 expect(logs.logs.some((v) => v.event === InstanceEvent.WAIT_START)).toBe(
388 true,
389 );
390 expect(
391 logs.logs.some((v) => v.event === InstanceEvent.WORKFLOW_SUCCESS),
392 ).toBe(true);
393 });
394
395 it("waitForEvent should receive events even if not active", async ({
396 expect,
397 }) => {
398 const engineStub = await runWorkflow(
399 "MOCK-INSTANCE-ID-WAIT-FOR-EVENT-NOT-ACTIVE",
400 async (_, step) => {
401 return await step.waitForEvent("i'm a event!", {
402 type: "event-type-1",
403 timeout: "10 seconds",
404 });
405 },
406 );
407
408 await vi.waitUntil(
409 async () => {
410 const logs = (await engineStub.readLogs()) as EngineLogs;
411 return logs.logs.some((val) => val.event === InstanceEvent.WAIT_START);
412 },
413 { timeout: 5000 },
414 );
415
416 try {
417 await runInDurableObject(engineStub, async (engine) => {
418 await engine.abort("kabooom");
419 });
420 } catch {
421 // supposed to error out
422 }
423
424 // Get a new stub since we've just aborted the durable object
425 const newStub = env.ENGINE.get(
426 env.ENGINE.idFromName("MOCK-INSTANCE-ID-WAIT-FOR-EVENT-NOT-ACTIVE"),
427 );
428
429 await newStub.receiveEvent({
430 type: "event-type-1",
431 timestamp: new Date(),
432 payload: {},
433 });
434
435 await vi.waitUntil(
436 async () => {
437 const logs = (await newStub.readLogs()) as EngineLogs;
438 return logs.logs.some(
439 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
440 );
441 },
442 { timeout: 5000 },
443 );
444
445 const logs = (await newStub.readLogs()) as EngineLogs;
446 expect(logs.logs.some((v) => v.event === InstanceEvent.WAIT_START)).toBe(
447 true,
448 );
449 expect(
450 logs.logs.some((v) => v.event === InstanceEvent.WORKFLOW_SUCCESS),
451 ).toBe(true);
452 });
453
454 it("waitForEvent should not deliver events to timed-out events with the same type", async ({
455 expect,
456 }) => {
457 const instanceId = "WAIT-FOR-EVENT-STALE-WAITER";
458 const engineStub = await runWorkflow(instanceId, async (_, step) => {
459 const results: Array<{
460 iteration: number;
461 received: boolean;
462 }> = [];
463
464 for (let i = 1; i <= 3; i++) {
465 try {
466 await step.waitForEvent(`my-event-waiter-${i}`, {
467 type: "my-event",
468 timeout: 500,
469 });
470 results.push({ iteration: i, received: true });
471 } catch {
472 results.push({ iteration: i, received: false });
473 }
474 }
475
476 return { results };
477 });
478
479 // 1st waitForEvent iteration - should receive event
480 await vi.waitUntil(
481 async () => {
482 const logs = (await engineStub.readLogs()) as EngineLogs;
483 return (
484 logs.logs.filter((val) => val.event === InstanceEvent.WAIT_START)
485 .length >= 1
486 );
487 },
488 { timeout: 5000 },
489 );
490
491 await engineStub.receiveEvent({
492 type: "my-event",
493 timestamp: new Date(),
494 payload: { iteration: 1 },
495 });
496
497 await vi.waitUntil(
498 async () => {
499 const logs = (await engineStub.readLogs()) as EngineLogs;
500 return (
501 logs.logs.filter((val) => val.event === InstanceEvent.WAIT_START)
502 .length >= 2
503 );
504 },
505 { timeout: 5000 },
506 );
507
508 // 2nd waitForEvent iteration - should timeout (500ms)
509 await vi.waitUntil(
510 async () => {
511 const logs = (await engineStub.readLogs()) as EngineLogs;
512 return logs.logs.some(
513 (val) => val.event === InstanceEvent.WAIT_TIMED_OUT,
514 );
515 },
516 { timeout: 5000 },
517 );
518
519 // 3rd waitForEvent iteration - should receive event
520 await vi.waitUntil(
521 async () => {
522 const logs = (await engineStub.readLogs()) as EngineLogs;
523 return (
524 logs.logs.filter((val) => val.event === InstanceEvent.WAIT_START)
525 .length >= 3
526 );
527 },
528 { timeout: 5000 },
529 );
530
531 await engineStub.receiveEvent({
532 type: "my-event",
533 timestamp: new Date(),
534 payload: { iteration: 3 },
535 });
536
537 await vi.waitUntil(
538 async () => {
539 const logs = (await engineStub.readLogs()) as EngineLogs;
540 return logs.logs.some(
541 (val) => val.event === InstanceEvent.WORKFLOW_SUCCESS,
542 );
543 },
544 { timeout: 5000 },
545 );
546
547 const logs = (await engineStub.readLogs()) as EngineLogs;
548 // Iterations 1 and 3 received events; iteration 2 timed out
549 expect(
550 logs.logs.filter((val) => val.event === InstanceEvent.WAIT_COMPLETE),
551 ).toHaveLength(2);
552 expect(
553 logs.logs.filter((val) => val.event === InstanceEvent.WAIT_TIMED_OUT),
554 ).toHaveLength(1);
555 });
556
557 it("should restore state from storage when accountId is undefined", async ({
558 expect,
559 }) => {
560 const instanceId = "RESTORE-TEST-INSTANCE";
561 const accountId = 12345;
562 const workflow: DatabaseWorkflow = {
563 name: "test-workflow",
564 id: "workflow-123",
565 created_on: new Date().toISOString(),
566 modified_on: new Date().toISOString(),
567 script_name: "test-script",
568 class_name: "TestWorkflow",
569 triggered_on: null,
570 };
571 const version: DatabaseVersion = {
572 id: "version-123",
573 class_name: "TestWorkflow",
574 created_on: new Date().toISOString(),
575 modified_on: new Date().toISOString(),
576 workflow_id: workflow.id,
577 mutable_pipeline_id: "pipeline-123",
578 };
579 const instance: DatabaseInstance = {
580 id: instanceId,
581 created_on: new Date().toISOString(),
582 modified_on: new Date().toISOString(),
583 workflow_id: workflow.id,
584 version_id: version.id,
585 status: InstanceStatus.Running,
586 started_on: new Date().toISOString(),
587 ended_on: null,
588 };
589 const event = {
590 payload: {},
591 timestamp: new Date(),
592 instanceId: instanceId,
593 workflowName: "",
594 };
595
596 const engineStub = await runWorkflow(instanceId, async () => {
597 return "test";
598 });
599
600 try {
601 await runInDurableObject(engineStub, async (engine) => {
602 await engine.init(accountId, workflow, version, instance, event);
603 await engine.setStatus(accountId, instanceId, InstanceStatus.Running);
604 await engine.abort(ABORT_REASONS.GRACE_PERIOD_COMPLETE);
605 });
606 } catch (e) {
607 // Expected - abort throws to break the DO
608 if (!isAbortError(e)) {
609 throw e;
610 }
611 }
612
613 const engineId = env.ENGINE.idFromName(instanceId);
614 const restartedStub = env.ENGINE.get(engineId);
615
616 const status = await runInDurableObject(restartedStub, (engine) => {
617 return engine.getStatus();
618 });
619
620 expect(status).toBe(InstanceStatus.Running);
621
622 const logs = (await restartedStub.readLogs()) as EngineLogs;
623 expect(
624 logs.logs.some((log) => log.event === InstanceEvent.WORKFLOW_START),
625 ).toBe(true);
626 });
627
628 it("should complete a step that returns a large Uint8Array without SQLITE_TOOBIG", async ({
629 expect,
630 }) => {
631 // Regression: JSON.stringify(Uint8Array) encodes each byte as a numeric key,
632 // producing a string far larger than byteLength. A 200 KB Uint8Array → ~2 MB JSON
633 // → SQLITE_TOOBIG. writeLog must use a replacer to sanitize TypedArrays.
634 const instanceId = "LARGE-UINT8ARRAY-RESULT";
635 const engineId = env.ENGINE.idFromName(instanceId);
636 const engineStub = env.ENGINE.get(engineId);
637
638 await runWorkflowAndAwait(instanceId, async (_event, step) => {
639 await step.do("large-binary-step", async () => {
640 return new Uint8Array(200_000); // ~200 KB, triggers SQLITE_TOOBIG without fix
641 });
642 });
643
644 const logs = (await engineStub.readLogs()) as EngineLogs;
645 expect(
646 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS),
647 ).toBe(true);
648 expect(
649 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE),
650 ).toBe(false);
651 });
652
653 it("should complete a step that returns an object containing a large Uint8Array without SQLITE_TOOBIG", async ({
654 expect,
655 }) => {
656 // Regression: nested TypedArrays inside objects also cause SQLITE_TOOBIG without
657 // the JSON.stringify replacer, since sanitization must be recursive.
658 const instanceId = "NESTED-UINT8ARRAY-RESULT";
659 const engineId = env.ENGINE.idFromName(instanceId);
660 const engineStub = env.ENGINE.get(engineId);
661
662 await runWorkflowAndAwait(instanceId, async (_event, step) => {
663 await step.do("nested-binary-step", async () => {
664 return { payload: new Uint8Array(200_000), label: "test" };
665 });
666 });
667
668 const logs = (await engineStub.readLogs()) as EngineLogs;
669 expect(
670 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS),
671 ).toBe(true);
672 expect(
673 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE),
674 ).toBe(false);
675 });
676
677 describe("step limits", () => {
678 it("should enforce step limit when exceeded", async ({ expect }) => {
679 const stepLimit = 3;
680 const instanceId = "STEP-LIMIT-EXCEEDED";
681 const engineId = env.ENGINE.idFromName(instanceId);
682 const engineStub = env.ENGINE.get(engineId);
683
684 await runInDurableObject(engineStub, (engine) => {
685 engine.stepLimit = stepLimit;
686 });
687
688 setTestWorkflowCallback(async (_event, step) => {
689 for (let i = 0; i < stepLimit + 1; i++) {
690 await step.do(`step-${i}`, async () => `result-${i}`);
691 }
692 });
693
694 await engineStub.init(
695 12346,
696 {} as DatabaseWorkflow,
697 {} as DatabaseVersion,
698 { id: instanceId } as DatabaseInstance,
699 { payload: {}, timestamp: new Date(), instanceId, workflowName: "" },
700 );
701
702 const logs = (await engineStub.readLogs()) as EngineLogs;
703
704 expect(
705 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE),
706 ).toBe(true);
707 });
708
709 it("should succeed when steps are exactly at the limit", async ({
710 expect,
711 }) => {
712 const stepLimit = 3;
713 const instanceId = "STEP-LIMIT-AT-LIMIT";
714 const engineId = env.ENGINE.idFromName(instanceId);
715 const engineStub = env.ENGINE.get(engineId);
716
717 await runInDurableObject(engineStub, (engine) => {
718 engine.stepLimit = stepLimit;
719 });
720
721 setTestWorkflowCallback(async (_event, step) => {
722 for (let i = 0; i < stepLimit; i++) {
723 await step.do(`step-${i}`, async () => `result-${i}`);
724 }
725 return "done";
726 });
727
728 await engineStub.init(
729 12346,
730 {} as DatabaseWorkflow,
731 {} as DatabaseVersion,
732 { id: instanceId } as DatabaseInstance,
733 { payload: {}, timestamp: new Date(), instanceId, workflowName: "" },
734 );
735
736 const logs = (await engineStub.readLogs()) as EngineLogs;
737
738 expect(
739 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS),
740 ).toBe(true);
741 expect(
742 logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE),
743 ).toBe(false);
744 });
745
746 it("should use DEFAULT_STEP_LIMIT when no limit is configured", async ({
747 expect,
748 }) => {
749 const engineId = env.ENGINE.idFromName("STEP-LIMIT-DEFAULT");
750 const freshStub = env.ENGINE.get(engineId);
751
752 const stepLimit = await runInDurableObject(
753 freshStub,
754 (engine) => engine.stepLimit,
755 );
756
757 expect(stepLimit).toBe(DEFAULT_STEP_LIMIT);
758 });
759 });
760
761 describe("lifecycle methods", () => {
762 it.for([
763 InstanceStatus.Complete,
764 InstanceStatus.Errored,
765 InstanceStatus.Terminated,
766 ])(
767 "should throw when calling terminate on instance in finite state: %s",
768 async (finiteStatus, { expect }) => {
769 const engineStub = await runWorkflowAndAwait(
770 `TERMINATE-${finiteStatus}-INSTANCE`,
771 async () => "done",
772 );
773
774 // If not Complete, manually set the status
775 if (finiteStatus !== InstanceStatus.Complete) {
776 await runInDurableObject(engineStub, async (_engine, state) => {
777 await state.storage.put("ENGINE_STATUS", finiteStatus);
778 });
779 }
780
781 await expect(
782 runInDurableObject(engineStub, async (engine) => {
783 await engine.changeInstanceStatus("terminate");
784 }),
785 ).rejects.toThrow(
786 "Cannot terminate instance since its on a finite state",
787 );
788 },
789 );
790
791 it.for([
792 InstanceStatus.Complete,
793 InstanceStatus.Errored,
794 InstanceStatus.Terminated,
795 InstanceStatus.Running,
796 InstanceStatus.Paused,
797 ])(
798 "should restart workflow from status: %s",
799 async (initialStatus, { expect }) => {
800 const instanceId = `RESTART-${initialStatus}-INSTANCE`;
801 const engineId = env.ENGINE.idFromName(instanceId);
802
803 const engineStub = await runWorkflowAndAwait(
804 instanceId,
805 async (_event: unknown, step: WorkflowStep) => {
806 await step.do("test-step", async () => "step-result");
807 return "done";
808 },
809 );
810
811 // Set the status to initialStatus
812 await runInDurableObject(engineStub, async (_engine, state) => {
813 await state.storage.put("ENGINE_STATUS", initialStatus);
814 });
815
816 try {
817 await runInDurableObject(engineStub, async (engine) => {
818 await engine.changeInstanceStatus("restart");
819 });
820 } catch (e) {
821 // Expected - abort throws to break the DO
822 if (!isAbortError(e)) {
823 throw e;
824 }
825 }
826
827 const restartedStub = env.ENGINE.get(engineId);
828
829 await runInDurableObject(restartedStub, async (engine) => {
830 await engine.attemptRestart();
831 });
832
833 await vi.waitUntil(
834 async () => {
835 const status = await runInDurableObject(restartedStub, (engine) =>
836 engine.getStatus(),
837 );
838 return status === InstanceStatus.Complete;
839 },
840 { timeout: 5000 },
841 );
842
843 // Verify the workflow ran again by checking logs
844 const logs = (await restartedStub.readLogs()) as EngineLogs;
845
846 expect(
847 logs.logs.some((log) => log.event === InstanceEvent.WORKFLOW_START),
848 ).toBe(true);
849
850 expect(
851 logs.logs.some((log) => log.event === InstanceEvent.STEP_START),
852 ).toBe(true);
853
854 expect(
855 logs.logs.some((log) => log.event === InstanceEvent.WORKFLOW_SUCCESS),
856 ).toBe(true);
857 },
858 );
859
860 it("should restart from a specific step and preserve earlier results", async ({
861 expect,
862 }) => {
863 const instanceId = "RESTART-FROM-STEP";
864 const engineId = env.ENGINE.idFromName(instanceId);
865
866 const engineStub = await runWorkflowAndAwait(
867 instanceId,
868 async (_event: unknown, step: WorkflowStep) => {
869 const a = await step.do("step-a", async () => "result-a");
870 const b = await step.do("step-b", async () => "result-b");
871 return { a, b };
872 },
873 );
874
875 const statusBefore = await runInDurableObject(engineStub, (engine) =>
876 engine.getStatus(),
877 );
878 expect(statusBefore).toBe(InstanceStatus.Complete);
879
880 try {
881 await runInDurableObject(engineStub, async (engine) => {
882 await engine.changeInstanceStatus("restart", {
883 name: "step-b",
884 });
885 });
886 } catch (e) {
887 if (!isAbortError(e)) {
888 throw e;
889 }
890 }
891
892 const restartedStub = env.ENGINE.get(engineId);
893
894 await runInDurableObject(restartedStub, async (engine) => {
895 await engine.attemptRestart();
896 });
897
898 await vi.waitUntil(
899 async () => {
900 const status = await runInDurableObject(restartedStub, (engine) =>
901 engine.getStatus(),
902 );
903 return status === InstanceStatus.Complete;
904 },
905 { timeout: 5000 },
906 );
907
908 const logs = (await restartedStub.readLogs()) as EngineLogs;
909
910 // step-a should still have its cached result from the first run
911 const stepSuccessLogs = logs.logs.filter(
912 (log) => log.event === InstanceEvent.STEP_SUCCESS,
913 );
914 expect(stepSuccessLogs.length).toBeGreaterThanOrEqual(2);
915
916 expect(
917 logs.logs.some((log) => log.event === InstanceEvent.WORKFLOW_SUCCESS),
918 ).toBe(true);
919 });
920
921 it("should default restart from step type to step.do", async ({
922 expect,
923 }) => {
924 const instanceId = "RESTART-FROM-STEP-DEFAULT-DO";
925 const engineId = env.ENGINE.idFromName(instanceId);
926
927 const engineStub = await runWorkflowAndAwait(
928 instanceId,
929 async (_event: unknown, step: WorkflowStep) => {
930 const setup = await step.do("setup", async () => crypto.randomUUID());
931 await step.sleep("checkpoint", 1);
932 const between = await step.do("between", async () =>
933 crypto.randomUUID(),
934 );
935 const checkpoint = await step.do("checkpoint", async () =>
936 crypto.randomUUID(),
937 );
938 const after = await step.do("after", async () => crypto.randomUUID());
939 return { setup, between, checkpoint, after };
940 },
941 );
942
943 const logsBefore = (await engineStub.readLogs()) as EngineLogs;
944 const stepResultsBefore = logsBefore.logs
945 .filter((log) => log.event === InstanceEvent.STEP_SUCCESS)
946 .map((log) => log.metadata.result);
947
948 try {
949 await runInDurableObject(engineStub, async (engine) => {
950 await engine.changeInstanceStatus("restart", {
951 name: "checkpoint",
952 });
953 });
954 } catch (e) {
955 if (!isAbortError(e)) {
956 throw e;
957 }
958 }
959
960 const restartedStub = env.ENGINE.get(engineId);
961
962 await runInDurableObject(restartedStub, async (engine) => {
963 await engine.attemptRestart();
964 });
965
966 await vi.waitUntil(
967 async () => {
968 const status = await runInDurableObject(restartedStub, (engine) =>
969 engine.getStatus(),
970 );
971 return status === InstanceStatus.Complete;
972 },
973 { timeout: 5000 },
974 );
975
976 const logsAfter = (await restartedStub.readLogs()) as EngineLogs;
977 const stepResultsAfter = logsAfter.logs
978 .filter((log) => log.event === InstanceEvent.STEP_SUCCESS)
979 .map((log) => log.metadata.result);
980
981 expect(stepResultsAfter[0]).toEqual(stepResultsBefore[0]);
982 expect(stepResultsAfter[1]).toEqual(stepResultsBefore[1]);
983 expect(stepResultsAfter[2]).not.toEqual(stepResultsBefore[2]);
984 expect(stepResultsAfter[3]).not.toEqual(stepResultsBefore[3]);
985 });
986
987 it("should throw when restarting from a non-existing step", async ({
988 expect,
989 }) => {
990 const instanceId = "RESTART-FROM-BAD-STEP";
991
992 const engineStub = await runWorkflowAndAwait(
993 instanceId,
994 async (_event: unknown, step: WorkflowStep) => {
995 await step.do("step-a", async () => "result-a");
996 return "done";
997 },
998 );
999
1000 await expect(
1001 runInDurableObject(engineStub, async (engine) => {
1002 await engine.changeInstanceStatus("restart", {
1003 name: "nonexistent-step",
1004 });
1005 }),
1006 ).rejects.toThrow("not found in execution history");
1007 });
1008
1009 it("should pause after in-flight step.do finishes", async ({ expect }) => {
1010 const instanceId = "PAUSE-AFTER-DO";
1011 const engineId = env.ENGINE.idFromName(instanceId);
1012
1013 const engineStub = await runWorkflow(instanceId, async (_event, step) => {
1014 await step.do("long-step", async () => {
1015 await scheduler.wait(500);
1016 return "first";
1017 });
1018 // step-2 should never run because pause takes effect after long-step
1019 await step.do("step-2", async () => "second");
1020 return "done";
1021 });
1022
1023 // Wait for long-step to start
1024 await vi.waitUntil(
1025 async () => {
1026 return await runInDurableObject(engineStub, (engine) => {
1027 const logs = engine.readLogs() as unknown as EngineLogs;
1028 return logs.logs.some(
1029 (log) => log.event === InstanceEvent.STEP_START,
1030 );
1031 });
1032 },
1033 { timeout: 5000 },
1034 );
1035
1036 // Request pause while long-step is in flight
1037 try {
1038 await runInDurableObject(engineStub, async (engine) => {
1039 await engine.changeInstanceStatus("pause");
1040 });
1041 } catch (e) {
1042 if (!isAbortError(e)) {
1043 throw e;
1044 }
1045 }
1046
1047 await vi.waitUntil(
1048 async () => {
1049 try {
1050 return await runInDurableObject(
1051 env.ENGINE.get(engineId),
1052 async (engine) =>
1053 (await engine.getStatus()) === InstanceStatus.Paused,
1054 );
1055 } catch (e) {
1056 if (isAbortError(e)) {
1057 return false;
1058 }
1059 throw e;
1060 }
1061 },
1062 { timeout: 5000 },
1063 );
1064
1065 const freshStub = env.ENGINE.get(engineId);
1066 const finalStatus = await runInDurableObject(freshStub, (engine) =>
1067 engine.getStatus(),
1068 );
1069 expect(finalStatus).toBe(InstanceStatus.Paused);
1070
1071 // Verify long-step completed but step-2 never ran
1072 const logs = await runInDurableObject(freshStub, (engine) => {
1073 return engine.readLogs() as unknown as EngineLogs;
1074 });
1075 const stepSuccesses = logs.logs.filter(
1076 (log) => log.event === InstanceEvent.STEP_SUCCESS,
1077 );
1078 expect(stepSuccesses).toHaveLength(1);
1079 });
1080
1081 it("should pause after multiple concurrent in-flight step.dos finish", async ({
1082 expect,
1083 }) => {
1084 const instanceId = "PAUSE-AFTER-CONCURRENT-DOS";
1085 const engineId = env.ENGINE.idFromName(instanceId);
1086
1087 const engineStub = await runWorkflow(instanceId, async (_event, step) => {
1088 const [resultA, resultB] = await Promise.all([
1089 step.do("slow-step-a", async () => {
1090 await scheduler.wait(500);
1091 return "a-done";
1092 }),
1093 step.do("slow-step-b", async () => {
1094 await scheduler.wait(500);
1095 return "b-done";
1096 }),
1097 ]);
1098
1099 // This step should never run
1100 await step.do("step-after-pause", async () => "should-not-run");
1101 return { resultA, resultB };
1102 });
1103
1104 await vi.waitUntil(
1105 async () => {
1106 return await runInDurableObject(engineStub, (engine) => {
1107 const logs = engine.readLogs() as unknown as EngineLogs;
1108 return (
1109 logs.logs.filter((log) => log.event === InstanceEvent.STEP_START)
1110 .length >= 2
1111 );
1112 });
1113 },
1114 { timeout: 5000 },
1115 );
1116
1117 // Request pause while both slow steps are in flight
1118 try {
1119 await runInDurableObject(engineStub, async (engine) => {
1120 await engine.changeInstanceStatus("pause");
1121 });
1122 } catch (e) {
1123 if (!isAbortError(e)) {
1124 throw e;
1125 }
1126 }
1127
1128 await vi.waitUntil(
1129 async () => {
1130 try {
1131 return await runInDurableObject(
1132 env.ENGINE.get(engineId),
1133 async (engine) =>
1134 (await engine.getStatus()) === InstanceStatus.Paused,
1135 );
1136 } catch (e) {
1137 if (isAbortError(e)) {
1138 return false;
1139 }
1140 throw e;
1141 }
1142 },
1143 { timeout: 5000 },
1144 );
1145
1146 const freshStub = env.ENGINE.get(engineId);
1147 const finalStatus = await runInDurableObject(freshStub, (engine) =>
1148 engine.getStatus(),
1149 );
1150 expect(finalStatus).toBe(InstanceStatus.Paused);
1151
1152 // Both concurrent steps should have completed, but step-after-pause should not
1153 const logs = await runInDurableObject(freshStub, (engine) => {
1154 return engine.readLogs() as unknown as EngineLogs;
1155 });
1156 const stepSuccesses = logs.logs.filter(
1157 (log) => log.event === InstanceEvent.STEP_SUCCESS,
1158 );
1159 expect(stepSuccesses).toHaveLength(2);
1160 });
1161
1162 it("should unblock concurrent steps blocked when resume cancels pending pause", async ({
1163 expect,
1164 }) => {
1165 const instanceId = "RESUME-UNBLOCKS-WAITING-STEPS";
1166 const engineId = env.ENGINE.idFromName(instanceId);
1167
1168 const engineStub = await runWorkflow(instanceId, async (_event, step) => {
1169 const [slowResult, fastResult] = await Promise.all([
1170 step.do("slow-step", async () => {
1171 await scheduler.wait(1000);
1172 return "slow-done";
1173 }),
1174 step.do("fast-step", async () => {
1175 return "fast-done";
1176 }),
1177 ]);
1178
1179 await step.do("step-after-resume", async () => "after-resume");
1180 return { slowResult, fastResult };
1181 });
1182
1183 await vi.waitUntil(
1184 async () => {
1185 return await runInDurableObject(engineStub, (engine) => {
1186 const logs = engine.readLogs() as unknown as EngineLogs;
1187 return logs.logs.some(
1188 (log) => log.event === InstanceEvent.STEP_START,
1189 );
1190 });
1191 },
1192 { timeout: 5000 },
1193 );
1194
1195 await runInDurableObject(engineStub, async (engine) => {
1196 await engine.changeInstanceStatus("pause");
1197 });
1198
1199 await vi.waitUntil(
1200 async () =>
1201 runInDurableObject(
1202 env.ENGINE.get(engineId),
1203 async (engine) =>
1204 (await engine.getStatus()) === InstanceStatus.WaitingForPause,
1205 ),
1206 { timeout: 5000 },
1207 );
1208
1209 // Resume before slow-step finishes
1210 await runInDurableObject(engineStub, async (engine) => {
1211 await engine.changeInstanceStatus("resume");
1212 });
1213
1214 // Verify status goes back to Running
1215 const statusAfterResume = await runInDurableObject(
1216 env.ENGINE.get(engineId),
1217 (engine) => engine.getStatus(),
1218 );
1219 expect(statusAfterResume).toBe(InstanceStatus.Running);
1220
1221 await vi.waitUntil(
1222 async () =>
1223 runInDurableObject(
1224 env.ENGINE.get(engineId),
1225 async (engine) =>
1226 (await engine.getStatus()) === InstanceStatus.Complete,
1227 ),
1228 { timeout: 5000 },
1229 );
1230
1231 const freshStub = env.ENGINE.get(engineId);
1232 const finalStatus = await runInDurableObject(freshStub, (engine) =>
1233 engine.getStatus(),
1234 );
1235 expect(finalStatus).toBe(InstanceStatus.Complete);
1236
1237 // All three steps should have completed
1238 const logs = await runInDurableObject(freshStub, (engine) => {
1239 return engine.readLogs() as unknown as EngineLogs;
1240 });
1241 const stepSuccesses = logs.logs.filter(
1242 (log) => log.event === InstanceEvent.STEP_SUCCESS,
1243 );
1244 expect(stepSuccesses).toHaveLength(3);
1245 });
1246
1247 it("should pause during a step.sleep, then resume and complete", async ({
1248 expect,
1249 }) => {
1250 const instanceId = "PAUSE-DURING-SLEEP-RESUME";
1251 const engineId = env.ENGINE.idFromName(instanceId);
1252
1253 const engineStub = await runWorkflow(instanceId, async (_event, step) => {
1254 await step.do("first-step", async () => "first-done");
1255
1256 await step.sleep("short-sleep", "1 second");
1257
1258 await step.do("after-sleep", async () => "after-sleep-done");
1259 return "done";
1260 });
1261
1262 // Wait for the first step to complete (workflow is now in the sleep)
1263 await vi.waitUntil(
1264 async () => {
1265 try {
1266 return await runInDurableObject(engineStub, (engine) => {
1267 const logs = engine.readLogs() as unknown as EngineLogs;
1268 return logs.logs.some(
1269 (log) => log.event === InstanceEvent.STEP_SUCCESS,
1270 );
1271 });
1272 } catch (e) {
1273 if (isAbortError(e)) {
1274 return false;
1275 }
1276 throw e;
1277 }
1278 },
1279 { timeout: 5000 },
1280 );
1281
1282 // Request pause while in step.sleep — should pause immediately
1283 try {
1284 await runInDurableObject(engineStub, async (engine) => {
1285 await engine.changeInstanceStatus("pause");
1286 });
1287 } catch (e) {
1288 if (!isAbortError(e)) {
1289 throw e;
1290 }
1291 }
1292
1293 await vi.waitUntil(
1294 async () => {
1295 try {
1296 return await runInDurableObject(
1297 env.ENGINE.get(engineId),
1298 async (engine) =>
1299 (await engine.getStatus()) === InstanceStatus.Paused,
1300 );
1301 } catch (e) {
1302 if (isAbortError(e)) {
1303 return false;
1304 }
1305 throw e;
1306 }
1307 },
1308 { timeout: 5000 },
1309 );
1310
1311 expect(
1312 await runInDurableObject(env.ENGINE.get(engineId), (engine) =>
1313 engine.getStatus(),
1314 ),
1315 ).toBe(InstanceStatus.Paused);
1316
1317 // Only the first step should have succeeded — sleep was interrupted
1318 const logsBeforeResume = await runInDurableObject(
1319 env.ENGINE.get(engineId),
1320 (engine) => engine.readLogs() as unknown as EngineLogs,
1321 );
1322 expect(
1323 logsBeforeResume.logs.filter(
1324 (log) => log.event === InstanceEvent.STEP_SUCCESS,
1325 ),
1326 ).toHaveLength(1);
1327
1328 // Resume the workflow
1329 try {
1330 await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
1331 await engine.changeInstanceStatus("resume");
1332 });
1333 } catch (e) {
1334 if (!isAbortError(e)) {
1335 throw e;
1336 }
1337 }
1338
1339 // Wait for the workflow to complete — sleep should finish and after-sleep step should run
1340 await vi.waitUntil(
1341 async () => {
1342 try {
1343 return await runInDurableObject(
1344 env.ENGINE.get(engineId),
1345 async (engine) =>
1346 (await engine.getStatus()) === InstanceStatus.Complete,
1347 );
1348 } catch (e) {
1349 if (isAbortError(e)) {
1350 return false;
1351 }
1352 throw e;
1353 }
1354 },
1355 { timeout: 5000 },
1356 );
1357
1358 const finalLogs = await runInDurableObject(
1359 env.ENGINE.get(engineId),
1360 (engine) => engine.readLogs() as unknown as EngineLogs,
1361 );
1362 const stepSuccesses = finalLogs.logs.filter(
1363 (log) => log.event === InstanceEvent.STEP_SUCCESS,
1364 );
1365 // first-step + after-sleep = 2 step successes
1366 expect(stepSuccesses).toHaveLength(2);
1367 });
1368
1369 it("should pause during a waitForEvent, then resume and complete", async ({
1370 expect,
1371 }) => {
1372 const instanceId = "PAUSE-DURING-WAIT-FOR-EVENT";
1373 const engineId = env.ENGINE.idFromName(instanceId);
1374
1375 const engineStub = await runWorkflow(instanceId, async (_event, step) => {
1376 await step.do("first-step", async () => "first-done");
1377
1378 await step.waitForEvent("wait-for-signal", {
1379 type: "continue",
1380 timeout: "30 seconds",
1381 });
1382
1383 await step.do("after-event", async () => "after-event-done");
1384 return "done";
1385 });
1386
1387 // Wait for the first step to complete (workflow is now waiting for event)
1388 await vi.waitUntil(
1389 async () => {
1390 try {
1391 return await runInDurableObject(engineStub, (engine) => {
1392 const logs = engine.readLogs() as unknown as EngineLogs;
1393 return logs.logs.some(
1394 (log) => log.event === InstanceEvent.WAIT_START,
1395 );
1396 });
1397 } catch (e) {
1398 if (isAbortError(e)) {
1399 return false;
1400 }
1401 throw e;
1402 }
1403 },
1404 { timeout: 5000 },
1405 );
1406
1407 // Request pause while in waitForEvent
1408 try {
1409 await runInDurableObject(engineStub, async (engine) => {
1410 await engine.changeInstanceStatus("pause");
1411 });
1412 } catch (e) {
1413 if (!isAbortError(e)) {
1414 throw e;
1415 }
1416 }
1417
1418 await vi.waitUntil(
1419 async () => {
1420 try {
1421 return await runInDurableObject(
1422 env.ENGINE.get(engineId),
1423 async (engine) =>
1424 (await engine.getStatus()) === InstanceStatus.Paused,
1425 );
1426 } catch (e) {
1427 if (isAbortError(e)) {
1428 return false;
1429 }
1430 throw e;
1431 }
1432 },
1433 { timeout: 5000 },
1434 );
1435
1436 expect(
1437 await runInDurableObject(env.ENGINE.get(engineId), (engine) =>
1438 engine.getStatus(),
1439 ),
1440 ).toBe(InstanceStatus.Paused);
1441
1442 // Resume the workflow
1443 try {
1444 await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
1445 await engine.changeInstanceStatus("resume");
1446 });
1447 } catch (e) {
1448 if (!isAbortError(e)) {
1449 throw e;
1450 }
1451 }
1452
1453 // Send the event after resume
1454 await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
1455 await engine.receiveEvent({
1456 timestamp: new Date(),
1457 payload: { done: true },
1458 type: "continue",
1459 });
1460 });
1461
1462 // Wait for the workflow to complete
1463 await vi.waitUntil(
1464 async () => {
1465 try {
1466 return await runInDurableObject(
1467 env.ENGINE.get(engineId),
1468 async (engine) =>
1469 (await engine.getStatus()) === InstanceStatus.Complete,
1470 );
1471 } catch (e) {
1472 if (isAbortError(e)) {
1473 return false;
1474 }
1475 throw e;
1476 }
1477 },
1478 { timeout: 5000 },
1479 );
1480
1481 const finalLogs = await runInDurableObject(
1482 env.ENGINE.get(engineId),
1483 (engine) => engine.readLogs() as unknown as EngineLogs,
1484 );
1485 const stepSuccesses = finalLogs.logs.filter(
1486 (log) => log.event === InstanceEvent.STEP_SUCCESS,
1487 );
1488 // first-step + after-event = 2 step successes
1489 expect(stepSuccesses).toHaveLength(2);
1490 });
1491
1492 it("should transition WaitingForPause to Paused on init() entry", async ({
1493 expect,
1494 }) => {
1495 const instanceId = "WAITING-FOR-PAUSE-INIT";
1496
1497 const engineStub = await runWorkflowAndAwait(
1498 instanceId,
1499 async () => "done",
1500 );
1501
1502 // Manually set status to WaitingForPause (simulating a DO restart scenario)
1503 await runInDurableObject(engineStub, async (_engine, state) => {
1504 await state.storage.put(
1505 "ENGINE_STATUS",
1506 InstanceStatus.WaitingForPause,
1507 );
1508 });
1509
1510 // Now call init() — it should detect WaitingForPause and transition to Paused
1511 await runInDurableObject(engineStub, async (engine) => {
1512 // Reset isRunning so init() doesn't short-circuit
1513 engine.isRunning = false;
1514 await engine.init(
1515 12346,
1516 {} as DatabaseWorkflow,
1517 {} as DatabaseVersion,
1518 { id: instanceId } as DatabaseInstance,
1519 {
1520 payload: {},
1521 timestamp: new Date(),
1522 instanceId,
1523 workflowName: "",
1524 },
1525 );
1526 });
1527
1528 const status = await runInDurableObject(engineStub, (engine) =>
1529 engine.getStatus(),
1530 );
1531 expect(status).toBe(InstanceStatus.Paused);
1532 });
1533 });
1534});
1535
1536// RPC-stubbed rollback fns can't reliably mutate test-scope closures —
1537// assert via engine log events instead.
1538describe("Rollback", () => {
1539 async function readLogsAfter(
1540 stub: { readLogs(): Promise<EngineLogs> | EngineLogs },
1541 predicate: (logs: EngineLogs) => boolean,
1542 timeout = 5000,
1543 ): Promise<EngineLogs> {
1544 await vi.waitUntil(
1545 async () => predicate((await stub.readLogs()) as EngineLogs),
1546 { timeout },
1547 );
1548 return (await stub.readLogs()) as EngineLogs;
1549 }
1550
1551 function targetsOf(
1552 logs: EngineLogs,
1553 event: InstanceEvent,
1554 ): Array<string | null> {
1555 return logs.logs.filter((l) => l.event === event).map((l) => l.target);
1556 }
1557
1558 function countOf(logs: EngineLogs, event: InstanceEvent): number {
1559 return logs.logs.filter((l) => l.event === event).length;
1560 }
1561
1562 function doWithRollback<T>(
1563 step: WorkflowStep,
1564 name: string,
1565 callback: (ctx: unknown) => Promise<T>,
1566 options: WorkflowStepRollbackOptions,
1567 ): Promise<unknown>;
1568 function doWithRollback<T>(
1569 step: WorkflowStep,
1570 name: string,
1571 config: WorkflowStepConfig,
1572 callback: (ctx: unknown) => Promise<T>,
1573 options: WorkflowStepRollbackOptions,
1574 ): Promise<unknown>;
1575 function doWithRollback<T>(
1576 step: WorkflowStep,
1577 name: string,
1578 configOrCallback: WorkflowStepConfig | ((ctx: unknown) => Promise<T>),
1579 callbackOrOptions:
1580 | ((ctx: unknown) => Promise<T>)
1581 | WorkflowStepRollbackOptions,
1582 options?: WorkflowStepRollbackOptions,
1583 ): Promise<unknown> {
1584 if (typeof configOrCallback === "function") {
1585 // @ts-expect-error -- rollback options are not in workers-types yet
1586 return step.do(name, configOrCallback, callbackOrOptions);
1587 }
1588 // @ts-expect-error -- rollback options are not in workers-types yet
1589 return step.do(name, configOrCallback, callbackOrOptions, options);
1590 }
1591
1592 async function noopRollback(_ctx: RollbackContext): Promise<void> {}
1593
1594 function rollbackOptions(
1595 fn: RollbackFn = noopRollback,
1596 ): WorkflowStepRollbackOptions {
1597 return { rollback: fn };
1598 }
1599
1600 function rollbackOptionsWithConfig(
1601 rollbackConfig: WorkflowStepConfig,
1602 fn: RollbackFn = noopRollback,
1603 ): WorkflowStepRollbackOptions {
1604 return { rollback: fn, rollbackConfig };
1605 }
1606
1607 it("runs rollback fns in LIFO order on workflow failure", async ({
1608 expect,
1609 }) => {
1610 const stub = await runWorkflowAndAwait("RB-LIFO", async (_e, step) => {
1611 await doWithRollback(
1612 step,
1613 "step-1",
1614 async () => "out-1",
1615 rollbackOptions(),
1616 );
1617 await doWithRollback(
1618 step,
1619 "step-2",
1620 async () => "out-2",
1621 rollbackOptions(),
1622 );
1623 await doWithRollback(
1624 step,
1625 "step-3",
1626 async () => "out-3",
1627 rollbackOptions(),
1628 );
1629 throw new Error("boom");
1630 });
1631 const logs = await readLogsAfter(stub, (l) =>
1632 l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE),
1633 );
1634 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
1635 "step-3-1",
1636 "step-2-1",
1637 "step-1-1",
1638 ]);
1639 expect(targetsOf(logs, InstanceEvent.ROLLBACK_ATTEMPT_SUCCESS)).toEqual([
1640 "step-3-1",
1641 "step-2-1",
1642 "step-1-1",
1643 ]);
1644 expect(countOf(logs, InstanceEvent.ROLLBACK_FAILED)).toBe(0);
1645 });
1646
1647 it("runs parallel rollbacks in reverse step start order", async ({
1648 expect,
1649 }) => {
1650 let firstStarted: () => void = () => {};
1651 let firstMayFinish: () => void = () => {};
1652 const firstStartedPromise = new Promise<void>((resolve) => {
1653 firstStarted = resolve;
1654 });
1655 const firstMayFinishPromise = new Promise<void>((resolve) => {
1656 firstMayFinish = resolve;
1657 });
1658
1659 const stub = await runWorkflowAndAwait("RB-PARALLEL", async (_e, step) => {
1660 const first = doWithRollback(
1661 step,
1662 "first",
1663 async () => {
1664 firstStarted();
1665 await firstMayFinishPromise;
1666 await scheduler.wait(50);
1667 return "out-1";
1668 },
1669 rollbackOptions(),
1670 );
1671 await firstStartedPromise;
1672
1673 const second = doWithRollback(
1674 step,
1675 "second",
1676 async () => {
1677 firstMayFinish();
1678 return "out-2";
1679 },
1680 rollbackOptions(),
1681 );
1682
1683 await Promise.all([first, second]);
1684 throw new Error("boom");
1685 });
1686 const logs = await readLogsAfter(stub, (l) =>
1687 l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE),
1688 );
1689 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
1690 "second-1",
1691 "first-1",
1692 ]);
1693 });
1694
1695 it("uses rollbackConfig when executing rollback", async ({ expect }) => {
1696 const rollbackConfig = {
1697 retries: { limit: 0, delay: 0, backoff: "constant" },
1698 timeout: "30 seconds",
1699 } satisfies WorkflowStepConfig;
1700 const stub = await runWorkflowAndAwait("RB-CONFIG", async (_e, step) => {
1701 await doWithRollback(
1702 step,
1703 "configured-step",
1704 async () => "out",
1705 rollbackOptionsWithConfig(rollbackConfig),
1706 );
1707 throw new Error("boom");
1708 });
1709 const logs = await readLogsAfter(stub, (l) =>
1710 l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE),
1711 );
1712 expect(
1713 logs.logs.find((l) => l.event === InstanceEvent.ROLLBACK_STEP_START)
1714 ?.metadata,
1715 ).toMatchObject({ config: rollbackConfig });
1716 });
1717
1718 it("passes rollback context to rollback fn", async ({ expect }) => {
1719 const stub = await runWorkflowAndAwait("RB-CONTEXT", async (_e, step) => {
1720 await doWithRollback(
1721 step,
1722 "ctx-step",
1723 async () => "out",
1724 rollbackOptions(async (ctx) => {
1725 if (
1726 ctx.error.name !== "Error" ||
1727 ctx.error.message !== "boom" ||
1728 ctx.output !== "out" ||
1729 ctx.ctx.step.name !== "ctx-step" ||
1730 ctx.ctx.step.count !== 1 ||
1731 `${ctx.ctx.step.name}-${ctx.ctx.step.count}` !== "ctx-step-1"
1732 ) {
1733 throw new Error("unexpected rollback context");
1734 }
1735 }),
1736 );
1737 throw new Error("boom");
1738 });
1739 const logs = await readLogsAfter(stub, (l) =>
1740 l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE),
1741 );
1742 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
1743 "ctx-step-1",
1744 ]);
1745 });
1746
1747 it("runs rollback for failed step", async ({ expect }) => {
1748 const stub = await runWorkflowAndAwait(
1749 "RB-FAILED-STEP",
1750 async (_e, step) => {
1751 await doWithRollback(
1752 step,
1753 "failed-step",
1754 { retries: { limit: 0, delay: 0, backoff: "constant" } },
1755 async () => {
1756 throw new Error("step-boom");
1757 },
1758 rollbackOptions(async (ctx) => {
1759 if (
1760 ctx.error.message !== "step-boom" ||
1761 ctx.output !== undefined ||
1762 ctx.ctx.step.name !== "failed-step" ||
1763 ctx.ctx.step.count !== 1 ||
1764 `${ctx.ctx.step.name}-${ctx.ctx.step.count}` !== "failed-step-1"
1765 ) {
1766 throw new Error("unexpected failed-step rollback context");
1767 }
1768 }),
1769 );
1770 },
1771 );
1772 const logs = await readLogsAfter(stub, (l) =>
1773 l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE),
1774 );
1775 const rollbackStepTargets = targetsOf(
1776 logs,
1777 InstanceEvent.ROLLBACK_STEP_SUCCESS,
1778 );
1779 const stepFailureTargets = targetsOf(logs, InstanceEvent.STEP_FAILURE);
1780
1781 expect(rollbackStepTargets).toEqual(["failed-step-1"]);
1782 expect(stepFailureTargets).toContain("failed-step-1");
1783 });
1784
1785 it("only runs rollbacks for steps with a registered fn", async ({
1786 expect,
1787 }) => {
1788 const stub = await runWorkflowAndAwait("RB-PARTIAL", async (_e, step) => {
1789 await step.do("plain-step", async () => "v1");
1790 await doWithRollback(
1791 step,
1792 "step-with-rollback",
1793 async () => "v2",
1794 rollbackOptions(),
1795 );
1796 await step.do("plain-step-after", async () => "v3");
1797 throw new Error("boom");
1798 });
1799 const logs = await readLogsAfter(stub, (l) =>
1800 l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_COMPLETE),
1801 );
1802 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
1803 "step-with-rollback-1",
1804 ]);
1805 expect(
1806 logs.logs.find((l) => l.event === InstanceEvent.ROLLBACK_START)?.metadata,
1807 ).toMatchObject({ totalSteps: 1 });
1808 });
1809
1810 it("stops at the first failing rollback and logs ROLLBACK_FAILED", async ({
1811 expect,
1812 }) => {
1813 const stub = await runWorkflowAndAwait("RB-FAILS", async (_e, step) => {
1814 await doWithRollback(step, "step-1", async () => "v1", rollbackOptions());
1815 await doWithRollback(
1816 step,
1817 "step-2",
1818 async () => "v2",
1819 rollbackOptionsWithConfig(
1820 { retries: { limit: 0, delay: 0, backoff: "constant" } },
1821 async () => {
1822 throw new Error("rollback-boom");
1823 },
1824 ),
1825 );
1826 await doWithRollback(step, "step-3", async () => "v3", rollbackOptions());
1827 throw new Error("boom");
1828 });
1829 const logs = await readLogsAfter(stub, (l) =>
1830 l.logs.some((r) => r.event === InstanceEvent.ROLLBACK_FAILED),
1831 );
1832 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
1833 "step-3-1",
1834 ]);
1835 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_FAILURE)).toEqual([
1836 "step-2-1",
1837 ]);
1838 expect(targetsOf(logs, InstanceEvent.ROLLBACK_ATTEMPT_FAILURE)).toEqual([
1839 "step-2-1",
1840 ]);
1841 expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(0);
1842 });
1843
1844 it("runs rollback when terminate requests it", async ({ expect }) => {
1845 const instanceId = "RB-TERMINATE";
1846 const engineStub = await runWorkflow(instanceId, async (_e, step) => {
1847 await doWithRollback(
1848 step,
1849 "setup-resource",
1850 async () => "resource-id",
1851 rollbackOptions(),
1852 );
1853 await step.sleep("wait-forever", "1 hour");
1854 });
1855
1856 await readLogsAfter(engineStub, (currentLogs) =>
1857 currentLogs.logs.some(
1858 (log) =>
1859 log.event === InstanceEvent.STEP_SUCCESS &&
1860 log.target === "setup-resource-1",
1861 ),
1862 );
1863
1864 try {
1865 await runInDurableObject(engineStub, async (engine) => {
1866 await engine.changeInstanceStatus("terminate", undefined, {
1867 rollback: true,
1868 });
1869 });
1870 } catch (error) {
1871 if (
1872 !(error instanceof Error) ||
1873 !error.message.startsWith("Aborting engine:")
1874 ) {
1875 throw error;
1876 }
1877 }
1878
1879 const logs = await readLogsAfter(
1880 env.ENGINE.get(env.ENGINE.idFromName(instanceId)),
1881 (currentLogs) =>
1882 currentLogs.logs.some(
1883 (log) => log.event === InstanceEvent.WORKFLOW_TERMINATED,
1884 ),
1885 );
1886 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
1887 "setup-resource-1",
1888 ]);
1889 expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
1890 });
1891
1892 it("replays cached steps before terminate rollback when registry is empty", async ({
1893 expect,
1894 }) => {
1895 const instanceId = "RB-TERMINATE-REPLAY";
1896 const engineStub = await runWorkflow(instanceId, async (_e, step) => {
1897 await doWithRollback(
1898 step,
1899 "setup-resource",
1900 async () => "resource-id",
1901 rollbackOptions(),
1902 );
1903 await step.sleep("wait-forever", "1 hour");
1904 });
1905
1906 await readLogsAfter(engineStub, (currentLogs) =>
1907 currentLogs.logs.some(
1908 (log) =>
1909 log.event === InstanceEvent.STEP_SUCCESS &&
1910 log.target === "setup-resource-1",
1911 ),
1912 );
1913
1914 await runInDurableObject(engineStub, async (engine) => {
1915 engine.rollbackRegistry.clear();
1916 });
1917
1918 try {
1919 await runInDurableObject(engineStub, async (engine) => {
1920 await engine.changeInstanceStatus("terminate", undefined, {
1921 rollback: true,
1922 });
1923 });
1924 } catch (error) {
1925 if (
1926 !(error instanceof Error) ||
1927 !error.message.startsWith("Aborting engine:")
1928 ) {
1929 throw error;
1930 }
1931 }
1932
1933 const logs = await readLogsAfter(
1934 env.ENGINE.get(env.ENGINE.idFromName(instanceId)),
1935 (currentLogs) =>
1936 currentLogs.logs.some(
1937 (log) => log.event === InstanceEvent.WORKFLOW_TERMINATED,
1938 ),
1939 );
1940 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
1941 "setup-resource-1",
1942 ]);
1943 expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
1944 });
1945
1946 it("replays started unfinished steps before terminate rollback when registry is empty", async ({
1947 expect,
1948 }) => {
1949 const instanceId = `RB-TERMINATE-STARTED-REPLAY-${crypto.randomUUID()}`;
1950 const engineId = env.ENGINE.idFromName(instanceId);
1951 const engineStub = await runWorkflow(instanceId, async (_e, step) => {
1952 await doWithRollback(
1953 step,
1954 "started-setup",
1955 async () => {
1956 await scheduler.wait(30_000);
1957 return "late-resource-id";
1958 },
1959 rollbackOptions(),
1960 );
1961 });
1962
1963 await readLogsAfter(engineStub, (currentLogs) =>
1964 currentLogs.logs.some(
1965 (log) =>
1966 log.event === InstanceEvent.ATTEMPT_START &&
1967 log.target === "started-setup-1",
1968 ),
1969 );
1970
1971 try {
1972 await runInDurableObject(engineStub, async (engine) => {
1973 await engine.abort("test restart before step completion");
1974 });
1975 } catch (error) {
1976 if (
1977 !(error instanceof Error) ||
1978 error.message !== "test restart before step completion"
1979 ) {
1980 throw error;
1981 }
1982 }
1983
1984 const restartedStub = env.ENGINE.get(engineId);
1985 await runInDurableObject(restartedStub, async (engine) => {
1986 engine.rollbackRegistry.clear();
1987 });
1988
1989 try {
1990 await runInDurableObject(restartedStub, async (engine) => {
1991 await engine.changeInstanceStatus("terminate", undefined, {
1992 rollback: true,
1993 });
1994 });
1995 } catch (error) {
1996 if (!isAbortError(error)) {
1997 throw error;
1998 }
1999 }
2000
2001 let logs: EngineLogs | undefined;
2002 await vi.waitUntil(
2003 async () => {
2004 try {
2005 logs = (await env.ENGINE.get(engineId).readLogs()) as EngineLogs;
2006 return logs.logs.some(
2007 (log) => log.event === InstanceEvent.WORKFLOW_TERMINATED,
2008 );
2009 } catch (error) {
2010 if (isAbortError(error)) {
2011 return false;
2012 }
2013 throw error;
2014 }
2015 },
2016 { timeout: 5000 },
2017 );
2018 if (logs === undefined) {
2019 throw new Error("Expected workflow logs after terminate");
2020 }
2021 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
2022 "started-setup-1",
2023 ]);
2024 expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
2025 });
2026
2027 it("fails terminate rollback when replay cannot rebuild an eligible handler", async ({
2028 expect,
2029 }) => {
2030 const instanceId = `RB-TERMINATE-MISSING-REPLAY-${crypto.randomUUID()}`;
2031 const engineId = env.ENGINE.idFromName(instanceId);
2032 const engineStub = await runWorkflow(instanceId, async (_e, step) => {
2033 await doWithRollback(
2034 step,
2035 "branchy-setup",
2036 async () => "resource-id",
2037 rollbackOptions(),
2038 );
2039 await step.sleep("wait-forever", "1 hour");
2040 });
2041
2042 await readLogsAfter(engineStub, (currentLogs) =>
2043 currentLogs.logs.some(
2044 (log) =>
2045 log.event === InstanceEvent.STEP_SUCCESS &&
2046 log.target === "branchy-setup-1",
2047 ),
2048 );
2049
2050 try {
2051 await runInDurableObject(engineStub, async (engine) => {
2052 await engine.abort("test restart before replay branch changes");
2053 });
2054 } catch (error) {
2055 if (
2056 !(error instanceof Error) ||
2057 error.message !== "test restart before replay branch changes"
2058 ) {
2059 throw error;
2060 }
2061 }
2062
2063 setTestWorkflowCallback(async () => {
2064 // Simulate nondeterministic replay control flow that no longer reaches
2065 // the persisted eligible step. Production treats this as rollback failure.
2066 });
2067
2068 try {
2069 await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
2070 await engine.changeInstanceStatus("terminate", undefined, {
2071 rollback: true,
2072 });
2073 });
2074 } catch (error) {
2075 if (!isAbortError(error)) {
2076 throw error;
2077 }
2078 }
2079
2080 const logs = await readLogsAfter(env.ENGINE.get(engineId), (currentLogs) =>
2081 currentLogs.logs.some(
2082 (log) => log.event === InstanceEvent.WORKFLOW_TERMINATED,
2083 ),
2084 );
2085 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_FAILURE)).toEqual([
2086 "branchy-setup-1",
2087 ]);
2088 expect(countOf(logs, InstanceEvent.ROLLBACK_FAILED)).toBe(1);
2089 expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(0);
2090 });
2091
2092 it("replays cached failed steps before terminate rollback when registry is empty", async ({
2093 expect,
2094 }) => {
2095 const instanceId = "RB-TERMINATE-FAILED-REPLAY";
2096 const engineStub = await runWorkflow(instanceId, async (_e, step) => {
2097 try {
2098 await doWithRollback(
2099 step,
2100 "failed-setup",
2101 { retries: { limit: 0, delay: 0, backoff: "constant" } },
2102 async () => {
2103 throw new Error("step-boom");
2104 },
2105 rollbackOptions(),
2106 );
2107 } catch {
2108 // Keep the workflow running so terminate rollback can replay the cached
2109 // failed step and re-register its rollback handler.
2110 }
2111 await step.sleep("wait-forever", "1 hour");
2112 });
2113
2114 await readLogsAfter(engineStub, (currentLogs) =>
2115 currentLogs.logs.some(
2116 (log) =>
2117 log.event === InstanceEvent.STEP_FAILURE &&
2118 log.target === "failed-setup-1",
2119 ),
2120 );
2121
2122 await runInDurableObject(engineStub, async (engine) => {
2123 engine.rollbackRegistry.clear();
2124 });
2125
2126 try {
2127 await runInDurableObject(engineStub, async (engine) => {
2128 await engine.changeInstanceStatus("terminate", undefined, {
2129 rollback: true,
2130 });
2131 });
2132 } catch (error) {
2133 if (
2134 !(error instanceof Error) ||
2135 !error.message.startsWith("Aborting engine:")
2136 ) {
2137 throw error;
2138 }
2139 }
2140
2141 const logs = await readLogsAfter(
2142 env.ENGINE.get(env.ENGINE.idFromName(instanceId)),
2143 (currentLogs) =>
2144 currentLogs.logs.some(
2145 (log) => log.event === InstanceEvent.WORKFLOW_TERMINATED,
2146 ),
2147 );
2148 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
2149 "failed-setup-1",
2150 ]);
2151 expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
2152 });
2153
2154 it("omits dynamic delay from rollback ctx.config on cached-error replay", async ({
2155 expect,
2156 }) => {
2157 const dynamicDelay: WorkflowDelayFunction = () => 1000;
2158 const instanceId = "RB-TERMINATE-FAILED-REPLAY-DYNAMIC-DELAY";
2159 const engineStub = await runWorkflow(instanceId, async (_e, step) => {
2160 try {
2161 await doWithRollback(
2162 step,
2163 "failed-setup",
2164 { retries: { limit: 0, delay: dynamicDelay, backoff: "constant" } },
2165 async () => {
2166 throw new Error("step-boom");
2167 },
2168 rollbackOptions(async (ctx) => {
2169 // A dynamic delay must be omitted from the user-facing config,
2170 // never surfaced as the internal "[dynamic]" marker string.
2171 if ("delay" in ctx.ctx.config.retries) {
2172 throw new Error(
2173 "dynamic delay marker leaked into rollback ctx.config",
2174 );
2175 }
2176 }),
2177 );
2178 } catch {
2179 // Keep the workflow running so terminate rollback can replay the cached
2180 // failed step and re-register its rollback handler.
2181 }
2182 await step.sleep("wait-forever", "1 hour");
2183 });
2184
2185 await readLogsAfter(engineStub, (currentLogs) =>
2186 currentLogs.logs.some(
2187 (log) =>
2188 log.event === InstanceEvent.STEP_FAILURE &&
2189 log.target === "failed-setup-1",
2190 ),
2191 );
2192
2193 await runInDurableObject(engineStub, async (engine) => {
2194 engine.rollbackRegistry.clear();
2195 });
2196
2197 try {
2198 await runInDurableObject(engineStub, async (engine) => {
2199 await engine.changeInstanceStatus("terminate", undefined, {
2200 rollback: true,
2201 });
2202 });
2203 } catch (error) {
2204 if (
2205 !(error instanceof Error) ||
2206 !error.message.startsWith("Aborting engine:")
2207 ) {
2208 throw error;
2209 }
2210 }
2211
2212 const logs = await readLogsAfter(
2213 env.ENGINE.get(env.ENGINE.idFromName(instanceId)),
2214 (currentLogs) =>
2215 currentLogs.logs.some(
2216 (log) => log.event === InstanceEvent.WORKFLOW_TERMINATED,
2217 ),
2218 );
2219 // ROLLBACK_STEP_SUCCESS only fires if the rollback fn did not throw, i.e.
2220 // the leaked marker was stripped by toEngineStepConfig on the replay path.
2221 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
2222 "failed-setup-1",
2223 ]);
2224 expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
2225 });
2226
2227 it("runs terminate rollback while paused with an empty registry", async ({
2228 expect,
2229 }) => {
2230 const instanceId = "RB-TERMINATE-PAUSED";
2231 const engineId = env.ENGINE.idFromName(instanceId);
2232 const engineStub = await runWorkflow(instanceId, async (_e, step) => {
2233 await doWithRollback(
2234 step,
2235 "setup-resource",
2236 async () => "resource-id",
2237 rollbackOptions(),
2238 );
2239 await step.sleep("wait-forever", "1 hour");
2240 });
2241
2242 await readLogsAfter(engineStub, (currentLogs) =>
2243 currentLogs.logs.some(
2244 (log) =>
2245 log.event === InstanceEvent.STEP_SUCCESS &&
2246 log.target === "setup-resource-1",
2247 ),
2248 );
2249
2250 try {
2251 await runInDurableObject(engineStub, async (engine) => {
2252 await engine.changeInstanceStatus("pause");
2253 });
2254 } catch (error) {
2255 if (!isAbortError(error)) {
2256 throw error;
2257 }
2258 }
2259
2260 await vi.waitUntil(
2261 async () =>
2262 runInDurableObject(
2263 env.ENGINE.get(engineId),
2264 async (engine) =>
2265 (await engine.getStatus()) === InstanceStatus.Paused,
2266 ),
2267 { timeout: 5000 },
2268 );
2269
2270 await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
2271 engine.rollbackRegistry.clear();
2272 });
2273
2274 try {
2275 await runInDurableObject(env.ENGINE.get(engineId), async (engine) => {
2276 await engine.changeInstanceStatus("terminate", undefined, {
2277 rollback: true,
2278 });
2279 });
2280 } catch (error) {
2281 if (!isAbortError(error)) {
2282 throw error;
2283 }
2284 }
2285
2286 const logs = await readLogsAfter(env.ENGINE.get(engineId), (currentLogs) =>
2287 currentLogs.logs.some(
2288 (log) => log.event === InstanceEvent.WORKFLOW_TERMINATED,
2289 ),
2290 );
2291 expect(targetsOf(logs, InstanceEvent.ROLLBACK_STEP_SUCCESS)).toEqual([
2292 "setup-resource-1",
2293 ]);
2294 expect(countOf(logs, InstanceEvent.ROLLBACK_COMPLETE)).toBe(1);
2295 });
2296
2297 it("does not run rollback when workflow succeeds", async ({ expect }) => {
2298 const stub = await runWorkflowAndAwait("RB-NOOP", async (_e, step) => {
2299 await doWithRollback(step, "a", async () => "ok", rollbackOptions());
2300 return "done";
2301 });
2302 const logs = await readLogsAfter(stub, (l) =>
2303 l.logs.some((r) => r.event === InstanceEvent.WORKFLOW_SUCCESS),
2304 );
2305 expect(countOf(logs, InstanceEvent.ROLLBACK_START)).toBe(0);
2306 });
2307});