[eslint] add rule to forbid async forEach bodies

packages/elastic-eslint-config-kibana/.eslintrc.js

@@ -92,5 +92,6 @@ module.exports = {
     ],
 
     '@kbn/eslint/no_async_promise_body': 'error',
+    '@kbn/eslint/no_async_foreach': 'error',
   },
 };

packages/kbn-eslint-plugin-eslint/index.js

@@ -14,5 +14,6 @@ module.exports = {
     module_migration: require('./rules/module_migration'),
     no_export_all: require('./rules/no_export_all'),
     no_async_promise_body: require('./rules/no_async_promise_body'),
+    no_async_foreach: require('./rules/no_async_foreach'),
   },
 };

packages/kbn-eslint-plugin-eslint/rules/no_async_foreach.js

@@ -0,0 +1,62 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+const tsEstree = require('@typescript-eslint/typescript-estree');
+const esTypes = tsEstree.AST_NODE_TYPES;
+
+/** @typedef {import("eslint").Rule.RuleModule} Rule */
+/** @typedef {import("@typescript-eslint/typescript-estree").TSESTree.Node} Node */
+/** @typedef {import("@typescript-eslint/typescript-estree").TSESTree.CallExpression} CallExpression */
+/** @typedef {import("@typescript-eslint/typescript-estree").TSESTree.FunctionExpression} FunctionExpression */
+/** @typedef {import("@typescript-eslint/typescript-estree").TSESTree.ArrowFunctionExpression} ArrowFunctionExpression */
+/** @typedef {import("eslint").Rule.RuleFixer} Fixer */
+
+const ERROR_MSG =
+  'Passing an async function to .forEach() prevents promise rejections from being handled. Use asyncForEach() or similar helper from "@kbn/std" instead.';
+
+/**
+ * @param {Node} node
+ * @returns {node is ArrowFunctionExpression | FunctionExpression}
+ */
+const isFunc = (node) =>
+  node.type === esTypes.ArrowFunctionExpression || node.type === esTypes.FunctionExpression;
+
+/**
+ * @param {any} context
+ * @param {CallExpression} node
+ */
+const isAsyncForEachCall = (node) => {
+  return (
+    node.callee.type === esTypes.MemberExpression &&
+    node.callee.property.type === esTypes.Identifier &&
+    node.callee.property.name === 'forEach' &&
+    node.arguments.length >= 1 &&
+    isFunc(node.arguments[0]) &&
+    node.arguments[0].async
+  );
+};
+
+/** @type {Rule} */
+module.exports = {
+  meta: {
+    fixable: 'code',
+    schema: [],
+  },
+  create: (context) => ({
+    CallExpression(_) {
+      const node = /** @type {CallExpression} */ (_);
+
+      if (isAsyncForEachCall(node)) {
+        context.report({
+          message: ERROR_MSG,
+          loc: node.arguments[0].loc,
+        });
+      }
+    },
+  }),
+};

packages/kbn-eslint-plugin-eslint/rules/no_async_foreach.test.js

@@ -0,0 +1,72 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+const { RuleTester } = require('eslint');
+const rule = require('./no_async_foreach');
+const dedent = require('dedent');
+
+const ruleTester = new RuleTester({
+  parser: require.resolve('@typescript-eslint/parser'),
+  parserOptions: {
+    sourceType: 'module',
+    ecmaVersion: 2018,
+    ecmaFeatures: {
+      jsx: true,
+    },
+  },
+});
+
+ruleTester.run('@kbn/eslint/no_async_foreach', rule, {
+  valid: [
+    {
+      code: dedent`
+        array.forEach((a) => {
+          b(a)
+        })
+      `,
+    },
+    {
+      code: dedent`
+        array.forEach(function (a) {
+          b(a)
+        })
+      `,
+    },
+  ],
+
+  invalid: [
+    {
+      code: dedent`
+        array.forEach(async (a) => {
+          await b(a)
+        })
+      `,
+      errors: [
+        {
+          line: 1,
+          message:
+            'Passing an async function to .forEach() prevents promise rejections from being handled. Use asyncForEach() or similar helper from "@kbn/std" instead.',
+        },
+      ],
+    },
+    {
+      code: dedent`
+        array.forEach(async function (a) {
+          await b(a)
+        })
+      `,
+      errors: [
+        {
+          line: 1,
+          message:
+            'Passing an async function to .forEach() prevents promise rejections from being handled. Use asyncForEach() or similar helper from "@kbn/std" instead.',
+        },
+      ],
+    },
+  ],
+});

packages/kbn-std/BUILD.bazel

@@ -9,7 +9,10 @@ SOURCE_FILES = glob(
   [
     "src/**/*.ts",
   ],
-  exclude = ["**/*.test.*"],
+  exclude = [
+    "**/*.test.*",
+    "**/test_helpers.ts",
+  ],
 )
 
 SRCS = SOURCE_FILES

packages/kbn-std/src/index.ts

@@ -18,3 +18,11 @@ export { unset } from './unset';
 export { getFlattenedObject } from './get_flattened_object';
 export { ensureNoUnsafeProperties } from './ensure_no_unsafe_properties';
 export * from './rxjs_7';
+export {
+  map$,
+  mapWithLimit$,
+  asyncMap,
+  asyncMapWithLimit,
+  asyncForEach,
+  asyncForEachWithLimit,
+} from './iteration';

packages/kbn-std/src/iteration/for_each.test.ts

@@ -0,0 +1,81 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import * as Rx from 'rxjs';
+
+import { asyncForEach, asyncForEachWithLimit } from './for_each';
+import { list, sleep } from './test_helpers';
+
+jest.mock('./observable');
+const mockMapWithLimit$: jest.Mock = jest.requireMock('./observable').mapWithLimit$;
+
+beforeEach(() => {
+  jest.clearAllMocks();
+});
+
+describe('asyncForEachWithLimit', () => {
+  it('calls mapWithLimit$ and resolves with undefined when it completes', async () => {
+    const iter = list(10);
+    const limit = 5;
+    const fn = jest.fn();
+
+    const result$ = new Rx.Subject();
+    mockMapWithLimit$.mockImplementation(() => result$);
+    const promise = asyncForEachWithLimit(iter, limit, fn);
+
+    let resolved = false;
+    promise.then(() => (resolved = true));
+
+    expect(mockMapWithLimit$).toHaveBeenCalledTimes(1);
+    expect(mockMapWithLimit$).toHaveBeenCalledWith(iter, limit, fn);
+
+    expect(resolved).toBe(false);
+    result$.next(1);
+    result$.next(2);
+    result$.next(3);
+
+    await sleep(10);
+    expect(resolved).toBe(false);
+
+    result$.complete();
+    await expect(promise).resolves.toBe(undefined);
+  });
+
+  it('resolves when iterator is empty', async () => {
+    mockMapWithLimit$.mockImplementation((x) => Rx.from(x));
+    await expect(asyncForEachWithLimit([], 100, async () => 'foo')).resolves.toBe(undefined);
+  });
+});
+
+describe('asyncForEach', () => {
+  it('calls mapWithLimit$ without limit and resolves with undefined when it completes', async () => {
+    const iter = list(10);
+    const fn = jest.fn();
+
+    const result$ = new Rx.Subject();
+    mockMapWithLimit$.mockImplementation(() => result$);
+    const promise = asyncForEach(iter, fn);
+
+    let resolved = false;
+    promise.then(() => (resolved = true));
+
+    expect(mockMapWithLimit$).toHaveBeenCalledTimes(1);
+    expect(mockMapWithLimit$).toHaveBeenCalledWith(iter, Infinity, fn);
+
+    expect(resolved).toBe(false);
+    result$.next(1);
+    result$.next(2);
+    result$.next(3);
+
+    await sleep(10);
+    expect(resolved).toBe(false);
+
+    result$.complete();
+    await expect(promise).resolves.toBe(undefined);
+  });
+});

packages/kbn-std/src/iteration/for_each.ts

@@ -0,0 +1,44 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { defaultIfEmpty } from 'rxjs/operators';
+
+import { lastValueFrom } from '../rxjs_7';
+import { mapWithLimit$ } from './observable';
+import { IterableInput, AsyncMapFn } from './types';
+
+/**
+ * Creates a promise which resolves with `undefined` after calling `fn` for each
+ * item in `iterable`. `fn` can return either a Promise or Observable. If `fn`
+ * returns observables then they will properly abort if an error occurs.
+ *
+ * @param iterable Items to iterate
+ * @param fn Function to call for each item
+ */
+export async function asyncForEach<T>(iterable: IterableInput<T>, fn: AsyncMapFn<T, any>) {
+  await lastValueFrom(mapWithLimit$(iterable, Infinity, fn).pipe(defaultIfEmpty()));
+}
+
+/**
+ * Creates a promise which resolves with `undefined` after calling `fn` for each
+ * item in `iterable`. `fn` can return either a Promise or Observable. If `fn`
+ * returns observables then they will properly abort if an error occurs.
+ *
+ * The number of concurrent executions of `fn` is limited by `limit`.
+ *
+ * @param iterable Items to iterate
+ * @param limit Maximum number of operations to run in parallel
+ * @param fn Function to call for each item
+ */
+export async function asyncForEachWithLimit<T>(
+  iterable: IterableInput<T>,
+  limit: number,
+  fn: AsyncMapFn<T, any>
+) {
+  await lastValueFrom(mapWithLimit$(iterable, limit, fn).pipe(defaultIfEmpty()));
+}

packages/kbn-std/src/iteration/index.ts

@@ -0,0 +1,11 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+export * from './observable';
+export * from './for_each';
+export * from './map';

packages/kbn-std/src/iteration/map.test.ts

@@ -0,0 +1,82 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import * as Rx from 'rxjs';
+import { mapTo } from 'rxjs/operators';
+
+import { asyncMap, asyncMapWithLimit } from './map';
+import { list } from './test_helpers';
+
+jest.mock('./observable');
+const mapWithLimit$: jest.Mock = jest.requireMock('./observable').mapWithLimit$;
+mapWithLimit$.mockImplementation(jest.requireActual('./observable').mapWithLimit$);
+
+beforeEach(() => {
+  jest.clearAllMocks();
+});
+
+describe('asyncMapWithLimit', () => {
+  it('calls mapWithLimit$ and resolves with properly sorted results', async () => {
+    const iter = list(10);
+    const limit = 5;
+    const fn = jest.fn((n) => (n % 2 ? Rx.timer(n) : Rx.timer(n * 4)).pipe(mapTo(n)));
+    const result = await asyncMapWithLimit(iter, limit, fn);
+
+    expect(result).toMatchInlineSnapshot(`
+      Array [
+        0,
+        1,
+        2,
+        3,
+        4,
+        5,
+        6,
+        7,
+        8,
+        9,
+      ]
+    `);
+
+    expect(mapWithLimit$).toHaveBeenCalledTimes(1);
+    expect(mapWithLimit$).toHaveBeenCalledWith(iter, limit, expect.any(Function));
+  });
+
+  it.each([
+    [list(0), []] as const,
+    [list(1), ['foo']] as const,
+    [list(2), ['foo', 'foo']] as const,
+  ])('resolves when iterator is %p', async (input, output) => {
+    await expect(asyncMapWithLimit(input, 100, async () => 'foo')).resolves.toEqual(output);
+  });
+});
+
+describe('asyncMap', () => {
+  it('calls mapWithLimit$ without limit and resolves with undefined when it completes', async () => {
+    const iter = list(10);
+    const fn = jest.fn((n) => (n % 2 ? Rx.timer(n) : Rx.timer(n * 4)).pipe(mapTo(n)));
+    const result = await asyncMap(iter, fn);
+
+    expect(result).toMatchInlineSnapshot(`
+      Array [
+        0,
+        1,
+        2,
+        3,
+        4,
+        5,
+        6,
+        7,
+        8,
+        9,
+      ]
+    `);
+
+    expect(mapWithLimit$).toHaveBeenCalledTimes(1);
+    expect(mapWithLimit$).toHaveBeenCalledWith(iter, Infinity, expect.any(Function));
+  });
+});