1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Test from "@/Test/Alchemy";
3import { expect } from "alchemy-test";
4import * as Effect from "effect/Effect";
5import { MinimumLogLevel } from "effect/References";
6import * as HttpClient from "effect/http/HttpClient";
7import * as pathe from "pathe";
8
9const { test } = Test.make({
10 providers: Cloudflare.providers(),
11 dev: true,
12});
13
14const logLevel = Effect.provideService(
15 MinimumLogLevel,
16 process.env.DEBUG ? "Debug" : "Info",
17);
18
19// Regression test for the secondary symptom of #1109: when no consumer is
20// registered for a queue, the dev registry's `ExternalQueueConsumer` accepts
21// and drops the message with a warning — but it must still settle the
22// producer's `send()` promise. It previously responded without draining the
23// request body, which left `send()` pending forever (workerd's queue client
24// only settles once the body is consumed).
25test.provider(
26 "plain worker send() to unconsumed queue resolves",
27 (stack) =>
28 Effect.gen(function* () {
29 yield* stack.destroy();
30 const deployed = yield* stack.deploy(
31 Effect.gen(function* () {
32 const queue = yield* Cloudflare.Queues.Queue("DropProbeQueue");
33 const worker = yield* Cloudflare.Worker("drop-probe-worker", {
34 main: pathe.resolve(
35 import.meta.dirname,
36 "fixtures/queue-local-worker.ts",
37 ),
38 env: { QUEUE: queue },
39 });
40 return { queue, worker };
41 }),
42 );
43 const client = yield* HttpClient.HttpClient;
44 const res = yield* client
45 .get(`${deployed.worker.url}/send?text=drop-me`)
46 .pipe(Effect.timeout("20 seconds"), Effect.orDie);
47 expect(res.status).toBe(200);
48 yield* stack.destroy();
49 }).pipe(logLevel),
50 {
51 tags: [
52 "provider:cloudflare",
53 "provider:cloudflare:queue",
54 "provider:cloudflare:worker",
55 "local",
56 ],
57 timeout: 120_000,
58 },
59);