WIP: [service-bus] Add in abortSignal support in createBatch, subscribe and receiveMessages #8994

richardpark-msft · 2020-05-19T00:51:03Z

Adding in abortSignal support to:

subscribe
receiveMessages
createBatch

There were some refactors to make this more consistent (testing is still happening, so this is just a draft):

Made the openLink (equivalent of the old .open()/_init() calls in MessageSender and MessageReceiver) a common function with no LinkEntity dependency (easier for testing)
Added in intermediate aborts into that function
Made a few abort signal helpers (in utils.ts) to make it easier to register with an abort signal and properly clean up (fixing a bug I left in MessageSender where I didn't clean up my abort signal handlers).

ramya-rao-a · 2020-05-20T22:45:55Z

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

@@ -173,6 +186,12 @@ export class BatchingReceiver extends MessageReceiver {
        }
      });

+      cleanupAbortSignal = checkAndRegisterWithAbortSignal(
+        () => finalAction(),


Why are we calling finalAction() when the abort signal is fired instead of throwing the AbortError?

I was keeping with the sentiment that we've got in the connectionErrorHandler and returning the messages we've got so far, rather than just error out. In receiveAndDelete mode this is particularly important. Less so in peekLock.

TBH, it's inconsistent but it seems like it's favorable in this spot.

@bterlson - curious what your arch take on this is. Would it be better to be consistent and throw an AbortError here or would it be okay to treat abort as a soft cancel of the current operation when you have partial results?

That's an interesting point...I could see abort meaning "I don't care anymore, just get me out of here!" or "Just give me what you have now!". Since it could mean potentially losing messages if in receiveAndDelete mode it seems safer to return those messages if you've got them, but also curious to hear @bterlson 's opinion.

ramya-rao-a · 2020-05-20T22:48:16Z

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

+
+        if (cleanupAbortSignal) {
+          cleanupAbortSignal();
+          cleanupAbortSignal = undefined;


Having the clean up being called from finalAction() ensures clean up on the happy path. What about cases when the promise returned from receive() is rejected?

Great point, will push a fix, just waiting for the test run to complete.

I'm just going to chain two promises together, rather than trying to intercept each exit from this particular promise.

richardpark-msft · 2020-05-21T00:44:35Z

/azp run js - servicebus - tests

azure-pipelines · 2020-05-21T00:44:40Z

Pull request contains merge conflicts.

richardpark-msft · 2020-05-21T01:52:53Z

/azp run js - servicebus - tests

azure-pipelines · 2020-05-21T01:53:02Z

Azure Pipelines successfully started running 1 pipeline(s).

richardpark-msft · 2020-05-21T20:59:42Z

/azp run js - servicebus - tests

azure-pipelines · 2020-05-21T20:59:46Z

Pull request contains merge conflicts.

richardpark-msft · 2020-05-21T22:03:42Z

/azp run js - servicebus - tests

azure-pipelines · 2020-05-21T22:03:55Z

Azure Pipelines successfully started running 1 pipeline(s).

ramya-rao-a

Hey @richardpark-msft,
I have reviewed and left my first round of comments, please take a look.

If it is possible, then I'd much prefer the work being done here to be done in phases in two separate PRs. Phase 1 would involve the utilities that you added for timeout and abort signal with them being used from link creation code path, Phase 2 would involve introducing openLink to consolidate the opening of AMQP links across senders and receivers.

I havent looked at the streaming and batching receiver yet

ramya-rao-a · 2020-05-27T05:47:19Z

sdk/servicebus/service-bus/src/serviceBusClient.ts

@@ -15,6 +15,7 @@ import { CreateSessionReceiverOptions, CreateSenderOptions } from "./models";
 import { Receiver, ReceiverImpl } from "./receivers/receiver";
 import { SessionReceiver, SessionReceiverImpl } from "./receivers/sessionReceiver";
 import { ReceivedMessageWithLock, ReceivedMessage } from "./serviceBusMessage";
+import { openLink } from "./shared/openLink";


Do we need to import openLink here and pass to the constructors of ReceiverImpl and SenderImpl? These constructors can import and use it directly right?

They could - this is to make unit testing possible by always injecting openLink.

Is injecting the only way to ensure unit testability? My concern is that this model is not intuitive and adds complexity that passing around a function brings.

If openLink was meant to be a common shared code between MessageSender and MessageReceiver, then I would expect it to be part of LinkEntity which is the base class for both. An independent helper method makes sense when it is being used from multiple unrelated places & is stateless. Given that this helper is only ever going to be used by classes that inherit from LinkEntity, the base class is a much better fit

Seeing the unit tests, I mainly see the below categories

Ensure abort signal is plumbed through to open()

Ensure open() is not called when abort signal is already fired

If this common code was on LinkEntity, then for the above unit tests, we can override the method for open(). I see that the plumbing needs to be tested from SenderImpl -> MessageSender -> LinkEntity. Would perhaps combining SenderImpl & MessageSender help?

export class SenderImpl extends LinkEntity implements Sender ?

Then you test code would be

class TestAbortSignalSender extends SenderImpl { async openLink(..) { if (options.abortSignal != null) { throw new Error("abortSignal was properly passed to open()!"); } else { throw new Error("No abortSignal was passed to open()!"); } } }

instead of

sender = new SenderImpl(clientEntityContext, {}, async (openArgs) => { if (openArgs.abortSignal != null) { throw new Error("abortSignal was properly passed to open()!"); } else { throw new Error("No abortSignal was passed to open()!"); } });

cc @chradek

Definitely not the only way. There are so many ways!

Inheritance for testing is also a valid approach and I don't think it's bad. In the code as it is I think inheritance is making it difficult to see where the lines are between what's mutable, and what's not.

Also, this is definitely not this code's final form but it is a step in the direction that I think helps with making the code easier to maintain. Some state is ultimately required (so we might want to consider how to do that) though but I didn't want to do too much in one PR.

ramya-rao-a · 2020-05-27T05:47:51Z

sdk/servicebus/service-bus/src/shared/openLink.ts

+export function openLink<T extends RheaLink>(args: OpenArgs<T>): Promise<T | undefined> {
+  const checkAborted = (): void => {
+    if (args.abortSignal?.aborted) {
+      throw new AbortError("open() has been cancelled by user.");


The user would have no idea what open() is, so this error message would not make much sense to them

ramya-rao-a · 2020-05-27T05:50:42Z

sdk/servicebus/service-bus/src/shared/openLink.ts

+  checkAborted();
+
+  if (args.isOpen() || args.isConnecting() || args.getCloseInitiated()) {
+    return Promise.resolve(undefined);


If openLink is declared with the async keyword, then this can simply be return

Another benefit to making openLink async!

sdk/servicebus/service-bus/src/shared/openLink.ts

ramya-rao-a · 2020-05-27T06:16:01Z

sdk/servicebus/service-bus/src/shared/openLink.ts

+
+    try {
+      if (args.isOpen()) {
+        return Promise.resolve(undefined);


Since this callback passed to lock acquire is declared as async, this can simply be return

sdk/servicebus/service-bus/src/util/utils.ts

ramya-rao-a · 2020-05-27T07:17:13Z

sdk/servicebus/service-bus/src/shared/openLink.ts

+ * we can stop passing in some of these functions.
+ *
+ * For instance
+ * - negotiateClaim + ensureTokenRenewal can be lifted out of LinkEntity


out of LinkEntity and into where?

Originally I was thinking just moving the methods into into openLink.ts since they're all pretty related to that activity.

However, I've been thinking about that some more. My main issue with them being here is that inheritance can make it difficult to see the proper separation between LinkEntity and its child classes. So when I was tracing through this it wasn't entirely obvious which parts were "internal" state to LinkEntity, etc...

Having been through it more I could see it staying but it'd be nice to make that first-time read a bit easier.

ramya-rao-a · 2020-05-27T07:18:45Z

sdk/servicebus/service-bus/src/shared/openLink.ts

+  // these are more candidates to move out from LinkEntity. They don't really depend on anything special
+  // in LinkEntity and are just clutter.
+  negotiateClaim(): Promise<void>;
+  ensureTokenRenewal(): Promise<void>;


After looking into _ensureTokenRenewal in LinkEntity, its not clear to me why this would be an async function. All it does is set a timer to negotiate claim

ramya-rao-a · 2020-05-27T07:24:55Z

sdk/servicebus/service-bus/src/shared/openLink.ts

+  // TODO: do we _need_ this lock at the moment? If we're not trying to get
+  // open and close to be mutually exclusive?
+  return defaultLock.acquire(args.openLock, async () => {
+    checkAborted();


There are a lot of calls to checkAborted() being done in here after acquiring the lock. I understand the need to use it so that we can exit as soon as possible instead of going through the different async operations, but we don't need it to throw the AbortError itself.

We can use checkAndRegisterWithAbortSignal() before calling openLink(). This way, in here, we only need to add early exits before every async operation. And then when these async operations can support abort signal in the future, these checks can go away.

- Removing unnecessary clearTimeout

- Adding in some commentary - Adding in a missing test - when the action throws an error.

richardpark-msft · 2020-05-27T21:51:04Z

Hey @richardpark-msft,
I have reviewed and left my first round of comments, please take a look.

If it is possible, then I'd much prefer the work being done here to be done in phases in two separate PRs. Phase 1 would involve the utilities that you added for timeout and abort signal with them being used from link creation code path, Phase 2 would involve introducing openLink to consolidate the opening of AMQP links across senders and receivers.

I havent looked at the streaming and batching receiver yet

Created PR#9138 for the abort helpers.

- Fixing some broken test

…e promise. - Also, make sure that finalActionHandler doesn't get called more than once.

… can run the same tests in the browser.

…ecific logger.

- Fixing up the logging so we still dump the options for created senders and receivers.

…e care of us)

…sb-track2-abortsignal-v2

… compatible.

…sb-track2-abortsignal-v2

ramya-rao-a · 2020-05-28T19:33:09Z

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

@@ -184,7 +184,7 @@ export class LinkEntity {
   * Ensures that the token is renewed within the predefined renewal margin.
   * @returns {void}
   */
-  protected async _ensureTokenRenewal(): Promise<void> {
+  protected _ensureTokenRenewal(): void {


There are few places from where this is called. All those places need to be updated to remove the await

ramya-rao-a · 2020-05-28T19:50:38Z

Resolved all my comments on the utils.ts file as they have either been addressed in #9138 or I have copied the conversation over to that PR

…anyways.

…sb-track2-abortsignal-v2

chradek · 2020-05-29T21:29:51Z

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

          try {
-            await Promise.race([this.open(), initTimeoutPromise]);
+            await waitForTimeoutOrAbortOrResolve({
+              actionFn: () => this.open({ abortSignal: options?.abortSignal }),


Nit: You already got the abortSignal so can do this:

Suggested change

actionFn: () => this.open({ abortSignal: options?.abortSignal }),

actionFn: () => this.open({ abortSignal }),

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

chradek · 2020-05-29T21:38:08Z

sdk/servicebus/service-bus/src/shared/openLink.ts

+import { AbortSignalLike } from "@azure/core-http";
+import { AbortError } from "@azure/abort-controller";
+
+export type RheaLink = AwaitableSender | RheaReceiver;


Nit: My obligatory "Add @internal and @ignores to anything exported that shouldn't be public!"

chradek · 2020-05-29T22:24:28Z

sdk/servicebus/service-bus/src/shared/openLink.ts

+export function openLink<T extends RheaLink>(args: OpenArgs<T>): Promise<T | undefined> {
+  const checkAborted = (): void => {
+    if (args.abortSignal?.aborted) {
+      throw new AbortError("open() has been cancelled by user.");


We just happened to talk about this in another PR! openLink is not currently an async method, so I'd expect that if this function gets called outside of the promise that openLink returns, it would return a Promise.reject(new AbortError("")).

This gets a bit complicated though since you re-use this function inside the promise that's returned, and I don't think (not certain) that returning a reject promise would trigger the catch block that's used to close a link.

Ultimately it might just be easier to make openLink an async function so you don't have to worry about this.

chradek · 2020-05-29T22:25:02Z

sdk/servicebus/service-bus/src/shared/openLink.ts

+  checkAborted();
+
+  if (args.isOpen() || args.isConnecting() || args.getCloseInitiated()) {
+    return Promise.resolve(undefined);


Another benefit to making openLink async!

chradek · 2020-05-29T23:04:30Z

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

    throwErrorIfConnectionClosed(this._context.namespace);
    this._onMessage = onMessage;
    this._onError = onError;

+    this._cleanupAbortHandler = checkAndRegisterWithAbortSignal(


We don't allow calling receive multiple times do we? Is there any risk that an existing _cleanupAbortHandler might get overwritten with a new one? (Should we be pushing the cleanup methods to a list and iterate through them when we call close?)

chradek · 2020-05-29T23:20:46Z

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

@@ -173,6 +186,12 @@ export class BatchingReceiver extends MessageReceiver {
        }
      });

+      cleanupAbortSignal = checkAndRegisterWithAbortSignal(
+        () => finalAction(),


That's an interesting point...I could see abort meaning "I don't care anymore, just get me out of here!" or "Just give me what you have now!". Since it could mean potentially losing messages if in receiveAndDelete mode it seems safer to return those messages if you've got them, but also curious to hear @bterlson 's opinion.

chradek · 2020-05-29T23:22:03Z

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

      // Final action to be performed after
      // - maxMessageCount is reached or
      // - maxWaitTime is passed or
      // - newMessageWaitTimeoutInSeconds is passed since the last message was received
      const finalAction = (): void => {
+        if (finalActionCalled) {


I feel like we used to have something like this before...

chradek · 2020-05-29T23:34:43Z

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

+      isConnecting: () => this.isConnecting,
+      setIsConnecting: (value: boolean) => (this.isConnecting = value),
+      getCloseInitiated: () => this.wasCloseInitiated,
+      create: async () => {


Slight readability improvement:

Suggested change

create: async () => {

create: async createReceiver() => {

chradek · 2020-05-29T23:43:02Z

sdk/servicebus/service-bus/test/abortSignal.spec.ts

+          throw new Error("abortSignal was properly passed to open()!");
+        } else {
+          throw new Error("No abortSignal was passed to open()!");


Nit: Might not be worth doing now, but it might be slightly easier to maintain/read if these messages were in variables like abortSignalPresent and abortSignalMissing. Then it'd be easy to tell in the asserts what's supposed to happen without having to parse the text (because my brain now parses variable names better than english words)

richardpark-msft · 2020-06-11T22:23:21Z

Closing this PR - the work has migrated into two separate PRs:
#9233 - sender
#9284 - receiver

richardpark-msft requested review from HarshaNalluru, ramya-rao-a and chradek May 19, 2020 00:51

ramya-rao-a reviewed May 20, 2020

View reviewed changes

richardpark-msft force-pushed the richardpark-sb-track2-abortsignal-v2 branch from 0b17a16 to 0869c5a Compare May 21, 2020 01:52

richardpark-msft force-pushed the richardpark-sb-track2-abortsignal-v2 branch from cc6bab1 to 19ed1f1 Compare May 21, 2020 22:00

ramya-rao-a suggested changes May 27, 2020

View reviewed changes

richardpark-msft added 3 commits May 27, 2020 11:42

Splitting out some abortSignal helpers

032c4b7

- Some renames

6fcf8f2

- Removing unnecessary clearTimeout

- Some more renames

620dba2

- Adding in some commentary - Adding in a missing test - when the action throws an error.

richardpark-msft mentioned this pull request May 27, 2020

[service-bus] AbortSignal helpers #9138

Merged

richardpark-msft added 8 commits May 27, 2020 14:57

Add in abortSignal support to createBatch, receiveBatch and subscribe().

d261f9f

- Add in early exit for openLink

12516a1

- Fixing some broken test

Properly clean up the abort signal regardless of the resolution of th…

8fe802b

…e promise. - Also, make sure that finalActionHandler doesn't get called more than once.

I've added a test that uses an EventEmitter. Adding in "events" so we…

7ac95d7

… can run the same tests in the browser.

Just use log.error() for now and remove the need to pass an entity sp…

dfc5ded

…ecific logger.

Update comment about future code movement.

c3eac87

Update for a build error

a3b54ac

ensureTokenRenewal didn't need to be async (or return a Promise)

1905b3a

richardpark-msft force-pushed the richardpark-sb-track2-abortsignal-v2 branch from 6d68bce to 1905b3a Compare May 28, 2020 00:10

richardpark-msft added 14 commits May 27, 2020 20:38

- Removing TODO comments have been addressed or are no longer valid.

252621b

- Fixing up the logging so we still dump the options for created senders and receivers.

abortSignal can't be null since we did a check above.

05e8ef8

Removing an unneeded clearing of the abortSignal (the .catch will tak…

e4d2c4b

…e care of us)

Remove unneeded ? check.

f07dce4

Remove .only

7dd35a9

Merge branch 'richardpark-sb-track2-a-aborthelpers' into richardpark-…

50982cf

…sb-track2-abortsignal-v2

Change over to return an empty function instead of undefined.

df16247

Merge branch 'richardpark-sb-track2-a-aborthelpers' into richardpark-…

a9e8bdd

…sb-track2-abortsignal-v2

Rename for idiomacy.

f9d9a28

Small rename (and to or)

86c63cb

Adding in abortError so a user can resolve() with it if they choose to.

1509c05

Merge branch 'richardpark-sb-track2-a-aborthelpers' into richardpark-…

988f4b6

…sb-track2-abortsignal-v2

Another nice optimization suggested by Ramya. Make ourselves reject…

3901fcd

… compatible.

Merge branch 'richardpark-sb-track2-a-aborthelpers' into richardpark-…

9785864

…sb-track2-abortsignal-v2

ramya-rao-a reviewed May 28, 2020

View reviewed changes

richardpark-msft added 7 commits May 28, 2020 21:03

Make the abort message non-optional

1f6c689

Just use AbortError for timeout - apparently we do this in core-http …

c917759

…anyways.

Swapping required and non-required args.

2284cd6

Change over to async and try/finally

62b234b

Use OperationTimeoutError

4871157

Update doc string

d1a246b

Merge branch 'richardpark-sb-track2-a-aborthelpers' into richardpark-…

c785c58

…sb-track2-abortsignal-v2

chradek reviewed May 30, 2020

View reviewed changes

richardpark-msft closed this Jun 11, 2020

	actionFn: () => this.open({ abortSignal: options?.abortSignal }),
	actionFn: () => this.open({ abortSignal }),

WIP: [service-bus] Add in abortSignal support in createBatch, subscribe and receiveMessages #8994

WIP: [service-bus] Add in abortSignal support in createBatch, subscribe and receiveMessages #8994

Conversation

richardpark-msft commented May 19, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

richardpark-msft commented May 21, 2020

azure-pipelines bot commented May 21, 2020

richardpark-msft commented May 21, 2020

azure-pipelines bot commented May 21, 2020

richardpark-msft commented May 21, 2020

azure-pipelines bot commented May 21, 2020

richardpark-msft commented May 21, 2020

azure-pipelines bot commented May 21, 2020

ramya-rao-a left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

richardpark-msft commented May 27, 2020

Choose a reason for hiding this comment

ramya-rao-a commented May 28, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

richardpark-msft commented Jun 11, 2020

ramya-rao-a left a comment •

edited

Loading