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 { DurableObject } from "cloudflare:workers";
5import { Context, REDACTED_STEP_OUTPUT } from "./context.ts";
6import {
7 INSTANCE_METADATA,
8 InstanceEvent,
9 InstanceStatus,
10 instanceStatusName,
11 InstanceTrigger,
12 toInstanceStatus,
13} from "./instance.ts";
14import { computeHash } from "./lib/cache.ts";
15import {
16 ABORT_REASONS,
17 createWorkflowError,
18 isAbortError,
19 PreservedNonRetryableError,
20 shouldPreserveNonRetryableError,
21 stepNotFoundError,
22 WorkflowFatalError,
23} from "./lib/errors.ts";
24import {
25 ENGINE_TIMEOUT,
26 GracePeriodSemaphore,
27 startGracePeriod,
28} from "./lib/gracePeriodSemaphore.ts";
29import {
30 readAndClearRestartFromStep,
31 resolveGroupKeysToWipe,
32 storeRestartFromStep,
33 wipeRestartState,
34} from "./lib/restart.ts";
35import {
36 clearRollbackRegistry,
37 disposeRollbackStub,
38 executeRollbacks,
39 registerRollbackFn,
40 ROLLBACK_CACHE_KEY_PREFIX,
41} from "./lib/rollback.ts";
42import { normalizeForStorage } from "./lib/serialization.ts";
43import {
44 createReplayReadableStream,
45 getInvalidStoredStreamOutputError,
46 getStoredStreamOutputPreview,
47 getStreamOutputMetaKey,
48 StreamOutputState,
49} from "./lib/streams.ts";
50import { TimePriorityQueue } from "./lib/timePriorityQueue.ts";
51import { MODIFIER_KEYS, WorkflowInstanceModifier } from "./modifier.ts";
52import {
53 isTerminalEvent,
54 parseResolvedStepConfig,
55 WorkflowSubscriptionTarget,
56} from "./subscription.ts";
57import type {
58 RestartFromStep,
59 WorkflowInstanceTerminateOptions,
60} from "./binding.ts";
61import type { Event } from "./context.ts";
62import type { InstanceMetadata, RawInstanceLog } from "./instance.ts";
63import type {
64 RollbackRegistration,
65 RollbackRegistryEntry,
66} from "./lib/rollback.ts";
67import type { StreamOutputMeta } from "./lib/streams.ts";
68import type {
69 WorkflowSubscriptionEvent,
70 WorkflowSubscriptionOptions,
71 WorkflowSubscriptionState,
72} from "./subscription.ts";
73import type {
74 WorkflowEntrypoint,
75 WorkflowEvent,
76 WorkflowStep,
77} from "cloudflare:workers";
78
79interface Env {
80 ENGINE: DurableObjectNamespace<Engine>;
81 USER_WORKFLOW: WorkflowEntrypoint;
82 MINIFLARE_LOOPBACK?: Fetcher;
83 WORKFLOW_NAME?: string;
84 STEP_LIMIT?: string; // JSON-encoded number from miniflare binding
85}
86
87export type DatabaseWorkflow = {
88 name: string;
89 id: string;
90 created_on: string;
91 modified_on: string;
92 script_name: string;
93 class_name: string | null;
94 triggered_on: string | null;
95};
96
97export type DatabaseVersion = {
98 id: string;
99 class_name: string;
100 created_on: string;
101 modified_on: string;
102 workflow_id: string;
103 mutable_pipeline_id: string;
104};
105
106export type DatabaseInstance = {
107 id: string;
108 created_on: string;
109 modified_on: string;
110 workflow_id: string;
111 version_id: string;
112 status: InstanceStatus;
113 started_on: string | null;
114 ended_on: string | null;
115};
116
117export type Log = {
118 event: InstanceEvent;
119 group: string | null;
120 target: string | null;
121 metadata: {
122 result: string;
123 payload: string;
124 error: { name: string; message: string };
125 };
126};
127
128export type EngineLogs = {
129 logs: Array<Log>;
130};
131
132const ENGINE_STATUS_KEY = "ENGINE_STATUS";
133const WORKFLOW_OUTPUT_KEY = "WORKFLOW_OUTPUT";
134
135const EVENT_MAP_PREFIX = "EVENT_MAP";
136
137export const DEFAULT_STEP_LIMIT = 10_000;
138
139const PAUSE_DATETIME = "PAUSE_DATETIME";
140
141export type RollbackPhase = "replay" | "rollback";
142
143/**
144 * JSON.stringify replacer that converts TypedArrays and ArrayBuffers to a
145 * human-readable description. Without this, JSON.stringify(Uint8Array) encodes
146 * each byte as a numeric key ({"0":1,"1":2,...}), producing a string ~10x larger
147 * than byteLength and causing SQLITE_TOOBIG for outputs above ~170 KB.
148 * The replacer is called recursively by JSON.stringify, so nested binary values
149 * inside objects or arrays are also handled.
150 */
151function binaryReplacer(_key: string, value: unknown): unknown {
152 if (typeof value === "bigint") {
153 return `[BigInt(${value})]`;
154 }
155 if (value instanceof ArrayBuffer) {
156 return `[ArrayBuffer(${value.byteLength} bytes)]`;
157 }
158 if (ArrayBuffer.isView(value) && !(value instanceof DataView)) {
159 return `[${value.constructor.name}(${(value as ArrayBufferView).byteLength} bytes)]`;
160 }
161 return value;
162}
163
164function isStepSuccessEvent(event: InstanceEvent): boolean {
165 return (
166 event === InstanceEvent.STEP_SUCCESS ||
167 event === InstanceEvent.ROLLBACK_STEP_SUCCESS
168 );
169}
170
171type EngineWorkflowSubscriptionEvent =
172 | WorkflowSubscriptionEvent
173 | (Pick<WorkflowSubscriptionEvent, "instanceId" | "eventId" | "timestamp"> & {
174 type: "internal";
175 });
176
177async function readStepConfig(
178 storage: DurableObjectStorage,
179 groupKey: string | null,
180) {
181 if (groupKey === null) {
182 return undefined;
183 }
184 return parseResolvedStepConfig(await storage.get(`${groupKey}-config`));
185}
186
187async function readStepCompletedOutput(
188 storage: DurableObjectStorage,
189 groupKey: string | null,
190) {
191 if (groupKey === null) {
192 return undefined;
193 }
194
195 const valueKey = `${groupKey}-value`;
196 const streamMetaKey = getStreamOutputMetaKey(groupKey);
197 const configKey = `${groupKey}-config`;
198 const stored = await storage.get([valueKey, streamMetaKey, configKey]);
199 const config = parseResolvedStepConfig(stored.get(configKey));
200 if (config === undefined) {
201 return undefined;
202 }
203 if (config.sensitive === "output") {
204 return REDACTED_STEP_OUTPUT;
205 }
206
207 const streamMeta = stored.get(streamMetaKey) as StreamOutputMeta | undefined;
208 if (streamMeta?.state === StreamOutputState.Complete) {
209 const integrityError = getInvalidStoredStreamOutputError(
210 storage,
211 groupKey,
212 streamMeta,
213 );
214 if (integrityError !== undefined) {
215 throw createWorkflowError(
216 "Step has completed but its stored stream output is corrupt or incomplete",
217 "instance.step_output_corrupt",
218 );
219 }
220 return createReplayReadableStream({
221 storage,
222 cacheKey: groupKey,
223 meta: streamMeta,
224 });
225 }
226
227 return (stored.get(valueKey) as { value: unknown } | undefined)?.value;
228}
229
230async function buildWorkflowSubscriptionEvent(
231 storage: DurableObjectStorage,
232 log: RawInstanceLog,
233 instanceId: string,
234 params: unknown,
235): Promise<EngineWorkflowSubscriptionEvent> {
236 const common = {
237 instanceId,
238 eventId: log.id,
239 timestamp: new Date(log.timestamp).valueOf(),
240 };
241 const parseMetadata = () =>
242 JSON.parse(log.metadata) as {
243 result?: unknown;
244 error: { name: string; message: string };
245 attempt: number;
246 retryDelayMs?: number;
247 durationMs: number;
248 event: string;
249 };
250 const stepEvent = (
251 createEvent: (
252 stepName: string,
253 ) => Extract<WorkflowSubscriptionEvent, { stepName: string }>,
254 ): EngineWorkflowSubscriptionEvent =>
255 log.target === null
256 ? { ...common, type: "internal" }
257 : createEvent(log.target);
258
259 switch (log.event) {
260 case InstanceEvent.WORKFLOW_QUEUED:
261 return { ...common, type: "workflow_queued" };
262 case InstanceEvent.WORKFLOW_START:
263 return { ...common, type: "workflow_started", params };
264 case InstanceEvent.WORKFLOW_RUNNING:
265 return { ...common, type: "workflow_running" };
266 case InstanceEvent.WORKFLOW_PAUSED:
267 return { ...common, type: "workflow_paused" };
268 case InstanceEvent.WORKFLOW_WAITING_FOR_PAUSE:
269 return { ...common, type: "workflow_waiting_for_pause" };
270 case InstanceEvent.WORKFLOW_WAITING:
271 return { ...common, type: "workflow_waiting" };
272 case InstanceEvent.WORKFLOW_SUCCESS: {
273 const storedOutput = await storage.get<{ value: unknown }>(
274 WORKFLOW_OUTPUT_KEY,
275 );
276 return {
277 ...common,
278 type: "workflow_completed",
279 output:
280 storedOutput === undefined
281 ? parseMetadata().result
282 : storedOutput.value,
283 };
284 }
285 case InstanceEvent.WORKFLOW_FAILURE:
286 return {
287 ...common,
288 type: "workflow_errored",
289 error: parseMetadata().error,
290 };
291 case InstanceEvent.WORKFLOW_TERMINATED:
292 return { ...common, type: "workflow_terminated" };
293 case InstanceEvent.STEP_START: {
294 const config = await readStepConfig(storage, log.groupKey);
295 return stepEvent((stepName) => ({
296 ...common,
297 type: "step_started",
298 stepName,
299 ...(config === undefined ? {} : { config }),
300 }));
301 }
302 case InstanceEvent.STEP_SUCCESS: {
303 const output = await readStepCompletedOutput(storage, log.groupKey);
304 return stepEvent((stepName) => ({
305 ...common,
306 type: "step_completed",
307 stepName,
308 ...(output === undefined ? {} : { output }),
309 }));
310 }
311 case InstanceEvent.STEP_FAILURE:
312 return stepEvent((stepName) => ({
313 ...common,
314 type: "step_errored",
315 stepName,
316 }));
317 case InstanceEvent.ATTEMPT_START:
318 return stepEvent((stepName) => ({
319 ...common,
320 type: "attempt_started",
321 stepName,
322 attempt: parseMetadata().attempt,
323 }));
324 case InstanceEvent.ATTEMPT_SUCCESS:
325 return stepEvent((stepName) => ({
326 ...common,
327 type: "attempt_completed",
328 stepName,
329 attempt: parseMetadata().attempt,
330 }));
331 case InstanceEvent.ATTEMPT_FAILURE: {
332 const metadata = parseMetadata();
333 return stepEvent((stepName) => ({
334 ...common,
335 type: "attempt_errored",
336 stepName,
337 attempt: metadata.attempt,
338 error: metadata.error,
339 ...(metadata.retryDelayMs === undefined
340 ? {}
341 : { retryDelayMs: metadata.retryDelayMs }),
342 }));
343 }
344 case InstanceEvent.SLEEP_START:
345 return stepEvent((stepName) => ({
346 ...common,
347 type: "sleep_started",
348 stepName,
349 durationMs: parseMetadata().durationMs,
350 }));
351 case InstanceEvent.SLEEP_COMPLETE:
352 return stepEvent((stepName) => ({
353 ...common,
354 type: "sleep_completed",
355 stepName,
356 }));
357 case InstanceEvent.WAIT_START:
358 return stepEvent((stepName) => ({
359 ...common,
360 type: "wait_started",
361 stepName,
362 eventType: parseMetadata().event,
363 }));
364 case InstanceEvent.WAIT_COMPLETE:
365 return stepEvent((stepName) => ({
366 ...common,
367 type: "wait_completed",
368 stepName,
369 }));
370 case InstanceEvent.WAIT_TIMED_OUT:
371 return stepEvent((stepName) => ({
372 ...common,
373 type: "wait_timed_out",
374 stepName,
375 }));
376 case InstanceEvent.ROLLBACK_START:
377 return { ...common, type: "rollback_started" };
378 case InstanceEvent.ROLLBACK_STEP_START: {
379 const config = await readStepConfig(storage, log.groupKey);
380 return stepEvent((stepName) => ({
381 ...common,
382 type: "rollback_step_started",
383 stepName,
384 ...(config === undefined ? {} : { config }),
385 }));
386 }
387 case InstanceEvent.ROLLBACK_STEP_SUCCESS:
388 return stepEvent((stepName) => ({
389 ...common,
390 type: "rollback_step_completed",
391 stepName,
392 }));
393 case InstanceEvent.ROLLBACK_STEP_FAILURE:
394 return stepEvent((stepName) => ({
395 ...common,
396 type: "rollback_step_errored",
397 stepName,
398 error: parseMetadata().error,
399 }));
400 case InstanceEvent.ROLLBACK_ATTEMPT_START:
401 return stepEvent((stepName) => ({
402 ...common,
403 type: "rollback_attempt_started",
404 stepName,
405 attempt: parseMetadata().attempt,
406 }));
407 case InstanceEvent.ROLLBACK_ATTEMPT_SUCCESS:
408 return stepEvent((stepName) => ({
409 ...common,
410 type: "rollback_attempt_completed",
411 stepName,
412 attempt: parseMetadata().attempt,
413 }));
414 case InstanceEvent.ROLLBACK_ATTEMPT_FAILURE: {
415 const metadata = parseMetadata();
416 return stepEvent((stepName) => ({
417 ...common,
418 type: "rollback_attempt_errored",
419 stepName,
420 attempt: metadata.attempt,
421 error: metadata.error,
422 ...(metadata.retryDelayMs === undefined
423 ? {}
424 : { retryDelayMs: metadata.retryDelayMs }),
425 }));
426 }
427 case InstanceEvent.ROLLBACK_COMPLETE:
428 return { ...common, type: "rollback_completed" };
429 case InstanceEvent.ROLLBACK_FAILED:
430 return { ...common, type: "rollback_errored" };
431 case InstanceEvent.__INTERNAL_PROD:
432 return { ...common, type: "internal" };
433 }
434}
435
436export class Engine extends DurableObject<Env> {
437 logs: Array<unknown> = [];
438
439 isRunning: boolean = false;
440 accountId: number | undefined;
441 instanceId: string | undefined;
442 workflowName: string | undefined;
443 timeoutHandler: GracePeriodSemaphore;
444 priorityQueue: TimePriorityQueue | undefined;
445 stepLimit: number;
446 engineAbortController: AbortController = new AbortController();
447 pauseController: AbortController = new AbortController();
448 rollbackPhase: RollbackPhase | undefined = undefined;
449 rollbackEligibleCacheKeys: Set<string> | undefined = undefined;
450
451 waiters: Map<
452 string,
453 Array<
454 [cacheKey: string, resolve: (event: Event | PromiseLike<Event>) => void]
455 >
456 > = new Map();
457 eventMap: Map<string, Array<Event>> = new Map();
458
459 // Not persisted: rollback fns are RPC stubs, dead across DO restarts.
460 rollbackRegistry: Map<string, RollbackRegistryEntry> = new Map();
461
462 subscribers = new Set<WorkflowSubscriptionState>();
463
464 constructor(state: DurableObjectState, env: Env) {
465 super(state, env);
466
467 this.stepLimit = env.STEP_LIMIT
468 ? JSON.parse(env.STEP_LIMIT)
469 : DEFAULT_STEP_LIMIT;
470
471 void this.ctx.blockConcurrencyWhile(async () => {
472 this.ctx.storage.transactionSync(() => {
473 try {
474 this.ctx.storage.sql.exec(`
475 CREATE TABLE IF NOT EXISTS priority_queue (
476 id INTEGER PRIMARY KEY NOT NULL,
477 created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
478 target_timestamp INTEGER NOT NULL,
479 action INTEGER NOT NULL, -- should only be 0 or 1 (1 for added, 0 for deleted),
480 entryType INTEGER NOT NULL,
481 hash TEXT NOT NULL,
482 CHECK (action IN (0, 1)), -- guararentee that action can only be 0 or 1
483 UNIQUE (action, entryType, hash)
484 );
485 CREATE TABLE IF NOT EXISTS states (
486 id INTEGER PRIMARY KEY NOT NULL,
487 timestamp TIMESTAMP DEFAULT (DATETIME('now','subsec')),
488 groupKey TEXT,
489 target TEXT,
490 metadata TEXT,
491 event INTEGER NOT NULL
492 );
493 CREATE TABLE IF NOT EXISTS streaming_step_chunks (
494 cache_key TEXT NOT NULL,
495 attempt INTEGER NOT NULL,
496 chunk_index INTEGER NOT NULL,
497 chunk BLOB NOT NULL,
498 PRIMARY KEY (cache_key, attempt, chunk_index)
499 ) WITHOUT ROWID
500 `);
501 } catch (e) {
502 console.error(e);
503 throw e;
504 }
505 });
506 });
507
508 this.timeoutHandler = new GracePeriodSemaphore(
509 startGracePeriod,
510 ENGINE_TIMEOUT,
511 );
512 }
513
514 writeLog(
515 event: InstanceEvent,
516 group: string | null,
517 target: string | null = null,
518 metadata: Record<string, unknown>,
519 ) {
520 this.ctx.storage.sql.exec(
521 "INSERT INTO states (event, groupKey, target, metadata) VALUES (?, ?, ?, ?)",
522 event,
523 group,
524 target,
525 JSON.stringify(metadata, binaryReplacer),
526 );
527
528 for (const subscriber of this.subscribers) {
529 const waiter = subscriber.waiter;
530 if (waiter !== undefined) {
531 subscriber.waiter = undefined;
532 waiter.resolve();
533 }
534 }
535
536 // Wake any waiters if this is a terminal step event
537 if (group) {
538 this.handleStepResultWaiter(group, event, metadata);
539 }
540 }
541
542 readEligibleRollbackStepsDesc(
543 limit?: number,
544 ): Array<{ cacheKey: string; target: string }> {
545 const rollbackTerminalGroups = new Set<string>();
546 const rollbackEligibleGroups = new Set<string>();
547 const stepStartsDesc: Array<{ groupKey: string; target: string | null }> =
548 [];
549 const rows = [
550 ...this.ctx.storage.sql.exec<{
551 event: InstanceEvent;
552 groupKey: string;
553 target: string | null;
554 metadata: string;
555 }>(
556 "SELECT event, groupKey, target, metadata FROM states WHERE groupKey IS NOT NULL ORDER BY id DESC",
557 ),
558 ];
559
560 for (const row of rows) {
561 if (row.event === InstanceEvent.STEP_START) {
562 stepStartsDesc.push({ groupKey: row.groupKey, target: row.target });
563 }
564
565 if (
566 row.event === InstanceEvent.ROLLBACK_STEP_SUCCESS ||
567 row.event === InstanceEvent.ROLLBACK_STEP_FAILURE
568 ) {
569 rollbackTerminalGroups.add(
570 row.groupKey.startsWith(ROLLBACK_CACHE_KEY_PREFIX)
571 ? row.groupKey.slice(ROLLBACK_CACHE_KEY_PREFIX.length)
572 : row.groupKey,
573 );
574 continue;
575 }
576
577 if (
578 row.event !== InstanceEvent.STEP_START &&
579 row.event !== InstanceEvent.STEP_SUCCESS &&
580 row.event !== InstanceEvent.STEP_FAILURE
581 ) {
582 continue;
583 }
584
585 try {
586 if (
587 (JSON.parse(row.metadata) as { hasRollback?: boolean })
588 .hasRollback === true
589 ) {
590 rollbackEligibleGroups.add(row.groupKey);
591 }
592 } catch {
593 // Ignore malformed metadata in local persisted logs.
594 }
595 }
596
597 const eligible: Array<{ cacheKey: string; target: string }> = [];
598 for (const { groupKey, target } of stepStartsDesc) {
599 if (
600 rollbackEligibleGroups.has(groupKey) &&
601 !rollbackTerminalGroups.has(groupKey)
602 ) {
603 eligible.push({ cacheKey: groupKey, target: target ?? groupKey });
604 if (limit !== undefined && eligible.length >= limit) {
605 break;
606 }
607 }
608 }
609
610 return eligible;
611 }
612
613 private getEligibleRollbackSteps(limit?: number): Array<string> {
614 return this.readEligibleRollbackStepsDesc(limit).map(
615 ({ cacheKey }) => cacheKey,
616 );
617 }
618
619 registerRollbackFn(registration: RollbackRegistration): void {
620 if (
621 this.rollbackPhase === "replay" &&
622 this.rollbackEligibleCacheKeys !== undefined &&
623 !this.rollbackEligibleCacheKeys.has(registration.cacheKey)
624 ) {
625 disposeRollbackStub(registration.fn);
626 return;
627 }
628
629 registerRollbackFn(this.rollbackRegistry, registration);
630 }
631
632 setRollbackPhase(phase: RollbackPhase | undefined): void {
633 this.rollbackPhase = phase;
634 }
635
636 // Lives here for access to the protected DurableObject `ctx`.
637 createRollbackContext(rollbackStep?: { cacheKey: string }): Context {
638 return new Context(this, this.ctx, rollbackStep);
639 }
640
641 readLogsFromStep(_cacheKey: string): Array<RawInstanceLog> {
642 return [];
643 }
644
645 async subscribe(
646 options?: WorkflowSubscriptionOptions,
647 ): Promise<WorkflowSubscriptionTarget> {
648 const { cursor, filter } = options ?? {};
649 const metadata =
650 await this.ctx.storage.get<InstanceMetadata>(INSTANCE_METADATA);
651 if (metadata === undefined) {
652 throw createWorkflowError(
653 "Instance does not exist",
654 "instance.not_found",
655 );
656 }
657
658 const state: WorkflowSubscriptionState = {
659 instanceId: metadata.instance.id,
660 params: metadata.event.payload,
661 lastEventId: cursor ?? -1,
662 filter: filter === undefined ? undefined : new Set(filter),
663 waiter: undefined,
664 closed: false,
665 };
666 const subscription = new WorkflowSubscriptionTarget(
667 () => this.nextWorkflowEvent(state),
668 () => {
669 state.closed = true;
670 this.subscribers.delete(state);
671 state.waiter?.resolve();
672 state.waiter = undefined;
673 },
674 );
675 this.subscribers.add(state);
676 return subscription;
677 }
678
679 private async nextWorkflowEvent(
680 state: WorkflowSubscriptionState,
681 ): Promise<IteratorResult<WorkflowSubscriptionEvent, undefined>> {
682 while (!state.closed) {
683 const row = this.ctx.storage.sql
684 .exec<RawInstanceLog>(
685 "SELECT id, timestamp, event, groupKey, target, metadata FROM states WHERE id > ? ORDER BY id ASC LIMIT 1",
686 state.lastEventId,
687 )
688 .toArray()[0];
689
690 if (row === undefined) {
691 const hasTerminalEvent =
692 this.ctx.storage.sql
693 .exec<{ id: number }>(
694 "SELECT id FROM states WHERE id <= ? AND event IN (?, ?, ?) LIMIT 1",
695 state.lastEventId,
696 InstanceEvent.WORKFLOW_SUCCESS,
697 InstanceEvent.WORKFLOW_FAILURE,
698 InstanceEvent.WORKFLOW_TERMINATED,
699 )
700 .toArray()[0] !== undefined;
701 if (hasTerminalEvent) {
702 state.closed = true;
703 return { done: true, value: undefined };
704 }
705
706 await new Promise<void>((resolve) => {
707 state.waiter = { resolve };
708 });
709 continue;
710 }
711
712 state.lastEventId = row.id;
713 const event = await buildWorkflowSubscriptionEvent(
714 this.ctx.storage,
715 row,
716 state.instanceId,
717 state.params,
718 );
719 if (event.type === "internal") {
720 continue;
721 }
722 if (state.filter === undefined || state.filter.has(event.type)) {
723 return { done: false, value: event };
724 }
725 if (isTerminalEvent(event)) {
726 return { done: true, value: undefined };
727 }
728 }
729
730 return { done: true, value: undefined };
731 }
732
733 readLogs(): EngineLogs {
734 const logs = [
735 ...this.ctx.storage.sql.exec<{
736 event: InstanceEvent;
737 groupKey: string | null;
738 target: string | null;
739 metadata: string;
740 }>("SELECT event, groupKey, target, metadata FROM states"),
741 ];
742
743 return {
744 logs: logs.map((log) => {
745 const metadata = JSON.parse(log.metadata);
746
747 if (!isStepSuccessEvent(log.event) || !metadata.streamOutput) {
748 return { ...log, metadata, group: log.groupKey };
749 }
750
751 const { cacheKey, meta } = metadata.streamOutput as {
752 cacheKey: string;
753 meta: StreamOutputMeta;
754 };
755 try {
756 const preview = getStoredStreamOutputPreview({
757 storage: this.ctx.storage,
758 cacheKey,
759 meta,
760 maxChars: 1024,
761 });
762 metadata.result =
763 preview.type === "text"
764 ? preview.output
765 : `[ReadableStream (binary): ${meta.totalBytes} bytes]`;
766 } catch {
767 metadata.result = `[ReadableStream: ${meta.totalBytes} bytes]`;
768 }
769 delete metadata.streamOutput;
770
771 return { ...log, metadata, group: log.groupKey };
772 }),
773 };
774 }
775
776 /**
777 * Returns detailed logs including timestamps, ordered by ID.
778 * Used by the local explorer to reconstruct step-level detail.
779 */
780 readDetailedLogs(): Array<{
781 id: number;
782 timestamp: string;
783 event: number;
784 group: string | null;
785 target: string | null;
786 metadata: Record<string, unknown>;
787 }> {
788 const rows = [
789 ...this.ctx.storage.sql.exec<{
790 id: number;
791 timestamp: string;
792 event: number;
793 groupKey: string | null;
794 target: string | null;
795 metadata: string;
796 }>(
797 "SELECT id, timestamp, event, groupKey, target, metadata FROM states ORDER BY id ASC",
798 ),
799 ];
800
801 return rows.map((row) => {
802 const metadata = JSON.parse(row.metadata) as Record<string, unknown>;
803
804 if (!isStepSuccessEvent(row.event) || !metadata.streamOutput) {
805 return {
806 id: row.id,
807 timestamp: String(row.timestamp).replace(" ", "T") + "Z",
808 event: row.event,
809 group: row.groupKey,
810 target: row.target,
811 metadata,
812 };
813 }
814
815 const { cacheKey, meta } = metadata.streamOutput as {
816 cacheKey: string;
817 meta: StreamOutputMeta;
818 };
819 try {
820 const preview = getStoredStreamOutputPreview({
821 storage: this.ctx.storage,
822 cacheKey,
823 meta,
824 maxChars: 1024,
825 });
826 metadata.result =
827 preview.type === "text"
828 ? preview.output
829 : `[ReadableStream (binary): ${meta.totalBytes} bytes]`;
830 } catch {
831 metadata.result = `[ReadableStream: ${meta.totalBytes} bytes]`;
832 }
833 delete metadata.streamOutput;
834
835 return {
836 id: row.id,
837 timestamp: String(row.timestamp).replace(" ", "T") + "Z",
838 event: row.event,
839 group: row.groupKey,
840 target: row.target,
841 metadata,
842 };
843 });
844 }
845
846 readLogsFromEvent(eventType: InstanceEvent): EngineLogs {
847 const logs = [
848 ...this.ctx.storage.sql.exec<{
849 event: InstanceEvent;
850 groupKey: string | null;
851 target: string | null;
852 metadata: string;
853 }>(
854 "SELECT event, groupKey, target, metadata FROM states WHERE event = ?",
855 eventType,
856 ),
857 ];
858
859 return {
860 logs: logs.map((log) => ({
861 ...log,
862 metadata: JSON.parse(log.metadata),
863 group: log.groupKey,
864 })),
865 };
866 }
867
868 async getStatus(): Promise<InstanceStatus> {
869 if (this.accountId === undefined) {
870 // Engine could have restarted, so we try to restore from its state
871 const metadata =
872 await this.ctx.storage.get<InstanceMetadata>(INSTANCE_METADATA);
873 if (metadata === undefined) {
874 // metadata was never set, so we assume the engine was never started
875 throw new Error("Engine was never started");
876 }
877
878 this.accountId = metadata.accountId;
879 this.instanceId = metadata.instance.id;
880 this.workflowName = metadata.workflow.name;
881 }
882
883 const res = await this.ctx.storage.get<InstanceStatus>(ENGINE_STATUS_KEY);
884
885 // NOTE(lduarte): if status don't exist, means that engine is running for the first time, so we assume queued
886 if (res === undefined) {
887 return InstanceStatus.Queued;
888 }
889 return res;
890 }
891
892 // Returns instance metadata for the local explorer.
893 async getInstanceMetadata(): Promise<{
894 instanceId: string;
895 status: InstanceStatus;
896 createdOn: string;
897 }> {
898 const status = await this.getStatus();
899 // Read the full metadata to get the created_on timestamp
900 const metadata =
901 await this.ctx.storage.get<InstanceMetadata>(INSTANCE_METADATA);
902 let createdOn = metadata?.instance?.created_on ?? "";
903
904 // Backfill for instances created before created_on was populated:
905 // try the earliest log entry, then the earliest priority queue entry.
906 if (!createdOn) {
907 const queries = [
908 "SELECT timestamp AS ts FROM states ORDER BY id ASC LIMIT 1",
909 "SELECT created_on AS ts FROM priority_queue ORDER BY id ASC LIMIT 1",
910 ];
911 for (const query of queries) {
912 if (createdOn) {
913 break;
914 }
915 try {
916 for (const row of this.ctx.storage.sql.exec(query)) {
917 if (row.ts) {
918 // SQLite DATETIME('now','subsec') returns "YYYY-MM-DD HH:MM:SS.sss"
919 // which is not valid ISO 8601 (needs T separator and Z timezone).
920 const raw = String(row.ts).replace(" ", "T") + "Z";
921 createdOn = raw;
922 }
923 }
924 } catch {
925 // Table may not exist
926 }
927 }
928 }
929
930 return {
931 instanceId: this.instanceId ?? "",
932 status,
933 createdOn,
934 };
935 }
936
937 async setStatus(
938 accountId: number,
939 instanceId: string,
940 status: InstanceStatus,
941 ): Promise<void> {
942 const previousStatus =
943 await this.ctx.storage.get<InstanceStatus>(ENGINE_STATUS_KEY);
944 await this.ctx.storage.put(ENGINE_STATUS_KEY, status);
945
946 if (previousStatus !== status) {
947 switch (status) {
948 case InstanceStatus.Queued: {
949 // Creation writes WORKFLOW_QUEUED before the first status is stored.
950 const hasQueuedEvent =
951 this.ctx.storage.sql
952 .exec<{ id: number }>(
953 "SELECT id FROM states WHERE event = ? LIMIT 1",
954 InstanceEvent.WORKFLOW_QUEUED,
955 )
956 .toArray()[0] !== undefined;
957 if (!(previousStatus === undefined && hasQueuedEvent)) {
958 this.writeLog(InstanceEvent.WORKFLOW_QUEUED, null, null, {});
959 }
960 break;
961 }
962 case InstanceStatus.Running:
963 this.writeLog(InstanceEvent.WORKFLOW_RUNNING, null, null, {});
964 break;
965 case InstanceStatus.Paused:
966 this.writeLog(InstanceEvent.WORKFLOW_PAUSED, null, null, {});
967 break;
968 case InstanceStatus.WaitingForPause:
969 this.writeLog(
970 InstanceEvent.WORKFLOW_WAITING_FOR_PAUSE,
971 null,
972 null,
973 {},
974 );
975 break;
976 case InstanceStatus.Waiting:
977 this.writeLog(InstanceEvent.WORKFLOW_WAITING, null, null, {});
978 break;
979 }
980 }
981
982 // check if anyone is waiting for this status
983 this.handleStatusWaiter(status);
984 }
985
986 private statusWaiters: Map<
987 InstanceStatus,
988 { resolve: () => void; reject: (e: unknown) => void }
989 > = new Map();
990 async waitForStatus(status: string): Promise<void> {
991 const targetStatus = toInstanceStatus(status);
992 const currentStatus =
993 await this.ctx.storage.get<InstanceStatus>(ENGINE_STATUS_KEY);
994
995 // if the workflow has already reached the desired state, resolve immediately
996 if (currentStatus === targetStatus) {
997 return;
998 }
999
1000 // if it hasn't reached the desired state, create a new promise and add its resolver to the waiters map
1001 return new Promise((resolve, reject) => {
1002 this.statusWaiters.set(targetStatus, { resolve, reject });
1003 // immediately reconcile against current status in case it's already finite
1004 this.handleStatusWaiter(currentStatus as InstanceStatus);
1005 });
1006 }
1007
1008 handleStatusWaiter(status: InstanceStatus): void {
1009 const waiter = this.statusWaiters.get(status);
1010
1011 // resolve if it reached the desired status
1012 if (waiter) {
1013 waiter.resolve();
1014 this.statusWaiters.delete(status);
1015 return;
1016 }
1017
1018 switch (status) {
1019 case InstanceStatus.Errored: {
1020 // if it reaches final status "errored", then it can't be waiting for it to complete or terminate
1021 const unreachableStatuses = [
1022 InstanceStatus.Complete,
1023 InstanceStatus.Terminated,
1024 ];
1025
1026 this.rejectUnreachableStatus(status, unreachableStatuses);
1027 break;
1028 }
1029 case InstanceStatus.Terminated: {
1030 // if it reaches final status "terminated", then it can't be waiting for it to complete or error
1031 const unreachableStatuses = [
1032 InstanceStatus.Complete,
1033 InstanceStatus.Errored,
1034 ];
1035
1036 this.rejectUnreachableStatus(status, unreachableStatuses);
1037 break;
1038 }
1039 case InstanceStatus.Complete: {
1040 // if it reaches final status "complete", then it can't be waiting for it to terminate or error
1041 const unreachableStatuses = [
1042 InstanceStatus.Terminated,
1043 InstanceStatus.Errored,
1044 ];
1045
1046 this.rejectUnreachableStatus(status, unreachableStatuses);
1047 break;
1048 }
1049 default:
1050 break;
1051 }
1052 }
1053
1054 rejectUnreachableStatus(
1055 reachedStatus: number,
1056 unreachableStatuses: Array<number>,
1057 ): void {
1058 if (unreachableStatuses) {
1059 for (const unreachableStatus of unreachableStatuses) {
1060 const waiter = this.statusWaiters.get(unreachableStatus);
1061 if (waiter) {
1062 waiter.reject(
1063 new Error(
1064 `[WorkflowIntrospector] The Workflow instance ${
1065 this.instanceId
1066 } has reached status '${instanceStatusName(
1067 reachedStatus,
1068 )}'. This is a finite status that prevents it from ever reaching the expected status of '${instanceStatusName(
1069 unreachableStatus,
1070 )}'.`,
1071 ),
1072 );
1073 this.statusWaiters.delete(unreachableStatus);
1074 return;
1075 }
1076 }
1077 }
1078 }
1079
1080 /**
1081 * Create a replay ReadableStream from stored stream output metadata.
1082 * Returns undefined if the stream data is not in a valid/complete state.
1083 */
1084 private replayStreamFromMeta(streamOutput: {
1085 cacheKey: string;
1086 meta: StreamOutputMeta;
1087 }): ReadableStream<Uint8Array> | undefined {
1088 if (streamOutput.meta.state !== StreamOutputState.Complete) {
1089 return undefined;
1090 }
1091 const integrityError = getInvalidStoredStreamOutputError(
1092 this.ctx.storage,
1093 streamOutput.cacheKey,
1094 streamOutput.meta,
1095 );
1096 if (integrityError !== undefined) {
1097 return undefined;
1098 }
1099 return createReplayReadableStream({
1100 storage: this.ctx.storage,
1101 cacheKey: streamOutput.cacheKey,
1102 meta: streamOutput.meta,
1103 });
1104 }
1105
1106 private stepResultWaiters: Map<
1107 string,
1108 { resolve: (v: unknown) => void; reject: (e: unknown) => void }
1109 > = new Map();
1110 async waitForStepResult(
1111 stepName: string,
1112 stepCount?: number,
1113 ): Promise<unknown> {
1114 const hash = await computeHash(stepName);
1115 const count = stepCount ?? 1;
1116 const cacheKey = `${hash}-${count}`;
1117
1118 // read latest log from step
1119 const rows = [
1120 ...this.ctx.storage.sql.exec<{
1121 event: InstanceEvent;
1122 metadata: string;
1123 }>(
1124 "SELECT event, metadata FROM states WHERE groupKey = ? ORDER BY id DESC LIMIT 1",
1125 cacheKey,
1126 ),
1127 ];
1128
1129 if (rows.length > 0) {
1130 const { event, metadata } = rows[0];
1131 const parsed = JSON.parse(metadata);
1132 if (event === InstanceEvent.STEP_SUCCESS) {
1133 if (parsed?.streamOutput) {
1134 return this.replayStreamFromMeta(parsed.streamOutput);
1135 }
1136 return parsed?.result;
1137 }
1138 if (event === InstanceEvent.STEP_FAILURE) {
1139 throw parsed?.error ?? parsed;
1140 }
1141 }
1142
1143 // if it hasn't completed the step, create a new promise to later resolve/reject
1144 return new Promise<unknown>((resolve, reject) => {
1145 this.stepResultWaiters.set(cacheKey, { resolve, reject });
1146 });
1147 }
1148
1149 handleStepResultWaiter(
1150 group: string,
1151 event: InstanceEvent,
1152 metadata: Record<string, unknown>,
1153 ) {
1154 const waiter = this.stepResultWaiters.get(group);
1155 if (!waiter) {
1156 return;
1157 }
1158 if (event === InstanceEvent.STEP_SUCCESS) {
1159 if (metadata?.streamOutput) {
1160 waiter.resolve(
1161 this.replayStreamFromMeta(
1162 metadata.streamOutput as {
1163 cacheKey: string;
1164 meta: StreamOutputMeta;
1165 },
1166 ),
1167 );
1168 } else {
1169 waiter.resolve(metadata?.result);
1170 }
1171 this.stepResultWaiters.delete(group);
1172 } else if (event === InstanceEvent.STEP_FAILURE) {
1173 const error = metadata?.error ?? new Error("Step failed");
1174 waiter.reject(error);
1175 this.stepResultWaiters.delete(group);
1176 }
1177 }
1178
1179 async getOutputOrError(isOutput: boolean): Promise<unknown> {
1180 const status = await this.getStatus();
1181
1182 if (isOutput) {
1183 if (status !== InstanceStatus.Complete) {
1184 throw new Error(
1185 `Cannot retrieve output: Workflow instance is in status "${instanceStatusName(
1186 status,
1187 )}" but must be "complete" to have an output available`,
1188 );
1189 }
1190 const logs = this.readLogsFromEvent(InstanceEvent.WORKFLOW_SUCCESS).logs;
1191 return logs.at(0)?.metadata.result;
1192 } else {
1193 if (status !== InstanceStatus.Errored) {
1194 throw new Error(
1195 `Cannot retrieve error: Workflow instance is in status "${instanceStatusName(
1196 status,
1197 )}" but must be "errored" to have error information available`,
1198 );
1199 }
1200 const logs = this.readLogsFromEvent(InstanceEvent.WORKFLOW_FAILURE).logs;
1201 const log = logs.at(0);
1202 if (!log?.metadata.error) {
1203 throw new Error(
1204 "Cannot retrieve error: No workflow instance failure log found",
1205 );
1206 }
1207 return log.metadata.error;
1208 }
1209 }
1210
1211 async abort(reason: string) {
1212 await this.ctx.storage.sync();
1213
1214 // Clean up pending JS operations
1215 this.timeoutHandler.dispose();
1216 this.engineAbortController.abort(new Error(reason));
1217
1218 this.ctx.abort(reason);
1219 }
1220
1221 // Called by the dispose function when introspecting the instance in tests
1222 // TODO: Ideally this abort should be done by `abortAllDurableObjects` from worked called by vitest-plugin
1223 async unsafeAbort(reason?: string) {
1224 await this.ctx.storage.sync();
1225 await this.ctx.storage.deleteAll();
1226
1227 this.ctx.abort(reason);
1228 }
1229
1230 async storeEventMap() {
1231 await this.ctx.blockConcurrencyWhile(async () => {
1232 const entries: Record<string, Event> = {};
1233 for (const [type, events] of this.eventMap.entries()) {
1234 for (let i = 0; i < events.length; i++) {
1235 entries[`${EVENT_MAP_PREFIX}\n${type}\n${i}`] = events[i];
1236 }
1237 }
1238 if (Object.keys(entries).length > 0) {
1239 await this.ctx.storage.put(entries);
1240 }
1241 });
1242 }
1243
1244 async restoreEventMap() {
1245 await this.ctx.blockConcurrencyWhile(async () => {
1246 // FIXME(lduarte): can this OoM the DO in the production?
1247 const entries = await this.ctx.storage.list<Event>({
1248 prefix: EVENT_MAP_PREFIX,
1249 });
1250 for (const [key, value] of entries) {
1251 const [_, eventType, _idx] = key.split("\n");
1252 // NOTE(lduarte): safe to do because list returns keys in ascending order, so
1253 // indexes will be correctly ordered
1254 const eventList = this.eventMap.get(eventType) ?? [];
1255 eventList.push(value);
1256 this.eventMap.set(eventType, eventList);
1257 }
1258 });
1259 }
1260
1261 async receiveEvent(event: {
1262 timestamp: Date;
1263 payload: unknown;
1264 type: string;
1265 }) {
1266 // There are four possible cases here:
1267 // - There is a callback waiting, send it
1268 // - There is no callback waiting but engine is alive, store it
1269 // - Engine is not awake and is in Waiting status, store it and start it up
1270 // - Engine is not awake and is in Paused (or another terminal) status, store it
1271 // - Engine is not awake and is Errored or Terminated, this should not get called
1272 let eventTypeQueue = this.eventMap.get(event.type) ?? [];
1273 eventTypeQueue.push(event as Event);
1274
1275 this.eventMap.set(event.type, eventTypeQueue);
1276 await this.storeEventMap();
1277
1278 // if the engine is running
1279 if (this.isRunning) {
1280 // Attempt to get the callback and run it
1281 const callbacks = this.waiters.get(event.type);
1282 if (callbacks) {
1283 const entry = callbacks[0];
1284 if (entry) {
1285 const [, resolve] = entry;
1286 resolve(event);
1287 // Remove it from the list of callbacks
1288 callbacks.shift();
1289 this.waiters.set(event.type, callbacks);
1290
1291 eventTypeQueue = this.eventMap.get(event.type) ?? [];
1292 eventTypeQueue.shift();
1293 this.eventMap.set(event.type, eventTypeQueue);
1294
1295 return;
1296 }
1297 }
1298 } else {
1299 const mockEvent = await this.ctx.storage.get(
1300 `${MODIFIER_KEYS.MOCK_EVENT}${event.type}`,
1301 );
1302 if (mockEvent) {
1303 return;
1304 }
1305
1306 const metadata =
1307 await this.ctx.storage.get<InstanceMetadata>(INSTANCE_METADATA);
1308
1309 if (metadata === undefined) {
1310 throw new Error("Engine was never started");
1311 }
1312
1313 void this.init(
1314 metadata.accountId,
1315 metadata.workflow,
1316 metadata.version,
1317 metadata.instance,
1318 metadata.event,
1319 );
1320 }
1321 }
1322
1323 getInstanceModifier(): WorkflowInstanceModifier {
1324 return new WorkflowInstanceModifier(this, this.ctx);
1325 }
1326
1327 async changeInstanceStatus(
1328 newStatus: "resume" | "pause" | "terminate" | "restart",
1329 from?: RestartFromStep,
1330 terminateOptions?: WorkflowInstanceTerminateOptions,
1331 ): Promise<void> {
1332 const metadata =
1333 await this.ctx.storage.get<InstanceMetadata>(INSTANCE_METADATA);
1334
1335 if (metadata === undefined) {
1336 throw createWorkflowError(
1337 "Instance does not exist",
1338 "instance.not_found",
1339 );
1340 }
1341
1342 switch (newStatus) {
1343 case "pause":
1344 await this.userTriggeredPause();
1345 break;
1346 case "resume": {
1347 const currentStatus = await this.getStatus();
1348 if (currentStatus === InstanceStatus.WaitingForPause) {
1349 // Engine is still running — cancel the pending pause
1350 this.timeoutHandler.cancelWaitingPromisesByType("pause");
1351 await this.setStatus(
1352 metadata.accountId,
1353 metadata.instance.id,
1354 InstanceStatus.Running,
1355 );
1356 } else if (currentStatus === InstanceStatus.Paused) {
1357 await this.attemptResume();
1358 }
1359 break;
1360 }
1361 case "terminate": {
1362 const currentStatus = await this.getStatus();
1363 if (
1364 [
1365 InstanceStatus.Terminated,
1366 InstanceStatus.Complete,
1367 InstanceStatus.Errored,
1368 ].includes(currentStatus)
1369 ) {
1370 throw createWorkflowError(
1371 "Cannot terminate instance since its on a finite state",
1372 "instance.cannot_terminate",
1373 );
1374 }
1375 await this.userTriggeredTerminate(terminateOptions);
1376 break;
1377 }
1378 case "restart":
1379 if (from) {
1380 if (!resolveGroupKeysToWipe(this.ctx.storage.sql, from)) {
1381 throw stepNotFoundError(from.name);
1382 }
1383 await storeRestartFromStep(this.ctx.storage, from);
1384 }
1385 await this.userTriggeredRestart();
1386 break;
1387 }
1388 }
1389
1390 private async replayRollbackRegistry(
1391 metadata: InstanceMetadata,
1392 ): Promise<void> {
1393 if (this.rollbackRegistry.size > 0) {
1394 return;
1395 }
1396
1397 const eligible = this.getEligibleRollbackSteps();
1398 if (eligible.length === 0) {
1399 return;
1400 }
1401
1402 this.rollbackEligibleCacheKeys = new Set(eligible);
1403 const stubStep = this.createRollbackContext();
1404 this.setRollbackPhase("replay");
1405 try {
1406 await this.env.USER_WORKFLOW.run(
1407 metadata.event,
1408 stubStep as unknown as WorkflowStep,
1409 );
1410 } catch (replayErr) {
1411 // Match the production engine: replay may stop on normal workflow control
1412 // flow; rollback execution uses whatever handlers replay registered.
1413 console.debug("Rollback replay stopped:", replayErr);
1414 } finally {
1415 this.setRollbackPhase(undefined);
1416 this.rollbackEligibleCacheKeys = undefined;
1417 }
1418 }
1419
1420 async userTriggeredTerminate(options?: WorkflowInstanceTerminateOptions) {
1421 const metadata =
1422 await this.ctx.storage.get<InstanceMetadata>(INSTANCE_METADATA);
1423
1424 if (metadata === undefined) {
1425 throw createWorkflowError(
1426 "Instance does not exist",
1427 "instance.not_found",
1428 );
1429 }
1430
1431 if (options?.rollback === true) {
1432 this.priorityQueue ??= new TimePriorityQueue(this.ctx, metadata);
1433 await this.replayRollbackRegistry(metadata);
1434
1435 const error = new Error("Instance terminated during rollback");
1436 error.name = "Terminated";
1437 this.setRollbackPhase("rollback");
1438 try {
1439 await executeRollbacks(this, error);
1440 } catch (rollbackErr) {
1441 console.error("Rollback execution failed:", rollbackErr);
1442 } finally {
1443 this.setRollbackPhase(undefined);
1444 }
1445 }
1446
1447 this.writeLog(InstanceEvent.WORKFLOW_TERMINATED, null, null, {
1448 trigger: {
1449 source: InstanceTrigger.API,
1450 },
1451 });
1452
1453 await this.setStatus(
1454 metadata.accountId,
1455 metadata.instance.id,
1456 InstanceStatus.Terminated,
1457 );
1458
1459 await this.abort(ABORT_REASONS.USER_TERMINATE);
1460 }
1461
1462 /** Deletes all instance state and aborts its current execution. */
1463 async deleteInstance(): Promise<void> {
1464 if ((await this.ctx.storage.get(INSTANCE_METADATA)) === undefined) {
1465 throw createWorkflowError(
1466 "Instance does not exist",
1467 "instance.not_found",
1468 );
1469 }
1470
1471 await this.ctx.storage.deleteAll();
1472
1473 if (
1474 this.env.MINIFLARE_LOOPBACK !== undefined &&
1475 this.env.WORKFLOW_NAME !== undefined
1476 ) {
1477 try {
1478 const response = await this.env.MINIFLARE_LOOPBACK.fetch(
1479 `http://localhost/core/workflow-storage/${encodeURIComponent(this.env.WORKFLOW_NAME)}/${this.ctx.id.toString()}?defer=1`,
1480 { method: "DELETE" },
1481 );
1482 if (!response.ok && response.status !== 404) {
1483 console.error("Failed to delete persisted workflow instance");
1484 }
1485 } catch (error) {
1486 console.error("Failed to delete persisted workflow instance", error);
1487 }
1488 }
1489
1490 await this.abort(ABORT_REASONS.USER_DELETE);
1491 }
1492
1493 async userTriggeredPause() {
1494 const status = await this.getStatus();
1495
1496 if (
1497 status === InstanceStatus.Paused ||
1498 status === InstanceStatus.WaitingForPause
1499 ) {
1500 return;
1501 }
1502
1503 if (
1504 status !== InstanceStatus.Running &&
1505 status !== InstanceStatus.Waiting
1506 ) {
1507 return;
1508 }
1509
1510 const metadata =
1511 await this.ctx.storage.get<InstanceMetadata>(INSTANCE_METADATA);
1512
1513 if (metadata === undefined) {
1514 throw createWorkflowError(
1515 "Instance does not exist",
1516 "instance.not_found",
1517 );
1518 }
1519
1520 await this.setStatus(
1521 metadata.accountId,
1522 metadata.instance.id,
1523 InstanceStatus.WaitingForPause,
1524 );
1525
1526 void this.timeoutHandler
1527 .waitUntilNothingIsRunning("pause", async () => {
1528 await this.ctx.storage.put(PAUSE_DATETIME, new Date());
1529 await this.setStatus(
1530 metadata.accountId,
1531 metadata.instance.id,
1532 InstanceStatus.Paused,
1533 );
1534 // Signal the pause controller to interrupt any active
1535 // scheduler.wait (sleep/waitForEvent). The workflow will
1536 // throw a pause error at the next step boundary via
1537 // #checkForPendingPause()
1538 this.pauseController.abort(ABORT_REASONS.USER_PAUSE);
1539 })
1540 .catch(() => {});
1541 }
1542
1543 async userTriggeredRestart() {
1544 // cleanup is done in attemptRestart() on the fresh DO instance
1545
1546 await this.abort(ABORT_REASONS.USER_RESTART);
1547 }
1548
1549 async attemptRestart() {
1550 const restartFromStep = await readAndClearRestartFromStep(this.ctx.storage);
1551
1552 let groupKeysToWipe: Set<string> | null = null;
1553 if (restartFromStep) {
1554 groupKeysToWipe = resolveGroupKeysToWipe(
1555 this.ctx.storage.sql,
1556 restartFromStep,
1557 );
1558 if (!groupKeysToWipe) {
1559 throw stepNotFoundError(restartFromStep.name);
1560 }
1561 }
1562
1563 await wipeRestartState(
1564 this.ctx.storage,
1565 ENGINE_STATUS_KEY,
1566 PAUSE_DATETIME,
1567 groupKeysToWipe,
1568 );
1569
1570 const metadata =
1571 await this.ctx.storage.get<InstanceMetadata>(INSTANCE_METADATA);
1572
1573 if (metadata === undefined) {
1574 throw createWorkflowError(
1575 "Instance does not exist",
1576 "instance.not_found",
1577 );
1578 }
1579
1580 const { accountId, workflow, version, instance, event } = metadata;
1581
1582 if (!groupKeysToWipe) {
1583 this.writeLog(InstanceEvent.WORKFLOW_QUEUED, null, null, {
1584 params: event.payload,
1585 versionId: version.id,
1586 trigger: {
1587 source: InstanceTrigger.API,
1588 },
1589 });
1590 this.writeLog(InstanceEvent.WORKFLOW_START, null, null, {});
1591 }
1592
1593 void this.init(accountId, workflow, version, instance, event);
1594 }
1595
1596 async attemptResume() {
1597 const metadata =
1598 await this.ctx.storage.get<InstanceMetadata>(INSTANCE_METADATA);
1599
1600 if (metadata === undefined) {
1601 throw createWorkflowError(
1602 "Instance does not exist",
1603 "instance.not_found",
1604 );
1605 }
1606
1607 const status =
1608 await this.ctx.storage.get<InstanceStatus>(ENGINE_STATUS_KEY);
1609 if (status !== InstanceStatus.Paused) {
1610 return;
1611 }
1612
1613 // Offset priority queue timers by the pause duration so that
1614 // sleeps/retries resume from where they left off
1615 const pausedDate = await this.ctx.storage.get<Date>(PAUSE_DATETIME);
1616 if (pausedDate !== undefined) {
1617 const offset = Date.now() - new Date(pausedDate).valueOf();
1618 if (this.priorityQueue) {
1619 this.priorityQueue.offsetAll(offset);
1620 } else {
1621 const pq = new TimePriorityQueue(this.ctx, metadata);
1622 pq.offsetAll(offset);
1623 }
1624 }
1625 await this.ctx.storage.delete(PAUSE_DATETIME);
1626
1627 const { accountId, workflow, version, instance, event } = metadata;
1628
1629 await this.ctx.storage.put(ENGINE_STATUS_KEY, InstanceStatus.Queued);
1630
1631 // Reset pause state for the new run. The DO stays alive across
1632 // pause/resume (no this.ctx.abort()), so we need to clear stale
1633 // in-memory state from the previous run to avoid duplicates.
1634 this.pauseController = new AbortController();
1635 this.waiters = new Map();
1636 this.eventMap = new Map();
1637 clearRollbackRegistry(this.rollbackRegistry);
1638
1639 void this.init(accountId, workflow, version, instance, event);
1640 }
1641
1642 async init(
1643 accountId: number,
1644 workflow: DatabaseWorkflow,
1645 version: DatabaseVersion,
1646 instance: DatabaseInstance,
1647 event: WorkflowEvent<unknown>,
1648 ) {
1649 if (this.priorityQueue === undefined) {
1650 this.priorityQueue = new TimePriorityQueue(
1651 this.ctx,
1652 // this.env,
1653 {
1654 accountId,
1655 workflow,
1656 version,
1657 instance,
1658 event,
1659 },
1660 );
1661 }
1662
1663 if (this.isRunning) {
1664 return;
1665 }
1666
1667 this.priorityQueue.popPastEntries();
1668 await this.priorityQueue.handleNextAlarm();
1669
1670 // We are not running and are possibly starting a new lifetime
1671 this.accountId = accountId;
1672 this.instanceId = instance.id;
1673 this.workflowName = workflow.name;
1674
1675 const status = await this.getStatus();
1676 if (
1677 [
1678 InstanceStatus.Errored, // TODO (WOR-85): Remove this once upgrade story is done
1679 InstanceStatus.Terminated,
1680 InstanceStatus.Complete,
1681 InstanceStatus.Paused,
1682 ].includes(status)
1683 ) {
1684 return;
1685 }
1686
1687 // If the DO restarted (e.g. from alarm) while in WaitingForPause state,
1688 // transition to Paused and return early — same as production behaviour.
1689 if (status === InstanceStatus.WaitingForPause) {
1690 await this.ctx.storage.put(PAUSE_DATETIME, new Date());
1691 await this.setStatus(accountId, instance.id, InstanceStatus.Paused);
1692 return;
1693 }
1694
1695 if ((await this.ctx.storage.get(INSTANCE_METADATA)) == undefined) {
1696 const instanceMetadata: InstanceMetadata = {
1697 accountId,
1698 workflow,
1699 version,
1700 instance,
1701 event,
1702 };
1703 await this.ctx.storage.put(INSTANCE_METADATA, instanceMetadata);
1704
1705 // TODO (WOR-78): We currently don't have a queue mechanism
1706 this.writeLog(InstanceEvent.WORKFLOW_QUEUED, null, null, {
1707 params: event.payload,
1708 versionId: version.id,
1709 trigger: {
1710 source: InstanceTrigger.API,
1711 },
1712 });
1713 this.writeLog(InstanceEvent.WORKFLOW_START, null, null, {});
1714 }
1715
1716 // restore eventMap so that waitForEvent across lifetimes works correctly
1717 await this.restoreEventMap();
1718
1719 const stubStep = new Context(this, this.ctx);
1720
1721 const workflowRunningHandler = async () => {
1722 await this.ctx.storage.transaction(async () => {
1723 // manually start the grace period
1724 // startGracePeriod(this, this.timeoutHandler.timeoutMs);
1725 await this.setStatus(accountId, instance.id, InstanceStatus.Running);
1726 });
1727 };
1728 this.isRunning = true;
1729
1730 void workflowRunningHandler();
1731 try {
1732 const target = this.env.USER_WORKFLOW;
1733 const result = await target.run(
1734 event,
1735 stubStep as unknown as WorkflowStep,
1736 );
1737 await this.ctx.storage.put(WORKFLOW_OUTPUT_KEY, {
1738 value: normalizeForStorage(result),
1739 });
1740 this.writeLog(InstanceEvent.WORKFLOW_SUCCESS, null, null, {
1741 result,
1742 });
1743 // NOTE(lduarte): we want to run this in a transaction to guarentee ordering with running setstatus call
1744 // in case that it returns immediately
1745 await this.ctx.storage.transaction(async () => {
1746 await this.setStatus(accountId, instance.id, InstanceStatus.Complete);
1747 });
1748 // Dispose dup'd stubs; otherwise they leak across DO lifetimes.
1749 clearRollbackRegistry(this.rollbackRegistry);
1750 this.isRunning = false;
1751 } catch (err) {
1752 if (isAbortError(err)) {
1753 this.isRunning = false;
1754 return;
1755 }
1756
1757 // Run before the terminal status so events land before WORKFLOW_FAILURE.
1758 try {
1759 await executeRollbacks(
1760 this,
1761 err instanceof Error ? err : new Error(String(err)),
1762 );
1763 } catch (rollbackErr) {
1764 console.error("Rollback execution failed:", rollbackErr);
1765 }
1766
1767 let error;
1768 if (err instanceof Error) {
1769 if (
1770 err.name === "NonRetryableError" ||
1771 err.message.startsWith("NonRetryableError")
1772 ) {
1773 const fatalError = shouldPreserveNonRetryableError()
1774 ? new PreservedNonRetryableError(err)
1775 : new WorkflowFatalError(
1776 `The execution of the Workflow instance was terminated, as a step threw an NonRetryableError and it was not handled`,
1777 );
1778
1779 this.writeLog(InstanceEvent.WORKFLOW_FAILURE, null, null, {
1780 error: fatalError,
1781 });
1782
1783 await this.setStatus(accountId, instance.id, InstanceStatus.Errored);
1784 await this.abort(ABORT_REASONS.NON_RETRYABLE_ERROR);
1785 this.isRunning = false;
1786 return;
1787 }
1788 error = {
1789 message: err.message,
1790 name: err.name,
1791 };
1792 } else {
1793 error = {
1794 name: "Error",
1795 message: err,
1796 };
1797 }
1798
1799 this.writeLog(InstanceEvent.WORKFLOW_FAILURE, null, null, {
1800 error,
1801 });
1802 // NOTE(lduarte): we want to run this in a transaction to guarentee ordering with running setstatus call
1803 // in case that it throws immediately
1804 await this.ctx.storage.transaction(async () => {
1805 await this.setStatus(accountId, instance.id, InstanceStatus.Errored);
1806 });
1807 this.isRunning = false;
1808 }
1809
1810 return {
1811 id: instance.id,
1812 };
1813 }
1814}