test(dsm): add regression test for DSM context leak in google-cloud-pubsub

Same root cause as the kafkajs test: ctx.currentStore is set by
startSpan before decodeDataStreamsContext/setCheckpoint call enterWith,
so the DSM context is never included in the bound store for async
continuations.

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

Author: robcarlan-datadog

packages/datadog-plugin-google-cloud-pubsub/test/dsm.spec.js

@@ -128,6 +128,72 @@ describe('Plugin', () => {
           })
         })
 
+        describe('concurrent context isolation', () => {
+          it('Should maintain separate DSM context for interleaved consume-produce flows', async () => {
+            const setCheckpointSpy = sinon.spy(DataStreamsProcessor.prototype, 'setCheckpoint')
+
+            try {
+              const topicAIn = (await pubsub.createTopic(`dsm-iso-a-in-${id()}`))[0]
+              const topicBIn = (await pubsub.createTopic(`dsm-iso-b-in-${id()}`))[0]
+              const topicAOut = (await pubsub.createTopic(`dsm-iso-a-out-${id()}`))[0]
+              const topicBOut = (await pubsub.createTopic(`dsm-iso-b-out-${id()}`))[0]
+
+              const subA = (await topicAIn.createSubscription(`sub-a-${id()}`))[0]
+              const subB = (await topicBIn.createSubscription(`sub-b-${id()}`))[0]
+
+              const fullTopicAIn = topicAIn.metadata?.name || topicAIn.name
+              const fullTopicBIn = topicBIn.metadata?.name || topicBIn.name
+              const fullTopicAOut = topicAOut.metadata?.name || topicAOut.name
+              const fullTopicBOut = topicBOut.metadata?.name || topicBOut.name
+
+              // Synchronization: both consumers must receive before either produces
+              let resolveAEntered, resolveBEntered
+              const aEntered = new Promise(resolve => { resolveAEntered = resolve })
+              const bEntered = new Promise(resolve => { resolveBEntered = resolve })
+              let doneCount = 0
+              const allDone = new Promise(resolve => {
+                const check = () => { if (++doneCount === 2) resolve() }
+                subA.on('message', async (msg) => {
+                  msg.ack()
+                  resolveAEntered()
+                  await bEntered
+                  await publish(topicAOut, { data: Buffer.from('from-a') })
+                  check()
+                })
+                subB.on('message', async (msg) => {
+                  msg.ack()
+                  resolveBEntered()
+                  await aEntered
+                  await publish(topicBOut, { data: Buffer.from('from-b') })
+                  check()
+                })
+              })
+
+              await publish(topicAIn, { data: Buffer.from('msg-a') })
+              await publish(topicBIn, { data: Buffer.from('msg-b') })
+
+              await allDone
+
+              const calls = setCheckpointSpy.getCalls()
+              const checkpoint = (dir, topic) => calls.find(c =>
+                c.args[0].includes(`direction:${dir}`) && c.args[0].includes(`topic:${topic}`)
+              )
+
+              const consumeA = checkpoint('in', fullTopicAIn)
+              const consumeB = checkpoint('in', fullTopicBIn)
+              const produceA = checkpoint('out', fullTopicAOut)
+              const produceB = checkpoint('out', fullTopicBOut)
+
+              assert.ok(produceA?.args[2], 'Process A produce should have a parent DSM context')
+              assert.ok(produceB?.args[2], 'Process B produce should have a parent DSM context')
+              assert.deepStrictEqual(produceA.args[2].hash, consumeA.returnValue.hash)
+              assert.deepStrictEqual(produceB.args[2].hash, consumeB.returnValue.hash)
+            } finally {
+              setCheckpointSpy.restore()
+            }
+          })
+        })
+
         describe('it should set a message payload size', () => {
           let recordCheckpointSpy