3.2.0 - queue.process becomes child of queue.push span (producer-consumer link)

Author: nvms

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@prsm/queue",
-  "version": "3.1.3",
+  "version": "3.2.0",
   "description": "Redis-backed distributed task queue with grouped concurrency, retries, and rate limiting",
   "type": "module",
   "exports": {

package.json

@@ -138,13 +138,12 @@ export default class Queue extends EventEmitter {
    */
   async push(payload, { group } = {}) {
     if (this._closed) throw new Error("Queue is closed")
-    const traceparent = this._tracer ? this._tracer.toTraceparent() : null
     const task = group
       ? { uuid: randomUUID(), payload, createdAt: Date.now(), group, attempts: 0 }
       : { uuid: randomUUID(), payload, createdAt: Date.now(), attempts: 0 }
-    if (traceparent) task.traceparent = traceparent
     this._pushed++
     const span = this._tracer?.startSpan('queue.push', { 'queue.group': group ?? null, 'task.uuid': task.uuid }, { kind: 'producer' })
+    if (span) task.traceparent = this._tracer.toTraceparent(span.context)
     try {
       await this._enqueue(task, group)
     } catch (err) {
@@ -165,11 +164,14 @@ export default class Queue extends EventEmitter {
    */
   async pushAndWait(payload, { group, timeout = 0 } = {}) {
     if (this._closed) throw new Error("Queue is closed")
-    const traceparent = this._tracer ? this._tracer.toTraceparent() : null
     const task = group
       ? { uuid: randomUUID(), payload, createdAt: Date.now(), group, attempts: 0 }
       : { uuid: randomUUID(), payload, createdAt: Date.now(), attempts: 0 }
-    if (traceparent) task.traceparent = traceparent
+    const tpSpan = this._tracer?.startSpan('queue.pushAndWait', { 'queue.group': group ?? null, 'task.uuid': task.uuid }, { kind: 'producer' })
+    if (tpSpan) {
+      task.traceparent = this._tracer.toTraceparent(tpSpan.context)
+      tpSpan.end()
+    }
     this._pushed++
     const { promise, ready } = this._awaitTask(task.uuid, timeout)
     promise.catch(() => {})

src/queue.js

@@ -0,0 +1,115 @@
+import { describe, it, expect, beforeEach, afterEach } from 'vitest'
+import { createClient } from 'redis'
+import { createTracer } from '@prsm/trace'
+import Queue from '../src/index.js'
+
+const REDIS = {}
+let admin
+
+async function flush() {
+  if (!admin) {
+    admin = createClient()
+    admin.on('error', () => {})
+    await admin.connect()
+  }
+  await admin.flushAll()
+}
+
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
+
+const queues = []
+function make(opts) {
+  const q = new Queue({ redisOptions: REDIS, ...opts })
+  queues.push(q)
+  return q
+}
+
+beforeEach(async () => { await flush() })
+
+afterEach(async () => {
+  while (queues.length) await queues.pop().close().catch(() => {})
+})
+
+describe('cross-instance trace propagation', () => {
+  it('producer span on one instance, consumer span on another, share traceId', async () => {
+    const tracer = createTracer({ service: 'svc' })
+    const spans = []
+    tracer.onSpan((s) => spans.push(s))
+
+    const producer = make({ tracer, concurrency: 0 })
+    const consumer = make({ tracer, concurrency: 2 })
+
+    await producer.ready()
+    await consumer.ready()
+
+    let processed = null
+    consumer.process(async (payload) => { processed = payload; return { ok: true } })
+
+    let pushedTraceId = null
+    await tracer.span('http.POST /order', async () => {
+      pushedTraceId = tracer.current().traceId
+      await producer.push({ orderId: 42 })
+    })
+
+    await sleep(400)
+
+    expect(processed).toEqual({ orderId: 42 })
+
+    const pushSpan = spans.find((s) => s.name === 'queue.push')
+    const processSpan = spans.find((s) => s.name === 'queue.process')
+    expect(pushSpan).toBeTruthy()
+    expect(processSpan).toBeTruthy()
+    expect(pushSpan.traceId).toBe(pushedTraceId)
+    expect(processSpan.traceId).toBe(pushedTraceId)
+    expect(processSpan.parentSpanId).toBe(pushSpan.spanId)
+  })
+
+  it('producer with no upstream context: push is root, process is its child', async () => {
+    const tracer = createTracer({ service: 'svc' })
+    const spans = []
+    tracer.onSpan((s) => spans.push(s))
+
+    const producer = make({ tracer, concurrency: 0 })
+    const consumer = make({ tracer, concurrency: 2 })
+    await producer.ready()
+    await consumer.ready()
+    consumer.process(async () => 'done')
+
+    // push outside any tracer.span — push becomes the trace root
+    await producer.push({ x: 1 })
+    await sleep(300)
+
+    const pushSpan = spans.find((s) => s.name === 'queue.push')
+    const processSpan = spans.find((s) => s.name === 'queue.process')
+    expect(pushSpan).toBeTruthy()
+    expect(processSpan).toBeTruthy()
+    expect(pushSpan.parentSpanId).toBeNull()
+    expect(processSpan.parentSpanId).toBe(pushSpan.spanId)
+    expect(processSpan.traceId).toBe(pushSpan.traceId)
+  })
+
+  it('grouped job carries trace context through the grouped queue path', async () => {
+    const tracer = createTracer({ service: 'svc' })
+    const spans = []
+    tracer.onSpan((s) => spans.push(s))
+
+    const producer = make({ tracer, concurrency: 0 })
+    const consumer = make({ tracer, groups: { concurrency: 1 } })
+    await producer.ready()
+    await consumer.ready()
+    consumer.process(async () => 'done')
+
+    let pushedTraceId = null
+    await tracer.span('grouped-entry', async () => {
+      pushedTraceId = tracer.current().traceId
+      await producer.push({ p: 1 }, { group: 'tenant-a' })
+    })
+
+    await sleep(400)
+
+    const pushSpan = spans.find((s) => s.name === 'queue.push')
+    const processSpan = spans.find((s) => s.name === 'queue.process')
+    expect(pushSpan.traceId).toBe(pushedTraceId)
+    expect(processSpan.traceId).toBe(pushedTraceId)
+  })
+})

tests/cross-instance-trace.test.js

@@ -2,5 +2,6 @@ export default {
   test: {
     reporters: ["verbose"],
     testTimeout: 15000,
+    fileParallelism: false,
   },
 }