1import {
2 BadGateway,
3 GatewayTimeout,
4 InternalServerError,
5 ServiceUnavailable,
6 TooManyRequests,
7} from "@distilled.cloud/fly-io/Errors";
8import * as machines from "@distilled.cloud/fly-io/machines";
9import * as Retry from "@distilled.cloud/fly-io/Retry";
10import * as Clock from "effect/Clock";
11import * as Data from "effect/Data";
12import * as Deferred from "effect/Deferred";
13import * as Duration from "effect/Duration";
14import * as Effect from "effect/Effect";
15import * as Fiber from "effect/Fiber";
16import * as Option from "effect/Option";
17import * as Schedule from "effect/Schedule";
18import * as FetchHttpClient from "effect/http/FetchHttpClient";
19import * as HttpClient from "effect/http/HttpClient";
20import * as HttpClientRequest from "effect/http/HttpClientRequest";
21
22const TTL_SECONDS = 120;
23const EXPIRY_MARGIN_MS = 15_000;
24
25const retrySchedule = <E>(): Schedule.Schedule<Duration.Duration, E> =>
26 Schedule.exponential("500 millis").pipe(
27 Schedule.modifyDelay<Duration.Duration, E>(({ input, duration }) =>
28 Effect.succeed(
29 (input instanceof TooManyRequests ||
30 input instanceof InternalServerError ||
31 input instanceof BadGateway ||
32 input instanceof ServiceUnavailable ||
33 input instanceof GatewayTimeout) &&
34 input.retryAfter
35 ? Math.min(Duration.toMillis(input.retryAfter), 60_000)
36 : Math.min(Duration.toMillis(duration), 5_000),
37 ),
38 ),
39 Schedule.addDelay<Duration.Duration, E>(() =>
40 Effect.sync(() => Duration.millis(Math.random() * 50)),
41 ),
42 );
43
44export class MachineLeaseBusy extends Data.TaggedError("Fly.MachineLeaseBusy")<{
45 appName: string;
46 machineId: string;
47}> {}
48
49export class MachineLeaseLost extends Data.TaggedError("Fly.MachineLeaseLost")<{
50 appName: string;
51 machineId: string;
52 reason: string;
53}> {
54 get message() {
55 return `Machine lease authority lost for ${this.appName}/${this.machineId}: ${this.reason}`;
56 }
57}
58
59export class MachineMutationUncertain extends Data.TaggedError(
60 "Fly.MachineMutationUncertain",
61)<{
62 appName: string;
63 machineId: string;
64}> {}
65
66interface HeldLease {
67 nonce: string;
68 deadline: number;
69 renewAt: number;
70}
71
72export const makeMachineLeases = Effect.fn(function* (appName: string) {
73 const scope = yield* Effect.scope;
74 const held = new Map<string, HeldLease>();
75 const deleting = new Set<string>();
76 const renewals = new Set<Fiber.Fiber<void, never>>();
77 const uncertain = new Set<string>();
78 const lost = yield* Deferred.make<never, MachineLeaseLost>();
79 let failure: MachineLeaseLost | undefined;
80 const lose = (machineId: string, reason: string) =>
81 Effect.gen(function* () {
82 failure ??= new MachineLeaseLost({ appName, machineId, reason });
83 yield* Deferred.fail(lost, failure);
84 return yield* Effect.fail(failure);
85 });
86 const check = Effect.gen(function* () {
87 if (failure) return yield* Effect.fail(failure);
88 const now = yield* Clock.currentTimeMillis;
89 for (const [machineId, lease] of held) {
90 if (now >= lease.deadline)
91 return yield* lose(machineId, "lease authority expired");
92 }
93 });
94 const guard = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
95 check.pipe(Effect.andThen(Effect.raceFirst(effect, Deferred.await(lost))));
96 const record = (
97 machineId: string,
98 response: machines.MachineLease,
99 requestedAt: number,
100 ) =>
101 Effect.gen(function* () {
102 const lease = response.data;
103 const now = yield* Clock.currentTimeMillis;
104 const deadline =
105 Math.min(
106 (lease?.expires_at ?? 0) * 1000,
107 requestedAt + TTL_SECONDS * 1000,
108 ) - EXPIRY_MARGIN_MS;
109 const captured = lease?.nonce
110 ? { nonce: lease.nonce, deadline, renewAt: requestedAt + 25_000 }
111 : undefined;
112 if (captured) held.set(machineId, captured);
113 if (!captured || !Number.isFinite(deadline) || deadline <= now) {
114 return yield* lose(machineId, "invalid or expired lease response");
115 }
116 return captured;
117 });
118 const acquire = (machineIds: readonly string[]) =>
119 Effect.gen(function* () {
120 for (const machineId of [...new Set(machineIds)].sort()) {
121 yield* check;
122 if (held.has(machineId)) continue;
123 // Save a returned nonce atomically; uncertain interrupted requests expire at Fly.
124 yield* Effect.uninterruptibleMask((restore) =>
125 Effect.gen(function* () {
126 const requestedAt = yield* Clock.currentTimeMillis;
127 const response = yield* restore(
128 machines
129 .createMachineLease({
130 app_name: appName,
131 machine_id: machineId,
132 ttl: TTL_SECONDS,
133 description: "Alchemy Machine lifecycle",
134 })
135 .pipe(
136 Retry.none,
137 Effect.retry({
138 times: 8,
139 schedule: retrySchedule<machines.CreateMachineLeaseError>(),
140 while: (error) =>
141 error._tag === "Conflict" ||
142 error._tag === "TooManyRequests",
143 }),
144 Effect.timeout("30 seconds"),
145 Effect.catchTag("Conflict", () =>
146 Effect.fail(new MachineLeaseBusy({ appName, machineId })),
147 ),
148 Effect.catchTag(
149 ["HttpClientError", "GatewayTimeout", "TimeoutError"],
150 (error) =>
151 lose(machineId, `acquisition uncertain: ${error._tag}`),
152 ),
153 ),
154 );
155 const lease = yield* record(machineId, response, requestedAt);
156 const renewal = yield* heartbeat(machineId, lease).pipe(
157 Effect.interruptible,
158 Effect.forkIn(scope),
159 );
160 renewals.add(renewal);
161 }),
162 );
163 yield* check;
164 }
165 });
166 const nonce = (machineId: string) =>
167 Effect.gen(function* () {
168 yield* check;
169 const lease = held.get(machineId);
170 if (!lease)
171 return yield* lose(
172 machineId,
173 "target is outside the acquired lease set",
174 );
175 return lease.nonce;
176 });
177 const mutate = <A, E extends { readonly _tag: string }, R>(
178 machineId: string,
179 operation: (nonce: string) => Effect.Effect<A, E, R>,
180 options: { idempotent?: boolean; timeoutMs?: number } = {},
181 ) =>
182 guard(
183 Effect.gen(function* () {
184 const value = yield* nonce(machineId);
185 return yield* operation(value).pipe(Retry.none);
186 }).pipe(
187 Effect.retry({
188 times: 8,
189 schedule: retrySchedule<E | MachineLeaseLost>(),
190 // Only explicit throttling is safe to replay for update/stop/delete.
191 while: (error) =>
192 error._tag === "TooManyRequests" ||
193 (options.idempotent === true &&
194 (error._tag === "InternalServerError" ||
195 error._tag === "BadGateway" ||
196 error._tag === "ServiceUnavailable" ||
197 error._tag === "GatewayTimeout")),
198 }),
199 Effect.timeout(options.timeoutMs ?? 120_000),
200 Effect.catch((error) =>
201 Effect.gen(function* () {
202 if (error._tag === "Conflict" || error._tag === "Forbidden") {
203 return yield* lose(
204 machineId,
205 `leased mutation rejected: ${error._tag}`,
206 );
207 }
208 // Transport errors retain the request, including its private nonce header.
209 if (
210 error._tag === "HttpClientError" ||
211 error._tag === "GatewayTimeout" ||
212 error._tag === "TimeoutError"
213 ) {
214 return yield* new MachineMutationUncertain({
215 appName,
216 machineId,
217 });
218 }
219 return yield* Effect.fail(error);
220 }),
221 ),
222 ),
223 );
224 const forget = (machineId: string) =>
225 Effect.sync(() => {
226 held.delete(machineId);
227 });
228 const confirmRemoved = (machineId: string) =>
229 guard(
230 machines
231 .getMachine({
232 app_name: appName,
233 machine_id: machineId,
234 })
235 .pipe(
236 Retry.none,
237 Effect.map(
238 (machine) =>
239 machine.id === machineId && machine.state === "destroyed",
240 ),
241 Effect.catchTag("NotFound", () => Effect.succeed(true)),
242 Effect.catchTag(
243 [
244 "TooManyRequests",
245 "InternalServerError",
246 "BadGateway",
247 "ServiceUnavailable",
248 "GatewayTimeout",
249 ],
250 () => Effect.succeed(false),
251 ),
252 Effect.catchTag("HttpClientError", (error) =>
253 error.reason._tag === "TransportError"
254 ? Effect.succeed(false)
255 : Effect.fail(error),
256 ),
257 Effect.repeat({
258 times: 8,
259 schedule: Schedule.spaced("2 seconds"),
260 until: (absent) => absent,
261 }),
262 Effect.timeout("30 seconds"),
263 Effect.catchTag("TimeoutError", () => Effect.succeed(false)),
264 Effect.catch((error) =>
265 Effect.gen(function* () {
266 if (!held.has(machineId)) return true;
267 if (error._tag === "Forbidden") {
268 return yield* lose(machineId, "removal observation forbidden");
269 }
270 return yield* Effect.fail(error);
271 }),
272 ),
273 // Another observer's confirmed removal wins over a late uncertain response.
274 Effect.map((absent) => absent || !held.has(machineId)),
275 Effect.tap((absent) => (absent ? forget(machineId) : Effect.void)),
276 ),
277 );
278 const remove = <A, E extends { readonly _tag: string }, R>(
279 machineId: string,
280 operation: (nonce: string) => Effect.Effect<A, E, R>,
281 ) =>
282 Effect.gen(function* () {
283 const http = yield* HttpClient.HttpClient;
284 const fetchOptions = yield* Effect.serviceOption(
285 FetchHttpClient.RequestInit,
286 );
287 yield* Effect.sync(() => {
288 deleting.add(machineId);
289 });
290 yield* mutate(machineId, (nonce) =>
291 operation(nonce).pipe(
292 // Bun can replay DELETE on a reused socket below the SDK retry policy.
293 Effect.provideService(
294 HttpClient.HttpClient,
295 HttpClient.mapRequest(
296 http,
297 HttpClientRequest.setHeader("connection", "close"),
298 ),
299 ),
300 Effect.provideService(FetchHttpClient.RequestInit, {
301 ...Option.getOrUndefined(fetchOptions),
302 keepalive: false,
303 redirect: "error",
304 }),
305 ),
306 ).pipe(
307 Effect.asVoid,
308 Effect.catch((error) =>
309 Effect.gen(function* () {
310 if (error._tag === "NotFound") return yield* forget(machineId);
311 if (
312 error._tag === "Fly.MachineMutationUncertain" ||
313 error._tag === "InternalServerError" ||
314 error._tag === "BadGateway" ||
315 error._tag === "ServiceUnavailable"
316 ) {
317 // Keep renewal aware of the removal until its exact target has been observed.
318 if (yield* confirmRemoved(machineId)) return;
319 }
320 return yield* Effect.fail(error);
321 }),
322 ),
323 );
324 yield* forget(machineId);
325 }).pipe(
326 Effect.ensuring(
327 Effect.sync(() => {
328 deleting.delete(machineId);
329 }),
330 ),
331 );
332 const nonceIfHeld = (machineId: string) =>
333 check.pipe(Effect.map(() => held.get(machineId)?.nonce));
334
335 yield* Effect.addFinalizer(() =>
336 Effect.gen(function* () {
337 // A late refresh can recreate a lease after a successful release.
338 yield* Fiber.interruptAll(renewals);
339 yield* Effect.forEach(
340 [...held.entries()]
341 .filter(([machineId]) => !uncertain.has(machineId))
342 .sort(([a], [b]) => a.localeCompare(b)),
343 ([machineId, lease]) =>
344 machines
345 .machinesReleaseLease({
346 app_name: appName,
347 machine_id: machineId,
348 lease_nonce: lease.nonce,
349 })
350 .pipe(
351 Retry.none,
352 Effect.timeout("5 seconds"),
353 Effect.interruptible,
354 Effect.catchTag("NotFound", () => Effect.void),
355 Effect.catch((error) =>
356 Effect.logWarning(
357 "Fly lease release was not confirmed; expiry remains the safety bound",
358 { appName, machineId, error: error._tag },
359 ),
360 ),
361 Effect.asVoid,
362 ),
363 { concurrency: 4 },
364 );
365 if (uncertain.size > 0)
366 yield* Effect.logWarning(
367 "Fly lease renewal was not confirmed; expiry remains the safety bound",
368 { appName, machineIds: [...uncertain] },
369 );
370 }),
371 );
372
373 const refresh = (machineId: string, lease: HeldLease) =>
374 Effect.gen(function* () {
375 yield* check;
376 if (held.get(machineId) !== lease) return;
377 const requestedAt = yield* Clock.currentTimeMillis;
378 const response = yield* machines
379 .createMachineLease({
380 app_name: appName,
381 machine_id: machineId,
382 ttl: TTL_SECONDS,
383 lease_nonce: lease.nonce,
384 description: "Alchemy Machine lifecycle",
385 })
386 .pipe(Retry.none);
387 if (held.get(machineId) !== lease) return;
388 if (response.data?.nonce !== lease.nonce) {
389 return yield* lose(machineId, "refresh returned a different nonce");
390 }
391 return yield* record(machineId, response, requestedAt);
392 }).pipe(
393 Effect.retry({
394 times: 8,
395 schedule: retrySchedule<
396 machines.CreateMachineLeaseError | MachineLeaseLost
397 >(),
398 while: (error) => error._tag === "TooManyRequests",
399 }),
400 Effect.timeout("75 seconds"),
401 Effect.catch((error) =>
402 Effect.gen(function* () {
403 if (held.get(machineId) !== lease) return;
404 uncertain.add(machineId);
405 // A missing lease during removal is not proof that the Machine is gone.
406 if (error._tag === "NotFound" && deleting.has(machineId)) {
407 const absent = yield* confirmRemoved(machineId).pipe(
408 Effect.catch((observation) =>
409 held.get(machineId) === lease
410 ? lose(
411 machineId,
412 `removal observation uncertain: ${observation._tag}`,
413 )
414 : Effect.succeed(true),
415 ),
416 );
417 if (absent || held.get(machineId) !== lease) return;
418 return yield* lose(
419 machineId,
420 "lease disappeared before removal was confirmed",
421 );
422 }
423 return yield* lose(machineId, `refresh uncertain: ${error._tag}`);
424 }),
425 ),
426 (operation) =>
427 Effect.uninterruptible(
428 Effect.gen(function* () {
429 const pending = yield* operation.pipe(
430 Effect.interruptible,
431 Effect.forkChild,
432 );
433 return yield* Fiber.join(pending);
434 }),
435 ),
436 );
437 // Each target renews serially on its own cadence within the operation scope.
438 const heartbeat = (machineId: string, acquired: HeldLease) =>
439 Effect.gen(function* () {
440 let lease: HeldLease | undefined = acquired;
441 while (lease && held.get(machineId) === lease) {
442 const now = yield* Clock.currentTimeMillis;
443 yield* Effect.sleep(Math.max(0, lease.renewAt - now));
444 lease = yield* refresh(machineId, lease);
445 }
446 }).pipe(Effect.catch(() => Effect.void));
447 yield* Effect.sleep("1 second").pipe(
448 Effect.andThen(check),
449 Effect.forever,
450 Effect.catch(() => Effect.void),
451 Effect.forkScoped,
452 );
453 const checkTarget = (machineId: string) =>
454 nonce(machineId).pipe(Effect.asVoid);
455 return {
456 acquire,
457 check,
458 checkTarget,
459 guard,
460 mutate,
461 remove,
462 forget,
463 nonceIfHeld,
464 };
465});
466
467export type MachineLeases = Effect.Success<
468 ReturnType<typeof makeMachineLeases>
469>;
470
471export const usingMachineLeases = <A, E, R>(
472 appName: string,
473 existing: MachineLeases | undefined,
474 use: (leases: MachineLeases) => Effect.Effect<A, E, R>,
475) =>
476 Effect.gen(function* () {
477 if (existing)
478 return yield* existing.guard(Effect.suspend(() => use(existing)));
479 return yield* Effect.scoped(
480 Effect.gen(function* () {
481 const leases = yield* makeMachineLeases(appName);
482 return yield* leases.guard(Effect.suspend(() => use(leases)));
483 }),
484 );
485 });