1import { Credentials } from "@distilled.cloud/gcp/Credentials";
2import * as cloudrun from "@distilled.cloud/gcp/run_v2";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { Unowned } from "../../AdoptPolicy.ts";
8import type * as Bundle from "../../Bundle/Bundle.ts";
9import { isResolved } from "../../Diff.ts";
10import { createPhysicalName } from "../../PhysicalName.ts";
11import {
12 Platform,
13 type Main,
14 type PlatformProps,
15 type PlatformServices,
16} from "../../Platform.ts";
17import * as Provider from "../../Provider.ts";
18import { Resource, type ResourceBinding } from "../../Resource.ts";
19import type { RuntimeContext } from "../../RuntimeContext.ts";
20import {
21 createContainerRuntimeContext,
22 type HostRuntimeContext,
23 type ServerHost,
24} from "../../Server/Process.ts";
25import type { Scope } from "effect/Scope";
26import { tagRecord } from "../../Tags.ts";
27import {
28 destroyHostImageRepository,
29 makeImageSource,
30} from "../ArtifactRegistry/ImageSource.ts";
31import { GcpEnvironment } from "../Environment.ts";
32import {
33 type LongRunningOperation,
34 waitForOperation as waitForLongRunningOperation,
35} from "../Operation.ts";
36import {
37 retryActAs,
38 type AppliedIamGrant,
39 type GcpHostBinding,
40} from "../Host.ts";
41import {
42 alchemyRuntimeEnv,
43 isManagedServiceAccount,
44 makeGcpBootstrap,
45 mergeContainerEnv,
46 releaseHostIdentity,
47 resolveHostIdentity,
48} from "../HostRuntime.ts";
49import {
50 createInternalLabels,
51 diffLabels,
52 hasAlchemyLabels,
53 stripInternalLabels,
54 toLabels,
55} from "../Labels.ts";
56import type { Providers } from "../Providers.ts";
57
58const DEFAULT_IMAGE = "us-docker.pkg.dev/cloudrun/container/worker-pool";
59const MAX_NAME_LENGTH = 49;
60
61export type WorkerPoolEnvVar = {
62 /** Environment variable name. */
63 name?: string;
64 /** Literal value. Mutually exclusive with `valueSource`. */
65 value?: string;
66 /** Secret Manager source for the value. */
67 valueSource?: {
68 secretKeyRef?: {
69 secret?: string;
70 version?: string;
71 };
72 };
73};
74
75export type WorkerPoolContainer = {
76 /** DNS_LABEL container name. */
77 name?: string;
78 /**
79 * Container image (Artifact Registry, GCR, or Docker Hub). Required
80 * unless `template` is omitted, in which case the public Cloud Run
81 * worker-pool image is used.
82 */
83 image?: string;
84 /** Entrypoint. */
85 command?: string[];
86 /** Arguments to the entrypoint. */
87 args?: string[];
88 /** Environment variables. */
89 env?: WorkerPoolEnvVar[];
90 /** CPU / memory / GPU requirements. */
91 resources?: {
92 /** Resource limits. Keys: `cpu`, `memory`, `nvidia.com/gpu`. */
93 limits?: Record<string, string>;
94 /** Allocate CPU only during requests. */
95 cpuIdle?: boolean;
96 /** Boost CPU on startup to reduce cold starts. */
97 startupCpuBoost?: boolean;
98 };
99 /** Working directory. */
100 workingDir?: string;
101 /** Volume mounts. Names must match `volumes`. */
102 volumeMounts?: Array<{
103 name: string;
104 mountPath: string;
105 subPath?: string;
106 }>;
107 /** Containers that must start before this one. */
108 dependsOn?: string[];
109};
110
111export type WorkerPoolRevisionTemplate = {
112 /** Containers that make up the revision. */
113 containers?: WorkerPoolContainer[];
114 /** Runtime service account email. Defaults to the project Compute SA. */
115 serviceAccount?: string;
116 /** Revision labels. */
117 labels?: Record<string, string>;
118 /** Revision annotations. */
119 annotations?: Record<string, string>;
120 /** Direct VPC egress / connector. */
121 vpcAccess?: cloudrun.GoogleCloudRunV2VpcAccess;
122 /** Volumes available to containers. */
123 volumes?: cloudrun.GoogleCloudRunV2VolumeList;
124 /** CMEK used to encrypt the container image. */
125 encryptionKey?: string;
126 /** Unique revision name. Generated from the worker pool name if omitted. */
127 revision?: string;
128 /** Node selector (e.g. GPU accelerator). */
129 nodeSelector?: cloudrun.GoogleCloudRunV2NodeSelector;
130 /** True if GPU zonal redundancy is disabled. */
131 gpuZonalRedundancyDisabled?: boolean;
132 /** Service mesh connectivity. */
133 serviceMesh?: cloudrun.GoogleCloudRunV2ServiceMesh;
134};
135
136export type InstanceSplit = {
137 /**
138 * Allocation type (`INSTANCE_SPLIT_ALLOCATION_TYPE_LATEST` or
139 * `INSTANCE_SPLIT_ALLOCATION_TYPE_REVISION`).
140 */
141 type?: string;
142 /** Revision to assign instances to when allocating by revision. */
143 revision?: string;
144 /** Percent of instances (0–100). */
145 percent?: number;
146};
147
148export type WorkerPoolScaling = {
149 /** Total instances in manual scaling mode. */
150 manualInstanceCount?: number;
151};
152
153export type WorkerPoolBinaryAuthorization = {
154 /** Use the project's default Binary Authorization policy. */
155 useDefault?: boolean;
156 /** Breakglass justification. Requires `useDefault`. */
157 breakglassJustification?: string;
158 /** Policy path `projects/{project}/platforms/cloudRun/{policy}`. */
159 policy?: string;
160};
161
162export type WorkerPoolProps = PlatformProps & {
163 /**
164 * Worker pool id (the `{workerPool}` segment of
165 * `projects/{project}/locations/{location}/workerPools/{workerPool}`).
166 * If omitted, a unique name is generated from the stack, stage, and
167 * logical id. Must begin with a letter, not end with a hyphen, and be
168 * fewer than 50 characters. Immutable — changing it replaces the pool.
169 */
170 workerPoolId?: string;
171 /**
172 * Region (`us-central1`, `europe-west1`, …). Immutable — changing it
173 * replaces the pool. `US-CENTRAL1` is accepted and normalized to
174 * `us-central1`.
175 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
176 */
177 location?: string;
178 /**
179 * User labels. Alchemy ownership labels are merged in automatically.
180 */
181 labels?: Record<string, string>;
182 /**
183 * Unstructured annotations. Cloud Run rejects `run.googleapis.com` /
184 * `cloud.googleapis.com` / Knative namespaces.
185 */
186 annotations?: Record<string, string>;
187 /**
188 * Human-readable description (max 512 characters).
189 */
190 description?: string;
191 /**
192 * Launch stage (`GA`, `BETA`, `ALPHA`). Defaults to GA.
193 */
194 launchStage?: string;
195 /**
196 * Binary Authorization settings.
197 */
198 binaryAuthorization?: WorkerPoolBinaryAuthorization;
199 /**
200 * Instance split. Empty / omitted sends 100% to the latest Ready
201 * revision.
202 */
203 instanceSplits?: InstanceSplit[];
204 /**
205 * Worker-pool-level scaling (manual instance count).
206 */
207 scaling?: WorkerPoolScaling;
208 /**
209 * Revision template. If omitted, a single container runs the public
210 * Cloud Run worker-pool image. The Effect-native `main` form fills
211 * `image` for you.
212 */
213 template?: WorkerPoolRevisionTemplate;
214 /**
215 * Module entrypoint for an Effect-native worker pool (typically
216 * `import.meta.url`). Alchemy bundles the program, builds a container,
217 * and deploys the pool. Bindings attach env + IAM onto the runtime SA.
218 */
219 main?: string;
220 /**
221 * Named export to load from `main`.
222 * @default "default"
223 */
224 handler?: string;
225 /**
226 * Additional environment variables for the Effect-native container.
227 */
228 env?: Record<string, any>;
229 /**
230 * Bundler configuration for `main`.
231 */
232 build?: Bundle.BundleConfig;
233};
234
235export type WorkerPool = Resource<
236 "GCP.Run.WorkerPool",
237 WorkerPoolProps,
238 {
239 /** Full resource name `projects/{project}/locations/{location}/workerPools/{workerPool}`. */
240 name: string;
241 /** Worker pool id (last path segment). */
242 workerPoolId: string;
243 /** Project id. */
244 project: string;
245 /** Region (`us-central1`, …). */
246 location: string;
247 /** User labels (Alchemy ownership labels stripped). */
248 labels: Record<string, string>;
249 /** User annotations. */
250 annotations: Record<string, string>;
251 /** Description. */
252 description: string | undefined;
253 /** Server-assigned UUID. */
254 uid: string | undefined;
255 /** Launch stage. */
256 launchStage: string | undefined;
257 /** Manual instance count. */
258 manualInstanceCount: number | undefined;
259 /** Latest created revision name. */
260 latestCreatedRevision: string | undefined;
261 /** Latest Ready revision name. */
262 latestReadyRevision: string | undefined;
263 /** Terminal condition state (`CONDITION_SUCCEEDED`, …). */
264 terminalConditionState: string | undefined;
265 /** True while Cloud Run is still reconciling the desired state. */
266 reconciling: boolean;
267 /** Observed generation. */
268 generation: string | undefined;
269 /** Image of the first container. */
270 image: string | undefined;
271 /** RFC3339 creation timestamp. */
272 createTime: string | undefined;
273 /** RFC3339 last-update timestamp. */
274 updateTime: string | undefined;
275 /** Runtime service account email. */
276 serviceAccount: string | undefined;
277 /** True when Alchemy minted the per-host runtime service account. */
278 managedServiceAccount: boolean;
279 /** IAM roles bindings granted to the runtime service account. */
280 iamGrants: AppliedIamGrant[];
281 /** Hash of the bundled `main` program and its bootstrap (Effect-native only). */
282 codeHash: string | undefined;
283 },
284 GcpHostBinding,
285 Providers
286>;
287
288export type WorkerPoolRuntimeContext = HostRuntimeContext;
289export type WorkerPoolServices = Credentials | GcpEnvironment | ServerHost;
290/**
291 * Effect-native Worker Pool shape: a `run` entry that executes when
292 * the container starts, and/or a `fetch` HTTP handler.
293 */
294export type WorkerPoolShape =
295 | void
296 | (Exclude<Main<WorkerPoolServices>, void> & {
297 run?: Effect.Effect<
298 void,
299 never,
300 WorkerPoolServices | PlatformServices | RuntimeContext | Scope
301 >;
302 });
303
304/**
305 * A Cloud Run worker pool (pull-based revision + instance split).
306 *
307 * Changing `workerPoolId` or `location` replaces the pool. Updates to
308 * `template` create a new revision.
309 *
310 * ### Creating a Worker Pool
311 * **Example:** Generated name, default worker image
312 * ```typescript
313 * const pool = yield* GCP.Run.WorkerPool("workers", {});
314 * ```
315 *
316 * **Example:** Explicit id, image, env, and labels
317 * ```typescript
318 * const pool = yield* GCP.Run.WorkerPool("workers", {
319 * workerPoolId: "order-workers",
320 * location: "us-central1",
321 * description: "order pull workers",
322 * labels: { env: "prod" },
323 * scaling: { manualInstanceCount: 1 },
324 * template: {
325 * containers: [
326 * {
327 * image: "us-docker.pkg.dev/cloudrun/container/worker-pool",
328 * env: [{ name: "ENV", value: "prod" }],
329 * },
330 * ],
331 * },
332 * });
333 * ```
334 *
335 * ### Reading a Worker Pool
336 * **Example:** Get the live worker pool
337 * ```typescript
338 * const getWorkerPool = yield* GCP.Run.GetWorkerPool(pool);
339 * const live = yield* getWorkerPool();
340 * ```
341 *
342 * ### Effect-native Worker Pool with bindings
343 * **Example:** Pull workers with Memorystore
344 * ```typescript
345 * export class Workers extends GCP.Run.WorkerPool<Workers>()(
346 * "Workers",
347 * { main: import.meta.url, scaling: { manualInstanceCount: 1 } },
348 * Effect.gen(function* () {
349 * const redis = yield* GCP.Redis.ReadWriteRedis(cache);
350 * return {
351 * run: redis.set("worker", "up"),
352 * };
353 * }).pipe(Effect.provide(GCP.Redis.ReadWriteRedisHttp)),
354 * ) {}
355 * ```
356 *
357 * @resource
358 * @category Run
359 */
360export const WorkerPool: Platform<
361 WorkerPool,
362 WorkerPoolServices,
363 WorkerPoolShape,
364 WorkerPoolRuntimeContext
365> = Platform("GCP.Run.WorkerPool", {
366 createRuntimeContext: createContainerRuntimeContext("GCP.Run.WorkerPool") as (
367 id: string,
368 ) => WorkerPoolRuntimeContext,
369});
370
371export class WorkerPoolNotResolved extends Data.TaggedError(
372 "GCP.Run.WorkerPoolNotResolved",
373)<{
374 name: string;
375}> {}
376
377export class WorkerPoolNotReady extends Data.TaggedError(
378 "GCP.Run.WorkerPoolNotReady",
379)<{
380 name: string;
381 state: string;
382 message: string;
383}> {}
384
385export class WorkerPoolReconciling extends Data.TaggedError(
386 "GCP.Run.WorkerPoolReconciling",
387)<{
388 name: string;
389 state: string;
390}> {}
391
392export class WorkerPoolStillExists extends Data.TaggedError(
393 "GCP.Run.WorkerPoolStillExists",
394)<{
395 name: string;
396}> {}
397
398const lastSegment = (value: string) => {
399 const trimmed = value.replace(/\/+$/, "");
400 const parts = trimmed.split("/");
401 return parts[parts.length - 1] || trimmed;
402};
403
404const normalizeLocation = (
405 location: string | undefined,
406 defaultLocation: string,
407) => lastSegment(location ?? defaultLocation).toLowerCase();
408
409const resourceName = (
410 project: string,
411 location: string,
412 workerPoolId: string,
413) => `projects/${project}/locations/${location}/workerPools/${workerPoolId}`;
414
415const parseName = (name: string, defaultLocation: string) => {
416 const parts = name.split("/").filter((part) => part.length > 0);
417 const poolsAt = parts.lastIndexOf("workerPools");
418 const locationsAt = parts.lastIndexOf("locations");
419 const projectsAt = parts.lastIndexOf("projects");
420 return {
421 project:
422 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
423 location:
424 locationsAt >= 0 && parts[locationsAt + 1]
425 ? parts[locationsAt + 1]!
426 : defaultLocation,
427 workerPoolId:
428 poolsAt >= 0 && parts[poolsAt + 1]
429 ? parts[poolsAt + 1]!
430 : lastSegment(name),
431 };
432};
433
434const userLabels = (
435 labels: Record<string, string | undefined> | null | undefined,
436): Record<string, string> => stripInternalLabels(tagRecord(labels));
437
438const userAnnotations = (
439 annotations: Record<string, string | undefined> | null | undefined,
440): Record<string, string> => tagRecord(annotations);
441
442const recordsEqual = (
443 left: Record<string, string>,
444 right: Record<string, string>,
445) => {
446 const leftKeys = Object.keys(left).sort();
447 const rightKeys = Object.keys(right).sort();
448 return (
449 leftKeys.length === rightKeys.length &&
450 leftKeys.every(
451 (key, index) => key === rightKeys[index] && left[key] === right[key],
452 )
453 );
454};
455
456const rfc1035 = (name: string): string => {
457 let next = name
458 .toLowerCase()
459 .replace(/[^a-z0-9-]/g, "-")
460 .replace(/-+/g, "-")
461 .replace(/^-+|-+$/g, "");
462 if (!/^[a-z]/.test(next)) next = `w${next}`;
463 next = next.slice(0, MAX_NAME_LENGTH).replace(/-+$/g, "");
464 return next.length > 0 ? next : "workerpool";
465};
466
467const toId = (
468 id: string,
469 workerPoolId: string | undefined,
470 existing?: string,
471) =>
472 Effect.gen(function* () {
473 if (workerPoolId !== undefined) return workerPoolId;
474 if (existing !== undefined) return existing;
475 return rfc1035(
476 yield* createPhysicalName({
477 id,
478 maxLength: MAX_NAME_LENGTH,
479 lowercase: true,
480 }),
481 );
482 });
483
484const desiredTemplate = (
485 news: WorkerPoolProps,
486): cloudrun.GoogleCloudRunV2WorkerPoolRevisionTemplate => {
487 const template = news.template ?? {};
488 return {
489 ...template,
490 containers: template.containers ?? [{ image: DEFAULT_IMAGE }],
491 };
492};
493
494const envFingerprint = (env: cloudrun.GoogleCloudRunV2EnvVarList | undefined) =>
495 JSON.stringify(
496 [...(env ?? [])]
497 .map((item) => ({
498 name: item.name ?? "",
499 value: item.value ?? "",
500 secret: item.valueSource?.secretKeyRef?.secret ?? "",
501 version: item.valueSource?.secretKeyRef?.version ?? "",
502 }))
503 .sort((left, right) => left.name.localeCompare(right.name)),
504 );
505
506const containerNeedsSync = (
507 desired: cloudrun.GoogleCloudRunV2ContainerList | undefined,
508 observed: cloudrun.GoogleCloudRunV2ContainerList | undefined,
509) => {
510 const want = desired ?? [];
511 const have = observed ?? [];
512 if (want.length !== have.length) return true;
513 return want.some((container, index) => {
514 const current = have[index];
515 if (current === undefined) return true;
516 if ((container.image ?? "") !== (current.image ?? "")) return true;
517 if (
518 container.command !== undefined &&
519 JSON.stringify(container.command) !==
520 JSON.stringify(current.command ?? [])
521 ) {
522 return true;
523 }
524 if (
525 container.args !== undefined &&
526 JSON.stringify(container.args) !== JSON.stringify(current.args ?? [])
527 ) {
528 return true;
529 }
530 if (
531 container.workingDir !== undefined &&
532 container.workingDir !== (current.workingDir ?? "")
533 ) {
534 return true;
535 }
536 if (
537 container.env !== undefined &&
538 envFingerprint(container.env) !== envFingerprint(current.env)
539 ) {
540 return true;
541 }
542 if (container.resources?.limits !== undefined) {
543 const wantLimits = tagRecord(container.resources.limits);
544 const haveLimits = tagRecord(current.resources?.limits);
545 for (const [key, value] of Object.entries(wantLimits)) {
546 if (haveLimits[key] !== value) return true;
547 }
548 }
549 if (
550 container.resources?.cpuIdle !== undefined &&
551 container.resources.cpuIdle !== current.resources?.cpuIdle
552 ) {
553 return true;
554 }
555 if (
556 container.resources?.startupCpuBoost !== undefined &&
557 container.resources.startupCpuBoost !== current.resources?.startupCpuBoost
558 ) {
559 return true;
560 }
561 if (
562 container.name !== undefined &&
563 container.name !== (current.name ?? "")
564 ) {
565 return true;
566 }
567 if (
568 container.volumeMounts !== undefined &&
569 JSON.stringify(container.volumeMounts) !==
570 JSON.stringify(current.volumeMounts ?? [])
571 ) {
572 return true;
573 }
574 if (
575 container.dependsOn !== undefined &&
576 JSON.stringify(container.dependsOn) !==
577 JSON.stringify(current.dependsOn ?? [])
578 ) {
579 return true;
580 }
581 return false;
582 });
583};
584
585const scalingFingerprint = (scaling: WorkerPoolScaling | undefined) =>
586 JSON.stringify({
587 manual: scaling?.manualInstanceCount ?? null,
588 });
589
590const instanceSplitFingerprint = (splits: InstanceSplit[] | undefined) =>
591 JSON.stringify(
592 (splits ?? []).map((split) => ({
593 type: split.type ?? "",
594 revision: split.revision ?? "",
595 percent: split.percent ?? 0,
596 })),
597 );
598
599const stable = (value: unknown): string =>
600 JSON.stringify(value, (_key, current) => {
601 if (current && typeof current === "object" && !Array.isArray(current)) {
602 return Object.fromEntries(
603 Object.entries(current as Record<string, unknown>)
604 .filter(([, item]) => item !== undefined)
605 .sort(([a], [b]) => a.localeCompare(b)),
606 );
607 }
608 return current;
609 });
610
611const templateNeedsSync = (
612 desired: cloudrun.GoogleCloudRunV2WorkerPoolRevisionTemplate,
613 observed: cloudrun.GoogleCloudRunV2WorkerPoolRevisionTemplate | undefined,
614) => {
615 const current = observed ?? {};
616 if (containerNeedsSync(desired.containers, current.containers)) {
617 return true;
618 }
619 if (
620 desired.serviceAccount !== undefined &&
621 desired.serviceAccount !== (current.serviceAccount ?? "")
622 ) {
623 return true;
624 }
625 if (
626 desired.encryptionKey !== undefined &&
627 desired.encryptionKey !== (current.encryptionKey ?? "")
628 ) {
629 return true;
630 }
631 if (
632 desired.revision !== undefined &&
633 desired.revision !== (current.revision ?? "")
634 ) {
635 return true;
636 }
637 if (
638 desired.gpuZonalRedundancyDisabled !== undefined &&
639 desired.gpuZonalRedundancyDisabled !==
640 (current.gpuZonalRedundancyDisabled === true)
641 ) {
642 return true;
643 }
644 if (
645 desired.vpcAccess !== undefined &&
646 stable(desired.vpcAccess) !== stable(current.vpcAccess ?? {})
647 ) {
648 return true;
649 }
650 if (
651 desired.volumes !== undefined &&
652 stable(desired.volumes) !== stable(current.volumes ?? [])
653 ) {
654 return true;
655 }
656 if (
657 desired.nodeSelector !== undefined &&
658 stable(desired.nodeSelector) !== stable(current.nodeSelector ?? {})
659 ) {
660 return true;
661 }
662 if (
663 desired.serviceMesh !== undefined &&
664 stable(desired.serviceMesh) !== stable(current.serviceMesh ?? {})
665 ) {
666 return true;
667 }
668 if (
669 desired.labels !== undefined &&
670 !recordsEqual(
671 userAnnotations(desired.labels),
672 userAnnotations(current.labels),
673 )
674 ) {
675 return true;
676 }
677 if (
678 desired.annotations !== undefined &&
679 !recordsEqual(
680 userAnnotations(desired.annotations),
681 userAnnotations(current.annotations),
682 )
683 ) {
684 return true;
685 }
686 return false;
687};
688
689const HOST_TYPE = "GCP.Run.WorkerPool";
690
691// Worker pools have no ingress, so they run the one-shot/loop bootstrap
692// (no HTTP server); a long-running `run` keeps the instance alive.
693const bootstrapFor = (news: WorkerPoolProps) =>
694 makeGcpBootstrap("CloudRunJob", news.handler ?? "default");
695
696const toAttrs = (
697 pool: cloudrun.GoogleCloudRunV2WorkerPool,
698 project: string,
699 region: string,
700 extras: { iamGrants?: AppliedIamGrant[]; codeHash?: string } = {},
701): WorkerPool["Attributes"] => {
702 const name = pool.name ?? "";
703 const parsed = parseName(name, region);
704 return {
705 name,
706 workerPoolId: parsed.workerPoolId,
707 project: parsed.project || project,
708 location: parsed.location,
709 labels: userLabels(pool.labels),
710 annotations: userAnnotations(pool.annotations),
711 description: pool.description,
712 uid: pool.uid,
713 launchStage: pool.launchStage,
714 manualInstanceCount: pool.scaling?.manualInstanceCount,
715 latestCreatedRevision: pool.latestCreatedRevision,
716 latestReadyRevision: pool.latestReadyRevision,
717 terminalConditionState: pool.terminalCondition?.state,
718 reconciling: pool.reconciling === true,
719 generation: pool.generation,
720 image: pool.template?.containers?.[0]?.image,
721 createTime: pool.createTime,
722 updateTime: pool.updateTime,
723 serviceAccount: pool.template?.serviceAccount,
724 managedServiceAccount: isManagedServiceAccount({
725 project: parsed.project || project,
726 hostType: HOST_TYPE,
727 resourceName: name,
728 serviceAccount: pool.template?.serviceAccount,
729 }),
730 iamGrants: extras.iamGrants ?? [],
731 codeHash: extras.codeHash,
732 };
733};
734
735const getByName = (name: string) =>
736 cloudrun
737 .getProjectsLocationsWorkerPools({ name })
738 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
739
740/** Waits on a Cloud Run long-running operation (revisions roll out in minutes). */
741const waitForOperation = (
742 operation: cloudrun.GoogleLongrunningOperation,
743 options?: { notFoundOk?: boolean },
744) =>
745 waitForLongRunningOperation(
746 operation,
747 (name) => {
748 const get = cloudrun.getProjectsLocationsOperations({ name });
749 return options?.notFoundOk === true
750 ? get.pipe(
751 Effect.catchTag("NotFound", () =>
752 Effect.succeed<LongRunningOperation>({ name, done: true }),
753 ),
754 )
755 : get.pipe(
756 // A just-returned operation can briefly 404 on read.
757 Effect.retry({
758 while: (error) => error._tag === "NotFound",
759 times: 5,
760 schedule: Schedule.exponential("250 millis"),
761 }),
762 );
763 },
764 { budget: "10 minutes" },
765 ).pipe(
766 // google.rpc.Code NOT_FOUND: the resource was already gone.
767 Effect.catchTag("GCP.OperationFailed", (error) =>
768 options?.notFoundOk === true && error.code === 5
769 ? Effect.succeed<LongRunningOperation>(operation)
770 : Effect.fail(error),
771 ),
772 );
773
774const isPendingPool = (pool: cloudrun.GoogleCloudRunV2WorkerPool) => {
775 const state = pool.terminalCondition?.state ?? "";
776 return (
777 pool.reconciling === true ||
778 state === "CONDITION_PENDING" ||
779 state === "CONDITION_RECONCILING" ||
780 state === "" ||
781 state === "STATE_UNSPECIFIED"
782 );
783};
784
785const waitUntilReady = (name: string) =>
786 getByName(name).pipe(
787 Effect.filterOrFail(
788 (pool): pool is cloudrun.GoogleCloudRunV2WorkerPool =>
789 pool !== undefined && pool.deleteTime === undefined,
790 () => new WorkerPoolNotResolved({ name }),
791 ),
792 Effect.filterOrFail(
793 (pool) => (pool.terminalCondition?.state ?? "") !== "CONDITION_FAILED",
794 (pool) =>
795 new WorkerPoolNotReady({
796 name,
797 state: pool.terminalCondition?.state ?? "",
798 message: pool.terminalCondition?.message ?? "revision failed",
799 }),
800 ),
801 Effect.filterOrFail(
802 (pool) => !isPendingPool(pool),
803 (pool) =>
804 new WorkerPoolReconciling({
805 name,
806 state: pool.terminalCondition?.state || "reconciling",
807 }),
808 ),
809 Effect.retry({
810 while: (error) =>
811 error._tag === "GCP.Run.WorkerPoolReconciling" ||
812 error._tag === "GCP.Run.WorkerPoolNotResolved",
813 times: 10,
814 schedule: Schedule.spaced("4 seconds"),
815 }),
816 );
817
818const waitUntilGone = (name: string) =>
819 getByName(name).pipe(
820 Effect.flatMap((pool) =>
821 pool === undefined
822 ? Effect.void
823 : Effect.fail(new WorkerPoolStillExists({ name })),
824 ),
825 Effect.retry({
826 while: (error) => error._tag === "GCP.Run.WorkerPoolStillExists",
827 times: 10,
828 schedule: Schedule.spaced("2 seconds"),
829 }),
830 );
831
832const toCreateBody = (
833 news: WorkerPoolProps,
834 labels: Record<string, string>,
835 template: cloudrun.GoogleCloudRunV2WorkerPoolRevisionTemplate,
836): cloudrun.GoogleCloudRunV2WorkerPool => ({
837 labels,
838 annotations: news.annotations,
839 description: news.description,
840 launchStage: news.launchStage,
841 binaryAuthorization: news.binaryAuthorization,
842 instanceSplits: news.instanceSplits,
843 scaling: news.scaling,
844 template,
845});
846
847const listAt = (project: string, location: string, region: string) =>
848 cloudrun.listProjectsLocationsWorkerPools
849 .pages({
850 parent: `projects/${project}/locations/${location}`,
851 pageSize: 1000,
852 })
853 .pipe(
854 Stream.flatMap((page) => Stream.fromIterable(page.workerPools ?? [])),
855 Stream.filter(
856 (pool) =>
857 pool.deleteTime === undefined &&
858 Object.keys(pool.labels ?? {}).some((key) =>
859 key.startsWith("alchemy-"),
860 ),
861 ),
862 Stream.map((pool) => toAttrs(pool, project, region)),
863 Stream.runCollect,
864 Effect.map((chunk) => Array.from(chunk)),
865 );
866
867export const WorkerPoolProvider = () =>
868 Provider.succeed(WorkerPool, {
869 stables: [
870 "name",
871 "workerPoolId",
872 "project",
873 "location",
874 "uid",
875 "createTime",
876 ],
877
878 diff: Effect.fn(function* ({ news, olds, output }) {
879 if (!isResolved(news)) return undefined;
880 const env = yield* GcpEnvironment.current;
881 const previousId = olds?.workerPoolId ?? output?.workerPoolId;
882 const nextId = news.workerPoolId ?? previousId;
883 const previousLocation = normalizeLocation(
884 olds?.location ?? output?.location,
885 env.region,
886 );
887 const nextLocation = normalizeLocation(
888 news.location ?? output?.location,
889 env.region,
890 );
891 const idChanged =
892 previousId !== undefined &&
893 nextId !== undefined &&
894 nextId !== previousId;
895 const locationChanged = previousLocation !== nextLocation;
896 if (idChanged || locationChanged) {
897 return { action: "replace" as const, deleteFirst: false };
898 }
899 // A code-only change leaves the props untouched; hash the bundled
900 // program (and bootstrap) so it still surfaces as an update.
901 if (output !== undefined && news.main !== undefined) {
902 const images = yield* makeImageSource;
903 const hash = yield* images.hash({
904 source: { main: news.main, handler: news.handler, build: news.build },
905 isExternal: news.isExternal,
906 bootstrap: bootstrapFor(news),
907 });
908 if (hash !== undefined && hash !== output.codeHash) {
909 return { action: "update" as const };
910 }
911 }
912 return undefined;
913 }),
914
915 read: Effect.fn(function* ({ id, olds, output }) {
916 const env = yield* GcpEnvironment.current;
917 const workerPoolId = yield* toId(
918 id,
919 olds?.workerPoolId,
920 output?.workerPoolId,
921 );
922 const location = normalizeLocation(
923 olds?.location ?? output?.location,
924 env.region,
925 );
926 const name =
927 output?.name ?? resourceName(env.project, location, workerPoolId);
928 const existing = yield* getByName(name);
929 if (existing === undefined || existing.deleteTime !== undefined) {
930 return undefined;
931 }
932 const attrs = toAttrs(existing, env.project, env.region, {
933 iamGrants: output?.iamGrants,
934 codeHash: output?.codeHash,
935 });
936 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
937 ? attrs
938 : Unowned(attrs);
939 }),
940
941 list: () =>
942 Effect.gen(function* () {
943 const env = yield* GcpEnvironment.current;
944 // WorkerPools list rejects the `-` wildcard; Services/Jobs accept it.
945 return yield* listAt(env.project, "-", env.region).pipe(
946 Effect.catchTag("LocationWildcardUnsupported", () =>
947 // `us-central1` was the default before `GCP.Region`; keep
948 // listing it so older pools are still found.
949 Effect.forEach(
950 [...new Set([env.region, "us-central1"])],
951 (location) => listAt(env.project, location, env.region),
952 ).pipe(Effect.map((groups) => groups.flat())),
953 ),
954 );
955 }),
956
957 reconcile: Effect.fn(function* ({ id, news, output, bindings, session }) {
958 const env = yield* GcpEnvironment.current;
959 const workerPoolId = yield* toId(
960 id,
961 news.workerPoolId,
962 output?.workerPoolId,
963 );
964 const location = normalizeLocation(
965 news.location ?? output?.location,
966 env.region,
967 );
968 const name = resourceName(env.project, location, workerPoolId);
969 const parent = `projects/${env.project}/locations/${location}`;
970 const desiredLabels = {
971 ...toLabels(news.labels),
972 ...(yield* createInternalLabels(id)),
973 };
974 const desiredAnnotations = news.annotations;
975 const template = desiredTemplate(news);
976 const identity = yield* resolveHostIdentity({
977 project: env.project,
978 hostType: HOST_TYPE,
979 resourceName: name,
980 userServiceAccount: template.serviceAccount,
981 effectNative: news.main !== undefined,
982 bindings: bindings as ResourceBinding<GcpHostBinding>[],
983 output,
984 });
985 template.serviceAccount = identity.serviceAccount;
986 let codeHash: string | undefined;
987 if (news.main !== undefined) {
988 const images = yield* makeImageSource;
989 const image = yield* images
990 .resolve({
991 id,
992 source: {
993 main: news.main,
994 handler: news.handler ?? "default",
995 build: news.build,
996 },
997 repositoryName: rfc1035(`${workerPoolId}-src`),
998 location,
999 isExternal: news.isExternal,
1000 bootstrap: bootstrapFor(news),
1001 session,
1002 })
1003 .pipe(Effect.onError(() => identity.cleanup));
1004 codeHash = image.codeHash;
1005 const container = template.containers?.[0] ?? {};
1006 template.containers = [
1007 {
1008 ...container,
1009 image: image.imageUri,
1010 env: mergeContainerEnv(
1011 container.env,
1012 identity.env,
1013 yield* alchemyRuntimeEnv,
1014 news.env,
1015 ),
1016 },
1017 ];
1018 } else {
1019 const runtimeEnv = { ...identity.env, ...news.env };
1020 if (Object.keys(runtimeEnv).length > 0 && template.containers) {
1021 template.containers = template.containers.map((container, index) =>
1022 index === 0
1023 ? {
1024 ...container,
1025 env: mergeContainerEnv(container.env, runtimeEnv),
1026 }
1027 : container,
1028 );
1029 }
1030 }
1031
1032 let current = yield* getByName(name);
1033 if (current?.deleteTime !== undefined) {
1034 yield* waitUntilGone(name);
1035 current = undefined;
1036 }
1037
1038 if (current === undefined) {
1039 const created = yield* cloudrun
1040 .createProjectsLocationsWorkerPools({
1041 parent,
1042 workerPoolId,
1043 body: toCreateBody(news, desiredLabels, template),
1044 })
1045 .pipe(
1046 retryActAs,
1047 Effect.catchTag("Conflict", () => Effect.succeed(undefined)),
1048 Effect.onError(() => identity.cleanup),
1049 );
1050 if (created !== undefined) {
1051 yield* waitForOperation(created);
1052 }
1053 current = yield* waitUntilReady(name);
1054 }
1055
1056 if (current === undefined) {
1057 return yield* new WorkerPoolNotResolved({ name });
1058 }
1059
1060 const observedLabels = tagRecord(current.labels);
1061 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
1062 const labelsChanged = upsert.length > 0 || removed.length > 0;
1063 const descriptionChanged =
1064 (current.description ?? "") !== (news.description ?? "");
1065 const annotationsChanged =
1066 desiredAnnotations !== undefined &&
1067 !recordsEqual(
1068 userAnnotations(current.annotations),
1069 userAnnotations(desiredAnnotations),
1070 );
1071 const launchStageChanged =
1072 news.launchStage !== undefined &&
1073 (current.launchStage ?? "GA") !== news.launchStage;
1074 const binaryAuthorizationChanged =
1075 news.binaryAuthorization !== undefined &&
1076 stable(news.binaryAuthorization) !==
1077 stable(current.binaryAuthorization);
1078 const scalingChanged =
1079 news.scaling !== undefined &&
1080 scalingFingerprint(current.scaling) !==
1081 scalingFingerprint(news.scaling);
1082 const instanceSplitsChanged =
1083 news.instanceSplits !== undefined &&
1084 instanceSplitFingerprint(current.instanceSplits) !==
1085 instanceSplitFingerprint(news.instanceSplits);
1086 const templateChanged =
1087 news.template !== undefined &&
1088 templateNeedsSync(template, current.template);
1089
1090 if (
1091 labelsChanged ||
1092 descriptionChanged ||
1093 annotationsChanged ||
1094 launchStageChanged ||
1095 binaryAuthorizationChanged ||
1096 scalingChanged ||
1097 instanceSplitsChanged ||
1098 templateChanged
1099 ) {
1100 const updateMask = [
1101 labelsChanged ? "labels" : undefined,
1102 descriptionChanged ? "description" : undefined,
1103 annotationsChanged ? "annotations" : undefined,
1104 launchStageChanged ? "launchStage" : undefined,
1105 binaryAuthorizationChanged ? "binaryAuthorization" : undefined,
1106 scalingChanged ? "scaling" : undefined,
1107 instanceSplitsChanged ? "instanceSplits" : undefined,
1108 templateChanged ? "template" : undefined,
1109 ].filter((field): field is string => field !== undefined);
1110
1111 const patched = yield* cloudrun.patchProjectsLocationsWorkerPools({
1112 name,
1113 updateMask: updateMask.join(","),
1114 body: {
1115 name,
1116 labels: desiredLabels,
1117 description: news.description,
1118 annotations: desiredAnnotations,
1119 launchStage: news.launchStage,
1120 binaryAuthorization: news.binaryAuthorization,
1121 scaling: news.scaling,
1122 instanceSplits: news.instanceSplits,
1123 template,
1124 },
1125 });
1126 yield* waitForOperation(patched);
1127 current = yield* waitUntilReady(name);
1128 }
1129
1130 if (current === undefined) {
1131 return yield* new WorkerPoolNotResolved({ name });
1132 }
1133
1134 return toAttrs(current, env.project, env.region, {
1135 iamGrants: identity.grants,
1136 codeHash,
1137 });
1138 }),
1139
1140 delete: Effect.fn(function* ({ id, output }) {
1141 const operation = yield* cloudrun
1142 .deleteProjectsLocationsWorkerPools({ name: output.name })
1143 .pipe(
1144 Effect.retry({
1145 while: (error) => error._tag === "Conflict",
1146 times: 8,
1147 schedule: Schedule.spaced("2 seconds"),
1148 }),
1149 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
1150 );
1151 if (operation !== undefined) {
1152 yield* waitForOperation(operation, { notFoundOk: true });
1153 }
1154 yield* waitUntilGone(output.name);
1155 // Effect-native hosts build into a per-host repository on reconcile.
1156 yield* destroyHostImageRepository(
1157 id,
1158 output,
1159 rfc1035(`${output.workerPoolId}-src`),
1160 );
1161 yield* releaseHostIdentity(output);
1162 }),
1163 });