Open History to see the last commit
1import * as AWS from "@alchemy.run/aws";
2import { executionNameOf } from "@alchemy.run/aws/Lambda/WorkflowFunction";
3import { describe, expect } from "@alchemy.run/test";
4import * as Test from "@alchemy.run/test/Alchemy";
5import * as Core from "@alchemy.run/test/Core";
6import * as Lambda from "@distilled.cloud/aws/lambda";
7import type { Status } from "alchemy/Workflows";
8import * as Data from "effect/Data";
9import * as Effect from "effect/Effect";
10import * as HttpClient from "effect/http/HttpClient";
11import * as Layer from "effect/Layer";
12import * as Schedule from "effect/Schedule";
13import { conformance } from "../../Workflows/conformance/conformance.ts";
14import CallerLive, { WorkflowCaller } from "./fixtures/durable-workflows/caller.ts";
15import HostLive, { TestWorkflows } from "./fixtures/durable-workflows/host.ts";
16
17const testOptions = { providers: AWS.providers() };
18const { test, beforeAll, afterAll } = Test.make(testOptions);
19const sharedStack = Core.scratchStack(testOptions, "LambdaWorkflowFunction");
20
21class NotReady extends Data.TaggedError("NotReady")<{ readonly message: string }> {}
22
23type RunStatus = Status<unknown, unknown>;
24
25/** A caller route's body, retried until it answers 200. */
26const request = <T>(target: string, method: "GET" | "POST" = "GET") =>
27 Effect.gen(function* () {
28 const client = yield* HttpClient.HttpClient;
29 const response = yield* method === "POST" ? client.post(target) : client.get(target);
30 const body = yield* response.text;
31 if (response.status !== 200) {
32 return yield* new NotReady({ message: `${method} ${target}: ${response.status}: ${body}` });
33 }
34 return (body.startsWith("{") ? JSON.parse(body) : body) as T;
35 }).pipe(
36 Effect.retry({
37 while: (error) => error._tag === "NotReady",
38 schedule: Schedule.spaced("1 second"),
39 times: 5,
40 }),
41 );
42
43/** A fresh id per run, so a retried test never reads an earlier attempt's execution. */
44const freshId = (prefix: string) =>
45 Effect.sync(() => `${prefix}-${crypto.randomUUID().slice(0, 8)}`);
46
47describe(
48 "Lambda WorkflowFunction",
49 { tags: ["provider:aws", "provider:aws:lambda", "live"] },
50 () => {
51 const deployed = beforeAll(
52 Effect.gen(function* () {
53 yield* sharedStack.destroy();
54 const outputs = yield* sharedStack.deploy(
55 Effect.gen(function* () {
56 const host = yield* TestWorkflows;
57 const caller = yield* WorkflowCaller;
58 return {
59 functionName: host.functionName,
60 version: host.version,
61 url: caller.functionUrl,
62 };
63 }).pipe(Effect.provide(Layer.mergeAll(HostLive, CallerLive))),
64 );
65 const url = outputs.url!.replace(/\/$/, "");
66 // A fresh Function URL can take a while to start serving.
67 yield* request(`${url}/ready`).pipe(
68 Effect.retry({ schedule: Schedule.spaced("2 seconds"), times: 30 }),
69 );
70 return { functionName: outputs.functionName, aliasVersion: outputs.version!, url };
71 }),
72 { timeout: 600_000 },
73 );
74 afterAll(sharedStack.destroy(), { timeout: 120_000 });
75
76 conformance({
77 test,
78 url: Effect.map(deployed, ({ url }) => url),
79 // Each wake-up is a fresh invocation of the host.
80 napsWithinMillis: 40_000,
81 // Lambda retries a durable step with its own policy.
82 recoveryWithinMillis: undefined,
83 timeout: 180_000,
84 unsupported: { raceAll: "aren't supported on Lambda durable functions" },
85 });
86
87 /** Start a Greet run through the caller, and wait for its result. */
88 const greet = (id: string) =>
89 Effect.gen(function* () {
90 const { url } = yield* deployed;
91 const { executionId } = yield* request<{ executionId: string }>(
92 `${url}/start/Greet/${id}`,
93 "POST",
94 );
95 const status = yield* request<RunStatus>(
96 `${url}/status/Greet/${id}?executionId=${executionId}`,
97 ).pipe(
98 Effect.repeat({
99 schedule: Schedule.spaced("2 seconds"),
100 until: (current) => current._tag !== "Running" && current._tag !== "NotStarted",
101 times: 45,
102 }),
103 );
104 return { executionId, status };
105 });
106
107 /** The durable executions of a run, on the host. */
108 const executionsOf = (workflow: string, executionId: string) =>
109 Effect.gen(function* () {
110 const { functionName } = yield* deployed;
111 return yield* Lambda.listDurableExecutionsByFunction({
112 FunctionName: functionName,
113 DurableExecutionName: executionNameOf(workflow, executionId),
114 }).pipe(
115 Effect.map((page) => page.DurableExecutions ?? []),
116 Effect.repeat({
117 schedule: Schedule.spaced("1 second"),
118 until: (executions) => executions.length > 0,
119 times: 20,
120 }),
121 );
122 });
123
124 test.provider(
125 "starts runs on the host's alias",
126 (_stack) =>
127 Effect.gen(function* () {
128 const { aliasVersion } = yield* deployed;
129 const run = yield* greet(yield* freshId("alias"));
130 expect(run.status).toEqual({ _tag: "Completed", result: "hello world" });
131 const [execution] = yield* executionsOf("Greet", run.executionId);
132 const { Version } = yield* Lambda.getDurableExecution({
133 DurableExecutionArn: execution!.DurableExecutionArn,
134 });
135 // The version the alias serves, not `$LATEST`.
136 expect(Version).toBe(aliasVersion);
137 }),
138 { timeout: 120_000 },
139 );
140
141 test.provider(
142 "starting the same execution twice makes one durable execution",
143 (_stack) =>
144 Effect.gen(function* () {
145 const { url } = yield* deployed;
146 const id = yield* freshId("twice");
147 yield* request(`${url}/start/Greet/${id}`, "POST");
148 const run = yield* greet(id);
149 expect(run.status).toEqual({ _tag: "Completed", result: "hello world" });
150 expect((yield* executionsOf("Greet", run.executionId)).length).toBe(1);
151 }),
152 { timeout: 120_000 },
153 );
154
155 test.provider(
156 "an interrupted run finishes its durable execution by itself",
157 (_stack) =>
158 Effect.gen(function* () {
159 const { url } = yield* deployed;
160 const id = yield* freshId("interrupt");
161 const { executionId } = yield* request<{ executionId: string }>(
162 `${url}/start/Approve/${id}`,
163 "POST",
164 );
165 // Interrupt once it's waiting on its signal.
166 yield* request<string[]>(`${url}/journal/${id}`).pipe(
167 Effect.repeat({
168 schedule: Schedule.spaced("1 second"),
169 until: (entries) => entries.length > 0,
170 times: 30,
171 }),
172 );
173 yield* request(`${url}/interrupt/Approve/${id}?executionId=${executionId}`, "POST");
174 // Woken through its callback and finished by itself, not stopped. The
175 // result is recorded before the durable execution settles, so poll
176 // until it leaves RUNNING.
177 const [execution] = yield* executionsOf("Approve", executionId).pipe(
178 Effect.repeat({
179 schedule: Schedule.spaced("2 seconds"),
180 until: ([first]) => first?.Status !== "RUNNING",
181 times: 30,
182 }),
183 );
184 expect(execution!.Status).toBe("SUCCEEDED");
185 }),
186 { timeout: 180_000 },
187 );
188 },
189);