Fix race in HTTP response shutdown handling (#105306) (#105327)

Similar to #97301, the fix in #105293 was still not quite correct: we
could in principle shut down the transport after checking `isOpen()` but
before sending the message. Applying the same fix as for the transport
layer here.

Author: DaveCTurner

docs/changelog/105306.yaml

@@ -0,0 +1,5 @@
+pr: 105306
+summary: Fix race in HTTP response shutdown handling
+area: Network
+type: bug
+issues: []

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpChannel.java

@@ -14,11 +14,11 @@
 import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.http.HttpChannel;
 import org.elasticsearch.http.HttpResponse;
+import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.netty4.Netty4TcpChannel;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
-import java.nio.channels.ClosedChannelException;
 
 public class Netty4HttpChannel implements HttpChannel {
 
@@ -32,12 +32,12 @@ public class Netty4HttpChannel implements HttpChannel {
 
     @Override
     public void sendResponse(HttpResponse response, ActionListener<Void> listener) {
-        if (isOpen()) {
-            channel.writeAndFlush(response, Netty4TcpChannel.addPromise(listener, channel));
-        } else {
-            // No need to dispatch to the event loop just to fail this listener; moreover the channel might be closed because the whole
-            // node is shutting down, in which case the event loop might not exist any more so the channel promise cannot be completed.
-            listener.onFailure(new ClosedChannelException());
+        // We need to both guard against double resolving the listener and not resolving it in case of event loop shutdown so we need to
+        // use #notifyOnce here until https://github.com/netty/netty/issues/8007 is resolved.
+        var wrapped = ActionListener.notifyOnce(listener);
+        channel.writeAndFlush(response, Netty4TcpChannel.addPromise(wrapped, channel));
+        if (channel.eventLoop().isShutdown()) {
+            wrapped.onFailure(new TransportException("Cannot send HTTP response, event loop is shutting down."));
         }
     }