feat!: require OpenTelemetry JS SDK 3

README.md

@@ -1011,6 +1011,8 @@ renewal that fails is logged at the `warn` level.
 
 `@boringnode/queue` ships with built-in OpenTelemetry instrumentation that creates **PRODUCER** spans for job dispatch and **CONSUMER** spans for job execution, following [OTel messaging semantic conventions](https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/).
 
+It requires OpenTelemetry JS SDK 3 (`@opentelemetry/api` 1.10, `@opentelemetry/core` 3, `@opentelemetry/instrumentation` 0.300+). Applications still on SDK 2 should stay on `@boringnode/queue` 0.9.x.
+
 ### Quick Setup
 
 ```typescript
@@ -1045,6 +1047,7 @@ The instrumentation uses standard [OTel messaging semantic conventions](https://
 | `messaging.job.group_id`        | Custom  | Queue-specific group identifier               |
 | `messaging.job.priority`        | Custom  | Queue-specific job priority                   |
 | `messaging.job.delay_ms`        | Custom  | Delay before the job becomes available        |
+| `messaging.job.dedup_outcome`   | Custom  | Outcome of `.dedup()` (e.g. `'skipped'`)      |
 | `messaging.job.queue_time_ms`   | Custom  | Time spent waiting in queue before processing |
 
 ### Trace Context Propagation

package.json

@@ -45,11 +45,11 @@
     "@japa/expect-type": "^2.0.4",
     "@japa/file-system": "^3.0.0",
     "@japa/runner": "^5.4.0",
-    "@opentelemetry/api": "^1.9.1",
-    "@opentelemetry/context-async-hooks": "^2.12.0",
-    "@opentelemetry/core": "^2.12.0",
-    "@opentelemetry/instrumentation": "^0.223.0",
-    "@opentelemetry/sdk-trace-base": "^2.12.0",
+    "@opentelemetry/api": "^1.10.0",
+    "@opentelemetry/context-async-hooks": "^3.1.1",
+    "@opentelemetry/core": "^3.1.1",
+    "@opentelemetry/instrumentation": "^0.301.1",
+    "@opentelemetry/sdk-trace": "^3.1.1",
     "@poppinss/ts-exec": "^1.5.1",
     "@release-it/conventional-changelog": "^12.0.2",
     "@types/better-sqlite3": "^9.6.0",
@@ -73,9 +73,9 @@
     "typescript": "^6.0.3"
   },
   "peerDependencies": {
-    "@opentelemetry/api": "^1.9.0",
-    "@opentelemetry/core": "^1.30.0 || ^2.0.0",
-    "@opentelemetry/instrumentation": ">=0.200.0 <0.300.0",
+    "@opentelemetry/api": "^1.10.0",
+    "@opentelemetry/core": "^3.0.0",
+    "@opentelemetry/instrumentation": ">=0.300.0 <0.400.0",
     "ioredis": "^5.0.0 || ^6.0.0",
     "knex": "^3.0.0",
     "kysely": "^0.29.0"

src/job_dispatch_runtime.ts

@@ -111,6 +111,7 @@ export class JobDispatchRuntime {
       if (result) {
         pushResult = result
         message.dedupOutcome = result.outcome
+        message.dedupJobId = result.jobId
       }
     }, message)
 

src/otel.ts

@@ -15,6 +15,7 @@
 import { dispatchChannel, executeChannel } from './tracing_channels.js'
 import type { AcquiredJob } from './contracts/adapter.js'
 import type { JobDispatchMessage, JobExecuteMessage } from './types/tracing_channels.js'
+import packageJson from '../package.json' with { type: 'json' }
 
 export interface QueueInstrumentationConfig extends InstrumentationConfig {
   /**
@@ -53,7 +54,7 @@
   #patchedManager?: { init: (...args: any[]) => any }
 
   constructor(config: QueueInstrumentationConfig = {}) {
-    super('@boringnode/queue', '0.1.0', config)
+    super(packageJson.name, packageJson.version, config)
   }
 
   get #messagingSystem(): string {
@@ -106,7 +107,7 @@
 
     this.#patchedManager = manager
     this.#originalInit = manager.init.bind(manager)
     const instrumentation = this
 
     manager.init = async (config: any) => {
       return this.#originalInit!({
@@ -192,6 +193,11 @@
     const span = this.dispatchSpans.get(message)
     if (!span) return
 
+    if (message.dedupOutcome) {
+      span.setAttribute('messaging.job.dedup_outcome', message.dedupOutcome)
+      if (message.dedupJobId) span.setAttribute('messaging.message.id', message.dedupJobId)
+    }
+
     if (message.error) {
       span.recordException(message.error)
       span.setStatus({ code: SpanStatusCode.ERROR, message: message.error.message })
@@ -206,6 +212,10 @@
    * execution in OTel context for proper child span parenting.
    */
   #wrapExecution<T>(fn: () => Promise<T>, job: AcquiredJob, queue: string): Promise<T> {
+    // The queue manager keeps this wrapper after disable(); without the
+    // execute subscription, nothing would end the span.
+    if (!this.subscribed) return fn()
+
     const extractedContext = this.#extractParentContext(job.traceContext)
     const parentSpanContext = trace.getSpanContext(extractedContext)
 

src/types/tracing_channels.ts

@@ -28,6 +28,12 @@ export type JobDispatchMessage = {
    */
   dedupOutcome?: DedupOutcome
 
+  /**
+   * ID of the job that holds the deduplication key, set with `dedupOutcome`.
+   * It differs from `jobs[0].id` when the dispatch did not add a new job.
+   */
+  dedupJobId?: string
+
   /** Error that caused the dispatch to fail */
   error?: Error
 }

tests/helpers/setup_tracing.ts

@@ -1,15 +1,11 @@
 import { trace, context, propagation } from '@opentelemetry/api'
-import {
-  BasicTracerProvider,
-  InMemorySpanExporter,
-  SimpleSpanProcessor,
-} from '@opentelemetry/sdk-trace-base'
+import { TracerProvider, InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace'
 import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks'
 import { W3CTraceContextPropagator } from '@opentelemetry/core'
 
 let initialized = false
 let exporter: InMemorySpanExporter
-let provider: BasicTracerProvider
+let provider: TracerProvider
 
 export function setupTracing() {
   if (!initialized) {
@@ -18,7 +14,7 @@ export function setupTracing() {
 
     context.setGlobalContextManager(contextManager.enable())
     propagation.setGlobalPropagator(new W3CTraceContextPropagator())
-    provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] })
+    provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] })
 
     trace.setGlobalTracerProvider(provider)
     initialized = true

tests/otel.spec.ts

@@ -6,6 +6,7 @@ import { dispatchChannel, executeChannel } from '../src/tracing_channels.js'
 import type { JobExecuteMessage, JobDispatchMessage } from '../src/types/tracing_channels.js'
 import type { AcquiredJob } from '../src/contracts/adapter.js'
 import type { JobData } from '../src/types/main.js'
+import packageJson from '../package.json' with { type: 'json' }
 
 function makeJob(overrides: Partial<AcquiredJob> = {}): AcquiredJob {
   return {
@@ -113,6 +114,8 @@ test.group('QueueInstrumentation | dispatch via DC', (group) => {
     assert.equal(spans[0].attributes['messaging.destination.name'], 'emails')
     assert.equal(spans[0].attributes['messaging.message.id'], 'job-1')
     assert.equal(spans[0].attributes['messaging.job.name'], 'SendEmailJob')
+    assert.equal(spans[0].instrumentationScope.name, '@boringnode/queue')
+    assert.equal(spans[0].instrumentationScope.version, packageJson.version)
 
     instrumentation.disable()
   })
@@ -156,6 +159,23 @@ test.group('QueueInstrumentation | dispatch via DC', (group) => {
     instrumentation.disable()
   })
 
+  test('reports the deduplicated job instead of the skipped one', async ({ assert }) => {
+    const { instrumentation } = await setupWithWrappers()
+
+    const jobData: JobData = { id: 'new-id', name: 'DedupJob', payload: {}, attempts: 0 }
+    const message: JobDispatchMessage = { jobs: [jobData], queue: 'default' }
+    await dispatchChannel.tracePromise(async () => {
+      message.dedupOutcome = 'skipped'
+      message.dedupJobId = 'existing-id'
+    }, message)
+
+    const spans = getFinishedSpans()
+    assert.equal(spans[0].attributes['messaging.message.id'], 'existing-id')
+    assert.equal(spans[0].attributes['messaging.job.dedup_outcome'], 'skipped')
+
+    instrumentation.disable()
+  })
+
   test('includes delay_ms and batch_count attributes', async ({ assert }) => {
     const { instrumentation } = await setupWithWrappers()
 
@@ -211,6 +231,29 @@ test.group('QueueInstrumentation | execute via executionWrapper', (group) => {
   })
   group.each.setup(() => resetSpans())
 
+  test('does not start a CONSUMER span after disable()', async ({ assert }) => {
+    const { instrumentation, executionWrapper } = await setupWithWrappers()
+    instrumentation.disable()
+
+    const job = makeJob({ id: 'disabled-1' })
+    const message: JobExecuteMessage = { job, queue: 'default', status: 'completed' }
+    let activeSpan: unknown
+    const result = await executeChannel.tracePromise(async () => {
+      return executionWrapper(
+        async () => {
+          activeSpan = trace.getActiveSpan()
+          return 'done'
+        },
+        job,
+        'default'
+      )
+    }, message)
+
+    assert.equal(result, 'done')
+    assert.isUndefined(activeSpan)
+    assert.lengthOf(getFinishedSpans(), 0)
+  })
+
   test('creates CONSUMER span with semconv attributes', async ({ assert }) => {
     const { instrumentation, executionWrapper } = await setupWithWrappers()
 

yarn.lock

@@ -47,11 +47,11 @@ __metadata:
     "@japa/file-system": "npm:^3.0.0"
     "@japa/runner": "npm:^5.4.0"
     "@lukeed/ms": "npm:^2.0.2"
-    "@opentelemetry/api": "npm:^1.9.1"
-    "@opentelemetry/context-async-hooks": "npm:^2.12.0"
-    "@opentelemetry/core": "npm:^2.12.0"
-    "@opentelemetry/instrumentation": "npm:^0.223.0"
-    "@opentelemetry/sdk-trace-base": "npm:^2.12.0"
+    "@opentelemetry/api": "npm:^1.10.0"
+    "@opentelemetry/context-async-hooks": "npm:^3.1.1"
+    "@opentelemetry/core": "npm:^3.1.1"
+    "@opentelemetry/instrumentation": "npm:^0.301.1"
+    "@opentelemetry/sdk-trace": "npm:^3.1.1"
     "@poppinss/ts-exec": "npm:^1.5.1"
     "@poppinss/utils": "npm:^7.1.2"
     "@release-it/conventional-changelog": "npm:^12.0.2"
@@ -76,9 +76,9 @@ __metadata:
     tsup: "npm:^9.0.0"
     typescript: "npm:^6.0.3"
   peerDependencies:
-    "@opentelemetry/api": ^1.9.0
-    "@opentelemetry/core": ^1.30.0 || ^2.0.0
-    "@opentelemetry/instrumentation": ">=0.200.0 <0.300.0"
+    "@opentelemetry/api": ^1.10.0
+    "@opentelemetry/core": ^3.0.0
+    "@opentelemetry/instrumentation": ">=0.300.0 <0.400.0"
     ioredis: ^5.0.0 || ^6.0.0
     knex: ^3.0.0
     kysely: ^0.29.0
@@ -927,98 +927,67 @@ __metadata:
   languageName: node
   linkType: hard
 
-"@opentelemetry/api-logs@npm:0.223.0":
-  version: 0.223.0
-  resolution: "@opentelemetry/api-logs@npm:0.223.0"
-  dependencies:
-    "@opentelemetry/api": "npm:^1.3.0"
-  checksum: 10c0/3be606243df08c7356d584d48194464050f3c379d63a92cc059de4f45dd5e81bbd7810f0a95419bed71fe4b790d2e495663eaf563f2e67bde30b381182add32e
+"@opentelemetry/api@npm:^1.10.0":
+  version: 1.10.0
+  resolution: "@opentelemetry/api@npm:1.10.0"
+  checksum: 10c0/c666f89d882e2e3ea3780b4e419de3ba7e00a80bf334269c3c8507a5e2b4e453e2bc8368f6230f894d789b206c14dde313ea6f51fa7bc629f0eff8480b5d356b
   languageName: node
   linkType: hard
 
-"@opentelemetry/api@npm:^1.3.0":
-  version: 1.9.0
-  resolution: "@opentelemetry/api@npm:1.9.0"
-  checksum: 10c0/9aae2fe6e8a3a3eeb6c1fdef78e1939cf05a0f37f8a4fae4d6bf2e09eb1e06f966ece85805626e01ba5fab48072b94f19b835449e58b6d26720ee19a58298add
-  languageName: node
-  linkType: hard
-
-"@opentelemetry/api@npm:^1.9.1":
-  version: 1.9.1
-  resolution: "@opentelemetry/api@npm:1.9.1"
-  checksum: 10c0/c608485fc8b5a91e1f7e05e843b45b509307456b31cd2ad365933d90813e40ebfedf179f1451c762037e82d7c76aa8500e95d2da3609f640a1206cde5322cd14
-  languageName: node
-  linkType: hard
-
-"@opentelemetry/context-async-hooks@npm:^2.12.0":
-  version: 2.12.0
-  resolution: "@opentelemetry/context-async-hooks@npm:2.12.0"
+"@opentelemetry/context-async-hooks@npm:^3.1.1":
+  version: 3.1.1
+  resolution: "@opentelemetry/context-async-hooks@npm:3.1.1"
   peerDependencies:
-    "@opentelemetry/api": ">=1.0.0 <1.10.0"
-  checksum: 10c0/fc4185cc643562527532cc5de985edc963a02a68aa1cf6ec517da5602cb51620ea9ffd2201f0fb75295eb9eb7d281a517598a081eeda79a905245a110f21282c
+    "@opentelemetry/api": ">=1.0.0 <1.11.0"
+  checksum: 10c0/70d26beee812852b6edc51282fe8053ca7190157987ccf0487020f53984ecc33ddda2561ae21d9ed02a966316f7c7c98d9fa901de65cc138abd89d9177fdf16f
   languageName: node
   linkType: hard
 
-"@opentelemetry/core@npm:2.12.0, @opentelemetry/core@npm:^2.12.0":
-  version: 2.12.0
-  resolution: "@opentelemetry/core@npm:2.12.0"
+"@opentelemetry/core@npm:3.1.1, @opentelemetry/core@npm:^3.1.1":
+  version: 3.1.1
+  resolution: "@opentelemetry/core@npm:3.1.1"
   dependencies:
     "@opentelemetry/semantic-conventions": "npm:^1.29.0"
   peerDependencies:
-    "@opentelemetry/api": ">=1.0.0 <1.10.0"
-  checksum: 10c0/e0d0338d314ca827b4395acf8c7eb8f88a58fd63f1effc6068ba4dd8e3443e02e5796333ae6723fb926a7e6c18dd6d2102f5cc7e531b185618562dadbf6cad71
+    "@opentelemetry/api": ">=1.10.0 <1.11.0"
+  checksum: 10c0/4776fb7f6dd08e96c1069f8b08e329ae854353ab617759539ca4b5bf925a1cce77b7c6f744f14565e75297cb73117a111c5d96b6489e64e789502032f041fceb
   languageName: node
   linkType: hard
 
-"@opentelemetry/instrumentation@npm:^0.223.0":
-  version: 0.223.0
-  resolution: "@opentelemetry/instrumentation@npm:0.223.0"
+"@opentelemetry/instrumentation@npm:^0.301.1":
+  version: 0.301.1
+  resolution: "@opentelemetry/instrumentation@npm:0.301.1"
   dependencies:
-    "@opentelemetry/api-logs": "npm:0.223.0"
     import-in-the-middle: "npm:^3.0.0"
     require-in-the-middle: "npm:^8.0.0"
   peerDependencies:
-    "@opentelemetry/api": ^1.3.0
-  checksum: 10c0/91d376a6625ed005655b27492922068f0d2fbe04372002407a6498d96e953b0b4669a97b9946cc49286a28082195b6db849ad35583b9772bc3a359b4e672756f
+    "@opentelemetry/api": ">=1.10.0 <1.11.0"
+  checksum: 10c0/3fa4a7a3fc8b77addcdf7374372175d46219aeba390bcd560f541920d52cbedda63a2d2cc85a909a97e191ced90212c6a10295eb2b921635cff3937bd28d2429
   languageName: node
   linkType: hard
 
-"@opentelemetry/resources@npm:2.12.0":
-  version: 2.12.0
-  resolution: "@opentelemetry/resources@npm:2.12.0"
-  dependencies:
-    "@opentelemetry/core": "npm:2.12.0"
-    "@opentelemetry/semantic-conventions": "npm:^1.29.0"
-  peerDependencies:
-    "@opentelemetry/api": ">=1.3.0 <1.10.0"
-  checksum: 10c0/ef3f8470b9ada8406fa4e4a0f002ca31537c1b7fe61f06102fb9023c97a56b0d978c687dd392539f1e005c20f0467f6a1c66f9c43e8d31c2900023164529f2bc
-  languageName: node
-  linkType: hard
-
-"@opentelemetry/sdk-trace-base@npm:^2.12.0":
-  version: 2.12.0
-  resolution: "@opentelemetry/sdk-trace-base@npm:2.12.0"
+"@opentelemetry/resources@npm:3.1.1":
+  version: 3.1.1
+  resolution: "@opentelemetry/resources@npm:3.1.1"
   dependencies:
-    "@opentelemetry/core": "npm:2.12.0"
-    "@opentelemetry/resources": "npm:2.12.0"
-    "@opentelemetry/sdk-trace": "npm:2.12.0"
+    "@opentelemetry/core": "npm:3.1.1"
     "@opentelemetry/semantic-conventions": "npm:^1.29.0"
   peerDependencies:
-    "@opentelemetry/api": ">=1.3.0 <1.10.0"
-  checksum: 10c0/cd1de7e881f60a98047d662915b74b7b39cde69a51e9fc146fb4f02af4df4441a785af57e7ad267ea22bbc6ac19c5497684b51d7407573c2aef51f9e402aea4b
+    "@opentelemetry/api": ">=1.10.0 <1.11.0"
+  checksum: 10c0/4ff4e28966b7ae33cae40e7d52b502277dd8391ecd985ee5479aff9855e7e08589018f571dabf8699ba0eb2e5b2ac55877059f3d45571d851658f7bf8dc4f53e
   languageName: node
   linkType: hard
 
-"@opentelemetry/sdk-trace@npm:2.12.0":
-  version: 2.12.0
-  resolution: "@opentelemetry/sdk-trace@npm:2.12.0"
+"@opentelemetry/sdk-trace@npm:^3.1.1":
+  version: 3.1.1
+  resolution: "@opentelemetry/sdk-trace@npm:3.1.1"
   dependencies:
-    "@opentelemetry/core": "npm:2.12.0"
-    "@opentelemetry/resources": "npm:2.12.0"
+    "@opentelemetry/core": "npm:3.1.1"
+    "@opentelemetry/resources": "npm:3.1.1"
     "@opentelemetry/semantic-conventions": "npm:^1.29.0"
   peerDependencies:
-    "@opentelemetry/api": ">=1.3.0 <1.10.0"
-  checksum: 10c0/bc36d6c3c0a007e0e1f928a7e3c9e6e6d130e157efab8da36a07a624000c2fd25309c0a3d42d4e58f6b30ae233cddfbc6270c2a8566c9f611b9d964f5b7ea8a6
+    "@opentelemetry/api": ">=1.10.0 <1.11.0"
+  checksum: 10c0/9373d5fe28fb8fe8d40d4b37a67033b9976d8c1d732fb05885250ec716df1e47e2e0285ee4fdfe4c587a13d5c50af192d197a0ffb7ecdfa96905899d804ae6e1
   languageName: node
   linkType: hard