Change Queues oldestMessageTimestamp to use kj::Date

Author: KennethRuan

src/workerd/api/queue.c++

@@ -23,6 +23,16 @@ namespace {
 // Header for the message format.
 static constexpr kj::StringPtr HDR_MSG_FORMAT = "X-Msg-Fmt"_kj;
 
+// The upstream service sends 0 when there is "no data" available on a timestamp field (e.g. no `oldestMessageTimestamp`).
+// This method converts it to kj::none so users see `undefined`.
+void clearEpochSentinel(jsg::Optional<kj::Date>& ts) {
+  KJ_IF_SOME(date, ts) {
+    if (date == kj::UNIX_EPOCH) {
+      ts = kj::none;
+    }
+  }
+}
+
 // Header for the message delivery delay.
 static constexpr kj::StringPtr HDR_MSG_DELAY = "X-Msg-Delay-Secs"_kj;
 
@@ -306,8 +316,10 @@ jsg::Promise<WorkerQueue::SendResponse> WorkerQueue::sendWithResponse(jsg::Lock&
   return context.awaitIo(
       js, kj::mv(promise), [&responseHandler](jsg::Lock& js, kj::String text) -> SendResponse {
     auto parsed = jsg::JsValue::fromJson(js, text);
-    return JSG_REQUIRE_NONNULL(
+    auto result = JSG_REQUIRE_NONNULL(
         responseHandler.tryUnwrap(js, parsed), Error, "Failed to parse queue send response", text);
+    clearEpochSentinel(result.metadata.metrics.oldestMessageTimestamp);
+    return kj::mv(result);
   });
 }
 
@@ -456,6 +468,7 @@ jsg::Promise<WorkerQueue::Metrics> WorkerQueue::metrics(
     auto parsed = jsg::JsValue::fromJson(js, text);
     auto result = JSG_REQUIRE_NONNULL(metricsHandler.tryUnwrap(js, parsed), Error,
         "Failed to parse queue metrics response", text);
+    clearEpochSentinel(result.oldestMessageTimestamp);
     return kj::mv(result);
   });
 }
@@ -569,8 +582,10 @@ jsg::Promise<WorkerQueue::SendBatchResponse> WorkerQueue::sendBatchWithResponse(
   return context.awaitIo(
       js, kj::mv(promise), [&responseHandler](jsg::Lock& js, kj::String text) -> SendBatchResponse {
     auto parsed = jsg::JsValue::fromJson(js, text);
-    return JSG_REQUIRE_NONNULL(
+    auto result = JSG_REQUIRE_NONNULL(
         responseHandler.tryUnwrap(js, parsed), Error, "Failed to parse queue send response", text);
+    clearEpochSentinel(result.metadata.metrics.oldestMessageTimestamp);
+    return kj::mv(result);
   });
 }
 
@@ -658,14 +673,19 @@ QueueEvent::QueueEvent(
   }
   messages = messagesBuilder.finish();
 
-  // Extract metadata. If the sender didn't set the field, capnp defaults all values to zero.
+  // Extract metadata. If the sender didn't set the field, capnp defaults all to the zero values.
   auto m = params.getMetadata().getMetrics();
+  jsg::Optional<kj::Date> oldestTimestamp;
+  if (m.getOldestMessageTimestamp() != 0) {
+    oldestTimestamp =
+        kj::UNIX_EPOCH + static_cast<int64_t>(m.getOldestMessageTimestamp()) * kj::MILLISECONDS;
+  }
   metadata = MessageBatchMetadata{
     .metrics =
         MessageBatchMetrics{
           .backlogCount = m.getBacklogCount(),
           .backlogBytes = m.getBacklogBytes(),
-          .oldestMessageTimestamp = m.getOldestMessageTimestamp(),
+          .oldestMessageTimestamp = kj::mv(oldestTimestamp),
         },
   };
 }
@@ -985,7 +1005,9 @@ kj::Promise<WorkerInterface::CustomEvent::Result> QueueCustomEvent::sendRpc(
         auto metricsBuilder = metadataBuilder.initMetrics();
         metricsBuilder.setBacklogCount(p.metadata.metrics.backlogCount);
         metricsBuilder.setBacklogBytes(p.metadata.metrics.backlogBytes);
-        metricsBuilder.setOldestMessageTimestamp(p.metadata.metrics.oldestMessageTimestamp);
+        KJ_IF_SOME(ts, p.metadata.metrics.oldestMessageTimestamp) {
+          metricsBuilder.setOldestMessageTimestamp((ts - kj::UNIX_EPOCH) / kj::MILLISECONDS);
+        }
       }
     }
   }

src/workerd/api/queue.h

@@ -27,18 +27,22 @@ class WorkerQueue: public jsg::Object {
   // representing this queue.
   WorkerQueue(uint subrequestChannel): subrequestChannel(subrequestChannel) {}
 
+  // The metrics structs below (Metrics, SendMetrics, SendBatchMetrics) are deserialized from
+  // JSON responses where the upstream service uses 0 as a sentinel for "no data" on timestamp
+  // fields. Callers MUST call clearEpochSentinel() on oldestMessageTimestamp after deserialization to convert the
+  // sentinel to kj::none (JS undefined).
   struct Metrics {
     double backlogCount;
     double backlogBytes;
-    double oldestMessageTimestamp;
+    jsg::Optional<kj::Date> oldestMessageTimestamp;
     JSG_STRUCT(backlogCount, backlogBytes, oldestMessageTimestamp);
     JSG_STRUCT_TS_OVERRIDE(QueueMetrics);
   };
 
   struct SendMetrics {
     double backlogCount;
     double backlogBytes;
-    double oldestMessageTimestamp;
+    jsg::Optional<kj::Date> oldestMessageTimestamp;
     JSG_STRUCT(backlogCount, backlogBytes, oldestMessageTimestamp);
     JSG_STRUCT_TS_OVERRIDE(QueueSendMetrics);
   };
@@ -58,7 +62,7 @@ class WorkerQueue: public jsg::Object {
   struct SendBatchMetrics {
     double backlogCount;
     double backlogBytes;
-    double oldestMessageTimestamp;
+    jsg::Optional<kj::Date> oldestMessageTimestamp;
     JSG_STRUCT(backlogCount, backlogBytes, oldestMessageTimestamp);
     JSG_STRUCT_TS_OVERRIDE(QueueSendBatchMetrics);
   };
@@ -171,10 +175,12 @@ class WorkerQueue: public jsg::Object {
 
 // Metadata delivered with a message batch in the queue() handler
 
+// Same sentinel caveat as WorkerQueue::Metrics above: the capnp path uses 0 to mean "no data"
+// for oldestMessageTimestamp. As such, we must explicitly set it to kj::none (JS undefined).
 struct MessageBatchMetrics {
   double backlogCount;
   double backlogBytes;
-  double oldestMessageTimestamp;
+  jsg::Optional<kj::Date> oldestMessageTimestamp;
   JSG_STRUCT(backlogCount, backlogBytes, oldestMessageTimestamp);
   JSG_STRUCT_TS_OVERRIDE(MessageBatchMetrics);
 };

src/workerd/api/tests/BUILD.bazel

@@ -164,6 +164,12 @@ wd_test(
     data = ["queue-metadata-test.js"],
 )
 
+wd_test(
+    src = "queue-metrics-sentinel-test.wd-test",
+    args = ["--experimental"],
+    data = ["queue-metrics-sentinel-test.js"],
+)
+
 wd_test(
     src = "queue-producer-metadata-test.wd-test",
     args = ["--experimental"],

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

@@ -25,17 +25,24 @@ export default {
     if (
       batch.metadata.metrics.backlogCount === 0 &&
       batch.metadata.metrics.backlogBytes === 0 &&
-      batch.metadata.metrics.oldestMessageTimestamp === 0
+      batch.metadata.metrics.oldestMessageTimestamp === undefined
     ) {
-      // If metadata is omitted → all values default to zero
+      // If metadata is omitted → counts default to zero, timestamp is undefined
       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);
+    assert.ok(
+      batch.metadata.metrics.oldestMessageTimestamp instanceof Date,
+      'Expected oldestMessageTimestamp to be a Date'
+    );
+    assert.strictEqual(
+      batch.metadata.metrics.oldestMessageTimestamp.getTime(),
+      1000000
+    );
     batch.ackAll();
   },
 

src/workerd/api/tests/queue-metrics-sentinel-test.js

@@ -0,0 +1,84 @@
+// 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
+
+// Tests that the upstream sentinel value of 0 for oldestMessageTimestamp is
+// correctly converted to undefined (kj::none) by clearEpochSentinel().
+
+import assert from 'node:assert';
+
+export default {
+  async fetch(request) {
+    const { pathname } = new URL(request.url);
+
+    if (pathname === '/metrics') {
+      return Response.json({
+        backlogCount: 5,
+        backlogBytes: 100,
+        oldestMessageTimestamp: 0,
+      });
+    }
+
+    if (pathname === '/message') {
+      await request.arrayBuffer();
+      return Response.json({
+        metadata: {
+          metrics: {
+            backlogCount: 5,
+            backlogBytes: 100,
+            oldestMessageTimestamp: 0,
+          },
+        },
+      });
+    }
+
+    if (pathname === '/batch') {
+      await request.arrayBuffer();
+      return Response.json({
+        metadata: {
+          metrics: {
+            backlogCount: 10,
+            backlogBytes: 200,
+            oldestMessageTimestamp: 0,
+          },
+        },
+      });
+    }
+
+    return new Response('Not Found', { status: 404 });
+  },
+
+  async test(ctrl, env) {
+    // Test metrics() zero-sentinel → undefined
+    const metrics = await env.QUEUE.metrics();
+    assert.strictEqual(metrics.backlogCount, 5);
+    assert.strictEqual(metrics.backlogBytes, 100);
+    assert.strictEqual(
+      metrics.oldestMessageTimestamp,
+      undefined,
+      'Expected oldestMessageTimestamp to be undefined when upstream sends 0'
+    );
+
+    // Test send() zero-sentinel → undefined
+    const sendResult = await env.QUEUE.send('abc', { contentType: 'text' });
+    assert.strictEqual(sendResult.metadata.metrics.backlogCount, 5);
+    assert.strictEqual(sendResult.metadata.metrics.backlogBytes, 100);
+    assert.strictEqual(
+      sendResult.metadata.metrics.oldestMessageTimestamp,
+      undefined,
+      'Expected send oldestMessageTimestamp to be undefined when upstream sends 0'
+    );
+
+    // Test sendBatch() zero-sentinel → undefined
+    const sendBatchResult = await env.QUEUE.sendBatch([
+      { body: 'def', contentType: 'text' },
+    ]);
+    assert.strictEqual(sendBatchResult.metadata.metrics.backlogCount, 10);
+    assert.strictEqual(sendBatchResult.metadata.metrics.backlogBytes, 200);
+    assert.strictEqual(
+      sendBatchResult.metadata.metrics.oldestMessageTimestamp,
+      undefined,
+      'Expected sendBatch oldestMessageTimestamp to be undefined when upstream sends 0'
+    );
+  },
+};

src/workerd/api/tests/queue-metrics-sentinel-test.wd-test

@@ -0,0 +1,17 @@
+using Workerd = import "/workerd/workerd.capnp";
+
+const unitTests :Workerd.Config = (
+  services = [
+    ( name = "queue-metrics-sentinel-test",
+      worker = (
+        modules = [
+          ( name = "worker", esModule = embed "queue-metrics-sentinel-test.js" )
+        ],
+        bindings = [
+          ( name = "QUEUE", queue = "queue-metrics-sentinel-test" ),
+        ],
+        compatibilityFlags = ["nodejs_compat", "queues_json_messages", "experimental", "capture_async_api_throws"],
+      )
+    ),
+  ],
+);

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

@@ -26,7 +26,11 @@ export default {
       const metrics = await env.QUEUE.metrics();
       assert.strictEqual(metrics.backlogCount, 100);
       assert.strictEqual(metrics.backlogBytes, 2048);
-      assert.strictEqual(metrics.oldestMessageTimestamp, 1000000);
+      assert.ok(
+        metrics.oldestMessageTimestamp instanceof Date,
+        'Expected oldestMessageTimestamp to be a Date'
+      );
+      assert.strictEqual(metrics.oldestMessageTimestamp.getTime(), 1000000);
     } else {
       // Flag OFF → metrics() should not be exposed on the binding
       assert.strictEqual(typeof env.QUEUE.metrics, 'undefined');

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

@@ -56,15 +56,23 @@ export default {
     if (responseBodyEnabled) {
       assert.strictEqual(sendResult.metadata.metrics.backlogCount, 100);
       assert.strictEqual(sendResult.metadata.metrics.backlogBytes, 2048);
+      assert.ok(
+        sendResult.metadata.metrics.oldestMessageTimestamp instanceof Date,
+        'Expected oldestMessageTimestamp to be a Date'
+      );
       assert.strictEqual(
-        sendResult.metadata.metrics.oldestMessageTimestamp,
+        sendResult.metadata.metrics.oldestMessageTimestamp.getTime(),
         1000000
       );
 
       assert.strictEqual(sendBatchResult.metadata.metrics.backlogCount, 200);
       assert.strictEqual(sendBatchResult.metadata.metrics.backlogBytes, 4096);
+      assert.ok(
+        sendBatchResult.metadata.metrics.oldestMessageTimestamp instanceof Date,
+        'Expected oldestMessageTimestamp to be a Date'
+      );
       assert.strictEqual(
-        sendBatchResult.metadata.metrics.oldestMessageTimestamp,
+        sendBatchResult.metadata.metrics.oldestMessageTimestamp.getTime(),
         2000000
       );
     } else {

types/generated-snapshot/experimental/index.d.ts

@@ -2409,7 +2409,7 @@ interface Queue<Body = unknown> {
 interface QueueSendMetrics {
   backlogCount: number;
   backlogBytes: number;
-  oldestMessageTimestamp: number;
+  oldestMessageTimestamp?: Date;
 }
 interface QueueSendMetadata {
   metrics: QueueSendMetrics;
@@ -2420,7 +2420,7 @@ interface QueueSendResponse {
 interface QueueSendBatchMetrics {
   backlogCount: number;
   backlogBytes: number;
-  oldestMessageTimestamp: number;
+  oldestMessageTimestamp?: Date;
 }
 interface QueueSendBatchMetadata {
   metrics: QueueSendBatchMetrics;
@@ -2443,12 +2443,12 @@ interface MessageSendRequest<Body = unknown> {
 interface QueueMetrics {
   backlogCount: number;
   backlogBytes: number;
-  oldestMessageTimestamp: number;
+  oldestMessageTimestamp?: Date;
 }
 interface MessageBatchMetrics {
   backlogCount: number;
   backlogBytes: number;
-  oldestMessageTimestamp: number;
+  oldestMessageTimestamp?: Date;
 }
 interface MessageBatchMetadata {
   metrics: MessageBatchMetrics;

types/generated-snapshot/experimental/index.ts

@@ -2412,7 +2412,7 @@ export interface Queue<Body = unknown> {
 export interface QueueSendMetrics {
   backlogCount: number;
   backlogBytes: number;
-  oldestMessageTimestamp: number;
+  oldestMessageTimestamp?: Date;
 }
 export interface QueueSendMetadata {
   metrics: QueueSendMetrics;
@@ -2423,7 +2423,7 @@ export interface QueueSendResponse {
 export interface QueueSendBatchMetrics {
   backlogCount: number;
   backlogBytes: number;
-  oldestMessageTimestamp: number;
+  oldestMessageTimestamp?: Date;
 }
 export interface QueueSendBatchMetadata {
   metrics: QueueSendBatchMetrics;
@@ -2446,12 +2446,12 @@ export interface MessageSendRequest<Body = unknown> {
 export interface QueueMetrics {
   backlogCount: number;
   backlogBytes: number;
-  oldestMessageTimestamp: number;
+  oldestMessageTimestamp?: Date;
 }
 export interface MessageBatchMetrics {
   backlogCount: number;
   backlogBytes: number;
-  oldestMessageTimestamp: number;
+  oldestMessageTimestamp?: Date;
 }
 export interface MessageBatchMetadata {
   metrics: MessageBatchMetrics;