fix(dsm): move DSM logic from bindStart to start for google-cloud-pubsub

Move DSM checkpoint logic from bindStart to start in both consumer and
producer plugins so DSM context is properly bound in the async store.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: robcarlan-datadog

packages/datadog-plugin-google-cloud-pubsub/src/consumer.js

@@ -96,6 +96,20 @@ class GoogleCloudPubsubConsumerPlugin extends ConsumerPlugin {
     })
   }
 
+  start (ctx) {
+    if (!this.config.dsmEnabled) return
+    const { message } = ctx
+    if (!message?.attributes) return
+
+    const { span } = ctx.currentStore
+    const subscription = message._subscriber._subscription
+    const topic = subscription?.metadata?.topic || message.attributes?.['pubsub.topic']
+    const payloadSize = getMessageSize(message)
+    this.tracer.decodeDataStreamsContext(message.attributes)
+    this.tracer
+      .setCheckpoint(['direction:in', `topic:${topic}`, 'type:google-pubsub'], span, payloadSize)
+  }
+
   bindStart (ctx) {
     const { message } = ctx
     const subscription = message._subscriber._subscription
@@ -181,13 +195,6 @@ class GoogleCloudPubsubConsumerPlugin extends ConsumerPlugin {
       }
     }
 
-    if (this.config.dsmEnabled && message?.attributes) {
-      const payloadSize = getMessageSize(message)
-      this.tracer.decodeDataStreamsContext(message.attributes)
-      this.tracer
-        .setCheckpoint(['direction:in', `topic:${topic}`, 'type:google-pubsub'], span, payloadSize)
-    }
-
     return ctx.currentStore
   }
 

packages/datadog-plugin-google-cloud-pubsub/src/producer.js

@@ -36,6 +36,24 @@ class GoogleCloudPubsubProducerPlugin extends ProducerPlugin {
     if (topicName) attributes['pubsub.topic'] = topicName
   }
 
+  start (ctx) {
+    if (!this.config.dsmEnabled) return
+    const { request } = ctx
+    const messages = request.messages || []
+    const topic = request.topic
+    const { span } = ctx.currentStore
+
+    for (const msg of messages) {
+      const attributes = msg.attributes ??= {}
+      const dataStreamsContext = this.tracer.setCheckpoint(
+        ['direction:out', `topic:${topic}`, 'type:google-pubsub'],
+        span,
+        getHeadersSize(msg)
+      )
+      DsmPathwayCodec.encode(dataStreamsContext, attributes)
+    }
+  }
+
   bindStart (ctx) {
     const { request, api, projectId } = ctx
     if (api !== 'publish') return
@@ -113,7 +131,6 @@ class GoogleCloudPubsubProducerPlugin extends ProducerPlugin {
 
     const messageCountStr = String(messageCount)
     const startTimeStr = String(Math.floor(batchSpan._startTime))
-    const dsmEnabled = this.config.dsmEnabled
 
     for (let i = 0; i < messageCount; i++) {
       const msg = messages[i]
@@ -137,15 +154,6 @@ class GoogleCloudPubsubProducerPlugin extends ProducerPlugin {
       if (batchTraceIdUpper) {
         attributes['_dd.pubsub_request.p.tid'] = batchTraceIdUpper
       }
-
-      if (dsmEnabled) {
-        const dataStreamsContext = this.tracer.setCheckpoint(
-          ['direction:out', `topic:${topic}`, 'type:google-pubsub'],
-          batchSpan,
-          getHeadersSize(msg)
-        )
-        DsmPathwayCodec.encode(dataStreamsContext, attributes)
-      }
     }
 
     ctx.batchSpan = batchSpan