[Miniflare 3] Implement Queues #566

mrbbot · 2023-05-04T12:20:20Z

This PR adds support for Queues to Miniflare 3. It builds on cloudflare/workerd#541, cloudflare/workerd#543, and cloudflare/workerd#545 (thanks @a-robinson!).

cloudflare/workerd#545 adds a queue() method on service bindings to dispatch queue events to Workers. It expects regular, de-serialised JavaScript objects, but the Queue producer bindings from cloudflare/workerd#541 send V8-serialised objects instead. Unfortunately, workerd doesn't expose the V8 (de)serialiser in Workers yet, so we need to re-serialise all messages into a format we can deserialise in a Worker. We use devalue for this. This is the first time we're using an npm dependency in an internal Miniflare Worker. To support this, and because it's something we've been wanting to do for a while, the core entrypoint Worker has been moved to a separate TypeScript file, which is built as an additional entrypoint with esbuild, and type-checked under @cloudflare/workers-types. This is a separate commit to aid reviewing. We'll migrate other internal Workers to this format soon.

Another minor issue is that workerd uses V8 serialisation version 15 when sending messages. This is only supported by V8 versions 10.0.29 and above. For reference, the V8 versions associated with notable Node versions are:

Miniflare's minimum supported version: Node 16.13.0 --> V8 9.4
Last Node 17/unsupported version: Node 17.9.1 --> V8 9.6
First supported version: Node 18.0.0 --> V8 10.1

This means that Queues will only be supported in Miniflare 3 if you're using Node 18 or above. I think this is ok, as I'm pretty sure Queues is still in beta, and it was previously marked experimental in Miniflare 2.

I've added some other questions that would be good to get opinions on too. 🙂

changeset-bot · 2023-05-04T12:20:25Z

⚠️ No Changeset found

Latest commit: 7fa2b72

Merging this PR will not cause a version bump for any packages. If these changes should not result in a new version, you're good to go. If these changes should result in a version bump, you need to add a changeset.

Click here to learn what changesets are, and how to add one.

Click here if you're a maintainer who wants to add a changeset to this PR

mrbbot · 2023-05-09T15:59:08Z

packages/tre/src/index.ts

+  for (const [queueName, consumer] of queueConsumers) {
+    if (consumer.deadLetterQueue !== undefined) {
+      // Check the dead letter queue isn't configured to be the queue itself
+      // (NOTE: Queues *does* permit DLQ cycles between multiple queues,


Queues does permit dead-letter queue cycles between multiple queues, but is this something we want to block in Miniflare? Are developers ever likely to need this?

mrbbot · 2023-05-09T16:01:57Z

packages/tre/src/plugins/queues/gateway.ts

+  else if (acked > 0) colour = yellow;
+  else colour = red;
+
+  let message = `${bold("QUEUE")} ${queueName} ${colour(`${acked}/${total}`)}`;


Should this log include anything else?

mrbbot · 2023-05-09T16:04:40Z

packages/tre/src/plugins/queues/gateway.ts

+    enqueueOn: QueueEnqueueOn,
+    consumer: QueueConsumer,


Ideally, we wouldn't need to pass enqueueOn and consumer each time. The problem is that gateways are created once and then reused across the lifetime of a Miniflare instance. If the consumer ever changes from for example a setOptions() call, we need to respect that. I wonder if there's a better way of doing this... 🤔

mrbbot · 2023-05-09T16:05:57Z

packages/tre/src/plugins/queues/index.ts

+  queueProducers: z
+    .union([z.record(z.string()), z.string().array()])
+    .optional(),
+  queueConsumers: z
+    .union([z.record(QueueConsumerOptionsSchema), z.string().array()])
+    .optional(),


This is a slightly different API to Miniflare 2...

miniflare/packages/queues/src/plugin.ts

Lines 16 to 32 in e3673a5

export interface BindingOptions {

name: string;

queueName: string;

}

export interface ConsumerOptions {

queueName: string;

maxBatchSize?: number;

maxWaitMs?: number;

maxRetries?: number;

deadLetterQueue?: string;

}

export interface QueuesOptions {

queueBindings?: BindingOptions[];

queueConsumers?: (string | ConsumerOptions)[];

}

...but I think it more closely matches the rest of Miniflare's options. Queues was previously marked as experimental, so I think this change is ok, but open to suggestions.

mrbbot · 2023-05-09T16:06:43Z

packages/tre/src/plugins/queues/index.ts

+export const _QUEUES_COMPATIBLE_V8_VERSION =
+  semiver(process.versions.v8, "10.0.29") >= 0;
+
+function assertCompatibleV8Version() {


Ideally, we'd just bump Miniflare 3's minimum supported Node version to 18, but I'm not sure if we want to do that for Wrangler 3 too?

If it's required for Queues...

packages/tre/src/workers/core/devalue.ts

penalosa

Looking good! How does this work with Wrangler? i.e. if we don't have both workers available as source

packages/tre/src/plugins/core/errors/index.ts

packages/tre/src/plugins/core/index.ts

packages/tre/src/workers/core/constants.ts

packages/tre/src/plugins/shared/constants.ts

packages/tre/src/runtime/config/workerd.capnp.d.ts

packages/tre/src/runtime/config/workerd.capnp.js

packages/tre/src/workers/core/devalue.ts

penalosa · 2023-05-11T19:50:28Z

packages/tre/src/workers/core/entry.worker.ts

+// @ts-expect-error "devalue" is ESM-only, so TypeScript requires us to use the
+//  `*.mts` extension, or set `"type": "module"` in our `package.json`. We can't
+//  do the first as ambient types from `@cloudflare/workers-types` don't seem to
+//  work, and the second would affect all consumers of `miniflare`.
+import { unflatten } from "devalue";


https://devblogs.microsoft.com/typescript/announcing-typescript-5-0-beta/#moduleresolution-bundler should fix this if the Typescript version can be upgraded to 5?

Oh nice! Doesn't look like api-extractor supports TypeScript 5 yet, but we should definitely switch to this when we upgrade. Alternatively, we just scrap API extractor and ship multiple *.d.ts files for our single bundled *.js file?

What would that look like? What's the benefit of api extractor over that approach?

api-extractor just means we have a single index.d.ts file corresponding to our index.js bundle. Without it, the structure of our src folder would make it to the final bundle as TypeScript definitions file. Essentially, I think you'd be able to do stuff like import { ... } from "@miniflare/tre/plugins/core" as plugins/core/index.d.ts would exist, but the code for it would be in the index.js bundle, so it wouldn't actually work.

packages/tre/test/plugins/queues/index.spec.ts

mrbbot · 2023-05-12T01:27:51Z

How does this work with Wrangler? i.e. if we don't have both workers available as source

Are you referring to the bundling of Miniflare into Wrangler here? I think the esbuild plugin for building Workers should alleviate this issue.

penalosa · 2023-05-12T10:33:48Z

No, I was wondering how this works if Wrangler only has access to one user Worker—do the queue producer and consumer have to be defined in the same script?

mrbbot · 2023-05-12T10:57:46Z

Ah, yeah they would have to be. We could probably do something similar to remote Durable Objects by defining a fake queue handler that proxies to another process which then sends the messages to the correct handler.

mrbbot · 2023-05-12T12:16:15Z

Squashing...

mrbbot · 2023-05-12T12:54:28Z

Hmmm, that Windows test failure on Node 20 looks like a real issue... 🙁 Time to debug...

mrbbot · 2023-05-12T16:03:44Z

Potential fix here: cloudflare/workerd#644

mrbbot · 2023-05-12T16:24:07Z

I think I'm going to merge this with broken tests... The fix for this will require a workerd release, and upgrading Miniflare to support some of the other changes in that (e.g. R2 multiple etag support) requires separate changes. This will also unblock reviewing the other PRs that depend on this change. 👍

mrbbot added the tre Relating to Miniflare 3 label May 4, 2023

mrbbot force-pushed the tre-new-storage-r2 branch from 1c413ad to 52c7c01 Compare May 9, 2023 14:17

Base automatically changed from tre-new-storage-r2 to tre May 9, 2023 14:20

mrbbot force-pushed the tre-queues branch 3 times, most recently from d23c667 to b6ce395 Compare May 9, 2023 15:42

mrbbot commented May 9, 2023

View reviewed changes

mrbbot requested a review from a team May 9, 2023 16:14

mrbbot marked this pull request as ready for review May 9, 2023 16:14

penalosa reviewed May 11, 2023

View reviewed changes

mrbbot requested a review from penalosa May 12, 2023 01:11

penalosa approved these changes May 12, 2023

View reviewed changes

mrbbot added 2 commits May 12, 2023 13:15

Extract entrypoint worker into separate file

2998365

Implement Queues

7fa2b72

mrbbot force-pushed the tre-queues branch from 244e3f5 to 7fa2b72 Compare May 12, 2023 12:16

mrbbot merged commit 5268fc0 into tre May 12, 2023
7 of 8 checks passed

mrbbot deleted the tre-queues branch May 12, 2023 16:24

mrbbot mentioned this pull request Jun 28, 2023

[Miniflare 3] Update API docs with routes, queue*, https*, and upstream #618

Merged

mrbbot mentioned this pull request Jun 30, 2023

[Miniflare 3] Minor API docs fixes #621

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Miniflare 3] Implement Queues #566

[Miniflare 3] Implement Queues #566

mrbbot commented May 4, 2023 •

edited

changeset-bot bot commented May 4, 2023 •

edited

mrbbot May 9, 2023

mrbbot May 9, 2023

mrbbot May 9, 2023

mrbbot May 9, 2023

mrbbot May 9, 2023

penalosa May 11, 2023

penalosa left a comment

penalosa May 11, 2023

mrbbot May 12, 2023

penalosa May 12, 2023

mrbbot May 12, 2023 •

edited

mrbbot commented May 12, 2023

penalosa commented May 12, 2023

mrbbot commented May 12, 2023

mrbbot commented May 12, 2023

mrbbot commented May 12, 2023

mrbbot commented May 12, 2023

mrbbot commented May 12, 2023

	export interface BindingOptions {
	name: string;
	queueName: string;
	}

	export interface ConsumerOptions {
	queueName: string;
	maxBatchSize?: number;
	maxWaitMs?: number;
	maxRetries?: number;
	deadLetterQueue?: string;
	}

	export interface QueuesOptions {
	queueBindings?: BindingOptions[];
	queueConsumers?: (string \| ConsumerOptions)[];
	}

[Miniflare 3] Implement Queues #566

[Miniflare 3] Implement Queues #566

Conversation

mrbbot commented May 4, 2023 • edited

changeset-bot bot commented May 4, 2023 • edited

⚠️ No Changeset found

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

penalosa left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mrbbot May 12, 2023 • edited

Choose a reason for hiding this comment

mrbbot commented May 12, 2023

penalosa commented May 12, 2023

mrbbot commented May 12, 2023

mrbbot commented May 12, 2023

mrbbot commented May 12, 2023

mrbbot commented May 12, 2023

mrbbot commented May 12, 2023

mrbbot commented May 4, 2023 •

edited

changeset-bot bot commented May 4, 2023 •

edited

mrbbot May 12, 2023 •

edited