Open History to see the last commit
1import * as Cloudflare from "@alchemy.run/cloudflare";
2import { CloudflareEnvironment } from "@alchemy.run/cloudflare/CloudflareEnvironment";
3import { expect } from "@alchemy.run/test";
4import * as Test from "@alchemy.run/test/Alchemy";
5import * as queues from "@distilled.cloud/cloudflare/queues";
6import * as Alchemy from "alchemy";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as HttpClient from "effect/http/HttpClient";
10import { MinimumLogLevel } from "effect/References";
11import * as Schedule from "effect/Schedule";
12import * as pathe from "pathe";
13
14// `dev: true` runs local providers behind the RPC sidecar proxy by default,
15// matching the process topology of the real `alchemy dev` command (see
16// MakeOptions.sidecar in Test/Core.ts). For Queue and Consumer this matters
17// doubly: both are RPC-backed providers, so under the proxy their entire
18// lifecycle (broker registration, consumer wiring, worker restart hooks)
19// runs in the sidecar process.
20const { test } = Test.make({
21 providers: Cloudflare.providers(),
22 dev: true,
23});
24
25const logLevel = Effect.provideService(MinimumLogLevel, process.env.DEBUG ? "Debug" : "Info");
26
27class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
28 status: number;
29}> {}
30
31const getJsonReady = (url: string) =>
32 Effect.gen(function* () {
33 const client = yield* HttpClient.HttpClient;
34 const res = yield* client.get(url).pipe(
35 Effect.flatMap((res) =>
36 res.status === 200
37 ? Effect.succeed(res)
38 : Effect.fail(new WorkerNotReady({ status: res.status })),
39 ),
40 Effect.retry({
41 while: (e): e is WorkerNotReady => e instanceof WorkerNotReady,
42 // Cap the backoff: an uncapped exponential over 10 recurs sums to
43 // ~8.5 minutes and turns a persistent non-200 into an apparent hang.
44 schedule: Schedule.max([
45 Schedule.min([Schedule.exponential("500 millis"), Schedule.spaced("2 seconds")]),
46 Schedule.recurs(10),
47 ]),
48 }),
49 );
50 return yield* res.json;
51 }).pipe(Effect.orDie);
52
53/**
54 * Under `alchemy dev` the Queue resource is emulated (a `dev:` queueId) and
55 * the worker's producer binding targets the local broker. The fixture both
56 * produces and consumes, so this pins end-to-end local delivery: send over
57 * the binding, broker dispatches to the `queue()` handler, poll the
58 * recorded bodies back over HTTP.
59 */
60test.provider(
61 "local queue delivers produced messages to the consumer",
62 (stack) =>
63 Effect.gen(function* () {
64 yield* stack.destroy();
65
66 const deployed = yield* stack.deploy(
67 Effect.gen(function* () {
68 const queue = yield* Cloudflare.Queues.Queue("LocalQueue");
69 const worker = yield* Cloudflare.Worker("queue-local-worker", {
70 main: pathe.resolve(import.meta.dirname, "fixtures/queue-local-worker.ts"),
71 env: { QUEUE: queue },
72 });
73 yield* Cloudflare.Queues.Consumer("LocalConsumer", {
74 queueId: queue.queueId,
75 scriptName: worker.workerName,
76 });
77 return { queue, worker };
78 }),
79 );
80
81 // The local provider fabricates a `dev:` id — proof no cloud call ran
82 // — and the worker serves from the local dev proxy.
83 expect(deployed.queue.queueId).toMatch(/^dev:/);
84 expect(deployed.worker.url).toMatch(/^http:\/\/localhost:\d+$/);
85
86 const sent = (yield* getJsonReady(`${deployed.worker.url}/send?text=local-hello`)) as {
87 sent: string;
88 };
89 expect(sent.sent).toBe("local-hello");
90
91 // Poll until the broker delivers to the fixture's queue() handler.
92 const received = yield* getJsonReady(`${deployed.worker.url}/received`).pipe(
93 Effect.map((body) => (body as { received: string[] }).received),
94 Effect.repeat({
95 schedule: Schedule.spaced("500 millis"),
96 until: (received) => received.includes("local-hello"),
97 times: 30,
98 }),
99 );
100 expect(received).toContain("local-hello");
101
102 yield* stack.destroy();
103 }).pipe(logLevel),
104 {
105 tags: [
106 "provider:cloudflare",
107 "provider:cloudflare:queue",
108 "provider:cloudflare:worker",
109 "local",
110 ],
111 timeout: 120_000,
112 },
113);
114
115/**
116 * Consumer `settings` reach the local broker with the field names and units
117 * the runtime expects (`batchSize` → `maxBatchSize`, `maxWaitTimeMs` (ms) →
118 * `maxBatchTimeout` (s)). With `batchSize: 2` and a 2s wait, five quick
119 * sends must arrive in batches of at most 2 — under the broker's defaults
120 * (batch of 5, 1s flush) they'd land as one batch of 5. The trailing
121 * single-message batch flushing within the poll window pins the ms→s
122 * conversion: an unconverted 2000 would stall it for over half an hour.
123 */
124test.provider(
125 "local consumer settings control broker batch size and flush timeout",
126 (stack) =>
127 Effect.gen(function* () {
128 yield* stack.destroy();
129
130 const deployed = yield* stack.deploy(
131 Effect.gen(function* () {
132 const queue = yield* Cloudflare.Queues.Queue("BatchSettingsQueue");
133 const worker = yield* Cloudflare.Worker("queue-batch-local-worker", {
134 main: pathe.resolve(import.meta.dirname, "fixtures/queue-local-worker.ts"),
135 env: { QUEUE: queue },
136 });
137 yield* Cloudflare.Queues.Consumer("BatchSettingsConsumer", {
138 queueId: queue.queueId,
139 scriptName: worker.workerName,
140 settings: { batchSize: 2, maxWaitTimeMs: 2000 },
141 });
142 return { queue, worker };
143 }),
144 );
145
146 expect(deployed.queue.queueId).toMatch(/^dev:/);
147
148 for (let i = 0; i < 5; i++) {
149 yield* getJsonReady(`${deployed.worker.url}/send?text=batch-${i}`);
150 }
151
152 // All five arrive: two full batches immediately, the leftover after
153 // the 2s flush timeout.
154 const received = yield* getJsonReady(`${deployed.worker.url}/received`).pipe(
155 Effect.map((body) => (body as { received: string[] }).received),
156 Effect.repeat({
157 schedule: Schedule.spaced("500 millis"),
158 until: (received) => received.length >= 5,
159 times: 30,
160 }),
161 );
162 expect(received.length).toBe(5);
163
164 const { batches } = (yield* getJsonReady(`${deployed.worker.url}/batches`)) as {
165 batches: number[];
166 };
167 expect(Math.max(...batches)).toBe(2);
168 expect(batches.reduce((a, b) => a + b, 0)).toBe(5);
169
170 yield* stack.destroy();
171 }).pipe(logLevel),
172 {
173 tags: [
174 "provider:cloudflare",
175 "provider:cloudflare:queue",
176 "provider:cloudflare:worker",
177 "local",
178 ],
179 timeout: 120_000,
180 },
181);
182
183/**
184 * Regression test for #988: a consumer configured with a dead-letter queue
185 * that the worker does not itself consume. The DLQ binding resolves through
186 * the local runtime's registry proxy; the worker previously failed to start
187 * with `WorkerdStartFailed` because the DLQ binding referenced the
188 * `cloudflare-runtime:registry-proxy` service before it was defined.
189 */
190test.provider(
191 "local queue consumer with an unconsumed dead-letter queue starts and delivers",
192 (stack) =>
193 Effect.gen(function* () {
194 yield* stack.destroy();
195
196 const deployed = yield* stack.deploy(
197 Effect.gen(function* () {
198 const queue = yield* Cloudflare.Queues.Queue("DlqSourceQueue");
199 const dlq = yield* Cloudflare.Queues.Queue("DlqTargetQueue");
200 const worker = yield* Cloudflare.Worker("queue-dlq-local-worker", {
201 main: pathe.resolve(import.meta.dirname, "fixtures/queue-local-worker.ts"),
202 env: { QUEUE: queue },
203 });
204 yield* Cloudflare.Queues.Consumer("DlqLocalConsumer", {
205 queueId: queue.queueId,
206 scriptName: worker.workerName,
207 deadLetterQueue: dlq.queueName,
208 settings: { maxRetries: 1 },
209 });
210 return { queue, dlq, worker };
211 }),
212 );
213
214 expect(deployed.queue.queueId).toMatch(/^dev:/);
215 expect(deployed.dlq.queueId).toMatch(/^dev:/);
216
217 // The worker starts (the DLQ binding resolves) and delivers normally.
218 const sent = (yield* getJsonReady(`${deployed.worker.url}/send?text=dlq-hello`)) as {
219 sent: string;
220 };
221 expect(sent.sent).toBe("dlq-hello");
222
223 const received = yield* getJsonReady(`${deployed.worker.url}/received`).pipe(
224 Effect.map((body) => (body as { received: string[] }).received),
225 Effect.repeat({
226 schedule: Schedule.spaced("500 millis"),
227 until: (received) => received.includes("dlq-hello"),
228 times: 30,
229 }),
230 );
231 expect(received).toContain("dlq-hello");
232
233 yield* stack.destroy();
234 }).pipe(logLevel),
235 {
236 tags: [
237 "provider:cloudflare",
238 "provider:cloudflare:queue",
239 "provider:cloudflare:worker",
240 "local",
241 ],
242 timeout: 120_000,
243 },
244);
245
246/**
247 * The full `Alchemy.remote()` queue roundtrip in dev: the local worker
248 * produces through the deployed shim into the REAL queue, and its local
249 * `queue()` handler receives the messages back through the runtime's pull
250 * loop (the Consumer's local provider attaches an `http_pull` consumer to
251 * the real queue; the pull loop drains it into the local broker).
252 */
253test.provider(
254 "Alchemy.remote() queue round-trips: local produce, real queue, local consume",
255 (stack) =>
256 Effect.gen(function* () {
257 yield* stack.destroy();
258
259 const deployed = yield* stack.deploy(
260 Effect.gen(function* () {
261 const queue = yield* Cloudflare.Queues.Queue("LiveRoundtripQueue").pipe(Alchemy.remote());
262 const worker = yield* Cloudflare.Worker("queue-roundtrip-worker", {
263 main: pathe.resolve(import.meta.dirname, "fixtures/queue-local-worker.ts"),
264 env: { QUEUE: queue },
265 });
266 yield* Cloudflare.Queues.Consumer("LiveRoundtripConsumer", {
267 queueId: queue.queueId,
268 scriptName: worker.workerName,
269 });
270 return { queue, worker };
271 }),
272 );
273
274 // The queue is real; the worker (and its queue() handler) is local.
275 expect(deployed.queue.queueId).not.toMatch(/^dev:/);
276
277 // Produce through the shim into the real queue.
278 const sent = (yield* getJsonReady(`${deployed.worker.url}/send?text=roundtrip-hello`)) as {
279 sent: string;
280 };
281 expect(sent.sent).toBe("roundtrip-hello");
282
283 // The pull loop drains the real queue into the local broker, which
284 // delivers to the fixture's queue() handler.
285 const received = yield* getJsonReady(`${deployed.worker.url}/received`).pipe(
286 Effect.map((body) => (body as { received: string[] }).received),
287 Effect.repeat({
288 schedule: Schedule.spaced("4 seconds"),
289 until: (received) => received.includes("roundtrip-hello"),
290 times: 10,
291 }),
292 );
293 expect(received).toContain("roundtrip-hello");
294
295 yield* stack.destroy();
296
297 // Destroy removed the real queue (with its pull consumer and shim).
298 const { accountId } = yield* yield* CloudflareEnvironment;
299 const gone = yield* queues.getQueue({ accountId, queueId: deployed.queue.queueId }).pipe(
300 Effect.as(false),
301 Effect.catchTag("QueueNotFound", () => Effect.succeed(true)),
302 );
303 expect(gone).toBe(true);
304 }).pipe(logLevel),
305 {
306 tags: [
307 "provider:cloudflare",
308 "provider:cloudflare:queue",
309 "provider:cloudflare:worker",
310 "live",
311 ],
312 timeout: 120_000,
313 },
314);
315
316/**
317 * `Alchemy.remote()` queues in dev: Cloudflare's remote-binding (preview)
318 * sessions reject queue bindings at the platform level
319 * (cloudflare/workers-sdk#9929), so production goes through the deployed
320 * shim instead — the eval registers a real shim worker holding the actual
321 * queue binding, and the local worker's producer binding relays the queue
322 * wire protocol to it over HTTPS with a minted bearer token.
323 *
324 * The proof is out-of-band: an `http_pull` consumer attached directly via
325 * the cloud API pulls the messages back from the REAL queue.
326 */
327test.provider(
328 "Alchemy.remote() queue in dev produces through the deployed shim",
329 (stack) =>
330 Effect.gen(function* () {
331 yield* stack.destroy();
332
333 const deployed = yield* stack.deploy(
334 Effect.gen(function* () {
335 const queue = yield* Cloudflare.Queues.Queue("LiveDevQueue").pipe(Alchemy.remote());
336 const worker = yield* Cloudflare.Worker("queue-live-worker", {
337 main: pathe.resolve(import.meta.dirname, "fixtures/queue-local-worker.ts"),
338 env: { QUEUE: queue },
339 });
340 return { queue, worker };
341 }),
342 );
343
344 // The queue is real (live provider), the worker is local.
345 expect(deployed.queue.queueId).not.toMatch(/^dev:/);
346 const { accountId } = yield* yield* CloudflareEnvironment;
347
348 // Out-of-band http_pull consumer so the test can pull from the real
349 // queue (a local queue() handler cannot receive from a live queue —
350 // Cloudflare pushes to deployed consumers only).
351 yield* queues.createConsumer({
352 accountId,
353 queueId: deployed.queue.queueId,
354 type: "http_pull",
355 });
356
357 // Produce through the local worker: binding → forwarder → shim →
358 // queue. The first send rides out the shim's workers.dev propagation
359 // (the forwarder retries 404/503 internally; the outer retry covers
360 // the tail).
361 const sent = (yield* getJsonReady(`${deployed.worker.url}/send?text=live-hello`)) as {
362 sent: string;
363 };
364 expect(sent.sent).toBe("live-hello");
365 const batch = (yield* getJsonReady(`${deployed.worker.url}/sendbatch?text=live-batch`)) as {
366 sent: number;
367 };
368 expect(batch.sent).toBe(2);
369
370 // Pull the messages back from the REAL queue. Pulled messages are
371 // leased (invisible to subsequent pulls until the timeout), so
372 // accumulate bodies across pulls instead of expecting one batch to
373 // carry everything.
374 const seen = new Set<string>();
375 const bodies = yield* queues
376 .pullMessage({
377 accountId,
378 queueId: deployed.queue.queueId,
379 batchSize: 10,
380 visibilityTimeoutMs: 1000,
381 })
382 .pipe(
383 Effect.map((res) => {
384 for (const m of res.messages ?? []) {
385 seen.add(String(m.body ?? ""));
386 }
387 return Array.from(seen);
388 }),
389 Effect.repeat({
390 schedule: Schedule.spaced("2 seconds"),
391 until: (bodies) =>
392 bodies.some((b) => b.includes("live-hello")) &&
393 bodies.some((b) => b.includes("live-batch-a")) &&
394 bodies.some((b) => b.includes("live-batch")),
395 times: 20,
396 }),
397 );
398 expect(bodies.some((b) => b.includes("live-hello"))).toBe(true);
399 expect(bodies.some((b) => b.includes("live-batch-a"))).toBe(true);
400
401 yield* stack.destroy();
402
403 // Destroy removed the real queue (and with it the shim + consumer).
404 const gone = yield* queues.getQueue({ accountId, queueId: deployed.queue.queueId }).pipe(
405 Effect.as(false),
406 Effect.catchTag("QueueNotFound", () => Effect.succeed(true)),
407 );
408 expect(gone).toBe(true);
409 }).pipe(logLevel),
410 {
411 tags: [
412 "provider:cloudflare",
413 "provider:cloudflare:queue",
414 "provider:cloudflare:worker",
415 "live",
416 ],
417 timeout: 120_000,
418 },
419);