1import * as Cloudflare from "@/Cloudflare";
2import * as Test from "@/Test/Alchemy";
3import { expect } from "alchemy-test";
4import * as Console from "effect/Console";
5import * as Effect from "effect/Effect";
6import { MinimumLogLevel } from "effect/References";
7import * as Schedule from "effect/Schedule";
8import * as Stream from "effect/Stream";
9import * as RpcClient from "effect/rpc/RpcClient";
10import { WorkerRpcs } from "./fixtures/rpc-http/group.ts";
11import Stack from "./fixtures/rpc-http/stack.ts";
12
13const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
14 providers: Cloudflare.providers(),
15});
16
17// `Test.rpcClientLayer` guards the transport against edge-generated HTML
18// bodies (the workers.dev placeholder — which serves with HTTP 200 —
19// 1101/1102 error pages, 429/1015 rate limits): the effect RPC HTTP protocol
20// never inspects status or content-type, so those would otherwise surface as
21// an opaque `RpcClientDefect: Error decoding HTTP response`. Non-ndjson
22// responses fail typed (status + body snippet) and are retried at the
23// transport level, so every RPC call in this file rides out edge-propagation
24// windows with one shared budget.
25const clientLayer = Test.rpcClientLayer;
26
27const logLevel = Effect.provideService(
28 MinimumLogLevel,
29 process.env.DEBUG ? "Debug" : "Info",
30);
31
32// Cap exponential backoff at 3s so readiness retries poll densely instead of
33// sleeping tens of seconds past the propagation window (an uncapped
34// exponential blows through the 30s test timeouts after ~6 attempts).
35const readinessSchedule = Schedule.min([
36 Schedule.exponential("500 millis"),
37 Schedule.spaced("3 seconds"),
38]);
39
40// The worker fixture wraps the DO calls in `Effect.orDie` / `Stream.orDie`,
41// so a transient `Worker not found.` (the worker→DO namespace binding hasn't
42// propagated to every Cloudflare edge yet) can arrive at the client as a
43// DEFECT — and `Effect.retry` does not retry defects. Promote defects to
44// failures so the readiness retry can absorb the propagation window (a
45// genuine bug simply keeps failing until the retry budget is exhausted).
46const retryReadyN =
47 (times: number) =>
48 <A, E, R>(eff: Effect.Effect<A, E, R>) =>
49 eff.pipe(
50 Effect.catchDefect((defect) => Effect.fail(defect)),
51 Effect.retry({ schedule: readinessSchedule, times }),
52 );
53
54const stack = beforeAll(
55 deploy(Stack).pipe(
56 // Ping the Worker to ensure it's ready.
57 // Subsequent calls should succeed without retries.
58 Effect.tap(({ url }) =>
59 Effect.gen(function* () {
60 const client = yield* RpcClient.make(WorkerRpcs);
61 const result = yield* client
62 .Ping({ message: "warmup" })
63 .pipe(Effect.tapError(Console.log), retryReadyN(20));
64 expect(result.echo).toBe("warmup");
65 expect(result.n).toBeGreaterThan(0);
66 }).pipe(Effect.scoped, Effect.provide(clientLayer(url))),
67 ),
68 // Gate on the worker→DO pathway too: under full-suite parallel load the
69 // DO namespace binding propagates noticeably slower than the worker
70 // itself, and the `*DO` tests below would otherwise race that window.
71 // While the binding is still propagating, the worker's internal DO client
72 // fails with an `HttpError` that the fixture's `Effect.orDie` turns into a
73 // server-sent Defect — so this gate has to both promote defects AND carry
74 // enough budget (~2 min of capped backoff) to outlast the slow windows a
75 // full-suite run produces; a 15-attempt/~40s budget was observed to
76 // exhaust and fail the whole file.
77 Effect.tap(({ url }) =>
78 Effect.gen(function* () {
79 const client = yield* RpcClient.make(WorkerRpcs);
80 yield* client.PingDO({ message: "warmup" }).pipe(retryReadyN(40));
81 yield* client
82 .CountDO({ upto: 1 })
83 .pipe(Stream.runCollect, retryReadyN(40));
84 }).pipe(Effect.scoped, Effect.provide(clientLayer(url))),
85 ),
86 // Let edge propagation settle before the (mostly un-retried) bodies run.
87 Effect.tap(() => Effect.sleep("5 seconds")),
88 ),
89 { timeout: 300_000 },
90);
91afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack));
92
93// The Cloudflare Worker fetch adapter (`workersHttpHandler`) currently
94// short-circuits Effect's standard HTTP lifecycle (it manually
95// provides `HttpServerRequest` and converts the response to a web
96// `Response` outside of `HttpEffect.toHandled`). PR #328 reported that
97// this can deadlock `RpcServer.toHttpEffect` under workerd. This test
98// hammers a real deployed Worker exposing an Effect RPC group to
99// surface lifecycle / per-request scope regressions.
100//
101// The `*DO` variants exercise the DO fetch pathway
102// (`DurableObjectBridge.fetch` -> `makeRequestEffect`) via an
103// `RpcClient` constructed inside the Worker handler whose transport
104// is `Cloudflare.toHttpClient(rpcDO.getByName(...))`. This mirrors the
105// HttpApi fixture's `getTaskDO` pattern.
106test(
107 "RpcServer.toHttpEffect: unary RPC response",
108 Effect.gen(function* () {
109 const { url } = yield* stack;
110 yield* Effect.log("url:", url);
111
112 yield* Effect.gen(function* () {
113 const client = yield* RpcClient.make(WorkerRpcs);
114 // Even after the warmup gate, a fresh request can land on a PoP that
115 // hasn't resolved the script yet and get Cloudflare's HTML error page,
116 // which is not valid ndjson (`RpcClientDefect`). `Ping` is idempotent,
117 // so retry through a bounded schedule.
118 const result = yield* client.Ping({ message: "hello" }).pipe(
119 Effect.tapError(Console.log),
120 Effect.retry({
121 schedule: Schedule.min([
122 Schedule.exponential("500 millis"),
123 Schedule.spaced("2 seconds"),
124 ]),
125 times: 10,
126 }),
127 );
128 expect(result.echo).toBe("hello");
129 expect(result.n).toBeGreaterThan(0);
130 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
131 }).pipe(logLevel),
132 {
133 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
134 timeout: 30_000,
135 },
136);
137
138test(
139 "RpcServer.toHttpEffect: streaming RPC response",
140 Effect.gen(function* () {
141 const { url } = yield* stack;
142
143 yield* Effect.gen(function* () {
144 const client = yield* RpcClient.make(WorkerRpcs);
145 // First streaming call can race edge propagation and hit a Cloudflare
146 // HTML error page; retry the whole collect through a bounded schedule.
147 const values = yield* client.Count({ upto: 5 }).pipe(
148 Stream.runCollect,
149 Effect.retry({
150 schedule: Schedule.exponential("500 millis"),
151 times: 10,
152 }),
153 );
154 expect(values).toEqual([1, 2, 3, 4, 5]);
155 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
156 }).pipe(logLevel),
157 {
158 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
159 timeout: 30_000,
160 },
161);
162
163test(
164 "RpcServer.toHttpEffect: array payload streams response items in order",
165 Effect.gen(function* () {
166 const { url } = yield* stack;
167
168 yield* Effect.gen(function* () {
169 const client = yield* RpcClient.make(WorkerRpcs);
170 const messages = ["a", "b", "c", "d"];
171 // First streaming call can race edge propagation and hit a Cloudflare
172 // HTML error page; retry the whole collect through a bounded schedule.
173 const values = yield* client.Echo({ messages }).pipe(
174 Stream.runCollect,
175 Effect.retry({
176 schedule: Schedule.exponential("500 millis"),
177 times: 10,
178 }),
179 );
180 expect(values).toEqual(
181 messages.map((message, index) => ({ index, message })),
182 );
183 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
184 }).pipe(logLevel),
185 {
186 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
187 timeout: 30_000,
188 },
189);
190
191test(
192 "RpcServer.toHttpEffect: 200 concurrent unary calls do not hang",
193 Effect.gen(function* () {
194 const { url } = yield* stack;
195
196 yield* Effect.gen(function* () {
197 const client = yield* RpcClient.make(WorkerRpcs);
198
199 const N = 200;
200 const results = yield* Effect.forEach(
201 Array.from({ length: N }, (_, i) => i),
202 (i) =>
203 client.Ping({ message: `m-${i}` }).pipe(
204 Effect.timeout("5 seconds"),
205 // 64-way fan-out opens many fresh connections, each of which can
206 // land on an edge host still serving an HTML page after the
207 // single-connection warmup succeeded. The transport guard retries
208 // those; this outer budget (matching the siblings' capped
209 // generosity) backstops timeouts and longer bursts — the previous
210 // uncapped `times: 3` (~3.5s window) was the flake.
211 Effect.retry({
212 schedule: Schedule.min([
213 Schedule.exponential("500 millis"),
214 Schedule.spaced("2 seconds"),
215 ]),
216 times: 10,
217 }),
218 ),
219 { concurrency: 64 },
220 );
221
222 expect(results).toHaveLength(N);
223 for (let i = 0; i < N; i++) {
224 expect(results[i].echo).toBe(`m-${i}`);
225 }
226 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
227 }).pipe(logLevel),
228 {
229 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
230 timeout: 60_000,
231 },
232);
233
234test(
235 "RpcServer.toHttpEffect: concurrent streaming calls do not hang",
236 Effect.gen(function* () {
237 const { url } = yield* stack;
238
239 yield* Effect.gen(function* () {
240 const client = yield* RpcClient.make(WorkerRpcs);
241
242 const N = 64;
243 const results = yield* Effect.forEach(
244 Array.from({ length: N }, (_, i) => i),
245 (i) =>
246 client.Count({ upto: 3 + (i % 3) }).pipe(
247 Stream.runCollect,
248 Effect.timeout("5 seconds"),
249 // 64 concurrent first-streams all race cold-start: a PoP that
250 // hasn't resolved the script yet returns Cloudflare's HTML error
251 // page, which is not valid ndjson and surfaces as an
252 // `RpcClientDefect` ("Error decoding HTTP response"). Match the
253 // single-stream tests' generosity (capped backoff, ~10 attempts)
254 // so the whole fan-out rides out propagation.
255 Effect.retry({
256 schedule: Schedule.min([
257 Schedule.exponential("500 millis"),
258 Schedule.spaced("2 seconds"),
259 ]),
260 times: 10,
261 }),
262 ),
263 { concurrency: N },
264 );
265
266 expect(results).toHaveLength(N);
267 for (let i = 0; i < N; i++) {
268 expect(results[i]).toEqual(
269 Array.from({ length: 3 + (i % 3) }, (_, n) => n + 1),
270 );
271 }
272 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
273 }).pipe(logLevel),
274 {
275 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
276 timeout: 60_000,
277 },
278);
279
280// === Durable Object pathway ===
281// These exercise the Worker's `*DO` handlers, which proxy through an
282// `RpcClient` whose transport is `Cloudflare.toHttpClient(rpcDO.getByName(...))`.
283
284test(
285 "RpcServer.toHttpEffect Durable Object unary RPC response",
286 Effect.gen(function* () {
287 const { url } = yield* stack;
288
289 yield* Effect.gen(function* () {
290 const client = yield* RpcClient.make(WorkerRpcs);
291 // First DO call can race edge propagation and hit a Cloudflare HTML
292 // error page (or a `Worker not found.` defect); retry through a
293 // bounded, defect-promoting schedule.
294 const result = yield* client
295 .PingDO({ message: "hello-do" })
296 .pipe(Effect.tapError(Console.log), retryReadyN(10));
297 expect(result.echo).toBe("hello-do");
298 expect(result.n).toBeGreaterThan(0);
299 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
300 }).pipe(logLevel),
301 {
302 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
303 timeout: 30_000,
304 },
305);
306
307test(
308 "RpcServer.toHttpEffect Durable Object streaming RPC response",
309 Effect.gen(function* () {
310 const { url } = yield* stack;
311
312 yield* Effect.gen(function* () {
313 const client = yield* RpcClient.make(WorkerRpcs);
314 // First DO streaming call can race edge propagation and hit a Cloudflare
315 // HTML error page (or a `Worker not found.` defect); retry the whole
316 // collect through a bounded, defect-promoting schedule.
317 const values = yield* client
318 .CountDO({ upto: 5 })
319 .pipe(Stream.runCollect, retryReadyN(10));
320 expect(values).toEqual([1, 2, 3, 4, 5]);
321 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
322 }).pipe(logLevel),
323 {
324 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
325 timeout: 30_000,
326 },
327);
328
329test(
330 "RpcServer.toHttpEffect Durable Object array payload streams response items in order",
331 Effect.gen(function* () {
332 const { url } = yield* stack;
333
334 yield* Effect.gen(function* () {
335 const client = yield* RpcClient.make(WorkerRpcs);
336 const messages = ["a", "b", "c", "d"];
337 // First streaming call can race edge propagation and hit a Cloudflare
338 // HTML error page (or a `Worker not found.` defect); retry the whole
339 // collect through a bounded, defect-promoting schedule.
340 const values = yield* client
341 .EchoDO({ messages })
342 .pipe(Stream.runCollect, retryReadyN(10));
343 expect(values).toEqual(
344 messages.map((message, index) => ({ index, message })),
345 );
346 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
347 }).pipe(logLevel),
348 {
349 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
350 timeout: 30_000,
351 },
352);
353
354test(
355 "RpcServer.toHttpEffect Durable Object concurrent unary calls do not hang",
356 Effect.gen(function* () {
357 const { url } = yield* stack;
358
359 yield* Effect.gen(function* () {
360 const client = yield* RpcClient.make(WorkerRpcs);
361
362 const N = 64;
363 const results = yield* Effect.forEach(
364 Array.from({ length: N }, (_, i) => i),
365 (i) =>
366 client
367 .PingDO({ message: `m-${i}` })
368 .pipe(Effect.timeout("10 seconds"), retryReadyN(5)),
369 { concurrency: 16 },
370 );
371
372 expect(results).toHaveLength(N);
373 for (let i = 0; i < N; i++) {
374 expect(results[i].echo).toBe(`m-${i}`);
375 }
376 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
377 }).pipe(logLevel),
378 {
379 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
380 timeout: 60_000,
381 },
382);
383
384test(
385 "RpcServer.toHttpEffect Durable Object concurrent streaming calls do not hang",
386 Effect.gen(function* () {
387 const { url } = yield* stack;
388
389 yield* Effect.gen(function* () {
390 const client = yield* RpcClient.make(WorkerRpcs);
391
392 const N = 32;
393 const results = yield* Effect.forEach(
394 Array.from({ length: N }, (_, i) => i),
395 (i) =>
396 client
397 .CountDO({ upto: 3 + (i % 3) })
398 .pipe(
399 Stream.runCollect,
400 Effect.timeout("10 seconds"),
401 retryReadyN(5),
402 ),
403 { concurrency: N },
404 );
405
406 expect(results).toHaveLength(N);
407 for (let i = 0; i < N; i++) {
408 expect(results[i]).toEqual(
409 Array.from({ length: 3 + (i % 3) }, (_, n) => n + 1),
410 );
411 }
412 }).pipe(Effect.scoped, Effect.provide(clientLayer(url)));
413 }).pipe(logLevel),
414 {
415 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
416 timeout: 30_000,
417 },
418);