Make handling of publisher confirmations transparent to the user

[lukebakken]

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

Mentioned here.

cc @danielmarbach @bording @Tornhoof

Describe the solution you'd like

• Creating a channel via CreateChannelAsync will automatically enable publisher confirmations for the channel, and will track them. A parameter will be introduced to disable this tracking.
• All BasicPublishAsync calls will have publisher confirmations tracked, and the mandatory flag will also be set to true. A confirmation must return within a certain period of time (configurable via the channel?) for the publish to be considered successful. Otherwise, a timeout exception will be raised.
• There will be a maximum count of outstanding confirmations allowed (also configurable). When this maximum is reached, calls to BasicPublishAsync will immediately return an exception.
• (optional) when the current count of outstanding confirmations starts to near the maximum, delays could be introduced in publish attempts to allow the broker to catch up.

Describe alternatives you've considered

Users will be allowed to disable automatic tracking and do it themselves via BasicAcksAsync and BasicNacksAsync

Additional context

#1676 (comment)

[bording]

This seems like a good list of features/changes for the client to make regarding publisher confirms now that everything is async.

After the investigation Daniel and I did regarding #1676, I have some observations about the existing confirmation tracking that I think make sense to include here.

The current confirmation tracking that was added relies on the separate use of the WaitForConfirms* APIs. This API structure is a leftover relic of the non-async past of this client, and integrating the tracking directly into the BasicPublish call as this issue suggests is definitely the right call.

When the tracking is moved into the BasicPublish, it enables the ability to track confirmations on per-message basis, which the current tracking code does not do.

We added this sort of per-message tracking to the NServiceBus RabbitMQ transport a while back, and you can see how this works here.

All BasicPublishAsync calls will have publisher confirmations tracked, and the mandatory flag will also be set to true

It's interesting that you mention the mandatory flag here, because that's another area that the current tracking code seems to be deficient.

Any BasicReturn messages that are sent from the broker need to also be accounted for when doing confirmation tracking in the BasicPublish calls. Otherwise, the BasicReturn is ignored, and then the BasicPublish would process the ack and indicate the message was published.

In the tracking code we added to the transport, we handle the following things and map them back to the TaskCompletionSource:

• ack
• nack
• return
• channel shutdown

The complication about handling BasicReturn messages is that the broker does not currently return the confirmation id for correlation like ack/nack do. We solved this in our transport by recording the sequence number in a header that we can extract from the returned message. I'm not sure how the client would be able to do this, but maybe a header wouldn't be a terrible option here either? Of course it would be great if the broker could include the id in the return message, maybe include it in the ReplyText? No clue how feasible such a change would be.

[lukebakken]

Of course it would be great if the broker could include the id in the return message, maybe include it in the ReplyText? No clue how feasible such a change would be.

I will look into this today.

[lukebakken]

Yes, I can see where adding the delivery tag to basic.return would be possible, but the protocol doesn't allow adding that information.

However, the protocol does guarantee that if a basic.return is sent to the client, the broker MUST immediately send basic.ack for that same message, which would allow correlation, client-side, in a very awkward way.

[bording]

Yes, I can see where adding the delivery tag to basic.return would be possible, but the protocol doesn't allow adding that information.

Can you explain a bit more about what this means? What's preventing it from being added to the existing text field of ReplyText? That would require parsing the text to find it, but that's not too terrible.

Is there something that makes the id just not available at all?

[lukebakken]

I just added this comment when you commented 😸 👍

AMQP 0.9.1 does guarantee that if a basic.return is sent to the client, the broker MUST immediately send basic.ack for that same message, which would allow correlation, client-side, in a very awkward way.

[lukebakken]

Here is where basic.return is constructed:

https://github.com/rabbitmq/rabbitmq-server/blob/main/deps/rabbit/src/rabbit_channel.erl#L1832-L1842

The Reason parameter is always not_found in this case, so the following function call...

https://github.com/rabbitmq/rabbitmq-server/blob/main/deps/rabbit/src/rabbit_channel.erl#L1836

...always returns NO_ROUTE for ReplyText. I suppose we could add more information to that field, but there's probably some application, somewhere, that depends on the existing behavior.

Adding a header is the best way to address this, or depend on the way the protocol works as I described.

[bording]

Adding a header is the best way to address this, or depend on the way the protocol works as I described.

Relying on the return + ack feels like something really hard to achieve. The header is a bit of a hack, but at least it's reliable and easy to implement.

I understand the concerns around adding something to the ReplyText, but that would be such a nice improvement!

[lukebakken]

Relying on the return + ack feels like something really hard to achieve

Maybe not ... ??? Thoughts? #1686

Update: nope, no guarantee of when the ack happens (https://www.rabbitmq.com/resources/specs/amqp0-9-1.extended.xml)

[bording]

Maybe not ... ??? Thoughts? #1686

Update: nope, no guarantee of when the ack happens (https://www.rabbitmq.com/resources/specs/amqp0-9-1.extended.xml)

I was going to say that I wasn't aware the broker had that strong of a guarantee of the sequencing of those, just that return would happen before ack, but I guess it doesn't!

[lukebakken]

I mean, the spec does say

Unroutable mandatory or immediate messages are acknowledged right after the Basic.Return method

...and this is what I see in my test program. I'm tracing the code rabbit_channel to see if this is in fact guaranteed.

I think the time guarantees section of the spec applies to "normal" acks, and not those that happen after basic.return cc @michaelklishin @kjnilsson @ansd may have some input.

Yes, it does appear that the state flow in rabbit_channel does guarantee that basic.ack for a basic.return-ed message will happen next but I want a second opinion.

[michaelklishin]

The docs say the same thing, basic.return is sent before a (publisher confirms) basic.ack.

[bording]

Publisher confirms are a RabbitMQ-specific extension of the AMQP spec, so I'm not sure how much the spec itself would help here anyway.

It does seem prudent to know what the broker actually guarantees and if that's actually something "guaranteed" or just an implementation detail.

The spike PR was relying on an assumption that the matching basic.ack would be the very next thing received on the channel after the basic.return with no possibility of there being something else in between.

[lukebakken]

Publisher confirms are a RabbitMQ-specific extension of the AMQP spec, so I'm not sure how much the spec itself would help here anyway.

I forgot about that bit 🤦

The spike PR was relying on an assumption that the matching basic.ack would be the very next thing received on the channel after the basic.return with no possibility of there being something else in between.

Yep, wouldn't that be nice?

[kjnilsson]

Shouldn't BasicPublishAsync block/wait when the maximum outstanding has been reached until the pending confirms fall below the limit?

[ansd]

#1682 (comment)

@lukebakken correct, RabbitMQ replies with basic.return before basic.ack (basic.nack).

Code:
https://github.com/rabbitmq/rabbitmq-server/blob/77bf0ad75fff178663527a8232bb97b095170093/deps/rabbit/src/rabbit_channel.erl#L2076-L2080

Documentation:
https://www.rabbitmq.com/docs/amqp#outcomes

[danielmarbach]

It seems the client currently doesn't set automatically a message id if non has been set. If we had a message ID that could be used to track states of the return and ACK/NACK sequence by having a secondary index from publish sequence number to IDs.

But then I wonder if it wouldn't be better to enforce a sequence number header when the confirms mode is enabled starting from v7. The overhead of that header is super minimal from a size perspective and comparably cheap compared to the decision of doing confirms anyway.

[danielmarbach]

So what about the following (raw ideas for discussion)

Change the ConfirmSelectAsync method to something like

        /// <summary>
        /// Asynchronously enable publisher confirmations.
        /// </summary>
        /// <param name="maximumNumberOfOutstandingConfirmations">Set to <c>null</c> if tracking via <see cref="BasicAcksAsync"/> and <see cref="BasicNacksAsync"/> yourself.</param>
        /// <param name="cancellationToken">CancellationToken for this operation.</param>
        Task ConfirmSelectAsync(ushort? maximumNumberOfOutstandingConfirmations = 10,
            CancellationToken cancellationToken = default);

by default it is set to some upper number of confirms that can be pending. If you want to track it yourself you can set it to null.

Internally the channel then makes sure the sequence number acquisition is properly guarded and enforcing the upper limit of number of oustanding confirms. When the upper limits are reached further publish attempts are hold up until for slots are available or the cancellation token triggers.

ConfirmSelectAsync would throw when it is called with new values when there are outstanding confirms.

In non confirms mode the publishes behave like today. In confirms mode they behave in the following way:

• When the number of outstanding confirms is below the threshold the publish only returns when either ACK, NACK or BasicReturn happened. Throwing in all cases when there is no ackknowledgement or when the cancellation token triggers
• When the number of oustanding confirms is reached the publish waits for available slots or until the cancellation token is triggered
• In all exception cases the channel is never closed. Making that judgement is something the user can decide by calling CloseAsync
• The sequence number is added to the properties using a well established known x- header

The methods WaitForConfirmsAsync and WaitForConfirmsOrDieAsync will be removed from the channel

Subsequent major releases might flip the default to have confirms enabled but that decision can be postponed.

[danielmarbach]

I'm also starting to wonder now that acquiring the sequence number is async whether that method should return a scope that makes sure the sequence number cannot be acquired while the scope is active.

That would give people tracking acks and nacks manually a chance to have guaranteed exclusive access to the sequence number value on the channel for the duration of the scope (internally acquiring / releasing the consumer semaphore)

[danielmarbach]

I think it would also make sense to configure the confirms defaults directly during the async channel creation instead of having a dedicated method and deal with all the ramifications of guarding against the value flipping mid way

[michaelklishin]

@danielmarbach there is no way to disable publisher confirms once enabled. But I have no objections to having it as a constructor argument (or equivalent) at channel creation time.

[danielmarbach]

Sorry I was thinking about the tracking but wrote confirms defaults (which is ambiguous) because you can call the method multiple times with different flag values today

[lukebakken]

@danielmarbach @bording I've done a little bit of work to get the ball rolling here:

#1687

Basically, just start working to remove the use of ConfirmSelectAsync, but still use the existing confirmation tracking code.