New name & tests for one-at-a-time /setup behavior

`firstPromiseBlocksAndFufills` for "the first promise created blocks others from being created, then fufills all with that first result"

x-pack/plugins/ingest_manager/server/services/setup.ts

@@ -26,7 +26,7 @@ import { packageConfigService } from './package_config';
 import { generateEnrollmentAPIKey } from './api_keys';
 import { settingsService } from '.';
 import { appContextService } from './app_context';
-import { limitOne } from './retry_setup';
+import { firstPromiseBlocksAndFufills } from './setup_utils';
 
 const FLEET_ENROLL_USERNAME = 'fleet_enroll';
 const FLEET_ENROLL_ROLE = 'fleet_enroll';
@@ -114,7 +114,7 @@ export async function setupIngestManager(
   soClient: SavedObjectsClientContract,
   callCluster: CallESAsCurrentUser
 ): Promise<SetupStatus> {
-  return limitOne(() => _setupIngestManager(soClient, callCluster));
+  return firstPromiseBlocksAndFufills(() => _setupIngestManager(soClient, callCluster));
 }
 
 export async function setupFleet(

x-pack/plugins/ingest_manager/server/services/setup_utils.test.ts

@@ -0,0 +1,164 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { firstPromiseBlocksAndFufills } from './setup_utils';
+
+async function sleep(ms: number) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+describe('limitOne', () => {
+  it('first promise called blocks others', async () => {
+    const fnA = jest.fn();
+    const fnB = jest.fn();
+    const fnC = jest.fn();
+    const fnD = jest.fn();
+    const promises = [
+      firstPromiseBlocksAndFufills(fnA),
+      firstPromiseBlocksAndFufills(fnB),
+      firstPromiseBlocksAndFufills(fnC),
+      firstPromiseBlocksAndFufills(fnD),
+    ];
+    await Promise.all(promises);
+
+    expect(fnA).toHaveBeenCalledTimes(1);
+    expect(fnB).toHaveBeenCalledTimes(0);
+    expect(fnC).toHaveBeenCalledTimes(0);
+    expect(fnD).toHaveBeenCalledTimes(0);
+  });
+
+  describe('first promise created, not necessarily first fufilled, sets value for all in queue', () => {
+    it('succeeds', async () => {
+      const fnA = jest.fn().mockImplementation(async () => {
+        await sleep(1000);
+        return 'called first';
+      });
+      const fnB = jest.fn().mockImplementation(async () => 'called second');
+      const fnC = jest.fn().mockImplementation(async () => 'called third');
+      const fnD = jest.fn().mockImplementation(async () => 'called fourth');
+      const promises = [
+        firstPromiseBlocksAndFufills(fnA),
+        firstPromiseBlocksAndFufills(fnB),
+        firstPromiseBlocksAndFufills(fnC),
+        firstPromiseBlocksAndFufills(fnD),
+      ];
+
+      expect(fnA).toHaveBeenCalledTimes(1);
+      expect(fnB).toHaveBeenCalledTimes(0);
+      expect(fnC).toHaveBeenCalledTimes(0);
+      expect(fnD).toHaveBeenCalledTimes(0);
+      await expect(Promise.all(promises)).resolves.toEqual([
+        'called first',
+        'called first',
+        'called first',
+        'called first',
+      ]);
+    });
+
+    it('throws', async () => {
+      const expectedError = new Error('error is called first');
+      const fnA = jest.fn().mockImplementation(async () => {
+        await sleep(1000);
+        throw expectedError;
+      });
+      const fnB = jest.fn().mockImplementation(async () => 'called second');
+      const fnC = jest.fn().mockImplementation(async () => 'called third');
+      const fnD = jest.fn().mockImplementation(async () => 'called fourth');
+      const promises = [
+        firstPromiseBlocksAndFufills(fnA),
+        firstPromiseBlocksAndFufills(fnB),
+        firstPromiseBlocksAndFufills(fnC),
+        firstPromiseBlocksAndFufills(fnD),
+      ];
+
+      await expect(Promise.all(promises)).rejects.toThrow(expectedError);
+      await expect(Promise.allSettled(promises)).resolves.toEqual([
+        { status: 'rejected', reason: expectedError },
+        { status: 'rejected', reason: expectedError },
+        { status: 'rejected', reason: expectedError },
+        { status: 'rejected', reason: expectedError },
+      ]);
+
+      expect(fnA).toHaveBeenCalledTimes(1);
+      expect(fnB).toHaveBeenCalledTimes(0);
+      expect(fnC).toHaveBeenCalledTimes(0);
+      expect(fnD).toHaveBeenCalledTimes(0);
+    });
+  });
+
+  it('does not block other calls after batch is fufilled. can call again for a new result', async () => {
+    const fnA = jest
+      .fn()
+      .mockImplementationOnce(() => 'fnA first')
+      .mockImplementationOnce(() => 'fnA second')
+      .mockImplementation(() => 'fnA default/2+');
+    const fnB = jest.fn();
+    const fnC = jest.fn();
+    const fnD = jest.fn();
+    let promises = [
+      firstPromiseBlocksAndFufills(fnA),
+      firstPromiseBlocksAndFufills(fnB),
+      firstPromiseBlocksAndFufills(fnC),
+      firstPromiseBlocksAndFufills(fnD),
+    ];
+    let results = await Promise.all(promises);
+
+    expect(fnA).toHaveBeenCalledTimes(1);
+    expect(fnB).toHaveBeenCalledTimes(0);
+    expect(fnC).toHaveBeenCalledTimes(0);
+    expect(fnD).toHaveBeenCalledTimes(0);
+    expect(results).toEqual(['fnA first', 'fnA first', 'fnA first', 'fnA first']);
+
+    promises = [
+      firstPromiseBlocksAndFufills(fnA),
+      firstPromiseBlocksAndFufills(fnB),
+      firstPromiseBlocksAndFufills(fnC),
+      firstPromiseBlocksAndFufills(fnD),
+    ];
+    results = await Promise.all(promises);
+    expect(fnA).toHaveBeenCalledTimes(2);
+    expect(fnB).toHaveBeenCalledTimes(0);
+    expect(fnC).toHaveBeenCalledTimes(0);
+    expect(fnD).toHaveBeenCalledTimes(0);
+    expect(results).toEqual(['fnA second', 'fnA second', 'fnA second', 'fnA second']);
+
+    promises = [
+      firstPromiseBlocksAndFufills(fnA),
+      firstPromiseBlocksAndFufills(fnB),
+      firstPromiseBlocksAndFufills(fnC),
+      firstPromiseBlocksAndFufills(fnD),
+    ];
+    results = await Promise.all(promises);
+    expect(fnA).toHaveBeenCalledTimes(3);
+    expect(fnB).toHaveBeenCalledTimes(0);
+    expect(fnC).toHaveBeenCalledTimes(0);
+    expect(fnD).toHaveBeenCalledTimes(0);
+    expect(results).toEqual([
+      'fnA default/2+',
+      'fnA default/2+',
+      'fnA default/2+',
+      'fnA default/2+',
+    ]);
+
+    promises = [
+      firstPromiseBlocksAndFufills(fnA),
+      firstPromiseBlocksAndFufills(fnB),
+      firstPromiseBlocksAndFufills(fnC),
+      firstPromiseBlocksAndFufills(fnD),
+    ];
+    results = await Promise.all(promises);
+    expect(fnA).toHaveBeenCalledTimes(4);
+    expect(fnB).toHaveBeenCalledTimes(0);
+    expect(fnC).toHaveBeenCalledTimes(0);
+    expect(fnD).toHaveBeenCalledTimes(0);
+    expect(results).toEqual([
+      'fnA default/2+',
+      'fnA default/2+',
+      'fnA default/2+',
+      'fnA default/2+',
+    ]);
+  });
+});

x-pack/plugins/ingest_manager/server/services/retry_setup.ts → x-pack/plugins/ingest_manager/server/services/setup_utils.ts

@@ -4,38 +4,36 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { SetupStatus } from './setup';
-
 // the promise which tracks the setup
-let setupIngestStatus: Promise<SetupStatus> | undefined;
+let status: Promise<unknown> | undefined;
 // default resolve to guard against "undefined is not a function" errors
-let onSetupResolve = (status: SetupStatus) => {};
-let onSetupReject = (reason: any) => {};
+let onResolve = (value?: unknown) => {};
+let onReject = (reason: any) => {};
 
-export async function limitOne(asyncFunction: Function) {
-  // pending pending or successful attempt
-  if (setupIngestStatus) {
+export async function firstPromiseBlocksAndFufills(asyncFunction: Function) {
+  // pending successful or failed attempt
+  if (status) {
     // don't run concurrent installs
-    return setupIngestStatus;
+    return status;
   } else {
     // create the initial promise
-    setupIngestStatus = new Promise((res, rej) => {
-      onSetupResolve = res;
-      onSetupReject = rej;
+    status = new Promise((res, rej) => {
+      onResolve = res;
+      onReject = rej;
     });
   }
   try {
     // if everything works, mark the tracking promise as resolved
     const result = await asyncFunction();
-    onSetupResolve(result);
+    onResolve(result);
     // * reset the tracking promise to try again next time
-    setupIngestStatus = undefined;
+    status = undefined;
     return result;
   } catch (error) {
     // if something fails
-    onSetupReject(error);
+    onReject(error);
     // * reset the tracking promise to try again next time
-    setupIngestStatus = undefined;
+    status = undefined;
     // * return the rejection so it can be dealt with now
     return Promise.reject(error);
   }