feat(MQ-1202): include metrics metadata in queue() handler message batch

Author: KennethRuan

src/workerd/api/http.c++

@@ -2230,8 +2230,10 @@ jsg::Promise<void> Fetcher::delete_(jsg::Lock& js, kj::String url) {
   return throwOnError(js, "DELETE", fetchImpl(js, JSG_THIS, kj::mv(url), kj::mv(subInit)));
 }
 
-jsg::Promise<Fetcher::QueueResult> Fetcher::queue(
-    jsg::Lock& js, kj::String queueName, kj::Array<ServiceBindingQueueMessage> messages) {
+jsg::Promise<Fetcher::QueueResult> Fetcher::queue(jsg::Lock& js,
+    kj::String queueName,
+    kj::Array<ServiceBindingQueueMessage> messages,
+    jsg::Optional<MessageBatchMetadata> metadata) {
   auto& ioContext = IoContext::current();
 
   auto encodedMessages = kj::heapArrayBuilder<IncomingQueueMessage>(messages.size());
@@ -2264,6 +2266,7 @@ jsg::Promise<Fetcher::QueueResult> Fetcher::queue(
   auto event = kj::refcounted<api::QueueCustomEvent>(QueueEvent::Params{
     .queueName = kj::mv(queueName),
     .messages = encodedMessages.finish(),
+    .metadata = kj::mv(metadata).orDefault({}),
   });
 
   auto eventRef =

src/workerd/api/http.h

@@ -392,8 +392,10 @@ class Fetcher: public JsRpcClientProvider {
     JSG_STRUCT(outcome, ackAll, retryBatch, explicitAcks, retryMessages);
   };
 
-  jsg::Promise<QueueResult> queue(
-      jsg::Lock& js, kj::String queueName, kj::Array<ServiceBindingQueueMessage> messages);
+  jsg::Promise<QueueResult> queue(jsg::Lock& js,
+      kj::String queueName,
+      kj::Array<ServiceBindingQueueMessage> messages,
+      jsg::Optional<MessageBatchMetadata> metadata);
 
   struct ScheduledOptions {
     jsg::Optional<kj::Date> scheduledTime;
@@ -446,7 +448,7 @@ class Fetcher: public JsRpcClientProvider {
       ) & {
         fetch(input: RequestInfo | URL, init?: RequestInit): Promise<Response>;
         connect(address: SocketAddress | string, options?: SocketOptions): Socket;
-        queue(queueName: string, messages: ServiceBindingQueueMessage[]): Promise<FetcherQueueResult>;
+        queue(queueName: string, messages: ServiceBindingQueueMessage[], metadata?: MessageBatchMetadata): Promise<FetcherQueueResult>;
         scheduled(options?: FetcherScheduledOptions): Promise<FetcherScheduledResult>;
       });
     } else {

src/workerd/api/queue.c++

@@ -445,11 +445,23 @@ QueueEvent::QueueEvent(
     messagesBuilder.add(js.alloc<QueueMessage>(js, incoming[i], result));
   }
   messages = messagesBuilder.finish();
+
+  // Extract metadata. If the sender didn't set the field, capnp defaults all values to zero.
+  auto m = params.getMetadata().getMetrics();
+  metadata = MessageBatchMetadata{
+    .metrics =
+        MessageBatchMetrics{
+          .backlogCount = m.getBacklogCount(),
+          .backlogBytes = m.getBacklogBytes(),
+          .oldestMessageTimestamp = m.getOldestMessageTimestamp(),
+        },
+  };
 }
 
 QueueEvent::QueueEvent(jsg::Lock& js, Params params, IoPtr<QueueEventResult> result)
     : ExtendableEvent("queue"),
       queueName(kj::mv(params.queueName)),
+      metadata(kj::mv(params.metadata)),
       result(result) {
   auto messagesBuilder = kj::heapArrayBuilder<jsg::Ref<QueueMessage>>(params.messages.size());
   for (auto i: kj::indices(params.messages)) {
@@ -742,6 +754,7 @@ kj::Promise<WorkerInterface::CustomEvent::Result> QueueCustomEvent::sendRpc(
     KJ_CASE_ONEOF(p, rpc::EventDispatcher::QueueParams::Reader) {
       req.setQueueName(p.getQueueName());
       req.setMessages(p.getMessages());
+      req.setMetadata(p.getMetadata());
     }
     KJ_CASE_ONEOF(p, QueueEvent::Params) {
       req.setQueueName(p.queueName);
@@ -755,6 +768,13 @@ kj::Promise<WorkerInterface::CustomEvent::Result> QueueCustomEvent::sendRpc(
         }
         messages[i].setAttempts(p.messages[i].attempts);
       }
+      {
+        auto metadataBuilder = req.initMetadata();
+        auto metricsBuilder = metadataBuilder.initMetrics();
+        metricsBuilder.setBacklogCount(p.metadata.metrics.backlogCount);
+        metricsBuilder.setBacklogBytes(p.metadata.metrics.backlogBytes);
+        metricsBuilder.setOldestMessageTimestamp(p.metadata.metrics.oldestMessageTimestamp);
+      }
     }
   }
 

src/workerd/api/queue.h

@@ -93,6 +93,22 @@ class WorkerQueue: public jsg::Object {
 
 // Event handler types
 
+// Metadata delivered with a message batch in the queue() handler
+
+struct MessageBatchMetrics {
+  double backlogCount;
+  double backlogBytes;
+  double oldestMessageTimestamp;
+  JSG_STRUCT(backlogCount, backlogBytes, oldestMessageTimestamp);
+  JSG_STRUCT_TS_OVERRIDE(MessageBatchMetrics);
+};
+
+struct MessageBatchMetadata {
+  MessageBatchMetrics metrics;
+  JSG_STRUCT(metrics);
+  JSG_STRUCT_TS_OVERRIDE(MessageBatchMetadata);
+};
+
 // Types for other workers passing messages into and responses out of a queue handler.
 
 struct IncomingQueueMessage {
@@ -212,6 +228,7 @@ class QueueEvent final: public ExtendableEvent {
   struct Params {
     kj::String queueName;
     kj::Array<IncomingQueueMessage> messages;
+    MessageBatchMetadata metadata;
   };
 
   explicit QueueEvent(jsg::Lock& js,
@@ -227,23 +244,37 @@ class QueueEvent final: public ExtendableEvent {
   kj::StringPtr getQueueName() {
     return queueName;
   }
+  MessageBatchMetadata getMetadata() {
+    return metadata;
+  }
 
   void retryAll(jsg::Optional<QueueRetryOptions> options);
   void ackAll();
 
-  JSG_RESOURCE_TYPE(QueueEvent) {
+  JSG_RESOURCE_TYPE(QueueEvent, CompatibilityFlags::Reader flags) {
     JSG_INHERIT(ExtendableEvent);
 
     JSG_LAZY_READONLY_INSTANCE_PROPERTY(messages, getMessages);
     JSG_READONLY_INSTANCE_PROPERTY(queue, getQueueName);
 
+    if (flags.getQueueBatchMetadata()) {
+      JSG_READONLY_INSTANCE_PROPERTY(metadata, getMetadata);
+    }
+
     JSG_METHOD(retryAll);
     JSG_METHOD(ackAll);
 
     JSG_TS_ROOT();
-    JSG_TS_OVERRIDE(QueueEvent<Body = unknown> {
-        readonly messages: readonly Message<Body>[];
-    });
+    if (flags.getQueueBatchMetadata()) {
+      JSG_TS_OVERRIDE(QueueEvent<Body = unknown> {
+          readonly messages: readonly Message<Body>[];
+          readonly metadata: MessageBatchMetadata;
+      });
+    } else {
+      JSG_TS_OVERRIDE(QueueEvent<Body = unknown> {
+          readonly messages: readonly Message<Body>[];
+      });
+    }
   }
 
   void visitForMemoryInfo(jsg::MemoryTracker& tracker) const {
@@ -274,6 +305,7 @@ class QueueEvent final: public ExtendableEvent {
   // array to avoid one intermediate copy?
   kj::Array<jsg::Ref<QueueMessage>> messages;
   kj::String queueName;
+  MessageBatchMetadata metadata;
   IoPtr<QueueEventResult> result;
   CompletionStatus completionStatus = Incomplete{};
 
@@ -293,24 +325,38 @@ class QueueController final: public jsg::Object {
   kj::StringPtr getQueueName() {
     return event->getQueueName();
   }
+  MessageBatchMetadata getMetadata() {
+    return event->getMetadata();
+  }
   void retryAll(jsg::Optional<QueueRetryOptions> options) {
     event->retryAll(options);
   }
   void ackAll() {
     event->ackAll();
   }
 
-  JSG_RESOURCE_TYPE(QueueController) {
+  JSG_RESOURCE_TYPE(QueueController, CompatibilityFlags::Reader flags) {
     JSG_READONLY_INSTANCE_PROPERTY(messages, getMessages);
     JSG_READONLY_INSTANCE_PROPERTY(queue, getQueueName);
 
+    if (flags.getQueueBatchMetadata()) {
+      JSG_READONLY_INSTANCE_PROPERTY(metadata, getMetadata);
+    }
+
     JSG_METHOD(retryAll);
     JSG_METHOD(ackAll);
 
     JSG_TS_ROOT();
-    JSG_TS_OVERRIDE(MessageBatch<Body = unknown> {
-      readonly messages: readonly Message<Body>[];
-    });
+    if (flags.getQueueBatchMetadata()) {
+      JSG_TS_OVERRIDE(MessageBatch<Body = unknown> {
+        readonly messages: readonly Message<Body>[];
+        readonly metadata: MessageBatchMetadata;
+      });
+    } else {
+      JSG_TS_OVERRIDE(MessageBatch<Body = unknown> {
+        readonly messages: readonly Message<Body>[];
+      });
+    }
   }
 
   void visitForMemoryInfo(jsg::MemoryTracker& tracker) const {
@@ -377,8 +423,9 @@ class QueueCustomEvent final: public WorkerInterface::CustomEvent, public kj::Re
 
 #define EW_QUEUE_ISOLATE_TYPES                                                                     \
   api::WorkerQueue, api::WorkerQueue::SendOptions, api::WorkerQueue::SendBatchOptions,             \
-      api::WorkerQueue::MessageSendRequest, api::IncomingQueueMessage, api::QueueRetryBatch,       \
-      api::QueueRetryMessage, api::QueueResponse, api::QueueRetryOptions, api::QueueMessage,       \
-      api::QueueEvent, api::QueueController, api::QueueExportedHandler
+      api::WorkerQueue::MessageSendRequest, api::MessageBatchMetrics, api::MessageBatchMetadata,   \
+      api::IncomingQueueMessage, api::QueueRetryBatch, api::QueueRetryMessage, api::QueueResponse, \
+      api::QueueRetryOptions, api::QueueMessage, api::QueueEvent, api::QueueController,            \
+      api::QueueExportedHandler
 
 }  // namespace workerd::api

src/workerd/api/tests/BUILD.bazel

@@ -119,6 +119,12 @@ wd_test(
     ],
 )
 
+wd_test(
+    src = "queue-metadata-test.wd-test",
+    args = ["--experimental"],
+    data = ["queue-metadata-test.js"],
+)
+
 wd_test(
     src = "r2-test.wd-test",
     args = ["--experimental"],

src/workerd/api/tests/queue-metadata-test.js

@@ -0,0 +1,76 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the Apache 2.0 license found in the LICENSE file or at:
+//     https://opensource.org/licenses/Apache-2.0
+
+import assert from 'node:assert';
+
+export default {
+  async queue(batch, env, ctx) {
+    const flagEnabled = env.METADATA_FLAG;
+
+    if (!flagEnabled) {
+      // Flag disabled → metadata property should not exist
+      assert.strictEqual(batch.metadata, undefined);
+      batch.ackAll();
+      return;
+    }
+
+    // Flag enabled → metadata should always be present
+    assert.ok(batch.metadata, 'Expected batch.metadata to be defined');
+    assert.ok(
+      batch.metadata.metrics,
+      'Expected batch.metadata.metrics to be defined'
+    );
+
+    if (
+      batch.metadata.metrics.backlogCount === 0 &&
+      batch.metadata.metrics.backlogBytes === 0 &&
+      batch.metadata.metrics.oldestMessageTimestamp === 0
+    ) {
+      // If metadata is omitted → all values default to zero
+      batch.ackAll();
+      return;
+    }
+
+    // Explicit metadata path
+    assert.strictEqual(batch.metadata.metrics.backlogCount, 100);
+    assert.strictEqual(batch.metadata.metrics.backlogBytes, 2048);
+    assert.strictEqual(batch.metadata.metrics.oldestMessageTimestamp, 1000000);
+    batch.ackAll();
+  },
+
+  async test(ctrl, env, ctx) {
+    const flagEnabled = env.METADATA_FLAG;
+    const timestamp = new Date();
+
+    if (flagEnabled) {
+      const response1 = await env.SERVICE.queue(
+        'test-queue',
+        [{ id: '0', timestamp, body: 'hello', attempts: 1 }],
+        {
+          metrics: {
+            backlogCount: 100,
+            backlogBytes: 2048,
+            oldestMessageTimestamp: 1000000,
+          },
+        }
+      );
+      assert.strictEqual(response1.outcome, 'ok');
+      assert(response1.ackAll);
+
+      // Test with omitted metadata
+      const response2 = await env.SERVICE.queue('test-queue', [
+        { id: '1', timestamp, body: 'world', attempts: 1 },
+      ]);
+      assert.strictEqual(response2.outcome, 'ok');
+      assert(response2.ackAll);
+    } else {
+      // Flag disabled → handler still works, metadata not visible
+      const response = await env.SERVICE.queue('test-queue', [
+        { id: '0', timestamp, body: 'foobar', attempts: 1 },
+      ]);
+      assert.strictEqual(response.outcome, 'ok');
+      assert(response.ackAll);
+    }
+  },
+};

src/workerd/api/tests/queue-metadata-test.wd-test

@@ -0,0 +1,30 @@
+using Workerd = import "/workerd/workerd.capnp";
+
+const unitTests :Workerd.Config = (
+  services = [
+    ( name = "queue-metadata-test",
+      worker = (
+        modules = [
+          ( name = "worker", esModule = embed "queue-metadata-test.js" )
+        ],
+        bindings = [
+          ( name = "SERVICE", service = "queue-metadata-test" ),
+          ( name = "METADATA_FLAG", json = "true" ),
+        ],
+        compatibilityFlags = ["nodejs_compat", "service_binding_extra_handlers", "queue_batch_metadata"],
+      )
+    ),
+    ( name = "queue-metadata-disabled-test",
+      worker = (
+        modules = [
+          ( name = "worker-disabled", esModule = embed "queue-metadata-test.js" )
+        ],
+        bindings = [
+          ( name = "SERVICE", service = "queue-metadata-disabled-test" ),
+          ( name = "METADATA_FLAG", json = "false" ),
+        ],
+        compatibilityFlags = ["nodejs_compat", "service_binding_extra_handlers", "no_queue_batch_metadata"],
+      )
+    ),
+  ],
+);

src/workerd/io/compatibility-date.capnp

@@ -1502,4 +1502,12 @@ struct CompatibilityFlags @0x8f8c1b68151b6cef {
   # When paired with `enable_version_api`, also exposes `ctx.version.metadata`. This is a separate
   # flag as we haven't decided on the exact behaviour of `ctx.version.metadata`, but the rest of
   # `ctx.version` is much more well defined. The behaviour of this flag will change in the future.
+
+  queueBatchMetadata @173 :Bool
+      $compatEnableFlag("queue_batch_metadata")
+      $compatDisableFlag("no_queue_batch_metadata")
+      $experimental;
+  # Enables the `metadata` property on MessageBatch (QueueController / QueueEvent),
+  # which provides queue metrics (backlog count, bytes, oldest message timestamp)
+  # delivered inline with each queue batch from the queue broker.
 }

src/workerd/io/worker-interface.capnp

@@ -366,6 +366,20 @@ struct QueueResponse @0x90e98932c0bfc0de {
   # List of retry options for messages that were explicitly marked for retry.
 }
 
+struct MessageBatchMetrics {
+  backlogCount @0 :Float64;
+  # Number of messages remaining in the queue backlog.
+  backlogBytes @1 :Float64;
+  # Total bytes of messages remaining in the queue backlog.
+  oldestMessageTimestamp @2 :Float64;
+  # Timestamp (ms since epoch) of the oldest message in the queue.
+}
+
+struct MessageBatchMetadata {
+  metrics @0 :MessageBatchMetrics;
+  # Best effort queue metrics at the time the batch was dispatched.
+}
+
 struct HibernatableWebSocketEventMessage {
   payload :union {
     text @0 :Text;
@@ -755,11 +769,13 @@ interface EventDispatcher @0xf20697475ec1752d {
   #   It would be cleaner to handle that inside the implementation so we could mark the entire
   #   interface (and file) with allowCancellation.
 
-  queue @8 (messages :List(QueueMessage), queueName :Text) -> (result :QueueResponse)
+  queue @8 (messages :List(QueueMessage), queueName :Text, metadata :MessageBatchMetadata)
+      -> (result :QueueResponse)
       $Cxx.allowCancellation;
   # Delivers a batch of queue messages to a worker's queue event handler. Returns information about
   # the success of the batch, including which messages should be considered acknowledged and which
-  # should be retried.
+  # should be retried. The optional metadata field carries queue metrics at the time the batch was
+  # dispatched; it is safe for the sender to omit this field (the consumer sees it as absent).
 
   jsRpcSession @9 () -> (topLevel :JsRpcTarget) $Cxx.allowCancellation;
   # Opens a JS rpc "session". The call does not return until the session is complete.