Avoid double decrement on current query counter

This commit fixes a double decrement bug on the current query
counter. The double decrement arises in a situation when the fetch phase
is inlined for a query that is only touching one shard. After the query
phase succeeds we decrement the current query counter. If the fetch
phase ultimately fails, an exception is thrown and we decrement the
current query counter again in the catch block. We also add assertions
that all current stats counters remain non-negative at all
times.

Relates #24922

Author: jasontedor

core/src/main/java/org/elasticsearch/index/search/stats/ShardSearchStats.java

@@ -82,8 +82,10 @@ public void onFailedQueryPhase(SearchContext searchContext) {
         computeStats(searchContext, statsHolder -> {
             if (searchContext.hasOnlySuggest()) {
                 statsHolder.suggestCurrent.dec();
+                assert statsHolder.suggestCurrent.count() >= 0;
             } else {
                 statsHolder.queryCurrent.dec();
+                assert statsHolder.queryCurrent.count() >= 0;
             }
         });
     }
@@ -94,9 +96,11 @@ public void onQueryPhase(SearchContext searchContext, long tookInNanos) {
             if (searchContext.hasOnlySuggest()) {
                 statsHolder.suggestMetric.inc(tookInNanos);
                 statsHolder.suggestCurrent.dec();
+                assert statsHolder.suggestCurrent.count() >= 0;
             } else {
                 statsHolder.queryMetric.inc(tookInNanos);
                 statsHolder.queryCurrent.dec();
+                assert statsHolder.queryCurrent.count() >= 0;
             }
         });
     }
@@ -116,6 +120,7 @@ public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
         computeStats(searchContext, statsHolder -> {
             statsHolder.fetchMetric.inc(tookInNanos);
             statsHolder.fetchCurrent.dec();
+            assert statsHolder.fetchCurrent.count() >= 0;
         });
     }
 
@@ -176,6 +181,7 @@ public void onNewScrollContext(SearchContext context) {
     @Override
     public void onFreeScrollContext(SearchContext context) {
         totalStats.scrollCurrent.dec();
+        assert totalStats.scrollCurrent.count() >= 0;
         totalStats.scrollMetric.inc(System.nanoTime() - context.getOriginNanoTime());
     }
 

core/src/main/java/org/elasticsearch/search/SearchService.java

@@ -252,6 +252,7 @@ public SearchPhaseResult executeQueryPhase(ShardSearchRequest request, SearchTas
         final SearchContext context = createAndPutContext(request);
         final SearchOperationListener operationListener = context.indexShard().getSearchOperationListener();
         context.incRef();
+        boolean queryPhaseSuccess = false;
         try {
             context.setTask(task);
             operationListener.onPreQueryPhase(context);
@@ -266,6 +267,7 @@ public SearchPhaseResult executeQueryPhase(ShardSearchRequest request, SearchTas
                 contextProcessedSuccessfully(context);
             }
             final long afterQueryTime = System.nanoTime();
+            queryPhaseSuccess = true;
             operationListener.onQueryPhase(context, afterQueryTime - time);
             if (request.numberOfShards() == 1) {
                 return executeFetchPhase(context, operationListener, afterQueryTime);
@@ -277,7 +279,9 @@ public SearchPhaseResult executeQueryPhase(ShardSearchRequest request, SearchTas
                 e = (e.getCause() == null || e.getCause() instanceof Exception) ?
                     (Exception) e.getCause() : new ElasticsearchException(e.getCause());
             }
-            operationListener.onFailedQueryPhase(context);
+            if (!queryPhaseSuccess) {
+                operationListener.onFailedQueryPhase(context);
+            }
             logger.trace("Query phase failed", e);
             processFailure(context, e);
             throw ExceptionsHelper.convertToRuntime(e);