1import * as queues from "@distilled.cloud/cloudflare/queues";
2import * as workers from "@distilled.cloud/cloudflare/workers";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as MutableHashMap from "effect/MutableHashMap";
6import * as Option from "effect/Option";
7import * as Schedule from "effect/Schedule";
8import * as Stream from "effect/Stream";
9import { isResolved } from "../../Diff.ts";
10import * as ProviderLayer from "../../Local/ProviderLayer.ts";
11import * as RpcProvider from "../../Local/RpcProvider.ts";
12import * as Provider from "../../Provider.ts";
13import { Resource } from "../../Resource.ts";
14import { Stack } from "../../Stack.ts";
15import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
16import { localAccountId } from "../LocalAccount.ts";
17import {
18 isLiveId,
19 LOCAL_PROVIDERS_URL,
20 LocalRuntimeState,
21 localRuntimeServices,
22} from "../LocalRuntime.ts";
23import type { Providers } from "../Providers.ts";
24
25export type ConsumerProps = {
26 /**
27 * The queue ID to attach the consumer to.
28 */
29 queueId: string;
30 /**
31 * Name of the Worker script that will consume messages.
32 */
33 scriptName: string;
34 /**
35 * Optional dead letter queue name for failed messages.
36 */
37 deadLetterQueue?: string;
38 /**
39 * Consumer settings.
40 */
41 settings?: ConsumerSettings;
42};
43
44export interface ConsumerSettings {
45 /**
46 * The maximum number of messages per batch.
47 * @default 10
48 */
49 batchSize?: number;
50 /**
51 * The maximum number of concurrent consumer invocations.
52 */
53 maxConcurrency?: number;
54 /**
55 * The maximum number of retries for a message.
56 * @default 3
57 */
58 maxRetries?: number;
59 /**
60 * The maximum time to wait for a batch to fill, in milliseconds.
61 * @default 5000
62 */
63 maxWaitTimeMs?: number;
64 /**
65 * The number of seconds to wait before retrying a message.
66 */
67 retryDelay?: number;
68}
69
70export type Consumer = Resource<
71 "Cloudflare.Queues.Consumer",
72 ConsumerProps,
73 {
74 consumerId: string;
75 queueId: string;
76 scriptName: string;
77 accountId: string;
78 deadLetterQueue?: string;
79 settings?: ConsumerSettings;
80 /**
81 * Dev only, live queues only: the real queue's name, resolved from the
82 * cloud so the local runtime can wire the broker + pull loop for a
83 * locally-running consumer of an `Alchemy.remote()` queue.
84 */
85 queueName?: string;
86 /**
87 * Dev only, live queues only: id of the `http_pull` consumer attached
88 * to the real queue so the local runtime can drain it via the HTTP
89 * pull API. Deleted when this resource is deleted.
90 */
91 pullConsumerId?: string;
92 },
93 never,
94 Providers
95>;
96
97/**
98 * A Cloudflare Queue Consumer that processes messages from a Queue.
99 *
100 * Register a Worker as a consumer of a Queue. The Worker's `queue()`
101 * handler will be invoked with batches of messages.
102 *
103 * Cloudflare allows at most one Worker consumer per queue (HTTP-pull
104 * consumers can coexist). The reconciler enforces this: if the queue
105 * already has a Worker consumer pointing at a different logical Worker's
106 * script, the deploy fails with a clear error rather than silently
107 * adopting it. A stranded consumer from a prior generation of the *same*
108 * Worker (identified by the scripts' ownership tags) is rebuilt in place.
109 * ### Registering a Consumer
110 * **Example:** Basic consumer
111 * ```typescript
112 * const queue = yield* Cloudflare.Queues.Queue("MyQueue");
113 * const worker = yield* Cloudflare.Worker("Worker", { ... });
114 *
115 * yield* Cloudflare.Queues.Consumer("MyConsumer", {
116 * queueId: queue.queueId,
117 * scriptName: worker.workerName,
118 * });
119 * ```
120 *
121 * **Example:** Consumer with settings
122 * ```typescript
123 * yield* Cloudflare.Queues.Consumer("MyConsumer", {
124 * queueId: queue.queueId,
125 * scriptName: worker.workerName,
126 * settings: {
127 * batchSize: 50,
128 * maxRetries: 5,
129 * maxWaitTimeMs: 10000,
130 * },
131 * });
132 * ```
133 *
134 * @resource
135 * @product Queues
136 * @category Storage & Databases
137 */
138export const Consumer = Resource<Consumer>("Cloudflare.Queues.Consumer", {
139 aliases: ["Cloudflare.QueueConsumer"],
140});
141
142/**
143 * Find and detach every worker queue-consumer pointing at `scriptName`.
144 * Queue consumers have no by-script lookup, so scan the account's queues
145 * (the list response inlines each queue's consumers). Waits until each
146 * detach propagates to the workers subsystem so a follow-up deleteScript
147 * doesn't re-race the conflict.
148 *
149 * Shared by the Worker provider's script delete (QueueConsumerConflict
150 * recovery) and the Queue provider's orphaned-script cleanup.
151 *
152 * @internal
153 */
154export const detachQueueConsumersOfScript = Effect.fn(function* (
155 accountId: string,
156 scriptName: string,
157) {
158 const pages = yield* queues.listQueues.pages({ accountId }).pipe(
159 Stream.runCollect,
160 Effect.catchTag("InvalidRoute", () => Effect.succeed([])),
161 );
162 const targets = Array.from(pages).flatMap((page) =>
163 (page.result ?? []).flatMap((queue) =>
164 (queue.consumers ?? []).flatMap((consumer) =>
165 queue.queueId &&
166 consumer.type === "worker" &&
167 consumer.consumerId &&
168 "scriptName" in consumer &&
169 consumer.scriptName === scriptName
170 ? [{ queueId: queue.queueId, consumerId: consumer.consumerId }]
171 : [],
172 ),
173 ),
174 );
175 yield* Effect.forEach(
176 targets,
177 ({ queueId, consumerId }) =>
178 queues.deleteConsumer({ accountId, queueId, consumerId }).pipe(
179 Effect.catchTag(
180 ["ConsumerNotFound", "QueueNotFound"],
181 () => Effect.void,
182 ),
183 Effect.andThen(
184 queues.getConsumer({ accountId, queueId, consumerId }).pipe(
185 Effect.flatMap(() => Effect.fail("still-attached" as const)),
186 Effect.catchTag(
187 ["ConsumerNotFound", "QueueNotFound"],
188 () => Effect.void,
189 ),
190 Effect.retry({
191 while: (e) => e === "still-attached",
192 schedule: Schedule.max([
193 Schedule.spaced("1 second"),
194 Schedule.recurs(30),
195 ]),
196 }),
197 Effect.ignore,
198 ),
199 ),
200 ),
201 { concurrency: 5, discard: true },
202 );
203});
204
205// Cloudflare allows a single Worker consumer per queue, so the
206// first match in the paginated stream is the only one. Using
207// `.items` defeats single-page lookups that would otherwise
208// miss late-arriving consumers under eventual consistency.
209const findWorkerConsumer = (acct: string, queueId: string) =>
210 queues.listConsumers.items({ accountId: acct, queueId }).pipe(
211 Stream.map(toObserved),
212 Stream.filter((c): c is ObservedConsumer => c !== undefined),
213 Stream.runHead,
214 Effect.map(Option.getOrUndefined),
215 // A queue created earlier in the same plan is eventually consistent:
216 // listConsumers can transiently 404 (`QueueNotFound`, code 11000) before
217 // the fresh queue is visible to the consumers API. Treat that as "no
218 // consumer yet" — the caller falls through to createConsumer, which
219 // retries QueueNotFound until the queue propagates.
220 Effect.catchTag("QueueNotFound", () => Effect.succeed(undefined)),
221 );
222
223export const ConsumerProviderLive = () =>
224 Provider.succeed(Consumer, {
225 // The `consumerId` is not marked as stable because if you start in dev mode, the ID will change on first deploy.
226 stables: ["accountId"],
227 // Queue consumers are sub-resources of a queue with no account-wide
228 // enumeration API, so fan out: list every queue in the account, then
229 // list each queue's consumers and keep the worker ones (the only kind
230 // this resource manages, matching `read`/`reconcile`).
231 list: Effect.fn(function* () {
232 const { accountId } = yield* yield* CloudflareEnvironment;
233 const queueIds = yield* queues.listQueues.pages({ accountId }).pipe(
234 Stream.runCollect,
235 Effect.map((chunk) =>
236 Array.from(chunk).flatMap((page) =>
237 (page.result ?? [])
238 .map((q) => q.queueId)
239 .filter((id): id is string => id != null),
240 ),
241 ),
242 // Account not entitled for Queues — nothing to enumerate.
243 Effect.catchTag("InvalidRoute", () => Effect.succeed<string[]>([])),
244 );
245 const rows = yield* Effect.forEach(
246 queueIds,
247 (queueId) =>
248 queues.listConsumers.pages({ accountId, queueId }).pipe(
249 Stream.runCollect,
250 Effect.map((chunk) =>
251 Array.from(chunk).flatMap((page) =>
252 (page.result ?? []).flatMap((c): Consumer["Attributes"][] => {
253 if (c.type !== "worker" || !c.consumerId) return [];
254 const s = c.settings ?? undefined;
255 return [
256 {
257 consumerId: c.consumerId,
258 queueId,
259 scriptName: c.scriptName ?? "",
260 accountId,
261 deadLetterQueue: c.deadLetterQueue ?? undefined,
262 settings: s
263 ? {
264 batchSize: s.batchSize ?? undefined,
265 maxConcurrency: s.maxConcurrency ?? undefined,
266 maxRetries: s.maxRetries ?? undefined,
267 maxWaitTimeMs: s.maxWaitTimeMs ?? undefined,
268 retryDelay: s.retryDelay ?? undefined,
269 }
270 : undefined,
271 },
272 ];
273 }),
274 ),
275 ),
276 // Queue deleted mid-list or partial entitlement — skip it.
277 Effect.catchTag(["QueueNotFound", "InvalidRoute"], () =>
278 Effect.succeed<Consumer["Attributes"][]>([]),
279 ),
280 ),
281 { concurrency: 10 },
282 );
283 return rows.flat();
284 }),
285 diff: Effect.fn(function* ({ olds, news, output }) {
286 const { accountId } = yield* yield* CloudflareEnvironment;
287 if (!isResolved(news)) return undefined;
288 // If either contain `dev:` IDs, we need to update to live ones.
289 // The live resource doesn't exist yet, so there's no need to replace even when we otherwise would.
290 if (!isLiveId(output?.queueId) || !isLiveId(output?.consumerId)) {
291 return { action: "update" };
292 }
293 if ((output?.accountId ?? accountId) !== accountId) {
294 return { action: "replace" } as const;
295 }
296 // Queue change requires replacement — consumerId is bound
297 // to a queue and the API has no "move consumer" verb.
298 if (output?.queueId && news.queueId !== output.queueId) {
299 return { action: "replace", deleteFirst: true } as const;
300 }
301 // Settings / DLQ / script drift is an update. We DON'T
302 // escalate scriptName changes to `replace` because the
303 // engine resolves cross-resource Output<string> refs (a
304 // sibling Worker's `workerName`) lazily — when the upstream
305 // Worker is created in the same plan, `news` is partially
306 // unresolved at diff time and `isResolved(news)` short-
307 // circuits up top. Falling through to "update" lets the
308 // engine call reconcile with fully-resolved `news`, where
309 // we detect script drift and rebuild the consumer in
310 // place (Cloudflare's PUT silently ignores `script_name`
311 // changes, so reconcile does delete-then-create).
312 if (
313 JSON.stringify(olds.settings ?? {}) !==
314 JSON.stringify(news.settings ?? {}) ||
315 (olds.deadLetterQueue ?? undefined) !==
316 (news.deadLetterQueue ?? undefined) ||
317 (output?.scriptName !== undefined &&
318 news.scriptName !== output.scriptName)
319 ) {
320 return { action: "update" } as const;
321 }
322 }),
323 reconcile: Effect.fn(function* ({ news, output }) {
324 const { accountId } = yield* yield* CloudflareEnvironment;
325 const acct = output?.accountId ?? accountId;
326 // Only use `output.queueId` if it's a live ID (i.e. no `dev:` prefix).
327 // Otherwise, the lookup will fail because the request is malformed.
328 const queueId = isLiveId(output?.queueId) ? output.queueId : news.queueId;
329
330 // Delete a consumer and wait for Cloudflare's worker subsystem to
331 // drop its claim on the old script so createConsumer below doesn't
332 // race the queue↔script propagation lag.
333 const detachConsumer = Effect.fn(function* (consumerId: string) {
334 yield* queues
335 .deleteConsumer({ accountId: acct, queueId, consumerId })
336 .pipe(Effect.catchTag("ConsumerNotFound", () => Effect.void));
337 yield* queues
338 .getConsumer({ accountId: acct, queueId, consumerId })
339 .pipe(
340 Effect.flatMap(() => Effect.fail("still-attached" as const)),
341 Effect.catchTag("ConsumerNotFound", () => Effect.void),
342 Effect.retry({
343 while: (e) => e === "still-attached",
344 schedule: Schedule.max([
345 Schedule.spaced("1 second"),
346 Schedule.recurs(30),
347 ]),
348 }),
349 Effect.ignore,
350 );
351 });
352
353 // Observe — prefer the cached consumerId, then fall back to
354 // listConsumers (paginated) to recover from out-of-band
355 // deletes or partial state-persistence failures. Track
356 // whether the observation came from the cached id or the
357 // list scan: a different-script worker consumer found via
358 // the list scan is potentially foreign (state was lost,
359 // someone else attached the consumer), and silently
360 // updating it could clobber another team's wiring.
361 let observed: ObservedConsumer | undefined;
362 let owned = false;
363 if (isLiveId(output?.consumerId)) {
364 const fetched = yield* queues
365 .getConsumer({
366 accountId: acct,
367 queueId,
368 consumerId: output.consumerId,
369 })
370 .pipe(
371 Effect.catchTag("ConsumerNotFound", () =>
372 Effect.succeed(undefined),
373 ),
374 );
375 if (fetched) {
376 observed = toObserved(fetched);
377 owned = observed !== undefined;
378 }
379 }
380 if (!observed) {
381 observed = yield* findWorkerConsumer(acct, queueId);
382 }
383
384 // Owned consumer pointing at a different script: rebuild
385 // it in place. Cloudflare's PUT consumer silently ignores
386 // `script_name` changes on existing consumers (the live
387 // record stays pinned to the original worker), and the
388 // platform allows only one Worker consumer per queue, so
389 // the only path to re-point is delete-then-create.
390 if (
391 owned &&
392 observed &&
393 observed.scriptName !== undefined &&
394 observed.scriptName !== news.scriptName
395 ) {
396 yield* detachConsumer(observed.consumerId);
397 observed = undefined;
398 owned = false;
399 }
400
401 // A consumer found via the list scan (state loss) pointing at a
402 // different script. Probe the script it points at: when it no
403 // longer exists, or is a *prior generation of the same logical
404 // Worker* (same `alchemy:stack/stage/id` ownership tags as the
405 // configured script — replacements mint a new physical name, and
406 // a crashed apply can strand the old generation's consumer with
407 // its state entry lost), the wiring is ours to rebuild. Any other
408 // script — another team's, or a *different* Worker in this stack
409 // (a real misconfiguration: two consumers declared for one queue)
410 // — is refused loudly rather than silently stolen.
411 if (
412 observed &&
413 !owned &&
414 observed.scriptName !== undefined &&
415 observed.scriptName !== news.scriptName
416 ) {
417 const scriptTags = Effect.fn(function* (scriptName: string) {
418 const settings = yield* workers
419 .getScriptScriptAndVersionSetting({
420 accountId: acct,
421 scriptName,
422 })
423 .pipe(
424 Effect.catchTag(["WorkerNotFound", "WorkerHasNoVersions"], () =>
425 Effect.succeed(undefined),
426 ),
427 );
428 return settings === undefined
429 ? undefined
430 : new Set(settings.tags ?? []);
431 });
432 const observedTags = yield* scriptTags(observed.scriptName);
433 let staleGeneration = observedTags === undefined;
434 if (observedTags !== undefined) {
435 const stack = yield* Stack;
436 const desiredTags = news.scriptName
437 ? yield* scriptTags(news.scriptName)
438 : undefined;
439 const observedId = Array.from(observedTags).find((t) =>
440 t.startsWith("alchemy:id:"),
441 );
442 staleGeneration =
443 observedId !== undefined &&
444 observedTags.has(`alchemy:stack:${stack.name}`) &&
445 observedTags.has(`alchemy:stage:${stack.stage}`) &&
446 desiredTags !== undefined &&
447 desiredTags.has(observedId) &&
448 desiredTags.has(`alchemy:stack:${stack.name}`) &&
449 desiredTags.has(`alchemy:stage:${stack.stage}`);
450 }
451 if (!staleGeneration) {
452 return yield* Effect.die(
453 `Cloudflare queue "${queueId}" already has a worker ` +
454 `consumer for script "${observed.scriptName}", but this ` +
455 `resource is configured for "${news.scriptName}" and ` +
456 `local state for the consumer was missing. Each queue ` +
457 `can have only one worker consumer — delete the ` +
458 `existing one, update scriptName to match, or restore ` +
459 `the consumer's state entry before redeploying.`,
460 );
461 }
462 yield* detachConsumer(observed.consumerId);
463 observed = undefined;
464 }
465
466 // Ensure — create if missing. ConsumerAlreadyExists is the
467 // race signal: another reconcile or peer beat us to it.
468 // Re-run the lookup; the paginated stream tolerates the
469 // single-page eventual-consistency window the previous
470 // implementation missed.
471 let consumerId: string;
472 if (!observed) {
473 const created = yield* queues
474 .createConsumer({
475 accountId: acct,
476 queueId,
477 scriptName: news.scriptName,
478 type: "worker",
479 deadLetterQueue: news.deadLetterQueue,
480 settings: news.settings,
481 })
482 .pipe(
483 // Two eventual-consistency races on a fresh deploy, both
484 // retried on the same bounded schedule:
485 // - QueueHandlerMissing (11001): the sibling Worker resource
486 // pre-creates a placeholder script with no `queue` handler;
487 // Cloudflare returns 11001 until the real reconcile uploads
488 // the handler.
489 // - QueueNotFound (11000): the queue was created earlier in
490 // this same plan but isn't yet visible to the consumers API.
491 Effect.tapError((e) =>
492 e._tag === "QueueHandlerMissing" || e._tag === "QueueNotFound"
493 ? Effect.logDebug(
494 `Consumer create: not ready for worker ` +
495 `"${news.scriptName}" (${e._tag}), retrying`,
496 )
497 : Effect.void,
498 ),
499 Effect.retry({
500 while: (e) =>
501 e._tag === "QueueHandlerMissing" || e._tag === "QueueNotFound",
502 schedule: queueHandlerReadinessSchedule,
503 }),
504 Effect.catchTag("ConsumerAlreadyExists", (cause) =>
505 Effect.gen(function* () {
506 const match = yield* findWorkerConsumer(acct, queueId);
507 if (!match) {
508 return yield* Effect.die(
509 `Cloudflare reported a worker consumer already ` +
510 `exists on queue "${queueId}", but listConsumers ` +
511 `returned none. Retry the deploy; if this ` +
512 `persists, the queue is in an inconsistent ` +
513 `state. Underlying error: ${cause.message}`,
514 );
515 }
516 if (
517 match.scriptName !== undefined &&
518 match.scriptName !== news.scriptName
519 ) {
520 return yield* Effect.die(
521 `Cloudflare queue "${queueId}" already has a ` +
522 `worker consumer for script "${match.scriptName}", ` +
523 `but this resource is configured for ` +
524 `"${news.scriptName}". Each queue can have only ` +
525 `one worker consumer — delete the existing one ` +
526 `or update scriptName to match before redeploying.`,
527 );
528 }
529 return match;
530 }),
531 ),
532 );
533 consumerId = created.consumerId!;
534 } else {
535 consumerId = observed.consumerId;
536 }
537
538 // Sync — Cloudflare replaces all mutable fields on
539 // updateConsumer, so always issue this so adoption converges
540 // and settings drift gets corrected on every reconcile.
541 // updateConsumer hits the same "queue handler missing" race
542 // window as create when the worker is mid-upload, so apply
543 // the same bounded retry.
544 yield* queues
545 .updateConsumer({
546 accountId: acct,
547 queueId,
548 consumerId,
549 scriptName: news.scriptName,
550 type: "worker",
551 settings: news.settings,
552 deadLetterQueue: news.deadLetterQueue,
553 })
554 .pipe(
555 Effect.retry({
556 while: (e) => e._tag === "QueueHandlerMissing",
557 schedule: queueHandlerReadinessSchedule,
558 }),
559 );
560
561 yield* queues.getConsumer({ accountId: acct, queueId, consumerId }).pipe(
562 Effect.flatMap((fetched) =>
563 toObserved(fetched)?.scriptName === news.scriptName
564 ? Effect.void
565 : Effect.fail("ScriptUnbound" as const),
566 ),
567 Effect.catchTag("ConsumerNotFound", () =>
568 Effect.fail("ScriptUnbound" as const),
569 ),
570 Effect.retry({
571 while: (e) => e === "ScriptUnbound",
572 schedule: queueHandlerReadinessSchedule,
573 }),
574 );
575
576 return {
577 consumerId,
578 queueId,
579 scriptName: news.scriptName!,
580 accountId: acct,
581 deadLetterQueue: news.deadLetterQueue,
582 settings: news.settings,
583 };
584 }),
585 delete: Effect.fn(function* ({ output }) {
586 // If the consumerId is a `dev:` ID, the resource only exists locally, so we don't need to delete it from Cloudflare.
587 if (!isLiveId(output.consumerId)) return;
588
589 yield* queues
590 .deleteConsumer({
591 accountId: output.accountId,
592 queueId: output.queueId,
593 consumerId: output.consumerId,
594 })
595 .pipe(Effect.catchTag("ConsumerNotFound", () => Effect.void));
596
597 // Block until Cloudflare's worker subsystem stops claiming
598 // the script as a queue consumer. Without this the
599 // sibling Worker.delete races on `QueueConsumerConflict`
600 // (code 10064) — `deleteConsumer` returns success on the
601 // queue subsystem before the script-side view propagates.
602 yield* queues
603 .getConsumer({
604 accountId: output.accountId,
605 queueId: output.queueId,
606 consumerId: output.consumerId,
607 })
608 .pipe(
609 Effect.flatMap(() => Effect.fail("still-attached" as const)),
610 Effect.catchTag("ConsumerNotFound", () => Effect.void),
611 Effect.retry({
612 while: (e) => e === "still-attached",
613 schedule: Schedule.max([
614 Schedule.spaced("1 second"),
615 Schedule.recurs(30),
616 ]),
617 }),
618 Effect.ignore,
619 );
620 }),
621 read: Effect.fn(function* ({ output }) {
622 if (output?.consumerId) {
623 const fetched = yield* queues
624 .getConsumer({
625 accountId: output.accountId,
626 queueId: output.queueId,
627 consumerId: output.consumerId,
628 })
629 .pipe(
630 Effect.catchTag("ConsumerNotFound", () =>
631 Effect.succeed(undefined),
632 ),
633 );
634 if (fetched) {
635 return {
636 consumerId: fetched.consumerId!,
637 queueId: output.queueId,
638 scriptName: toObserved(fetched)?.scriptName ?? output.scriptName,
639 accountId: output.accountId,
640 deadLetterQueue: output.deadLetterQueue,
641 settings: output.settings,
642 };
643 }
644 }
645 // Fallback: a state loss can leave us without a consumerId
646 // even though the consumer is still alive on Cloudflare. The
647 // queue allows only one worker consumer, so finding it via
648 // listConsumers is unambiguous.
649 if (output?.queueId && output?.accountId) {
650 const match = yield* findWorkerConsumer(
651 output.accountId,
652 output.queueId,
653 );
654 if (match) {
655 return {
656 consumerId: match.consumerId,
657 queueId: output.queueId,
658 scriptName: match.scriptName ?? output.scriptName,
659 accountId: output.accountId,
660 deadLetterQueue: output.deadLetterQueue,
661 settings: output.settings,
662 };
663 }
664 }
665 return undefined;
666 }),
667 });
668
669type ObservedConsumer = {
670 consumerId: string;
671 scriptName: string | undefined;
672};
673
674const toObserved = (c: {
675 consumerId?: string | null;
676 scriptName?: string | null;
677 type?: "worker" | "http_pull" | null;
678}): ObservedConsumer | undefined =>
679 c.consumerId && c.type === "worker"
680 ? { consumerId: c.consumerId, scriptName: c.scriptName ?? undefined }
681 : undefined;
682
683// ~60s budget — Worker reconcile uploads typically land in 2–10s,
684// but a fresh container/asset deploy can stretch that.
685const queueHandlerReadinessSchedule = Schedule.max([
686 Schedule.spaced("2 seconds"),
687 Schedule.recurs(30),
688]);
689
690export const ConsumerProviderLocal = () =>
691 RpcProvider.effect(
692 Consumer,
693 LOCAL_PROVIDERS_URL,
694 Effect.gen(function* () {
695 const localRuntimeState = yield* LocalRuntimeState;
696
697 // Restart the locally running workerd instances (if any) for the
698 // given scripts so start-time queue-consumer wiring is re-read from
699 // `localRuntimeState.queueConsumers`. See `workerRestarts` on
700 // {@link LocalRuntimeState}.
701 const restartScripts = (scriptNames: ReadonlyArray<string>) =>
702 Effect.forEach(
703 scriptNames,
704 (scriptName) =>
705 MutableHashMap.get(
706 localRuntimeState.workerRestarts,
707 scriptName,
708 ).pipe(
709 Option.match({
710 onNone: () => Effect.void,
711 onSome: (restart) => restart,
712 }),
713 ),
714 { discard: true },
715 );
716
717 return {
718 list: () =>
719 Effect.sync(() =>
720 Array.from(MutableHashMap.values(localRuntimeState.queueConsumers)),
721 ),
722 diff: Effect.fn(function* ({ news, output }) {
723 if (!output) return { action: "update" };
724 // A real (non-`dev:`) consumerId on a local-mode row is legacy
725 // damage from pre-stamping dev runs — replace so the new
726 // generation mints a true local identity (delete best-effort
727 // detaches the stray live consumer).
728 if (isLiveId(output.consumerId)) {
729 return { action: "replace" };
730 }
731 if (!isResolved(news)) return undefined;
732 const accountId = isLiveId(news.queueId)
733 ? (yield* yield* CloudflareEnvironment).accountId
734 : yield* localAccountId;
735 if (
736 output.queueId !== news.queueId ||
737 output.accountId !== accountId
738 ) {
739 return { action: "replace" };
740 }
741 if (
742 JSON.stringify(output.settings ?? {}) !==
743 JSON.stringify(news.settings ?? {}) ||
744 output.scriptName !== news.scriptName ||
745 output.deadLetterQueue !== news.deadLetterQueue
746 ) {
747 return { action: "update" };
748 }
749 // If the resource is a noop, add it to the local runtime state so it's available downstream.
750 // We do it here instead of in the reconcile function so it doesn't appear as an update.
751 MutableHashMap.set(
752 localRuntimeState.queueConsumers,
753 output.consumerId,
754 output,
755 );
756 return { action: "noop" };
757 }),
758 read: Effect.fn(function* ({ output }) {
759 if (!output?.consumerId) return undefined;
760 return MutableHashMap.get(
761 localRuntimeState.queueConsumers,
762 output.consumerId,
763 ).pipe(Option.getOrUndefined);
764 }),
765 reconcile: Effect.fn(function* ({ news, output }) {
766 const accountId = isLiveId(news.queueId)
767 ? (yield* yield* CloudflareEnvironment).accountId
768 : yield* localAccountId;
769 // A LIVE queue (`Alchemy.remote()`) consumed by a LOCAL worker:
770 // Cloudflare only pushes to deployed consumers, so the local
771 // runtime drains the real queue via the HTTP pull API instead.
772 // Observe the queue (name + existing consumers) and ensure an
773 // `http_pull` consumer exists for the pull loop to lease from.
774 let queueName: string | undefined;
775 let pullConsumerId: string | undefined;
776 if (isLiveId(news.queueId)) {
777 const queue = yield* queues.getQueue({
778 accountId,
779 queueId: news.queueId,
780 });
781 queueName = queue.queueName ?? undefined;
782 const existingPull = (queue.consumers ?? []).find(
783 (c) => c.type === "http_pull",
784 );
785 if (existingPull?.consumerId) {
786 pullConsumerId = existingPull.consumerId;
787 } else {
788 const created = yield* queues.createConsumer({
789 accountId,
790 queueId: news.queueId,
791 type: "http_pull",
792 settings: {
793 batchSize: news.settings?.batchSize,
794 maxRetries: news.settings?.maxRetries,
795 retryDelay: news.settings?.retryDelay,
796 },
797 });
798 pullConsumerId =
799 ("consumerId" in created ? created.consumerId : undefined) ??
800 undefined;
801 }
802 }
803 const consumer: Consumer["Attributes"] = {
804 // Never carry a real (non-`dev:`) consumer id forward onto a
805 // local row — it belongs to a live consumer this row no longer
806 // manages (legacy pre-stamping damage).
807 consumerId:
808 output?.consumerId && !isLiveId(output.consumerId)
809 ? output.consumerId
810 : `dev:${crypto.randomUUID()}`,
811 queueId: news.queueId,
812 scriptName: news.scriptName,
813 deadLetterQueue: news.deadLetterQueue,
814 accountId,
815 settings: news.settings,
816 queueName,
817 pullConsumerId,
818 };
819 MutableHashMap.set(
820 localRuntimeState.queueConsumers,
821 consumer.consumerId,
822 consumer,
823 );
824 // A local workerd instance only reads its queue-consumer wiring
825 // at start time, and this reconcile races the sibling Worker's
826 // start (the Worker's `precreate` resolves `scriptName` before
827 // workerd is up). Restart any already-running instance so it
828 // picks up the new wiring; if the worker hasn't served yet this
829 // is a no-op and its first serve observes the map we just set.
830 yield* restartScripts(
831 output && output.scriptName !== consumer.scriptName
832 ? [output.scriptName, consumer.scriptName]
833 : [consumer.scriptName],
834 );
835 return consumer;
836 }),
837 delete: Effect.fn(function* ({ output }) {
838 MutableHashMap.remove(
839 localRuntimeState.queueConsumers,
840 output.consumerId,
841 );
842 yield* restartScripts([output.scriptName]);
843 // Legacy local-mode rows written before providerMode stamping can
844 // carry a real consumer's id — detach the live consumer too so
845 // migrating the row to a true local identity doesn't strand it on
846 // the queue (a stranded worker consumer blocks every future
847 // createConsumer with ConsumerAlreadyExists).
848 if (isLiveId(output.consumerId) && isLiveId(output.queueId)) {
849 yield* queues
850 .deleteConsumer({
851 accountId: output.accountId,
852 queueId: output.queueId,
853 consumerId: output.consumerId,
854 })
855 .pipe(
856 Effect.catchTag(
857 ["ConsumerNotFound", "QueueNotFound"],
858 () => Effect.void,
859 ),
860 );
861 }
862 // Remove the http_pull consumer this row attached to its live
863 // queue. Idempotent: gone-already (or queue deleted first) is
864 // success.
865 if (output.pullConsumerId && isLiveId(output.queueId)) {
866 yield* queues
867 .deleteConsumer({
868 accountId: output.accountId,
869 queueId: output.queueId,
870 consumerId: output.pullConsumerId,
871 })
872 .pipe(
873 Effect.catchTag(
874 ["ConsumerNotFound", "QueueNotFound"],
875 () => Effect.void,
876 ),
877 );
878 }
879 }),
880 };
881 }),
882 );
883
884export const ConsumerProvider = () =>
885 ProviderLayer.dual(Consumer, {
886 local: () =>
887 ConsumerProviderLocal().pipe(Layer.provide(localRuntimeServices())),
888 live: () => ConsumerProviderLive(),
889 });