1import * as Cloudflare from "@/Cloudflare";
2import * as Test from "@/Test/Alchemy";
3import { expect } from "alchemy-test";
4import * as Effect from "effect/Effect";
5import { MinimumLogLevel } from "effect/References";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import * as HttpClient from "effect/http/HttpClient";
9import * as RpcClient from "effect/rpc/RpcClient";
10import Stack from "./fixtures/rpc-do-namespace-do-rpc/stack.ts";
11import { WorkerRpcs as RpcWorkerWorkerRpcs } from "./fixtures/rpc-worker-rpc-http/group.ts";
12import RpcWorkerStack from "./fixtures/rpc-worker-rpc-http/stack.ts";
13
14const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
15 providers: Cloudflare.providers(),
16});
17
18// `Test.rpcClientLayer` guards the transport against edge-generated HTML
19// bodies (workers.dev placeholder, error pages) that the RPC protocol would
20// otherwise surface as an opaque `RpcClientDefect`; see Test/Http.ts.
21const rpcClientLayer = Test.rpcClientLayer;
22
23const logLevel = Effect.provideService(
24 MinimumLogLevel,
25 process.env.DEBUG ? "Debug" : "Info",
26);
27
28// Cap exponential backoff at 3s so retries stay bounded when CF edge is
29// slow (otherwise the geometric blow-up dominates wall time).
30const readinessSchedule = Schedule.min([
31 Schedule.exponential("500 millis"),
32 Schedule.spaced("3 seconds"),
33]);
34
35// Suffix DO instance ids with a per-process random tag so reruns under
36// `NO_DESTROY=1` don't collide with persisted state from earlier runs
37// (the DO's `count` lives in `state.storage`).
38const runId = Math.random().toString(36).slice(2, 10);
39const k = (name: string) => `${name}-${runId}`;
40
41// Retry transient 5xx/network errors from a freshly-deployed worker. The
42// worker→DO binding can take a while to propagate to every edge POP, during
43// which requests fail with 500 before the DO method ever runs — so retrying
44// non-idempotent calls (increment) through this window is safe (the
45// 100-concurrent test below relies on the same property).
46const retryHttp = <A, E, R>(eff: Effect.Effect<A, E, R>) =>
47 eff.pipe(Effect.retry({ schedule: readinessSchedule, times: 10 }));
48
49const resetCounter = (url: string, id: string) =>
50 Effect.gen(function* () {
51 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
52 yield* client.post(`${url}/counter/${id}/reset`).pipe(retryHttp);
53 });
54
55// Full-suite concurrency + workers.dev propagation routinely needs more than
56// a dozen attempts before the DO binding / subdomain is routable.
57const readinessRetries = 30;
58
59// The `*DO` RPC handlers forward to the Durable Object via `getByName(...)`.
60// On a freshly-deployed worker the DO-namespace binding hasn't propagated to
61// every Cloudflare edge yet, so the first calls fail with `Worker not found.`.
62// The worker fixture wraps the DO call in `Effect.orDie` / `Stream.orDie`, so
63// that error arrives at the client as a DEFECT — and `Effect.retry` does not
64// retry defects. Promote defects to failures so the readiness retry can absorb
65// the transient binding-propagation error (a genuine bug would simply keep
66// failing until the retry budget is exhausted).
67const retryReadyN =
68 (times: number) =>
69 <A, E, R>(eff: Effect.Effect<A, E, R>) =>
70 eff.pipe(
71 Effect.catchDefect((defect) => Effect.fail(defect)),
72 Effect.retry({ schedule: readinessSchedule, times }),
73 );
74
75const retryReady = retryReadyN(readinessRetries);
76
77// Gate HTTP deploy on workers.dev + DO binding readiness. Without this the
78// first test burns its 30s budget on 404s while the subdomain propagates.
79const stack = beforeAll(
80 deploy(Stack).pipe(
81 Effect.tap((outputs) =>
82 Effect.gen(function* () {
83 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
84 yield* client
85 .post(`${outputs.url}/counter/${k("warmup")}/reset`)
86 .pipe(retryReadyN(readinessRetries));
87 }),
88 ),
89 ),
90);
91afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack));
92
93// Gate the deploy on the worker→DO binding having propagated to the edge:
94// hit both the unary and streaming `*DO` paths once, retrying through the
95// transient `Worker not found.` window, so individual tests can call the
96// `*DO` RPCs directly without each having to re-implement the readiness retry.
97const rpcWorkerStack = beforeAll(
98 deploy(RpcWorkerStack).pipe(
99 Effect.tap((outputs) =>
100 Effect.gen(function* () {
101 const c = yield* RpcClient.make(RpcWorkerWorkerRpcs);
102 yield* c.PingDO({ message: "warmup" }).pipe(retryReady);
103 yield* c.CountDO({ upto: 1 }).pipe(Stream.runCollect, retryReady);
104 }).pipe(Effect.scoped, Effect.provide(rpcClientLayer(outputs.url))),
105 ),
106 // just give it some extra time to propagate
107 Effect.tap(Effect.sleep("5 seconds")),
108 ),
109);
110afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(RpcWorkerStack));
111
112test(
113 "RpcDurableObject: Increment / Get round-trip via Worker",
114 Effect.gen(function* () {
115 const { url } = yield* stack;
116 const alpha = k("alpha");
117 yield* resetCounter(url, alpha);
118 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
119
120 const incRes = yield* client
121 .post(`${url}/counter/${alpha}/increment`)
122 .pipe(retryHttp);
123 expect(incRes.status).toBe(200);
124 const inc = (yield* incRes.json) as { count: number };
125 expect(inc.count).toBe(1);
126
127 yield* client.post(`${url}/counter/${alpha}/increment`).pipe(retryHttp);
128 yield* client.post(`${url}/counter/${alpha}/increment`).pipe(retryHttp);
129
130 const getRes = yield* client.get(`${url}/counter/${alpha}`).pipe(retryHttp);
131 expect(getRes.status).toBe(200);
132 const got = (yield* getRes.json) as { count: number };
133 expect(got.count).toBe(3);
134 }).pipe(logLevel),
135 {
136 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
137 timeout: 60_000,
138 },
139);
140
141test(
142 "RpcDurableObject: separate getByName(id) instances are isolated",
143 Effect.gen(function* () {
144 const { url } = yield* stack;
145 const betaId = k("beta");
146 const gammaId = k("gamma");
147 yield* resetCounter(url, betaId);
148 yield* resetCounter(url, gammaId);
149 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
150
151 yield* client.post(`${url}/counter/${betaId}/increment`).pipe(retryHttp);
152
153 const beta = (yield* (yield* client
154 .get(`${url}/counter/${betaId}`)
155 .pipe(retryHttp)).json) as {
156 count: number;
157 };
158 const gamma = (yield* (yield* client
159 .get(`${url}/counter/${gammaId}`)
160 .pipe(retryHttp)).json) as {
161 count: number;
162 };
163 expect(beta.count).toBe(1);
164 expect(gamma.count).toBe(0);
165 }).pipe(logLevel),
166 {
167 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
168 timeout: 30_000,
169 },
170);
171
172test(
173 "RpcDurableObject: streaming RPC via getByName(id).CountUpTo",
174 Effect.gen(function* () {
175 const { url } = yield* stack;
176 const delta = k("delta");
177 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
178
179 // The stream route commits a 200 before the DO stream produces data, so
180 // a transient `Worker not found.` during binding propagation dies AFTER
181 // the headers are sent and surfaces as a truncated/empty 200 body —
182 // invisible to status-based retries. `CountUpTo` is a pure stream, so
183 // retry on content until all four lines arrive.
184 const lines = yield* client
185 .get(`${url}/counter/${delta}/stream?upto=4`)
186 .pipe(
187 Effect.flatMap((res) => res.text),
188 Effect.flatMap((body) => {
189 const lines = body.split("\n").filter((l) => l.length > 0);
190 return lines.length === 4
191 ? Effect.succeed(lines)
192 : Effect.fail(
193 new Error(`truncated stream body: ${JSON.stringify(lines)}`),
194 );
195 }),
196 retryHttp,
197 );
198 expect(lines).toEqual(["1", "2", "3", "4"]);
199 }).pipe(logLevel),
200 {
201 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
202 timeout: 30_000,
203 },
204);
205
206test(
207 "RpcWorker + RpcDurableObject: Worker proxies *DO RPCs through the typed namespace",
208 Effect.gen(function* () {
209 const { url } = yield* rpcWorkerStack;
210
211 yield* Effect.gen(function* () {
212 const c = yield* RpcClient.make(RpcWorkerWorkerRpcs);
213 const ping = yield* c.Ping({ message: "hi" }).pipe(retryReady);
214 expect(ping.echo).toBe("hi");
215
216 const pingDO = yield* c.PingDO({ message: "via DO" }).pipe(retryReady);
217 expect(pingDO.echo).toBe("via DO");
218 }).pipe(Effect.scoped, Effect.provide(rpcClientLayer(url)));
219 }).pipe(logLevel),
220 {
221 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
222 timeout: 30_000,
223 },
224);
225
226test(
227 "RpcDurableObject: 100 concurrent Increment calls do not hang",
228 Effect.gen(function* () {
229 const { url } = yield* stack;
230 const concurrent = k("concurrent");
231 yield* resetCounter(url, concurrent);
232 const client = HttpClient.filterStatusOk(yield* HttpClient.HttpClient);
233
234 yield* client
235 .post(`${url}/counter/${concurrent}/increment`)
236 .pipe(retryHttp);
237
238 const N = 100;
239 const results = yield* Effect.forEach(
240 Array.from({ length: N }, (_, i) => i),
241 () =>
242 client.post(`${url}/counter/${concurrent}/increment`).pipe(
243 Effect.flatMap((res) => res.json),
244 Effect.timeout("10 seconds"),
245 // Under full-suite load the account churns through worker
246 // deploys/deletes and workers.dev routing intermittently 404s an
247 // existing worker for several seconds; `times: 3` (~3.5s) is not
248 // enough to ride out a blip when 100 requests each get 4 chances.
249 // Retry only HTTP failures (a 404/5xx never reached the DO, so the
250 // increment can't double-count); a timeout is ambiguous and stays
251 // un-retried so the final-count assertion holds.
252 Effect.retry({
253 while: (e) => e._tag === "HttpClientError",
254 schedule: readinessSchedule,
255 times: 10,
256 }),
257 ),
258 { concurrency: 32 },
259 );
260
261 expect(results).toHaveLength(N);
262 const finalRes = yield* client
263 .get(`${url}/counter/${concurrent}`)
264 .pipe(retryHttp);
265 const final = (yield* finalRes.json) as { count: number };
266 expect(final.count).toBe(N + 1);
267 }).pipe(logLevel),
268 {
269 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
270 timeout: 30_000,
271 },
272);
273
274test(
275 "RpcWorker + RpcDurableObject: 100 concurrent unary RPCs do not hang",
276 Effect.gen(function* () {
277 const { url } = yield* rpcWorkerStack;
278
279 yield* Effect.gen(function* () {
280 const c = yield* RpcClient.make(RpcWorkerWorkerRpcs);
281
282 const N = 100;
283 const results = yield* Effect.forEach(
284 Array.from({ length: N }, (_, i) => i),
285 (i) =>
286 c.Ping({ message: `m-${i}` }).pipe(
287 Effect.timeout("10 seconds"),
288 // A cold PoP mid-propagation returns Cloudflare's HTML error
289 // page, which is not valid ndjson and surfaces as a retryable
290 // `RpcClientError` (`RpcClientDefect: Error decoding HTTP
291 // response`); `times: 3` (~6s) is not enough to ride out a
292 // multi-second blip across 100 requests. Ping is idempotent.
293 retryReadyN(8),
294 ),
295 { concurrency: 32 },
296 );
297
298 expect(results).toHaveLength(N);
299 for (let i = 0; i < N; i++) {
300 expect(results[i].echo).toBe(`m-${i}`);
301 }
302 }).pipe(Effect.scoped, Effect.provide(rpcClientLayer(url)));
303 }).pipe(logLevel),
304 {
305 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
306 timeout: 60_000,
307 },
308);
309
310test(
311 "RpcWorker + RpcDurableObject: 100 concurrent *DO unary RPCs do not hang",
312 Effect.gen(function* () {
313 const { url } = yield* rpcWorkerStack;
314
315 yield* Effect.gen(function* () {
316 const c = yield* RpcClient.make(RpcWorkerWorkerRpcs);
317
318 const N = 100;
319 const results = yield* Effect.forEach(
320 Array.from({ length: N }, (_, i) => i),
321 (i) =>
322 c
323 .PingDO({ message: `m-${i}` })
324 .pipe(Effect.timeout("10 seconds"), retryReadyN(5)),
325 { concurrency: 16 },
326 );
327
328 expect(results).toHaveLength(N);
329 for (let i = 0; i < N; i++) {
330 expect(results[i].echo).toBe(`m-${i}`);
331 }
332 }).pipe(Effect.scoped, Effect.provide(rpcClientLayer(url)));
333 }).pipe(logLevel),
334 {
335 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
336 timeout: 30_000,
337 },
338);
339
340test(
341 "RpcWorker + RpcDurableObject: 100 concurrent streaming *DO RPCs do not hang",
342 Effect.gen(function* () {
343 const { url } = yield* rpcWorkerStack;
344
345 yield* Effect.gen(function* () {
346 const c = yield* RpcClient.make(RpcWorkerWorkerRpcs);
347
348 const N = 100;
349 const results = yield* Effect.forEach(
350 Array.from({ length: N }, (_, i) => i),
351 (i) =>
352 c
353 .CountDO({ upto: 3 + (i % 3) })
354 .pipe(
355 Stream.runCollect,
356 Effect.timeout("10 seconds"),
357 retryReadyN(5),
358 ),
359 { concurrency: 16 },
360 );
361
362 expect(results).toHaveLength(N);
363 for (let i = 0; i < N; i++) {
364 expect(results[i]).toEqual(
365 Array.from({ length: 3 + (i % 3) }, (_, n) => n + 1),
366 );
367 }
368 }).pipe(Effect.scoped, Effect.provide(rpcClientLayer(url)));
369 }).pipe(logLevel),
370 {
371 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
372 timeout: 30_000,
373 },
374);