feat: trace queue worker transactions + structured span test infrastructure

CHANGELOG.md

@@ -8,6 +8,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/).
 
 ### Added
 
+- Queue worker transactions are traced as coherent `<service> - tx` spans under the `cds.spawn - run task` root, instead of orphaned per-call spans
+
 ### Changed
 
 ### Fixed

lib/tracing/cds.js

@@ -46,6 +46,25 @@ module.exports = () => {
     }
   })
 
+  // Wrap `srv.tx(fn)` so the queue worker's two transactions (SELECT+UPDATE lock tx,
+  // then handle+DELETE dispatch tx) appear as child spans of the `cds.spawn - run task`
+  // root instead of each top-level CAP call inside them becoming an orphan root.
+  // Only wraps when there is no current `srv.context` (i.e. not already inside a request).
+  const _tx_proto = cds.Service.prototype.tx
+  cds.Service.prototype.tx = wrap(_tx_proto, {
+    wrapper: function tx() {
+      const fnIdx = typeof arguments[0] === 'function' ? 0 : typeof arguments[1] === 'function' ? 1 : -1
+      if (fnIdx < 0) return _tx_proto.apply(this, arguments)
+      // Skip if this service is already handling a request (has an active EventContext),
+      // or if this is a nested .tx() call (cds.Service.tx is a no-op when already in tx).
+      // Do NOT skip for a bare {} context set by processInboundMsg — that's the entry point
+      // for file-based messaging consumer delivery and should get a root span.
+      if (this.context instanceof cds.EventContext) return _tx_proto.apply(this, arguments)
+      const name = `${this.name || 'cds'} - tx`
+      return trace(name, _tx_proto, this, arguments, {})
+    }
+  })
+
   const { spawn: _spawn } = cds
   cds.spawn = wrap(_spawn, {
     wrapper: function spawn() {

test/bookshop/.cdsrc.json

@@ -82,6 +82,18 @@
       }
     }
   },
+  "[tracing-in-memory]": {
+    "requires": {
+      "telemetry": {
+        "tracing": {
+          "exporter": {
+            "module": "./lib/MyInMemorySpanExporter.js",
+            "class": "MyInMemorySpanExporter"
+          }
+        }
+      }
+    }
+  },
   "[persistent-outbox]": {
     "requires": {
       "messaging": {
@@ -90,6 +102,15 @@
       }
     }
   },
+  "[inboxed]": {
+    "requires": {
+      "messaging": {
+        "kind": "file-based-messaging",
+        "file": "../inboxed",
+        "inboxed": true
+      }
+    }
+  },
   "[without-outbox]": {
     "requires": {
       "messaging": {

test/bookshop/lib/MyInMemorySpanExporter.js

@@ -0,0 +1,59 @@
+// In-memory span exporter for tests. Spans are accumulated in a module-level array that
+// tests can import directly via `require('./lib/MyInMemorySpanExporter').captured`.
+// Wired into the tracer provider via .cdsrc.json profile config (no provider-poking from tests).
+
+const { ExportResultCode } = require('@opentelemetry/core')
+
+const captured = []
+
+class MyInMemorySpanExporter {
+  export(spans, resultCallback) {
+    captured.push(...spans)
+    resultCallback({ code: ExportResultCode.SUCCESS })
+  }
+
+  shutdown() {
+    return Promise.resolve()
+  }
+
+  forceFlush() {
+    return Promise.resolve()
+  }
+}
+
+// Returns the captured spans grouped by traceId, each group is a hierarchy:
+//   { traceId, root, all, byParent }
+// `root` is the span with no parent inside the group (the visible root for the exporter's
+// "elapsed times:" primer logic — i.e. spans whose parentSpanId is not present in this group).
+function groupedByTrace() {
+  const byTrace = new Map()
+  for (const s of captured) {
+    const tid = s.spanContext().traceId
+    if (!byTrace.has(tid)) byTrace.set(tid, [])
+    byTrace.get(tid).push(s)
+  }
+
+  return [...byTrace.entries()].map(([traceId, all]) => {
+    const ids = new Set(all.map(s => s.spanContext().spanId))
+    const roots = all.filter(s => !s.parentSpanContext?.spanId || !ids.has(s.parentSpanContext.spanId))
+    const byParent = new Map()
+    for (const s of all) {
+      const pid = s.parentSpanContext?.spanId
+      if (!byParent.has(pid)) byParent.set(pid, [])
+      byParent.get(pid).push(s)
+    }
+    return { traceId, root: roots[0], roots, all, byParent }
+  })
+}
+
+// Returns just the visible "root" spans across all captured traces. These correspond 1:1 to
+// "elapsed times:" primers our ConsoleSpanExporter would emit for the same data.
+function rootSpans() {
+  return groupedByTrace().flatMap(g => g.roots)
+}
+
+function reset() {
+  captured.length = 0
+}
+
+module.exports = { MyInMemorySpanExporter, captured, groupedByTrace, rootSpans, reset }

test/bookshop/srv/admin-service.cds

@@ -7,6 +7,9 @@ service AdminService @(requires: 'admin') {
 
   action test_spawn();
   action test_emit();
+  action test_outboxed_send();
+  action test_outboxed_send_batch();
+  action test_scheduled();
 
   event foo {
     bar : String;

test/bookshop/srv/admin-service.js

@@ -32,6 +32,30 @@ module.exports = class AdminService extends cds.ApplicationService {
       await messaging.emit('foo', { bar: 'baz' })
     })
 
+    // test_outboxed_send: writes a task to the persistent outbox addressed to ExternalServiceOne,
+    // whose handler the test installs. Exercises the queue-worker path (scan, lock, dispatch).
+    this.on('test_outboxed_send', async () => {
+      const externalOne = await cds.connect.to('ExternalServiceOne')
+      await cds.queued(externalOne).send('call', {})
+    })
+
+    // test_outboxed_send_batch: writes multiple tasks to the persistent outbox to exercise chunkSize > 1 fan-out.
+    this.on('test_outboxed_send_batch', async () => {
+      const externalOne = await cds.connect.to('ExternalServiceOne')
+      const queued = cds.queued(externalOne)
+      await Promise.all([
+        queued.send('call', {}),
+        queued.send('call', {}),
+        queued.send('call', {})
+      ])
+    })
+
+    // test_scheduled: schedules a one-shot task to fire after a short delay.
+    this.on('test_scheduled', async () => {
+      const externalOne = await cds.connect.to('ExternalServiceOne')
+      await cds.queued(externalOne).schedule('call', {}).after(10)
+    })
+
     return super.init()
   }
 }