Increment Request Before Serializing it in OutboundHandler (#74256) (#74270)

If there are outside ways by which a request can be decremented (e.g.
due to cancelling a recovery concurrently) we may run into a situation where
we try to send a `refcount == 0` request. We have to avoid this by incrementing
a request before serializing and decrementing after it returns to make sure we
don't corrupt the request bytes while they're being serialized
or sent over the wire.

Closes #74253

Author: original-brownbear

server/src/main/java/org/elasticsearch/indices/recovery/RemoteRecoveryTargetHandler.java

@@ -10,6 +10,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.store.RateLimiter;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
@@ -243,8 +244,19 @@ private <T extends TransportResponse> void executeRetryableAction(String action,
 
             @Override
             public void tryAction(ActionListener<T> listener) {
-                transportService.sendRequest(targetNode, action, request, options,
-                    new ActionListenerResponseHandler<>(listener, reader, ThreadPool.Names.GENERIC));
+                if (request.tryIncRef()) {
+                    transportService.sendRequest(
+                        targetNode,
+                        action,
+                        request,
+                        options,
+                        new ActionListenerResponseHandler<>(
+                            ActionListener.runBefore(listener, request::decRef),
+                            reader,
+                            ThreadPool.Names.GENERIC));
+                } else {
+                    listener.onFailure(new AlreadyClosedException("already closed"));
+                }
             }
 
             @Override

server/src/main/java/org/elasticsearch/transport/BytesTransportRequest.java

@@ -67,7 +67,7 @@ public void incRef() {
 
     @Override
     public boolean tryIncRef() {
-        return bytes.decRef();
+        return bytes.tryIncRef();
     }
 
     @Override

server/src/main/java/org/elasticsearch/transport/OutboundHandler.java

@@ -11,6 +11,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.NotifyOnceListener;
@@ -81,8 +82,17 @@ void sendRequest(final DiscoveryNode node, final TcpChannel channel, final long
         Version version = Version.min(this.version, channelVersion);
         OutboundMessage.Request message = new OutboundMessage.Request(threadPool.getThreadContext(), features, request, version, action,
             requestId, isHandshake, compressRequest);
-        ActionListener<Void> listener = ActionListener.wrap(() ->
-            messageListener.onRequestSent(node, requestId, action, request, options));
+        if (request.tryIncRef() == false) {
+            assert false : "request [" + request + "] has been released already";
+            throw new AlreadyClosedException("request [" + request + "] has been released already");
+        }
+        ActionListener<Void> listener = ActionListener.wrap(() -> {
+            try {
+                messageListener.onRequestSent(node, requestId, action, request, options);
+            } finally {
+                request.decRef();
+            }
+        });
         sendMessage(channel, message, listener);
     }