Fix assertion error on multiple node search reduce errors

[cbuescher]

When receiving exceptions from more than one node on data node search reduction, SearchQueryThenFetchAsyncAction will raise more than one phase failure. This can lead to calling the listener in AbstractSearchAsyncAction more than once, which in turn in tests trips an assertion in ActionListener#assertFirstRun.

[elasticsearchmachine]

Pinging @elastic/es-search-foundations (Team:Search Foundations)

[cbuescher]

I ran into this issue while working on #129445 where another test in this suite fails for probably other reasons ("testSortMixedFieldTypes"). The test provokes IllegalArgumentExceptions when sorting on two indices with incompatible sort types.
When I couldn't reproduce the issue in #129445, I increased the index size to more than just one document and frequenty began running into another issue, namely getting errors of this kind:

https://gradle-enterprise.elastic.co/s/hjhqq6ium3hru/tests/task/:server:internalClusterTest/details/org.elasticsearch.search.sort.FieldSortIT/testSortMixedFieldTypesSeveralDocs?top-execution=1

WARNING: Uncaught exception in thread: Thread[#144,elasticsearch-error-rethrower,5,TGRP-FieldSortIT]
java.lang.AssertionError: java.lang.AssertionError: [org.elasticsearch.action.ActionListenerImplementations$RunBeforeActionListener/org.elasticsearch.action.search.TransportSearchAction$SearchResponseActionListener/org.elasticsearch.action.ActionListenerImplementations$RunBeforeActionListener/org.elasticsearch.tasks.TaskManager$1{org.elasticsearch.client.internal.support.AbstractClient$RefCountedFuture@4419e4fa}{CancellableTask{Task{id=30, type='transport', action='indices:data/read/search', description='null', parentTask=unset, startTime=1752232559616, headers={}, startTimeNanos=14563851254166}, reason='null', isCancelled=false}}/org.elasticsearch.action.support.TransportAction$$Lambda/0x000003000192f800@68f5045c/org.elasticsearch.action.search.AbstractSearchAsyncAction$$Lambda/0x0000030001b7c490@44774f6e]
	at __randomizedtesting.SeedInfo.seed([B5D33B2621614AE9]:0)
	at org.elasticsearch.action.ActionListener$3.assertFirstRun(ActionListener.java:404)
	at org.elasticsearch.action.ActionListener$3.onFailure(ActionListener.java:422)
	at org.elasticsearch.action.search.AbstractSearchAsyncAction.raisePhaseFailure(AbstractSearchAsyncAction.java:652)
	at org.elasticsearch.action.search.AbstractSearchAsyncAction.onPhaseFailure(AbstractSearchAsyncAction.java:629)
	at org.elasticsearch.action.search.SearchQueryThenFetchAsyncAction.onPhaseFailure(SearchQueryThenFetchAsyncAction.java:81)
	at org.elasticsearch.action.search.SearchQueryThenFetchAsyncAction$1.handleException(SearchQueryThenFetchAsyncAction.java:525)
	at 
[...]
Caused by: java.lang.AssertionError: [org.elasticsearch.action.ActionListenerImplementations$RunBeforeActionListener/org.elasticsearch.action.search.TransportSearchAction$SearchResponseActionListener/org.elasticsearch.action.ActionListenerImplementations$RunBeforeActionListener/org.elasticsearch.tasks.TaskManager$1{org.elasticsearch.client.internal.support.AbstractClient$RefCountedFuture@4419e4fa}{CancellableTask{Task{id=30, type='transport', action='indices:data/read/search', description='null', parentTask=unset, startTime=1752232559616, headers={}, startTimeNanos=14563851254166}, reason='null', isCancelled=false}}/org.elasticsearch.action.support.TransportAction$$Lambda/0x000003000192f800@68f5045c/org.elasticsearch.action.search.AbstractSearchAsyncAction$$Lambda/0x0000030001b7c490@44774f6e]
	... 42 more
Caused by: org.elasticsearch.ElasticsearchException: executed already
	at org.elasticsearch.action.ActionListener$3.assertFirstRun(ActionListener.java:403)
	... 41 more

[2025-07-11T14:15:59,717][WARN ][o.e.t.TcpTransport       ][node_s3][transport_worker][T#1] exception caught on transport layer [Netty4TcpChannel{localAddress=/127.0.0.1:49428, remoteAddress=127.0.0.1/127.0.0.1:14833, profile=default}], closing connection
org.elasticsearch.ElasticsearchException: executed already
	at org.elasticsearch.action.ActionListener$3.assertFirstRun(ActionListener.java:403) ~[main/:?]
	at org.elasticsearch.action.ActionListener$3.onFailure(ActionListener.java:422) ~[main/:?]
	at org.elasticsearch.action.search.AbstractSearchAsyncAction.raisePhaseFailure(AbstractSearchAsyncAction.java:652) ~[main/:?]
	at org.elasticsearch.action.search.AbstractSearchAsyncAction.onPhaseFailure(AbstractSearchAsyncAction.java:629) ~[main/:?]
	at org.elasticsearch.action.search.SearchQueryThenFetchAsyncAction.onPhaseFailure(SearchQueryThenFetchAsyncAction.java:81) ~[main/:?]
	at org.elasticsearch.action.search.SearchQueryThenFetchAsyncAction$1.handleException(SearchQueryThenFetchAsyncAction.java:525) ~[main/:?]
	at org.elasticsearch.transport.TransportService$UnregisterChildTransportResponseHandler.handleException(TransportService.java:1705) ~[main/:?]
	at org.elasticsearch.transport.TransportService$ContextRestoreResponseHandler.handleException(TransportService.java:1485) ~[main/:?]

Without the change in AbstractSearchAsyncAction#onPhaseFailure in this PR this is reproducible for me with:

./gradlew ":server:internalClusterTest" --tests "org.elasticsearch.search.sort.FieldSortIT.testSortMixedFieldTypesSeveralDocs" -Dtests.seed=B5D33B2621614AE9:3C91376D67454DCD --no-build-cache

I'm not sure preventing propagating any subsequent phase failure in AbstractSearchAsyncAction like suggested here is the right way to solve this, also I'm not really sure what big of an issue calling the ActionListener twice is in practice but the test assertion seems to be there for a reason. Also this might still be related to the weird errors we get in #129445 since that issue also seems to be related to search exception handling.

[benchaplin]

I agree it seems wrong to raise multiple phase failures for multiple data node failures.

I'm hesitant about this change, is it possible we're relying on multiple runs of raisePhaseFailure / listener.onFailure(exception) here, one per node search exception? Would love to hear more of your thoughts on that.

[benchaplin]

Maybe this is what we want, but this change would be lighter weight if this check was moved to the caller in SearchQueryThenFetchAsyncAction.

[benchaplin]

I've looked into this more after reviewing a similar exception in TransportSearchIT due to batched query execution. I'm now convinced this fix is correct - I think it makes the most sense to protect against calling raisePhaseFailure twice.

[cbuescher]

@javanna @benchaplin fyi just linked #134151 which looks like a similar failure.

Holding off as we continue to discuss...

[cbuescher]

@benchaplin I rebased and updated this fix as suggested in #134151, also verified that the reproducing test case from that issue passes now. I'm not sure if it is worth adding that repro since its pretty specific for aggs. Also I was considering moving the test added here to some other location than FieldSortIT since its main focus now should be testing the failure handling. With your recent changes in that area maybe you can suggest an alternative location?

[cbuescher]

@benchaplin after reading your comment on #134151 (comment) I think this PR either needs a follow up or should be closed in favor of another approach you sketched out in that comment. I'll take a look later how I would approach freeing contexts of in-flight requests as you suggested as a modification for this PR but will probably time-box that if it gets too involved.

[benchaplin]

Agreed @cbuescher, this change might be part of a solution but definitely not the whole thing. Don't sweat looking into it to much, I've been doing the same already :). Thanks for staying in the loop on this stuff with me!

[cbuescher]

@benchaplin quickly checking if I should close this in favor of #136889, wdyt?

[benchaplin]

Yep @cbuescher let's close this.