1import type { Credentials } from "@distilled.cloud/aws/Credentials";
2import * as Lambda from "@distilled.cloud/aws/lambda";
3import type { InputProps } from "alchemy/Input";
4import * as Output from "alchemy/Output";
5import {
6 CurrentActivity,
7 decodeExit,
8 decodeResult,
9 encodeExit,
10 encodeResult,
11 makeCurrentActivity,
12 payloadCodec,
13} from "alchemy/WorkflowEncoding";
14import { type Implemented, implemented } from "alchemy/Workflows";
15import type { ConfigError } from "effect/Config";
16import * as Context from "effect/Context";
17import * as Duration from "effect/Duration";
18import * as Effect from "effect/Effect";
19import * as Effectable from "effect/Effectable";
20import * as Exit from "effect/Exit";
21import type * as HttpClient from "effect/http/HttpClient";
22import * as Layer from "effect/Layer";
23import * as Option from "effect/Option";
24import * as Redacted from "effect/Redacted";
25import * as Schedule from "effect/Schedule";
26import * as Schema from "effect/Schema";
27import * as Scope from "effect/Scope";
28import * as Stream from "effect/Stream";
29import { Activity, Workflow, WorkflowEngine } from "effect/workflow";
30import * as AWSBindings from "../AWSBindings.ts";
31import { DurableStep } from "./Durable.ts";
32import {
33 DurableFunction,
34 type DurableFunctionImpl,
35 type DurableFunctionInitServices,
36 type DurableFunctionProps,
37} from "./DurableFunction.ts";
38import { Function } from "./Function.ts";
39import { qualifierOf } from "./QualifiedArn.ts";
40
41type WorkflowNames<W> = W extends Workflow.Workflow<infer Name, any, any, any> ? Name : never;
42
43type ImplementedEach<Name> = Name extends string ? Implemented<Name> : never;
44
45/** Workflows a host lists whose implementation is missing from the Layer it's given. */
46type Missing<ROut, Name> = Name extends string
47 ? Implemented<Name> extends ROut
48 ? never
49 : Name
50 : never;
51
52type ImplementsAll<ROut, Name> = [Missing<ROut, Name>] extends [never]
53 ? unknown
54 : { readonly "Missing workflow implementations": Missing<ROut, Name> };
55
56/**
57 * A Lambda durable function that runs workflows: the class a
58 * `WorkflowFunction` declares. `yield*` it for the Lambda function itself.
59 */
60export interface WorkflowFunctionClass<
61 Self,
62 W extends Workflow.Any = Workflow.Any,
63> extends Effect.Effect<Function, never, Function["Providers"]> {
64 readonly LogicalId: string;
65 /** The workflows this function runs. */
66 readonly workflows: ReadonlyArray<W>;
67 /**
68 * Deploy the function with the Layer that implements its workflows. The
69 * Layer must implement every workflow the function lists.
70 */
71 make<ROut, E, RIn, PropsReq = never>(
72 props:
73 | InputProps<DurableFunctionProps>
74 | Effect.Effect<InputProps<DurableFunctionProps>, ConfigError, PropsReq>,
75 workflows: Layer.Layer<ROut, E, RIn> & ImplementsAll<ROut, WorkflowNames<W>>,
76 ): Layer.Layer<
77 Self,
78 never,
79 | Function["Providers"]
80 | Exclude<PropsReq | RIn, DurableFunctionInitServices | WorkflowEngine.WorkflowEngine>
81 >;
82 new (_: never): {};
83}
84
85/**
86 * A Lambda durable function that runs Effect workflows. Each run is one
87 * durable execution of the function, named after the workflow and its
88 * execution id.
89 *
90 * Steps run as durable steps, so a finished step's result is replayed rather
91 * than run again. `DurableClock.sleep` is a durable wait: the invocation ends,
92 * and Lambda invokes the function again when it's due, with no compute billed
93 * in between. `DurableDeferred.await` waits for a callback, completed with a
94 * run's `signal`. A workflow's child workflows run in the same function.
95 *
96 * Other functions start and read its runs with
97 * {@link DurableFunctionEngine | AWS.Lambda.DurableFunctionEngine}.
98 *
99 * ### Declaring a workflow function
100 * **Example:** One function for the order workflows
101 * ```typescript
102 * // src/OrderWorkflows.ts
103 * import * as AWS from "@alchemy.run/aws";
104 * import * as Layer from "effect/Layer";
105 * import { PlaceOrder, PlaceOrderLive, RefundOrder, RefundOrderLive } from "./orders.ts";
106 *
107 * export class OrderWorkflows extends AWS.Lambda.WorkflowFunction<OrderWorkflows>()(
108 * "OrderWorkflows",
109 * { workflows: [PlaceOrder, RefundOrder] },
110 * ) {}
111 *
112 * export default OrderWorkflows.make(
113 * { main: import.meta.url, executionTimeout: "7 days" },
114 * Layer.mergeAll(PlaceOrderLive, RefundOrderLive),
115 * );
116 * ```
117 *
118 * The Layer must implement every workflow the function lists: leaving one out
119 * is a type error.
120 *
121 * @resource
122 */
123export const WorkflowFunction =
124 <Self>() =>
125 <const W extends ReadonlyArray<Workflow.Any>>(
126 id: string,
127 options: {
128 /** The workflows this function runs. */
129 readonly workflows: W;
130 },
131 ): WorkflowFunctionClass<Self, W[number]> => {
132 const durable = (DurableFunction as any)()(id);
133 const make = durable.make;
134 // The Lambda function, from anywhere: its own init, or another function's.
135 const self = (Function as any)()(id);
136 return Object.assign(
137 durable,
138 {
139 workflows: options.workflows,
140 make: (props: unknown, workflows: Layer.Layer<any, any, any>) =>
141 make(props, makeWorkflowRunner(workflows)),
142 },
143 Effectable.Prototype({
144 label: `AWS.Lambda.WorkflowFunction<${id}>`,
145 evaluate: () => self,
146 }),
147 );
148 };
149
150// ---------------------------------------------------------------- callers
151
152type HostedBy<Host> = Host extends WorkflowFunctionClass<any, infer W> ? WorkflowNames<W> : never;
153
154/**
155 * Run workflows hosted by {@link WorkflowFunction | AWS.Lambda.WorkflowFunction}s
156 * from another Lambda function: a `WorkflowEngine` that starts, reads,
157 * signals and interrupts each workflow's runs on the function that hosts it.
158 *
159 * Provide it to the calling function, and resolve each workflow's client with
160 * `Workflows.RunWorkflow` during init. It grants the caller the IAM it needs
161 * on each host. Running a workflow none of the hosts list is a type error.
162 *
163 * ### Running workflows from another function
164 * **Example:** An API that starts orders
165 * ```typescript
166 * import * as AWS from "@alchemy.run/aws";
167 * import * as Workflows from "alchemy/Workflows";
168 * import * as Effect from "effect/Effect";
169 * import * as HttpServerResponse from "effect/http/HttpServerResponse";
170 * import { OrderWorkflows } from "./OrderWorkflows.ts";
171 * import { PlaceOrder } from "./orders.ts";
172 *
173 * export class Api extends AWS.Lambda.Function<Api>()("Api") {}
174 *
175 * export default Api.make(
176 * { main: import.meta.url, functionUrl: true },
177 * Effect.gen(function* () {
178 * const placeOrder = yield* Workflows.RunWorkflow(PlaceOrder);
179 * return {
180 * fetch: Effect.gen(function* () {
181 * const run = yield* placeOrder.start({ orderId: "o-1" });
182 * return HttpServerResponse.text(run.executionId);
183 * }),
184 * };
185 * }).pipe(Effect.provide(AWS.Lambda.DurableFunctionEngine(OrderWorkflows))),
186 * );
187 * ```
188 *
189 * **Example:** Workflows across several functions
190 * ```typescript
191 * Effect.provide(AWS.Lambda.DurableFunctionEngine(OrderWorkflows, BillingWorkflows))
192 * ```
193 *
194 * @layer
195 * @provides WorkflowEngine
196 * @product Lambda
197 */
198export const DurableFunctionEngine = <
199 const Hosts extends ReadonlyArray<WorkflowFunctionClass<any, any>>,
200>(
201 ...hosts: Hosts
202): Layer.Layer<
203 WorkflowEngine.WorkflowEngine | ImplementedEach<HostedBy<Hosts[number]>>,
204 never,
205 LambdaServices | Function["Providers"]
206> =>
207 Layer.merge(
208 Layer.effect(
209 WorkflowEngine.WorkflowEngine,
210 Effect.gen(function* () {
211 const bind = yield* AWSBindings.Bind;
212 const context = yield* Effect.context<LambdaServices>();
213 const routes = new Map<string, Calls>();
214 for (const host of hosts) {
215 const fn = yield* host;
216 if (!globalThis.__ALCHEMY_RUNTIME__) {
217 yield* bind`AWS.Lambda.DurableFunctionEngine(${fn})`({
218 policyStatements: [
219 {
220 Effect: "Allow",
221 Action: ["lambda:InvokeFunction"],
222 Resource: [fn.functionArn, Output.interpolate`${fn.functionArn}:*`],
223 },
224 // Durable execution ARNs aren't function ARNs, so reading and
225 // signalling runs is granted account-wide, as the host's own is.
226 {
227 Effect: "Allow",
228 Action: [
229 "lambda:ListDurableExecutionsByFunction",
230 "lambda:GetDurableExecution",
231 "lambda:GetDurableExecutionHistory",
232 "lambda:StopDurableExecution",
233 "lambda:SendDurableExecutionCallbackSuccess",
234 "lambda:SendDurableExecutionCallbackFailure",
235 ],
236 Resource: ["*"],
237 },
238 ],
239 });
240 }
241 const functionName = yield* fn.functionName;
242 // Runs start on the host's alias when it sets one.
243 // oxlint-disable-next-line effecttsgo/unsafe-effect-type-assertion -- the host provides RuntimeContext at runtime
244 const qualifier = yield* qualifierOf(fn).bind(
245 `${fn.LogicalId}.qualifier`,
246 ) as Effect.Effect<Effect.Effect<string>>;
247 const calls = makeCalls(
248 (effect) => effect.pipe(Effect.provide(context)),
249 functionName,
250 qualifier,
251 );
252 for (const workflow of host.workflows) routes.set(workflow._tag, calls);
253 }
254 return makeEngine({
255 route: (workflowName) => {
256 const calls = routes.get(workflowName);
257 return calls
258 ? Effect.succeed(calls)
259 : Effect.die(new Error(`No WorkflowFunction hosts workflow ${workflowName}`));
260 },
261 // Workflows run on their hosts; nothing registers here.
262 register: () => undefined,
263 });
264 }),
265 ),
266 implemented(
267 hosts.flatMap((host) => host.workflows.map((workflow) => workflow._tag)),
268 ) as Layer.Layer<ImplementedEach<HostedBy<Hosts[number]>>>,
269 ) as Layer.Layer<
270 WorkflowEngine.WorkflowEngine | ImplementedEach<HostedBy<Hosts[number]>>,
271 never,
272 LambdaServices | Function["Providers"]
273 >;
274
275// ---------------------------------------------------------------- hosting
276
277/** What a durable function running workflows is started with. */
278interface WorkflowInvocation {
279 /** The workflow's tag. */
280 readonly workflow: string;
281 readonly executionId: string;
282 /** The payload, encoded. */
283 readonly payload: unknown;
284}
285
286/** The services the engines call Lambda with. */
287type LambdaServices = Credentials | HttpClient.HttpClient;
288
289/**
290 * The durable function body that runs the given workflows: each invocation is
291 * one run of one workflow. Child workflows start in the same function.
292 */
293const makeWorkflowRunner = <ROut, E, RIn>(
294 workflows: Layer.Layer<ROut, E, RIn>,
295): Effect.Effect<
296 DurableFunctionImpl<WorkflowInvocation, string>,
297 E,
298 Exclude<RIn, WorkflowEngine.WorkflowEngine> | LambdaServices
299> =>
300 Effect.gen(function* () {
301 const context = yield* Effect.context<LambdaServices>();
302 const functionName = yield* Effect.sync(() => process.env.AWS_LAMBDA_FUNCTION_NAME);
303 // Child workflows start on the version this function is running.
304 const version = yield* Effect.sync(() => process.env.AWS_LAMBDA_FUNCTION_VERSION ?? "$LATEST");
305 const handlers = new Map<string, Registration>();
306 const calls = makeCalls(
307 (effect) => effect.pipe(Effect.provide(context)),
308 functionName
309 ? Effect.succeed(functionName)
310 : Effect.die(new Error("AWS_LAMBDA_FUNCTION_NAME is not set")),
311 Effect.succeed(version),
312 );
313 const engine = makeEngine({
314 route: (workflowName) =>
315 handlers.has(workflowName)
316 ? Effect.succeed(calls)
317 : Effect.die(new Error(`Workflow ${workflowName} is not hosted by this function`)),
318 register: (workflow, execute) => handlers.set(workflow._tag, { workflow, execute }),
319 });
320 // Workflows register for the life of the function instance.
321 yield* Layer.buildWithScope(
322 workflows.pipe(Layer.provide(Layer.succeed(WorkflowEngine.WorkflowEngine, engine))),
323 yield* Scope.make(),
324 );
325
326 return (invocation: WorkflowInvocation) =>
327 Effect.gen(function* () {
328 const registration = handlers.get(invocation.workflow);
329 if (!registration) {
330 return yield* Effect.die(new Error(`Workflow ${invocation.workflow} is not hosted here`));
331 }
332 const durable = yield* DurableStep;
333 const runContext = yield* Effect.context<never>();
334 const { workflow, execute } = registration;
335 const payload = yield* Schema.decodeEffect(payloadCodec(workflow))(invocation.payload).pipe(
336 Effect.orDie,
337 );
338 const instance = WorkflowEngine.WorkflowInstance.initial(workflow, invocation.executionId);
339 const result = yield* execute(payload as object, invocation.executionId).pipe(
340 Workflow.intoResult,
341 Effect.provide(
342 Layer.mergeAll(
343 Layer.succeed(WorkflowEngine.WorkflowInstance, instance),
344 Layer.succeed(WorkflowEngine.WorkflowEngine, engine),
345 Layer.succeed(CurrentRun, {
346 durable,
347 context: runContext,
348 clocks: new Map(),
349 instance,
350 waited: new Map(),
351 }),
352 ),
353 ),
354 );
355 // Every wait blocks inside the durable execution, so a run always finishes.
356 if (result._tag === "Suspended") {
357 return yield* Effect.die(new Error(`Workflow ${workflow._tag} suspended unexpectedly`));
358 }
359 return yield* encodeResult(workflow, result.exit);
360 });
361 }) as Effect.Effect<
362 DurableFunctionImpl<WorkflowInvocation, string>,
363 E,
364 Exclude<RIn, WorkflowEngine.WorkflowEngine> | LambdaServices
365 >;
366
367// ---------------------------------------------------------------- the engine
368
369type Execute = (
370 payload: object,
371 executionId: string,
372) => Effect.Effect<
373 unknown,
374 unknown,
375 WorkflowEngine.WorkflowInstance | WorkflowEngine.WorkflowEngine
376>;
377
378interface Registration {
379 readonly workflow: Workflow.AnyWithProps;
380 readonly execute: Execute;
381}
382
383/** The run in progress: its durable context, and the clocks it has scheduled. */
384class CurrentRun extends Context.Service<
385 CurrentRun,
386 {
387 readonly durable: DurableStep["Service"];
388 readonly context: Context.Context<never>;
389 readonly clocks: Map<string, Duration.Duration>;
390 readonly instance: WorkflowEngine.WorkflowInstance["Service"];
391 /** Waits run outside the activity that reached them, by deferred name. */
392 readonly waited: Map<string, string>;
393 /** Inside an activity's step, which can't wait: the wait it reached. */
394 readonly step?: { pending?: Pending };
395 }
396>()("alchemy/AWS/Lambda/DurableWorkflows/CurrentRun") {}
397
398/** A wait an activity reached inside its step, recorded so a replay repeats it. */
399interface Pending {
400 readonly name: string;
401 readonly clockMillis?: number;
402}
403
404/** An interrupt fails the callback a run waits on: the run is interrupted there. */
405const interruptible = <A>(run: CurrentRun["Service"], wait: Effect.Effect<A>) =>
406 wait.pipe(
407 Effect.catchDefect((defect) =>
408 isInterrupt(defect)
409 ? Effect.sync(() => {
410 run.instance.interrupted = true;
411 run.instance.suspended = false;
412 }).pipe(Effect.andThen(Effect.interrupt))
413 : Effect.die(defect),
414 ),
415 );
416
417/** Wait for a deferred: a clock's sleep is a durable wait, a signal a callback. Its exit, encoded. */
418const waitFor = (run: CurrentRun["Service"], pending: Pending) =>
419 Effect.gen(function* () {
420 if (pending.clockMillis !== undefined) {
421 yield* run.durable.wait(
422 pending.name,
423 Duration.max(Duration.millis(pending.clockMillis), Duration.seconds(1)),
424 );
425 return yield* encodeExit(Exit.void);
426 }
427 const text = yield* interruptible(
428 run,
429 run.durable.waitForCallback<string>({ name: pending.name, submitter: () => Effect.void }),
430 );
431 return typeof text === "string" ? text : JSON.stringify(text);
432 });
433
434const currentRun = (operation: string) =>
435 Effect.flatMap(Effect.serviceOption(CurrentRun), (run) =>
436 Option.isSome(run)
437 ? Effect.succeed(run.value)
438 : Effect.die(new Error(`WorkflowFunction: ${operation} only runs inside a workflow run`)),
439 );
440
441/** A callback failed with this type interrupts the run waiting on it. */
442const INTERRUPT = "alchemy/WorkflowInterrupted";
443
444/**
445 * What an activity's step records when the activity suspends: on a failure,
446 * or on a wait it reached inside its step, followed by that wait as JSON.
447 */
448const SUSPENDED = "alchemy-suspended:";
449
450/** The callback a run suspended on a failure waits on, which `resume` completes. */
451const isSuspension = (name: string) => name.includes("/suspended-");
452
453/** The SDK's callback error arrives wrapped, with the type in its cause chain. */
454const isInterrupt = (defect: unknown) => {
455 let node = defect;
456 for (let depth = 0; node !== null && typeof node === "object" && depth < 5; depth++) {
457 const error = node as {
458 readonly name?: unknown;
459 readonly message?: unknown;
460 readonly cause?: unknown;
461 };
462 if (
463 error.name === INTERRUPT ||
464 (typeof error.message === "string" && error.message.includes(INTERRUPT))
465 ) {
466 return true;
467 }
468 node = error.cause;
469 }
470 return false;
471};
472
473/** Lambda's calls for one host function. */
474interface Calls {
475 readonly start: (executionName: string, invocation: WorkflowInvocation) => Effect.Effect<void>;
476 readonly find: (executionName: string) => Effect.Effect<string | undefined>;
477 readonly get: (
478 executionArn: string,
479 ) => Effect.Effect<{ readonly status: string; readonly result: string | undefined }>;
480 readonly history: (executionArn: string) => Effect.Effect<ReadonlyArray<Lambda.Event>>;
481 readonly stop: (executionArn: string) => Effect.Effect<void>;
482 readonly succeed: (callbackId: string, result: string) => Effect.Effect<void>;
483 readonly fail: (callbackId: string, errorType: string) => Effect.Effect<void>;
484}
485
486const unwrap = (value: string | Redacted.Redacted<string> | undefined) =>
487 value === undefined ? undefined : Redacted.isRedacted(value) ? Redacted.value(value) : value;
488
489const makeCalls = (
490 provide: <A, E>(effect: Effect.Effect<A, E, LambdaServices>) => Effect.Effect<A, E>,
491 functionName: Effect.Effect<string>,
492 qualifier: Effect.Effect<string>,
493): Calls => {
494 const run = <A, E>(effect: Effect.Effect<A, E, LambdaServices>) =>
495 provide(effect).pipe(Effect.orDie);
496 return {
497 start: (executionName, invocation) =>
498 Effect.flatMap(Effect.all([functionName, qualifier]), ([FunctionName, Qualifier]) =>
499 run(
500 Lambda.invoke({
501 FunctionName,
502 Qualifier,
503 InvocationType: "Event",
504 DurableExecutionName: executionName,
505 Payload: JSON.stringify(invocation),
506 }).pipe(
507 // Already started: the same key always names the same run.
508 Effect.catchTag("DurableExecutionAlreadyStartedException", () => Effect.void),
509 Effect.asVoid,
510 ),
511 ),
512 ),
513 find: (executionName) =>
514 Effect.flatMap(functionName, (FunctionName) =>
515 run(
516 Lambda.listDurableExecutionsByFunction({
517 FunctionName,
518 DurableExecutionName: executionName,
519 }).pipe(Effect.map((page) => page.DurableExecutions?.[0]?.DurableExecutionArn)),
520 ),
521 ),
522 get: (executionArn) =>
523 run(
524 Lambda.getDurableExecution({ DurableExecutionArn: executionArn }).pipe(
525 Effect.map((execution) => ({
526 status: execution.Status,
527 result: unwrap(execution.Result),
528 })),
529 ),
530 ),
531 history: (executionArn) =>
532 run(
533 Lambda.getDurableExecutionHistory.items({ DurableExecutionArn: executionArn }).pipe(
534 Stream.runCollect,
535 Effect.map((events) => [...events]),
536 ),
537 ),
538 stop: (executionArn) =>
539 run(Lambda.stopDurableExecution({ DurableExecutionArn: executionArn }).pipe(Effect.asVoid)),
540 succeed: (callbackId, result) =>
541 run(
542 Lambda.sendDurableExecutionCallbackSuccess({ CallbackId: callbackId, Result: result }).pipe(
543 Effect.asVoid,
544 ),
545 ),
546 fail: (callbackId, errorType) =>
547 run(
548 Lambda.sendDurableExecutionCallbackFailure({
549 CallbackId: callbackId,
550 Error: { ErrorType: errorType, ErrorMessage: errorType },
551 }).pipe(Effect.asVoid),
552 ),
553 };
554};
555
556/**
557 * A durable execution's name: the workflow and execution id, within Lambda's
558 * limit of 64 letters, digits, `-` and `_`.
559 */
560/** @internal */
561export const executionNameOf = (workflowName: string, executionId: string) => {
562 const name = `${workflowName}-${executionId}`.replace(/[^A-Za-z0-9_-]/g, "_");
563 if (name.length <= 64) return name;
564 let hash = 2166136261;
565 for (const char of name) hash = Math.imul(hash ^ char.charCodeAt(0), 16777619) >>> 0;
566 return `${name.slice(0, 55)}-${hash.toString(16).padStart(8, "0")}`;
567};
568
569/**
570 * A run's callbacks: the ones it is waiting on, started and not yet completed,
571 * and the names of the ones already completed. A callback's name is on the
572 * `waitForCallback` context it was started in, its parent.
573 */
574const callbacksOf = (events: ReadonlyArray<Lambda.Event>) => {
575 const names = new Map<string, string>();
576 const started = new Map<string, { readonly name: string; readonly callbackId: string }>();
577 const completed = new Set<string>();
578 for (const event of events) {
579 if (event.SubType === "WaitForCallback" && event.Id && event.Name) {
580 names.set(event.Id, event.Name);
581 }
582 if (event.CallbackStartedDetails && event.Id) {
583 started.set(event.Id, {
584 name: (event.ParentId && names.get(event.ParentId)) ?? event.Name ?? "",
585 callbackId: event.CallbackStartedDetails.CallbackId,
586 });
587 }
588 if (
589 event.Id &&
590 (event.CallbackSucceededDetails ||
591 event.CallbackFailedDetails ||
592 event.CallbackTimedOutDetails)
593 ) {
594 const callback = started.get(event.Id);
595 if (callback) completed.add(callback.name);
596 started.delete(event.Id);
597 }
598 }
599 return { pending: [...started.values()], completed };
600};
601
602const pendingCallbacks = (events: ReadonlyArray<Lambda.Event>) => callbacksOf(events).pending;
603
604/** Whether a run is in a durable wait: a sleep, or a parent checking on its child. */
605const sleeping = (events: ReadonlyArray<Lambda.Event>) => {
606 const waits = new Set<string>();
607 for (const event of events) {
608 if (!event.Id) continue;
609 if (event.WaitStartedDetails) waits.add(event.Id);
610 if (event.WaitSucceededDetails || event.WaitCancelledDetails) waits.delete(event.Id);
611 }
612 return waits.size > 0;
613};
614
615/** Whether a run has finished, so there's nothing left to signal or interrupt. */
616const finished = (calls: Calls, executionArn: string) =>
617 Effect.map(calls.get(executionArn), ({ status }) => status !== "RUNNING");
618
619/**
620 * Find a waiting run's callback, retrying while the run is on its way to the
621 * wait. Undefined once the wait is over, or the run is: a signal sent again,
622 * or after the run finished, has nothing to complete.
623 */
624const findCallback = (calls: Calls, executionName: string, deferredName: string) =>
625 Effect.gen(function* () {
626 const arn = yield* calls.find(executionName);
627 if (!arn) return yield* Effect.fail("not started" as const);
628 const { pending, completed } = callbacksOf(yield* calls.history(arn));
629 const callback = pending.find((waiting) => waiting.name === deferredName);
630 if (callback) return callback.callbackId;
631 if (completed.has(deferredName) || (yield* finished(calls, arn))) return undefined;
632 return yield* Effect.fail("not waiting" as const);
633 }).pipe(
634 Effect.retry({
635 schedule: Schedule.min([Schedule.exponential("200 millis"), Schedule.spaced("5 seconds")]),
636 times: 40,
637 }),
638 Effect.catch((reason) =>
639 Effect.die(
640 new Error(
641 `Workflow execution ${executionName} is ${reason === "not started" ? "not started" : `not waiting on ${deferredName}`}`,
642 ),
643 ),
644 ),
645 );
646
647/** Read a run's result from Lambda: undefined while it runs. */
648const readResult = (calls: Calls, workflow: Workflow.AnyWithProps, executionArn: string) =>
649 Effect.gen(function* () {
650 const { status, result } = yield* calls.get(executionArn);
651 switch (status) {
652 case "RUNNING":
653 return undefined;
654 case "SUCCEEDED":
655 return yield* decodeResult(workflow, JSON.parse(result ?? "null") as string);
656 case "STOPPED":
657 return new Workflow.Complete({ exit: Exit.interrupt() });
658 default:
659 return new Workflow.Complete({
660 exit: Exit.die(new Error(`Workflow execution ${executionArn} ended ${status}`)),
661 });
662 }
663 });
664
665const makeEngine = (options: {
666 /** The calls for the function that hosts a workflow. */
667 readonly route: (workflowName: string) => Effect.Effect<Calls>;
668 readonly register: (workflow: Workflow.AnyWithProps, execute: Execute) => void;
669}): WorkflowEngine.WorkflowEngine["Service"] => {
670 const { route } = options;
671 const engine: WorkflowEngine.WorkflowEngine["Service"] = WorkflowEngine.makeUnsafe({
672 register: (workflow, execute) =>
673 Effect.sync(() => options.register(workflow as Workflow.AnyWithProps, execute)),
674
675 execute: (workflow, opts) =>
676 Effect.gen(function* () {
677 const calls = yield* route(workflow._tag);
678 const executionName = executionNameOf(workflow._tag, opts.executionId);
679 const payload = yield* Schema.encodeEffect(payloadCodec(workflow as Workflow.AnyWithProps))(
680 opts.payload,
681 ).pipe(Effect.orDie);
682 yield* calls.start(executionName, {
683 workflow: workflow._tag,
684 executionId: opts.executionId,
685 payload,
686 });
687 if (opts.discard) return undefined;
688 const run = yield* Effect.serviceOption(CurrentRun);
689 if (Option.isNone(run)) {
690 const arn = yield* calls.find(executionName);
691 const result = arn
692 ? yield* readResult(calls, workflow as Workflow.AnyWithProps, arn)
693 : undefined;
694 return result ?? Workflow.Suspended.make({});
695 }
696 // A child workflow: wait for it here, with durable waits between checks.
697 for (let check = 0; ; check++) {
698 const text = yield* run.value.durable.step({
699 name: `child/${executionName}/${check}`,
700 effect: Effect.gen(function* () {
701 const arn = yield* calls.find(executionName);
702 const result = arn
703 ? yield* readResult(calls, workflow as Workflow.AnyWithProps, arn)
704 : undefined;
705 return result
706 ? yield* encodeResult(workflow as Workflow.AnyWithProps, result.exit)
707 : "";
708 }),
709 });
710 if (text !== "") return yield* decodeResult(workflow as Workflow.AnyWithProps, text);
711 yield* run.value.durable.wait(`child-wait/${executionName}/${check}`, "5 seconds");
712 }
713 }) as Effect.Effect<any>,
714
715 poll: (workflow, executionId) =>
716 Effect.gen(function* () {
717 const calls = yield* route(workflow._tag);
718 const arn = yield* calls.find(executionNameOf(workflow._tag, executionId));
719 if (!arn) return Option.none();
720 const result = yield* readResult(calls, workflow as Workflow.AnyWithProps, arn);
721 return Option.some(result ?? Workflow.Suspended.make({}));
722 }),
723
724 // A run waiting on a signal is woken with an interrupt, so its
725 // compensations run. A sleeping run is stopped. A run still working,
726 // or not listed yet, gets a while to reach one or the other.
727 interrupt: (workflow, executionId) =>
728 Effect.gen(function* () {
729 const calls = yield* route(workflow._tag);
730 const executionName = executionNameOf(workflow._tag, executionId);
731 const target = yield* Effect.gen(function* () {
732 const arn = yield* calls.find(executionName);
733 if (!arn) return yield* Effect.fail({ arn } as const);
734 const events = yield* calls.history(arn);
735 const pending = pendingCallbacks(events);
736 if (pending.length > 0) return { arn, pending, stop: false };
737 if (sleeping(events)) return { arn, pending, stop: true };
738 if (yield* finished(calls, arn)) return { arn, pending, stop: false };
739 return yield* Effect.fail({ arn } as const);
740 }).pipe(
741 Effect.retry({
742 schedule: Schedule.min([
743 Schedule.exponential("200 millis"),
744 Schedule.spaced("2 seconds"),
745 ]),
746 times: 25,
747 }),
748 // Still working after all that: it can only be stopped.
749 Effect.catch(({ arn }) => Effect.succeed({ arn, pending: [], stop: true })),
750 );
751 if (!target.arn) return;
752 if (target.stop) return yield* calls.stop(target.arn);
753 yield* Effect.forEach(
754 target.pending,
755 (callback) => calls.fail(callback.callbackId, INTERRUPT),
756 {
757 discard: true,
758 },
759 );
760 }),
761
762 interruptUnsafe: (workflow, executionId) =>
763 Effect.gen(function* () {
764 const calls = yield* route(workflow._tag);
765 const arn = yield* calls.find(executionNameOf(workflow._tag, executionId));
766 if (arn) yield* calls.stop(arn);
767 }),
768
769 // Only a run suspended on a failure waits to be resumed: complete the
770 // callback it waits on.
771 resume: (workflow, executionId) =>
772 Effect.gen(function* () {
773 const calls = yield* route(workflow._tag);
774 const arn = yield* calls.find(executionNameOf(workflow._tag, executionId));
775 if (!arn) return;
776 const suspended = pendingCallbacks(yield* calls.history(arn)).filter((callback) =>
777 isSuspension(callback.name),
778 );
779 yield* Effect.forEach(suspended, (callback) => calls.succeed(callback.callbackId, "{}"), {
780 discard: true,
781 });
782 }),
783
784 activityExecute: (activity, attempt) =>
785 Effect.gen(function* () {
786 const run = yield* currentRun("activity execution");
787 const instance = yield* WorkflowEngine.WorkflowInstance;
788 const activityInstance = WorkflowEngine.WorkflowInstance.initial(
789 instance.workflow,
790 instance.executionId,
791 );
792 activityInstance.interrupted = instance.interrupted;
793 // A step can't hold a durable operation. An activity that reaches a
794 // wait inside its step suspends, and records the wait: it runs here,
795 // and the activity again under a new step name, which finds it done.
796 // An activity that suspends on a failure waits for a resume, or an
797 // interrupt, the same way.
798 for (let generation = 0; ; generation++) {
799 const step: { pending?: Pending } = {};
800 const text = yield* run.durable.step({
801 name:
802 generation === 0
803 ? `${activity.name}/${attempt}`
804 : `${activity.name}/${attempt}/resumed-${generation}`,
805 effect: activity.executeEncoded.pipe(
806 Workflow.intoResult,
807 Effect.flatMap((result) =>
808 result._tag === "Complete"
809 ? encodeExit(result.exit)
810 : Effect.succeed(`${SUSPENDED}${JSON.stringify(step.pending ?? null)}`),
811 ),
812 // The run's services, with the step's own instance.
813 Effect.provide(
814 Layer.succeedContext(
815 run.context.pipe(
816 Context.add(WorkflowEngine.WorkflowInstance, activityInstance),
817 // The step runs on its own fiber: carry the attempt `Activity.retry` set.
818 Context.add(Activity.CurrentAttempt, attempt),
819 // A durable clock or deferred inside the activity needs the engine.
820 Context.add(WorkflowEngine.WorkflowEngine, engine),
821 Context.add(CurrentRun, { ...run, step }),
822 Context.add(
823 CurrentActivity,
824 makeCurrentActivity(
825 instance.workflow._tag,
826 instance.executionId,
827 activity.name,
828 ),
829 ),
830 ),
831 ),
832 ),
833 ) as Effect.Effect<string>,
834 });
835 if (!text.startsWith(SUSPENDED)) {
836 return new Workflow.Complete({ exit: yield* decodeExit(text) });
837 }
838 const pending = JSON.parse(text.slice(SUSPENDED.length)) as Pending | null;
839 if (pending) {
840 run.waited.set(pending.name, yield* waitFor(run, pending));
841 continue;
842 }
843 yield* interruptible(
844 run,
845 run.durable.waitForCallback<string>({
846 name: `${activity.name}/${attempt}/suspended-${generation}`,
847 submitter: () => Effect.void,
848 }),
849 );
850 }
851 }),
852
853 // A clock's sleep is a durable wait; a signal is a callback to wait for.
854 deferredResult: (deferred) =>
855 Effect.gen(function* () {
856 const run = yield* currentRun("DurableDeferred reads");
857 // A race asks for its recorded winner before it runs. Lambda replays
858 // operations in order, so a winner can't be looked up first.
859 if (deferred.name.startsWith("raceAll/")) {
860 return yield* Effect.die(
861 new Error(
862 "Activity.raceAll and DurableDeferred.raceAll aren't supported on Lambda durable functions: a step's result can't be looked up before the step runs.",
863 ),
864 );
865 }
866 const waited = run.waited.get(deferred.name);
867 if (waited !== undefined) return Option.some(yield* decodeExit(waited));
868 const duration = run.clocks.get(deferred.name);
869 const pending: Pending = {
870 name: deferred.name,
871 ...(duration === undefined ? {} : { clockMillis: Duration.toMillis(duration) }),
872 };
873 // Inside a step: the activity suspends, and the wait runs after it.
874 if (run.step) {
875 run.step.pending = pending;
876 return Option.none();
877 }
878 return Option.some(yield* decodeExit(yield* waitFor(run, pending)));
879 }),
880
881 deferredDone: (opts) =>
882 Effect.gen(function* () {
883 const calls = yield* route(opts.workflowName);
884 const executionName = executionNameOf(opts.workflowName, opts.executionId);
885 const callbackId = yield* findCallback(calls, executionName, opts.deferredName);
886 if (callbackId === undefined) return;
887 yield* calls.succeed(callbackId, yield* encodeExit(opts.exit));
888 }),
889
890 scheduleClock: (_workflow, opts) =>
891 Effect.flatMap(currentRun("DurableClock scheduling"), (run) =>
892 Effect.sync(() => void run.clocks.set(opts.clock.deferred.name, opts.clock.duration)),
893 ),
894 });
895 return engine;
896};