[service-bus] Closing some open areas where we could mask errors, and fixing flaky tests to be more reliable. (#15761)

Making changes to simplify a flaky test (and hopefully make it more reliable).

The main issue with the 'handle interrupted detach' method was that it relied on too many moving parts to work reliably. We could just eternally loop like we'd expect customers to do, but in the end we have a very simple test we're trying to perform - we want to receive, and while we're in the process of draining, cause a detach and have it early exit and reject/resolve immediately rather than waiting for the timeout.

I reworked the test to make that simpler by just removing the unneeded connection.idle() and just calling directly into the onDetached method. Because it happens prior to rhea even seeing that we're draining we should reliably win that race each time. 

There were a couple of other things changed for this PR as well:
- The max time per test was lowered accidentally. Bringing it back what's been used as the standard time in other libraries
- Fixed a spot in receiveMessages() where, if the link had been closed, we'd falsly return an empty array instead of throwing an exception indicating the link closed. This didn't appear to be related to the bug but it's incorrect and can hide bugs so I'd rather just throw the error than eat the condition and return an empty array. It's rare, but when it does happen the empty array response isn't right either - we're probably in the middle of a connection reset/change event.

Fixes #13461

Author: richardpark-msft

sdk/servicebus/service-bus/package.json

@@ -60,7 +60,7 @@
     "extract-api": "tsc -p . && api-extractor run --local",
     "format": "prettier --write --config ../../../.prettierrc.json --ignore-path ../../../.prettierignore \"samples/**/*.{ts,js}\" \"src/**/*.ts\" \"test/**/*.ts\" \"samples-dev/**/*.ts\" \"*.{js,json}\"",
     "integration-test:browser": "karma start --single-run",
-    "integration-test:node": "nyc mocha -r esm --require source-map-support/register --reporter ../../../common/tools/mocha-multi-reporter.js --timeout 120000 --full-trace \"dist-esm/test/internal/**/*.spec.js\" \"dist-esm/test/public/**/*.spec.js\"",
+    "integration-test:node": "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\"",
     "integration-test": "npm run integration-test:node && npm run integration-test:browser",
     "lint:fix": "eslint package.json api-extractor.json src test --ext .ts --fix --fix-type [problem,suggestion]",
     "lint": "eslint package.json api-extractor.json src test --ext .ts -f html -o service-bus-lintReport.html || exit 0",
@@ -71,7 +71,7 @@
     "test:node": "npm run clean && npm run build:test:node && npm run integration-test:node",
     "test": "npm run test:node && npm run test:browser",
     "unit-test:browser": "echo skipped",
-    "unit-test:node": "mocha -r esm -r ts-node/register --require source-map-support/register --reporter ../../../common/tools/mocha-multi-reporter.js --timeout 120000 --full-trace \"test/internal/unit/*.spec.ts\" \"test/internal/node/*.spec.ts\"",
+    "unit-test:node": "mocha -r esm -r ts-node/register --require source-map-support/register --reporter ../../../common/tools/mocha-multi-reporter.js --timeout 1200000 --full-trace \"test/internal/unit/*.spec.ts\" \"test/internal/node/*.spec.ts\"",
     "unit-test": "npm run unit-test:node && npm run unit-test:browser",
     "docs": "typedoc --excludePrivate --excludeNotExported --excludeExternals --stripInternal --mode file --out ./dist/docs ./src"
   },

sdk/servicebus/service-bus/src/core/batchingReceiver.ts

@@ -281,7 +281,7 @@ export class BatchingReceiverLite {
 
       if (receiver == null) {
         // (was somehow closed in between the init() and the return)
-        return [];
+        throw new ServiceBusError("Link closed before receiving messages.", "GeneralError");
       }
 
       const messages = await new Promise<ServiceBusMessageImpl[]>((resolve, reject) =>

sdk/servicebus/service-bus/test/internal/backupMessageSettlement.spec.ts

@@ -19,6 +19,7 @@ import {
   ServiceBusMessageImpl,
   ServiceBusReceivedMessage
 } from "../../src/serviceBusMessage";
+import { disableCommonLoggers, enableCommonLoggers, testLogger } from "./utils/misc";
 
 const should = chai.should();
 chai.use(chaiAsPromised);
@@ -256,10 +257,17 @@ describe("Message settlement After Receiver is Closed - Through ManagementLink",
     const testMessages = entityNames.usesSessions
       ? TestMessage.getSessionSample()
       : TestMessage.getSample();
+
+    testLogger.info(`sending (and receiving) initial message`);
+
     const msg = await sendReceiveMsg(testMessages);
 
+    testLogger.info(`Done sending initial messages`);
+
     const msgDeliveryLink = (msg as ServiceBusMessageImpl).delivery.link.name;
 
+    testLogger.info(`About to close the underlying link.`);
+
     if (entityNames.usesSessions) {
       await (receiver as ServiceBusReceiverImpl)["_context"].messageSessions[
         msgDeliveryLink
@@ -270,10 +278,18 @@ describe("Message settlement After Receiver is Closed - Through ManagementLink",
       ].close();
     }
 
+    testLogger.info(
+      `Underlying link should be closed: ${receiver.isClosed}. This will force us to use the management link to settle. Will now attempt to dead letter.`
+    );
+
     let errorWasThrown = false;
     try {
       await receiver.deadLetterMessage(msg);
+
+      testLogger.info(`Message has been dead lettered`);
     } catch (err) {
+      testLogger.error(`Exception thrown`, err);
+
       should.equal(
         err.message,
         `Failed to ${DispositionType.deadletter} the message as the AMQP link with which the message was received is no longer alive.`,
@@ -288,6 +304,9 @@ describe("Message settlement After Receiver is Closed - Through ManagementLink",
       should.equal(errorWasThrown, false, "Error was thrown for sessions without session-id");
     }
 
+    testLogger.info(
+      `Creating a peek lock dead letter receiver and attempting to receive the dead lettered message`
+    );
     receiver = await serviceBusClient.test.createPeekLockReceiver(entityNames);
 
     if (!entityNames.usesSessions) {
@@ -310,22 +329,30 @@ describe("Message settlement After Receiver is Closed - Through ManagementLink",
         "MessageId is different than expected"
       );
 
+      testLogger.info(`Attempting to complete the message: ${deadLetterMsgsBatch[0].messageId}`);
       await receiver.completeMessage(deadLetterMsgsBatch[0]);
-
       await testPeekMsgsLength(deadLetterReceiver, 0);
     } else {
       const messageBatch = await receiver.receiveMessages(1);
-      await receiver.completeMessage(messageBatch[0]);
 
+      testLogger.info(`Attempting to complete the message: ${messageBatch[0].messageId}`);
+      await receiver.completeMessage(messageBatch[0]);
       await testPeekMsgsLength(receiver, 0);
     }
+
+    testLogger.info(`Done testing dead letter`);
   }
 
   it(
     noSessionTestClientType + ": deadLetter() moves message to deadletter queue",
     async function(): Promise<void> {
-      await beforeEachTest(noSessionTestClientType);
-      await testDeadletter();
+      enableCommonLoggers();
+      try {
+        await beforeEachTest(noSessionTestClientType);
+        await testDeadletter();
+      } finally {
+        disableCommonLoggers();
+      }
     }
   );