[Search] Set keep_alive parameter in async search (elastic#73712)

* [Search] Set keep_alive parameter in async search

* Revert accidental change

* Add batched_reduce_size

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

x-pack/plugins/data_enhanced/server/search/es_search_strategy.test.ts

@@ -113,4 +113,19 @@ describe('ES search strategy', () => {
     expect(method).toBe('POST');
     expect(path).toBe('/foo-%E7%A8%8B/_rollup_search');
   });
+
+  it('sets wait_for_completion_timeout and keep_alive in the request', async () => {
+    mockApiCaller.mockResolvedValueOnce(mockAsyncResponse);
+
+    const params = { index: 'foo-*', body: {} };
+    const esSearch = await enhancedEsSearchStrategyProvider(mockConfig$, mockLogger);
+
+    await esSearch.search((mockContext as unknown) as RequestHandlerContext, { params });
+
+    expect(mockApiCaller).toBeCalled();
+    expect(mockApiCaller.mock.calls[0][0]).toBe('transport.request');
+    const { query } = mockApiCaller.mock.calls[0][1];
+    expect(query).toHaveProperty('wait_for_completion_timeout');
+    expect(query).toHaveProperty('keep_alive');
+  });
 });

x-pack/plugins/data_enhanced/server/search/es_search_strategy.ts

@@ -80,8 +80,15 @@ async function asyncSearch(
   const method = request.id ? 'GET' : 'POST';
   const path = encodeURI(request.id ? `/_async_search/${request.id}` : `/${index}/_async_search`);
 
-  // Wait up to 1s for the response to return
-  const query = toSnakeCase({ waitForCompletionTimeout: '100ms', ...queryParams });
+  // Only report partial results every 64 shards; this should be reduced when we actually display partial results
+  const batchedReduceSize = request.id ? undefined : 64;
+
+  const query = toSnakeCase({
+    waitForCompletionTimeout: '100ms', // Wait up to 100ms for the response to return
+    keepAlive: '1m', // Extend the TTL for this search request by one minute
+    ...(batchedReduceSize && { batchedReduceSize }),
+    ...queryParams,
+  });
 
   const { id, response, is_partial, is_running } = (await caller(
     'transport.request',