[7.6] Abort cancelled search requests to Elasticsearch (#56788)

src/plugins/data/server/autocomplete/value_suggestions_route.ts

@@ -24,6 +24,7 @@ import { IRouter, SharedGlobalConfig } from 'kibana/server';
 import { Observable } from 'rxjs';
 import { first } from 'rxjs/operators';
 import { IFieldType, indexPatterns, esFilters } from '../index';
+import { getRequestAbortedSignal } from '../lib';
 
 export function registerValueSuggestionsRoute(
   router: IRouter,
@@ -54,6 +55,7 @@ export function registerValueSuggestionsRoute(
       const { field: fieldName, query, boolFilter } = request.body;
       const { index } = request.params;
       const { dataClient } = context.core.elasticsearch;
+      const signal = getRequestAbortedSignal(request.events.aborted$);
 
       const autocompleteSearchOptions = {
         timeout: `${config.kibana.autocompleteTimeout.asMilliseconds()}ms`,
@@ -69,7 +71,7 @@ export function registerValueSuggestionsRoute(
       const body = await getBody(autocompleteSearchOptions, field || fieldName, query, boolFilter);
 
       try {
-        const result = await dataClient.callAsCurrentUser('search', { index, body });
+        const result = await dataClient.callAsCurrentUser('search', { index, body }, { signal });
 
         const buckets: any[] =
           get(result, 'aggregations.suggestions.buckets') ||

src/plugins/data/server/lib/get_request_aborted_signal.test.ts

@@ -0,0 +1,45 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { Subject } from 'rxjs';
+import { getRequestAbortedSignal } from './get_request_aborted_signal';
+
+describe('abortableRequestHandler', () => {
+  jest.useFakeTimers();
+
+  it('should call abort if disconnected', () => {
+    const abortedSubject = new Subject<void>();
+    const aborted$ = abortedSubject.asObservable();
+    const onAborted = jest.fn();
+
+    const signal = getRequestAbortedSignal(aborted$);
+    signal.addEventListener('abort', onAborted);
+
+    // Shouldn't be aborted or call onAborted prior to disconnecting
+    expect(signal.aborted).toBe(false);
+    expect(onAborted).not.toBeCalled();
+
+    abortedSubject.next();
+    jest.runAllTimers();
+
+    // Should be aborted and call onAborted after disconnecting
+    expect(signal.aborted).toBe(true);
+    expect(onAborted).toBeCalled();
+  });
+});

src/plugins/data/server/lib/get_request_aborted_signal.ts

@@ -0,0 +1,33 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { Observable } from 'rxjs';
+// @ts-ignore not typed
+import { AbortController } from 'abortcontroller-polyfill/dist/cjs-ponyfill';
+
+/**
+ * A simple utility function that returns an `AbortSignal` corresponding to an `AbortController`
+ * which aborts when the given request is aborted.
+ * @param aborted$ The observable of abort events (usually `request.events.aborted$`)
+ */
+export function getRequestAbortedSignal(aborted$: Observable<void>): AbortSignal {
+  const controller = new AbortController();
+  aborted$.subscribe(() => controller.abort());
+  return controller.signal;
+}

src/plugins/data/server/lib/index.ts

@@ -0,0 +1,20 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+export { getRequestAbortedSignal } from './get_request_aborted_signal';

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

@@ -19,6 +19,7 @@
 
 import { schema } from '@kbn/config-schema';
 import { IRouter } from '../../../../core/server';
+import { getRequestAbortedSignal } from '../lib';
 
 export function registerSearchRoute(router: IRouter): void {
   router.post(
@@ -35,8 +36,10 @@ export function registerSearchRoute(router: IRouter): void {
     async (context, request, res) => {
       const searchRequest = request.body;
       const strategy = request.params.strategy;
+      const signal = getRequestAbortedSignal(request.events.aborted$);
+
       try {
-        const response = await context.search!.search(searchRequest, {}, strategy);
+        const response = await context.search!.search(searchRequest, { signal }, strategy);
         return res.ok({ body: response });
       } catch (err) {
         return res.internalError({ body: err });