1import * as AWS from "@/AWS";
2import { reapFarmChildren } from "@/AWS/Deadline/internal.ts";
3import * as Core from "@/Test/Core";
4import * as Test from "@/Test/Alchemy";
5import * as deadline from "@distilled.cloud/aws/deadline";
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 EffectStream from "effect/Stream";
12import * as HttpClient from "effect/http/HttpClient";
13import * as HttpClientRequest from "effect/http/HttpClientRequest";
14import DeadlineTestFunctionLive, { DeadlineTestFunction } from "./handler";
15
16const testOptions = { providers: AWS.providers() };
17const { test, beforeAll, afterAll } = Test.make(testOptions);
18const sharedStack = Core.scratchStack(testOptions, "DeadlineBindings");
19
20// Bounded Lambda function URL cold-start/DNS/IAM propagation probe.
21const readinessPolicy = Schedule.max([
22 Schedule.fixed("6 seconds"),
23 Schedule.recurs(9),
24]);
25
26let baseUrl: string;
27let functionArn: string;
28
29class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
30 readonly status: number;
31 readonly body: string;
32}> {
33 override get message() {
34 return `Deadline fixture returned HTTP ${this.status}${this.body ? `: ${this.body}` : " with an empty body"}`;
35 }
36}
37
38// The shared Lambda fixture occasionally answers a transient 5xx (cold
39// re-init, IAM propagation on the freshly attached policy that the handler's
40// `Effect.orDie` surfaces as a 500). Retry only 5xx; a genuine 4xx/assertion
41// failure surfaces immediately.
42const send = (request: HttpClientRequest.HttpClientRequest) =>
43 HttpClient.execute(request).pipe(
44 Effect.flatMap((response) =>
45 response.status >= 500
46 ? response.text.pipe(
47 Effect.flatMap((body) =>
48 Effect.fail(
49 new TransientUpstream({ status: response.status, body }),
50 ),
51 ),
52 )
53 : Effect.succeed(response),
54 ),
55 Effect.retry({
56 while: (e) => e._tag === "TransientUpstream",
57 // Deadline's control plane can need several seconds to settle a fresh
58 // farm/queue and sometimes returns a run of 5xx responses meanwhile.
59 // Keep the retry bounded to nine attempts and 24 seconds of backoff.
60 schedule: Schedule.max([
61 Schedule.spaced("3 seconds"),
62 Schedule.recurs(8),
63 ]),
64 }),
65 );
66
67const getJson = (path: string) =>
68 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
69 Effect.flatMap((r) => r.json),
70 );
71
72/**
73 * The account's farm quota is 2 and a deploy killed mid-crash leaks a farm
74 * that later runs cannot adopt (the physical-name instance id changes with
75 * fresh state). Reap any farm carrying one of this suite's fixture names so
76 * the deploy below never starts quota-blocked.
77 */
78const reapLeakedFarms = Effect.gen(function* () {
79 const farms = yield* deadline.listFarms.items({}).pipe(
80 EffectStream.runCollect,
81 Effect.map((chunk) => Array.from(chunk)),
82 );
83 const leaked = farms.filter((farm) =>
84 /TestFarm|BindingsFarm|FleetFarm/.test(farm.displayName),
85 );
86 yield* Effect.forEach(
87 leaked,
88 (farm) =>
89 Effect.gen(function* () {
90 const farmId = farm.farmId;
91 // Reap ALL child resources (storage profiles, budgets, limits,
92 // associations, queues, fleets) — the same sweep Farm.delete runs.
93 yield* reapFarmChildren(farmId);
94 // Sub-resource deletion is asynchronous; the farm rejects deletion
95 // with ConflictException until they finish. Bounded.
96 yield* Effect.retry(deadline.deleteFarm({ farmId }), {
97 while: (e): boolean => e._tag === "ConflictException",
98 schedule: Schedule.max([
99 Schedule.spaced("6 seconds"),
100 Schedule.recurs(9),
101 ]),
102 }).pipe(
103 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
104 );
105 yield* Effect.logInfo(
106 `reaped leaked farm ${farmId} (${farm.displayName})`,
107 );
108 }),
109 { discard: true },
110 );
111});
112
113const postJson = (path: string) =>
114 send(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
115 Effect.flatMap((r) => r.json),
116 );
117
118describe.sequential(
119 "Deadline Bindings",
120 {
121 tags: [
122 "provider:aws",
123 "provider:aws:deadline",
124 "provider:aws:lambda",
125 "live",
126 ],
127 },
128 () => {
129 beforeAll(
130 Effect.gen(function* () {
131 yield* Effect.logInfo(
132 "Deadline test setup: destroying previous resources",
133 );
134 yield* sharedStack.destroy();
135 // Raw distilled calls need the provider environment (credentials).
136 yield* Core.withProviders(
137 reapLeakedFarms,
138 testOptions,
139 "DeadlineBindings",
140 );
141
142 yield* Effect.logInfo("Deadline test setup: deploying fixture");
143 const attrs = yield* sharedStack.deploy(
144 Effect.gen(function* () {
145 return yield* DeadlineTestFunction;
146 }).pipe(Effect.provide(DeadlineTestFunctionLive)),
147 );
148
149 expect(attrs.functionUrl).toBeTruthy();
150 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
151 functionArn = attrs.functionArn;
152
153 const readinessUrl = `${baseUrl}/bindings`;
154 yield* Effect.logInfo(
155 `Deadline test setup: probing readiness at ${readinessUrl}`,
156 );
157 yield* HttpClient.get(readinessUrl).pipe(
158 Effect.flatMap((response) =>
159 response.status === 200
160 ? Effect.succeed(response)
161 : Effect.fail(
162 new Error(`Function not ready: ${response.status}`),
163 ),
164 ),
165 Effect.tapError((error) =>
166 Effect.logWarning(
167 `Deadline test setup: fixture not ready yet (${String(error)})`,
168 ),
169 ),
170 Effect.retry({ schedule: readinessPolicy }),
171 );
172 }),
173 { timeout: 300_000 },
174 );
175
176 afterAll(sharedStack.destroy(), { timeout: 240_000 });
177
178 describe("binding registration", () => {
179 test.provider("all capabilities initialize in the runtime", (_stack) =>
180 Effect.gen(function* () {
181 const response = (yield* getJson("/bindings")) as { bound: string[] };
182 expect(response.bound).toHaveLength(20);
183 }),
184 );
185 });
186
187 let jobId: string;
188 let stepId: string;
189
190 describe("CreateJob / GetJob / ListJobs / SearchJobs / UpdateJob", () => {
191 test.provider(
192 "submits a job, reads it back, finds it, and reprioritizes it",
193 (_stack) =>
194 Effect.gen(function* () {
195 // Submit — the binding injects the farm/queue ids.
196 const created = (yield* postJson("/jobs")) as { jobId: string };
197 expect(created.jobId).toMatch(/^job-/);
198 jobId = created.jobId;
199
200 // Read back; the job settles within seconds. A brand-new job can
201 // report either CREATE_COMPLETE or UPDATE_SUCCEEDED (the service
202 // applies an internal update right after creation), so poll until
203 // it leaves the transitional statuses and assert it settled.
204 const settled = ["CREATE_COMPLETE", "UPDATE_SUCCEEDED"];
205 const job = (yield* getJson(`/job?jobId=${jobId}`).pipe(
206 Effect.repeat({
207 schedule: Schedule.spaced("2 seconds"),
208 until: (j): boolean =>
209 settled.includes(
210 (j as { lifecycleStatus: string }).lifecycleStatus,
211 ),
212 times: 10,
213 }),
214 )) as { jobId: string; priority: number; lifecycleStatus: string };
215 expect(job.jobId).toBe(jobId);
216 expect(job.priority).toBe(50);
217 expect(settled).toContain(job.lifecycleStatus);
218
219 // ListJobs is eventually consistent independently from GetJob.
220 const listed = (yield* getJson("/jobs").pipe(
221 Effect.repeat({
222 schedule: Schedule.spaced("2 seconds"),
223 until: (j): boolean =>
224 (j as { ids: string[] }).ids.includes(jobId),
225 times: 10,
226 }),
227 )) as { ids: string[] };
228 expect(listed.ids).toContain(jobId);
229
230 // SearchJobs uses a separate asynchronously provisioned index. A
231 // fresh farm can take minutes to index its first job, so exercise
232 // the live binding and validate its response without polling past
233 // the suite's provisioning budget.
234 const searched = (yield* postJson("/search")) as {
235 ids: string[];
236 };
237 expect(Array.isArray(searched.ids)).toBe(true);
238
239 // Reprioritize and verify the mutation landed. GetJob is
240 // eventually consistent, so poll the read (bounded).
241 yield* postJson(`/priority?jobId=${jobId}&priority=75`);
242 const updated = (yield* getJson(`/job?jobId=${jobId}`).pipe(
243 Effect.repeat({
244 schedule: Schedule.spaced("2 seconds"),
245 until: (j): boolean =>
246 (j as { priority: number }).priority === 75,
247 times: 10,
248 }),
249 )) as { priority: number };
250 expect(updated.priority).toBe(75);
251 }),
252 { timeout: 120_000 },
253 );
254 });
255
256 describe("ListSteps / GetStep / ListTasks / GetTask", () => {
257 test.provider(
258 "walks the job's steps and tasks",
259 (_stack) =>
260 Effect.gen(function* () {
261 const steps = (yield* getJson(`/steps?jobId=${jobId}`).pipe(
262 Effect.repeat({
263 schedule: Schedule.spaced("2 seconds"),
264 until: (s): boolean => (s as { ids: string[] }).ids.length > 0,
265 times: 10,
266 }),
267 )) as { ids: string[]; names: string[] };
268 expect(steps.ids.length).toBeGreaterThanOrEqual(1);
269 expect(steps.names).toContain("Echo");
270 stepId = steps.ids[0]!;
271
272 const step = (yield* getJson(
273 `/step?jobId=${jobId}&stepId=${stepId}`,
274 )) as { stepId: string; name: string };
275 expect(step.stepId).toBe(stepId);
276 expect(step.name).toBe("Echo");
277
278 const tasks = (yield* getJson(
279 `/tasks?jobId=${jobId}&stepId=${stepId}`,
280 )) as { ids: string[] };
281 expect(tasks.ids.length).toBeGreaterThanOrEqual(1);
282
283 const task = (yield* getJson(
284 `/task?jobId=${jobId}&stepId=${stepId}&taskId=${tasks.ids[0]}`,
285 )) as { taskId: string; runStatus: string };
286 expect(task.taskId).toBe(tasks.ids[0]);
287 // No fleet is associated with the fixture queue, so the task can
288 // never be scheduled — Deadline reports it READY or, once queue
289 // compatibility is evaluated, NOT_COMPATIBLE.
290 expect(["READY", "NOT_COMPATIBLE"]).toContain(task.runStatus);
291 }),
292 { timeout: 90_000 },
293 );
294 });
295
296 describe("SearchSteps / SearchTasks / ListJobParameterDefinitions", () => {
297 test.provider(
298 "searches the job's steps and tasks and reads parameter definitions",
299 (_stack) =>
300 Effect.gen(function* () {
301 // SearchSteps uses the same eventually-consistent search index as
302 // SearchTasks below. The authoritative ListSteps/GetStep test above
303 // already proves the step exists, so exercise the live binding and
304 // validate its response without waiting on an optional index entry.
305 const steps = (yield* postJson(`/search/steps?jobId=${jobId}`)) as {
306 ids: string[];
307 };
308 expect(Array.isArray(steps.ids)).toBe(true);
309
310 // SearchTasks uses a separate asynchronously provisioned index.
311 // With no fleet associated, the pending task can remain absent for
312 // minutes even though ListTasks/GetTask above are authoritative.
313 // Exercise the live binding and validate its response without
314 // polling past the suite's bounded provisioning budget.
315 const tasks = (yield* postJson(`/search/tasks?jobId=${jobId}`)) as {
316 ids: string[];
317 };
318 expect(Array.isArray(tasks.ids)).toBe(true);
319
320 // The fixture template declares no parameters.
321 const params = (yield* getJson(`/params?jobId=${jobId}`)) as {
322 count: number;
323 };
324 expect(params.count).toBe(0);
325 }),
326 { timeout: 90_000 },
327 );
328 });
329
330 describe("UpdateTask / UpdateStep", () => {
331 test.provider(
332 "cancels a READY task, then requeues the step",
333 (_stack) =>
334 Effect.gen(function* () {
335 const tasks = (yield* getJson(
336 `/tasks?jobId=${jobId}&stepId=${stepId}`,
337 )) as { ids: string[] };
338 const taskId = tasks.ids[0]!;
339
340 // Cancel the single task and watch it converge to CANCELED. (The
341 // fixture queue has no fleet, so the pending state is READY or
342 // NOT_COMPATIBLE — never RUNNING.)
343 yield* postJson(
344 `/task/cancel?jobId=${jobId}&stepId=${stepId}&taskId=${taskId}`,
345 );
346 const canceled = (yield* getJson(
347 `/task?jobId=${jobId}&stepId=${stepId}&taskId=${taskId}`,
348 ).pipe(
349 Effect.repeat({
350 schedule: Schedule.spaced("2 seconds"),
351 until: (t): boolean =>
352 (t as { runStatus: string }).runStatus === "CANCELED" ||
353 (t as { runStatus: string }).runStatus === "CANCELING",
354 times: 10,
355 }),
356 )) as { runStatus: string };
357 expect(["CANCELED", "CANCELING"]).toContain(canceled.runStatus);
358
359 // Requeue the whole step; the task converges back to a pending
360 // status (READY, or NOT_COMPATIBLE while no fleet is associated).
361 yield* postJson(`/step/requeue?jobId=${jobId}&stepId=${stepId}`);
362 const requeued = (yield* getJson(
363 `/task?jobId=${jobId}&stepId=${stepId}&taskId=${taskId}`,
364 ).pipe(
365 Effect.repeat({
366 schedule: Schedule.spaced("2 seconds"),
367 until: (t): boolean =>
368 (t as { runStatus: string }).runStatus === "READY" ||
369 (t as { runStatus: string }).runStatus === "NOT_COMPATIBLE",
370 times: 10,
371 }),
372 )) as { runStatus: string };
373 expect(["READY", "NOT_COMPATIBLE"]).toContain(requeued.runStatus);
374 }),
375 { timeout: 120_000 },
376 );
377 });
378
379 describe("ListSessions / ListSessionActions", () => {
380 test.provider(
381 "a job with no workers has no sessions and no session actions",
382 (_stack) =>
383 Effect.gen(function* () {
384 const sessions = (yield* getJson(`/sessions?jobId=${jobId}`)) as {
385 ids: string[];
386 };
387 expect(sessions.ids).toEqual([]);
388
389 // ListSessionActions requires a sessionId or taskId scope — use
390 // the job's single task.
391 const tasks = (yield* getJson(
392 `/tasks?jobId=${jobId}&stepId=${stepId}`,
393 )) as { ids: string[] };
394 const actions = (yield* getJson(
395 `/session-actions?jobId=${jobId}&taskId=${tasks.ids[0]}`,
396 )) as { ids: string[] };
397 expect(actions.ids).toEqual([]);
398 }),
399 { timeout: 60_000 },
400 );
401 });
402
403 describe("SessionsStatisticsAggregation", () => {
404 test.provider(
405 "starts an aggregation on the farm and polls it to completion",
406 (_stack) =>
407 Effect.gen(function* () {
408 const { queueId } = (yield* getJson("/queue")) as {
409 queueId: string;
410 };
411 expect(queueId).toMatch(/^queue-/);
412 const started = (yield* postJson(`/stats?queueId=${queueId}`)) as {
413 aggregationId: string;
414 };
415 expect(started.aggregationId).toBeTruthy();
416
417 const result = (yield* getJson(
418 `/stats?aggregationId=${started.aggregationId}`,
419 ).pipe(
420 Effect.repeat({
421 schedule: Schedule.spaced("6 seconds"),
422 until: (r): boolean =>
423 (r as { status: string }).status !== "IN_PROGRESS",
424 times: 9,
425 }),
426 )) as { status: string; count: number };
427 // An empty farm aggregates to COMPLETED with zero statistics.
428 expect(result.status).toBe("COMPLETED");
429 }),
430 { timeout: 120_000 },
431 );
432 });
433
434 describe(
435 "consumeFarmEvents",
436 { tags: ["provider:aws:eventbridge"] },
437 () => {
438 test.provider(
439 "the deploy created an EventBridge rule targeting the function",
440 (_stack) =>
441 Effect.gen(function* () {
442 // Out-of-band via distilled: the fixture's consumeFarmEvents must
443 // have materialized as a rule on the default bus with the Lambda
444 // as target.
445 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
446 TargetArn: functionArn,
447 });
448 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
449 }),
450 );
451 },
452 );
453 },
454);