[eslint] add rule to forbid async subscribe bodies

dev_docs/tutorials/testing_plugins.mdx

@@ -493,7 +493,7 @@ export const renderApp = (
 
   // uiSettings subscription
   const uiSettingsClient = core.uiSettings.client;
-  const pollingSubscription = uiSettingClient.get$('mysetting1').subscribe(async mySetting1 => {
+  const pollingSubscription = uiSettingClient.get$('mysetting1').subscribe((mySetting1) => {
     const value = core.http.fetch(/** use `mySetting1` in request **/);
     // ...
   });

examples/embeddable_examples/public/book/book_embeddable.tsx

@@ -70,7 +70,7 @@ export class BookEmbeddable
   ) {
     super(initialInput, {} as BookEmbeddableOutput, parent);
 
-    this.subscription = this.getInput$().subscribe(async () => {
+    this.subscription = this.getInput$().subscribe(() => {
       const savedObjectId = (this.getInput() as BookByReferenceInput).savedObjectId;
       const attributes = (this.getInput() as BookByValueInput).attributes;
       if (this.attributes !== attributes || this.savedObjectId !== savedObjectId) {

examples/embeddable_examples/public/todo/todo_ref_embeddable.tsx

@@ -9,6 +9,7 @@
 import React from 'react';
 import ReactDOM from 'react-dom';
 import { Subscription } from 'rxjs';
+import { mergeMap } from 'rxjs/operators';
 import { TodoSavedObjectAttributes } from 'examples/embeddable_examples/common';
 import { SavedObjectsClientContract } from 'kibana/public';
 import {
@@ -17,6 +18,7 @@ import {
   EmbeddableOutput,
   SavedObjectEmbeddableInput,
 } from '../../../../src/plugins/embeddable/public';
+import { FatalErrorEvent } from '../../../../src/core/public';
 import { TodoRefEmbeddableComponent } from './todo_ref_component';
 
 // Notice this is not the same value as the 'todo' saved object type. Many of our
@@ -83,30 +85,38 @@ export class TodoRefEmbeddable extends Embeddable<TodoRefInput, TodoRefOutput> {
     super(initialInput, { hasMatch: false }, parent);
     this.savedObjectsClient = savedObjectsClient;
 
-    this.subscription = this.getInput$().subscribe(async () => {
-      // There is a little more work today for this embeddable because it has
-      // more output it needs to update in response to input state changes.
-      let savedAttributes: TodoSavedObjectAttributes | undefined;
+    this.subscription = this.getInput$()
+      .pipe(
+        mergeMap(async () => {
+          // There is a little more work today for this embeddable because it has
+          // more output it needs to update in response to input state changes.
+          let savedAttributes: TodoSavedObjectAttributes | undefined;
 
-      // Since this is an expensive task, we save a local copy of the previous
-      // savedObjectId locally and only retrieve the new saved object if the id
-      // actually changed.
-      if (this.savedObjectId !== this.input.savedObjectId) {
-        this.savedObjectId = this.input.savedObjectId;
-        const todoSavedObject = await this.savedObjectsClient.get<TodoSavedObjectAttributes>(
-          'todo',
-          this.input.savedObjectId
-        );
-        savedAttributes = todoSavedObject?.attributes;
-      }
+          // Since this is an expensive task, we save a local copy of the previous
+          // savedObjectId locally and only retrieve the new saved object if the id
+          // actually changed.
+          if (this.savedObjectId !== this.input.savedObjectId) {
+            this.savedObjectId = this.input.savedObjectId;
+            const todoSavedObject = await this.savedObjectsClient.get<TodoSavedObjectAttributes>(
+              'todo',
+              this.input.savedObjectId
+            );
+            savedAttributes = todoSavedObject?.attributes;
+          }
 
-      // The search string might have changed as well so we need to make sure we recalculate
-      // hasMatch.
-      this.updateOutput({
-        hasMatch: getHasMatch(this.input.search, savedAttributes),
-        savedAttributes,
+          // The search string might have changed as well so we need to make sure we recalculate
+          // hasMatch.
+          this.updateOutput({
+            hasMatch: getHasMatch(this.input.search, savedAttributes),
+            savedAttributes,
+          });
+        })
+      )
+      .subscribe({
+        error: (error) => {
+          window.dispatchEvent(new FatalErrorEvent(error));
+        },
       });
-    });
   }
 
   public render(node: HTMLElement) {

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()));
+}