[data.search.bsearch] Forward request abortSignal to search strategy (#169041)

## Summary

Creates an `abortSignal` from the request disconnected event that is
forwarded to the search strategy. In practice this means that when a
bsearch call is disconnected (either due to client disconnect or server
timeout) the corresponding call to ES is also cancelled.

### Checklist

Delete any items that are not applicable to this PR.

- [ ] Any text added follows [EUI's writing
guidelines](https://elastic.github.io/eui/#/guidelines/writing), uses
sentence case text and includes [i18n
support](https://github.com/elastic/kibana/blob/main/packages/kbn-i18n/README.md)
- [ ]
[Documentation](https://www.elastic.co/guide/en/kibana/master/development-documentation.html)
was added for features that require explanation or tutorials
- [ ] [Unit or functional
tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html)
were updated or added to match the most common scenarios
- [ ] Any UI touched in this PR is usable by keyboard only (learn more
about [keyboard accessibility](https://webaim.org/techniques/keyboard/))
- [ ] Any UI touched in this PR does not create any new axe failures
(run axe in browser:
[FF](https://addons.mozilla.org/en-US/firefox/addon/axe-devtools/),
[Chrome](https://chrome.google.com/webstore/detail/axe-web-accessibility-tes/lhdoppojpmngadmnindnejefpokejbdd?hl=en-US))
- [ ] If a plugin configuration key changed, check if it needs to be
allowlisted in the cloud and added to the [docker
list](https://github.com/elastic/kibana/blob/main/src/dev/build/tasks/os_packages/docker_generator/resources/base/bin/kibana-docker)
- [ ] This renders correctly on smaller devices using a responsive
layout. (You can test this [in your
browser](https://www.browserstack.com/guide/responsive-testing-on-local-server))
- [ ] This was checked for [cross-browser
compatibility](https://www.elastic.co/support/matrix#matrix_browsers)

---------

Co-authored-by: Stratoula Kalafateli <efstratia.kalafateli@elastic.co>
Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

src/plugins/data/common/search/utils.ts

@@ -23,7 +23,9 @@ export const isAbortResponse = (response?: IKibanaSearchResponse) => {
 /**
  * @returns true if request is still running
  */
-export const isRunningResponse = (response?: IKibanaSearchResponse) => response?.isRunning ?? false;
+export const isRunningResponse = (response?: IKibanaSearchResponse) => {
+  return response?.isRunning ?? false;
+};
 
 export const getUserTimeZone = (
   getConfig: AggTypesDependencies['getConfig'],

src/plugins/data/server/search/routes/bsearch.ts

@@ -11,6 +11,7 @@ import { catchError } from 'rxjs/operators';
 import { BfetchServerSetup } from '@kbn/bfetch-plugin/server';
 import type { ExecutionContextSetup } from '@kbn/core/server';
 import apm from 'elastic-apm-node';
+import { getRequestAbortedSignal } from '../..';
 import {
   IKibanaSearchRequest,
   IKibanaSearchResponse,
@@ -28,6 +29,7 @@ export function registerBsearchRoute(
     IKibanaSearchResponse
   >('/internal/bsearch', (request) => {
     const search = getScoped(request);
+    const abortSignal = getRequestAbortedSignal(request.events.aborted$);
     return {
       /**
        * @param requestOptions
@@ -39,7 +41,7 @@ export function registerBsearchRoute(
           apm.addLabels(executionContextService.getAsLabels());
 
           return firstValueFrom(
-            search.search(requestData, restOptions).pipe(
+            search.search(requestData, { ...restOptions, abortSignal }).pipe(
               catchError((err) => {
                 // Re-throw as object, to get attributes passed to the client
                 // eslint-disable-next-line no-throw-literal

src/plugins/data/server/search/strategies/ese_search/ese_search_strategy.test.ts

@@ -259,6 +259,38 @@ describe('ES search strategy', () => {
 
         expect(mockApiCaller).toBeCalledTimes(0);
       });
+
+      it('should delete when aborted', async () => {
+        mockSubmitCaller.mockResolvedValueOnce({
+          ...mockAsyncResponse,
+          body: {
+            ...mockAsyncResponse.body,
+            is_running: true,
+          },
+        });
+
+        const params = { index: 'logstash-*', body: { query: {} } };
+        const esSearch = await enhancedEsSearchStrategyProvider(
+          mockLegacyConfig$,
+          mockSearchConfig,
+          mockLogger
+        );
+        const abortController = new AbortController();
+        const abortSignal = abortController.signal;
+
+        // Abort after an incomplete first response is returned
+        setTimeout(() => abortController.abort(), 100);
+
+        let err: KbnServerError | undefined;
+        try {
+          await esSearch.search({ params }, { abortSignal }, mockDeps).toPromise();
+        } catch (e) {
+          err = e;
+        }
+        expect(mockSubmitCaller).toBeCalled();
+        expect(err).not.toBeUndefined();
+        expect(mockDeleteCaller).toBeCalled();
+      });
     });
 
     describe('with sessionId', () => {
@@ -366,6 +398,44 @@ describe('ES search strategy', () => {
         expect(request).toHaveProperty('wait_for_completion_timeout');
         expect(request).not.toHaveProperty('keep_alive');
       });
+
+      it('should not delete a saved session when aborted', async () => {
+        mockSubmitCaller.mockResolvedValueOnce({
+          ...mockAsyncResponse,
+          body: {
+            ...mockAsyncResponse.body,
+            is_running: true,
+          },
+        });
+
+        const params = { index: 'logstash-*', body: { query: {} } };
+        const esSearch = await enhancedEsSearchStrategyProvider(
+          mockLegacyConfig$,
+          mockSearchConfig,
+          mockLogger
+        );
+        const abortController = new AbortController();
+        const abortSignal = abortController.signal;
+
+        // Abort after an incomplete first response is returned
+        setTimeout(() => abortController.abort(), 100);
+
+        let err: KbnServerError | undefined;
+        try {
+          await esSearch
+            .search(
+              { params },
+              { abortSignal, sessionId: '1', isSearchStored: true, isStored: true },
+              mockDeps
+            )
+            .toPromise();
+        } catch (e) {
+          err = e;
+        }
+        expect(mockSubmitCaller).toBeCalled();
+        expect(err).not.toBeUndefined();
+        expect(mockDeleteCaller).not.toBeCalled();
+      });
     });
 
     it('throws normalized error if ResponseError is thrown', async () => {

src/plugins/data/server/search/strategies/ese_search/ese_search_strategy.ts

@@ -82,7 +82,7 @@ export const enhancedEsSearchStrategyProvider = (
     };
 
     const cancel = async () => {
-      if (id) {
+      if (id && !options.isStored) {
         await cancelAsyncSearch(id, esClient);
       }
     };