1import * as Clock from "effect/Clock";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import * as HttpClient from "effect/http/HttpClient";
6import * as ChildProcess from "effect/process/ChildProcess";
7import { expect } from "alchemy-test";
8
9export const functionRolloutTimeout = 900_000;
10
11/** Poll immediately, then require eight consecutive current-version samples. */
12export const functionRolloutSamples = <A, E, R>(
13 sample: Effect.Effect<A, E, R>,
14 isCurrent: (value: A) => boolean,
15) =>
16 Effect.gen(function* () {
17 const started = yield* Clock.currentTimeMillis;
18 const samples: A[] = [];
19 let consecutive = 0;
20 let firstCurrentMs: number | undefined;
21 let lastStaleMs: number | undefined;
22 const poll = Effect.gen(function* () {
23 const value = yield* sample.pipe(Effect.timeout("25 seconds"));
24 const elapsedMs = (yield* Clock.currentTimeMillis) - started;
25 const current = isCurrent(value);
26 samples.push(value);
27 consecutive = current ? consecutive + 1 : 0;
28 if (current) firstCurrentMs ??= elapsedMs;
29 else lastStaleMs = elapsedMs;
30 yield* Effect.logInfo(
31 JSON.stringify({
32 neonRolloutSample: {
33 round: samples.length,
34 elapsedMs,
35 current,
36 consecutive,
37 },
38 }),
39 );
40 return consecutive >= 8;
41 });
42 const converged = yield* poll.pipe(
43 Effect.repeat({
44 schedule: Schedule.spaced("5 seconds"),
45 times: 119,
46 until: (done) => done,
47 }),
48 Effect.timeout("10 minutes"),
49 );
50 yield* Effect.logInfo(
51 JSON.stringify({
52 neonRollout: {
53 converged,
54 observations: samples.length,
55 firstCurrentMs,
56 lastStaleMs,
57 confirmedAtMs: (yield* Clock.currentTimeMillis) - started,
58 },
59 }),
60 );
61 expect(converged).toBe(true);
62 return samples.slice(-8);
63 });
64
65/** Compare the shared HTTP client with a fresh, proxy-bypassing connection. */
66export const functionTextSamples = (
67 url: string,
68 isCurrent: (body: string) => boolean,
69) =>
70 functionRolloutSamples(
71 Effect.gen(function* () {
72 const response = yield* HttpClient.get(url, {
73 headers: { "cache-control": "no-cache" },
74 });
75 expect(response.status).toBe(200);
76 const text = yield* response.text;
77 const child = yield* ChildProcess.make("curl", [
78 "--silent",
79 "--show-error",
80 "--fail",
81 "--max-time",
82 "10",
83 "--http1.1",
84 "--noproxy",
85 "*",
86 "-H",
87 "Connection: close",
88 "-H",
89 "Cache-Control: no-cache",
90 url,
91 ]);
92 const [exit, fresh] = yield* Effect.all(
93 [child.exitCode, child.stdout.pipe(Stream.decodeText, Stream.mkString)],
94 { concurrency: "unbounded" },
95 );
96 expect(Number(exit)).toBe(0);
97 return [text, fresh] as const;
98 }),
99 (samples) => samples.every(isCurrent),
100 ).pipe(Effect.map((samples) => samples.flat()));