Handle SocketException while configuring an accepted socket in TcpListener

AUTHORS.md

@@ -12,6 +12,7 @@ Contributors
 * Ilya Loginov — [@iloginov](https://github.com/iloginov)
 * James Hurst — [@JamesWhurst](https://github.com/JamesWhurst)
 * Jay Shelton — [@jayshelton](https://github.com/jayshelton)
+* Jonas Follesø — [@follesoe](https://github.com/follesoe)
 * Miguel Angel Jimenez — [@majimenezp](https://github.com/majimenezp)
 * Nicolas Dextraze — [@ndextraze-pbp](https://github.com/ndextraze-pbp)
 * Peter H. Merkel — [@mph911](https://github.com/mph911)

src/NetMQ.Tests/TcpListenerTests.cs

@@ -0,0 +1,61 @@
+using System;
+using System.Net;
+using System.Net.Sockets;
+using NetMQ.Sockets;
+using Xunit;
+
+namespace NetMQ.Tests
+{
+    public class TcpListenerTests : IClassFixture<CleanupAfterFixture>
+    {
+        public TcpListenerTests() => NetMQConfig.Cleanup();
+
+        [Fact]
+        public void AcceptedSocketFailingSetupDoesNotCrashProcessOrStopListener()
+        {
+            using (var pub = new PublisherSocket())
+            {
+                int port = pub.BindRandomPort("tcp://127.0.0.1");
+
+                // Hammer the listener with connections that are reset immediately:
+                // SO_LINGER=0 + Close sends RST, which can land between the proactor
+                // completing an accept and TcpListener applying socket options to the
+                // accepted socket. On macOS the option calls then throw
+                // SocketException (EINVAL). Unhandled, that exception killed the
+                // proactor thread and with it the process; swallowed carelessly, it
+                // skips the re-arming Accept() and the listener goes deaf instead.
+                for (int i = 0; i < 400; i++)
+                {
+                    using (var raw = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
+                    {
+                        raw.LingerState = new LingerOption(true, 0);
+                        try
+                        {
+                            raw.Connect(IPAddress.Loopback, port);
+                        }
+                        catch (SocketException)
+                        {
+                        }
+                    }
+                }
+
+                // The listener must still be alive and accepting: a subscriber that
+                // connects after the storm must complete the handshake and receive.
+                using (var sub = new SubscriberSocket())
+                {
+                    sub.Connect("tcp://127.0.0.1:" + port);
+                    sub.SubscribeToAnyTopic();
+
+                    var received = false;
+                    for (int i = 0; i < 100 && !received; i++)
+                    {
+                        pub.SendFrame("hello");
+                        received = sub.TryReceiveFrameString(TimeSpan.FromMilliseconds(100), out string? _);
+                    }
+
+                    Assert.True(received, "Listener stopped accepting connections after a flood of immediately-reset connects");
+                }
+            }
+        }
+    }
+}

src/NetMQ/Core/Transports/Tcp/TcpListener.cs

@@ -199,11 +199,14 @@ public void InCompleted(SocketError socketError, int bytesTransferred)
                     // TODO: check TcpFilters
                     var acceptedSocket = m_handle.GetAcceptedSocket();
 
+                    StreamEngine engine;
+                    try
+                    {
                         acceptedSocket.NoDelay = true;
 
-                    if (m_options.TcpKeepalive != -1)
-                    {
-                        acceptedSocket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, m_options.TcpKeepalive);
+                        if (m_options.TcpKeepalive != -1)
+                        {
+                            acceptedSocket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, m_options.TcpKeepalive);
 #if NET
                             if (m_options.TcpKeepaliveIdle != -1)
                                 acceptedSocket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, m_options.TcpKeepaliveIdle / 1000);
@@ -213,24 +216,54 @@ public void InCompleted(SocketError socketError, int bytesTransferred)
                                 acceptedSocket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount, m_options.TcpKeepaliveCnt);
 #else
 
-                        if (m_options.TcpKeepaliveIdle != -1 && m_options.TcpKeepaliveIntvl != -1)
-                        {
-                            var bytes = new ByteArraySegment(new byte[12]);
+                            if (m_options.TcpKeepaliveIdle != -1 && m_options.TcpKeepaliveIntvl != -1)
+                            {
+                                var bytes = new ByteArraySegment(new byte[12]);
 
-                            Endianness endian = BitConverter.IsLittleEndian ? Endianness.Little : Endianness.Big;
+                                Endianness endian = BitConverter.IsLittleEndian ? Endianness.Little : Endianness.Big;
 
-                            bytes.PutInteger(endian, m_options.TcpKeepalive, 0);
-                            bytes.PutInteger(endian, m_options.TcpKeepaliveIdle, 4);
-                            bytes.PutInteger(endian, m_options.TcpKeepaliveIntvl, 8);
+                                bytes.PutInteger(endian, m_options.TcpKeepalive, 0);
+                                bytes.PutInteger(endian, m_options.TcpKeepaliveIdle, 4);
+                                bytes.PutInteger(endian, m_options.TcpKeepaliveIntvl, 8);
 
 
-                            acceptedSocket.IOControl(IOControlCode.KeepAliveValues, (byte[])bytes, null);
-                        }
+                                acceptedSocket.IOControl(IOControlCode.KeepAliveValues, (byte[])bytes, null);
+                            }
 #endif
+                        }
+
+                        // Create the engine object for this connection. The engine
+                        // constructor also touches the socket (send/receive buffer
+                        // sizes), so it shares the failure window below.
+                        engine = new StreamEngine(acceptedSocket, m_options, m_endpoint);
                     }
+                    catch (SocketException ex)
+                    {
+                        // The peer can reset the connection between the accept
+                        // completing and the socket options above being applied; the
+                        // accepted socket is then already dead and the option calls
+                        // throw (EINVAL/InvalidArgument on macOS and Linux). Without
+                        // this catch the exception escapes on the proactor thread and
+                        // terminates the process. Treat it as a failed accept, as
+                        // libzmq does: drop the socket and keep listening. The raw
+                        // SocketError is deliberately not passed through ToErrorCode,
+                        // which Debug.Asserts on unmapped values; ConnectionReset is
+                        // the effective outcome whatever the exact reported code.
+                        Debug.WriteLine($"TcpListener dropping accepted socket that failed setup: {ex.SocketErrorCode}");
+
+                        try
+                        {
+                            acceptedSocket.Dispose();
+                        }
+                        catch (SocketException)
+                        {
+                        }
+
+                        m_socket.EventAcceptFailed(m_endpoint, ErrorCode.ConnectionReset);
 
-                    // Create the engine object for this connection.
-                    var engine = new StreamEngine(acceptedSocket, m_options, m_endpoint);
+                        Accept();
+                        break;
+                    }
 
                     // Choose I/O thread to run connector in. Given that we are already
                     // running in an I/O thread, there must be at least one available.