1// Alchemy modifications are licensed under Apache-2.0.
2// This file includes third-party code; see /THIRD_PARTY_LICENSES.md.
3import { ms } from "itty-time";
4import { DelayFunctionError } from "./retries.ts";
5import type {
6 WorkflowDelayDuration,
7 WorkflowDelayFunction,
8 WorkflowDynamicDelayContext,
9} from "cloudflare:workers";
10
11export const DEFAULT_RETRY_DELAY_MS = 1000;
12
13export const DELAY_FUNCTION_TIMEOUT_MS = ms("5 seconds");
14
15type DelayLogger = {
16 warn(...msgs: unknown[]): void;
17 debug(...msgs: unknown[]): void;
18};
19
20type Waiter = (ms: number, opts?: { signal?: AbortSignal }) => Promise<void>;
21
22type AbortRaceResult<T> = { aborted: true } | { aborted: false; value: T };
23
24export async function raceAgainstAbort<T>(
25 promise: Promise<T>,
26 signal: AbortSignal
27): Promise<AbortRaceResult<T>> {
28 const resultPromise = promise.then(
29 (value): AbortRaceResult<T> => ({
30 aborted: false,
31 value,
32 })
33 );
34 if (signal.aborted) {
35 // Observe a later rejection from the losing promise so it doesn't become
36 // an unhandled rejection after returning the already-aborted result.
37 void resultPromise.catch(() => undefined);
38 return { aborted: true };
39 }
40
41 let resolveAbort: ((result: AbortRaceResult<T>) => void) | undefined;
42 const abortPromise = new Promise<AbortRaceResult<T>>((resolve) => {
43 resolveAbort = resolve;
44 });
45 const onAbort = (): void => resolveAbort?.({ aborted: true });
46 signal.addEventListener("abort", onAbort, { once: true });
47
48 try {
49 return await Promise.race([resultPromise, abortPromise]);
50 } finally {
51 signal.removeEventListener("abort", onAbort);
52 }
53}
54
55export async function invokeDelayFunction(
56 delay: WorkflowDelayFunction,
57 input: WorkflowDynamicDelayContext,
58 options: {
59 timeoutMs: number;
60 wait: Waiter;
61 signal?: AbortSignal;
62 logger?: DelayLogger;
63 }
64): Promise<WorkflowDelayDuration> {
65 const { wait, signal, logger } = options;
66 const logFields = { step: input.ctx.step.name, attempt: input.ctx.attempt };
67
68 const settled = new AbortController();
69 const timeoutSignal = signal
70 ? AbortSignal.any([signal, settled.signal])
71 : settled.signal;
72
73 type Raced =
74 | { timedOut: true }
75 | { timedOut: false; value: WorkflowDelayDuration };
76
77 try {
78 const result = await Promise.race<Raced>([
79 Promise.resolve(delay(input)).then((value) => ({
80 timedOut: false,
81 value,
82 })),
83 wait(options.timeoutMs, { signal: timeoutSignal })
84 .then((): Raced => ({ timedOut: true }))
85 .catch((): Raced => ({ timedOut: true })),
86 ]);
87 if (result.timedOut) {
88 if (signal?.aborted) {
89 logger?.debug(
90 "delay function aborted by engine shutdown; using default delay",
91 logFields
92 );
93 return DEFAULT_RETRY_DELAY_MS;
94 }
95 logger?.warn(
96 `delay function did not resolve within ${options.timeoutMs}ms`,
97 logFields
98 );
99 throw new DelayFunctionError(
100 `did not return within ${options.timeoutMs / 1000} seconds`
101 );
102 }
103 return result.value;
104 } catch (e) {
105 if (e instanceof DelayFunctionError) {
106 throw e;
107 }
108 const message = e instanceof Error ? e.message : String(e);
109 logger?.warn("delay function threw", { ...logFields, error: message });
110 throw new DelayFunctionError(`threw an error: ${message}`);
111 } finally {
112 settled.abort();
113 }
114}