Make AbstractSearchAsyncAction release resources earlier

We can release resources earlier by releasing before responding to the
listener (everything that needs retaining is retained via the search
response already) as well as make the GCs life a little easier and get
obvious correctness by using the listener that nulls out resources in a
thread-safe manner intead of a non-thread-safe and mutable list shared
across all kinds of places in the code.

Author: original-brownbear

server/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java

@@ -27,7 +27,6 @@
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.core.Releasable;
-import org.elasticsearch.core.Releasables;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.SearchContextMissingException;
 import org.elasticsearch.search.SearchPhaseResult;
@@ -102,7 +101,7 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
     private final AtomicBoolean requestCancelled = new AtomicBoolean();
 
     // protected for tests
-    protected final List<Releasable> releasables = new ArrayList<>();
+    protected final SubscribableListener<Void> doneFuture = new SubscribableListener<>();
 
     AbstractSearchAsyncAction(
         String name,
@@ -151,7 +150,7 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
         this.executor = executor;
         this.request = request;
         this.task = task;
-        this.listener = ActionListener.runAfter(listener, () -> Releasables.close(releasables));
+        this.listener = ActionListener.runBefore(listener, () -> doneFuture.onResponse(null));
         this.nodeIdToConnection = nodeIdToConnection;
         this.concreteIndexBoosts = concreteIndexBoosts;
         this.clusterStateVersion = clusterState.version();
@@ -182,7 +181,12 @@ protected void notifyListShards(
      * Registers a {@link Releasable} that will be closed when the search request finishes or fails.
      */
     public void addReleasable(Releasable releasable) {
-        releasables.add(releasable);
+        var doneFuture = this.doneFuture;
+        if (doneFuture.isDone()) {
+            releasable.close();
+        } else {
+            doneFuture.addListener(ActionListener.releasing((releasable)));
+        }
     }
 
     /**

server/src/test/java/org/elasticsearch/action/search/MockSearchPhaseContext.java

@@ -18,7 +18,6 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.core.Nullable;
-import org.elasticsearch.core.Releasables;
 import org.elasticsearch.search.SearchPhaseResult;
 import org.elasticsearch.search.SearchShardTarget;
 import org.elasticsearch.search.internal.ShardSearchContextId;
@@ -104,8 +103,7 @@ public void sendSearchResponse(SearchResponseSections internalSearchResponse, At
                 searchContextId
             )
         );
-        Releasables.close(releasables);
-        releasables.clear();
+        doneFuture.onResponse(null);
         if (existing != null) {
             existing.decRef();
         }