From 786dec4fa3b226574357d84752b4e6aac4ad4264 Mon Sep 17 00:00:00 2001 From: Jeremy Meng Date: Thu, 10 Feb 2022 01:17:24 +0000 Subject: [PATCH 1/5] [ServiceBus] add delay between infinite retry cycles Errors that are not retryable will be re-thrown out of the normal retry() call. However, we don't have any delay before restarting the retry cycle. This PR adds a delay before continuing the infinite retry cycles. --- .../src/receivers/receiverCommon.ts | 55 +++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts b/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts index 1bf4ed443d1f..2ae857fe318c 100644 --- a/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts +++ b/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts @@ -15,9 +15,12 @@ import { import { DispositionStatusOptions } from "../core/managementClient"; import { ConnectionContext } from "../connectionContext"; import { + Constants, ErrorNameConditionMapper, + delay, retry, RetryConfig, + RetryMode, RetryOperationType, RetryOptions, } from "@azure/core-amqp"; @@ -279,6 +282,28 @@ export interface RetryForeverArgs { logPrefix: string; } +/** + * Calculates delay between retries, in milliseconds. + * @internal + */ +function calculateDelay( + attemptCount: number, + retryDelayInMs: number, + maxRetryDelayInMs: number, + mode: RetryMode +): number { + if (mode === RetryMode.Exponential) { + const boundedRandDelta = + retryDelayInMs * 0.8 + + Math.floor(Math.random() * (retryDelayInMs * 1.2 - retryDelayInMs * 0.8)); + + const incrementDelta = boundedRandDelta * (Math.pow(2, attemptCount) - 1); + return Math.min(incrementDelta, maxRetryDelayInMs); + } + + return retryDelayInMs; +} + /** * Retry infinitely until success, reporting in between retry attempts. * @@ -293,6 +318,22 @@ export async function retryForever( retryFn: typeof retry = retry ): Promise { let numRetryCycles = 0; + const config = args.retryConfig; + if (!config.retryOptions) { + config.retryOptions = {}; + } + if (config.retryOptions.retryDelayInMs == undefined || config.retryOptions.retryDelayInMs < 0) { + config.retryOptions.retryDelayInMs = Constants.defaultDelayBetweenOperationRetriesInMs; + } + if ( + config.retryOptions.maxRetryDelayInMs == undefined || + config.retryOptions.maxRetryDelayInMs < 0 + ) { + config.retryOptions.maxRetryDelayInMs = Constants.defaultMaxDelayForExponentialRetryInMs; + } + if (config.retryOptions.mode == undefined) { + config.retryOptions.mode = RetryMode.Fixed; + } // The retries are broken up into cycles, giving the user some control over how often // we actually attempt to retry. @@ -325,6 +366,20 @@ export async function retryForever( args.retryConfig ); + const delayInMs = calculateDelay( + numRetryCycles, + config.retryOptions.retryDelayInMs, + config.retryOptions.maxRetryDelayInMs, + config.retryOptions.mode + ); + logger.verbose( + "[%s] Sleeping for %d milliseconds for '%s'.", + config.connectionId, + delayInMs, + config.operationType + ); + await delay(delayInMs, config.abortSignal, "Retry cycle has been cancelled by the user."); + continue; } } From a044202918d8928a23c407e3c59f58a7265c9409 Mon Sep 17 00:00:00 2001 From: Jeremy Meng Date: Thu, 10 Feb 2022 13:42:05 -0800 Subject: [PATCH 2/5] Add a test and update changelog also fixed the issue where we don't use `maxRetryDelayInMs` in Fixed retry mode. --- sdk/core/core-amqp/src/retry.ts | 2 +- sdk/servicebus/service-bus/CHANGELOG.md | 3 +- .../src/receivers/receiverCommon.ts | 17 ++--- .../test/internal/unit/receiverCommon.spec.ts | 65 +++++++++++++++++++ 4 files changed, 77 insertions(+), 10 deletions(-) diff --git a/sdk/core/core-amqp/src/retry.ts b/sdk/core/core-amqp/src/retry.ts index 2e0a2cf6f8e0..3f61b3753310 100644 --- a/sdk/core/core-amqp/src/retry.ts +++ b/sdk/core/core-amqp/src/retry.ts @@ -153,7 +153,7 @@ function calculateDelay( return Math.min(incrementDelta, maxRetryDelayInMs); } - return retryDelayInMs; + return Math.min(retryDelayInMs, maxRetryDelayInMs); } /** diff --git a/sdk/servicebus/service-bus/CHANGELOG.md b/sdk/servicebus/service-bus/CHANGELOG.md index dba057209556..ff8a24ed57f7 100644 --- a/sdk/servicebus/service-bus/CHANGELOG.md +++ b/sdk/servicebus/service-bus/CHANGELOG.md @@ -1,6 +1,6 @@ # Release History -## 7.5.0 (2022-02-08) +## 7.5.0 (2022-02-14) ### Features Added @@ -10,6 +10,7 @@ ### Bugs Fixed - The `processError` callback to `subscribe()` was previously called only for errors on setting up the receiver, errors on message settlement or message lock renewal and not for errors on AMQP link or session. This is now fixed. [PR #19189](https://github.com/Azure/azure-sdk-for-js/pull/19189) +- Fix an issue where we don't respect retry options before starting the next retry cycle [PR #20316](https://github.com/Azure/azure-sdk-for-js/pull/20316) ## 7.4.0 (2021-11-08) diff --git a/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts b/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts index 2ae857fe318c..7ed490701830 100644 --- a/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts +++ b/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts @@ -301,7 +301,7 @@ function calculateDelay( return Math.min(incrementDelta, maxRetryDelayInMs); } - return retryDelayInMs; + return Math.min(retryDelayInMs, maxRetryDelayInMs); } /** @@ -322,16 +322,13 @@ export async function retryForever( if (!config.retryOptions) { config.retryOptions = {}; } - if (config.retryOptions.retryDelayInMs == undefined || config.retryOptions.retryDelayInMs < 0) { + if (!config.retryOptions.retryDelayInMs || config.retryOptions.retryDelayInMs < 0) { config.retryOptions.retryDelayInMs = Constants.defaultDelayBetweenOperationRetriesInMs; } - if ( - config.retryOptions.maxRetryDelayInMs == undefined || - config.retryOptions.maxRetryDelayInMs < 0 - ) { + if (!config.retryOptions.maxRetryDelayInMs || config.retryOptions.maxRetryDelayInMs < 0) { config.retryOptions.maxRetryDelayInMs = Constants.defaultMaxDelayForExponentialRetryInMs; } - if (config.retryOptions.mode == undefined) { + if (!config.retryOptions.mode) { config.retryOptions.mode = RetryMode.Fixed; } @@ -378,7 +375,11 @@ export async function retryForever( delayInMs, config.operationType ); - await delay(delayInMs, config.abortSignal, "Retry cycle has been cancelled by the user."); + await delay( + delayInMs, + config.abortSignal, + "Retry cycle has been cancelled by the user." + ); continue; } diff --git a/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts b/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts index 20cc4f95ddb0..75942f3c04d4 100644 --- a/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts +++ b/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts @@ -278,6 +278,9 @@ describe("shared receiver code", () => { }, connectionId: "id", operationType: RetryOperationType.connection, + retryOptions: { + maxRetryDelayInMs: 2000, + }, }, }, fakeRetry @@ -290,6 +293,68 @@ describe("shared receiver code", () => { assert.equal(numRetryCalls, 2 + 1); }); + + it("respects retry options", async () => { + const errorMessages: string[] = []; + let numRetryCalls = 0; + + const fakeRetry = async (): Promise => { + ++numRetryCalls; + + if (numRetryCalls < 4) { + // force retry<> to get called three times (3x because + // we "failed" and threw exceptions and 1 more time where + // we succeed. + throw new Error(`Attempt ${numRetryCalls}: Force another call of retry<>`); + } + + return Promise.resolve({} as T); + }; + + const retryDelayInMs = 2000; + let previousAttemptTime = Date.now(); + await retryForever( + { + logPrefix: "logPrefix", + logger: logger, + onError: (err) => { + errorMessages.push(err.message); + if (numRetryCalls > 1) { + // not the first attempt + const currentTime = Date.now(); + if (currentTime - previousAttemptTime < retryDelayInMs) { + // wait at least 100ms between attempts + errorMessages.push( + `Unexpected, Should've waited at least ${retryDelayInMs} between attempts` + ); + } + previousAttemptTime = currentTime; + } + }, + retryConfig: { + operation: async () => { + ++numRetryCalls; + + return 1; + }, + connectionId: "id", + operationType: RetryOperationType.connection, + retryOptions: { + retryDelayInMs, + }, + }, + }, + fakeRetry + ); + + assert.deepEqual(errorMessages, [ + "Attempt 1: Force another call of retry<>", + "Attempt 2: Force another call of retry<>", + "Attempt 3: Force another call of retry<>", + ]); + + assert.equal(numRetryCalls, 3 + 1); + }); }); }); From 57d6df2628b1d65bf43594b709553b1b35e91f79 Mon Sep 17 00:00:00 2001 From: Jeremy Meng Date: Thu, 10 Feb 2022 22:40:39 +0000 Subject: [PATCH 3/5] Address CR feedback --- .../service-bus/src/receivers/receiverCommon.ts | 1 - .../service-bus/test/internal/unit/receiverCommon.spec.ts | 8 ++++---- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts b/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts index 7ed490701830..88f94d975fa3 100644 --- a/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts +++ b/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts @@ -284,7 +284,6 @@ export interface RetryForeverArgs { /** * Calculates delay between retries, in milliseconds. - * @internal */ function calculateDelay( attemptCount: number, diff --git a/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts b/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts index 75942f3c04d4..6480b615a387 100644 --- a/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts +++ b/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts @@ -296,13 +296,14 @@ describe("shared receiver code", () => { it("respects retry options", async () => { const errorMessages: string[] = []; + const errorCount = 3; let numRetryCalls = 0; const fakeRetry = async (): Promise => { ++numRetryCalls; - if (numRetryCalls < 4) { - // force retry<> to get called three times (3x because + if (numRetryCalls < errorCount + 1) { + // force retry<> to get called ${errorCount} times (because // we "failed" and threw exceptions and 1 more time where // we succeed. throw new Error(`Attempt ${numRetryCalls}: Force another call of retry<>`); @@ -323,7 +324,6 @@ describe("shared receiver code", () => { // not the first attempt const currentTime = Date.now(); if (currentTime - previousAttemptTime < retryDelayInMs) { - // wait at least 100ms between attempts errorMessages.push( `Unexpected, Should've waited at least ${retryDelayInMs} between attempts` ); @@ -353,7 +353,7 @@ describe("shared receiver code", () => { "Attempt 3: Force another call of retry<>", ]); - assert.equal(numRetryCalls, 3 + 1); + assert.equal(numRetryCalls, errorCount + 1); }); }); }); From 55e4c0df9f56454c49f4bf8c1e05f034f155931c Mon Sep 17 00:00:00 2001 From: Jeremy Meng Date: Fri, 11 Feb 2022 11:03:33 -0800 Subject: [PATCH 4/5] Update sdk/servicebus/service-bus/CHANGELOG.md Co-authored-by: Ramya Rao --- sdk/servicebus/service-bus/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/servicebus/service-bus/CHANGELOG.md b/sdk/servicebus/service-bus/CHANGELOG.md index ff8a24ed57f7..9b3118323964 100644 --- a/sdk/servicebus/service-bus/CHANGELOG.md +++ b/sdk/servicebus/service-bus/CHANGELOG.md @@ -10,7 +10,7 @@ ### Bugs Fixed - The `processError` callback to `subscribe()` was previously called only for errors on setting up the receiver, errors on message settlement or message lock renewal and not for errors on AMQP link or session. This is now fixed. [PR #19189](https://github.com/Azure/azure-sdk-for-js/pull/19189) -- Fix an issue where we don't respect retry options before starting the next retry cycle [PR #20316](https://github.com/Azure/azure-sdk-for-js/pull/20316) +- Fix an issue where we don't respect retry options before starting the next retry cycle when using the `subscribe()` method. [PR #20316](https://github.com/Azure/azure-sdk-for-js/pull/20316) ## 7.4.0 (2021-11-08) From e48a827fb0fdac3343b0caa99f097fd02c347bf1 Mon Sep 17 00:00:00 2001 From: Jeremy Meng Date: Fri, 11 Feb 2022 17:48:13 -0800 Subject: [PATCH 5/5] Revert max delay limit in Fixed retry mode as the setting only applies to Exponential retry mode. --- sdk/core/core-amqp/src/retry.ts | 2 +- sdk/servicebus/service-bus/src/receivers/receiverCommon.ts | 2 +- .../service-bus/test/internal/unit/receiverCommon.spec.ts | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/sdk/core/core-amqp/src/retry.ts b/sdk/core/core-amqp/src/retry.ts index 3f61b3753310..2e0a2cf6f8e0 100644 --- a/sdk/core/core-amqp/src/retry.ts +++ b/sdk/core/core-amqp/src/retry.ts @@ -153,7 +153,7 @@ function calculateDelay( return Math.min(incrementDelta, maxRetryDelayInMs); } - return Math.min(retryDelayInMs, maxRetryDelayInMs); + return retryDelayInMs; } /** diff --git a/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts b/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts index 88f94d975fa3..7514a6e41f93 100644 --- a/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts +++ b/sdk/servicebus/service-bus/src/receivers/receiverCommon.ts @@ -300,7 +300,7 @@ function calculateDelay( return Math.min(incrementDelta, maxRetryDelayInMs); } - return Math.min(retryDelayInMs, maxRetryDelayInMs); + return retryDelayInMs; } /** diff --git a/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts b/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts index 6480b615a387..cb1dabef0dff 100644 --- a/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts +++ b/sdk/servicebus/service-bus/test/internal/unit/receiverCommon.spec.ts @@ -279,7 +279,7 @@ describe("shared receiver code", () => { connectionId: "id", operationType: RetryOperationType.connection, retryOptions: { - maxRetryDelayInMs: 2000, + retryDelayInMs: 2000, }, }, },