Must catch IOException from channel.read and channel.write to avoid repeated exception reporting from the reaktor

[cmebarrow]

I accidentally hit this while working on a test. The console output was:

java.io.IOException: An established connection was aborted by the software in your host machine
    at sun.nio.ch.SocketDispatcher.read0(Native Method)
    at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:43)
    at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
    at sun.nio.ch.IOUtil.read(IOUtil.java:197)
    at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
    at org.reaktivity.nukleus.tcp.internal.reader.stream.StreamFactory$Stream.handleRead(StreamFactory.java:113)
    at org.reaktivity.nukleus.tcp.internal.reader.stream.StreamFactory$Stream.handleStream(StreamFactory.java:89)
    at org.reaktivity.nukleus.tcp.internal.reader.stream.StreamFactory$Stream.access$2(StreamFactory.java:85)
    at org.reaktivity.nukleus.tcp.internal.reader.stream.StreamFactory$$Lambda$52/2030708124.getAsInt(Unknown Source)
    at org.reaktivity.nukleus.tcp.internal.reader.Source.processKey(Source.java:105)
    at org.reaktivity.nukleus.tcp.internal.reader.Source$$Lambda$36/749126367.applyAsInt(Unknown Source)
    at org.agrona.nio.NioSelectedKeySet.forEach(NioSelectedKeySet.java:128)
    at org.reaktivity.nukleus.tcp.internal.reader.Source.process(Source.java:65)
    at org.reaktivity.nukleus.Nukleus$Composite.process(Nukleus.java:52)
    at org.reaktivity.nukleus.Nukleus$Composite.process(Nukleus.java:52)
    at org.reaktivity.nukleus.Nukleus$Composite.process(Nukleus.java:52)
    at org.reaktivity.reaktor.test.NukleusRule$1.lambda$0(NukleusRule.java:166)
    at org.reaktivity.reaktor.test.NukleusRule$1$$Lambda$12/1627960023.run(Unknown Source)
    at java.lang.Thread.run(Thread.java:745)
java.io.IOException: An established connection was aborted by the software in your host machine
    at sun.nio.ch.SocketDispatcher.read0(Native Method)
    at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:43)
    at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
    at sun.nio.ch.IOUtil.read(IOUtil.java:197)
etc (repeated 510 times)

This can be reproduced by adding the following test case to ServerIT and running it several times (the problem does not always occur, it obviously depends on timing):

    @Test
    @Specification({
        "${route}/input/new/controller",
        "${streams}/server.sent.data/server/target"
    })
    public void shouldNotGetRepeatedIOExceptionsFromReaderStreamRead() throws Exception
    {
        k3po.start();
        k3po.awaitBarrier("ROUTED_INPUT");

        try(Socket socket = new Socket("127.0.0.1", 0x1f90))
        {
            socket.shutdownInput();
            Thread.sleep(500);
        }

        k3po.finish();
    }

[cmebarrow]

We should not report this condition like an unexpected error (as we do other exceptions caught in the reaktor) because it is a frequent condition that clients abruptly close a connection (e.g. mobile phones). We should report it in a separate counter. The same applies for subsequent attempts to write to the channel on which the IOException (RST) occurred.

[cmebarrow]

Looking at the jdk source code and the Microsoft Windows Socket API documentation there is no reliable way to distinguish an IOException from SocketChannel.read due to client sending a RST from any other one. Proposed solution from discussion with John is:

• [ ] Catch the IOException from channel.read and do the following
  • close the channel (CloseHelper.quietClose(channel);)
  • report end of stream (as we do when read returns -1.
• [ ] Do not rethrow IOException from write, just report a reset
• [x] Add spec tests as follows (these should use ByteMan to reliably cause read to throw an IOException since the behavior on Windows may not be typical and it does not really seem correct that socket.shut downInput would cause a RST to be sent when more data comes in)):
  1. IOException from read (RST from remote) should result in EOS on input (can use spec tests client.close and server.close)
  2. IOException from read should cause subsequent attempt to write to result in RESET (can use spec tests xxx.sent.data.received.reset)
  3. END write after IOException from read should not result in error (could use server.close/server/target.rpt? may need a variant without the close/closed on input stream)

Testing shows any subsequent attempt to write to the socket channel will result in an exception (which will be reported as a RESET frame to the writer). It doesn't appear there is any value in immediately reporting a RESET to the writer when we catch the exception from the read.

[cmebarrow]

Orderly Versus Abortive Connection Release in Java is a very useful link.

[jfallows]

Fix merged, closing.