From 5a5dece396e52c689ad89e09a9b84c86d605bd64 Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Mon, 5 Oct 2026 13:27:13 +0200 Subject: [PATCH 1/6] fix(server-utils): Record amqplib consumer info before the first delivery --- .../tracing/amqplib/scenario-topic-noack.mjs | 66 +++++++++++++++++++ .../suites/tracing/amqplib/test.ts | 63 ++++++++++++++++++ .../server-utils/src/integrations/amqplib.ts | 58 ++++++++++++---- .../src/orchestrion/config/amqplib.ts | 10 ++- 4 files changed, 184 insertions(+), 13 deletions(-) create mode 100644 dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs new file mode 100644 index 000000000000..a2767f2612a4 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs @@ -0,0 +1,66 @@ +import * as Sentry from '@sentry/node'; +import amqp from 'amqplib'; + +// A topic exchange, so the routing key and the queue name differ. +const exchangeName = 'orders'; +const queueName = 'orders-worker'; +const routingKey = 'order.created.12345'; +const amqpUsername = 'sentry'; +const amqpPassword = 'sentry'; + +const AMQP_URL = `amqp://${amqpUsername}:${amqpPassword}@localhost:5672/`; + +// A `noAck` consumer never settles a message, so its span has to end when dispatch returns. +const NO_ACKNOWLEDGEMENT = { noAck: true }; + +(async () => { + const { connection, channel } = await connectToRabbitMQ(); + + await channel.assertExchange(exchangeName, 'topic', { durable: false, autoDelete: true }); + await channel.assertQueue(queueName, { durable: false, exclusive: false, autoDelete: true }); + await channel.bindQueue(queueName, exchangeName, 'order.created.#'); + + // Publish before consuming, so the broker sends `BasicConsumeOk` and the delivery back to back. + await Sentry.startSpan({ name: 'root span' }, async () => { + channel.publish(exchangeName, routingKey, Buffer.from(JSON.stringify({ foo: 'bar01' }))); + }); + + await consumeMessageFromQueue(queueName, channel); + + await channel.close(); + await connection.close(); +})(); + +async function connectToRabbitMQ() { + // The broker accepts TCP before AMQP handshakes succeed, so early connects can reject while it boots. + let lastError; + for (let attempt = 0; attempt < 5; attempt++) { + try { + const connection = await amqp.connect(AMQP_URL); + const channel = await connection.createChannel(); + return { connection, channel }; + } catch (err) { + lastError = err; + await new Promise(resolve => setTimeout(resolve, 1000)); + } + } + throw lastError; +} + +async function consumeMessageFromQueue(queueName, channel) { + return new Promise((resolve, reject) => { + channel + .consume( + queueName, + message => { + if (message) { + resolve(); + } else { + reject(new Error('No message received')); + } + }, + NO_ACKNOWLEDGEMENT, + ) + .catch(reject); + }); +} diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts index fade84b952a4..b929d8809e69 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts @@ -180,8 +180,71 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_ }, { additionalDependencies }, ); + + createEsmAndCjsTests( + __dirname, + 'scenario-topic-noack.mjs', + 'instrument.mjs', + (createTestRunner, test) => { + test('names a noAck consumer span after its queue and ends it as ok', { timeout: 60_000 }, async () => { + const receivedTransactions: TransactionEvent[] = []; + + await createTestRunner() + .expect({ + transaction: (transaction: TransactionEvent) => { + receivedTransactions.push(transaction); + }, + }) + .expect({ + transaction: (transaction: TransactionEvent) => { + receivedTransactions.push(transaction); + + const consumer = receivedTransactions.find( + t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.amqplib.consumer', + ); + + expect(consumer).toBeDefined(); + expect(consumer!.transaction).toBe('orders-worker process'); + expect(consumer!.contexts?.trace?.status).toBe('ok'); + expect(consumer!.contexts?.trace?.data?.['messaging.destination.name']).toBe('orders'); + expect(consumer!.contexts?.trace?.data?.['messaging.rabbitmq.destination.routing_key']).toBe( + 'order.created.12345', + ); + }, + }) + .start() + .completed(); + }); + }, + { additionalDependencies }, + ); }); + createEsmAndCjsTests( + __dirname, + 'scenario-topic-noack.mjs', + 'instrument-span-streaming.mjs', + (createTestRunner, test) => { + test('ends a streamed noAck consumer span as ok', { timeout: 60_000 }, async () => { + await createTestRunner() + .ignore('event') + .expect({ + span: container => { + const consumerSpan = container.items.find( + span => span.attributes['sentry.origin']?.value === 'auto.amqplib.consumer', + ); + expect(consumerSpan).toBeDefined(); + expect(consumerSpan!.status).toBe('ok'); + expect(consumerSpan!.name).toBe('process orders'); + expect(consumerSpan!.attributes['messaging.destination.name']?.value).toBe('orders'); + }, + }) + .start() + .completed(); + }); + }, + ); + createEsmAndCjsTests(__dirname, 'scenario.mjs', 'instrument-span-streaming.mjs', (createTestRunner, test) => { test('names streamed spans after the messaging conventions', { timeout: 60_000 }, async () => { await createTestRunner() diff --git a/packages/server-utils/src/integrations/amqplib.ts b/packages/server-utils/src/integrations/amqplib.ts index b67b546d8b2f..238af64b6780 100644 --- a/packages/server-utils/src/integrations/amqplib.ts +++ b/packages/server-utils/src/integrations/amqplib.ts @@ -68,6 +68,7 @@ const MESSAGE_STORED_SPAN: unique symbol = Symbol('sentry.amqplib.message.stored const CHANNEL_SPANS_NOT_ENDED: unique symbol = Symbol('sentry.amqplib.channel.spans-not-ended'); const CHANNEL_CONSUME_TIMEOUT_TIMER: unique symbol = Symbol('sentry.amqplib.channel.consume-timeout-timer'); const CHANNEL_CONSUMER_INFO: unique symbol = Symbol('sentry.amqplib.channel.consumer-info'); +const CHANNEL_PENDING_CONSUMERS: unique symbol = Symbol('sentry.amqplib.channel.pending-consumers'); const CHANNEL_IS_CONFIRM_PUBLISHING: unique symbol = Symbol('sentry.amqplib.channel.is-confirm-publishing'); const CONNECTION_ATTRIBUTES: unique symbol = Symbol('sentry.amqplib.connection.attributes'); @@ -108,12 +109,19 @@ interface ConsumerInfo { queue: string; } +/** A `consume` call waiting for the broker to assign it a consumer tag. */ +interface PendingConsumer { + callback: unknown; + info: ConsumerInfo; +} + interface ChannelLike { connection?: ConnectionLike; on?: (event: string, listener: (...args: unknown[]) => void) => unknown; [CHANNEL_SPANS_NOT_ENDED]?: { msg: ConsumeMessage; timeOfConsume: number }[]; [CHANNEL_CONSUME_TIMEOUT_TIMER]?: ReturnType; [CHANNEL_CONSUMER_INFO]?: Map; + [CHANNEL_PENDING_CONSUMERS]?: PendingConsumer[]; [CHANNEL_IS_CONFIRM_PUBLISHING]?: boolean; } @@ -156,6 +164,7 @@ function instrumentAmqplib(): void { subscribePublish(); subscribeConfirmPublish(); subscribeConsume(); + subscribeRegisterConsumer(); subscribeDispatch(); subscribeSettle(); } @@ -199,29 +208,55 @@ function subscribeConfirmPublish(): void { }); } -/** - * Records `consumerTag -> { noAck, queue }` when a consumer is registered, so the per-message - * dispatch hook can name the span after the queue and know when to end it. - */ +/** Stashes the `{ noAck, queue }` each `consume` asks for until the broker assigns it a tag. */ function subscribeConsume(): void { const channel = diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_CONSUME); - // A `start` subscriber is required for orchestrion to wrap `consume` at all. - channel.start.subscribe(NOOP); - channel.asyncEnd.subscribe(message => { + channel.start.subscribe(message => { const data = message as AmqpChannelContext; const consumerChannel = data.self; - const result = data.result as { consumerTag?: string } | undefined; - const consumerTag = result?.consumerTag; - if (!consumerChannel || !consumerTag) { + if (!consumerChannel) { return; } ensureChannelState(consumerChannel); const queueArg = data.arguments[0]; const queue = typeof queueArg === 'string' ? queueArg : ''; + const callback = data.arguments[1]; const options = data.arguments[2] as { noAck?: boolean } | undefined; - consumerChannel[CHANNEL_CONSUMER_INFO]?.set(consumerTag, { noAck: !!options?.noAck, queue }); + consumerChannel[CHANNEL_PENDING_CONSUMERS]?.push({ callback, info: { noAck: !!options?.noAck, queue } }); + }); +} + +/** + * Records `consumerTag -> { noAck, queue }` so the per-message dispatch hook can name the span after + * the queue and know when to end it. + * + * `consume`'s promise is too late for this: amqplib can decode `BasicConsumeOk` and the first delivery + * from one socket read, before any promise callback runs. It always registers a consumer before it + * dispatches to it. + */ +function subscribeRegisterConsumer(): void { + const channel = diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_REGISTER_CONSUMER); + + channel.start.subscribe(message => { + const data = message as AmqpChannelContext; + const consumerChannel = data.self; + const consumerTag = data.arguments[0]; + const callback = data.arguments[1]; + if (!consumerChannel || typeof consumerTag !== 'string') { + return; + } + + const pending = consumerChannel[CHANNEL_PENDING_CONSUMERS] ?? []; + // A failed `consume` leaves its entry behind, so pair by callback, not by position. + const index = pending.findIndex(entry => entry.callback === callback); + if (index === -1) { + return; + } + + const [entry] = pending.splice(index, 1); + consumerChannel[CHANNEL_CONSUMER_INFO]?.set(consumerTag, entry!.info); }); } @@ -356,6 +391,7 @@ function ensureChannelState(channel: ChannelLike): void { channel[CHANNEL_SPANS_NOT_ENDED] = []; channel[CHANNEL_CONSUMER_INFO] = new Map(); + channel[CHANNEL_PENDING_CONSUMERS] = []; const timer = setInterval(() => checkConsumeTimeoutOnChannel(channel), CONSUME_TIMEOUT_MS); timer.unref?.(); diff --git a/packages/server-utils/src/orchestrion/config/amqplib.ts b/packages/server-utils/src/orchestrion/config/amqplib.ts index a8f0e4fe61b4..2521d4546f33 100644 --- a/packages/server-utils/src/orchestrion/config/amqplib.ts +++ b/packages/server-utils/src/orchestrion/config/amqplib.ts @@ -29,13 +29,18 @@ export const amqplibConfig = [ module: { ...module, filePath: 'lib/channel_model.js' }, functionQuery: { className: 'ConfirmChannel', methodName: 'publish', kind: 'Callback' }, }, - // Records `consumerTag -> { noAck, queue }` so the per-message dispatch hook knows how to name and - // when to end the consumer span. + // Together record `consumerTag -> { noAck, queue }` so the per-message dispatch hook knows how to + // name and when to end the consumer span: `consume` knows the queue, `registerConsumer` the tag. { channelName: 'consume', module: { ...module, filePath: 'lib/channel_model.js' }, functionQuery: { className: 'Channel', methodName: 'consume', kind: 'Async' }, }, + { + channelName: 'registerConsumer', + module: { ...module, filePath: 'lib/channel.js' }, + functionQuery: { className: 'BaseChannel', methodName: 'registerConsumer', kind: 'Sync' }, + }, // Per delivered message: creates the consumer span and runs the user callback under it. { channelName: 'dispatch', @@ -83,6 +88,7 @@ export const amqplibChannels = { AMQPLIB_PUBLISH: 'orchestrion:amqplib:publish', AMQPLIB_CONFIRM_PUBLISH: 'orchestrion:amqplib:confirmPublish', AMQPLIB_CONSUME: 'orchestrion:amqplib:consume', + AMQPLIB_REGISTER_CONSUMER: 'orchestrion:amqplib:registerConsumer', AMQPLIB_DISPATCH: 'orchestrion:amqplib:dispatch', AMQPLIB_ACK: 'orchestrion:amqplib:ack', AMQPLIB_NACK: 'orchestrion:amqplib:nack', From 4fdc0fa289aa8060b05aacb72e8e38fc601beed0 Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Mon, 5 Oct 2026 14:05:42 +0200 Subject: [PATCH 2/6] update comments and address AI review --- .../tracing/amqplib/scenario-topic-noack.mjs | 2 +- .../server-utils/src/integrations/amqplib.ts | 16 +++++++--------- .../src/orchestrion/config/amqplib.ts | 4 ++-- 3 files changed, 10 insertions(+), 12 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs index a2767f2612a4..966b716ae01f 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs @@ -17,7 +17,7 @@ const NO_ACKNOWLEDGEMENT = { noAck: true }; const { connection, channel } = await connectToRabbitMQ(); await channel.assertExchange(exchangeName, 'topic', { durable: false, autoDelete: true }); - await channel.assertQueue(queueName, { durable: false, exclusive: false, autoDelete: true }); + await channel.assertQueue(queueName, { durable: false, exclusive: true }); await channel.bindQueue(queueName, exchangeName, 'order.created.#'); // Publish before consuming, so the broker sends `BasicConsumeOk` and the delivery back to back. diff --git a/packages/server-utils/src/integrations/amqplib.ts b/packages/server-utils/src/integrations/amqplib.ts index 238af64b6780..741c893ca414 100644 --- a/packages/server-utils/src/integrations/amqplib.ts +++ b/packages/server-utils/src/integrations/amqplib.ts @@ -229,12 +229,9 @@ function subscribeConsume(): void { } /** - * Records `consumerTag -> { noAck, queue }` so the per-message dispatch hook can name the span after - * the queue and know when to end it. - * - * `consume`'s promise is too late for this: amqplib can decode `BasicConsumeOk` and the first delivery - * from one socket read, before any promise callback runs. It always registers a consumer before it - * dispatches to it. + * Records `consumerTag -> { noAck, queue }` for the dispatch hook. `consume`'s promise resolves too late, + * because amqplib can decode `BasicConsumeOk` and the first delivery in one socket read. amqplib always + * registers a consumer before it dispatches to it. */ function subscribeRegisterConsumer(): void { const channel = diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_REGISTER_CONSUMER); @@ -251,12 +248,13 @@ function subscribeRegisterConsumer(): void { const pending = consumerChannel[CHANNEL_PENDING_CONSUMERS] ?? []; // A failed `consume` leaves its entry behind, so pair by callback, not by position. const index = pending.findIndex(entry => entry.callback === callback); - if (index === -1) { + const entry = pending[index]; + if (!entry) { return; } - const [entry] = pending.splice(index, 1); - consumerChannel[CHANNEL_CONSUMER_INFO]?.set(consumerTag, entry!.info); + pending.splice(index, 1); + consumerChannel[CHANNEL_CONSUMER_INFO]?.set(consumerTag, entry.info); }); } diff --git a/packages/server-utils/src/orchestrion/config/amqplib.ts b/packages/server-utils/src/orchestrion/config/amqplib.ts index 2521d4546f33..ea449400ea5d 100644 --- a/packages/server-utils/src/orchestrion/config/amqplib.ts +++ b/packages/server-utils/src/orchestrion/config/amqplib.ts @@ -29,8 +29,8 @@ export const amqplibConfig = [ module: { ...module, filePath: 'lib/channel_model.js' }, functionQuery: { className: 'ConfirmChannel', methodName: 'publish', kind: 'Callback' }, }, - // Together record `consumerTag -> { noAck, queue }` so the per-message dispatch hook knows how to - // name and when to end the consumer span: `consume` knows the queue, `registerConsumer` the tag. + // `consume` knows the queue and `noAck`, and `registerConsumer` knows the tag. Together they tell the + // dispatch hook how to name the consumer span and when to end it. { channelName: 'consume', module: { ...module, filePath: 'lib/channel_model.js' }, From 359a0715c47d762acc8b64751ba57d7524fe4a98 Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Tue, 6 Oct 2026 11:52:37 +0200 Subject: [PATCH 3/6] fix: drop entry when consume fails --- .../tracing/amqplib/scenario-topic-noack.mjs | 23 ++++++------- .../server-utils/src/integrations/amqplib.ts | 33 ++++++++++++++----- 2 files changed, 35 insertions(+), 21 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs index 966b716ae01f..819f1181cb1f 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-topic-noack.mjs @@ -49,18 +49,15 @@ async function connectToRabbitMQ() { async function consumeMessageFromQueue(queueName, channel) { return new Promise((resolve, reject) => { - channel - .consume( - queueName, - message => { - if (message) { - resolve(); - } else { - reject(new Error('No message received')); - } - }, - NO_ACKNOWLEDGEMENT, - ) - .catch(reject); + const onMessage = message => (message ? resolve() : reject(new Error('No message received'))); + + // Invalid `arguments` throw before the RPC, so the channel stays open for the next `consume`. + try { + channel.consume('orders-archive', onMessage, { arguments: 'invalid' }); + } catch { + // Expected. + } + + channel.consume(queueName, onMessage, NO_ACKNOWLEDGEMENT).catch(reject); }); } diff --git a/packages/server-utils/src/integrations/amqplib.ts b/packages/server-utils/src/integrations/amqplib.ts index 741c893ca414..81da73c32453 100644 --- a/packages/server-utils/src/integrations/amqplib.ts +++ b/packages/server-utils/src/integrations/amqplib.ts @@ -143,6 +143,11 @@ interface AmqpDispatchContext extends AmqpChannelContext { _sentryNoAck?: boolean; } +interface AmqpConsumeContext extends AmqpChannelContext { + // The entry this call queued, so the error hook can drop it. + _sentryPendingConsumer?: PendingConsumer; +} + interface AmqpConnectContext { arguments?: unknown[]; result?: unknown; @@ -210,10 +215,10 @@ function subscribeConfirmPublish(): void { /** Stashes the `{ noAck, queue }` each `consume` asks for until the broker assigns it a tag. */ function subscribeConsume(): void { - const channel = diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_CONSUME); + const channel = diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_CONSUME); channel.start.subscribe(message => { - const data = message as AmqpChannelContext; + const data = message as AmqpConsumeContext; const consumerChannel = data.self; if (!consumerChannel) { return; @@ -224,7 +229,18 @@ function subscribeConsume(): void { const queue = typeof queueArg === 'string' ? queueArg : ''; const callback = data.arguments[1]; const options = data.arguments[2] as { noAck?: boolean } | undefined; - consumerChannel[CHANNEL_PENDING_CONSUMERS]?.push({ callback, info: { noAck: !!options?.noAck, queue } }); + const entry = { callback, info: { noAck: !!options?.noAck, queue } }; + consumerChannel[CHANNEL_PENDING_CONSUMERS]?.push(entry); + data._sentryPendingConsumer = entry; + }); + + // Some failures, such as invalid `arguments`, leave the channel open. A stale entry would label the + // next consumer that shares this callback. + channel.error.subscribe(message => { + const data = message as AmqpConsumeContext; + if (data.self && data._sentryPendingConsumer) { + removePendingConsumer(data.self, data._sentryPendingConsumer); + } }); } @@ -245,19 +261,20 @@ function subscribeRegisterConsumer(): void { return; } - const pending = consumerChannel[CHANNEL_PENDING_CONSUMERS] ?? []; - // A failed `consume` leaves its entry behind, so pair by callback, not by position. - const index = pending.findIndex(entry => entry.callback === callback); - const entry = pending[index]; + const entry = consumerChannel[CHANNEL_PENDING_CONSUMERS]?.find(pending => pending.callback === callback); if (!entry) { return; } - pending.splice(index, 1); + removePendingConsumer(consumerChannel, entry); consumerChannel[CHANNEL_CONSUMER_INFO]?.set(consumerTag, entry.info); }); } +function removePendingConsumer(channel: ChannelLike, entry: PendingConsumer): void { + channel[CHANNEL_PENDING_CONSUMERS] = channel[CHANNEL_PENDING_CONSUMERS]?.filter(pending => pending !== entry); +} + /** * Per delivered message (`BaseChannel.prototype.dispatchMessage`): continues the producer's trace, * opens a CONSUMER span, and runs the user callback under it. Manual-ack consumers keep the span open From 2b5a531b0c39b444192f3d092d45037cbe092636 Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Tue, 6 Oct 2026 12:13:13 +0200 Subject: [PATCH 4/6] hook the whole file, instrument callback API --- .../tracing/amqplib/scenario-callback-api.mjs | 57 ++++++++++ .../suites/tracing/amqplib/test.ts | 28 +++++ .../src/orchestrion/config/amqplib.ts | 103 +++++++++--------- 3 files changed, 138 insertions(+), 50 deletions(-) create mode 100644 dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-callback-api.mjs diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-callback-api.mjs b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-callback-api.mjs new file mode 100644 index 000000000000..21fa2d678f28 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-callback-api.mjs @@ -0,0 +1,57 @@ +import * as Sentry from '@sentry/node'; +import amqp from 'amqplib/callback_api.js'; + +const queueName = 'callback-queue'; +const amqpUsername = 'sentry'; +const amqpPassword = 'sentry'; + +const AMQP_URL = `amqp://${amqpUsername}:${amqpPassword}@localhost:5672/`; + +(async () => { + const connection = await connectToRabbitMQ(); + const channel = await toPromise(cb => connection.createChannel(cb)); + await toPromise(cb => channel.assertQueue(queueName, { durable: false, exclusive: true }, cb)); + + const received = new Promise((resolve, reject) => { + channel.consume( + queueName, + message => { + if (message) { + channel.ack(message); + resolve(); + } else { + reject(new Error('No message received')); + } + }, + { noAck: false }, + err => err && reject(err), + ); + }); + + await Sentry.startSpan({ name: 'root span' }, async () => { + channel.sendToQueue(queueName, Buffer.from(JSON.stringify({ foo: 'bar01' }))); + }); + + await received; + + await toPromise(cb => channel.close(cb)); + await toPromise(cb => connection.close(cb)); +})(); + +async function connectToRabbitMQ() { + // The broker accepts TCP before AMQP handshakes succeed, so early connects can fail while it boots. + let lastError; + for (let attempt = 0; attempt < 5; attempt++) { + try { + return await toPromise(cb => amqp.connect(AMQP_URL, cb)); + } catch (err) { + lastError = err; + await new Promise(resolve => setTimeout(resolve, 1000)); + } + } + throw lastError; +} + +function toPromise(fn) { + return new Promise((resolve, reject) => fn((err, result) => (err ? reject(err) : resolve(result)))); +} diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts index b929d8809e69..cc2bf8f79fb6 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts @@ -275,4 +275,32 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_ .completed(); }); }); + + createEsmAndCjsTests( + __dirname, + 'scenario-callback-api.mjs', + 'instrument-span-streaming.mjs', + (createTestRunner, test) => { + test('instruments publish and ack on the callback API', { timeout: 60_000 }, async () => { + await createTestRunner() + .ignore('event') + .expect({ + span: container => { + const producerSpan = container.items.find( + span => span.attributes['sentry.origin']?.value === 'auto.amqplib.publisher', + ); + expect(producerSpan?.name).toBe('send callback-queue'); + + const consumerSpan = container.items.find( + span => span.attributes['sentry.origin']?.value === 'auto.amqplib.consumer', + ); + expect(consumerSpan?.name).toBe('process callback-queue'); + expect(consumerSpan?.status).toBe('ok'); + }, + }) + .start() + .completed(); + }); + }, + ); }); diff --git a/packages/server-utils/src/orchestrion/config/amqplib.ts b/packages/server-utils/src/orchestrion/config/amqplib.ts index ea449400ea5d..350184a692df 100644 --- a/packages/server-utils/src/orchestrion/config/amqplib.ts +++ b/packages/server-utils/src/orchestrion/config/amqplib.ts @@ -2,9 +2,11 @@ import type { InstrumentationConfig } from '../apmTypes'; import { getModuleNames } from './module-names'; -// `amqplib` splits its API across three files: -// - `lib/channel_model.js` holds `class Channel` (publish/consume/ack/nack/reject/…) and -// `class ConfirmChannel extends Channel` (a `publish` that takes a broker-confirm callback). +// `amqplib` splits its API across these files: +// - `lib/channel_model.js` (promise API) and `lib/callback_model.js` (callback API) each hold a +// `class Channel` (publish/consume/ack/nack/reject/…) and a `class ConfirmChannel extends Channel` +// (a `publish` that takes a broker-confirm callback). The method signatures match, so both files +// publish to the same channels. // - `lib/channel.js` holds `class BaseChannel` whose `dispatchMessage` invokes the registered // consumer callback once per delivered message — the natural per-message hook for consumer spans // (`Channel.consume` itself only registers the callback, it isn't called per message). @@ -15,27 +17,54 @@ import { getModuleNames } from './module-names'; const module = { name: 'amqplib', versionRange: '>=0.5.5 <3' } as const; export const amqplibConfig = [ - // Producer span + trace-header injection. `sendToQueue` delegates to `publish`, so it's covered. - { - channelName: 'publish', - module: { ...module, filePath: 'lib/channel_model.js' }, - functionQuery: { className: 'Channel', methodName: 'publish', kind: 'Sync' }, - }, - // Confirm-channel producer span; the trailing broker-confirm callback ends the span when the - // broker acks/nacks. It internally calls `super.publish`, so the subscriber guards against the - // base `publish` channel double-instrumenting. - { - channelName: 'confirmPublish', - module: { ...module, filePath: 'lib/channel_model.js' }, - functionQuery: { className: 'ConfirmChannel', methodName: 'publish', kind: 'Callback' }, - }, - // `consume` knows the queue and `noAck`, and `registerConsumer` knows the tag. Together they tell the - // dispatch hook how to name the consumer span and when to end it. - { - channelName: 'consume', - module: { ...module, filePath: 'lib/channel_model.js' }, - functionQuery: { className: 'Channel', methodName: 'consume', kind: 'Async' }, - }, + ...['lib/channel_model.js', 'lib/callback_model.js'].flatMap((filePath): InstrumentationConfig[] => [ + // Producer span + trace-header injection. `sendToQueue` delegates to `publish`, so it's covered. + { + channelName: 'publish', + module: { ...module, filePath }, + functionQuery: { className: 'Channel', methodName: 'publish', kind: 'Sync' }, + }, + // Confirm-channel producer span; the trailing broker-confirm callback ends the span when the + // broker acks/nacks. It internally calls the base `publish`, so the subscriber skips that nested call. + { + channelName: 'confirmPublish', + module: { ...module, filePath }, + functionQuery: { className: 'ConfirmChannel', methodName: 'publish', kind: 'Callback' }, + }, + // `consume` knows the queue and `noAck`, and `registerConsumer` knows the tag. Together they tell the + // dispatch hook how to name the consumer span and when to end it. + { + channelName: 'consume', + module: { ...module, filePath }, + functionQuery: { className: 'Channel', methodName: 'consume', kind: 'Async' }, + }, + // End the consumer span when the user settles the message. + { + channelName: 'ack', + module: { ...module, filePath }, + functionQuery: { className: 'Channel', methodName: 'ack', kind: 'Sync' }, + }, + { + channelName: 'nack', + module: { ...module, filePath }, + functionQuery: { className: 'Channel', methodName: 'nack', kind: 'Sync' }, + }, + { + channelName: 'reject', + module: { ...module, filePath }, + functionQuery: { className: 'Channel', methodName: 'reject', kind: 'Sync' }, + }, + { + channelName: 'ackAll', + module: { ...module, filePath }, + functionQuery: { className: 'Channel', methodName: 'ackAll', kind: 'Sync' }, + }, + { + channelName: 'nackAll', + module: { ...module, filePath }, + functionQuery: { className: 'Channel', methodName: 'nackAll', kind: 'Sync' }, + }, + ]), { channelName: 'registerConsumer', module: { ...module, filePath: 'lib/channel.js' }, @@ -47,32 +76,6 @@ export const amqplibConfig = [ module: { ...module, filePath: 'lib/channel.js' }, functionQuery: { className: 'BaseChannel', methodName: 'dispatchMessage', kind: 'Sync' }, }, - // End the consumer span when the user settles the message. - { - channelName: 'ack', - module: { ...module, filePath: 'lib/channel_model.js' }, - functionQuery: { className: 'Channel', methodName: 'ack', kind: 'Sync' }, - }, - { - channelName: 'nack', - module: { ...module, filePath: 'lib/channel_model.js' }, - functionQuery: { className: 'Channel', methodName: 'nack', kind: 'Sync' }, - }, - { - channelName: 'reject', - module: { ...module, filePath: 'lib/channel_model.js' }, - functionQuery: { className: 'Channel', methodName: 'reject', kind: 'Sync' }, - }, - { - channelName: 'ackAll', - module: { ...module, filePath: 'lib/channel_model.js' }, - functionQuery: { className: 'Channel', methodName: 'ackAll', kind: 'Sync' }, - }, - { - channelName: 'nackAll', - module: { ...module, filePath: 'lib/channel_model.js' }, - functionQuery: { className: 'Channel', methodName: 'nackAll', kind: 'Sync' }, - }, // Stashes connection attributes (url/host/port/protocol/server product) on the connection object // for span-time reads via `channel.connection`. { From b10ddb3dcc1ff480f9457a1ddbbbbb171b65e9cc Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Tue, 6 Oct 2026 14:19:43 +0200 Subject: [PATCH 5/6] fix AI pr comments --- .../suites/tracing/amqplib/scenario-callback-api.mjs | 12 +++--------- .../suites/tracing/amqplib/test.ts | 2 +- packages/server-utils/src/integrations/amqplib.ts | 4 ++-- .../server-utils/src/orchestrion/config/amqplib.ts | 5 +++-- 4 files changed, 9 insertions(+), 14 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-callback-api.mjs b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-callback-api.mjs index 21fa2d678f28..4870e99da9bd 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-callback-api.mjs +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/scenario-callback-api.mjs @@ -12,18 +12,12 @@ const AMQP_URL = `amqp://${amqpUsername}:${amqpPassword}@localhost:5672/`; const channel = await toPromise(cb => connection.createChannel(cb)); await toPromise(cb => channel.assertQueue(queueName, { durable: false, exclusive: true }, cb)); + // Without the info from `consume`, a `noAck` span stays open until the channel closes and ends as an error. const received = new Promise((resolve, reject) => { channel.consume( queueName, - message => { - if (message) { - channel.ack(message); - resolve(); - } else { - reject(new Error('No message received')); - } - }, - { noAck: false }, + message => (message ? resolve() : reject(new Error('No message received'))), + { noAck: true }, err => err && reject(err), ); }); diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts index cc2bf8f79fb6..78379052ecce 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts @@ -281,7 +281,7 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_ 'scenario-callback-api.mjs', 'instrument-span-streaming.mjs', (createTestRunner, test) => { - test('instruments publish and ack on the callback API', { timeout: 60_000 }, async () => { + test('instruments publish and noAck consume on the callback API', { timeout: 60_000 }, async () => { await createTestRunner() .ignore('event') .expect({ diff --git a/packages/server-utils/src/integrations/amqplib.ts b/packages/server-utils/src/integrations/amqplib.ts index 646d1b26cb89..9768cf27a26c 100644 --- a/packages/server-utils/src/integrations/amqplib.ts +++ b/packages/server-utils/src/integrations/amqplib.ts @@ -234,8 +234,8 @@ function subscribeConsume(): void { data._sentryPendingConsumer = entry; }); - // Some failures, such as invalid `arguments`, leave the channel open. A stale entry would label the - // next consumer that shares this callback. + // A synchronous throw, such as for invalid `arguments`, leaves the channel open. RPC failures close it. + // A stale entry would label the next consumer that shares this callback. channel.error.subscribe(message => { const data = message as AmqpConsumeContext; if (data.self && data._sentryPendingConsumer) { diff --git a/packages/server-utils/src/orchestrion/config/amqplib.ts b/packages/server-utils/src/orchestrion/config/amqplib.ts index 350184a692df..184844a8d799 100644 --- a/packages/server-utils/src/orchestrion/config/amqplib.ts +++ b/packages/server-utils/src/orchestrion/config/amqplib.ts @@ -32,11 +32,12 @@ export const amqplibConfig = [ functionQuery: { className: 'ConfirmChannel', methodName: 'publish', kind: 'Callback' }, }, // `consume` knows the queue and `noAck`, and `registerConsumer` knows the tag. Together they tell the - // dispatch hook how to name the consumer span and when to end it. + // dispatch hook how to name the consumer span and when to end it. `Sync`, because the callback API + // returns the channel and only a synchronous throw leaves the channel open. { channelName: 'consume', module: { ...module, filePath }, - functionQuery: { className: 'Channel', methodName: 'consume', kind: 'Async' }, + functionQuery: { className: 'Channel', methodName: 'consume', kind: 'Sync' }, }, // End the consumer span when the user settles the message. { From c546f20b38c2c1b782eb04837128e5f1e0d6b96c Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Fri, 9 Oct 2026 10:19:49 +0200 Subject: [PATCH 6/6] fix test --- .../suites/tracing/amqplib/test.ts | 101 +++--------------- 1 file changed, 15 insertions(+), 86 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts index 20a65f16783b..cd86d01459ec 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts @@ -174,28 +174,19 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_ 'scenario-topic-noack.mjs', 'instrument.mjs', (createTestRunner, test) => { - test('names a noAck consumer span after its queue and ends it as ok', { timeout: 60_000 }, async () => { - const receivedTransactions: TransactionEvent[] = []; - + test('ends a noAck consumer span as ok', { timeout: 60_000 }, async () => { await createTestRunner() .expect({ - transaction: (transaction: TransactionEvent) => { - receivedTransactions.push(transaction); - }, - }) - .expect({ - transaction: (transaction: TransactionEvent) => { - receivedTransactions.push(transaction); - - const consumer = receivedTransactions.find( - t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.amqplib.consumer', + span: container => { + const consumer = container.items.find( + span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.amqplib.consumer', ); expect(consumer).toBeDefined(); - expect(consumer!.transaction).toBe('orders-worker process'); - expect(consumer!.contexts?.trace?.status).toBe('ok'); - expect(consumer!.contexts?.trace?.data?.['messaging.destination.name']).toBe('orders'); - expect(consumer!.contexts?.trace?.data?.['messaging.rabbitmq.destination.routing_key']).toBe( + expect(consumer!.name).toBe('process orders'); + expect(consumer!.status).toBe('ok'); + expect(consumer!.attributes[MESSAGING_DESTINATION_NAME]?.value).toBe('orders'); + expect(consumer!.attributes[MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY]?.value).toBe( 'order.created.12345', ); }, @@ -208,87 +199,25 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_ ); }); - createEsmAndCjsTests( - __dirname, - 'scenario-topic-noack.mjs', - 'instrument-span-streaming.mjs', - (createTestRunner, test) => { - test('ends a streamed noAck consumer span as ok', { timeout: 60_000 }, async () => { - await createTestRunner() - .ignore('event') - .expect({ - span: container => { - const consumerSpan = container.items.find( - span => span.attributes['sentry.origin']?.value === 'auto.amqplib.consumer', - ); - expect(consumerSpan).toBeDefined(); - expect(consumerSpan!.status).toBe('ok'); - expect(consumerSpan!.name).toBe('process orders'); - expect(consumerSpan!.attributes['messaging.destination.name']?.value).toBe('orders'); - }, - }) - .start() - .completed(); - }); - }, - ); - - createEsmAndCjsTests(__dirname, 'scenario.mjs', 'instrument-span-streaming.mjs', (createTestRunner, test) => { - test('names streamed spans after the messaging conventions', { timeout: 60_000 }, async () => { + createEsmAndCjsTests(__dirname, 'scenario-callback-api.mjs', 'instrument.mjs', (createTestRunner, test) => { + test('instruments publish and noAck consume on the callback API', { timeout: 60_000 }, async () => { await createTestRunner() - .ignore('event') .expect({ span: container => { - // `sendToQueue` publishes to the default exchange, which has no name. Its routing key is the - // queue name, so it is the destination rather than per-message data. - for (const origin of ['auto.amqplib.publisher', 'auto.amqplib.consumer']) { - const span = container.items.find(item => item.attributes['sentry.origin']?.value === origin); - expect(span).toBeDefined(); - expect(span!.attributes['messaging.destination.name']?.value).toBe('queue1'); - expect(span!.attributes['messaging.rabbitmq.destination.routing_key']?.value).toBe('queue1'); - } - const producerSpan = container.items.find( - span => span.attributes['sentry.origin']?.value === 'auto.amqplib.publisher', + span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.amqplib.publisher', ); - expect(producerSpan!.name).toBe('send queue1'); + expect(producerSpan?.name).toBe('send callback-queue'); const consumerSpan = container.items.find( - span => span.attributes['sentry.origin']?.value === 'auto.amqplib.consumer', + span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.amqplib.consumer', ); - expect(consumerSpan!.name).toBe('process queue1'); + expect(consumerSpan?.name).toBe('process callback-queue'); + expect(consumerSpan?.status).toBe('ok'); }, }) .start() .completed(); }); }); - - createEsmAndCjsTests( - __dirname, - 'scenario-callback-api.mjs', - 'instrument-span-streaming.mjs', - (createTestRunner, test) => { - test('instruments publish and noAck consume on the callback API', { timeout: 60_000 }, async () => { - await createTestRunner() - .ignore('event') - .expect({ - span: container => { - const producerSpan = container.items.find( - span => span.attributes['sentry.origin']?.value === 'auto.amqplib.publisher', - ); - expect(producerSpan?.name).toBe('send callback-queue'); - - const consumerSpan = container.items.find( - span => span.attributes['sentry.origin']?.value === 'auto.amqplib.consumer', - ); - expect(consumerSpan?.name).toBe('process callback-queue'); - expect(consumerSpan?.status).toBe('ok'); - }, - }) - .start() - .completed(); - }); - }, - ); });