Retry logic for blockchain calls

Signed-off-by: Matthew Whitehead <matthew1001@gmail.com>

Author: matthew1001

README.md

@@ -29,8 +29,9 @@ are additional methods used by the token connector to guess at the contract ABI
 but is the preferred method for most use cases.
 
 To leverage this capability in a running FireFly environment, you must:
+
 1. [Upload the token contract ABI to FireFly](https://hyperledger.github.io/firefly/tutorials/custom_contracts/ethereum.html)
-as a contract interface.
+   as a contract interface.
 2. Include the `interface` parameter when [creating the pool on FireFly](https://hyperledger.github.io/firefly/tutorials/tokens).
 
 This will cause FireFly to parse the interface and provide ABI details
@@ -119,7 +120,29 @@ that specific token. If omitted, the approval covers all tokens.
 
 The following APIs are not part of the fftokens standard, but are exposed under `/api/v1`:
 
-* `GET /receipt/:id` - Get receipt for a previous request
+- `GET /receipt/:id` - Get receipt for a previous request
+
+## Retry behaviour
+
+Most short-term outages should be handled by the blockchain connector. For example if the blockchain node returns `HTTP 429` due to rate limiting
+it is the blockchain connector's responsibility to use appropriate back-off retries to attempt to make the required blockchain call successfully.
+
+There are cases where the token connector may need to perform it's own back-off retry for a blockchain action. For example if the blockchain connector
+microservice has crashed and is in the process of restarting just as the token connector is trying to query an NFT token URI to enrich a token event, if
+the token connector doesn't perform a retry then the event will be returned without the token URI populated.
+
+The token connector has configurable retry behaviour for all blockchain related calls. By default the connector will perform up to 15 retries with a back-off
+interval between each one. The default first retry interval is 100ms and doubles up to a maximum of 10s per retry interval. Retries are only performed where
+the error returned from the REST call matches a configurable regular expression retry condition. The default retry condition is `._ECONN._` which ensures
+retries take place for common TCP errors such as `ECONNRESET` and `ECONNREFUSED`.
+
+Setting the retry condition to "" disables retries. The configurable retry settings are:
+
+- `RETRY_BACKOFF_FACTOR` (default `2`)
+- `RETRY_BACKOFF_LIMIT_MS` (default `10000`)
+- `RETRY_BACKOFF_INITIAL_MS` (default `100`)
+- `RETRY_CONDITION` (default `.*ECONN.*`)
+- `RETRY_MAX_ATTEMPTS` (default `15`)
 
 ## Running the service
 

src/main.ts

@@ -84,6 +84,13 @@ async function bootstrap() {
   const legacyERC20 = config.get<string>('USE_LEGACY_ERC20_SAMPLE', '').toLowerCase() === 'true';
   const legacyERC721 = config.get<string>('USE_LEGACY_ERC721_SAMPLE', '').toLowerCase() === 'true';
 
+  // Configuration for retries
+  const retryBackOffFactor = config.get<number>('RETRY_BACKOFF_FACTOR', 2);
+  const retryBackOffLimit = config.get<number>('RETRY_BACKOFF_LIMIT_MS', 10000);
+  const retryBackOffInitial = config.get<number>('RETRY_BACKOFF_INITIAL_MS', 100);
+  const retryCondition = config.get<string>('RETRY_CONDITION', '.*ECONN.*');
+  const retriesMax = config.get<number>('RETRY_MAX_ATTEMPTS', 15);
+
   const passthroughHeaders: string[] = [];
   for (const h of passthroughHeaderString.split(',')) {
     passthroughHeaders.push(h.toLowerCase());
@@ -93,7 +100,18 @@ async function bootstrap() {
   app.get(TokensService).configure(ethConnectUrl, topic, factoryAddress);
   app
     .get(BlockchainConnectorService)
-    .configure(ethConnectUrl, fftmUrl, username, password, passthroughHeaders);
+    .configure(
+      ethConnectUrl,
+      fftmUrl,
+      username,
+      password,
+      passthroughHeaders,
+      retryBackOffFactor,
+      retryBackOffLimit,
+      retryBackOffInitial,
+      retryCondition,
+      retriesMax,
+    );
   app.get(AbiMapperService).configure(legacyERC20, legacyERC721);
 
   if (autoInit.toLowerCase() !== 'false') {

src/tokens/blockchain.service.ts

@@ -43,6 +43,12 @@ export class BlockchainConnectorService {
   password: string;
   passthroughHeaders: string[];
 
+  retryBackOffFactor: number;
+  retryBackOffLimit: number;
+  retryBackOffInitial: number;
+  retryCondition: string;
+  retriesMax: number;
+
   constructor(public http: HttpService) {}
 
   configure(
@@ -51,12 +57,22 @@ export class BlockchainConnectorService {
     username: string,
     password: string,
     passthroughHeaders: string[],
+    retryBackOffFactor: number,
+    retryBackOffLimit: number,
+    retryBackOffInitial: number,
+    retryCondition: string,
+    retriesMax: number,
   ) {
     this.baseUrl = baseUrl;
     this.fftmUrl = fftmUrl;
     this.username = username;
     this.password = password;
     this.passthroughHeaders = passthroughHeaders;
+    this.retryBackOffFactor = retryBackOffFactor;
+    this.retryBackOffLimit = retryBackOffLimit;
+    this.retryBackOffInitial = retryBackOffInitial;
+    this.retryCondition = retryCondition;
+    this.retriesMax = retriesMax;
   }
 
   private requestOptions(ctx: Context): AxiosRequestConfig {
@@ -88,15 +104,60 @@ export class BlockchainConnectorService {
     });
   }
 
+  // Check if retry condition matches the err that's been hit
+  private matchesRetryCondition(err: any): boolean {
+    return this.retryCondition != '' && err?.toString().match(this.retryCondition) !== null;
+  }
+
+  // Delay by the appropriate amount of time given the iteration the caller is in
+  private async backoffDelay(iteration: number) {
+    const delay = Math.min(
+      this.retryBackOffInitial * Math.pow(this.retryBackOffFactor, iteration),
+      this.retryBackOffLimit,
+    );
+    await new Promise(resolve => setTimeout(resolve, delay));
+  }
+
+  // Generic helper function that makes an given blockchain function retryable
+  // by using synchronous, back-off delays for cases where the function returns
+  // an error which matches the configured retry condition
+  private async retryableCall<T = any>(
+    blockchainFunction: () => Promise<AxiosResponse<T>>,
+  ): Promise<AxiosResponse<T>> {
+    let response: any;
+    for (let retries = 0; retries <= this.retriesMax; retries++) {
+      try {
+        response = await blockchainFunction();
+        break;
+      } catch (e) {
+        if (this.matchesRetryCondition(e)) {
+          this.logger.debug(`Retry condition matched for error ${e}`);
+          // Wait for a backed-off delay before trying again
+          await this.backoffDelay(retries);
+        } else {
+          // Whatever the error was it's not one we will retry for
+          break;
+        }
+      }
+    }
+
+    return response;
+  }
+
   async query(ctx: Context, to: string, method?: IAbiMethod, params?: any[]) {
-    const response = await this.wrapError(
-      lastValueFrom(
-        this.http.post<EthConnectReturn>(
-          this.baseUrl,
-          { headers: { type: queryHeader }, to, method, params },
-          this.requestOptions(ctx),
-        ),
-      ),
+    const url = this.baseUrl;
+    const response = await this.retryableCall<EthConnectReturn>(
+      async (): Promise<AxiosResponse<EthConnectReturn>> => {
+        return await this.wrapError(
+          lastValueFrom(
+            this.http.post(
+              url,
+              { headers: { type: queryHeader }, to, method, params },
+              this.requestOptions(ctx),
+            ),
+          ),
+        );
+      },
     );
     return response.data;
   }
@@ -110,26 +171,36 @@ export class BlockchainConnectorService {
     params?: any[],
   ) {
     const url = this.fftmUrl !== undefined && this.fftmUrl !== '' ? this.fftmUrl : this.baseUrl;
-    const response = await this.wrapError(
-      lastValueFrom(
-        this.http.post<EthConnectAsyncResponse>(
-          url,
-          { headers: { id, type: sendTransactionHeader }, from, to, method, params },
-          this.requestOptions(ctx),
-        ),
-      ),
+
+    const response = await this.retryableCall<EthConnectAsyncResponse>(
+      async (): Promise<AxiosResponse<EthConnectAsyncResponse>> => {
+        return await this.wrapError(
+          lastValueFrom(
+            this.http.post(
+              url,
+              { headers: { id, type: sendTransactionHeader }, from, to, method, params },
+              this.requestOptions(ctx),
+            ),
+          ),
+        );
+      },
     );
     return response.data;
   }
 
   async getReceipt(ctx: Context, id: string): Promise<EventStreamReply> {
-    const response = await this.wrapError(
-      lastValueFrom(
-        this.http.get<EventStreamReply>(new URL(`/reply/${id}`, this.baseUrl).href, {
-          validateStatus: status => status < 300 || status === 404,
-          ...this.requestOptions(ctx),
-        }),
-      ),
+    const url = this.baseUrl;
+    const response = await this.retryableCall<EventStreamReply>(
+      async (): Promise<AxiosResponse<EventStreamReply>> => {
+        return await this.wrapError(
+          lastValueFrom(
+            this.http.get(new URL(`/reply/${id}`, url).href, {
+              validateStatus: status => status < 300 || status === 404,
+              ...this.requestOptions(ctx),
+            }),
+          ),
+        );
+      },
     );
     if (response.status === 404) {
       throw new NotFoundException();

src/tokens/tokens.service.spec.ts

@@ -235,7 +235,7 @@ describe('TokensService', () => {
     service = module.get(TokensService);
     service.configure(BASE_URL, TOPIC, '');
     blockchain = module.get(BlockchainConnectorService);
-    blockchain.configure(BASE_URL, '', '', '', []);
+    blockchain.configure(BASE_URL, '', '', '', [], 2, 1000, 250, '.*ECONN.*', 15);
   });
 
   it('should be defined', () => {

test/app.e2e-context.ts

@@ -71,7 +71,9 @@ export class TestContext {
 
     this.app.get(EventStreamProxyGateway).configure('url', TOPIC);
     this.app.get(TokensService).configure(BASE_URL, TOPIC, '');
-    this.app.get(BlockchainConnectorService).configure(BASE_URL, '', '', '', []);
+    this.app
+      .get(BlockchainConnectorService)
+      .configure(BASE_URL, '', '', '', [], 2, 1000, 250, '.*ECONN.*', 15);
 
     (this.app.getHttpServer() as Server).listen();
     this.server = request(this.app.getHttpServer());