Repository navigation
Conversation
|
bugbot run |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
size-limit report 📦
|
isaacs
left a comment
isaacs
left a comment
There was a problem hiding this comment.
I'd recommend adding the error handler, but this looks good. Even if all the suggestions get put off to follow-up PRs, it's a net improvement.
| 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]; |
There was a problem hiding this comment.
A failed consume on a live channel can poison a later consume that reuses the same callback.
(Also relevant to line 227 above)
Pairing by callback avoids the problem only when the next consume uses a different callback. Sharing one handler across several queues is a common pattern.
So the sequence is:
channel.consume(badName, handler)rejects, but the channel stays open. This happens for client-side encode errors, for example a queue name orconsumerTagover 255 bytes (defs.jsthrowsTypeError: Field 'queue' is the wrong type). Server-side errors (404, 403) close the channel, so they are harmless here.channel.consume('jobs', handler, { noAck: true })succeeds.registerConsumerfinds the stale entry first (findIndexreturns the oldest match) and records{ queue: badName, noAck: false }for thejobstag.
Every message on jobs then gets the wrong static span name, and its spans stay open until the 1-minute timeout or channel close, then end as errors.
That's the same bug being fixed here, but narrower.
Suggestion: drop the entry when the consume call fails. The tracing-channel context object is the same object for start and error, so keep a reference to the entry on it:
channel.start.subscribe(message => {
// ...
const entry = { callback, info: { noAck: !!options?.noAck, queue } };
consumerChannel[CHANNEL_PENDING_CONSUMERS]?.push(entry);
(data as AmqpConsumeContext)._sentryPendingConsumer = entry;
});
channel.error.subscribe(message => {
const data = message as AmqpConsumeContext;
const pending = data.self?.[CHANNEL_PENDING_CONSUMERS];
const index = pending?.indexOf(data._sentryPendingConsumer!) ?? -1;
if (index >= 0) pending!.splice(index, 1);
});| // `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', |
There was a problem hiding this comment.
The callback API is still affected by the bug, right? Only lib/channel_model.js's Channel.consume is hooked, and lib/callback_model.js has its own Channel.consume (it also passes callback unchanged to registerConsumer).
Dispatch runs through the shared BaseChannel, so callback-API consumers get consumer spans but never get consumer info. Their spans always use the routing key as the static name, and noAck spans always leak until close and end as errors.
This PR makes the fix cheap: subscribeConsume uses only start, so one more config entry for lib/callback_model.js with kind: 'Sync' would be enough, and the new registerConsumer hook already covers both APIs.
Could be a follow-up PR (with an integration scenario using amqplib/callback_api), or include it here since the hook is shared now.
There was a problem hiding this comment.
I did it in this PR and added a test.
The classes and signatures from callback_model.js match channel_model.js, so the files share one flatMap and also the same channels.
s1gr1d
changed the title
![sentry[bot]](https://avatars.githubusercontent.com/in/12637?s=60&v=4){kind=small}
|
bugbot run |
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 2 potential issues.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Want reviews to match your repository better? Bugbot Learning can learn team-specific rules from PR activity. A team admin can enable Learning in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 2b5a531. Configure here.
This branch has not been deployed
3 participants
When a worker subscribes to a queue that already has messages, amqplib can decode
BasicConsumeOkand the first message in one socket read. We recorded each consumer's queue andnoAckflag whenconsume()resolved. That happens too late:The first message's span got the routing key as its name, and
noAckspans ended as errors at channel close. Now we record the info inregisterConsumer, which amqplib always calls before dispatch.This PR also instruments the callback API (
lib/callback_model.js). It has the same classes and signatures as the promise API.Follow-up on this PR (noticed the bug while working on this): #23652