Implement consumer-timeout for classic queue type

[the-mikedavis]

Is your feature request related to a problem? Please describe.

v4.3 moved the consumer timeout ("message locks") feature from rabbit_channel (AMQP 0-9-1 only) to the queue type (release notes). Quorum queues and the JMS queue type implement consumer timeouts but not classic queues or streams. Client applications might implicitly rely on the consumer timeout feature to take the message back from a completely stuck consumer, so that other consumers may handle it. Ideally classic queue consumers could get the same improved basic.cancel on a timeout if supported by the client.

Describe the solution you'd like

We can look at implementing the consumer timeout concept for the classic queue type.

Describe alternatives you've considered

It would be hacky but we could add back the concept of consumer timeout to rabbit_channel so that stuck consumers see the same behavior as v4.2.x and below. That feels like going backwards though.

Additional context

I don't think the same is necessary for stream queues because of the non-destructive consumption.

[the-mikedavis]

I think there's a sublety to msg_id in CQs versus QQs that makes this tricky to implement. rabbit_classic_queue:settle/5 doesn't pass along the consumer tag, just the message IDs. So if there's a timeout and that message is checked out to another channel, a stale ack/settle from the timed-out channel would settle that other consumer's checkout of the same message. And it gets more complicated if the same message is delivered back to the channel that timed out on the first attempt.

[Payal2000]

Following up on the msg_id subtlety, I want to make sure I understand the failure mode before suggesting anything.

Since acks travel client → channel → queue, the channel does learn about the timeout via the cancel, so it could invalidate the affected delivery tags locally and drop a late ack from the client before it ever reaches the queue. That doesn't close the window completely: an ack already in the queue process's mailbox when the requeue happens would still land and alias onto the new checkout. Is that residual race the thing that makes channel-side invalidation insufficient, or is there a case I'm missing where it fails more broadly?

If per-delivery ack identity is the answer, is changing the rabbit_queue_type settle callback on the table (with a feature flag for mixed-version clusters), or would you rather keep the change contained inside rabbit_classic_queue/rabbit_amqqueue_process?

Also curious whether you see this landing in main only, or whether there's any appetite for a stopgap on 4.3.x so upgrades from 4.2.x aren't silently affected.

[the-mikedavis]

Actually the delivery tags are channel-local counters, so the ambiguity doesn't reach the client, just the seem between the channel and queue. The channel forwards the raw message tag/ID back to the queue when settling. So the channel needs a concept of a checkout's generation/attempt along with the message ID or else the message targetted by a settle can be ambiguous.

I have a change locally (edit: #17037) that starts with a refactor to map an ack_id() to the raw SeqId and to use that ack_id() as the msg_id instead. This is closer to what quorum queues do and it makes the settle unambiguous. This msg_id type is opaque, so it can change without needing a change to rabbit_queue_type, rabbit_channel or rabbit_amqp_session. It's not an especially large change in terms of lines, but it's basically a rewrite of rabbit_queue_consumers, so I think it would be more appropriate for a minor version than a patch that ends up in v4.3.x. I think it would be useful to have consumer timeout on v4.3 though.