1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as batch from "@distilled.cloud/aws/batch";
5import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
6import * as eventbridge from "@distilled.cloud/aws/eventbridge";
7import { describe, expect } from "alchemy-test";
8import * as Data from "effect/Data";
9import * as Effect from "effect/Effect";
10import * as Schedule from "effect/Schedule";
11import * as Stream from "effect/Stream";
12import * as HttpClient from "effect/http/HttpClient";
13import * as HttpClientRequest from "effect/http/HttpClientRequest";
14import BatchTestFunctionLive, { BatchTestFunction } from "./handler";
15
16const testOptions = { providers: AWS.providers() };
17const { test, beforeAll, afterAll } = Test.make(testOptions);
18const sharedStack = Core.scratchStack(testOptions, "BatchBindings");
19
20let baseUrl: string;
21const submittedJobIds = new Set<string>();
22
23// AWS Batch auto-creates the shared, account-level `/aws/batch/job` log group
24// the first time a job writes logs — it is NOT deleted with the compute
25// environment / job queue / job definition. The JobDefinition provider reaps
26// the family's log *streams* on delete, but the group itself lingers, and a
27// stream flushed by a job finishing while the stack is mid-destroy can land
28// after that reap. So on teardown: delete every stream this suite caused
29// (family names are prefixed `{stackName}-`), then delete the group itself
30// only when zero streams remain (nothing else is using it). The Batch service
31// recreates the group on the next job, so deleting the empty group is safe.
32// Idempotent: every delete tolerates ResourceNotFoundException; any other
33// error is a defect so this is a valid `Effect.ensuring` finalizer.
34const reapBatchJobLogGroup = Core.withProviders(
35 Effect.gen(function* () {
36 const logGroupName = "/aws/batch/job";
37 const streams = yield* logs.describeLogStreams.pages({ logGroupName }).pipe(
38 Stream.runCollect,
39 Effect.map((pages) =>
40 Array.from(pages).flatMap((page) => page.logStreams ?? []),
41 ),
42 Effect.catchTag("ResourceNotFoundException", () =>
43 Effect.succeed(undefined),
44 ),
45 );
46 if (streams === undefined) return; // group never created / already reaped
47
48 const isTestOwned = (name: string | undefined): boolean =>
49 name !== undefined && name.startsWith(`${sharedStack.name}-`);
50 yield* Effect.forEach(
51 streams.filter((s) => isTestOwned(s.logStreamName)),
52 (s) =>
53 logs
54 .deleteLogStream({ logGroupName, logStreamName: s.logStreamName! })
55 .pipe(
56 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
57 ),
58 { concurrency: 4 },
59 );
60
61 // Shared group: only remove it when no foreign streams remain.
62 if (streams.some((s) => !isTestOwned(s.logStreamName))) return;
63 yield* logs
64 .deleteLogGroup({ logGroupName })
65 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
66 }).pipe(Effect.orDie),
67 testOptions,
68 sharedStack.name,
69);
70
71class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
72 readonly status: number;
73 readonly body: string;
74}> {}
75
76// Fresh Lambda role + Batch permissions propagate eventually — the first
77// submits can 500 with AccessDenied under the handler's `Effect.orDie`.
78// Retry 5xx only; a genuine 4xx fails immediately.
79const send = (request: HttpClientRequest.HttpClientRequest) =>
80 HttpClient.execute(request).pipe(
81 Effect.flatMap((response) =>
82 response.status >= 500
83 ? response.text.pipe(
84 Effect.flatMap((body) =>
85 Effect.fail(
86 new TransientUpstream({ status: response.status, body }),
87 ),
88 ),
89 )
90 : Effect.succeed(response),
91 ),
92 Effect.retry({
93 while: (e) => e._tag === "TransientUpstream",
94 schedule: Schedule.max([
95 Schedule.exponential("1 second"),
96 Schedule.recurs(8),
97 ]),
98 }),
99 );
100
101const submit = (jobName: string) =>
102 Effect.gen(function* () {
103 const response = yield* send(
104 HttpClientRequest.post(`${baseUrl}/submit`).pipe(
105 HttpClientRequest.bodyJsonUnsafe({ jobName }),
106 ),
107 );
108 expect(response.status).toBe(200);
109 const submitted = (yield* response.json) as {
110 jobId: string;
111 jobName: string;
112 jobArn?: string;
113 };
114 submittedJobIds.add(submitted.jobId);
115 return submitted;
116 });
117
118/** Out-of-band job status via distilled. */
119const jobStatus = (jobId: string) =>
120 batch
121 .describeJobs({ jobs: [jobId] })
122 .pipe(Effect.map((res) => res.jobs?.[0]?.status));
123
124const terminalJobStatuses = new Set(["SUCCEEDED", "FAILED"]);
125
126/**
127 * Release Fargate tasks before deleting the queue and compute environment.
128 *
129 * Several binding tests intentionally stop once a submitted job is visible;
130 * leaving those jobs active makes Batch drain its managed ENIs only after
131 * stack teardown has already started. Terminating every non-terminal job here
132 * moves that asynchronous wait to the front of teardown, before the VPC graph
133 * competes for the suite's hard timeout.
134 */
135const drainSubmittedJobs = Core.withProviders(
136 Effect.gen(function* () {
137 const jobIds = [...submittedJobIds];
138 if (jobIds.length === 0) return;
139
140 const describeSubmitted = batch
141 .describeJobs({ jobs: jobIds })
142 .pipe(Effect.map((response) => response.jobs ?? []));
143 const jobs = yield* describeSubmitted;
144 yield* Effect.forEach(
145 jobs.filter((job) => !terminalJobStatuses.has(job.status ?? "")),
146 (job) =>
147 batch.terminateJob({
148 jobId: job.jobId!,
149 reason: "Batch binding test teardown",
150 }),
151 { concurrency: 4, discard: true },
152 );
153
154 yield* describeSubmitted.pipe(
155 Effect.repeat({
156 schedule: Schedule.spaced("2 seconds"),
157 until: (observed) =>
158 observed.every((job) => terminalJobStatuses.has(job.status ?? "")),
159 times: 10,
160 }),
161 );
162 submittedJobIds.clear();
163 }).pipe(Effect.orDie),
164 testOptions,
165 sharedStack.name,
166);
167
168describe(
169 "Batch Bindings",
170 {
171 tags: [
172 "provider:aws",
173 "provider:aws:batch",
174 "provider:aws:ec2",
175 "provider:aws:iam",
176 "provider:aws:lambda",
177 "live",
178 ],
179 },
180 () => {
181 beforeAll(
182 Effect.gen(function* () {
183 yield* Effect.logInfo(
184 "Batch test setup: destroying previous resources",
185 );
186 yield* sharedStack.destroy();
187
188 yield* Effect.logInfo(
189 "Batch test setup: deploying CE -> queue -> job definition -> Lambda",
190 );
191 const { functionUrl } = yield* sharedStack.deploy(
192 Effect.gen(function* () {
193 return yield* BatchTestFunction;
194 }).pipe(Effect.provide(BatchTestFunctionLive)),
195 );
196
197 expect(functionUrl).toBeTruthy();
198 baseUrl = functionUrl!.replace(/\/+$/, "");
199
200 // Readiness probe — fresh function URLs take seconds (sometimes over a
201 // minute) to serve 200s.
202 yield* HttpClient.get(`${baseUrl}/status?jobId=readiness-probe`).pipe(
203 Effect.flatMap((response) =>
204 response.status === 200
205 ? Effect.succeed(response)
206 : Effect.fail(
207 new Error(`Function not ready: ${response.status}`),
208 ),
209 ),
210 Effect.retry({
211 schedule: Schedule.max([
212 Schedule.fixed("2 seconds"),
213 Schedule.recurs(75),
214 ]),
215 }),
216 );
217 }),
218 { timeout: 300_000 },
219 );
220 afterAll(
221 drainSubmittedJobs.pipe(
222 Effect.andThen(sharedStack.destroy()),
223 Effect.ensuring(reapBatchJobLogGroup),
224 ),
225 { timeout: 240_000 },
226 );
227
228 describe("SubmitJob", () => {
229 test.provider(
230 "submits a job that leaves the queue head (RUNNABLE or beyond)",
231 () =>
232 Effect.gen(function* () {
233 const { jobId, jobName } = yield* submit("alchemy-e2e-echo");
234 expect(jobName).toBe("alchemy-e2e-echo");
235 expect(jobId).toBeTruthy();
236
237 // Out-of-band: the job progresses past SUBMITTED/PENDING within
238 // ~2 minutes on a healthy Fargate CE (bounded poll).
239 const status = yield* jobStatus(jobId).pipe(
240 Effect.repeat({
241 schedule: Schedule.spaced("5 seconds"),
242 until: (s) =>
243 s === "RUNNABLE" ||
244 s === "STARTING" ||
245 s === "RUNNING" ||
246 s === "SUCCEEDED" ||
247 s === "FAILED",
248 times: 24,
249 }),
250 );
251 expect(["RUNNABLE", "STARTING", "RUNNING", "SUCCEEDED"]).toContain(
252 status,
253 );
254 }),
255 { timeout: 240_000 },
256 );
257
258 // The full RUNNABLE→STARTING→SUCCEEDED round-trip takes 1-3 minutes of
259 // Fargate provisioning — gated so the default suite stays in budget.
260 test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
261 "submitted echo job runs to SUCCEEDED (AWS_TEST_SLOW=1)",
262 () =>
263 Effect.gen(function* () {
264 const { jobId } = yield* submit("alchemy-e2e-echo-succeed");
265 const status = yield* jobStatus(jobId).pipe(
266 Effect.repeat({
267 schedule: Schedule.spaced("10 seconds"),
268 until: (s) => s === "SUCCEEDED" || s === "FAILED",
269 times: 42,
270 }),
271 );
272 expect(status).toBe("SUCCEEDED");
273 }),
274 { timeout: 480_000 },
275 );
276 });
277
278 describe("DescribeJobs", () => {
279 test.provider(
280 "describes a submitted job from the runtime",
281 () =>
282 Effect.gen(function* () {
283 const { jobId } = yield* submit("alchemy-e2e-describe");
284
285 const response = yield* send(
286 HttpClientRequest.get(`${baseUrl}/status?jobId=${jobId}`),
287 );
288 expect(response.status).toBe(200);
289 const body = (yield* response.json) as {
290 status?: string;
291 jobQueue?: string;
292 };
293 expect(body.status).toBeTruthy();
294 expect(body.jobQueue).toContain("job-queue/");
295 }),
296 { timeout: 120_000 },
297 );
298 });
299
300 describe("CancelJob", () => {
301 test.provider(
302 "cancels a queued job from the runtime",
303 () =>
304 Effect.gen(function* () {
305 const { jobId } = yield* submit("alchemy-e2e-cancel");
306
307 // Cancel immediately — the job is still SUBMITTED/PENDING/RUNNABLE
308 // (Fargate placement takes ~a minute), so cancellation applies.
309 const response = yield* send(
310 HttpClientRequest.post(`${baseUrl}/cancel`).pipe(
311 HttpClientRequest.bodyJsonUnsafe({
312 jobId,
313 reason: "alchemy e2e cancel test",
314 }),
315 ),
316 );
317 expect(response.status).toBe(200);
318 expect((yield* response.json) as object).toEqual({
319 cancelled: true,
320 });
321
322 // Out-of-band: a cancelled-before-starting job lands in FAILED.
323 // (SUCCEEDED only if the cancel raced past STARTING — accepted to
324 // keep the test deterministic, but FAILED is the expected path.)
325 const status = yield* jobStatus(jobId).pipe(
326 Effect.repeat({
327 schedule: Schedule.spaced("5 seconds"),
328 until: (s): boolean => s === "FAILED" || s === "SUCCEEDED",
329 times: 36,
330 }),
331 );
332 expect(["FAILED", "SUCCEEDED"]).toContain(status);
333 }),
334 { timeout: 240_000 },
335 );
336 });
337
338 describe("ListJobs", () => {
339 test.provider(
340 "lists a submitted job in the bound queue from the runtime",
341 () =>
342 Effect.gen(function* () {
343 const { jobId } = yield* submit("alchemy-e2e-list");
344
345 // The job moves through SUBMITTED→PENDING→RUNNABLE→STARTING→…;
346 // poll the runtime ListJobs across the non-terminal statuses until
347 // the submitted job shows up.
348 const findJob = Effect.gen(function* () {
349 for (const jobStatus of [
350 "SUBMITTED",
351 "PENDING",
352 "RUNNABLE",
353 "STARTING",
354 "RUNNING",
355 "SUCCEEDED",
356 ]) {
357 const response = yield* send(
358 HttpClientRequest.get(
359 `${baseUrl}/jobs?jobStatus=${jobStatus}`,
360 ),
361 );
362 expect(response.status).toBe(200);
363 const body = (yield* response.json) as {
364 jobs: { jobId: string; jobName: string }[];
365 };
366 const match = body.jobs.find((job) => job.jobId === jobId);
367 if (match) return match;
368 }
369 return undefined;
370 });
371 const found = yield* findJob.pipe(
372 Effect.repeat({
373 schedule: Schedule.spaced("5 seconds"),
374 until: (match): boolean => match !== undefined,
375 times: 24,
376 }),
377 );
378 expect(found?.jobId).toBe(jobId);
379 expect(found?.jobName).toBe("alchemy-e2e-list");
380 }),
381 { timeout: 240_000 },
382 );
383 });
384
385 describe("GetJobQueueSnapshot", () => {
386 test.provider(
387 "snapshots the front of the bound queue from the runtime",
388 () =>
389 Effect.gen(function* () {
390 // The queue-ARN-scoped IAM grant + injected `jobQueue` selector:
391 // a 200 with a jobs array proves the call path end to end (the
392 // snapshot only surfaces RUNNABLE jobs, so membership is racy and
393 // not asserted).
394 const response = yield* send(
395 HttpClientRequest.get(`${baseUrl}/snapshot`),
396 );
397 expect(response.status).toBe(200);
398 const body = (yield* response.json) as { jobs: unknown };
399 expect(Array.isArray(body.jobs)).toBe(true);
400 }),
401 { timeout: 120_000 },
402 );
403 });
404
405 describe("consumeJobEvents", { tags: ["provider:aws:eventbridge"] }, () => {
406 test.provider(
407 "created the EventBridge rule for batch job state changes",
408 () =>
409 Effect.gen(function* () {
410 // The rule's physical name embeds the fixture's logical id
411 // (`BatchTestFunction-BatchJobEvents`); find it on the default bus
412 // with bounded manual pagination.
413 let rule: eventbridge.Rule | undefined;
414 let nextToken: string | undefined;
415 for (let page = 0; page < 10 && !rule; page++) {
416 const result = yield* eventbridge.listRules({
417 NextToken: nextToken,
418 });
419 rule = (result.Rules ?? []).find((candidate) =>
420 candidate.Name?.includes("BatchJobEvents"),
421 );
422 nextToken = result.NextToken;
423 if (!nextToken) break;
424 }
425 expect(rule).toBeDefined();
426 expect(rule?.EventPattern).toContain("aws.batch");
427 expect(rule?.EventPattern).toContain("Batch Job State Change");
428 }),
429 { timeout: 60_000 },
430 );
431 });
432
433 describe("TerminateJob", () => {
434 test.provider(
435 "terminates a submitted job from the runtime",
436 () =>
437 Effect.gen(function* () {
438 const { jobId } = yield* submit("alchemy-e2e-terminate");
439
440 const response = yield* send(
441 HttpClientRequest.post(`${baseUrl}/terminate`).pipe(
442 HttpClientRequest.bodyJsonUnsafe({
443 jobId,
444 reason: "alchemy e2e terminate test",
445 }),
446 ),
447 );
448 expect(response.status).toBe(200);
449 expect((yield* response.json) as object).toEqual({
450 terminated: true,
451 });
452
453 // Out-of-band: a job terminated before running lands in FAILED.
454 const status = yield* jobStatus(jobId).pipe(
455 Effect.repeat({
456 schedule: Schedule.spaced("5 seconds"),
457 until: (s) => s === "FAILED" || s === "SUCCEEDED",
458 times: 24,
459 }),
460 );
461 expect(status).toBe("FAILED");
462 }),
463 { timeout: 240_000 },
464 );
465 });
466 },
467);