1import * as AWS from "@/AWS";
2import { Subnet } from "@/AWS/EC2";
3import { Cluster } from "@/AWS/ECS/Cluster.ts";
4import * as Core from "@/Test/Core";
5import * as Test from "@/Test/Alchemy";
6import { describe, expect } from "alchemy-test";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as Schedule from "effect/Schedule";
10import * as HttpClient from "effect/http/HttpClient";
11import * as HttpClientRequest from "effect/http/HttpClientRequest";
12import OneShotTask from "./fixtures/oneshot-task.ts";
13import { getDefaultVpc } from "../DefaultVpc.ts";
14import EcsBindingsTestFunctionLive, {
15 EcsBindingsTestFunction,
16} from "./handler.ts";
17
18const testOptions = { providers: AWS.providers() };
19const { test, beforeAll, afterAll } = Test.make(testOptions);
20const sharedStack = Core.scratchStack(testOptions, "EcsBindings");
21
22// Infra the fixture Lambda references via `Resource.ref(...)` (see
23// handler.ts): a public subnet in the default VPC, a scratch cluster, and the
24// one-shot busybox task definition. Yielded by BOTH deploy phases so the
25// second (Lambda) deploy doesn't delete them.
26const infra = Effect.gen(function* () {
27 // ECS bindings are not EC2 VPC coverage. Reuse the standing default VPC so
28 // this fixture does not compete for the account's small VPC quota during a
29 // high-concurrency AWS sweep. The stack still owns its subnet, preserving
30 // Resource.ref coverage and deterministic cleanup.
31 const vpc = yield* getDefaultVpc;
32 const subnet = yield* Subnet("EcsBindingsSubnet", {
33 vpcId: vpc.vpcId,
34 cidrBlock: vpc.subnetCidrBlock(240),
35 mapPublicIpOnLaunch: true,
36 });
37 const cluster = yield* Cluster("EcsBindingsCluster", {
38 clusterName: "alchemy-test-ecs-bindings",
39 });
40 const task = yield* OneShotTask;
41 return { cluster, task, subnet };
42});
43
44// Deploy is heavy: default-VPC subnet + cluster + the one-shot task definition
45// + Lambda. Keep readiness polling bounded after the deploy completes.
46const readinessPolicy = Schedule.max([
47 Schedule.fixed("2 seconds"),
48 Schedule.recurs(10),
49]);
50
51let baseUrl: string;
52
53class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
54 readonly status: number;
55 readonly body: string;
56}> {}
57
58// Retry transient fixture 5xx (cold re-init under parallel load); a genuine
59// 4xx/assertion failure is surfaced immediately.
60const send = (request: HttpClientRequest.HttpClientRequest) =>
61 HttpClient.execute(request).pipe(
62 Effect.flatMap((response) =>
63 response.status >= 500
64 ? response.text.pipe(
65 Effect.flatMap((body) =>
66 Effect.fail(
67 new TransientUpstream({ status: response.status, body }),
68 ),
69 ),
70 )
71 : Effect.succeed(response),
72 ),
73 Effect.retry({
74 while: (e) => e._tag === "TransientUpstream",
75 schedule: Schedule.max([
76 Schedule.exponential("500 millis"),
77 Schedule.recurs(6),
78 ]),
79 }),
80 );
81
82interface RunResponse {
83 taskArn?: string;
84 lastStatus?: string;
85 failures?: { reason?: string; detail?: string }[];
86}
87
88interface DescribedTask {
89 taskArn?: string;
90 lastStatus?: string;
91 stoppedReason?: string;
92 startedBy?: string;
93 containers?: { name?: string; exitCode?: number; lastStatus?: string }[];
94}
95
96const runTask = (body: { command?: string[]; startedBy?: string }) =>
97 send(
98 HttpClientRequest.bodyJsonUnsafe(
99 HttpClientRequest.post(`${baseUrl}/run`),
100 body,
101 ),
102 ).pipe(
103 Effect.flatMap((r) => r.json),
104 Effect.map((json) => json as RunResponse),
105 );
106
107const describeTask = (taskArn: string) =>
108 send(
109 HttpClientRequest.get(
110 `${baseUrl}/describe?task=${encodeURIComponent(taskArn)}`,
111 ),
112 ).pipe(
113 Effect.flatMap((r) => r.json),
114 Effect.map(
115 (json) =>
116 (json as { tasks?: DescribedTask[] }).tasks?.[0] as
117 | DescribedTask
118 | undefined,
119 ),
120 );
121
122// Poll the fixture's /describe route until the task reaches STOPPED.
123// Bounded: 16 × 5s = 80s of polling. A one-shot busybox task typically stops
124// within ~40s, but ECS can remain DEPROVISIONING for around a minute when the
125// account is under a high-concurrency test sweep.
126const describeUntilStopped = (taskArn: string) =>
127 describeTask(taskArn).pipe(
128 Effect.repeat({
129 schedule: Schedule.spaced("5 seconds"),
130 until: (t) => t?.lastStatus === "STOPPED",
131 times: 16,
132 }),
133 );
134
135describe(
136 "ECS Bindings",
137 {
138 tags: [
139 "provider:aws",
140 "provider:aws:ec2",
141 "provider:aws:ecs",
142 "provider:aws:lambda",
143 "live",
144 ],
145 },
146 () => {
147 beforeAll(
148 Effect.gen(function* () {
149 // scratchStack state is in-memory, so this destroy is a tag-based
150 // cloud sweep — it is what reclaims resources any previous
151 // (possibly interrupted) run left behind. Do not remove it.
152 yield* Effect.logInfo("ECS bindings setup: destroying previous stack");
153 yield* sharedStack.destroy();
154
155 // Phase 1: deploy the infra alone so its state exists before the
156 // Lambda's `Resource.ref(...)`s resolve (refs read stack state, not
157 // the in-flight plan — see handler.ts).
158 yield* Effect.logInfo("ECS bindings setup: deploying infra (phase 1)");
159 yield* sharedStack.deploy(infra);
160
161 // Phase 2: same infra (idempotent re-reconcile) + the fixture Lambda.
162 yield* Effect.logInfo(
163 "ECS bindings setup: deploying fixture (phase 2)",
164 );
165 const { functionUrl } = yield* sharedStack.deploy(
166 Effect.gen(function* () {
167 yield* infra;
168 return yield* EcsBindingsTestFunction;
169 }).pipe(Effect.provide(EcsBindingsTestFunctionLive)),
170 );
171
172 expect(functionUrl).toBeTruthy();
173 baseUrl = functionUrl!.replace(/\/+$/, "");
174 const readinessUrl = `${baseUrl}/list`;
175
176 yield* Effect.logInfo(
177 `ECS bindings setup: probing readiness at ${readinessUrl}`,
178 );
179 yield* HttpClient.get(readinessUrl).pipe(
180 Effect.flatMap((response) =>
181 response.status === 200
182 ? Effect.succeed(response)
183 : Effect.fail(
184 new Error(`Function not ready: ${response.status}`),
185 ),
186 ),
187 Effect.tapError((error) =>
188 Effect.logWarning(
189 `ECS bindings setup: fixture not ready yet (${String(error)})`,
190 ),
191 ),
192 Effect.retry({ schedule: readinessPolicy }),
193 );
194 yield* Effect.logInfo("ECS bindings setup: fixture ready");
195 }),
196 // High-concurrency full-suite sweeps starve the container build/push +
197 // phase-2 Lambda bundle. After a nuke the ECR repo is gone, so every
198 // round is a cold build — 210s tripped at c96 (convergence round 6).
199 { timeout: 330_000 },
200 );
201
202 // NO_DESTROY=1 keeps the deployment around between runs while iterating —
203 // without it an interrupted run tears everything down and the next run
204 // pays the full cold build again.
205 afterAll.skipIf(!!process.env.NO_DESTROY)(sharedStack.destroy(), {
206 timeout: 180_000,
207 });
208
209 describe("RunTask", () => {
210 test.provider(
211 "runs a one-shot Fargate task",
212 (_stack) =>
213 Effect.gen(function* () {
214 const response = yield* runTask({ startedBy: "alchemy-run-test" });
215 expect(response.failures ?? []).toEqual([]);
216 expect(response.taskArn).toMatch(/:task\//);
217 // RunTask returns the task in its initial state.
218 expect(response.lastStatus).toBeTruthy();
219 }),
220 { timeout: 120_000 },
221 );
222 });
223
224 describe("DescribeTasks", () => {
225 test.provider(
226 "observes a one-shot task run to completion with exit code 0",
227 (_stack) =>
228 Effect.gen(function* () {
229 const run = yield* runTask({ startedBy: "alchemy-describe-test" });
230 expect(run.taskArn).toBeTruthy();
231
232 const stopped = yield* describeUntilStopped(run.taskArn!);
233 expect(stopped?.lastStatus).toBe("STOPPED");
234 expect(stopped?.startedBy).toBe("alchemy-describe-test");
235 expect(stopped?.containers?.[0]?.exitCode).toBe(0);
236 }),
237 { timeout: 240_000 },
238 );
239
240 test.provider(
241 "reports MISSING for an unknown task id",
242 (_stack) =>
243 Effect.gen(function* () {
244 const response = yield* send(
245 HttpClientRequest.get(
246 `${baseUrl}/describe?task=00000000000000000000000000000000`,
247 ),
248 ).pipe(Effect.flatMap((r) => r.json));
249 const failures = (response as { failures?: { reason?: string }[] })
250 .failures;
251 expect(failures?.length).toBe(1);
252 expect(failures?.[0]?.reason).toBe("MISSING");
253 }),
254 { timeout: 60_000 },
255 );
256 });
257
258 describe("ListTasks", () => {
259 test.provider(
260 "lists tasks in the bound cluster by startedBy",
261 (_stack) =>
262 Effect.gen(function* () {
263 const run = yield* runTask({ startedBy: "alchemy-list-test" });
264 expect(run.taskArn).toBeTruthy();
265
266 // A one-shot echo task can stop before we ever observe it RUNNING,
267 // so poll the STOPPED listing until it appears (stopped tasks stay
268 // listed for ~an hour).
269 const taskArns = yield* send(
270 HttpClientRequest.get(
271 `${baseUrl}/list?status=STOPPED&startedBy=alchemy-list-test`,
272 ),
273 ).pipe(
274 Effect.flatMap((r) => r.json),
275 Effect.map(
276 (json) => (json as { taskArns: string[] }).taskArns ?? [],
277 ),
278 Effect.repeat({
279 schedule: Schedule.spaced("5 seconds"),
280 until: (arns) => arns.includes(run.taskArn!),
281 times: 10,
282 }),
283 );
284 expect(taskArns).toContain(run.taskArn!);
285 }),
286 { timeout: 240_000 },
287 );
288 });
289
290 describe("ListServices", () => {
291 test.provider(
292 "lists services in the bound cluster (none deployed)",
293 (_stack) =>
294 Effect.gen(function* () {
295 const response = yield* send(
296 HttpClientRequest.get(`${baseUrl}/services`),
297 ).pipe(Effect.flatMap((r) => r.json));
298 expect((response as { serviceArns: string[] }).serviceArns).toEqual(
299 [],
300 );
301 }),
302 { timeout: 60_000 },
303 );
304 });
305
306 describe("DescribeServices", () => {
307 test.provider(
308 "reports MISSING for an unknown service",
309 (_stack) =>
310 Effect.gen(function* () {
311 const response = yield* send(
312 HttpClientRequest.get(
313 `${baseUrl}/describe-services?service=alchemy-no-such-service`,
314 ),
315 ).pipe(Effect.flatMap((r) => r.json));
316 const failures = (response as { failures?: { reason?: string }[] })
317 .failures;
318 expect(failures?.length).toBe(1);
319 expect(failures?.[0]?.reason).toBe("MISSING");
320 }),
321 { timeout: 60_000 },
322 );
323 });
324
325 describe("ListContainerInstances", () => {
326 test.provider(
327 "returns no container instances for a Fargate-only cluster",
328 (_stack) =>
329 Effect.gen(function* () {
330 const response = yield* send(
331 HttpClientRequest.get(`${baseUrl}/container-instances`),
332 ).pipe(Effect.flatMap((r) => r.json));
333 expect(
334 (response as { containerInstanceArns: string[] })
335 .containerInstanceArns,
336 ).toEqual([]);
337 }),
338 { timeout: 60_000 },
339 );
340 });
341
342 describe("TaskProtection", () => {
343 test.provider(
344 "get/update task protection round-trips for a standalone task",
345 (_stack) =>
346 Effect.gen(function* () {
347 // Task protection only applies to service-managed tasks; for a
348 // standalone RunTask task AWS reports TASK_NOT_VALID (as a failure
349 // entry or a typed InvalidParameterException). Either way the
350 // probe proves IAM, the cluster injection, and (for /protect) the
351 // Duration → expiresInMinutes mapping reached AWS intact.
352 const run = yield* runTask({
353 command: ["sh", "-c", "sleep 300"],
354 startedBy: "alchemy-protection-test",
355 });
356 expect(run.taskArn).toBeTruthy();
357
358 interface ProtectionResponse {
359 error?: string;
360 failures?: { reason?: string }[];
361 protectedTasks?: unknown[];
362 }
363 const isTaskNotValid = (r: ProtectionResponse): boolean =>
364 r.error === "InvalidParameterException" ||
365 r.failures?.[0]?.reason === "TASK_NOT_VALID";
366
367 const protect = (yield* send(
368 HttpClientRequest.bodyJsonUnsafe(
369 HttpClientRequest.post(`${baseUrl}/protect`),
370 { taskArn: run.taskArn },
371 ),
372 ).pipe(Effect.flatMap((r) => r.json))) as ProtectionResponse;
373 expect(isTaskNotValid(protect)).toBe(true);
374
375 const protection = (yield* send(
376 HttpClientRequest.get(
377 `${baseUrl}/protection?task=${encodeURIComponent(run.taskArn!)}`,
378 ),
379 ).pipe(Effect.flatMap((r) => r.json))) as ProtectionResponse;
380 expect(isTaskNotValid(protection)).toBe(true);
381
382 // Clean up the sleeping task.
383 yield* send(
384 HttpClientRequest.bodyJsonUnsafe(
385 HttpClientRequest.post(`${baseUrl}/stop`),
386 { taskArn: run.taskArn, reason: "protection test done" },
387 ),
388 );
389 }),
390 { timeout: 120_000 },
391 );
392 });
393
394 describe("StopTask", () => {
395 test.provider(
396 "stops a long-running task with a reason",
397 (_stack) =>
398 Effect.gen(function* () {
399 // Override the one-shot command so the task stays alive until
400 // stopped (containerOverrides replaces the image CMD).
401 const run = yield* runTask({
402 command: ["sh", "-c", "sleep 300"],
403 startedBy: "alchemy-stop-test",
404 });
405 expect(run.taskArn).toBeTruthy();
406
407 const stop = yield* send(
408 HttpClientRequest.bodyJsonUnsafe(
409 HttpClientRequest.post(`${baseUrl}/stop`),
410 { taskArn: run.taskArn, reason: "alchemy stop test" },
411 ),
412 ).pipe(Effect.flatMap((r) => r.json));
413 expect((stop as { taskArn?: string }).taskArn).toBe(run.taskArn);
414 expect((stop as { desiredStatus?: string }).desiredStatus).toBe(
415 "STOPPED",
416 );
417
418 const stopped = yield* describeUntilStopped(run.taskArn!);
419 expect(stopped?.lastStatus).toBe("STOPPED");
420 expect(stopped?.stoppedReason).toBe("alchemy stop test");
421 }),
422 { timeout: 240_000 },
423 );
424 });
425 },
426);