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 } from "cloudflare:workers";
5import { ms } from "itty-time";
6import {
7 INSTANCE_METADATA,
8 InstanceEvent,
9 InstanceStatus,
10} from "./instance.ts";
11import { computeHash } from "./lib/cache.ts";
12import {
13 DEFAULT_RETRY_DELAY_MS,
14 DELAY_FUNCTION_TIMEOUT_MS,
15 invokeDelayFunction,
16 raceAgainstAbort,
17} from "./lib/delay.ts";
18import {
19 ABORT_REASONS,
20 InvalidStepReadableStreamError,
21 NonRetryableDelayError,
22 OversizedStreamChunkError,
23 PreservedNonRetryableError,
24 shouldPreserveNonRetryableError,
25 StreamOutputStorageLimitError,
26 UnsupportedStreamChunkError,
27 WorkflowFatalError,
28 WorkflowInternalError,
29 WorkflowTimeoutError,
30} from "./lib/errors.ts";
31import { calcRetryDuration, DelayFunctionError } from "./lib/retries.ts";
32import {
33 parseRollbackOptions,
34 ROLLBACK_CACHE_KEY_PREFIX,
35} from "./lib/rollback.ts";
36import { normalizeForStorage } from "./lib/serialization.ts";
37import {
38 cleanupPendingStreamOutput,
39 createReplayReadableStream,
40 getInvalidStoredStreamOutputError,
41 getStreamOutputMetaKey,
42 isReadableStreamLike,
43 rollbackStreamOutput,
44 StreamOutputState,
45 writeStreamOutput,
46} from "./lib/streams.ts";
47import {
48 isValidStepConfig,
49 isValidStepName,
50 MAX_STEP_NAME_LENGTH,
51 SENSITIVE_STEP_OUTPUT,
52} from "./lib/validators.ts";
53import { MODIFIER_KEYS } from "./modifier.ts";
54import type { Engine } from "./engine.ts";
55import type { InstanceMetadata } from "./instance.ts";
56import type {
57 RollbackFn,
58 WorkflowStepRollbackOptions,
59} from "./lib/rollback.ts";
60import type { StreamOutputMeta } from "./lib/streams.ts";
61import type {
62 WorkflowBackoff,
63 WorkflowDelayDuration,
64 WorkflowDelayFunction,
65 WorkflowSleepDuration,
66 WorkflowStepConfig,
67 WorkflowStepEvent,
68 WorkflowStepSensitivity,
69 WorkflowTimeoutDuration,
70} from "cloudflare:workers";
71
72export type Event = {
73 timestamp: Date;
74 payload: unknown;
75 type: string;
76};
77
78// Persisted in place of a dynamic delay function (functions can't be
79// serialized); the live function is re-read from the user's step config on each
80// execution.
81const SERIALIZABLE_DELAY_MARKER = "[dynamic]";
82type SerializableDelayMarker = typeof SERIALIZABLE_DELAY_MARKER;
83
84export const REDACTED_STEP_OUTPUT = "[REDACTED]";
85
86// The persisted, fully-merged config. A dynamic delay is stored as the marker.
87export type ResolvedStepConfig = {
88 retries: {
89 limit: number;
90 delay: WorkflowDelayDuration | SerializableDelayMarker;
91 backoff?: WorkflowBackoff;
92 };
93 timeout: WorkflowTimeoutDuration;
94 sensitive?: WorkflowStepSensitivity;
95};
96
97// The user-facing config (passed to step callbacks). The dynamic-delay marker is
98// never exposed: `delay` is omitted entirely when the delay is a function.
99export type EngineStepConfig = {
100 retries: {
101 limit: number;
102 delay?: WorkflowDelayDuration;
103 backoff?: WorkflowBackoff;
104 };
105 timeout: WorkflowTimeoutDuration;
106 sensitive?: WorkflowStepSensitivity;
107};
108
109export function toEngineStepConfig(
110 config: ResolvedStepConfig,
111): EngineStepConfig {
112 const { delay, ...retries } = config.retries;
113 if (
114 typeof delay === "number" ||
115 (typeof delay === "string" && delay !== SERIALIZABLE_DELAY_MARKER)
116 ) {
117 return { ...config, retries: { ...retries, delay } };
118 }
119 return { ...config, retries };
120}
121
122// Signal-aware wrapper around the global `scheduler.wait`. `scheduler.wait`
123// can't itself be cancelled, so an aborted signal resolves the returned promise
124// early and the dangling timer becomes a no-op.
125function schedulerWait(
126 durationMs: number,
127 opts?: { signal?: AbortSignal },
128): Promise<void> {
129 return new Promise<void>((resolve) => {
130 const signal = opts?.signal;
131 if (signal?.aborted) {
132 resolve();
133 return;
134 }
135 let done = false;
136 const finish = (): void => {
137 if (!done) {
138 done = true;
139 resolve();
140 }
141 };
142 void scheduler.wait(durationMs).then(finish);
143 signal?.addEventListener("abort", finish, { once: true });
144 });
145}
146
147const defaultConfig: ResolvedStepConfig = {
148 retries: {
149 limit: 5,
150 delay: 1000,
151 backoff: "exponential",
152 },
153 timeout: "10 minutes",
154};
155
156export interface UserErrorField {
157 isUserError?: boolean;
158}
159
160export type StepState = {
161 attemptedCount: number;
162};
163
164export type WorkflowStepContext = {
165 step: {
166 name: string;
167 count: number;
168 };
169 attempt: number;
170 config: EngineStepConfig;
171};
172
173const PAUSE_DATETIME = "PAUSE_DATETIME";
174
175export class Context extends RpcTarget {
176 #engine: Engine;
177 #state: DurableObjectState;
178
179 #counters: Map<string, number> = new Map();
180 #lifetimeStepCounter: number = 0;
181
182 #rollbackStep: { cacheKey: string } | undefined;
183
184 constructor(
185 engine: Engine,
186 state: DurableObjectState,
187 rollbackStep?: { cacheKey: string },
188 ) {
189 super();
190 this.#engine = engine;
191 this.#state = state;
192 this.#rollbackStep = rollbackStep;
193 }
194
195 async #checkForPendingPause(): Promise<void> {
196 if (this.#engine.timeoutHandler.isRunningStep()) {
197 return;
198 }
199
200 const status = await this.#engine.getStatus();
201
202 if (status === InstanceStatus.Paused) {
203 throw new Error(ABORT_REASONS.USER_PAUSE);
204 }
205
206 if (status === InstanceStatus.WaitingForPause) {
207 await this.#state.storage.put(PAUSE_DATETIME, new Date());
208 const metadata =
209 await this.#state.storage.get<InstanceMetadata>(INSTANCE_METADATA);
210 if (metadata) {
211 await this.#engine.setStatus(
212 metadata.accountId,
213 metadata.instance.id,
214 InstanceStatus.Paused,
215 );
216 }
217 throw new Error(ABORT_REASONS.USER_PAUSE);
218 }
219 }
220
221 #getCount(name: string): number {
222 let val = this.#counters.get(name) ?? 0;
223 // 1-indexed, as we're increasing the value before write
224 val++;
225 this.#counters.set(name, val);
226
227 return val;
228 }
229
230 #registerRollback(options: {
231 cacheKey: string;
232 rollbackFn: RollbackFn | undefined;
233 stepContext: WorkflowStepContext;
234 output?: unknown;
235 rollbackConfig?: WorkflowStepConfig;
236 }): void {
237 const { cacheKey, rollbackFn, stepContext, output, rollbackConfig } =
238 options;
239 if (rollbackFn && this.#rollbackStep === undefined) {
240 this.#engine.registerRollbackFn({
241 cacheKey,
242 fn: rollbackFn,
243 stepContext,
244 ...("output" in options && { output }),
245 ...(rollbackConfig !== undefined && { config: rollbackConfig }),
246 });
247 }
248 }
249
250 do(
251 name: string,
252 callback: (ctx: WorkflowStepContext) => Promise<unknown>,
253 rollbackOptions?: WorkflowStepRollbackOptions,
254 ): Promise<unknown>;
255 do(
256 name: string,
257 config: WorkflowStepConfig,
258 callback: (ctx: WorkflowStepContext) => Promise<unknown>,
259 rollbackOptions?: WorkflowStepRollbackOptions,
260 ): Promise<unknown>;
261
262 async do<T>(
263 name: string,
264 ...rest: Array<unknown>
265 ): Promise<unknown | void | undefined> {
266 let closure: (ctx: WorkflowStepContext) => Promise<T>;
267 let stepConfig: WorkflowStepConfig;
268 let rollbackOptions: WorkflowStepRollbackOptions | undefined;
269
270 const first = rest[0];
271 if (typeof first === "function") {
272 closure = first as (ctx: WorkflowStepContext) => Promise<T>;
273 stepConfig = {};
274 rollbackOptions = parseRollbackOptions(name, rest[1]);
275 } else {
276 stepConfig = (first ?? {}) as WorkflowStepConfig;
277 closure = rest[1] as (ctx: WorkflowStepContext) => Promise<T>;
278 if (typeof closure !== "function") {
279 const error = new WorkflowFatalError(
280 `Step "${name}" requires a callback function`,
281 ) as Error & UserErrorField;
282 error.isUserError = true;
283 throw error;
284 }
285 rollbackOptions = parseRollbackOptions(name, rest[2]);
286 }
287 const { rollback: rollbackFn, rollbackConfig } = rollbackOptions ?? {};
288
289 const isRollback = this.#rollbackStep !== undefined;
290 if (this.#engine.rollbackPhase === "rollback" && !isRollback) {
291 throw new WorkflowFatalError(
292 "Cannot execute steps during rollback phase",
293 );
294 }
295
296 const events = isRollback
297 ? {
298 start: InstanceEvent.ROLLBACK_STEP_START,
299 attemptStart: InstanceEvent.ROLLBACK_ATTEMPT_START,
300 attemptSuccess: InstanceEvent.ROLLBACK_ATTEMPT_SUCCESS,
301 attemptFailure: InstanceEvent.ROLLBACK_ATTEMPT_FAILURE,
302 success: InstanceEvent.ROLLBACK_STEP_SUCCESS,
303 failure: InstanceEvent.ROLLBACK_STEP_FAILURE,
304 }
305 : {
306 start: InstanceEvent.STEP_START,
307 attemptStart: InstanceEvent.ATTEMPT_START,
308 attemptSuccess: InstanceEvent.ATTEMPT_SUCCESS,
309 attemptFailure: InstanceEvent.ATTEMPT_FAILURE,
310 success: InstanceEvent.STEP_SUCCESS,
311 failure: InstanceEvent.STEP_FAILURE,
312 };
313
314 if (!isRollback) {
315 this.#lifetimeStepCounter++;
316
317 const stepLimit = this.#engine.stepLimit;
318 if (this.#lifetimeStepCounter > stepLimit) {
319 throw new WorkflowFatalError(
320 `The limit of ${stepLimit} steps has been reached. This limit can be changed in your worker configuration.`,
321 );
322 }
323 }
324
325 if (!isValidStepName(name)) {
326 // NOTE(lduarte): marking errors as user error allows the observability layer to avoid leaking
327 // user errors to sentry while making everything more observable. `isUserError` is not serialized
328 // into userland code due to how workerd serialzises errors over RPC - we also set it as undefined
329 // in the obs layer in case changes to workerd happen
330 const error = new WorkflowFatalError(
331 `Step name "${name}" exceeds max length (${MAX_STEP_NAME_LENGTH} chars) or invalid characters found`,
332 ) as Error & UserErrorField;
333 error.isUserError = true;
334 throw error;
335 }
336
337 if (!isValidStepConfig(stepConfig)) {
338 const error = new WorkflowFatalError(
339 `Step config for "${name}" is in a invalid format. See https://developers.cloudflare.com/workflows/build/sleeping-and-retrying/`,
340 ) as Error & UserErrorField;
341 error.isUserError = true;
342 throw error;
343 }
344
345 let liveDelay: WorkflowDelayDuration | WorkflowDelayFunction =
346 stepConfig.retries?.delay ?? DEFAULT_RETRY_DELAY_MS;
347
348 let config: ResolvedStepConfig = {
349 ...defaultConfig,
350 ...stepConfig,
351 retries: {
352 ...defaultConfig.retries,
353 ...stepConfig.retries,
354 delay:
355 typeof liveDelay === "function"
356 ? SERIALIZABLE_DELAY_MARKER
357 : liveDelay,
358 },
359 };
360
361 let cacheKey: string;
362 let count: number;
363 let stepNameWithCounter: string;
364 const rollbackStep = this.#rollbackStep;
365 if (rollbackStep !== undefined) {
366 cacheKey = `${ROLLBACK_CACHE_KEY_PREFIX}${rollbackStep.cacheKey}`;
367 count = 1;
368 stepNameWithCounter = name;
369 } else {
370 const hash = await computeHash(name);
371 count = this.#getCount("run-" + name);
372 cacheKey = `${hash}-${count}`;
373 stepNameWithCounter = `${name}-${count}`;
374 }
375
376 const valueKey = `${cacheKey}-value`;
377 const streamMetaKey = getStreamOutputMetaKey(cacheKey);
378 const configKey = `${cacheKey}-config`;
379 const errorKey = `${cacheKey}-error`;
380 const stepStateKey = `${cacheKey}-metadata`;
381 const retryDelayDisableKey = `${MODIFIER_KEYS.DISABLE_RETRY_DELAY}${valueKey}`;
382
383 const maybeMap = await this.#state.storage.get([
384 valueKey,
385 streamMetaKey,
386 configKey,
387 errorKey,
388 stepStateKey,
389 ]);
390
391 // Check cache -- streams first, then plain values
392 const maybeStreamMeta = maybeMap.get(streamMetaKey) as
393 | StreamOutputMeta
394 | undefined
395 | null;
396 const cachedConfig = maybeMap.get(configKey) as
397 | ResolvedStepConfig
398 | undefined;
399 if (maybeStreamMeta?.state === StreamOutputState.Complete) {
400 const cachedState = maybeMap.get(stepStateKey) as StepState | undefined;
401 const maybeOutputError = getInvalidStoredStreamOutputError(
402 this.#state.storage,
403 cacheKey,
404 maybeStreamMeta,
405 );
406 if (maybeOutputError !== undefined) {
407 throw new WorkflowInternalError(
408 `Stored output for ${stepNameWithCounter} is corrupt or incomplete.`,
409 );
410 }
411
412 const result = createReplayReadableStream({
413 storage: this.#state.storage,
414 cacheKey,
415 meta: maybeStreamMeta,
416 }) as T;
417 this.#registerRollback({
418 cacheKey,
419 rollbackFn,
420 stepContext: {
421 step: { name, count },
422 attempt: cachedState?.attemptedCount ?? 1,
423 config: toEngineStepConfig(cachedConfig ?? config),
424 },
425 output: result,
426 rollbackConfig,
427 });
428 return result;
429 } else if (maybeStreamMeta !== undefined && maybeStreamMeta !== null) {
430 // We're not in a complete state - means we crashed while persisting a stream on a previous invocation - need to cleanup
431 await cleanupPendingStreamOutput(this.#state.storage, cacheKey).catch(
432 () => {},
433 );
434 }
435
436 const maybeResult = maybeMap.get(valueKey);
437
438 if (maybeResult) {
439 const cachedState = maybeMap.get(stepStateKey) as StepState | undefined;
440 const result = (maybeResult as { value: T }).value;
441 this.#registerRollback({
442 cacheKey,
443 rollbackFn,
444 stepContext: {
445 step: { name, count },
446 attempt: cachedState?.attemptedCount ?? 1,
447 config: toEngineStepConfig(cachedConfig ?? config),
448 },
449 output: result,
450 rollbackConfig,
451 });
452 return result;
453 }
454
455 const maybeError: (Error & UserErrorField) | undefined = maybeMap.get(
456 errorKey,
457 ) as Error | undefined;
458
459 if (maybeError) {
460 const cachedState = maybeMap.get(stepStateKey) as StepState | undefined;
461 this.#registerRollback({
462 cacheKey,
463 rollbackFn,
464 stepContext: {
465 step: { name, count },
466 attempt: cachedState?.attemptedCount ?? 1,
467 config: toEngineStepConfig(cachedConfig ?? config),
468 },
469 rollbackConfig,
470 });
471 maybeError.isUserError = true;
472 throw maybeError;
473 }
474
475 // Persist initial config because user can pass in dynamic config
476 if (cachedConfig === undefined) {
477 await this.#state.storage.put(configKey, config);
478 } else {
479 config = cachedConfig;
480 // Recover the live delay from the (non-serializable) user config: a
481 // persisted marker means the user passed a function, so re-read it.
482 liveDelay =
483 config.retries.delay === SERIALIZABLE_DELAY_MARKER
484 ? typeof stepConfig.retries?.delay === "function"
485 ? stepConfig.retries.delay
486 : DEFAULT_RETRY_DELAY_MS
487 : config.retries.delay;
488 }
489
490 if (this.#engine.rollbackPhase === "replay" && !isRollback) {
491 const cachedState = maybeMap.get(stepStateKey) as StepState | undefined;
492 this.#registerRollback({
493 cacheKey,
494 rollbackFn,
495 stepContext: {
496 step: { name, count },
497 attempt: cachedState?.attemptedCount ?? 1,
498 config: toEngineStepConfig(config),
499 },
500 rollbackConfig,
501 });
502 return undefined;
503 }
504
505 const attemptLogs = this.#engine
506 .readLogsFromStep(cacheKey)
507 .filter((val) =>
508 [
509 events.attemptSuccess,
510 events.attemptFailure,
511 events.attemptStart,
512 ].includes(val.event),
513 );
514
515 // this means that the the engine died while executing this step - we can mark the latest attempt as failed
516 if (
517 attemptLogs.length > 0 &&
518 attemptLogs.at(-1)?.event === events.attemptStart
519 ) {
520 // TODO: We should get this from SQL
521 const stepState = ((await this.#state.storage.get(
522 stepStateKey,
523 )) as StepState) ?? {
524 attemptedCount: 1,
525 };
526
527 const priorityQueueHash = `${cacheKey}-${stepState.attemptedCount}`;
528
529 // @ts-expect-error priorityQueue is initiated in init
530 const timeoutEntryPQ = this.#engine.priorityQueue.getFirst(
531 (a) => a.hash === priorityQueueHash && a.type === "timeout",
532 );
533 // if there's a timeout on the PQ we pop it, because we wont need it
534 if (timeoutEntryPQ !== undefined) {
535 // @ts-expect-error priorityQueue is initiated in init
536 this.#engine.priorityQueue.remove(timeoutEntryPQ);
537 }
538 this.#engine.writeLog(
539 events.attemptFailure,
540 cacheKey,
541 stepNameWithCounter,
542 {
543 attempt: stepState.attemptedCount,
544 error: {
545 name: "WorkflowInternalError",
546 message: `Attempt failed due to internal workflows error`,
547 },
548 },
549 );
550
551 await this.#state.storage.put(stepStateKey, stepState);
552 }
553
554 const doWrapper = async (
555 doWrapperClosure: (ctx: WorkflowStepContext) => Promise<unknown>,
556 ): Promise<unknown | void | undefined> => {
557 const stepState = ((await this.#state.storage.get(
558 stepStateKey,
559 )) as StepState) ?? {
560 attemptedCount: 0,
561 };
562 const forwardStepContext = (): WorkflowStepContext => ({
563 step: { name, count },
564 attempt: stepState.attemptedCount,
565 config: toEngineStepConfig(config),
566 });
567
568 // NOTE(caio): this might be a stream returning step - if so cleanup stale data from previous lifetimes
569 await cleanupPendingStreamOutput(this.#state.storage, cacheKey).catch(
570 () => {},
571 );
572
573 await this.#engine.timeoutHandler.acquire(this.#engine);
574
575 if (stepState.attemptedCount == 0) {
576 this.#engine.writeLog(events.start, cacheKey, stepNameWithCounter, {
577 config: toEngineStepConfig(config),
578 ...(!isRollback && rollbackFn ? { hasRollback: true } : {}),
579 });
580 } else {
581 // in case the engine dies while retrying and wakes up before the retry period
582 const priorityQueueHash = `${cacheKey}-${stepState.attemptedCount}`;
583 // @ts-expect-error priorityQueue is initiated in init
584 const retryEntryPQ = this.#engine.priorityQueue.getFirst(
585 (a) => a.hash === priorityQueueHash && a.type === "retry",
586 );
587 // complete sleep if it didn't finish for some reason
588 if (retryEntryPQ !== undefined) {
589 const disableAllRetryDelays = await this.#state.storage.get(
590 MODIFIER_KEYS.DISABLE_ALL_RETRY_DELAYS,
591 );
592 const disableThisRetryDelay =
593 await this.#state.storage.get(retryDelayDisableKey);
594 const disableRetryDelay =
595 disableAllRetryDelays || disableThisRetryDelay;
596
597 await this.#engine.timeoutHandler.release(this.#engine);
598 await scheduler.wait(
599 disableRetryDelay ? 0 : retryEntryPQ.targetTimestamp - Date.now(),
600 );
601 await this.#engine.timeoutHandler.acquire(this.#engine);
602 // @ts-expect-error priorityQueue is initiated in init
603 this.#engine.priorityQueue.remove({
604 hash: priorityQueueHash,
605 type: "retry",
606 });
607 }
608 }
609
610 let result;
611
612 const instanceMetadata =
613 await this.#state.storage.get<InstanceMetadata>(INSTANCE_METADATA);
614 if (!instanceMetadata) {
615 throw new Error("instanceMetadata is undefined");
616 }
617 const { accountId, instance } = instanceMetadata;
618
619 let streamResultSeen = false;
620 let lastStreamMeta: StreamOutputMeta | undefined;
621 const abortController = new AbortController();
622 const stepExecutionSignal = AbortSignal.any([
623 abortController.signal,
624 this.#engine.engineAbortController.signal,
625 ]);
626
627 try {
628 const timeoutPromise = async (): Promise<never> => {
629 const priorityQueueHash = `${cacheKey}-${stepState.attemptedCount}`;
630 let timeout = ms(config.timeout);
631 if (forceStepTimeout) {
632 timeout = 0;
633 }
634 // @ts-expect-error priorityQueue is initiated in init
635 await this.#engine.priorityQueue.add({
636 hash: priorityQueueHash,
637 targetTimestamp: Date.now() + timeout,
638 type: "timeout",
639 });
640 await scheduler.wait(timeout);
641 // if we reach here, means that we can try to delete the timeout from the PQ
642 // because we managed to wait in the same lifetime
643 // @ts-expect-error priorityQueue is initiated in init
644 await this.#engine.priorityQueue.remove({
645 hash: priorityQueueHash,
646 type: "timeout",
647 });
648 const error = new WorkflowTimeoutError(
649 `Execution timed out after ${timeout}ms`,
650 );
651 abortController.abort(error);
652 throw error;
653 };
654
655 this.#engine.writeLog(
656 events.attemptStart,
657 cacheKey,
658 stepNameWithCounter,
659 {
660 attempt: stepState.attemptedCount + 1,
661 },
662 );
663 stepState.attemptedCount++;
664 this.#registerRollback({
665 cacheKey,
666 rollbackFn,
667 stepContext: forwardStepContext(),
668 rollbackConfig,
669 });
670 await this.#state.storage.put(stepStateKey, stepState);
671 const priorityQueueHash = `${cacheKey}-${stepState.attemptedCount}`;
672
673 const mockErrorKey = `${MODIFIER_KEYS.MOCK_STEP_ERROR}${valueKey}`;
674 const persistentMockError = await this.#state.storage.get<{
675 name: string;
676 message: string;
677 }>(mockErrorKey);
678 const transientMockError = await this.#state.storage.get<{
679 name: string;
680 message: string;
681 }>(`${mockErrorKey}-${stepState.attemptedCount}`);
682 const mockErrorPayload = persistentMockError || transientMockError;
683
684 // if a mocked error exists, throw it immediately
685 if (mockErrorPayload) {
686 const errorToThrow = new Error(mockErrorPayload.message);
687 errorToThrow.name = mockErrorPayload.name;
688 throw errorToThrow;
689 }
690
691 const replaceResult = await this.#state.storage.get(
692 `${MODIFIER_KEYS.REPLACE_RESULT}${valueKey}`,
693 );
694
695 const forceStepTimeoutKey = `${MODIFIER_KEYS.FORCE_STEP_TIMEOUT}${valueKey}`;
696 const persistentStepTimeout =
697 await this.#state.storage.get(forceStepTimeoutKey);
698 const transientStepTimeout = await this.#state.storage.get(
699 `${forceStepTimeoutKey}-${stepState.attemptedCount}`,
700 );
701 const forceStepTimeout = persistentStepTimeout || transientStepTimeout;
702
703 let timeoutTask: Promise<never> | undefined;
704
705 const persistStepResult = async (
706 value: unknown,
707 activeTimeoutTask?: Promise<never>,
708 ): Promise<unknown> => {
709 if (!isReadableStreamLike(value)) {
710 // Typed-array views anywhere in the value tree are copied
711 // into a tight backing buffer so the full backing buffer
712 // does not ride along with each view (issue #14101). View
713 // types are preserved so cached replays observe the same
714 // constructor as the live execution path. The caller still
715 // receives the original `value` below — only the stored
716 // shape changes.
717 const stored = normalizeForStorage(value);
718 await this.#state.storage.put(valueKey, { value: stored });
719 abortController.abort("step finished");
720 // @ts-expect-error priorityQueue is initiated in init
721 this.#engine.priorityQueue.remove({
722 hash: priorityQueueHash,
723 type: "timeout",
724 });
725 return value;
726 }
727
728 streamResultSeen = true;
729 const streamMeta = await writeStreamOutput({
730 storage: this.#state.storage,
731 cacheKey,
732 attempt: stepState.attemptedCount,
733 stream: value as ReadableStream<unknown>,
734 signal: stepExecutionSignal,
735 timeoutTask: activeTimeoutTask,
736 });
737 lastStreamMeta = streamMeta;
738
739 abortController.abort("step finished");
740 // @ts-expect-error priorityQueue is initiated in init
741 this.#engine.priorityQueue.remove({
742 hash: priorityQueueHash,
743 type: "timeout",
744 });
745 return createReplayReadableStream({
746 storage: this.#state.storage,
747 cacheKey,
748 meta: streamMeta,
749 });
750 };
751
752 if (forceStepTimeout) {
753 result = await timeoutPromise();
754 } else if (replaceResult) {
755 // Check if the mocked result is a stream sentinel (from mockStepResult with ReadableStream)
756 if (
757 replaceResult &&
758 typeof replaceResult === "object" &&
759 (replaceResult as Record<string, unknown>).__mockStreamOutput
760 ) {
761 const sentinel = replaceResult as {
762 __mockStreamOutput: true;
763 cacheKey: string;
764 meta: StreamOutputMeta;
765 };
766 result = createReplayReadableStream({
767 storage: this.#state.storage,
768 cacheKey: sentinel.cacheKey,
769 meta: sentinel.meta,
770 });
771 } else {
772 result = replaceResult;
773 }
774 } else {
775 timeoutTask = timeoutPromise();
776 result = await Promise.race([
777 doWrapperClosure({
778 step: { name, count },
779 attempt: stepState.attemptedCount,
780 config: toEngineStepConfig(config),
781 }),
782 timeoutTask,
783 ]);
784 }
785
786 // We store the value of `output` in an object with a `value` property. This allows us to store `undefined`,
787 // in the case that it's returned from the user's code. This is because DO storage will error if you try to
788 // store undefined directly.
789 try {
790 result = await persistStepResult(result, timeoutTask);
791 } catch (e) {
792 abortController.abort("step errored");
793 // @ts-expect-error priorityQueue is initiated in init
794 this.#engine.priorityQueue.remove({
795 hash: priorityQueueHash,
796 type: "timeout",
797 });
798
799 if (e instanceof WorkflowTimeoutError) {
800 throw e;
801 }
802
803 // Fatal serialization/storage errors abort the DO immediately, so
804 // previously registered rollbacks do not run for these paths.
805 // This matches the existing terminal behavior for unrecoverable output.
806 // Stream-specific fatal errors
807 if (
808 e instanceof InvalidStepReadableStreamError ||
809 e instanceof OversizedStreamChunkError ||
810 e instanceof UnsupportedStreamChunkError
811 ) {
812 this.#engine.writeLog(
813 events.attemptFailure,
814 cacheKey,
815 stepNameWithCounter,
816 {
817 attempt: stepState.attemptedCount,
818 error: new WorkflowFatalError(e.message),
819 },
820 );
821 this.#engine.writeLog(
822 events.failure,
823 cacheKey,
824 stepNameWithCounter,
825 {},
826 );
827 this.#engine.writeLog(InstanceEvent.WORKFLOW_FAILURE, null, null, {
828 error: new WorkflowFatalError(
829 `The execution of the Workflow instance was terminated, as the step "${name}" returned an invalid ReadableStream output. ${e.message}`,
830 ),
831 });
832
833 await this.#engine.setStatus(
834 accountId,
835 instance.id,
836 InstanceStatus.Errored,
837 );
838 await this.#engine.timeoutHandler.release(this.#engine);
839 await this.#engine.abort(ABORT_REASONS.NOT_SERIALISABLE);
840 return;
841 }
842
843 if (e instanceof StreamOutputStorageLimitError) {
844 this.#engine.writeLog(
845 events.attemptFailure,
846 cacheKey,
847 stepNameWithCounter,
848 {
849 attempt: stepState.attemptedCount,
850 error: new WorkflowFatalError(e.message),
851 },
852 );
853 this.#engine.writeLog(
854 events.failure,
855 cacheKey,
856 stepNameWithCounter,
857 {},
858 );
859 this.#engine.writeLog(InstanceEvent.WORKFLOW_FAILURE, null, null, {
860 error: new WorkflowFatalError(
861 "The instance has exceeded the 1GiB storage limit",
862 ),
863 });
864
865 await this.#engine.setStatus(
866 accountId,
867 instance.id,
868 InstanceStatus.Errored,
869 );
870 await this.#engine.timeoutHandler.release(this.#engine);
871 await this.#engine.abort(ABORT_REASONS.STORAGE_LIMIT_EXCEEDED);
872 return;
873 }
874
875 // something that cannot be written to storage
876 if (e instanceof Error && e.name === "DataCloneError") {
877 this.#engine.writeLog(
878 events.attemptFailure,
879 cacheKey,
880 stepNameWithCounter,
881 {
882 attempt: stepState.attemptedCount,
883 error: new WorkflowFatalError(
884 `Value returned from step "${name}" is not serialisable`,
885 ),
886 },
887 );
888 this.#engine.writeLog(
889 events.failure,
890 cacheKey,
891 stepNameWithCounter,
892 {},
893 );
894 this.#engine.writeLog(InstanceEvent.WORKFLOW_FAILURE, null, null, {
895 error: new WorkflowFatalError(
896 `The execution of the Workflow instance was terminated, as the step "${name}" returned a value which is not serialisable`,
897 ),
898 });
899
900 await this.#engine.setStatus(
901 accountId,
902 instance.id,
903 InstanceStatus.Errored,
904 );
905 await this.#engine.timeoutHandler.release(this.#engine);
906 await this.#engine.abort(ABORT_REASONS.NOT_SERIALISABLE);
907 } else if (
908 e instanceof Error &&
909 e.message.includes("string or blob too big: SQLITE_TOOBIG")
910 ) {
911 throw new WorkflowInternalError(
912 `Step ${stepNameWithCounter} output is too large. Maximum allowed size is 1MiB.`,
913 );
914 } else {
915 // TODO (WOR-77): Send this to Sentry
916 throw new WorkflowInternalError(
917 `Storage failure for ${stepNameWithCounter} due to internal error.`,
918 );
919 }
920 return;
921 }
922
923 // if we reach here, means that the closure ran successfully and we can remove the timeout from the PQ
924 // @ts-expect-error priorityQueue is initiated in init
925 this.#engine.priorityQueue.remove({
926 hash: priorityQueueHash,
927 type: "timeout",
928 });
929
930 this.#engine.writeLog(
931 events.attemptSuccess,
932 cacheKey,
933 stepNameWithCounter,
934 {
935 attempt: stepState.attemptedCount,
936 },
937 );
938 } catch (e) {
939 const error = e as Error;
940 // if we reach here, means that the closure ran but errored out and we can remove the timeout from the PQ
941 // @ts-expect-error priorityQueue is initiated in init
942 this.#engine.priorityQueue.remove({
943 hash: `${cacheKey}-${stepState.attemptedCount}`,
944 type: "timeout",
945 });
946
947 // Clean up any partial stream output from this failed attempt
948 if (streamResultSeen) {
949 try {
950 await rollbackStreamOutput(
951 this.#state.storage,
952 cacheKey,
953 stepState.attemptedCount,
954 );
955 } catch {
956 // Best-effort cleanup
957 }
958 }
959
960 if (
961 e instanceof Error &&
962 (error.name === "NonRetryableError" ||
963 error.message.startsWith("NonRetryableError"))
964 ) {
965 const attemptError = shouldPreserveNonRetryableError()
966 ? new PreservedNonRetryableError(e)
967 : new WorkflowFatalError(
968 `Step threw a NonRetryableError with message "${e.message}"`,
969 );
970
971 this.#engine.writeLog(
972 events.attemptFailure,
973 cacheKey,
974 stepNameWithCounter,
975 {
976 attempt: stepState.attemptedCount,
977 error: attemptError,
978 },
979 );
980 this.#engine.writeLog(
981 events.failure,
982 cacheKey,
983 stepNameWithCounter,
984 !isRollback && rollbackFn ? { hasRollback: true } : {},
985 );
986 this.#registerRollback({
987 cacheKey,
988 rollbackFn,
989 stepContext: forwardStepContext(),
990 rollbackConfig,
991 });
992
993 throw error;
994 }
995
996 await this.#state.storage.put(stepStateKey, stepState);
997
998 const willRetry = stepState.attemptedCount <= config.retries.limit;
999 const priorityQueueHash = `${cacheKey}-${stepState.attemptedCount}`;
1000
1001 // Resolve the delay once per attempt. A broken delay function turns
1002 // the retry into a non-retryable failure.
1003 let retryDelayMs: number | undefined;
1004 let delayFailure: (Error & UserErrorField) | undefined;
1005 if (willRetry) {
1006 try {
1007 let resolvedDelay: unknown;
1008 if (typeof liveDelay === "function") {
1009 // Park a default-delayed retry before invoking the user's
1010 // delay function so a crash while the function is in flight
1011 // still reschedules the step. The PQ table is append-only
1012 // with a UNIQUE(action, entryType, hash) constraint, so this
1013 // exact entry is reused as the retry entry below — it is
1014 // never removed and re-added (that would violate UNIQUE).
1015 // @ts-expect-error priorityQueue is initiated in init
1016 await this.#engine.priorityQueue.add({
1017 hash: priorityQueueHash,
1018 targetTimestamp: Date.now() + DEFAULT_RETRY_DELAY_MS,
1019 type: "retry",
1020 });
1021
1022 const stepCtx: WorkflowStepContext = {
1023 step: { name, count },
1024 attempt: stepState.attemptedCount,
1025 config: toEngineStepConfig(config),
1026 };
1027
1028 resolvedDelay = await invokeDelayFunction(
1029 liveDelay,
1030 { ctx: stepCtx, error },
1031 {
1032 timeoutMs: DELAY_FUNCTION_TIMEOUT_MS,
1033 wait: schedulerWait,
1034 signal: this.#engine.engineAbortController.signal,
1035 },
1036 );
1037 } else {
1038 resolvedDelay = liveDelay;
1039 }
1040 retryDelayMs = calcRetryDuration(config, stepState, resolvedDelay);
1041 } catch (delayErr) {
1042 if (!(delayErr instanceof DelayFunctionError)) {
1043 throw delayErr;
1044 }
1045 // @ts-expect-error priorityQueue is initiated in init
1046 this.#engine.priorityQueue.remove({
1047 hash: priorityQueueHash,
1048 type: "retry",
1049 });
1050 const userError = new NonRetryableDelayError(
1051 `The delay function for step "${stepNameWithCounter}" ${delayErr.message}`,
1052 ) as Error & UserErrorField;
1053 userError.isUserError = true;
1054 delayFailure = userError;
1055 }
1056 }
1057
1058 this.#engine.writeLog(
1059 events.attemptFailure,
1060 cacheKey,
1061 stepNameWithCounter,
1062 {
1063 attempt: stepState.attemptedCount,
1064 error: {
1065 name: error.name,
1066 message: error.message,
1067 // TODO (WOR-79): Stacks are all incorrect over RPC and need work
1068 // stack: error.stack,
1069 },
1070 ...(retryDelayMs !== undefined ? { retryDelayMs } : {}),
1071 },
1072 );
1073
1074 if (willRetry && delayFailure === undefined) {
1075 // TODO (WOR-71): Think through if every Error should transition
1076 const durationMs = retryDelayMs ?? DEFAULT_RETRY_DELAY_MS;
1077 const disableAllRetryDelays = await this.#state.storage.get(
1078 MODIFIER_KEYS.DISABLE_ALL_RETRY_DELAYS,
1079 );
1080 const disableThisRetryDelay =
1081 await this.#state.storage.get(retryDelayDisableKey);
1082 const disableRetryDelay =
1083 disableAllRetryDelays || disableThisRetryDelay;
1084 const effectiveDuration = disableRetryDelay ? 0 : durationMs;
1085
1086 // A dynamic delay already parked its retry entry under this hash
1087 // above (the append-only PQ rejects re-adding the same hash), so
1088 // only static delays add here. The live wait below uses
1089 // effectiveDuration regardless; the parked entry's timestamp only
1090 // drives crash recovery.
1091 //
1092 // Known trade-off: for a dynamic delay the parked entry keeps the
1093 // DEFAULT_RETRY_DELAY_MS placeholder timestamp set before the delay
1094 // function ran (see the add above) — the UNIQUE(action, entryType,
1095 // hash) constraint means it can't be overwritten in place with the
1096 // computed value without a remove+re-add that would violate the
1097 // constraint. So if the engine crashes mid-wait, recovery resumes
1098 // the step at the placeholder delay rather than the computed one.
1099 // This is accepted for local dev: the step still retries with the
1100 // correct semantics, only the post-crash wait can be shorter.
1101 if (typeof liveDelay !== "function") {
1102 // @ts-expect-error priorityQueue is initiated in init
1103 await this.#engine.priorityQueue.add({
1104 hash: priorityQueueHash,
1105 targetTimestamp: Date.now() + effectiveDuration,
1106 type: "retry",
1107 });
1108 }
1109 await this.#engine.timeoutHandler.release(this.#engine);
1110 // Race retry wait against the pause signal so pause
1111 // takes effect immediately during retries
1112 {
1113 const retryPauseSignal = this.#engine.pauseController.signal;
1114 await raceAgainstAbort(
1115 scheduler.wait(effectiveDuration),
1116 retryPauseSignal,
1117 );
1118 const retryStatus = await this.#engine.getStatus();
1119 const pausedDuringRetry =
1120 retryStatus === InstanceStatus.Paused ||
1121 retryStatus === InstanceStatus.WaitingForPause;
1122 if (pausedDuringRetry) {
1123 throw new Error(ABORT_REASONS.USER_PAUSE);
1124 }
1125 }
1126
1127 // if it ever reaches here, we can try to remove it from the priority queue since it's no longer useful
1128 // @ts-expect-error priorityQueue is initiated in init
1129 this.#engine.priorityQueue.remove({
1130 hash: priorityQueueHash,
1131 type: "retry",
1132 });
1133
1134 return doWrapper(doWrapperClosure);
1135 } else {
1136 await this.#engine.timeoutHandler.release(this.#engine);
1137 // Clean up any leftover stream chunks on retry exhaustion
1138 try {
1139 await rollbackStreamOutput(
1140 this.#state.storage,
1141 cacheKey,
1142 stepState.attemptedCount,
1143 );
1144 } catch {
1145 // Best-effort cleanup
1146 }
1147 this.#engine.writeLog(
1148 events.failure,
1149 cacheKey,
1150 stepNameWithCounter,
1151 !isRollback && rollbackFn ? { hasRollback: true } : {},
1152 );
1153 this.#registerRollback({
1154 cacheKey,
1155 rollbackFn,
1156 stepContext: forwardStepContext(),
1157 rollbackConfig,
1158 });
1159
1160 const finalError = delayFailure ?? error;
1161 await this.#state.storage.put(errorKey, finalError);
1162 throw finalError;
1163 }
1164 }
1165
1166 const redactOutput = config.sensitive === SENSITIVE_STEP_OUTPUT;
1167 this.#engine.writeLog(events.success, cacheKey, stepNameWithCounter, {
1168 // TODO (WOR-86): Add limits, figure out serialization
1169 result: redactOutput
1170 ? REDACTED_STEP_OUTPUT
1171 : lastStreamMeta
1172 ? undefined
1173 : result,
1174 ...(!redactOutput &&
1175 lastStreamMeta && {
1176 streamOutput: { cacheKey, meta: lastStreamMeta },
1177 }),
1178 ...(!isRollback && rollbackFn ? { hasRollback: true } : {}),
1179 });
1180 this.#registerRollback({
1181 cacheKey,
1182 rollbackFn,
1183 stepContext: forwardStepContext(),
1184 output: result,
1185 rollbackConfig,
1186 });
1187 await this.#engine.timeoutHandler.release(this.#engine);
1188 return result;
1189 };
1190
1191 const result = await doWrapper(closure);
1192
1193 // Check if a pause was requested while this step was running.
1194 // Rollback steps may run while the instance is paused; don't let stale pause
1195 // state abort rollback cleanup after the rollback step itself succeeds.
1196 if (!isRollback) {
1197 await this.#checkForPendingPause();
1198 }
1199
1200 return result;
1201 }
1202
1203 async sleep(name: string, duration: WorkflowSleepDuration): Promise<void> {
1204 if (this.#engine.rollbackPhase === "replay") {
1205 return;
1206 }
1207
1208 if (typeof duration == "string") {
1209 duration = ms(duration);
1210 }
1211
1212 const hash = await computeHash(name + duration.toString());
1213 const count = this.#getCount("sleep-" + name + duration.toString());
1214 const cacheKey = `${hash}-${count}`;
1215 const sleepNameWithCounter = `${name}-${count}`;
1216
1217 const sleepKey = `${cacheKey}-value`;
1218 const sleepLogWrittenKey = `${cacheKey}-log-written`;
1219 const maybeResult = await this.#state.storage.get(sleepKey);
1220
1221 const sleepNameCountHash = await computeHash(
1222 name + this.#getCount("sleep-" + name),
1223 );
1224 const disableThisSleep = await this.#state.storage.get(
1225 `${MODIFIER_KEYS.DISABLE_SLEEP}${sleepNameCountHash}`,
1226 );
1227 const disableAllSleeps = await this.#state.storage.get(
1228 MODIFIER_KEYS.DISABLE_ALL_SLEEPS,
1229 );
1230
1231 const disableSleep = disableAllSleeps || disableThisSleep;
1232
1233 if (maybeResult != undefined) {
1234 // @ts-expect-error priorityQueue is initiated in init
1235 const entryPQ = this.#engine.priorityQueue.getFirst(
1236 (a) => a.hash === cacheKey && a.type === "sleep",
1237 );
1238 // in case the engine dies while sleeping and wakes up before the retry period
1239 if (entryPQ !== undefined) {
1240 await scheduler.wait(
1241 disableSleep ? 0 : entryPQ.targetTimestamp - Date.now(),
1242 );
1243 // @ts-expect-error priorityQueue is initiated in init
1244 this.#engine.priorityQueue.remove({ hash: cacheKey, type: "sleep" });
1245 }
1246 const shouldWriteLog =
1247 (await this.#state.storage.get(sleepLogWrittenKey)) == undefined;
1248 if (shouldWriteLog) {
1249 this.#engine.writeLog(
1250 InstanceEvent.SLEEP_COMPLETE,
1251 cacheKey,
1252 sleepNameWithCounter,
1253 {},
1254 );
1255 await this.#state.storage.put(sleepLogWrittenKey, true);
1256 }
1257 return;
1258 }
1259
1260 this.#engine.writeLog(
1261 InstanceEvent.SLEEP_START,
1262 cacheKey,
1263 sleepNameWithCounter,
1264 {
1265 durationMs: duration,
1266 },
1267 );
1268 const instanceMetadata =
1269 await this.#state.storage.get<InstanceMetadata>(INSTANCE_METADATA);
1270 if (!instanceMetadata) {
1271 throw new Error("instanceMetadata is undefined");
1272 }
1273
1274 // TODO(lduarte): not sure of this order of operations
1275 await this.#state.storage.put(sleepKey, true); // Any value will do for cache hit
1276
1277 // @ts-expect-error priorityQueue is initiated in init
1278 await this.#engine.priorityQueue.add({
1279 hash: cacheKey,
1280 targetTimestamp: Date.now() + (disableSleep ? 0 : duration),
1281 type: "sleep",
1282 });
1283
1284 // Race the sleep against the pause signal
1285 const pauseSignal = this.#engine.pauseController.signal;
1286 const sleepDuration = disableSleep ? 0 : duration;
1287
1288 await raceAgainstAbort(scheduler.wait(sleepDuration), pauseSignal);
1289
1290 // Check if we were paused during the sleep
1291 const statusAfterSleep = await this.#engine.getStatus();
1292 const pausedDuringSleep =
1293 statusAfterSleep === InstanceStatus.Paused ||
1294 statusAfterSleep === InstanceStatus.WaitingForPause;
1295
1296 if (pausedDuringSleep) {
1297 // Throw pause error
1298 throw new Error(ABORT_REASONS.USER_PAUSE);
1299 }
1300
1301 this.#engine.writeLog(
1302 InstanceEvent.SLEEP_COMPLETE,
1303 cacheKey,
1304 sleepNameWithCounter,
1305 {},
1306 );
1307 await this.#state.storage.put(sleepLogWrittenKey, true);
1308
1309 // @ts-expect-error priorityQueue is initiated in init
1310 this.#engine.priorityQueue.remove({ hash: cacheKey, type: "sleep" });
1311 }
1312
1313 async sleepUntil(name: string, timestamp: Date | number): Promise<void> {
1314 if (this.#engine.rollbackPhase === "replay") {
1315 return;
1316 }
1317
1318 if (timestamp instanceof Date) {
1319 timestamp = timestamp.valueOf();
1320 }
1321
1322 const now = Date.now();
1323 // timestamp needs to be in the future, throw if not
1324 if (timestamp < now) {
1325 throw new Error(
1326 "You can't sleep until a time in the past, time-traveler",
1327 );
1328 }
1329
1330 return this.sleep(name, timestamp - now);
1331 }
1332
1333 async waitForEvent<T>(
1334 name: string,
1335 options: {
1336 type: string;
1337 timeout?: string | number;
1338 },
1339 ): Promise<WorkflowStepEvent<T>> {
1340 if (this.#engine.rollbackPhase === "rollback") {
1341 throw new WorkflowFatalError(
1342 "Cannot execute steps during rollback phase",
1343 );
1344 }
1345
1346 if (!options.timeout) {
1347 options.timeout = "24 hours";
1348 }
1349
1350 const count = this.#getCount("waitForEvent-" + name);
1351 const waitForEventNameWithCounter = `${name}-${count}`;
1352 const hash = await computeHash(waitForEventNameWithCounter);
1353 const cacheKey = `${hash}-${count}`;
1354 const waitForEventKey = `${cacheKey}-value`;
1355 const errorKey = `${cacheKey}-error`;
1356
1357 const pendingWaiterRegistered = `${cacheKey}-pending`;
1358
1359 const timeoutError = new WorkflowTimeoutError(
1360 `Execution timed out after ${ms(options.timeout)}ms`,
1361 ) as Error & UserErrorField;
1362
1363 const maybeResult = await this.#state.storage.get<Event>(waitForEventKey);
1364
1365 if (this.#engine.rollbackPhase === "replay") {
1366 return maybeResult as WorkflowStepEvent<T>;
1367 }
1368
1369 if (maybeResult) {
1370 const shouldWriteLog =
1371 (await this.#state.storage.get(waitForEventKey)) == undefined;
1372 if (shouldWriteLog) {
1373 this.#engine.writeLog(
1374 InstanceEvent.WAIT_COMPLETE,
1375 cacheKey,
1376 waitForEventNameWithCounter,
1377 maybeResult,
1378 );
1379 }
1380 return maybeResult as WorkflowStepEvent<T>;
1381 }
1382 const maybeError: (Error & UserErrorField) | undefined =
1383 (await this.#state.storage.get(errorKey)) as Error | undefined;
1384
1385 if (maybeError) {
1386 maybeError.isUserError = true;
1387 throw maybeError;
1388 }
1389
1390 const maybeRegistered = await this.#state.storage.get(
1391 pendingWaiterRegistered,
1392 );
1393
1394 if (!maybeRegistered) {
1395 this.#engine.writeLog(
1396 InstanceEvent.WAIT_START,
1397 cacheKey,
1398 waitForEventNameWithCounter,
1399 {
1400 event: options.type,
1401 },
1402 );
1403
1404 await this.#state.storage.put(pendingWaiterRegistered, true);
1405 }
1406
1407 // if there's a timeout on the PQ we pop it, because we wont need it
1408 // @ts-expect-error priorityQueue is initiated in init
1409 const timeoutEntryPQ = this.#engine.priorityQueue.getFirst(
1410 (a) => a.hash === cacheKey && a.type === "timeout",
1411 );
1412 const forceEventTimeout = await this.#state.storage.get(
1413 `${MODIFIER_KEYS.FORCE_EVENT_TIMEOUT}${waitForEventKey}`,
1414 );
1415 if (
1416 (timeoutEntryPQ === undefined &&
1417 this.#engine.priorityQueue !== undefined &&
1418 this.#engine.priorityQueue.checkIfExistedInPast({
1419 hash: cacheKey,
1420 type: "timeout",
1421 })) ||
1422 (timeoutEntryPQ !== undefined &&
1423 timeoutEntryPQ.targetTimestamp < Date.now()) ||
1424 forceEventTimeout
1425 ) {
1426 this.#engine.writeLog(
1427 InstanceEvent.WAIT_TIMED_OUT,
1428 cacheKey,
1429 waitForEventNameWithCounter,
1430 {
1431 name: timeoutError.name,
1432 message: timeoutError.message,
1433 },
1434 );
1435 await this.#state.storage.put(errorKey, timeoutError);
1436 throw timeoutError;
1437 }
1438
1439 const timeoutPromise = async (timeoutToWait: number, addToPQ: boolean) => {
1440 const priorityQueueHash = cacheKey;
1441 if (addToPQ) {
1442 // @ts-expect-error priorityQueue is initiated in init
1443 await this.#engine.priorityQueue.add({
1444 hash: priorityQueueHash,
1445 targetTimestamp: Date.now() + timeoutToWait,
1446 type: "timeout",
1447 });
1448 }
1449 await scheduler.wait(timeoutToWait);
1450 // if we reach here, means that we can try to delete the timeout from the PQ
1451 // because we managed to wait in the same lifetime
1452
1453 // @ts-expect-error priorityQueue is initiated in init
1454 this.#engine.priorityQueue.remove({
1455 hash: priorityQueueHash,
1456 type: "timeout",
1457 });
1458 // NOTE(lduarte): marking errors as user error allows the observability layer to avoid leaking
1459 // user errors to sentry while making everything more observable. `isUserError` is not serialized
1460 // into userland code due to how workerd serialzises errors over RPC - we also set it as undefined
1461 // in the obs layer in case changes to workerd happen
1462 const error = timeoutError;
1463 error.isUserError = true;
1464 throw error;
1465 };
1466
1467 const eventPromise = new Promise<Event>((resolve) => {
1468 // TODO: This might need to be the name, not the event type
1469
1470 const eventTypeQueue = this.#engine.eventMap.get(options.type);
1471 if (eventTypeQueue) {
1472 const event = eventTypeQueue.shift();
1473 if (event) {
1474 this.#engine.eventMap.set(options.type, eventTypeQueue);
1475 return resolve(event);
1476 }
1477 }
1478 const callbacks = this.#engine.waiters.get(options.type) ?? [];
1479 callbacks.push([cacheKey, resolve]);
1480
1481 this.#engine.waiters.set(options.type, callbacks);
1482 });
1483
1484 // Race event and timeout against the pause signal.
1485 const pauseSignal = this.#engine.pauseController.signal;
1486 const raceResult = await raceAgainstAbort(
1487 Promise.race([
1488 eventPromise,
1489 timeoutEntryPQ !== undefined
1490 ? timeoutPromise(timeoutEntryPQ.targetTimestamp - Date.now(), false)
1491 : timeoutPromise(ms(options.timeout), true),
1492 ]),
1493 pauseSignal,
1494 ).catch(async (error) => {
1495 const callbacks = this.#engine.waiters.get(options.type);
1496 if (callbacks) {
1497 const idx = callbacks.findIndex(([key]) => key === cacheKey);
1498 if (idx !== -1) {
1499 callbacks.splice(idx, 1);
1500 }
1501 }
1502
1503 this.#engine.writeLog(
1504 InstanceEvent.WAIT_TIMED_OUT,
1505 cacheKey,
1506 waitForEventNameWithCounter,
1507 error,
1508 );
1509 await this.#state.storage.put(errorKey, error);
1510 throw error;
1511 });
1512
1513 // Pause signal won the race — throw to stop the workflow
1514 if (raceResult.aborted) {
1515 throw new Error(ABORT_REASONS.USER_PAUSE);
1516 }
1517 const event = raceResult.value;
1518
1519 this.#engine.writeLog(
1520 InstanceEvent.WAIT_COMPLETE,
1521 cacheKey,
1522 waitForEventNameWithCounter,
1523 event,
1524 );
1525 await this.#state.storage.put(waitForEventKey, event);
1526
1527 return event as WorkflowStepEvent<T>;
1528 }
1529}