Repository: netty/netty (Java)

Issue:
Http2MultiplexCodec.DefaultHttp2StreamChannel.Http2ChannelUnsafe has a different order of close events than other channel implementation
The `HttpObjectAggregator`, when it encounters a too-large-message, [does the following](https://github.com/netty/netty/blob/4.1/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectAggregator.java#L281) when handling a too large response:
```java
ctx.close();
throw new TooLongFrameException("Response entity too large: " + oversized);
```
The tail of the pipeline gets an `exceptionCaught` event before receiving a `channelInactive` event because channel deregistration is always done '[later](https://github.com/netty/netty/blob/4.1/transport/src/main/java/io/netty/channel/AbstractChannel.java#L813)'. This is not the case in the `Http2ChannelUnsafe` which triggers shutdown events [immediately](https://github.com/netty/netty/blob/4.1/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2MultiplexCodec.java#L925). That means when using the `HttpObjectAggregator` in a stream channel, we will receive  the`channelInactive` event before the `exceptionCaught` event, making it difficult to handle gracefully.

### Netty version

4.1.26-SNAPSHOT

### JVM version (e.g. `java -version`)
openjdk version "1.8.0_162"

### OS version (e.g. `uname -a`)
Darwin


Hints:
Presuming I haven't messed this up, I'd love to get this fixed before 4.1.26 since it would make it really difficult to use symmetric pipelines for our HTTP/2 and HTTP/1.x clients.
This sounds like a bug... interested in working on a PR?
Absolutely, just wanted confirmation that I wasn't just using it wrong. 😄 
Just to be clear, I think the deferred close operations is what we want, correct? I could see it going the other way: technically the `ctx.close` was called first, but the operation is asynchronous so it seem like legal behavior to defer the events.
Yes just as for normal channels 

Patch:
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2MultiplexCodec.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2MultiplexCodec.java
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2MultiplexCodec.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2MultiplexCodec.java
@@ -43,11 +43,14 @@
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.ThrowableUtil;
 import io.netty.util.internal.UnstableApi;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
 
 import java.net.SocketAddress;
 import java.nio.channels.ClosedChannelException;
 import java.util.ArrayDeque;
 import java.util.Queue;
+import java.util.concurrent.RejectedExecutionException;
 
 import static io.netty.handler.codec.http2.Http2CodecUtil.HTTP_UPGRADE_STREAM_ID;
 import static io.netty.handler.codec.http2.Http2CodecUtil.isStreamIdValid;
@@ -104,6 +107,8 @@
 @UnstableApi
 public class Http2MultiplexCodec extends Http2FrameCodec {
 
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultHttp2StreamChannel.class);
+
     private static final ChannelFutureListener CHILD_CHANNEL_REGISTRATION_LISTENER = new ChannelFutureListener() {
         @Override
         public void operationComplete(ChannelFuture future) throws Exception {
@@ -899,6 +904,8 @@ public void operationComplete(ChannelFuture future) throws Exception {
                 closePending = false;
                 fireChannelReadPending = false;
 
+                final boolean wasActive = isActive();
+
                 // Only ever send a reset frame if the connection is still alive as otherwise it makes no sense at
                 // all anyway.
                 if (parent().isActive() && !streamClosedWithoutError && isStreamIdValid(stream().id())) {
@@ -922,10 +929,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
                 closePromise.setSuccess();
                 promise.setSuccess();
 
-                pipeline().fireChannelInactive();
-                if (isRegistered()) {
-                    deregister(unsafe().voidPromise());
-                }
+                fireChannelInactiveAndDeregister(voidPromise(), wasActive);
             }
 
             @Override
@@ -935,15 +939,70 @@ public void closeForcibly() {
 
             @Override
             public void deregister(ChannelPromise promise) {
+                fireChannelInactiveAndDeregister(promise, false);
+            }
+
+            private void fireChannelInactiveAndDeregister(final ChannelPromise promise,
+                                                          final boolean fireChannelInactive) {
                 if (!promise.setUncancellable()) {
                     return;
                 }
-                if (registered) {
-                    registered = true;
+
+                if (!registered) {
                     promise.setSuccess();
-                    pipeline().fireChannelUnregistered();
-                } else {
-                    promise.setFailure(new IllegalStateException("Not registered"));
+                    return;
+                }
+
+                // As a user may call deregister() from within any method while doing processing in the ChannelPipeline,
+                // we need to ensure we do the actual deregister operation later. This is necessary to preserve the
+                // behavior of the AbstractChannel, which always invokes channelUnregistered and channelInactive
+                // events 'later' to ensure the current events in the handler are completed before these events.
+                //
+                // See:
+                // https://github.com/netty/netty/issues/4435
+                invokeLater(new Runnable() {
+                    @Override
+                    public void run() {
+                        if (fireChannelInactive) {
+                            pipeline.fireChannelInactive();
+                        }
+                        // Some transports like local and AIO does not allow the deregistration of
+                        // an open channel.  Their doDeregister() calls close(). Consequently,
+                        // close() calls deregister() again - no need to fire channelUnregistered, so check
+                        // if it was registered.
+                        if (registered) {
+                            registered = false;
+                            pipeline.fireChannelUnregistered();
+                        } else {
+                            promise.setFailure(new IllegalStateException("Not registered"));
+                        }
+                        safeSetSuccess(promise);
+                    }
+                });
+            }
+
+            private void safeSetSuccess(ChannelPromise promise) {
+                if (!(promise instanceof VoidChannelPromise) && !promise.trySuccess()) {
+                    logger.warn("Failed to mark a promise as success because it is done already: {}", promise);
+                }
+            }
+
+            private void invokeLater(Runnable task) {
+                try {
+                    // This method is used by outbound operation implementations to trigger an inbound event later.
+                    // They do not trigger an inbound event immediately because an outbound operation might have been
+                    // triggered by another inbound event handler method.  If fired immediately, the call stack
+                    // will look like this for example:
+                    //
+                    //   handlerA.inboundBufferUpdated() - (1) an inbound handler method closes a connection.
+                    //   -> handlerA.ctx.close()
+                    //     -> channel.unsafe.close()
+                    //       -> handlerA.channelInactive() - (2) another inbound handler method called while in (1) yet
+                    //
+                    // which means the execution of two inbound handler methods of the same handler overlap undesirably.
+                    eventLoop().execute(task);
+                } catch (RejectedExecutionException e) {
+                    logger.warn("Can't invoke task later as EventLoop rejected it", e);
                 }
             }
 


Test patch:
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexCodecTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexCodecTest.java
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexCodecTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexCodecTest.java
@@ -34,6 +34,7 @@
 import java.net.InetSocketAddress;
 import java.nio.channels.ClosedChannelException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.After;
 import org.junit.Before;
@@ -583,6 +584,54 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         assertFalse(channelActive.get());
     }
 
+    @Test
+    public void channelInactiveHappensAfterExceptionCaughtEvents() throws Exception {
+        final AtomicInteger count = new AtomicInteger(0);
+        final AtomicInteger exceptionCaught = new AtomicInteger(-1);
+        final AtomicInteger channelInactive = new AtomicInteger(-1);
+        final AtomicInteger channelUnregistered = new AtomicInteger(-1);
+        Http2StreamChannel childChannel = newOutboundStream();
+
+        childChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
+
+            @Override
+            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+                ctx.close();
+                throw new Exception("exception");
+            }
+        });
+
+        childChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
+
+            @Override
+            public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+                channelInactive.set(count.getAndIncrement());
+                super.channelInactive(ctx);
+            }
+
+            @Override
+            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+                exceptionCaught.set(count.getAndIncrement());
+                super.exceptionCaught(ctx, cause);
+            }
+
+            @Override
+            public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
+                channelUnregistered.set(count.getAndIncrement());
+                super.channelUnregistered(ctx);
+            }
+        });
+
+        childChannel.pipeline().fireUserEventTriggered(new Object());
+        parentChannel.runPendingTasks();
+
+        // The events should have happened in this order because the inactive and deregistration events
+        // get deferred as they do in the AbstractChannel.
+        assertEquals(0, exceptionCaught.get());
+        assertEquals(1, channelInactive.get());
+        assertEquals(2, channelUnregistered.get());
+    }
+
     @Ignore("not supported anymore atm")
     @Test
     public void cancellingWritesBeforeFlush() {