Failure to timeout messages when underlying socket fails

[ricsiLT]

Describe the bug

Encountered this during publishing of messages that coincided with network gear change.
I assume messages were published mid-change as I see a socket exception in the logs.

I expected to still receive all messages, albeit with a timeout status.
Instead, I got a hanging process :(

Relevant logs:

Sep 13, 2023 @ 04:26:58.779 "Producer reference: YC4974-46e1e9ba-1b8d-4b94-82d0-d6940525a7cf, stream: prod " trying to reconnect due to exception
exceptions.ClassName: System.TimeoutException
exceptions.Message: The operation has timed out.
exceptions.StackTraceString: at RabbitMQ.Stream.Client.ManualResetValueTaskSource`1.System.Threading.Tasks.Sources.IValueTaskSource<T>.GetResult(Int16 token) in /_/RabbitMQ.Stream.Client/Client.cs:line 734
   at RabbitMQ.Stream.Client.Client.Request[TIn,TOut](Func`2 request, Nullable`1 timeout) in /_/RabbitMQ.Stream.Client/Client.cs:line 385
   at RabbitMQ.Stream.Client.Client.Request[TIn,TOut](Func`2 request, Nullable`1 timeout) in /_/RabbitMQ.Stream.Client/Client.cs:line 385
   at RabbitMQ.Stream.Client.Client.QueryMetadata(String[] streams) in /_/RabbitMQ.Stream.Client/Client.cs:line 654
   at RabbitMQ.Stream.Client.StreamSystem.CreateRawProducer(RawProducerConfig rawProducerConfig, ILogger logger) in /_/RabbitMQ.Stream.Client/StreamSystem.cs:line 269
   at RabbitMQ.Stream.Client.Reliable.ProducerFactory.StandardProducer() in /_/RabbitMQ.Stream.Client/Reliable/ProducerFactory.cs:line 101
   at RabbitMQ.Stream.Client.Reliable.ProducerFactory.CreateProducer() in /_/RabbitMQ.Stream.Client/Reliable/ProducerFactory.cs:line 30
   at RabbitMQ.Stream.Client.Reliable.Producer.CreateNewEntity(Boolean boot) in /_/RabbitMQ.Stream.Client/Reliable/Producer.cs:line 183
   at RabbitMQ.Stream.Client.Reliable.ReliableBase.Init(Boolean boot, IReconnectStrategy reconnectStrategy) in /_/RabbitMQ.Stream.Client/Reliable/ReliableBase.cs:line 89

Sep 13, 2023 @ 04:29:12.051 Error reading the socket
exceptions.ClassName: System.InvalidOperationException
exceptions.Message: Operation is not valid due to the current state of the object.
exceptions.StackTraceString: at 
System.Threading.Tasks.Sources.ManualResetValueTaskSourceCore`1.SignalCompletion()
   at RabbitMQ.Stream.Client.Client.HandleCorrelatedResponse[T](T 
command) in /_/RabbitMQ.Stream.Client/Client.cs:line 575
   at RabbitMQ.Stream.Client.Client.HandleCorrelatedCommand(UInt16 tag, 
ReadOnlySequence`1& frame) in 
/_/RabbitMQ.Stream.Client/Client.cs:line 562
   at RabbitMQ.Stream.Client.Client.HandleIncoming(Memory`1 frameMemory)
 in /_/RabbitMQ.Stream.Client/Client.cs:line 478
   at RabbitMQ.Stream.Client.Connection.ProcessIncomingFrames() in 
/_/RabbitMQ.Stream.Client/Connection.cs:line 190

Sep 13, 2023 @ 04:29:12.939 Error reading the socket
exceptions.ClassName: System.ArgumentNullException
exceptions.Message: Value cannot be null. (Parameter 'target')
exceptions.StackTraceString: at 
System.Threading.Tasks.Dataflow.DataflowBlock.SendAsync[TInput](ITargetBlock`1
 target, TInput item, CancellationToken cancellationToken)
   at 
RabbitMQ.Stream.Client.Reliable.ProducerFactory.<StandardProducer>b__4_2(Confirmation
 confirmation) in 
/_/RabbitMQ.Stream.Client/Reliable/ProducerFactory.cs:line 99
   at 
RabbitMQ.Stream.Client.RawProducer.<Init>b__18_1(ReadOnlyMemory`1 
publishingIds) in /_/RabbitMQ.Stream.Client/RawProducer.cs:line 113
   at RabbitMQ.Stream.Client.Client.HandleIncoming(Memory`1 frameMemory)
 in /_/RabbitMQ.Stream.Client/Client.cs:line 478
   at RabbitMQ.Stream.Client.Connection.ProcessIncomingFrames() in 
/_/RabbitMQ.Stream.Client/Connection.cs:line 190

Sep 13, 2023 @ 04:29:12.961  error while Process Buffer
exceptions.ClassName: System.ObjectDisposedException
exceptions.Message: Cannot access a disposed object. Object name: 'System.Net.Security.SslStream'.
exceptions.StackTraceString: at System.Net.Security.SslStream.<ThrowIfExceptional>g__ThrowExceptional|123_0(ExceptionDispatchInfo e)
   at System.Net.Security.SslStream.WriteAsync(ReadOnlyMemory`1 buffer, CancellationToken cancellationToken)
   at System.IO.Pipelines.StreamPipeWriter.FlushAsyncInternal(Boolean writeToStream, ReadOnlyMemory`1 data, CancellationToken cancellationToken)
   at System.Runtime.CompilerServices.PoolingAsyncValueTaskMethodBuilder`1.StateMachineBox`1.System.Threading.Tasks.Sources.IValueTaskSource<TResult>.GetResult(Int16 token)
   at RabbitMQ.Stream.Client.Connection.WriteCommand[T](T command) in /_/RabbitMQ.Stream.Client/Connection.cs:line 121
   at RabbitMQ.Stream.Client.Connection.Write[T](T command) in /_/RabbitMQ.Stream.Client/Connection.cs:line 97
   at RabbitMQ.Stream.Client.Client.Publish(Publish publishMsg) in /_/RabbitMQ.Stream.Client/Client.cs:line 267
   at RabbitMQ.Stream.Client.RawProducer.SendMessages(List`1 messages, Boolean clearMessagesList) in /_/RabbitMQ.Stream.Client/RawProducer.cs:line 221
   at RabbitMQ.Stream.Client.RawProducer.ProcessBuffer() in /_/RabbitMQ.Stream.Client/RawProducer.cs:line 302

Reproduction steps

1. Yoink cable during publishing?
2. Observe?

No clue how you'd repro this :/

Expected behavior

Messages that were queued for publishing should just be returned to me with a timeout/other status, indicating failure.

Additional context

@Gsantomaggio sorry for not responding on Slack. Corporate recently banned all external comms and I have yet to copy my user/psw to home workstation to reply :(

[ricsiLT]

Would this be mitigated by having RawProducer instead of ReliableProducer? Would I be able to hook into disconnect?

[Gsantomaggio]

You can use the RawProducer but you will lose all the features like "confirmation timeout" the messages sent/received trace and metadata update.

On the ProducerFactory there is the ConnectionClosedHandler what we could is to provide and interface like:

IConnectionClosedHandler

The producer will use the current implementation by default; otherwise, it will delegate to the user what to do.

[Gsantomaggio]

something like this #294

[ricsiLT]

Yeah, looks like exactly what we need. Would also play well with IReconnectionStrategy because presumably some strategies would involve giving up.