1import * as deadline from "@distilled.cloud/aws/deadline";
2import * as Data from "effect/Data";
3import type * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import type * as Redacted from "effect/Redacted";
6import * as Schedule from "effect/Schedule";
7import * as EffectStream from "effect/Stream";
8import { Unowned } from "../../AdoptPolicy.ts";
9import { isResolved } from "../../Diff.ts";
10import { createPhysicalName } from "../../PhysicalName.ts";
11import * as Provider from "../../Provider.ts";
12import { Resource } from "../../Resource.ts";
13import { createInternalTags, hasAlchemyTags, type Tags } from "../../Tags.ts";
14import { toWireHours, toWireSeconds } from "../../Util/Duration.ts";
15import type { Providers } from "../Providers.ts";
16import {
17 asPlain,
18 deadlineArnOf,
19 fetchDeadlineTags,
20 retryThroughIamPropagation,
21 retryWhileConflict,
22 retryWhileFarmSettling,
23 syncDeadlineTags,
24} from "./internal.ts";
25
26export type FleetStatus = deadline.FleetStatus;
27
28/**
29 * Worker auto-scaling settings for a fleet, with idle time expressed as a
30 * `Duration.Input` (converted to whole seconds on the wire).
31 */
32export interface FleetAutoScalingConfiguration {
33 /**
34 * Number of idle workers kept on standby to absorb bursts.
35 * @default 0
36 */
37 standbyWorkerCount?: number;
38 /**
39 * How long a worker may sit idle before it is scaled in
40 * (wire: `workerIdleDurationSeconds`).
41 * @default "5 minutes"
42 */
43 workerIdleDuration?: Duration.Input;
44 /**
45 * Maximum number of workers added per minute when scaling out.
46 */
47 scaleOutWorkersPerMinute?: number;
48}
49
50/**
51 * Customer-managed fleet configuration (you run the worker hosts).
52 */
53export interface CustomerManagedFleetConfiguration extends Omit<
54 deadline.CustomerManagedFleetConfiguration,
55 "autoScalingConfiguration"
56> {
57 /**
58 * Auto-scaling behavior for `EVENT_BASED_AUTO_SCALING` mode.
59 */
60 autoScalingConfiguration?: FleetAutoScalingConfiguration;
61}
62
63/**
64 * Persistent EBS volume settings for service-managed EC2 workers, with the
65 * reuse TTL expressed as a `Duration.Input` (converted to whole hours on the
66 * wire).
67 */
68export interface FleetPersistentVolumeConfiguration extends Omit<
69 deadline.PersistentVolumeConfiguration,
70 "lastUsedTtlHours"
71> {
72 /**
73 * How long an unused persistent volume is retained for reuse before it is
74 * deleted (wire: `lastUsedTtlHours`).
75 */
76 lastUsedTtl?: Duration.Input;
77}
78
79/**
80 * Service-managed EC2 fleet configuration (Deadline provisions instances).
81 */
82export interface ServiceManagedEc2FleetConfiguration extends Omit<
83 deadline.ServiceManagedEc2FleetConfiguration,
84 "autoScalingConfiguration" | "persistentVolumeConfiguration"
85> {
86 /**
87 * Auto-scaling behavior for the EC2 workers.
88 */
89 autoScalingConfiguration?: FleetAutoScalingConfiguration;
90 /**
91 * Persistent EBS volume reused across workers.
92 */
93 persistentVolumeConfiguration?: FleetPersistentVolumeConfiguration;
94}
95
96/**
97 * Fleet configuration — either `customerManaged` (you run the workers) or
98 * `serviceManagedEc2` (Deadline provisions EC2 instances).
99 */
100export type FleetConfiguration =
101 | {
102 customerManaged: CustomerManagedFleetConfiguration;
103 serviceManagedEc2?: never;
104 }
105 | {
106 customerManaged?: never;
107 serviceManagedEc2: ServiceManagedEc2FleetConfiguration;
108 };
109
110/**
111 * Startup script run on each worker host, with the timeout expressed as a
112 * `Duration.Input` (converted to whole seconds on the wire).
113 */
114export interface FleetHostConfiguration {
115 /**
116 * The script body executed when a worker host starts.
117 */
118 scriptBody: string | Redacted.Redacted<string>;
119 /**
120 * Maximum time the startup script may run before the worker is marked
121 * unhealthy (wire: `scriptTimeoutSeconds`).
122 * @default "5 minutes"
123 */
124 scriptTimeout?: Duration.Input;
125}
126
127const toWireAutoScaling = (
128 config: FleetAutoScalingConfiguration | undefined,
129): deadline.CustomerManagedAutoScalingConfiguration | undefined => {
130 if (config === undefined) return undefined;
131 const { workerIdleDuration, ...rest } = config;
132 return {
133 ...rest,
134 workerIdleDurationSeconds: toWireSeconds(workerIdleDuration),
135 };
136};
137
138const toWirePersistentVolume = (
139 config: FleetPersistentVolumeConfiguration | undefined,
140): deadline.PersistentVolumeConfiguration | undefined => {
141 if (config === undefined) return undefined;
142 const { lastUsedTtl, ...rest } = config;
143 return { ...rest, lastUsedTtlHours: toWireHours(lastUsedTtl) };
144};
145
146const toWireConfiguration = (
147 config: FleetConfiguration,
148): deadline.FleetConfiguration =>
149 config.customerManaged !== undefined
150 ? {
151 customerManaged: {
152 ...config.customerManaged,
153 autoScalingConfiguration: toWireAutoScaling(
154 config.customerManaged.autoScalingConfiguration,
155 ),
156 },
157 }
158 : {
159 serviceManagedEc2: {
160 ...config.serviceManagedEc2,
161 autoScalingConfiguration: toWireAutoScaling(
162 config.serviceManagedEc2.autoScalingConfiguration,
163 ),
164 persistentVolumeConfiguration: toWirePersistentVolume(
165 config.serviceManagedEc2.persistentVolumeConfiguration,
166 ),
167 },
168 };
169
170const toWireHostConfiguration = (
171 config: FleetHostConfiguration | undefined,
172): deadline.HostConfiguration | undefined => {
173 if (config === undefined) return undefined;
174 const { scriptTimeout, ...rest } = config;
175 return { ...rest, scriptTimeoutSeconds: toWireSeconds(scriptTimeout) };
176};
177
178export interface FleetProps {
179 /**
180 * The identifier of the farm the fleet belongs to. Changing it replaces
181 * the fleet.
182 */
183 farmId: string;
184 /**
185 * Display name of the fleet.
186 * @default ${app}-${stage}-${id}
187 */
188 displayName?: string;
189 /**
190 * A description of the fleet.
191 */
192 description?: string;
193 /**
194 * ARN of the IAM role fleet workers assume (trusted by
195 * `credentials.deadline.amazonaws.com`).
196 */
197 roleArn: string;
198 /**
199 * Minimum number of workers the fleet keeps running.
200 * @default 0
201 */
202 minWorkerCount?: number;
203 /**
204 * Maximum number of workers the fleet scales to.
205 */
206 maxWorkerCount: number;
207 /**
208 * Fleet configuration — either `customerManaged` (you run the workers) or
209 * `serviceManagedEc2` (Deadline provisions EC2 instances).
210 */
211 configuration: FleetConfiguration;
212 /**
213 * Script run on each worker host when it starts.
214 */
215 hostConfiguration?: FleetHostConfiguration;
216 /**
217 * Tags to associate with the fleet.
218 */
219 tags?: Record<string, string>;
220}
221
222export interface Fleet extends Resource<
223 "AWS.Deadline.Fleet",
224 FleetProps,
225 {
226 /**
227 * The identifier of the farm the fleet belongs to.
228 */
229 farmId: string;
230 /**
231 * Service-assigned unique identifier of the fleet (`fleet-...`).
232 */
233 fleetId: string;
234 /**
235 * ARN of the fleet.
236 */
237 fleetArn: string;
238 /**
239 * The fleet's display name.
240 */
241 displayName: string;
242 /**
243 * Current lifecycle status of the fleet.
244 */
245 status: FleetStatus;
246 /**
247 * Number of workers currently in the fleet.
248 */
249 workerCount: number;
250 /**
251 * The configured minimum worker count.
252 */
253 minWorkerCount: number;
254 /**
255 * The configured maximum worker count.
256 */
257 maxWorkerCount: number;
258 /**
259 * ARN of the fleet's worker role.
260 */
261 roleArn: string;
262 /**
263 * Current tags reported for the fleet.
264 */
265 tags: Record<string, string>;
266 },
267 never,
268 Providers
269> {}
270
271/**
272 * An AWS Deadline Cloud fleet — a group of workers (customer-managed hosts
273 * or service-managed EC2 instances) that run render jobs from associated
274 * queues.
275 *
276 * ### Creating Fleets
277 * **Example:** Customer-Managed Fleet
278 * ```typescript
279 * import * as AWS from "alchemy/AWS";
280 *
281 * const fleet = yield* AWS.Deadline.Fleet("Workers", {
282 * farmId: farm.farmId,
283 * roleArn: fleetRole.roleArn,
284 * maxWorkerCount: 10,
285 * configuration: {
286 * customerManaged: {
287 * mode: "NO_SCALING",
288 * workerCapabilities: {
289 * vCpuCount: { min: 1 },
290 * memoryMiB: { min: 1024 },
291 * osFamily: "LINUX",
292 * cpuArchitectureType: "x86_64",
293 * },
294 * },
295 * },
296 * });
297 * ```
298 *
299 * **Example:** Service-Managed EC2 Fleet
300 * ```typescript
301 * const fleet = yield* AWS.Deadline.Fleet("Workers", {
302 * farmId: farm.farmId,
303 * roleArn: fleetRole.roleArn,
304 * minWorkerCount: 0,
305 * maxWorkerCount: 5,
306 * configuration: {
307 * serviceManagedEc2: {
308 * instanceCapabilities: {
309 * vCpuCount: { min: 2, max: 8 },
310 * memoryMiB: { min: 4096 },
311 * osFamily: "LINUX",
312 * cpuArchitectureType: "x86_64",
313 * },
314 * instanceMarketOptions: { type: "spot" },
315 * },
316 * },
317 * });
318 * ```
319 *
320 * @resource
321 */
322export const Fleet = Resource<Fleet>("AWS.Deadline.Fleet");
323
324const createFleetName = (
325 id: string,
326 props: { displayName?: string | undefined },
327) =>
328 props.displayName
329 ? Effect.succeed(props.displayName)
330 : createPhysicalName({ id, maxLength: 100 });
331
332interface FleetState {
333 attrs: Fleet["Attributes"];
334 described: deadline.GetFleetResponse;
335}
336
337const readFleetById = Effect.fn(function* (
338 farmId: string,
339 fleetId: string,
340 arnOf: (path: string) => string,
341) {
342 const described = yield* deadline
343 .getFleet({ farmId, fleetId })
344 .pipe(
345 Effect.catchTag("ResourceNotFoundException", () =>
346 Effect.succeed(undefined),
347 ),
348 );
349 if (!described) return undefined;
350 const fleetArn = arnOf(`farm/${described.farmId}/fleet/${described.fleetId}`);
351 const state: FleetState = {
352 described,
353 attrs: {
354 farmId: described.farmId,
355 fleetId: described.fleetId,
356 fleetArn,
357 displayName: described.displayName,
358 status: described.status,
359 workerCount: described.workerCount,
360 minWorkerCount: described.minWorkerCount,
361 maxWorkerCount: described.maxWorkerCount,
362 roleArn: described.roleArn,
363 tags: yield* fetchDeadlineTags(fleetArn),
364 },
365 };
366 return state;
367});
368
369const findFleetByDisplayName = Effect.fn(function* (
370 farmId: string,
371 displayName: string,
372 arnOf: (path: string) => string,
373) {
374 const summaries = yield* deadline.listFleets.items({ farmId }).pipe(
375 EffectStream.runCollect,
376 Effect.map((chunk) => Array.from(chunk)),
377 // The parent farm may itself be gone.
378 Effect.catchTag("ResourceNotFoundException", () =>
379 Effect.succeed([] as deadline.FleetSummary[]),
380 ),
381 );
382 const match = summaries.find(
383 (summary) => summary.displayName === displayName,
384 );
385 if (!match) return undefined;
386 return yield* readFleetById(farmId, match.fleetId, arnOf);
387});
388
389/**
390 * A fleet still transitioning toward the awaited state — retried by the
391 * bounded wait schedules below.
392 */
393class FleetNotReady extends Data.TaggedError("FleetNotReady")<{
394 readonly fleetId: string;
395 readonly status: string | undefined;
396}> {}
397
398/**
399 * A fleet whose asynchronous provisioning converged to a terminal failed
400 * status (`CREATE_FAILED` / `UPDATE_FAILED`).
401 */
402export class FleetProvisioningFailed extends Data.TaggedError(
403 "FleetProvisioningFailed",
404)<{
405 readonly fleetId: string;
406 readonly status: string;
407 readonly message: string | undefined;
408}> {}
409
410const retryWhileNotReady = <A, E extends { readonly _tag: string }, R>(
411 self: Effect.Effect<A, E, R>,
412): Effect.Effect<A, E, R> =>
413 Effect.retry(self, {
414 while: (e) => e._tag === "FleetNotReady",
415 // Fleet activation is usually well under a minute; stay within the
416 // provider-wide bounded provisioning budget.
417 schedule: Schedule.max([Schedule.spaced("6 seconds"), Schedule.recurs(9)]),
418 });
419
420const waitForFleetActive = (
421 farmId: string,
422 fleetId: string,
423 arnOf: (path: string) => string,
424) =>
425 retryWhileNotReady(
426 Effect.gen(function* () {
427 const state = yield* readFleetById(farmId, fleetId, arnOf);
428 if (state === undefined) {
429 return yield* Effect.fail(
430 new FleetNotReady({ fleetId, status: undefined }),
431 );
432 }
433 if (
434 state.described.status === "CREATE_FAILED" ||
435 state.described.status === "UPDATE_FAILED"
436 ) {
437 return yield* Effect.fail(
438 new FleetProvisioningFailed({
439 fleetId,
440 status: state.described.status,
441 message: state.described.statusMessage,
442 }),
443 );
444 }
445 if (
446 state.described.status !== "ACTIVE" &&
447 state.described.status !== "SUSPENDED"
448 ) {
449 return yield* Effect.fail(
450 new FleetNotReady({ fleetId, status: state.described.status }),
451 );
452 }
453 return state;
454 }),
455 );
456
457const waitUntilFleetGone = (farmId: string, fleetId: string) =>
458 retryWhileNotReady(
459 Effect.gen(function* () {
460 const described = yield* deadline
461 .getFleet({ farmId, fleetId })
462 .pipe(
463 Effect.catchTag("ResourceNotFoundException", () =>
464 Effect.succeed(undefined),
465 ),
466 );
467 if (described !== undefined) {
468 return yield* Effect.fail(
469 new FleetNotReady({ fleetId, status: described.status }),
470 );
471 }
472 }),
473 ).pipe(
474 // Exhausted retries: deletion is already converging server-side.
475 Effect.catchTag("FleetNotReady", () => Effect.void),
476 );
477
478export const FleetProvider = () =>
479 Provider.effect(
480 Fleet,
481 Effect.gen(function* () {
482 return {
483 stables: ["farmId", "fleetId", "fleetArn", "roleArn"],
484 // Keyed by a parent farm — sub-resource list() convention.
485 list: () => Effect.succeed([]),
486 read: Effect.fn(function* ({ id, olds, output }) {
487 const arnOf = yield* deadlineArnOf;
488 const farmId = output?.farmId ?? olds?.farmId;
489 if (farmId === undefined) return undefined;
490 const state = output?.fleetId
491 ? yield* readFleetById(farmId, output.fleetId, arnOf)
492 : yield* findFleetByDisplayName(
493 farmId,
494 yield* createFleetName(id, olds ?? {}),
495 arnOf,
496 );
497 if (!state) return undefined;
498 return (yield* hasAlchemyTags(id, state.attrs.tags as Tags))
499 ? state.attrs
500 : Unowned(state.attrs);
501 }),
502 diff: Effect.fn(function* ({ news, olds }) {
503 if (!isResolved(news)) return;
504 if (olds === undefined) return;
505 // The parent farm is fixed at creation.
506 if (olds.farmId !== news.farmId) {
507 return { action: "replace" } as const;
508 }
509 }),
510 reconcile: Effect.fn(function* ({ id, news, output, session }) {
511 if (news === undefined) {
512 return yield* Effect.fail(
513 new Error("AWS.Deadline.Fleet requires props"),
514 );
515 }
516 const arnOf = yield* deadlineArnOf;
517 const farmId = news.farmId;
518 const displayName = yield* createFleetName(id, news);
519 const internalTags = yield* createInternalTags(id);
520 const desiredTags = { ...internalTags, ...news.tags };
521
522 // Observe.
523 let state = output?.fleetId
524 ? yield* readFleetById(farmId, output.fleetId, arnOf)
525 : yield* findFleetByDisplayName(farmId, displayName, arnOf);
526
527 // Ensure — create if missing, then wait for ACTIVE.
528 if (state === undefined) {
529 const created = yield* retryWhileFarmSettling(
530 retryThroughIamPropagation(
531 deadline.createFleet({
532 farmId,
533 displayName,
534 description: news.description,
535 roleArn: news.roleArn,
536 minWorkerCount: news.minWorkerCount,
537 maxWorkerCount: news.maxWorkerCount,
538 configuration: toWireConfiguration(news.configuration),
539 hostConfiguration: toWireHostConfiguration(
540 news.hostConfiguration,
541 ),
542 tags: desiredTags,
543 }),
544 ),
545 );
546 yield* session.note(
547 `Creating fleet ${displayName} (${created.fleetId})...`,
548 );
549 state = yield* waitForFleetActive(farmId, created.fleetId, arnOf);
550 }
551
552 // Sync mutable settings — only when drifted from OBSERVED state.
553 const described = state.described;
554 const needsUpdate =
555 displayName !== described.displayName ||
556 (news.description !== undefined &&
557 news.description !== (asPlain(described.description) ?? "")) ||
558 news.roleArn !== described.roleArn ||
559 (news.minWorkerCount !== undefined &&
560 news.minWorkerCount !== described.minWorkerCount) ||
561 news.maxWorkerCount !== described.maxWorkerCount ||
562 news.configuration !== undefined;
563 if (needsUpdate) {
564 yield* retryWhileConflict(
565 deadline.updateFleet({
566 farmId,
567 fleetId: state.attrs.fleetId,
568 displayName,
569 description: news.description,
570 roleArn: news.roleArn,
571 minWorkerCount: news.minWorkerCount,
572 maxWorkerCount: news.maxWorkerCount,
573 configuration: toWireConfiguration(news.configuration),
574 hostConfiguration: toWireHostConfiguration(
575 news.hostConfiguration,
576 ),
577 }),
578 );
579 state = yield* waitForFleetActive(
580 farmId,
581 state.attrs.fleetId,
582 arnOf,
583 );
584 yield* session.note(`Updated fleet ${displayName}`);
585 }
586
587 // Sync tags — diff against observed cloud tags.
588 yield* syncDeadlineTags(state.attrs.fleetArn, desiredTags);
589
590 yield* session.note(state.attrs.fleetArn);
591 const final = yield* readFleetById(
592 farmId,
593 state.attrs.fleetId,
594 arnOf,
595 );
596 if (!final) {
597 return yield* Effect.fail(
598 new Error(`failed to read reconciled fleet ${displayName}`),
599 );
600 }
601 return final.attrs;
602 }),
603 delete: Effect.fn(function* ({ output }) {
604 yield* retryWhileConflict(
605 deadline.deleteFleet({
606 farmId: output.farmId,
607 fleetId: output.fleetId,
608 }),
609 ).pipe(
610 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
611 );
612 // Fleet deletion drains workers asynchronously; wait until gone so
613 // the parent farm's deletion does not hit a dependency conflict.
614 yield* waitUntilFleetGone(output.farmId, output.fleetId);
615 }),
616 };
617 }),
618 );