[Event Hubs] Rewrite partition receiver (Azure#24731)

# Re-implementing the Event Receiver
This PR re-implements the event receiver using promises and a single
queue to fix an ordering issue and to correct waiting behavior.
## Problem Statement [Issue Azure#23993]
A customer reported that the list of events passed into the
`processEvents` callback is not always ordered by `sequenceNumber`. This
leads to processing the events in a wrong order. The customer provided a
sample that prints an out of order message when the `sequenceNumber` of
received messages is not in order and I confirm that I see the message
printed sometimes.
## Analysis
The customer-provided callback, `processEvents`, gets called every time
a batch of events is received from the service. This batch is coming
from a single partition. Events are ordered within a partition by their
`sequenceNumber`, and events received by `processEvents` should be in
the same order. However currently, the list of events the
`processEvents` callback gets called on is not always in-order. Upon
further investigation, it was found that the library implements a
complex logic to read events from the service. It maintains two queues
for reading events, one for building a batch of events that will be sent
to the next call of the `processEvents` callback, and another for when
errors occur or there are no active listeners. The coordination to read
events from the two queues is subtle and is the source of the ordering
bug.
## Re-design
The most straightforward way to simplify this design and to ensure
ordering is to use a single queue and add incoming events to it in the
order they're received. Reading from this queue is as simple as the
following:
- If the queue contains any events, check if their count is already the
`maxMessageCount` or more:
  - If yes, remove `maxMessageCount` events and return them immediately
- If no, wait for a few milliseconds and then remove up to
`maxMessageCount` and return them
- If the queue doesn't contain any events, wait until the
`maxWaitTimeInSeconds` and then return an empty list, or until one or
more event arrive and then return those

### Abstraction

The idea is concisely captured by `waitForEvents`, a newly introduced
function that races a list of promises, one for each of the scenarios
listed above:
https://github.com/Azure/azure-sdk-for-js/blob/10826927554e7254dce0a4849f1e0c8219373522/sdk/eventhub/event-hubs/src/eventHubReceiver.ts#L733-L739

The first promise resolves right away and is returned if the queue
already has `maxMessageCount` events or more. It corresponds to the
first scenario listed above.

The second promise is created by the `checkOnInterval` function. The
promise is resolved only if the queue has any events in it. Otherwise,
it keeps checking every number of milliseconds. Note that chained to it
is a timer promise that waits another number of milliseconds to give the
service a chance to send more events. This corresponds to the second
scenario listed above.

The third promise is a simple timer promise that is resolved after the
`maxWaitTime` has elapsed. This promise corresponds to the third
scenario.

### Rewrite

In addition to some other minor improvements, the `receiveBatch` method
is concisely rewritten using that abstraction as follows:
https://github.com/Azure/azure-sdk-for-js/blob/10826927554e7254dce0a4849f1e0c8219373522/sdk/eventhub/event-hubs/src/eventHubReceiver.ts#L578-L628

Notice that the chain of promises makes the algorithm simple to read: a
link is established first, credits are added to it as needed, and then
the waiting starts.

Also, notice that at this point, no actual events were read from the
queue yet, all what this does is waiting until one of the promises
resolve. The actual reading from the queue is thened to that chain so
that it happens only after everything else is said and done. For
example, if an error occurred, it should be handled and we don't want to
prematurely mutate the queue. The reading from the queue is as simple as
the following:
https://github.com/Azure/azure-sdk-for-js/blob/10826927554e7254dce0a4849f1e0c8219373522/sdk/eventhub/event-hubs/src/eventHubReceiver.ts#L630

## Other changes

### Exporting `core-util`'s `createAbortablePromise`

This function was added in
Azure#24821 and proved to be
useful in this re-write so I am exporting it. I am planning on using it
in core-lro too.

### Updating tests

There are two tests updated, one for authentication and one for
returning events in the presence of retryable and non-retryable errors.

In the former, the receiver is expected to receive events after the auth
token has been invalidated but not yet refreshed. However, I am
observing that a disconnected event has been received at that moment and
the receiver has been deleted. The old receiver's behavior is to
continue receiving despite the deletion but the new one's behavior
correctly cleans up the receiver. I deleted this expectation for now.

In the latter, the test forces an error on the receiver after 50
milliseconds but the receiver already finishes around 40 milliseconds,
so I updated the forced error to happen sooner, at 10 milliseconds:
https://github.com/Azure/azure-sdk-for-js/blob/10826927554e7254dce0a4849f1e0c8219373522/sdk/eventhub/event-hubs/test/internal/receiveBatch.spec.ts#L107

Finally, a couple test suites were added for `waitForEvents` and
`checkOnInterval` functions.

## Updates in action

Live tests succeed
[[here](https://dev.azure.com/azure-sdk/internal/_build/results?buildId=2201768&view=results)].
Please ignore the timeout in the deployed resources script in canary, it
is an unrelated service issue, see
[[here](https://dev.azure.com/azure-sdk/internal/_build/results?buildId=2198994&view=results)].

A log for how the updated receiver behaves when used by the customer
sample can be found in
[log2.txt](https://github.com/Azure/azure-sdk-for-js/files/10775378/log2.txt).
Notice that the out of order message was never printed.

## Reviewing tips

The changes in `eventHubReceiver.ts` are too many and the diff is not
easily readable. I highly suggest to review
Azure@1082692
instead because it is on top of a deleting commit so there is no diff to
wrestle with. The main changes are in `receiveBatch` but please feel
free to review the rest of the module too.

Author: deyaaeldeen

sdk/core/core-util/CHANGELOG.md

@@ -1,9 +1,11 @@
 # Release History
 
-## 1.1.2 (Unreleased)
+## 1.2.0 (Unreleased)
 
 ### Features Added
 
+- Add `createAbortablePromise` which creates promises that can be aborted.
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/core/core-util/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@azure/core-util",
-  "version": "1.1.2",
+  "version": "1.2.0",
   "description": "Core library for shared utility methods",
   "sdk-type": "client",
   "main": "dist/index.js",

sdk/core/core-util/review/core-util.api.md

@@ -12,6 +12,16 @@ export function computeSha256Hash(content: string, encoding: "base64" | "hex"):
 // @public
 export function computeSha256Hmac(key: string, stringToSign: string, encoding: "base64" | "hex"): Promise<string>;
 
+// @public
+export function createAbortablePromise<T>(buildPromise: (resolve: (value: T | PromiseLike<T>) => void, reject: (reason?: any) => void) => void, options?: CreateAbortablePromiseOptions): Promise<T>;
+
+// @public
+export interface CreateAbortablePromiseOptions {
+    abortErrorMsg?: string;
+    abortSignal?: AbortSignalLike;
+    cleanupBeforeAbort?: () => void;
+}
+
 // @public
 export function delay(timeInMs: number, options?: DelayOptions): Promise<void>;
 

sdk/core/core-util/src/createAbortablePromise.ts

@@ -0,0 +1,63 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import { AbortError, AbortSignalLike } from "@azure/abort-controller";
+
+/**
+ * Options for the createAbortablePromise function.
+ */
+export interface CreateAbortablePromiseOptions {
+  /** A function to be called if the promise was aborted */
+  cleanupBeforeAbort?: () => void;
+  /** An abort signal */
+  abortSignal?: AbortSignalLike;
+  /** An abort error message */
+  abortErrorMsg?: string;
+}
+
+/**
+ * Creates an abortable promise.
+ * @param buildPromise - A function that takes the resolve and reject functions as parameters.
+ * @param options - The options for the abortable promise.
+ * @returns A promise that can be aborted.
+ */
+export function createAbortablePromise<T>(
+  buildPromise: (
+    resolve: (value: T | PromiseLike<T>) => void,
+    reject: (reason?: any) => void
+  ) => void,
+  options?: CreateAbortablePromiseOptions
+): Promise<T> {
+  const { cleanupBeforeAbort, abortSignal, abortErrorMsg } = options ?? {};
+  return new Promise((resolve, reject) => {
+    function rejectOnAbort(): void {
+      reject(new AbortError(abortErrorMsg ?? "The operation was aborted."));
+    }
+    function removeListeners(): void {
+      abortSignal?.removeEventListener("abort", onAbort);
+    }
+    function onAbort(): void {
+      cleanupBeforeAbort?.();
+      removeListeners();
+      rejectOnAbort();
+    }
+    if (abortSignal?.aborted) {
+      return rejectOnAbort();
+    }
+    try {
+      buildPromise(
+        (x) => {
+          removeListeners();
+          resolve(x);
+        },
+        (x) => {
+          removeListeners();
+          reject(x);
+        }
+      );
+    } catch (err) {
+      reject(err);
+    }
+    abortSignal?.addEventListener("abort", onAbort);
+  });
+}

sdk/core/core-util/src/delay.ts

@@ -1,7 +1,8 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
-import { AbortError, AbortSignalLike } from "@azure/abort-controller";
+import { AbortSignalLike } from "@azure/abort-controller";
+import { createAbortablePromise } from "./createAbortablePromise";
 
 const StandardAbortMessage = "The delay was aborted.";
 
@@ -19,58 +20,6 @@ export interface DelayOptions {
   abortErrorMsg?: string;
 }
 
-/**
- * Creates an abortable promise.
- * @param buildPromise - A function that takes the resolve and reject functions as parameters.
- * @param options - The options for the abortable promise.
- * @returns A promise that can be aborted.
- * @internal
- */
-export function createAbortablePromise<T>(
-  buildPromise: (
-    resolve: (value: T | PromiseLike<T>) => void,
-    reject: (reason?: any) => void
-  ) => void,
-  options?: {
-    cleanupBeforeAbort?: () => void;
-    abortSignal?: AbortSignalLike;
-    abortErrorMsg?: string;
-  }
-): Promise<T> {
-  const { cleanupBeforeAbort, abortSignal, abortErrorMsg } = options ?? {};
-  return new Promise((resolve, reject) => {
-    function rejectOnAbort(): void {
-      reject(new AbortError(abortErrorMsg ?? "The operation was aborted."));
-    }
-    function removeListeners(): void {
-      abortSignal?.removeEventListener("abort", onAbort);
-    }
-    function onAbort(): void {
-      cleanupBeforeAbort?.();
-      removeListeners();
-      rejectOnAbort();
-    }
-    if (abortSignal?.aborted) {
-      return rejectOnAbort();
-    }
-    try {
-      buildPromise(
-        (x) => {
-          removeListeners();
-          resolve(x);
-        },
-        (x) => {
-          removeListeners();
-          reject(x);
-        }
-      );
-    } catch (err) {
-      reject(err);
-    }
-    abortSignal?.addEventListener("abort", onAbort);
-  });
-}
-
 /**
  * A wrapper for setTimeout that resolves a promise after timeInMs milliseconds.
  * @param timeInMs - The number of milliseconds to be delayed.

sdk/core/core-util/src/index.ts

@@ -3,6 +3,7 @@
 
 export { isNode } from "./isNode";
 export { delay, DelayOptions } from "./delay";
+export { createAbortablePromise, CreateAbortablePromiseOptions } from "./createAbortablePromise";
 export { getRandomIntegerInclusive } from "./random";
 export { isObject, UnknownObject } from "./object";
 export { isError, getErrorMessage } from "./error";

…/internal/createAbortablePromise.spec.ts …st/public/createAbortablePromise.spec.tssdk/core/core-util/test/internal/createAbortablePromise.spec.ts renamed to sdk/core/core-util/test/public/createAbortablePromise.spec.ts sdk/core/core-util/test/internal/createAbortablePromise.spec.ts renamed to sdk/core/core-util/test/public/createAbortablePromise.spec.ts

@@ -5,7 +5,7 @@ import * as sinon from "sinon";
 import { AbortController, AbortSignalLike } from "@azure/abort-controller";
 import chai from "chai";
 import chaiAsPromised from "chai-as-promised";
-import { createAbortablePromise } from "../../src/delay";
+import { createAbortablePromise } from "../../src/createAbortablePromise";
 
 chai.use(chaiAsPromised);
 const { assert } = chai;

sdk/eventhub/event-hubs/CHANGELOG.md

@@ -1,15 +1,19 @@
 # Release History
 
-## 5.8.1 (Unreleased)
+## 5.9.0 (Unreleased)
 
 ### Features Added
 
 ### Breaking Changes
 
 ### Bugs Fixed
 
+- Fixing a bug where events were not always received in order [#23993](https://github.com/Azure/azure-sdk-for-js/issues/23993).
+
 ### Other Changes
 
+- The receiver no longer attempts to build batches of `maxMessageCount` size, instead, it returns batches as soon as they are received from the service, up to `maxMessageCount`.
+
 ## 5.8.0 (2022-05-10)
 
 ### Breaking Changes

sdk/eventhub/event-hubs/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@azure/event-hubs",
   "sdk-type": "client",
-  "version": "5.8.1",
+  "version": "5.9.0",
   "description": "Azure Event Hubs SDK for JS.",
   "author": "Microsoft Corporation",
   "license": "MIT",
@@ -49,7 +49,7 @@
     "build:node": "tsc -p . && dev-tool run bundle",
     "build:samples": "echo Obsolete.",
     "build:test:browser": "tsc -p . && rollup -c rollup.test.config.js 2>&1",
-    "build:test:node": "tsc -p . && dev-tool run bundle",
+    "build:test:node": "tsc -p .",
     "build:test": "tsc -p . && rollup -c rollup.test.config.js 2>&1 && npm run generate-certs && copyfiles -f ./test/internal/node/partitionKeyHashMap.json ./dist-esm/test/internal/node",
     "build:types": "downlevel-dts types/latest types/3.1",
     "build": "npm run clean && tsc -p . && dev-tool run bundle && api-extractor run --local && npm run build:types",
@@ -70,7 +70,7 @@
     "test:node": "npm run build:test && npm run unit-test:node && npm run integration-test:node",
     "test": "npm run build:test && npm run unit-test && npm run integration-test",
     "unit-test:browser": "echo skipped",
-    "unit-test:node": "cross-env NODE_EXTRA_CA_CERTS=\"./certs/my-private-root-ca.crt.pem\" TEST_TARGET=mock DISABLE_MULTI_VERSION_TESTING=true nyc mocha -r esm --require source-map-support/register --reporter ../../../common/tools/mocha-multi-reporter.js --timeout 1200000 --full-trace \"dist-esm/test/internal/*.spec.js\" \"dist-esm/test/public/*.spec.js\" \"dist-esm/test/public/**/*.spec.js\" \"dist-esm/test/internal/**/*.spec.js\"",
+    "unit-test:node": "cross-env NODE_EXTRA_CA_CERTS=\"./certs/my-private-root-ca.crt.pem\" TEST_TARGET=mock DISABLE_MULTI_VERSION_TESTING=true mocha -r esm -r ts-node/register --reporter ../../../common/tools/mocha-multi-reporter.js --timeout 1200000 --full-trace \"test/internal/*.spec.ts\" \"test/public/*.spec.ts\" \"test/public/**/*.spec.ts\" \"test/internal/**/*.spec.ts\"",
     "unit-test": "npm run unit-test:node && npm run unit-test:browser"
   },
   "//metadata": {