Mitigation

.github/workflows/ci-prb.yml

@@ -35,7 +35,7 @@ jobs:
         if: runner.os == 'Linux'
         env:
           JAVA_TOOL_OPTIONS: -Dfile.encoding=UTF-8
-        run:  sudo -E env "PATH=$PATH" bash -c "ulimit -l 65536 && ulimit -a && ./gradlew --no-daemon :servicetalk-http-netty:test --tests io.servicetalk.http.netty.SslAndNonSslConnectionsTest"
+        run:  sudo -E env "PATH=$PATH" bash -c "ulimit -l 65536 && ulimit -a && ./gradlew --no-daemon --parallel clean test"
       - name: Build and Test (non-Linux)
         if: runner.os != 'Linux'
         env:

servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/DefaultNettyConnection.java

@@ -890,11 +890,6 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
             // NettyChannelPublisher will force closure of the channel in case of exception after the cause is
             // propagated to users. In case users don't have offloading, there is a risk to retry on the same IO thread.
             // We should notify LoadBalancer that this connection is closing to avoid retrying on the same connection.
-            if (cause instanceof Errors.NativeIoException) {
-                // TODO: doesn't appear to be the pathway.
-                cause = new Errors.NativeIoException("Boomed here in DefaultNettyConnection",
-                        ((Errors.NativeIoException) cause).expectedErr(), true);
-            }
             connection.notifyOnClosing();
             connection.nettyChannelPublisher.channelOnError(unwrapThrowable(cause));
         }
@@ -907,6 +902,17 @@ private static Throwable unwrapThrowable(final Throwable t) {
             if (t instanceof DecoderException && (cause = t.getCause()) instanceof SSLException) {
                 return cause;
             }
+            if (t instanceof Errors.NativeIoException) {
+                if (t.getMessage().contains("Connection reset by peer")) {
+                    // TODO: this is messy and probably not the right place to intercept this.
+                    // This is really a channel closed exception that happens when netty tries to read from a
+                    // channel that is closed.
+                    Throwable channelClosedException = StacklessClosedChannelException.newInstance(
+                            DefaultNettyConnection.class, "exceptionCaught(ChannelInputShutdownReadComplete)");
+                    channelClosedException.initCause(t);
+                    return channelClosedException;
+                }
+            }
             return t;
         }
 
@@ -944,7 +950,6 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                 // ChannelInputShutdownEvent is not always triggered and can get triggered before we tried to read
                 // all the available data. ChannelInputShutdownReadComplete is the one that seems to (at least in
                 // the current netty version) gets triggered reliably at the appropriate time.
-                // TODO: This is the standard pathway the connection is shut down.
                 connection.nettyChannelPublisher.channelOnError(StacklessClosedChannelException.newInstance(
                         DefaultNettyConnection.class, "userEventTriggered(ChannelInputShutdownReadComplete)"));
             } else if (evt instanceof SslHandshakeCompletionEvent) {

servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/WriteStreamSubscriber.java

@@ -27,7 +27,6 @@
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.DefaultChannelPromise;
-import io.netty.channel.unix.Errors;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
@@ -441,10 +440,6 @@ void sourceTerminated(@Nullable Throwable cause, boolean markCancelled) {
                 // We have terminated prematurely perhaps due to write failure.
                 return;
             }
-            if (cause instanceof Errors.NativeIoException) {
-                cause = new Errors.NativeIoException("sourceTerminated",
-                        ((Errors.NativeIoException) cause).expectedErr(), true);
-            }
             this.failureCause = cause;
             state = set(state, SOURCE_TERMINATED);
             if (markCancelled) {
@@ -593,11 +588,6 @@ private void terminateSubscriber(@Nullable Throwable cause) {
                     closeHandler.closeChannelOutbound(channel);
                 }
             } else {
-                if (cause instanceof Errors.NativeIoException) {
-                    // TODO: doesn't appear to be the pathway.
-                    cause = new Errors.NativeIoException("It blew up in WriteStreamSubscriber.",
-                            ((Errors.NativeIoException) cause).expectedErr(), true);
-                }
                 Throwable enrichedCause = enrichProtocolError.apply(cause);
                 assignConnectionError(channel, enrichedCause);
                 enrichedCause = !written ? new AbortedFirstWriteException(enrichedCause) : enrichedCause;