1import * as machines from "@distilled.cloud/fly-io/machines";
2import * as Retry from "@distilled.cloud/fly-io/Retry";
3import * as Fly from "@/Fly";
4import { makeMachineLeases } from "@/Fly/leases";
5import * as Test from "@/Test/Alchemy";
6import { describe, expect } from "alchemy-test";
7import * as Clock from "effect/Clock";
8import * as Effect from "effect/Effect";
9import * as Fiber from "effect/Fiber";
10import * as Result from "effect/Result";
11import * as HttpClient from "effect/http/HttpClient";
12import * as HttpClientRequest from "effect/http/HttpClientRequest";
13import { assertAppGone } from "./fixtures/bluegreen.ts";
14import { transportProxy, type TransportEvent } from "./fixtures/transport.ts";
15
16const { test } = Test.make({ providers: Fly.providers() });
17
18const targets = Effect.gen(function* () {
19 const app = yield* Fly.App("LeaseSite");
20 return yield* Fly.Machine("LeaseTargets", {
21 app,
22 region: "iad",
23 image: "nginx:alpine",
24 guest: { cpus: 1, memoryMb: 256 },
25 count: 2,
26 skipLaunch: true,
27 });
28});
29
30const sanitizeFailure = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
31 effect.pipe(
32 Effect.mapError(
33 (error) =>
34 new Error(
35 error instanceof Error ? error.name : "Fly lease regression failed",
36 ),
37 ),
38 );
39
40const isLease = (machineId: string) => (event: TransportEvent) =>
41 event.method === "POST" &&
42 event.path.endsWith(`/machines/${machineId}/lease`);
43
44const isAbsent = (event: TransportEvent) =>
45 event.status === 404 || event.state === "destroyed";
46
47// Faults hold or drop real responses; no Fly status, lease, or nonce is fabricated.
48describe.sequential(
49 "native lease controller concurrency",
50 {
51 tags: ["provider:fly", "provider:fly:app", "provider:fly:machine", "live"],
52 },
53 () => {
54 for (const confirmed of [true, false]) {
55 test.provider(
56 confirmed
57 ? "FLY-REVIEW-2 a late absence timeout cannot override confirmed removal"
58 : "FLY-REVIEW-2 unconfirmed removal still cancels shared authority",
59 (stack) =>
60 Effect.gen(function* () {
61 yield* stack.destroy();
62 const created = yield* stack.deploy(targets);
63 const [removedId, siblingId] = [...created.machineIds].sort();
64 expect(created.machineIds).toHaveLength(2);
65 const target = {
66 app_name: created.appName,
67 machine_id: removedId!,
68 };
69 const sibling = {
70 app_name: created.appName,
71 machine_id: siblingId!,
72 };
73 yield* Effect.gen(function* () {
74 const first = yield* transportProxy();
75 const second = yield* transportProxy();
76 const client = yield* HttpClient.HttpClient;
77 let observations = 0;
78 const routed = HttpClient.mapRequest(client, (request) => {
79 if (!request.url.startsWith("https://api.machines.dev/"))
80 return request;
81 const observing =
82 request.method === "GET" &&
83 request.url.endsWith(`/machines/${removedId}`);
84 const proxy = observing && ++observations > 1 ? second : first;
85 return HttpClientRequest.setUrl(
86 request,
87 request.url.replace("https://api.machines.dev", proxy.url),
88 );
89 });
90 yield* Effect.gen(function* () {
91 const leases = yield* makeMachineLeases(created.appName);
92 yield* leases.acquire(created.machineIds);
93 const before = yield* machines
94 .getMachineLease(sibling)
95 .pipe(Retry.none);
96 const deleting = (event: TransportEvent) =>
97 event.method === "DELETE" &&
98 event.path.endsWith(`/machines/${removedId}`);
99 const observing = (event: TransportEvent) =>
100 event.method === "GET" &&
101 event.path.endsWith(`/machines/${removedId}`);
102 yield* Effect.sync(() => {
103 first.arm({
104 match: deleting,
105 action: "drop-response",
106 remaining: 1,
107 });
108 first.arm({
109 match: observing,
110 action: "hold-response",
111 remaining: 1,
112 });
113 second.arm({
114 match: observing,
115 action: "hold-response",
116 remaining: Infinity,
117 });
118 });
119 // Start absence observation just before the target's first renewal.
120 yield* leases.guard(Effect.sleep("20 seconds"));
121 const removal = yield* leases
122 .remove(removedId!, (lease_nonce) =>
123 machines.deleteMachine({
124 ...target,
125 lease_nonce,
126 force: true,
127 }),
128 )
129 .pipe(Effect.result, Effect.forkScoped);
130 const dropped = yield* first.wait(
131 (event) => deleting(event) && event.stage === "dropped",
132 );
133 expect(dropped.status! >= 200 && dropped.status! < 300).toBe(
134 true,
135 );
136 const initialObservation = yield* first.wait(
137 (event) => observing(event) && event.stage === "held",
138 );
139 expect(isAbsent(initialObservation)).toBe(true);
140 const lateObservation = yield* second.wait(
141 (event) => observing(event) && event.stage === "held",
142 );
143 expect(isAbsent(lateObservation)).toBe(true);
144 expect(
145 first.events.some(
146 (event) =>
147 isLease(removedId!)(event) &&
148 event.stage === "forwarded" &&
149 event.status === 404,
150 ),
151 ).toBe(true);
152 if (confirmed) {
153 yield* Effect.sync(first.release);
154 const removed = yield* Fiber.join(removal).pipe(
155 Effect.timeout("5 seconds"),
156 );
157 expect(Result.isSuccess(removed)).toBe(true);
158 expect(yield* leases.nonceIfHeld(removedId!)).toBeUndefined();
159 }
160 // The renewal observer remains held until its own 30-second deadline.
161 const outcome = yield* leases
162 .guard(Effect.sleep("32 seconds"))
163 .pipe(Effect.result);
164 expect(Result.isSuccess(outcome)).toBe(confirmed);
165 if (Result.isFailure(outcome)) {
166 expect(outcome.failure._tag).toBe("Fly.MachineLeaseLost");
167 expect(outcome.failure.machineId).toBe(removedId);
168 expect(outcome.failure.reason).toBe(
169 "lease disappeared before removal was confirmed",
170 );
171 }
172 expect(
173 second.events.some(
174 (event) => observing(event) && event.stage === "forwarded",
175 ),
176 ).toBe(false);
177 if (confirmed) {
178 yield* leases.checkTarget(siblingId!);
179 const after = yield* machines
180 .getMachineLease(sibling)
181 .pipe(Retry.none);
182 expect(after.data?.expires_at).toBeGreaterThan(
183 before.data!.expires_at!,
184 );
185 expect(
186 first.events.some(
187 (event) =>
188 isLease(siblingId!)(event) &&
189 event.stage === "forwarded" &&
190 event.sequence > dropped.sequence &&
191 event.status! >= 200 &&
192 event.status! < 300,
193 ),
194 ).toBe(true);
195 } else {
196 const removed = yield* Fiber.join(removal).pipe(
197 Effect.timeout("5 seconds"),
198 );
199 expect(Result.isFailure(removed)).toBe(true);
200 }
201 expect(
202 first.events.filter(
203 (event) => deleting(event) && event.stage === "request",
204 ),
205 ).toHaveLength(1);
206 }).pipe(
207 Effect.scoped,
208 Effect.provideService(HttpClient.HttpClient, routed),
209 );
210 }).pipe(Effect.scoped);
211 yield* stack.destroy();
212 yield* assertAppGone(created.appName);
213 }).pipe(sanitizeFailure),
214 { timeout: 300_000 },
215 );
216 }
217
218 test.provider(
219 "FLY-REVIEW-4 held renewal responses do not consume another target's renewal interval",
220 (stack) =>
221 Effect.gen(function* () {
222 yield* stack.destroy();
223 const created = yield* stack.deploy(targets);
224 const [a, b] = [...created.machineIds].sort();
225 expect(created.machineIds).toHaveLength(2);
226 yield* Effect.gen(function* () {
227 const first = yield* transportProxy();
228 const second = yield* transportProxy();
229 const client = yield* HttpClient.HttpClient;
230 const routed = HttpClient.mapRequest(client, (request) => {
231 if (!request.url.startsWith("https://api.machines.dev/"))
232 return request;
233 const proxy = request.url.includes(`/machines/${a}/`)
234 ? first
235 : second;
236 return HttpClientRequest.setUrl(
237 request,
238 request.url.replace("https://api.machines.dev", proxy.url),
239 );
240 });
241 yield* Effect.gen(function* () {
242 const leases = yield* makeMachineLeases(created.appName);
243 yield* leases.acquire(created.machineIds);
244 let renewalsA = 0;
245 yield* Effect.sync(() => {
246 first.arm({
247 match: (event) => isLease(a!)(event) && ++renewalsA === 2,
248 action: "hold-response",
249 remaining: 1,
250 });
251 second.arm({
252 match: isLease(b!),
253 action: "hold-response",
254 remaining: 1,
255 });
256 });
257 const heldB = yield* leases.guard(
258 second.wait(
259 (event) => isLease(b!)(event) && event.stage === "held",
260 ),
261 );
262 expect(heldB.status! >= 200 && heldB.status! < 300).toBe(true);
263 const heldAt = yield* Clock.currentTimeMillis;
264 const releaseB = yield* Effect.sleep("60 seconds").pipe(
265 Effect.andThen(Effect.sync(second.release)),
266 Effect.forkScoped,
267 );
268 const heldA = yield* leases
269 .guard(
270 first.wait(
271 (event) => isLease(a!)(event) && event.stage === "held",
272 ),
273 )
274 .pipe(Effect.timeout("45 seconds"));
275 expect(heldA.status! >= 200 && heldA.status! < 300).toBe(true);
276 expect((yield* Clock.currentTimeMillis) - heldAt).toBeLessThan(
277 45_000,
278 );
279 expect(
280 second.events.some(
281 (event) =>
282 event.sequence === heldB.sequence &&
283 event.stage === "forwarded",
284 ),
285 ).toBe(false);
286 const releaseA = yield* Effect.sleep("60 seconds").pipe(
287 Effect.andThen(Effect.sync(first.release)),
288 Effect.forkScoped,
289 );
290 yield* leases.guard(
291 Effect.all([Fiber.join(releaseA), Fiber.join(releaseB)]),
292 );
293 yield* leases.guard(Effect.sleep("5 seconds"));
294 for (const machineId of created.machineIds)
295 yield* leases.checkTarget(machineId);
296 for (const [proxy, held, machineId] of [
297 [first, heldA, a!],
298 [second, heldB, b!],
299 ] as const) {
300 const forwarded = proxy.events.findIndex(
301 (event) =>
302 event.sequence === held.sequence &&
303 event.stage === "forwarded",
304 );
305 const started = proxy.events.findIndex(
306 (event) =>
307 event.sequence === held.sequence && event.stage === "held",
308 );
309 expect(forwarded).toBeGreaterThan(started);
310 expect(
311 proxy.events
312 .slice(started + 1, forwarded)
313 .some(
314 (event) =>
315 isLease(machineId)(event) && event.stage === "request",
316 ),
317 ).toBe(false);
318 expect(
319 proxy.events
320 .slice(forwarded + 1)
321 .some(
322 (event) =>
323 isLease(machineId)(event) &&
324 event.stage === "forwarded" &&
325 event.status! >= 200 &&
326 event.status! < 300,
327 ),
328 ).toBe(true);
329 }
330 }).pipe(
331 Effect.scoped,
332 Effect.provideService(HttpClient.HttpClient, routed),
333 );
334 }).pipe(Effect.scoped);
335 yield* stack.destroy();
336 yield* assertAppGone(created.appName);
337 }).pipe(sanitizeFailure),
338 { timeout: 300_000 },
339 );
340 },
341);