Buffered endpoints restart the listener immediately after back pressure is tripped regardless of the restart parameter

[IWillDonut]

Describe the bug

Apologies if I'm misunderstanding the buffering semantics. The following is based on how I've interpreted the documentation and related blog posts.

After tripping BufferingLimits::Maximum, the BackPressureAgent appears to restart the listener on the next invocation of BackPressureAgent::CheckNowAsync regardless of BufferingLimits::Restart. Note, this is not a bug in the BackPressureAgent itself.

Given the comment in the code, I suspect there is a very good reason that ListeningAgent::_receiver is disposed of and nulled out here. However, this has a side effect of driving _agent.QueueCount to 0 regardless of the state of the BufferedReciever.

To Reproduce
I haven't had time to attempt to write a test that might cover this, but I can try to work on that in the next few days.

Steps to reproduce the behavior:

1. Create a minimal project with a single buffered endpoint with low BufferingLimits (I tested with RabbitMQ)
2. Add a single handler for a test message that delays for X seconds
3. Flood the queue with enough messages to trip the back pressure limits
4. Observe the listener is restarted on the next tick of the BackPressureAgent::CheckNowAsync regardless of the receiver state

Expected behavior
A buffered endpoint should only restart the listener once the the receiver's queue drops below the BufferingLimits::Restart threshold.

Desktop:

• OS: Windows 11
• Version 4.12.2

Additional context
This might have some additional unintended consequences where multiple receivers are active for a listening agent. The disposed receiver appears to continue processing its internal queue in parallel with the receiver that's created on listener restart.

[jeremydmiller]

Sigh, sounds like an unintended bug. If there's anyway you could do a PR w/ a test that reproduces the issue, that would be very helpful. Or I can look through the tests for that. They're flaky and slow, so don't get put in CI sadly enough.

[IWillDonut]

Of course! Might not be able to get to it for a day or two, but I'll see what I can do.

[jeremydmiller]

@IWillDonut I'm reversing the code you saw -- and I'm more than a little sad this took this long to find:( I'll deal w/ whatever test failures pop up from this.

[konigm97]

I'm sorry if I'm misunderstanding any of the principles, but I'm trying to do the following:

• Use batching for a certain messages type. The batching of the type is only on one queue
• Buffer in memory, but set limits

I'm working with RabbitMq. And currently I just want to test with as much limits as possible.

Queue and batching configuration (see the image below):

My handler has a single line of code: await Task.Delay(1000) - so, just waiting for 1 sec. Naturally, with very restricted batching of just 1 message per batch, and with Buffer Limits of max: 2, restart: 1, queue on RabbitMq should drop really slowly (for sure not instantly). But please find the image below showing the RabbitMq queue which I'm talking about:

As you can see, the queue dropped really quickly. Now, I don't know whether I'm doing something wrong or the bug still persists. I even tried setting prefetch size to 1, the result's didn't change.

How I tested:

• I had my service offline until I've filled the queue up, then I started my service (consumer) later on.
• I pulled the main branch of this repository and added Wolverine.csproj and Wolverine.RabbitMQ.csproj.
• Also I've checked that the ListeningAgent.cs has the fix mentioned in the commit on this thread.

This issue (or my misconfiguration) is causing the service to run out of memory, as it get's bombarded with messages more that it can handle at the time.

PS: @jeremydmiller - I didn't know whether I had to open another issue or not, please do as you find fitting.