Add support for scheduled retry with retryAfter

README.md

@@ -196,6 +196,33 @@ const catWatch = new ServiceClient({
 });
 ```
 
+#### Scheduled retries
+
+Timeouts are often the reason for retries. You might want to retry your request *without* waiting for the initial request to timeout, giving more time to the service you are calling to reply to initial request but still keeping the same overal requests timeout.
+
+For this purpose you can use `retryAfter` in `retryOptions` in combination with `dropAllRequestsAfter` in options root. If you have retries configured and initial request has not resolved before `retryAfter`, another request will be sent and whichever requests will resolve first will be treated, aborting the other one(s).
+
+`retryAfter` logic is immediate and will not take into account minTimeout or maxTimeout.
+
+```js
+const {ServiceClient} = require('perron');
+
+const catWatch = new ServiceClient({
+    hostname: 'catwatch.opensource.zalan.do',
+    dropAllRequestsAfter: 1000
+    retryOptions: {
+        retries: 1,
+        retryAfter: 400,
+        shouldRetry(err, req) {
+            return (err && err.response && err.response.statusCode >= 500);
+        },
+        onRetry(currentAttempt, err, req) {
+            console.log('Retry attempt #' + currentAttempt + ' for ' + req.path + ' due to ' + err);
+        }
+    }
+});
+```
+
 ## Filters
 
 It's quite often necessary to do some pre- or post-processing of the request. For this purpose `perron` implements a concept of filters, that are just an object with 2 optional methods: `request` and `response`.

lib/client.ts

@@ -5,6 +5,7 @@ import {
   CircuitBreakerPublicApi
 } from "./circuit-breaker";
 import { operation } from "./retry";
+import { performance } from "perf_hooks";
 import * as url from "url";
 import {
   ConnectionTimeoutError,
@@ -84,6 +85,7 @@ export class ServiceClientOptions {
   public autoParseJson?: boolean;
   public retryOptions?: {
     retries?: number;
+    retryAfter?: number;
     factor?: number;
     minTimeout?: number;
     maxTimeout?: number;
@@ -97,6 +99,7 @@ export class ServiceClientOptions {
   };
   public circuitBreaker?: false | CircuitBreakerOptions | CircuitBreakerFactory;
   public defaultRequestOptions?: Partial<ServiceClientRequestOptions>;
+  public dropAllRequestsAfter?: number;
 }
 
 /**
@@ -109,6 +112,7 @@ class ServiceClientStrictOptions {
   public autoParseJson: boolean;
   public retryOptions: {
     retries: number;
+    retryAfter: number;
     factor: number;
     minTimeout: number;
     maxTimeout: number;
@@ -121,6 +125,7 @@ class ServiceClientStrictOptions {
     ) => void;
   };
   public defaultRequestOptions: ServiceClientRequestOptions;
+  public dropAllRequestsAfter: number;
 
   constructor(options: ServiceClientOptions) {
     if (!options.hostname) {
@@ -140,6 +145,7 @@ class ServiceClientStrictOptions {
       minTimeout: 200,
       randomize: true,
       retries: 0,
+      retryAfter: 0,
       shouldRetry() {
         return true;
       },
@@ -163,6 +169,8 @@ class ServiceClientStrictOptions {
       timeout: 2000,
       ...options.defaultRequestOptions
     };
+
+    this.dropAllRequestsAfter = options.dropAllRequestsAfter || 0;
   }
 }
 
@@ -586,6 +594,7 @@ export class ServiceClient {
 
     const {
       retries,
+      retryAfter,
       factor,
       minTimeout,
       maxTimeout,
@@ -604,22 +613,54 @@ export class ServiceClient {
 
     const retryErrors: ServiceClientError[] = [];
     return new Promise<ServiceClientResponse>((resolve, reject) => {
+      const timerInitial = performance.now();
       const breaker = this.getCircuitBreaker(params);
-      const retryOperation = operation(opts, (currentAttempt: number) => {
+      const requestsAbortCallbacks: (() => void)[] = [];
+      const retryOperation = operation(opts, scheduledRetry => {
         breaker.run(
           (success: () => void, failure: () => void) => {
+            if (this.options.dropAllRequestsAfter) {
+              const timerLeft =
+                this.options.dropAllRequestsAfter -
+                (performance.now() - timerInitial);
+              if (params.dropRequestAfter) {
+                params.dropRequestAfter = Math.min(
+                  params.dropRequestAfter,
+                  timerLeft
+                );
+              } else {
+                params.dropRequestAfter = timerLeft;
+              }
+            }
+
+            if (retryAfter) {
+              params.registerAbortCallback = cb =>
+                requestsAbortCallbacks.push(cb);
+            }
+
             return requestWithFilters(
               this,
               params,
               this.options.filters || [],
               this.options.autoParseJson
             )
               .then((result: ServiceClientResponse) => {
+                if (retryOperation.isResolved()) {
+                  return;
+                }
+                retryOperation.resolved();
                 success();
                 result.retryErrors = retryErrors;
                 resolve(result);
+                if (retryAfter) {
+                  requestsAbortCallbacks.forEach(cb => cb());
+                  requestsAbortCallbacks.length = 0;
+                }
               })
               .catch((error: ServiceClientError) => {
+                if (retryOperation.isResolved()) {
+                  return;
+                }
                 retryErrors.push(error);
                 failure();
                 if (!shouldRetry(error, params)) {
@@ -628,11 +669,12 @@ export class ServiceClient {
                   );
                   return;
                 }
-                if (!retryOperation.retry()) {
+                const currentAttempt = retryOperation.retry();
+                if (!currentAttempt) {
                   // Wrapping error when user does not want retries would result
                   // in bad developer experience where you always have to unwrap it
                   // knowing there is only one error inside, so we do not do that.
-                  if (retries === 0) {
+                  if (retries === 0 || scheduledRetry) {
                     reject(error);
                   } else {
                     reject(
@@ -654,6 +696,20 @@ export class ServiceClient {
         );
       });
       retryOperation.attempt();
+
+      if (retryAfter) {
+        const retryAfterTimeout = () =>
+          setTimeout(() => {
+            if (!retryOperation.isResolved()) {
+              const currentAttempt = retryOperation.retry(true);
+              if (currentAttempt) {
+                onRetry(currentAttempt + 1, undefined, params);
+                retryAfterTimeout();
+              }
+            }
+          }, retryAfter);
+        retryAfterTimeout();
+      }
     }).catch((error: unknown) => {
       const rawError =
         error instanceof Error ? error : new Error(String(error));

lib/request.ts

@@ -70,6 +70,7 @@ export interface ServiceClientRequestOptions extends RequestOptions {
    * Opentracing like span interface to log events
    */
   span?: Span;
+  registerAbortCallback?: (abortFunction: () => void) => void;
 }
 
 export class ServiceClientResponse {
@@ -218,6 +219,28 @@ export const request = (
     }
 
     const requestObject = httpRequestFn(options);
+
+    let dropRequestAfterTimeout: NodeJS.Timer;
+    if (options.dropRequestAfter) {
+      dropRequestAfterTimeout = setTimeout(() => {
+        abortCallback();
+      }, options.dropRequestAfter);
+    }
+
+    function abortCallback() {
+      clearTimeout(dropRequestAfterTimeout);
+      if (!hasRequestEnded) {
+        requestObject.abort();
+        const err = new UserTimeoutError(options, timings);
+        logEvent(EventSource.HTTP_REQUEST, EventName.ERROR, err.message);
+        reject(err);
+      }
+    }
+
+    if (options.registerAbortCallback) {
+      options.registerAbortCallback(abortCallback);
+    }
+
     requestObject.setTimeout(readTimeout, () => {
       logEvent(EventSource.HTTP_REQUEST, EventName.TIMEOUT);
       requestObject.socket.destroy();
@@ -343,17 +366,6 @@ export const request = (
       });
     });
 
-    if (options.dropRequestAfter) {
-      setTimeout(() => {
-        if (!hasRequestEnded) {
-          requestObject.abort();
-          const err = new UserTimeoutError(options, timings);
-          logEvent(EventSource.HTTP_REQUEST, EventName.ERROR, err.message);
-          reject(err);
-        }
-      }, options.dropRequestAfter);
-    }
-
     if (options.body) {
       requestObject.write(options.body);
     }

lib/retry.ts

@@ -1,6 +1,6 @@
 export function operation(
   options: OperationOptions,
-  fn: (currentAttempt: number) => void
+  fn: (scheduledRetry: boolean) => void
 ) {
   return new RetryOperation(timeouts(options), fn);
 }
@@ -69,28 +69,43 @@ interface CreateTimeoutOptions {
 
 class RetryOperation {
   private readonly _timeouts: number[];
-  private readonly _fn: (currentAttempt: number) => void;
+  private readonly _fn: (scheduledRetry: boolean) => void;
+  private _resolved: boolean;
   private _attempts: number;
-  constructor(timeouts: number[], fn: (currentAttempt: number) => void) {
+  constructor(timeouts: number[], fn: (scheduledRetry: boolean) => void) {
     this._timeouts = timeouts;
     this._fn = fn;
+    this._resolved = false;
     this._attempts = 1;
   }
 
-  retry() {
+  retry(immediate: boolean = false) {
     if (this._attempts > this._timeouts.length) {
       return false;
     }
-    let timeout = this._timeouts[this._attempts - 1];
+    if (immediate) {
+      this._attempts++;
+      this._fn(immediate);
+      return this._attempts - 1;
+    }
+
     setTimeout(() => {
       this._attempts++;
-      this._fn(this._attempts);
-    }, timeout);
+      this._fn(immediate);
+    }, this._timeouts[this._attempts - 1]);
 
-    return true;
+    return this._attempts;
   }
 
   attempt() {
-    this._fn(this._attempts);
+    this._fn(false);
+  }
+
+  resolved() {
+    this._resolved = true;
+  }
+
+  isResolved() {
+    return this._resolved;
   }
 }