Add possible cancel of DeferredConfirmation

[sapk]

In case of network error (supposed), a program using DeferredConfirmation can wait indefinitely on .Wait() method.
Linked issue in benthos: redpanda-data/connect#1299

The program can implement a timeout over the wait method but it will not clear it in amqp lib from confirms/deferredConfirmations of the channel.

First, is ok for you to add a cancel solution (new method, add context, ...) ?
If yes, I see many possible solution:

• Add cancellable context to PublishWithDeferredConfirm and clear linked confirmation at context cancel
• Add a CancelConfirm method to Channel that would call confirm.One (or new method Cancel)
• Add cancellable context to (d *DeferredConfirmation) Wait() (this need change so that it can clear it self from deferredConfirmations)
• Add Cancel method to DeferredConfirmation (this need change so that it can clear it self from deferredConfirmations)

Which one suit you the most ? or if you have an other to suggest ?

[lukebakken]

Is using .Wait() the right thing to do here?

You are synchronously waiting on a publisher confirmation. For the highest performance, our recommendation is to keep track of an acceptable number of outstanding confirmations, perhaps throttling your publishing if this number reaches a threshold.

But, we still need to figure out how to cancel a Wait(). Seems like this is the idomatic way - https://pkg.go.dev/context#pkg-overview. Though that may be a pretty major change in this library.

[sapk]

I am not sure at 100% but I think benthos spawn multiple write that wait each on their confirmation.
It seems to have a common abstraction of an asyncwriter over this part of code meaning it can have a certain number of call to (a *amqp09Writer) WriteWithContext in parallel defined by MaxInFlight configuration.

I see two common ways to do it in go, the net/http way with a context at query level or io way with Close method. The context seems to be the more common and allow for more advanced use case.

Briefly looking at the code, I would say the easiest would be to add a context to PublishWithDeferredConfirm and start a goroutine in (c *confirms) Publish() or (d *deferredConfirmations) Add(tag uint64) that end on confirmation (d *DeferredConfirmation) Wait() or context.Done() to clean the confirmation.