Add error handling best practices to documentation.

[TroelsL]

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

We are using the client in production, and we ran into an issue where we have lost connectivity, but the reliable consumer keeps retrying indefinitely, and we have discovered no immediate way to detect that we have a serious problem.

We have looked at the source code and attempted to create a custom ReconnectStrategy, but haven't readily been able to find a good solution.

Please feel free to propose a user-space fix for this as well. This is quite critical to us that we can detect message losses. However, this is also a request to add documentation describing how to handle what I assume is a very common scenario. The answer may very well be that we should be using the RawProducer/Consumers instead of the Reliable ones.

Describe the solution you'd like

In the event of a longer disconnect from the server, I want to raise an exception or at least be able to get a status on the fact that I am not connected to any stream.

We tried producer.IsOpen(), !System.IsClosed, etc. But they all seem to show no sign of trouble. Private member Producer._inReconnection seems to contain some information.

Some advice on this would be greatly appreciated.

Describe alternatives you've considered

No response

Additional context

No response

[Gsantomaggio]

Client version?

keeps retrying indefinitely,

Can you please post some logs? With this kind of behaviour?

We tested the connection lost in different scenarios; it would be helpful to understand why the client could not reconnect to the server.

[Gsantomaggio]

In the event of a longer disconnect from the server, I want to raise an exception or at least be able to get a status on the fact that I am not connected to any stream.

Maybe we can add a callback with the status Producer._inReconnection

Btw you should have the messages in the messages in not confirmed status in the ConfirmationHandler here:

     ConfirmationHandler = confirmation =>
                    {
                        switch (confirmation.Status)
                        {
 ....

[TroelsL]

Thanks for responding so quickly. Here is the log

dbug: RabbitMQ.Stream.Client.Reliable.Producer[0]
      Server properties: [cluster_name, <SUBSTITUTED_HOSTNAME2>], [copyright, Copyright (c) 2007-2023 VMware, Inc. or its affiliates.], [information, Licensed under the MPL 2.0. Website: https://rabbitmq.com], [platform, Erlang/OTP 25.3], [product, RabbitMQ], [version, 3.11.10]
dbug: RabbitMQ.Stream.Client.Reliable.Producer[0]
      Sasl mechanism: PLAIN, AMQPLAIN
dbug: RabbitMQ.Stream.Client.Reliable.Producer[0]
      Open: ConnectionProperties: [advertised_port, 5552], [advertised_host, <SUBSTITUTED_HOSTNAME>]
dbug: RabbitMQ.Stream.Client.Reliable.Producer[0]
      Extracted BrokerVersion version: 3.11.10
info: RabbitMQ.Stream.Client.Reliable.Producer[0]
      Producer reference: , stream: TML_TEST_STREAM  reconnected successfully
dbug: RabbitMQ.Stream.Client.Reliable.Producer[0]
      Producer: (null) created for Stream: TML_TEST_STREAM
dbug: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Server properties: [cluster_name, <SUBSTITUTED_HOSTNAME2>], [copyright, Copyright (c) 2007-2023 VMware, Inc. or its affiliates.], [information, Licensed under the MPL 2.0. Website: https://rabbitmq.com], [platform, Erlang/OTP 25.3], [product, RabbitMQ], [version, 3.11.10]
dbug: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Sasl mechanism: PLAIN, AMQPLAIN
dbug: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Open: ConnectionProperties: [advertised_port, 5552], [advertised_host, <SUBSTITUTED_HOSTNAME>]
dbug: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Extracted BrokerVersion version: 3.11.10
dbug: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      creating consumer c7f5f1b3-96e1-49eb-8b35-15747a5173f2 with initial credits 2, offset RabbitMQ.Stream.Client.OffsetTypeNext, is single active consumer False, super stream (null), client provided name dotnet-stream-consumer
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer reference: c7f5f1b3-96e1-49eb-8b35-15747a5173f2, stream: TML_TEST_STREAM  reconnected successfully
dbug: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer: c7f5f1b3-96e1-49eb-8b35-15747a5173f2 created for Stream: TML_TEST_STREAM
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
fail: RabbitMQ.Stream.Client.Reliable.Producer[0]
      Error reading the socket
      System.IO.IOException: Unable to read data from the transport connection: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond..
       ---> System.Net.Sockets.SocketException (10060): A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond.
         --- End of inner exception stack trace ---
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.ThrowException(SocketError error, CancellationToken cancellationToken)
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.System.Threading.Tasks.Sources.IValueTaskSource<System.Int32>.GetResult(Int16 token)
         at System.IO.Pipelines.StreamPipeReader.<ReadAsync>g__Core|36_0(StreamPipeReader reader, CancellationTokenSource tokenSource, CancellationToken cancellationToken)
         at RabbitMQ.Stream.Client.Connection.ProcessIncomingFrames() in /_/RabbitMQ.Stream.Client/Connection.cs:line 132
info: RabbitMQ.Stream.Client.Reliable.Producer[0]
      Producer reference: , stream: TML_TEST_STREAM  disconnected, check if reconnection needed in 200 ms
info: RabbitMQ.Stream.Client.Reliable.Producer[0]
      Producer reference: , stream: TML_TEST_STREAM  is disconnected. Client will try reconnect
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
!!!!!ClientTimeoutError
fail: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Error reading the socket
      System.IO.IOException: Unable to read data from the transport connection: An established connection was aborted by the software in your host machine..
       ---> System.Net.Sockets.SocketException (10053): An established connection was aborted by the software in your host machine.
         --- End of inner exception stack trace ---
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.ThrowException(SocketError error, CancellationToken cancellationToken)
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.System.Threading.Tasks.Sources.IValueTaskSource<System.Int32>.GetResult(Int16 token)
         at System.IO.Pipelines.StreamPipeReader.<ReadAsync>g__Core|36_0(StreamPipeReader reader, CancellationTokenSource tokenSource, CancellationToken cancellationToken)
         at RabbitMQ.Stream.Client.Connection.ProcessIncomingFrames() in /_/RabbitMQ.Stream.Client/Connection.cs:line 132
dbug: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      The ProcessChunks task has been closed due to cancellation
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer reference: c7f5f1b3-96e1-49eb-8b35-15747a5173f2, stream: TML_TEST_STREAM  disconnected, check if reconnection needed in 200 ms
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer reference: c7f5f1b3-96e1-49eb-8b35-15747a5173f2, stream: TML_TEST_STREAM  is disconnected. Client will try reconnect
fail: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer reference: c7f5f1b3-96e1-49eb-8b35-15747a5173f2, stream: TML_TEST_STREAM  trying to reconnect due to exception
      System.Net.Sockets.SocketException (11001): No such host is known.
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.ThrowException(SocketError error, CancellationToken cancellationToken)
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.System.Threading.Tasks.Sources.IValueTaskSource.GetResult(Int16 token)
         at System.Threading.Tasks.ValueTask.ValueTaskSourceAsTask.<>c.<.cctor>b__4_0(Object state)
      --- End of stack trace from previous location ---
         at RabbitMQ.Stream.Client.Connection.Create(EndPoint endpoint, Func'2 commandCallback, Func'2 closedCallBack, SslOption sslOption, ILogger logger) in /_/RabbitMQ.Stream.Client/Connection.cs:line 67
         at RabbitMQ.Stream.Client.Client.Create(ClientParameters parameters, ILogger logger) in /_/RabbitMQ.Stream.Client/Client.cs:line 209
         at RabbitMQ.Stream.Client.StreamSystem.MayBeReconnectLocator() in /_/RabbitMQ.Stream.Client/StreamSystem.cs:line 115
         at RabbitMQ.Stream.Client.StreamSystem.CreateRawConsumer(RawConsumerConfig rawConsumerConfig, ILogger logger) in /_/RabbitMQ.Stream.Client/StreamSystem.cs:line 353
         at RabbitMQ.Stream.Client.Reliable.ConsumerFactory.StandardConsumer(Boolean boot) in /_/RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs:line 45
         at RabbitMQ.Stream.Client.Reliable.ConsumerFactory.CreateConsumer(Boolean boot) in /_/RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs:line 32
         at RabbitMQ.Stream.Client.Reliable.Consumer.CreateNewEntity(Boolean boot) in /_/RabbitMQ.Stream.Client/Reliable/Consumer.cs:line 183
         at RabbitMQ.Stream.Client.Reliable.ReliableBase.Init(Boolean boot, IReconnectStrategy reconnectStrategy) in /_/RabbitMQ.Stream.Client/Reliable/ReliableBase.cs:line 65
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer reference: c7f5f1b3-96e1-49eb-8b35-15747a5173f2, stream: TML_TEST_STREAM  disconnected, check if reconnection needed in 400 ms
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer reference: c7f5f1b3-96e1-49eb-8b35-15747a5173f2, stream: TML_TEST_STREAM  is disconnected. Client will try reconnect
fail: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer reference: c7f5f1b3-96e1-49eb-8b35-15747a5173f2, stream: TML_TEST_STREAM  trying to reconnect due to exception
      System.Net.Sockets.SocketException (11001): No such host is known.
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.ThrowException(SocketError error, CancellationToken cancellationToken)
         at System.Net.Sockets.Socket.AwaitableSocketAsyncEventArgs.System.Threading.Tasks.Sources.IValueTaskSource.GetResult(Int16 token)
         at System.Threading.Tasks.ValueTask.ValueTaskSourceAsTask.<>c.<.cctor>b__4_0(Object state)
      --- End of stack trace from previous location ---
         at RabbitMQ.Stream.Client.Connection.Create(EndPoint endpoint, Func'2 commandCallback, Func'2 closedCallBack, SslOption sslOption, ILogger logger) in /_/RabbitMQ.Stream.Client/Connection.cs:line 67
         at RabbitMQ.Stream.Client.Client.Create(ClientParameters parameters, ILogger logger) in /_/RabbitMQ.Stream.Client/Client.cs:line 209
         at RabbitMQ.Stream.Client.StreamSystem.MayBeReconnectLocator() in /_/RabbitMQ.Stream.Client/StreamSystem.cs:line 115
         at RabbitMQ.Stream.Client.StreamSystem.CreateRawConsumer(RawConsumerConfig rawConsumerConfig, ILogger logger) in /_/RabbitMQ.Stream.Client/StreamSystem.cs:line 353
         at RabbitMQ.Stream.Client.Reliable.ConsumerFactory.StandardConsumer(Boolean boot) in /_/RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs:line 45
         at RabbitMQ.Stream.Client.Reliable.ConsumerFactory.CreateConsumer(Boolean boot) in /_/RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs:line 32
         at RabbitMQ.Stream.Client.Reliable.Consumer.CreateNewEntity(Boolean boot) in /_/RabbitMQ.Stream.Client/Reliable/Consumer.cs:line 183
         at RabbitMQ.Stream.Client.Reliable.ReliableBase.Init(Boolean boot, IReconnectStrategy reconnectStrategy) in /_/RabbitMQ.Stream.Client/Reliable/ReliableBase.cs:line 65
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer reference: c7f5f1b3-96e1-49eb-8b35-15747a5173f2, stream: TML_TEST_STREAM  disconnected, check if reconnection needed in 800 ms
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
      Consumer reference: c7f5f1b3-96e1-49eb-8b35-15747a5173f2, stream: TML_TEST_STREAM  is disconnected. Client will try reconnect

The lines prefixed with !!!! are written in the ConfirmationHandler of the producer when it fails.

I haven't been able to find a terminating condition in the source code with an upper limit for how long it should stay in reconnecting "mode". When trying to implement the custom IReconnectStrategy, returning false in WhenDisconnected doesn't raise any exceptions or change any state that is readable from the outside of my consumer. IsOpen() always returns true, it seems.

The way I am testing is quite simple. I connect a producer and consumer in the same app with more or less default settings. (This log is with the default retry handling). Then I disconnect from WiFi on my laptop. What is happening is that the application will continue running, but failing silently. I would like it to throw an exception after a timeout or some other way to handle this scenario. Specifically, we use Streams in our containers, and if we would be able to add a heartbeat on the connection, our orchestrator would restart the container. But in this case, we had a service that ran over the weekend with a passing health check. In other words, I am looking for a property to query for the connetion status.

In terms of the ConfirmationHandler, I did indeed find that for the publisher. I have yet to investigate whether we need to add custom logic around this. My initial survey seems to indicate that the publisher will eventually fail, but we cannot determine whether or not a message was actually sent other than doing confirmation collection in this handler. Out of the box, the service sends message into oblivion for a short period until detected in the confirmationhandler. This might be a problem for Queue->Stream services where you'd want to ACK messages on the queue synchroniously. However, this is solvable currently. But would require a custom implementation for every user as far as I can see.

My main problem right now is with the consumer. Ideally, we would have a healthcheck on when we last received a message, but we have cases where messages are sporadic and it's hard to set a good timeout interval. Here, we would like to restart the container if the underlying stream connection had been in "retrying" for more than, say, 5 minutes.

[Gsantomaggio]

on question:

• does the consumer/producer eventually reconnect to the server? when the connection is restored?

[TroelsL]

Maybe we can add a callback with the status Producer._inReconnection

I am personally open to many different ways of getting the information into the hands of the user. Technically, I could do it with reflection, but that is iffy.

If both Producer and Consumer exposed a ConnectionState enum e.g. { Initialized, Connected, Reconnecting, TimedOut, Closed } that would solve it for me.

[Gsantomaggio]

Thank you for all the info.
We need to provide feedback on what's happening during the reconnect stage.
We will try to work on this next week.

[Gsantomaggio]

Maybe we can add a callback with the status Producer._inReconnection

I am personally open to many different ways of getting the information into the hands of the user. Technically, I could do it with reflection, but that is iffy.

If both Producer and Consumer exposed a ConnectionState enum e.g. { Initialized, Connected, Reconnecting, TimedOut, Closed } that would solve it for me.

So basically, you are looking for a way to understand the producer/consumer status.
It makes sense

[Gsantomaggio]

Per conversation with @Zerpet and @acogoluegnes.

The idea would be to implement a sort of event bus where the stream system receives general events.
The event bus can receive different events, and the user can filter on what is interesting.

cc @TroelsL

[TroelsL]

I am not sure what the event bus suggestion entails. We currently have two problems:

1. We observed a problem reconnecting to the stream for a consumer after the server was upgraded to a newer version. It was stuck in a retry loop, and we would have liked either to be able to query the state of the connection, or to set a reconnection timeout (when it should stop retrying an assume that the connection is permanently dead). If the state changes are publised to an internal event bus, we would likely create a wrapper class that exposed the most recent state.
2. For our producer, we see a potential issue. Since ValueTask Producer.Send(Message message) appears to return regardless of success, we will need to create a custom confirmationhandler to store the unconfirmed messages. However, this seems solvable in user space with the current solution. A nice-to-have feature would be to have the package handle this logic for us.

The first issue, we haven't been able to find a good solution for as IsOpen always returns true, even when not connected.

[Gsantomaggio]

We observed a problem reconnecting to the stream for a consumer after the server was upgraded to a newer version

Ok we can re-test it even though we did it in the past.

Producer.Send(Message message)

The send is asynchronous and sends the messages regardless of the client status. The confirmation handles the situation of whether a message is stored or not.

[Zerpet]

We observed a problem reconnecting to the stream for a consumer after the server was upgraded to a newer version. It was stuck in a retry loop, and we would have liked either to be able to query the state of the connection,

That's the idea behind the event bus. You will receive an event that tells you the internal state of the producer. Something like "Connected", "Reconnecting" or "Closed"; we haven't settled on that.

You can also implement your own ReconnectStrategy in the producer config, to give up after a number of attempts. Our default strategy does not give up, because we assume that the connection will be recovered eventually.

When trying to implement the custom IReconnectStrategy, returning false in WhenDisconnected doesn't raise any exceptions or change any state that is readable from the outside of my consumer. IsOpen() always returns true, it seems.

That's a bit surprising, because TryReconnect closes the producer, which sets _isOpen to false, and IsOpen() returns the value of _isOpen.

rabbitmq-stream-dotnet-client/RabbitMQ.Stream.Client/Reliable/ReliableBase.cs

Lines 105 to 126 in 301afb5

	protected async Task TryToReconnect(IReconnectStrategy reconnectStrategy)
	{
	_inReconnection = true;
	try
	{
	switch (await reconnectStrategy.WhenDisconnected(ToString()).ConfigureAwait(false) && _isOpen)
	{
	case true:
	BaseLogger.LogInformation("{Identity} is disconnected. Client will try reconnect", ToString());
	await Init(false, reconnectStrategy).ConfigureAwait(false);
	break;
	case false:
	BaseLogger.LogInformation("{Identity} is asked to be closed", ToString());
	await Close().ConfigureAwait(false);
	break;
	}
	}
	finally
	{
	_inReconnection = false;
	}
	}

we will need to create a custom confirmationhandler to store the unconfirmed messages

The library does this for you, it's just hidden away, so that you don't have to worry about tracking the publishing IDs. There's also a confirmation handler available via the Producer config, where you receive the messages that have been confirmed. You can keep a record of sent messages, and remove them after you've received the confirmation, if you'd like to keep track of them on your app.

The first issue, we haven't been able to find a good solution for as IsOpen always returns true, even when not connected.

IsOpen does not report the state of the underlying connection, but the state of the producer. The producer is open, but it may be reconnecting, for example.

[Gsantomaggio]

General info:

We made some decisions on the Producer and Consumer side to make it as reliable as possible. We expected that some use cases could not be covered.
The idea is to improve the current implementation to avoid creating a "Frankenstein" class with many events and behaviour overrides.

There are still raw-level classes where you can implement your behaviour in case of disconnection, metadata update and messages confirmation and consumer disconnection.

The event bus idea is transparent to the current implementation.

---

The best approach in this case is:

• describe the issue
• provide a way to reproduce
• Identify the code with the possible error ( is a plus )
• Provide a PR ( would be great)

Users tend to suggest a solution for their use case. but we have to think about the big picture.

We want to analyze one problem at a time to see how to solve it.

We will start with the upgrade scenario.

Thank you

[Gsantomaggio]

We tried different fail-over scenarios like:

• upgrade cluster
• disconnect the network
• block the IP address

Producers and Consumers can reconnect to the server.

Let me add that we do these kinds of tests regularly.

about:

it was stuck in a retry loop,

Looking the logs it seems that the client can't actually lookup the host No such host is known. So it is working correctly ... apparently!

[TroelsL]

Looking the logs it seems that the client can't actually lookup the host No such host is known. So it is working correctly ... apparently!

Just to be clear - the log you received is from an MRE, not our production system. We are still investigating what would cause our service not being able to reconnect for several days. But that is not what our main problem is - our problem is not being able to detect this.

I will do another test to double check my statements regarding the IsOpen() method and the ReconnectStrategy. I will let you know if my findings.

(And just to be clear - I have no expectation to dictate a solution here. I am only proposing examples of what would solve the issue we are seeing in hopes of providing a bit more detail)

[Gsantomaggio]

OK :)
Let me try to implement a POC for the service bus, so I will show you how it would work!

[TroelsL]

Sounds great.

I have just made a simpler MRE, and can indeed confirm that:

• With the default BackOffReconnectStrategy, IsOpen() will permanently return true.
• With a custom IReconnectStrategy where WhenDisconnected always returns false, IsOpen() also returns false for both Producer and Consumer. I apologize any confusion added.

In our case, we can stitch something together to make it throw an exception when it finally gives up on reconnecting. (Either in the ReconnectionStrategy or by polling IsOpen() in combination with a custom ReconnectionStrategy. I'm still all for being able to get information out about the state changes - that would make this easier. So look forward to what you are cooking up.

For now, we have temporarily exposed _inReconnection similar to IsOpen():

public bool IsReconnecting() => _inReconnection;

[Gsantomaggio]

Thank you @TroelsL for all the details and tests you are doing.

Please have a bit of patience we are a bit overloaded at the moment.

[TroelsL]

Thank you @Gsantomaggio for being very helpful about this. I have no expectations regarding a timeframe on this (or even that you would take action). My initial assumption was that there was something I misunderstood.

[Gsantomaggio]

@TroelsL please keep under control che 1.8 version.

We added the ReliableEntityStatus https://github.com/rabbitmq/rabbitmq-stream-dotnet-client/blob/main/RabbitMQ.Stream.Client/Reliable/ReliableBase.cs#L35 to handle the status

In the next PRs, we add an event when the status changes. So you will be aware of what is happening.

[Gsantomaggio]

Will close the issue.