1import * as Fly from "@/Fly";
2import { HttpServer, NodeHttpServer } from "@/Http";
3import { ServerHost } from "@/Server/Process";
4import * as Redis from "@/Redis";
5import * as Config from "effect/Config";
6import * as ConfigProvider from "effect/ConfigProvider";
7import * as Deferred from "effect/Deferred";
8import * as Effect from "effect/Effect";
9import * as Fiber from "effect/Fiber";
10import * as Redacted from "effect/Redacted";
11import * as Scope from "effect/Scope";
12import * as Stream from "effect/Stream";
13import { HttpServerRequest } from "effect/http/HttpServerRequest";
14import * as HttpServerResponse from "effect/http/HttpServerResponse";
15import {
16 Cache,
17 WorkerSite,
18 scripts,
19 services,
20 type Job,
21} from "./bluegreen-worker-shared.ts";
22
23export class Worker extends Fly.Service<Worker>()("Worker") {}
24export interface WorkerOptions {
25 version: string;
26 timeout: "10 seconds" | "30 seconds" | "60 seconds";
27 signal: "SIGTERM" | "SIGINT";
28 runOnly?: boolean;
29 mode?: string;
30 afterSignalMs?: number;
31 workers?: number;
32}
33export const workerLayer = (options: WorkerOptions) =>
34 Worker.make(
35 {
36 app: WorkerSite,
37 main: import.meta.url,
38 region: "iad",
39 port: 3000,
40 guest: { cpuKind: "shared", cpus: 1, memoryMb: 256 },
41 env: {
42 VERSION: options.version,
43 MODE: options.mode ?? "drain",
44 AFTER_SIGNAL_MS: String(options.afterSignalMs ?? 1000),
45 WORKERS: String(options.workers ?? 1),
46 RUN_ONLY: options.runOnly ? "1" : "0",
47 },
48 deploy: { strategy: "bluegreen", healthTimeout: "45 seconds" },
49 shutdown: { signal: options.signal, timeout: options.timeout },
50 services: options.runOnly ? [] : services,
51 checks: options.runOnly
52 ? {
53 ready: {
54 type: "http",
55 port: 3000,
56 path: "/health",
57 interval: "2s",
58 timeout: "1s",
59 },
60 }
61 : undefined,
62 },
63 Effect.gen(function* () {
64 yield* Fly.ReadWriteRedis(Cache);
65 const cache = yield* Cache;
66 const boundUrl = yield* cache.url;
67 const stopping = yield* Deferred.make<void>();
68 const { count, runOnly } = globalThis.__ALCHEMY_RUNTIME__
69 ? yield* Effect.all({
70 count: Config.Number("WORKERS"),
71 runOnly: Config.String("RUN_ONLY").pipe(
72 Effect.map((value) => value === "1"),
73 ),
74 }).pipe(
75 // These controls are plain Machine env, not packed runtime bindings.
76 Effect.provideService(
77 ConfigProvider.ConfigProvider,
78 ConfigProvider.fromEnv(),
79 ),
80 Effect.orDie,
81 )
82 : { count: options.workers ?? 1, runOnly: options.runOnly ?? false };
83 const ready = new Set<string>();
84 let sharedOpen = true;
85 const url = Effect.gen(function* () {
86 const value = yield* boundUrl;
87 if (!value) return yield* Effect.die(new Error("Redis URL missing"));
88 return Redacted.value(value);
89 });
90 const identify = Effect.gen(function* () {
91 return {
92 machine: yield* Config.String("FLY_MACHINE_ID"),
93 version: yield* Config.String("VERSION"),
94 };
95 }).pipe(Effect.orDie);
96 const connect = url.pipe(
97 Effect.flatMap(Redis.connect),
98 Effect.timeout("5 seconds"),
99 Effect.orDie,
100 );
101 const event = (
102 client: Redis.Connection,
103 name: string,
104 fields: Record<string, unknown> = {},
105 ) =>
106 identify.pipe(
107 Effect.flatMap((identity) =>
108 client.send("EVAL", [
109 scripts.event,
110 0,
111 JSON.stringify({ ...identity, event: name, ...fields }),
112 ]),
113 ),
114 Effect.timeout("5 seconds"),
115 Effect.orDie,
116 );
117 if (globalThis.__ALCHEMY_RUNTIME__) {
118 yield* Effect.addFinalizer(() =>
119 Effect.gen(function* () {
120 const client = yield* connect;
121 yield* event(client, "shared-closed");
122 sharedOpen = false;
123 }).pipe(Effect.scoped),
124 );
125 }
126 const runWorker = (name: string) =>
127 Effect.gen(function* () {
128 const client = yield* connect;
129 const identity = yield* identify;
130 const mode =
131 name === "a"
132 ? yield* Config.String("MODE").pipe(Effect.orDie)
133 : "drain";
134 const delay = yield* Config.Number("AFTER_SIGNAL_MS").pipe(
135 Effect.orDie,
136 );
137 const call = (operation: keyof typeof scripts, args: string[]) =>
138 client
139 .send("EVAL", [scripts[operation], 0, ...args])
140 .pipe(Effect.timeout("5 seconds"), Effect.orDie);
141 yield* Effect.addFinalizer(() =>
142 event(client, "client-released", { worker: name }),
143 );
144 const workScope = yield* Scope.make();
145 yield* Effect.addFinalizer((exit) =>
146 Scope.close(workScope, exit).pipe(
147 Effect.andThen(event(client, "work-closed", { worker: name })),
148 ),
149 );
150 const pollScope = yield* Scope.make();
151 yield* Effect.addFinalizer((exit) => Scope.close(pollScope, exit));
152 const jobs: Fiber.Fiber<void, never>[] = [];
153 let accepting = true;
154 const poll = Effect.gen(function* () {
155 if (!accepting) return;
156 if (name === "a")
157 yield* call("tick", [identity.machine, identity.version]);
158 const response = yield* call("claim", [
159 identity.machine,
160 identity.version,
161 ]);
162 if (typeof response !== "string") return;
163 const job = JSON.parse(response) as Job;
164 const task = Effect.gen(function* () {
165 let finished = false;
166 yield* Effect.addFinalizer(() =>
167 finished
168 ? Effect.void
169 : call("checkpoint", [
170 job.id,
171 job.job,
172 identity.machine,
173 ]).pipe(Effect.asVoid),
174 );
175 if (!job.checkpoint && job.kind !== "quick") {
176 yield* Deferred.await(stopping);
177 if (job.kind === "checkpoint" || mode === "job-hang")
178 yield* Effect.never;
179 yield* Effect.sleep(delay);
180 } else yield* Effect.sleep("100 millis");
181 if (!sharedOpen)
182 return yield* Effect.die(
183 new Error("shared dependency closed before job"),
184 );
185 yield* call("finish", [job.id, job.job, identity.machine]).pipe(
186 Effect.tap(() =>
187 Effect.sync(() => {
188 finished = true;
189 }),
190 ),
191 Effect.uninterruptible,
192 );
193 }).pipe(
194 Effect.scoped,
195 Effect.interruptible,
196 Effect.forkIn(workScope),
197 );
198 jobs.push(yield* task);
199 }).pipe(Effect.uninterruptible);
200 const poller = yield* poll.pipe(
201 Effect.andThen(Effect.sleep("250 millis")),
202 Effect.forever,
203 Effect.forkIn(pollScope),
204 );
205 yield* Effect.addFinalizer(() =>
206 Effect.gen(function* () {
207 yield* event(client, "stop-started", { worker: name });
208 yield* Deferred.succeed(stopping, undefined);
209 if (mode === "stop-hang") yield* Effect.never;
210 if (mode === "stop-delay") yield* Effect.sleep("1 second");
211 accepting = false;
212 yield* Fiber.interrupt(poller);
213 if (mode === "stop-fail") {
214 yield* event(client, "stop-failed", { worker: name });
215 yield* Scope.close(workScope, yield* Effect.exit(Effect.void));
216 return yield* Effect.die(new Error("application stop failure"));
217 }
218 yield* event(client, "stopped", { worker: name });
219 if (mode === "checkpoint")
220 yield* Scope.close(workScope, yield* Effect.exit(Effect.void));
221 yield* Effect.forEach(jobs, Fiber.await, {
222 concurrency: "unbounded",
223 });
224 yield* event(client, "drained", { worker: name });
225 }),
226 );
227 yield* event(client, "worker-ready", { worker: name });
228 ready.add(name);
229 yield* Effect.never;
230 }).pipe(Effect.scoped);
231 const fetch = Effect.gen(function* () {
232 const request = yield* HttpServerRequest;
233 if (request.url === "/health") {
234 return HttpServerResponse.text(
235 ready.size === count ? "ready" : "starting",
236 {
237 status: ready.size === count ? 200 : 503,
238 },
239 );
240 }
241 const client = yield* connect;
242 const identity = yield* identify;
243 if (request.url === "/")
244 return yield* HttpServerResponse.json(identity);
245 yield* event(client, "request-started");
246 yield* Effect.addFinalizer(() => event(client, "request-finalized"));
247 const delay = yield* Config.Number("AFTER_SIGNAL_MS").pipe(
248 Effect.orDie,
249 );
250 const afterStop = Deferred.await(stopping).pipe(
251 Effect.andThen(Effect.sleep(delay)),
252 Effect.andThen(
253 Effect.sync(() => {
254 if (!sharedOpen)
255 throw new Error("shared dependency closed before response");
256 }),
257 ),
258 Effect.andThen(event(client, "response-finished")),
259 );
260 if (request.url === "/stream")
261 return HttpServerResponse.stream(
262 Stream.make("first\n".repeat(32768)).pipe(
263 Stream.concat(
264 Stream.fromEffect(
265 afterStop.pipe(Effect.as("last\n".repeat(32768))),
266 ),
267 ),
268 Stream.mapEffect((chunk) =>
269 Effect.sync(() => new TextEncoder().encode(chunk)),
270 ),
271 ),
272 );
273 yield* afterStop;
274 return yield* HttpServerResponse.json(identity);
275 });
276 const host = yield* ServerHost;
277 if (count > 1) yield* host.run(runWorker("b"));
278 if (runOnly) {
279 yield* host.run(
280 Effect.gen(function* () {
281 const server = yield* HttpServer;
282 yield* server.serve(fetch);
283 yield* Effect.never;
284 }).pipe(
285 Effect.provide(NodeHttpServer({ hostname: "::" })),
286 Effect.scoped,
287 ),
288 );
289 }
290 const run = runWorker("a");
291 return runOnly ? { run } : { run, fetch };
292 }).pipe(Effect.provide(Fly.ReadWriteRedisHttp)),
293 );
294
295export default workerLayer({
296 version: "default",
297 timeout: "30 seconds",
298 signal: "SIGTERM",
299});