Schemaregistry oauth retry (#170)

* Add simple-oauth types to schema registry

* Add oauth retry and unit tests for retry

* Add oauth retry and unit tests for retry

rename retry helper

Rename retry helper

* Added boom error

* Add boom and boomify

* remove retryHelper

Author: Claimundefine

package-lock.json

@@ -35,9 +35,11 @@
     "@bufbuild/buf": "^1.37.0",
     "@bufbuild/protoc-gen-es": "^2.0.0",
     "@eslint/js": "^9.9.0",
+    "@hapi/boom": "^10.0.1",
     "@types/eslint__js": "^8.42.3",
     "@types/jest": "^29.5.13",
     "@types/node": "^20.16.1",
+    "axios-mock-adapter": "^2.1.0",
     "bluebird": "^3.5.3",
     "eslint": "^8.57.0",
     "eslint-plugin-jest": "^28.6.0",
@@ -53,7 +55,7 @@
     "@mapbox/node-pre-gyp": "^1.0.11",
     "bindings": "^1.3.1",
     "nan": "^2.17.0"
-    },
+  },
   "engines": {
     "node": ">=18.0.0"
   },
@@ -62,4 +64,4 @@
     "schemaregistry",
     "schemaregistry-examples"
   ]
-}
+}

package.json

@@ -1,13 +1,20 @@
 import { ModuleOptions, ClientCredentials, ClientCredentialTokenConfig, AccessToken } from 'simple-oauth2';
+import { sleep, fullJitter, isRetriable } from '../retry-helper';
+import { isBoom } from '@hapi/boom';
 
 const TOKEN_EXPIRATION_THRESHOLD_SECONDS = 30 * 60; // 30 minutes
 
 export class OAuthClient {
   private client: ClientCredentials;
   private token: AccessToken | undefined;
   private tokenParams: ClientCredentialTokenConfig;
+  private maxRetries: number;
+  private retriesWaitMs: number;
+  private retriesMaxWaitMs: number;
 
-  constructor(clientId: string, clientSecret: string, tokenHost: string, tokenPath: string, scope: string) {
+  constructor(clientId: string, clientSecret: string, tokenHost: string, tokenPath: string, scope: string,
+    maxRetries: number, retriesWaitMs: number, retriesMaxWaitMs: number
+  ) {
     const clientConfig: ModuleOptions = {
       client: {
         id: clientId,
@@ -22,6 +29,10 @@ export class OAuthClient {
     this.tokenParams = { scope };
 
     this.client = new ClientCredentials(clientConfig);
+
+    this.maxRetries = maxRetries;
+    this.retriesWaitMs = retriesWaitMs;
+    this.retriesMaxWaitMs = retriesMaxWaitMs;
   }
 
   async getAccessToken(): Promise<string> {
@@ -33,14 +44,21 @@ export class OAuthClient {
   }
 
   async generateAccessToken(): Promise<void> {
-    try {
-      const token = await this.client.getToken(this.tokenParams);
-      this.token = token;
-    } catch (error) {
-      if (error instanceof Error) {
-        throw new Error(`Failed to get token from server: ${error.message}`);
+    for (let i = 0; i < this.maxRetries + 1; i++) {
+      try {
+        const token = await this.client.getToken(this.tokenParams);
+        this.token = token;
+      } catch (error: any) {
+        if (isBoom(error) && i < this.maxRetries) {
+          const statusCode = error.output.statusCode;
+          if (isRetriable(statusCode)) {
+            const waitTime = fullJitter(this.retriesWaitMs, this.retriesMaxWaitMs, i);
+            await sleep(waitTime);
+            continue;
+          }
+        } 
+        throw new Error(`Failed to get token from server: ${error}`);
       }
-      throw new Error(`Failed to get token from server: ${error}`);
     }
   }
 
@@ -54,3 +72,4 @@ export class OAuthClient {
     throw new Error('Access token is not available');
   }
 }
+

schemaregistry/oauth/oauth-client.ts

@@ -38,7 +38,9 @@
     "@criteria/json-schema-validation": "^0.10.0",
     "@google-cloud/kms": "^4.5.0",
     "@hackbg/miscreant-esm": "^0.3.2-patch.3",
+    "@hapi/boom": "^10.0.1",
     "@smithy/types": "^3.3.0",
+    "@types/simple-oauth2": "^5.0.7",
     "@types/validator": "^13.12.0",
     "ajv": "^8.17.1",
     "async-mutex": "^0.5.0",

schemaregistry/package.json

@@ -2,6 +2,7 @@ import axios, { AxiosInstance, AxiosRequestConfig, AxiosResponse, CreateAxiosDef
 import { OAuthClient } from './oauth/oauth-client';
 import { RestError } from './rest-error';
 import axiosRetry from "axios-retry";
+import { fullJitter, isRetriable } from './retry-helper';
 /*
  * Confluent-Schema-Registry-TypeScript - Node.js wrapper for Confluent Schema Registry
  *
@@ -62,10 +63,10 @@ export class RestService {
     axiosRetry(this.client, {
       retries: maxRetries ?? 2,
       retryDelay: (retryCount) => {
-        return this.fullJitter(retriesWaitMs ?? 1000, retriesMaxWaitMs ?? 20000, retryCount - 1)
+        return fullJitter(retriesWaitMs ?? 1000, retriesMaxWaitMs ?? 20000, retryCount - 1)
       },
       retryCondition: (error) => {
-        return this.isRetriable(error.response?.status ?? 0);
+        return isRetriable(error.response?.status ?? 0);
       }
     });
     this.baseURLs = baseURLs;
@@ -76,16 +77,7 @@ export class RestService {
     this.setHeaders({ 'Content-Type': 'application/vnd.schemaregistry.v1+json' });
 
     this.handleBasicAuth(basicAuthCredentials);
-    this.handleBearerAuth(bearerAuthCredentials);
-  }
-
-  isRetriable(statusCode: number): boolean {
-    return statusCode == 408 || statusCode == 429
-      || statusCode == 500 || statusCode == 502 || statusCode == 503 || statusCode == 504;
-  }
-
-  fullJitter(baseDelayMs: number, maxDelayMs: number, retriesAttempted: number): number {
-    return Math.random() * Math.min(maxDelayMs, baseDelayMs * 2 ** retriesAttempted)
+    this.handleBearerAuth(maxRetries ?? 2, retriesWaitMs ?? 1000, retriesMaxWaitMs ?? 20000, bearerAuthCredentials);
   }
 
   handleBasicAuth(basicAuthCredentials?: BasicAuthCredentials): void {
@@ -119,7 +111,8 @@ export class RestService {
     }
   }
 
-  handleBearerAuth(bearerAuthCredentials?: BearerAuthCredentials): void {
+  handleBearerAuth(maxRetries: number, 
+    retriesWaitMs: number, retriesMaxWaitMs: number, bearerAuthCredentials?: BearerAuthCredentials): void {
     if (bearerAuthCredentials) {
       delete this.client.defaults.auth;
 
@@ -157,7 +150,8 @@ export class RestService {
           }
           const issuerEndPointUrl = new URL(bearerAuthCredentials.issuerEndpointUrl!);
           this.oauthClient = new OAuthClient(bearerAuthCredentials.clientId!, bearerAuthCredentials.clientSecret!,
-            issuerEndPointUrl.origin, issuerEndPointUrl.pathname, bearerAuthCredentials.scope!);
+            issuerEndPointUrl.origin, issuerEndPointUrl.pathname, bearerAuthCredentials.scope!, 
+            maxRetries, retriesWaitMs, retriesMaxWaitMs);
           break;
         default:
           throw new Error('Invalid bearer auth credentials source');

schemaregistry/rest-service.ts

@@ -0,0 +1,12 @@
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+function fullJitter(baseDelayMs: number, maxDelayMs: number, retriesAttempted: number): number {
+  return Math.random() * Math.min(maxDelayMs, baseDelayMs * 2 ** retriesAttempted)
+}
+
+function isRetriable(statusCode: number): boolean {
+  return statusCode == 408 || statusCode == 429
+    || statusCode == 500 || statusCode == 502 || statusCode == 503 || statusCode == 504;
+}
+
+export { sleep, fullJitter, isRetriable };