1// Alchemy modifications are licensed under Apache-2.0.
2// This file includes third-party code; see /THIRD_PARTY_LICENSES.md.
3import { InstanceEvent } from "../instance.ts";
4import { WorkflowFatalError } from "./errors.ts";
5import { isValidStepConfig } from "./validators.ts";
6import type { WorkflowStepContext } from "../context.ts";
7import type { Engine } from "../engine.ts";
8import type { WorkflowStepConfig } from "cloudflare:workers";
9
10type UserErrorField = {
11 isUserError?: boolean;
12};
13
14// `:` can't appear in user-step cacheKeys (sha1-hex + `-` only).
15export const ROLLBACK_CACHE_KEY_PREFIX = "rollback:";
16
17export type RollbackContext = {
18 ctx: WorkflowStepContext;
19 error: Error;
20 output: unknown;
21 /** @deprecated Use `${ctx.step.name}-${ctx.step.count}` instead. */
22 stepName: string;
23};
24
25// dup() to outlive the originating step.do call; Symbol.dispose locally
26// (calling `.dispose()` would RPC to a non-existent remote method).
27export type RollbackFn = ((ctx: RollbackContext) => Promise<void>) & {
28 dup?: () => RollbackFn;
29 [Symbol.dispose]?: () => void;
30};
31
32export type WorkflowStepRollbackConfig = Pick<
33 WorkflowStepConfig,
34 "retries" | "timeout"
35>;
36
37export type WorkflowStepRollbackOptions = {
38 rollback: RollbackFn;
39 rollbackConfig?: WorkflowStepRollbackConfig;
40};
41
42export type RollbackRegistryEntry = {
43 fn: RollbackFn;
44 stepContext: WorkflowStepContext;
45 output?: unknown;
46 config?: WorkflowStepConfig;
47};
48
49export type RollbackRegistration = RollbackRegistryEntry & {
50 cacheKey: string;
51};
52
53export function parseRollbackOptions(
54 stepName: string,
55 options: unknown
56): WorkflowStepRollbackOptions | undefined {
57 if (options === undefined) {
58 return undefined;
59 }
60
61 if (
62 typeof options !== "object" ||
63 options === null ||
64 Array.isArray(options)
65 ) {
66 const error = new WorkflowFatalError(
67 `Rollback options for "${stepName}" must be an object`
68 ) as Error & UserErrorField;
69 error.isUserError = true;
70 throw error;
71 }
72
73 const rollbackOptions = options as Partial<WorkflowStepRollbackOptions>;
74 if (typeof rollbackOptions.rollback !== "function") {
75 const error = new WorkflowFatalError(
76 `Rollback for "${stepName}" must be a function`
77 ) as Error & UserErrorField;
78 error.isUserError = true;
79 throw error;
80 }
81
82 if (
83 rollbackOptions.rollbackConfig !== undefined &&
84 !isValidStepConfig(rollbackOptions.rollbackConfig)
85 ) {
86 const error = new WorkflowFatalError(
87 `Rollback config for "${stepName}" is in a invalid format. See https://developers.cloudflare.com/workflows/build/sleeping-and-retrying/`
88 ) as Error & UserErrorField;
89 error.isUserError = true;
90 throw error;
91 }
92
93 return rollbackOptions as WorkflowStepRollbackOptions;
94}
95
96export function dupRollbackStub(fn: RollbackFn): RollbackFn {
97 return fn.dup ? fn.dup() : fn;
98}
99
100export function disposeRollbackStub(fn: RollbackFn): void {
101 try {
102 fn[Symbol.dispose]?.();
103 } catch (err) {
104 console.warn("Failed to dispose rollback stub", err);
105 }
106}
107
108export function registerRollbackFn(
109 registry: Map<string, RollbackRegistryEntry>,
110 registration: RollbackRegistration
111): void {
112 const { cacheKey, fn, stepContext, output, config } = registration;
113 const existing = registry.get(cacheKey);
114 if (existing) {
115 // Existing entry already owns the duped rollback stub. Duplicate registrations
116 // refresh context/output only; this helper has not duped the incoming fn.
117 registry.set(cacheKey, {
118 ...existing,
119 stepContext,
120 ...("output" in registration && { output }),
121 });
122 return;
123 }
124 registry.set(cacheKey, {
125 fn: dupRollbackStub(fn),
126 stepContext,
127 ...("output" in registration && { output }),
128 ...(config !== undefined && { config }),
129 });
130}
131
132export function clearRollbackRegistry(
133 registry: Map<string, RollbackRegistryEntry>
134): void {
135 for (const entry of registry.values()) {
136 disposeRollbackStub(entry.fn);
137 }
138 registry.clear();
139}
140
141// LIFO; halts on first failure. Goes through Context.do so each rollback
142// inherits retries/timeouts/attempt-logging.
143export async function executeRollbacks(
144 engine: Engine,
145 triggerError: Error
146): Promise<{ ranAny: boolean; allSucceeded: boolean }> {
147 const eligibleSteps = engine.readEligibleRollbackStepsDesc();
148 if (eligibleSteps.length === 0) {
149 clearRollbackRegistry(engine.rollbackRegistry);
150 return { ranAny: false, allSucceeded: true };
151 }
152
153 engine.writeLog(InstanceEvent.ROLLBACK_START, null, null, {
154 triggerError: { name: triggerError.name, message: triggerError.message },
155 totalSteps: eligibleSteps.length,
156 });
157
158 let allSucceeded = true;
159 let completed = 0;
160
161 try {
162 for (const step of eligibleSteps) {
163 const entry = engine.rollbackRegistry.get(step.cacheKey);
164 if (entry === undefined) {
165 engine.writeLog(
166 InstanceEvent.ROLLBACK_STEP_FAILURE,
167 step.cacheKey,
168 step.target,
169 {
170 error: {
171 name: "RollbackMissing",
172 message: "Rollback function not available in registry",
173 },
174 }
175 );
176 allSucceeded = false;
177 break;
178 }
179
180 const ctx = engine.createRollbackContext({ cacheKey: step.cacheKey });
181 try {
182 await ctx.do(step.target, entry.config ?? {}, async () => {
183 await entry.fn({
184 ctx: structuredClone(entry.stepContext),
185 error: triggerError,
186 output: entry.output,
187 stepName: step.target,
188 });
189 });
190 completed++;
191 } catch {
192 // Context.do already wrote ROLLBACK_STEP_FAILURE; halt the chain.
193 allSucceeded = false;
194 break;
195 } finally {
196 disposeRollbackStub(entry.fn);
197 engine.rollbackRegistry.delete(step.cacheKey);
198 }
199 }
200 } finally {
201 clearRollbackRegistry(engine.rollbackRegistry);
202 }
203
204 engine.writeLog(
205 allSucceeded
206 ? InstanceEvent.ROLLBACK_COMPLETE
207 : InstanceEvent.ROLLBACK_FAILED,
208 null,
209 null,
210 { totalSteps: eligibleSteps.length, completedSteps: completed }
211 );
212 return { ranAny: completed > 0, allSucceeded };
213}