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 for non-tuple array types to match the repository convention.
4import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers";
5import { InstanceEvent, instanceStatusName } from "./instance.ts";
6import {
7 isUserTriggeredDelete,
8 isUserTriggeredPause,
9 isUserTriggeredRestart,
10 createWorkflowError,
11 isUserTriggeredTerminate,
12 WorkflowError,
13} from "./lib/errors.ts";
14import {
15 isValidAddressableWorkflowInstanceId,
16 isValidWorkflowInstanceId,
17} from "./lib/validators.ts";
18import { parseWorkflowSubscriptionOptions } from "./subscription.ts";
19import type {
20 DatabaseInstance,
21 DatabaseVersion,
22 DatabaseWorkflow,
23 Engine,
24 EngineLogs,
25} from "./engine.ts";
26import type { InstanceStatus as EngineInstanceStatus } from "./instance.ts";
27import type {
28 WorkflowSubscription,
29 WorkflowSubscriptionOptions,
30} from "./subscription.ts";
31import type {
32 WorkflowInstanceModifier,
33 WorkflowIntrospectionOperation,
34 WorkflowIntrospectionStreamResult,
35} from "./types.ts";
36
37type Env = {
38 ENGINE: DurableObjectNamespace<Engine>;
39 BINDING_NAME: string;
40 WORKFLOW_NAME: string;
41 MINIFLARE_LOOPBACK?: Fetcher;
42};
43
44/** Waits for Miniflare to finish deleting an instance's persistence files. */
45async function waitForPersistedInstanceDelete(
46 env: Env,
47 id: string | undefined,
48): Promise<void> {
49 if (id === undefined || env.MINIFLARE_LOOPBACK === undefined) {
50 return;
51 }
52
53 const hexId = env.ENGINE.idFromName(id).toString();
54 const response = await env.MINIFLARE_LOOPBACK.fetch(
55 `http://localhost/core/workflow-storage/${encodeURIComponent(env.WORKFLOW_NAME)}/${hexId}?waitForPendingDelete=1`,
56 );
57 if (!response.ok) {
58 throw new Error(
59 `Failed to wait for persisted workflow instance '${id}' deletion`,
60 );
61 }
62}
63
64/** Aborts an Engine object before removing its persistence files. */
65async function deletePersistedInstance(env: Env, id: string): Promise<void> {
66 if (env.MINIFLARE_LOOPBACK === undefined) {
67 return;
68 }
69
70 const stub = env.ENGINE.get(env.ENGINE.idFromName(id));
71 try {
72 await stub.unsafeAbort();
73 } catch {
74 // Aborting the Durable Object rejects its RPC.
75 }
76
77 const response = await env.MINIFLARE_LOOPBACK.fetch(
78 `http://localhost/core/workflow-storage/${encodeURIComponent(env.WORKFLOW_NAME)}/${stub.id.toString()}?defer=1`,
79 { method: "DELETE" },
80 );
81 if (!response.ok) {
82 throw new Error(`Failed to delete persisted workflow instance '${id}'`);
83 }
84 await waitForPersistedInstanceDelete(env, id);
85}
86
87type WorkflowIntrospectionSession = {
88 id: string;
89 operations: Array<WorkflowIntrospectionOperation>;
90 instanceIds: Array<string>;
91};
92
93// workerd may construct a fresh WorkflowBinding object for each RPC call. Store
94// sessions at module scope so start/modify/get/dispose calls, and later
95// WorkflowBinding.create() calls, all see the same active Workflow session.
96const workflowIntrospectionSessions = new Map<
97 string,
98 WorkflowIntrospectionSession
99>();
100
101function getWorkflowIntrospectionSession(
102 workflowName: string,
103 sessionId: string,
104): WorkflowIntrospectionSession {
105 const session = workflowIntrospectionSessions.get(workflowName);
106 if (session?.id !== sessionId) {
107 throw new Error(
108 `Workflow ${JSON.stringify(workflowName)} does not have an active introspection session for this introspector.`,
109 );
110 }
111 return session;
112}
113
114function isWorkflowIntrospectionStreamResult(
115 value: unknown,
116): value is WorkflowIntrospectionStreamResult {
117 return (
118 value !== null &&
119 typeof value === "object" &&
120 "__workflowIntrospectionStreamResult" in value &&
121 "chunks" in value &&
122 value.__workflowIntrospectionStreamResult === true &&
123 Array.isArray(value.chunks)
124 );
125}
126
127function createWorkflowIntrospectionReadableStream(
128 result: WorkflowIntrospectionStreamResult,
129): ReadableStream<Uint8Array> {
130 return new ReadableStream<Uint8Array>({
131 start(controller) {
132 for (const chunk of result.chunks) {
133 controller.enqueue(chunk.slice());
134 }
135 controller.close();
136 },
137 });
138}
139
140async function applyWorkflowIntrospectionOperation(
141 modifier: WorkflowInstanceModifier,
142 operation: WorkflowIntrospectionOperation,
143) {
144 switch (operation.type) {
145 case "disableSleeps":
146 await modifier.disableSleeps(operation.steps);
147 break;
148 case "disableRetryDelays":
149 await modifier.disableRetryDelays(operation.steps);
150 break;
151 case "mockStepResult":
152 await modifier.mockStepResult(
153 operation.step,
154 isWorkflowIntrospectionStreamResult(operation.stepResult)
155 ? createWorkflowIntrospectionReadableStream(operation.stepResult)
156 : operation.stepResult,
157 );
158 break;
159 case "mockStepError": {
160 const error = new Error(operation.error.message);
161 error.name = operation.error.name;
162 await modifier.mockStepError(operation.step, error, operation.times);
163 break;
164 }
165 case "forceStepTimeout":
166 await modifier.forceStepTimeout(operation.step, operation.times);
167 break;
168 case "mockEvent":
169 await modifier.mockEvent(operation.event);
170 break;
171 case "forceEventTimeout":
172 await modifier.forceEventTimeout(operation.step);
173 break;
174 }
175}
176
177// TODO(vaish): import from @cloudflare/workers-types once restart options are published
178export interface RestartFromStep {
179 name: string;
180 count?: number;
181 type?: "do" | "sleep" | "waitForEvent";
182}
183
184export interface WorkflowInstanceTerminateOptions {
185 /**
186 * If true, run registered rollback handlers before terminating the instance.
187 */
188 rollback?: boolean;
189}
190
191export interface WorkflowInstanceRestartOptions {
192 from?: RestartFromStep;
193}
194
195// this.env.WORKFLOW is WorkflowBinding
196export class WorkflowBinding extends WorkerEntrypoint<Env> {
197 constructor(ctx: ExecutionContext, env: Env) {
198 super(ctx, env);
199 }
200
201 public async create({
202 id = crypto.randomUUID(),
203 params = {},
204 }: WorkflowInstanceCreateOptions = {}): Promise<{
205 id: string;
206 }> {
207 if (!isValidWorkflowInstanceId(id)) {
208 throw new WorkflowError("Workflow instance has invalid id");
209 }
210
211 await waitForPersistedInstanceDelete(this.env, id);
212 const stubId = this.env.ENGINE.idFromName(id);
213 const stub = this.env.ENGINE.get(stubId);
214 const introspectionSession = workflowIntrospectionSessions.get(
215 this.env.WORKFLOW_NAME,
216 );
217
218 if (introspectionSession !== undefined) {
219 const modifier = stub.getInstanceModifier();
220 introspectionSession.instanceIds.push(id);
221 for (const operation of introspectionSession.operations) {
222 await applyWorkflowIntrospectionOperation(modifier, operation);
223 }
224 }
225
226 const now = new Date().toISOString();
227 const initPromise = stub
228 .init(
229 0, // accountId: number,
230 {} as DatabaseWorkflow, // workflow: DatabaseWorkflow,
231 {} as DatabaseVersion, // version: DatabaseVersion,
232 {
233 id,
234 created_on: now,
235 modified_on: now,
236 workflow_id: "",
237 version_id: "",
238 status: 0, // InstanceStatus.Queued
239 started_on: now,
240 ended_on: null,
241 } satisfies DatabaseInstance,
242 {
243 timestamp: new Date(),
244 payload: params as Readonly<typeof params>,
245 instanceId: id,
246 workflowName: this.env.WORKFLOW_NAME,
247 },
248 )
249 .then((val) => {
250 if (val !== undefined) {
251 val[Symbol.dispose]();
252 }
253 })
254 .catch(() => {
255 // Suppress all rejections: create() should queue and
256 // return immediately
257 });
258
259 this.ctx.waitUntil(initPromise);
260
261 return {
262 id,
263 };
264 }
265
266 public async get(id: string): Promise<WorkflowInstance> {
267 const stubId = this.env.ENGINE.idFromName(id);
268 const stub = this.env.ENGINE.get(stubId);
269
270 // Pass a getter function so WorkflowHandle can get a fresh stub after abort
271 const getStub = () => this.env.ENGINE.get(this.env.ENGINE.idFromName(id));
272
273 const handle = new WorkflowHandle(id, stub, getStub);
274
275 try {
276 await handle.status();
277 } catch {
278 throw new Error("instance.not_found");
279 }
280
281 return handle;
282 }
283
284 public async createBatch(
285 batch: Array<WorkflowInstanceCreateOptions<unknown>>,
286 ): Promise<Array<{ id: string }>> {
287 if (batch.length === 0) {
288 throw new Error(
289 "WorkflowError: batchCreate should have at least 1 instance",
290 );
291 }
292
293 return await Promise.all(
294 batch.map(async (val) => {
295 const res = await this.create(val);
296 return res;
297 }),
298 );
299 }
300
301 /**
302 * Deletes an instance. Named `deleteInstance` because `Fetcher.delete()` shadows
303 * a same-named JSRPC method.
304 */
305 public async deleteInstance(id: string): Promise<void> {
306 if (!isValidAddressableWorkflowInstanceId(id)) {
307 throw createWorkflowError(
308 "Instance ID is invalid",
309 "instance.invalid_id",
310 );
311 }
312
313 const stub = this.env.ENGINE.get(this.env.ENGINE.idFromName(id));
314 try {
315 await stub.deleteInstance();
316 } catch (error) {
317 // delete aborts the instance
318 if (!isUserTriggeredDelete(error)) {
319 throw error;
320 }
321 }
322 await waitForPersistedInstanceDelete(this.env, id);
323 }
324
325 /** Deletes each unique instance once while preserving duplicate results. */
326 public async deleteBatch(options: {
327 instances: Array<string>;
328 }): Promise<WorkflowBatchDeleteResult> {
329 const instanceIds = options?.instances;
330 if (!Array.isArray(instanceIds)) {
331 throw createWorkflowError("Provided argument is invalid", "body");
332 }
333 if (instanceIds.length > 100) {
334 throw createWorkflowError(
335 "batchDeleteInstances only supports 100 instances at a time",
336 "body",
337 );
338 }
339 if (instanceIds.length === 0) {
340 throw createWorkflowError(
341 "batchDeleteInstances should have at least 1 instance",
342 "body",
343 );
344 }
345 if (!instanceIds.every(isValidAddressableWorkflowInstanceId)) {
346 throw createWorkflowError(
347 "Instance ID is invalid",
348 "instance.invalid_id",
349 );
350 }
351
352 const uniqueIds = [...new Set(instanceIds)];
353 const settled = await Promise.allSettled(
354 uniqueIds.map((id) => this.deleteInstance(id)),
355 );
356 const resultsById = new Map(
357 uniqueIds.map((id, index) => [id, settled[index]]),
358 );
359 const result: WorkflowBatchDeleteResult = { deleted: [], errors: [] };
360 for (const id of instanceIds) {
361 const deletion = resultsById.get(id);
362 if (deletion === undefined) {
363 throw new Error("Missing batch deletion result");
364 }
365 if (
366 deletion.status === "fulfilled" ||
367 isUserTriggeredDelete(deletion.reason)
368 ) {
369 result.deleted.push({ id });
370 continue;
371 }
372
373 const isNotFound =
374 deletion.reason instanceof Error &&
375 deletion.reason.message.includes("(instance.not_found)");
376 result.errors.push({
377 id,
378 code: isNotFound ? 10400 : 10001,
379 message: isNotFound
380 ? "workflows.api.error.instance.not_found"
381 : "workflows.api.error.internal_server",
382 });
383 }
384
385 const missingIds = new Set(
386 result.errors.filter(({ code }) => code === 10400).map(({ id }) => id),
387 );
388 const cleanupIds = [...missingIds];
389 const cleanups = await Promise.allSettled(
390 cleanupIds.map((id) => deletePersistedInstance(this.env, id)),
391 );
392 const failedCleanupIds = new Set(
393 cleanupIds.filter((_, index) => cleanups[index]?.status === "rejected"),
394 );
395 if (failedCleanupIds.size === 0) {
396 return result;
397 }
398
399 const errorsById = new Map(result.errors.map((error) => [error.id, error]));
400 return {
401 deleted: result.deleted.filter(({ id }) => !failedCleanupIds.has(id)),
402 errors: instanceIds.flatMap((id) => {
403 if (failedCleanupIds.has(id)) {
404 return [
405 { id, code: 10001, message: "workflows.api.error.internal_server" },
406 ];
407 }
408 const error = errorsById.get(id);
409 return error === undefined ? [] : [error];
410 }),
411 };
412 }
413
414 public async unsafeGetBindingName(): Promise<string> {
415 // async because of rpc
416 return this.env.BINDING_NAME;
417 }
418
419 public async unsafeStartIntrospection(): Promise<string> {
420 if (workflowIntrospectionSessions.has(this.env.WORKFLOW_NAME)) {
421 throw new Error(
422 `Workflow ${JSON.stringify(this.env.WORKFLOW_NAME)} already has an active introspection session for binding ${JSON.stringify(this.env.BINDING_NAME)}.`,
423 );
424 }
425
426 const sessionId = crypto.randomUUID();
427 workflowIntrospectionSessions.set(this.env.WORKFLOW_NAME, {
428 id: sessionId,
429 operations: [],
430 instanceIds: [],
431 });
432 return sessionId;
433 }
434
435 public async unsafeStopIntrospection(sessionId: string): Promise<void> {
436 const session = workflowIntrospectionSessions.get(this.env.WORKFLOW_NAME);
437 if (session?.id === sessionId) {
438 workflowIntrospectionSessions.delete(this.env.WORKFLOW_NAME);
439 }
440 }
441
442 public async unsafeSetIntrospectionOperations(
443 sessionId: string,
444 operations: Array<WorkflowIntrospectionOperation>,
445 ): Promise<void> {
446 const session = getWorkflowIntrospectionSession(
447 this.env.WORKFLOW_NAME,
448 sessionId,
449 );
450 session.operations = operations;
451 }
452
453 public async unsafeGetIntrospectionInstances(
454 sessionId: string,
455 ): Promise<Array<string>> {
456 return getWorkflowIntrospectionSession(this.env.WORKFLOW_NAME, sessionId)
457 .instanceIds;
458 }
459
460 public async unsafeGetInstanceModifier(instanceId: string): Promise<unknown> {
461 // async because of rpc
462 const stubId = this.env.ENGINE.idFromName(instanceId);
463 const stub = this.env.ENGINE.get(stubId);
464
465 const instanceModifier = stub.getInstanceModifier();
466
467 return instanceModifier;
468 }
469
470 public async unsafeWaitForStepResult(
471 instanceId: string,
472 name: string,
473 index?: number,
474 ): Promise<unknown> {
475 const stubId = this.env.ENGINE.idFromName(instanceId);
476 const stub = this.env.ENGINE.get(stubId);
477
478 return await stub.waitForStepResult(name, index);
479 }
480
481 public async unsafeAbort(instanceId: string, reason?: string): Promise<void> {
482 const stubId = this.env.ENGINE.idFromName(instanceId);
483 const stub = this.env.ENGINE.get(stubId);
484
485 try {
486 await stub.unsafeAbort(reason);
487 } catch {
488 // do nothing because we want to dispose this instance
489 }
490 }
491
492 public async unsafeWaitForStatus(
493 instanceId: string,
494 status: string,
495 ): Promise<void> {
496 const stubId = this.env.ENGINE.idFromName(instanceId);
497 const stub = this.env.ENGINE.get(stubId);
498 return await stub.waitForStatus(status);
499 }
500
501 public async unsafeGetOutputOrError(
502 instanceId: string,
503 isOutput: boolean,
504 ): Promise<unknown> {
505 const stubId = this.env.ENGINE.idFromName(instanceId);
506 const stub = this.env.ENGINE.get(stubId);
507 return await stub.getOutputOrError(isOutput);
508 }
509}
510
511export class WorkflowHandle extends RpcTarget implements WorkflowInstance {
512 private stub: DurableObjectStub<Engine>;
513
514 constructor(
515 public id: string,
516 stub: DurableObjectStub<Engine>,
517 private getStub: () => DurableObjectStub<Engine>,
518 ) {
519 super();
520 this.stub = stub;
521 }
522
523 public async pause(): Promise<void> {
524 try {
525 await this.stub.changeInstanceStatus("pause");
526 } catch (e) {
527 // pause causes instance abortion
528 if (!isUserTriggeredPause(e)) {
529 throw e;
530 }
531 }
532 }
533
534 public async resume(): Promise<void> {
535 await this.stub.changeInstanceStatus("resume");
536 }
537
538 public async terminate(
539 options?: WorkflowInstanceTerminateOptions,
540 ): Promise<void> {
541 try {
542 await this.stub.changeInstanceStatus("terminate", undefined, options);
543 } catch (e) {
544 // terminate causes instance abortion
545 if (!isUserTriggeredTerminate(e)) {
546 throw e;
547 }
548 }
549 }
550
551 public async delete(): Promise<void> {
552 try {
553 await this.stub.deleteInstance();
554 } catch (e) {
555 // delete aborts the instance
556 if (!isUserTriggeredDelete(e)) {
557 throw e;
558 }
559 }
560 }
561
562 public async restart(
563 options?: WorkflowInstanceRestartOptions,
564 ): Promise<void> {
565 try {
566 await this.stub.changeInstanceStatus("restart", options?.from);
567 } catch (e) {
568 // restart causes instance abortion
569 if (!isUserTriggeredRestart(e)) {
570 throw e;
571 }
572 }
573
574 // trigger restart flow after abortion
575 this.stub = this.getStub();
576 await this.stub.attemptRestart();
577 }
578
579 public async status(): Promise<
580 InstanceStatus & { __LOCAL_DEV_STEP_OUTPUTS: Array<unknown> }
581 > {
582 // Both getStatus() and readLogs() must use the same fresh stub.
583 // After pause/restart/terminate aborts the DO, the stub goes stale
584 const fetchStatusAndLogs = async () => {
585 const status = await this.stub.getStatus();
586
587 // NOTE(lduarte): for some reason, sync functions over RPC are typed as never instead of Promise<EngineLogs>
588 const logs = await (this.stub.readLogs() as unknown as Promise<
589 EngineLogs & Disposable
590 >);
591
592 return { status, logs };
593 };
594
595 let result: {
596 status: EngineInstanceStatus;
597 logs: EngineLogs & Disposable;
598 };
599 try {
600 result = await fetchStatusAndLogs();
601 } catch {
602 this.stub = this.getStub();
603 result = await fetchStatusAndLogs();
604 }
605 // Dispose the RPC handle when the method scope exits
606 using logs = result.logs;
607
608 const filteredLogs = logs.logs.filter(
609 (log) =>
610 log.event === InstanceEvent.STEP_SUCCESS ||
611 log.event === InstanceEvent.WAIT_COMPLETE,
612 );
613
614 const stepOutputs = filteredLogs.map((log) =>
615 log.event === InstanceEvent.STEP_SUCCESS
616 ? log.metadata.result
617 : log.metadata.payload,
618 );
619
620 const workflowOutput =
621 logs.logs.find((log) => log.event === InstanceEvent.WORKFLOW_SUCCESS)
622 ?.metadata.result ?? null;
623
624 const workflowError = logs.logs.find(
625 (log) => log.event === InstanceEvent.WORKFLOW_FAILURE,
626 )?.metadata.error;
627
628 return {
629 status: instanceStatusName(result.status),
630 __LOCAL_DEV_STEP_OUTPUTS: stepOutputs,
631 output: workflowOutput,
632 error: workflowError,
633 };
634 }
635
636 public async subscribe(
637 options?: WorkflowSubscriptionOptions,
638 ): Promise<WorkflowSubscription> {
639 const parsedOptions = parseWorkflowSubscriptionOptions(options);
640 return this.stub.subscribe(parsedOptions);
641 }
642
643 public async sendEvent(args: {
644 payload: unknown;
645 type: string;
646 }): Promise<void> {
647 await this.stub.receiveEvent({
648 payload: args.payload,
649 type: args.type,
650 timestamp: new Date(),
651 });
652 }
653}