Add API credit support to BurstRateLimiter (#205)

Author: karen-stepanyan

src/rate-limiting/burst.ts

@@ -6,6 +6,8 @@ const logger = makeLogger('BurstRateLimiter')
 
 /**
  * This rate limiter is the simplest stateful option.
+ * It will keep track of used API credits (by default, one request equals one credit) and
+ * upon reaching the limits in a fixed time window, will block the requests until API credits are available again.
  * On startup, it'll compare the different thresholds for each tier, calculate them all
  * in the finest window we'll use (seconds), and use the most restrictive one.
  * This is so if the EA were to restart, we don't need to worry about persisting state
@@ -14,9 +16,9 @@ const logger = makeLogger('BurstRateLimiter')
  */
 export class BurstRateLimiter implements RateLimiter {
   latestSecondInterval = 0
-  requestsThisSecond = 0
+  creditsThisSecond = 0
   latestMinuteInterval = 0
-  requestsThisMinute = 0
+  creditsThisMinute = 0
   perSecondLimit!: number
   perMinuteLimit!: number
 
@@ -29,9 +31,8 @@ export class BurstRateLimiter implements RateLimiter {
     this.perMinuteLimit = Math.min(limits?.rateLimit1m || Infinity, perHourLimit)
     this.perSecondLimit = limits?.rateLimit1s || Infinity
     logger.debug(
-      `Using rate limiting settings: perMinute = ${this.perMinuteLimit} | perSecond: = ${this.perSecondLimit}`,
+      `Using API credit limit settings: perMinute = ${this.perMinuteLimit} | perSecond: = ${this.perSecondLimit}`,
     )
-
     return this
   }
 
@@ -46,18 +47,18 @@ export class BurstRateLimiter implements RateLimiter {
     // Ops should be "thread safe". Thank JS and its infinite single threaded dumbness.
     if (nearestSecondInterval !== this.latestSecondInterval) {
       logger.trace(
-        `Clearing latest second interval, # of requests logged was: ${this.requestsThisSecond} `,
+        `Clearing latest second interval, # of credits logged was: ${this.creditsThisSecond} `,
       )
       this.latestSecondInterval = nearestSecondInterval
-      this.requestsThisSecond = 0
+      this.creditsThisSecond = 0
     }
 
     if (nearestMinuteInterval !== this.latestMinuteInterval) {
       logger.trace(
-        `Clearing latest second minute, # of requests logged was: ${this.requestsThisMinute} `,
+        `Clearing latest second minute, # of credits logged was: ${this.creditsThisMinute} `,
       )
       this.latestMinuteInterval = nearestMinuteInterval
-      this.requestsThisMinute = 0
+      this.creditsThisMinute = 0
     }
 
     return {
@@ -76,33 +77,33 @@ export class BurstRateLimiter implements RateLimiter {
     const { now, nextSecondInterval, nextMinuteInterval } = this.updateIntervals()
 
     const timeToWaitForNextSecond =
-      this.requestsThisSecond < this.perSecondLimit ? 0 : nextSecondInterval - now
+      this.creditsThisSecond < this.perSecondLimit ? 0 : nextSecondInterval - now
     const timeToWaitForNextMinute =
-      this.requestsThisMinute < this.perMinuteLimit ? 0 : nextMinuteInterval - now
+      this.creditsThisMinute < this.perMinuteLimit ? 0 : nextMinuteInterval - now
     const timeToWait = Math.max(timeToWaitForNextSecond, timeToWaitForNextMinute)
 
     return timeToWait
   }
 
-  async waitForRateLimit(): Promise<void> {
+  async waitForRateLimit(creditCost = 1): Promise<void> {
     const timeToWait = this.msUntilNextExecution()
 
     if (timeToWait === 0) {
       logger.trace(
-        `Request under limits, current count: (S = ${this.requestsThisSecond} | M = ${this.requestsThisMinute})`,
+        `API credits under limits, current count: (S = ${this.creditsThisSecond} | M = ${this.creditsThisMinute}, | C = ${creditCost})`,
       )
     } else {
       logger.trace(
-        `Capacity for provider requests has been reached this interval (S = ${this.requestsThisSecond} | M = ${this.requestsThisMinute}), need to wait ${timeToWait}ms`,
+        `Capacity for provider API credits has been reached this interval (S = ${this.creditsThisSecond} | M = ${this.creditsThisMinute} | C = ${creditCost}), need to wait ${timeToWait}ms`,
       )
       await sleep(timeToWait)
       this.updateIntervals()
     }
+    this.creditsThisSecond += creditCost
+    this.creditsThisMinute += creditCost
 
-    this.requestsThisSecond++
-    this.requestsThisMinute++
     logger.trace(
-      `Request is now ready to go, updated count: (S = ${this.requestsThisSecond} | M = ${this.requestsThisMinute})`,
+      `Request is now ready to go, updated count: (S = ${this.creditsThisSecond} | M = ${this.creditsThisMinute})`,
     )
 
     return

src/rate-limiting/index.ts

@@ -31,8 +31,10 @@ export interface RateLimiter {
   /**
    * This method will block (if necessary) until the rate limiter can make sure the
    * next outbound request will be within the specified limits.
+   *
+   * @param cost - credit cost of an API request sent to Data Provider. Used for burst rate limiter
    */
-  waitForRateLimit(): Promise<void>
+  waitForRateLimit(cost?: number): Promise<void>
 
   /**
    * Returns the time in milliseconds until the next request would be able to be fired

src/transports/http.ts

@@ -47,6 +47,10 @@ export type ProviderRequestConfig<T extends HttpTransportGenerics> = {
 
   /** An endpoint name override for the requester key to allow the coalescing of requests across endpoints */
   endpointNameOverride?: string
+
+  /** The API credit cost of the request sent to the data provider.
+   * Applies only for burst rate limit strategy and is ignored if another strategy is used. */
+  cost?: number
 }
 
 /**
@@ -235,6 +239,7 @@ export class HttpTransport<T extends HttpTransportGenerics> extends Subscription
           transportName: this.name,
         }),
         requestConfig.request,
+        requestConfig.cost,
       )
 
       // Parse responses and apply timestamps

src/util/requester.ts

@@ -66,6 +66,7 @@ interface QueuedRequest<T = unknown> {
   key: string
   config: AxiosRequestConfig
   retries: number
+  cost?: number
   promise: Promise<RequesterResult<T>>
   reject: (err: unknown) => void
   resolve: (req: RequesterResult<T>) => void
@@ -158,9 +159,14 @@ export class Requester {
    *
    * @param key - a key to uniquely identify this request, and coalesce new ones that match
    * @param req - a request to send to a data provider
+   * @param cost - Data Provider API credit cost of the request
    * @returns a promise that will resolve whenever the request is popped from the queue, sent, and a response is received
    */
-  async request<T>(key: string, req: AxiosRequestConfig): Promise<RequesterResult<T>> {
+  async request<T>(
+    key: string,
+    req: AxiosRequestConfig,
+    cost?: number,
+  ): Promise<RequesterResult<T>> {
     // If there's already a queued request, reuse it's existing promise
     const existingQueuedRequest = this.map[key]
     if (existingQueuedRequest) {
@@ -174,6 +180,7 @@ export class Requester {
       key,
       config: req,
       retries: 0,
+      cost,
     } as QueuedRequest<T>
 
     // This dual promise layer is built so the queuedRequest can hold both the resolve and reject handlers,
@@ -212,7 +219,7 @@ export class Requester {
     )
 
     // Wait until the rate limiter allows the request to be executed
-    await this.rateLimiter.waitForRateLimit()
+    await this.rateLimiter.waitForRateLimit(next.cost)
 
     // Fire off to complete in the background. We don't wait here to be able to fire off multiple requests concurrently
     this.executeRequest.bind(this)(next)

test/rate-limit-config.test.ts

@@ -487,7 +487,7 @@ test('test build highest rate limits from config second, minute)', async (t) =>
   t.is(highestRateLimitTier, 4)
 })
 
-test('returns 0 when limit is set to infinity, no tier limit specified', async (t) => {
+test('returns 0 when limit is set to infinity, no tier limit specified (burst rate limiter)', async (t) => {
   const burstRateLimiter = RateLimiterFactory.buildRateLimiter(
     RateLimitingStrategy.BURST,
   ).initialize([], {})

test/transports/http.test.ts

@@ -104,6 +104,7 @@ class MockHttpTransport extends HttpTransport<HttpTransportTypes> {
             pairs: params.map((p) => ({ base: p.from, quote: p.to })),
           },
         },
+        cost: requestCost,
       }),
       parseResponse: (
         params,
@@ -145,6 +146,7 @@ const from = 'ETH'
 const to = 'USD'
 const price = 1234
 const volume = 4567
+const requestCost = 2 // Static API credit cost of the request
 
 axiosMock
   .onPost(URL + endpoint, {
@@ -254,16 +256,18 @@ test.serial(
       t.is(transport.backgroundExecuteCalls, expectedValue)
     }
 
+    let expected = rateLimit1m * Math.ceil(WARMUP_SUBSCRIPTION_TTL / 60_000)
+    await subtest(RateLimitingStrategy.FIXED_INTERVAL, expected)
+
     /**
-     * The expected number of requests is at follows:
-     * - At minute 0, a burst of `rateLimit1m` requests will be fired.
+     * The expected number of requests using burst limiter is at follows:
+     * - At minute 0, a burst of N requests will be fired, until the number of used API credits equals or exceeds `rateLimit1m`.
      * - The next request after those will be attempted immediately, only to hit the rate limiter blocking sleep.
      * - This will be repeated at the beginning of each minute interval (hence the rounding up when dividing the TTL).
      * - Finally, the +1 for the burst strategy will be the last request that was attempted to be fired then blocked,
      *     and by the time it's fired the subscription is no longer active.
      */
-    const expected = rateLimit1m * Math.ceil(WARMUP_SUBSCRIPTION_TTL / 60_000)
-    await subtest(RateLimitingStrategy.FIXED_INTERVAL, expected)
+    expected = Math.ceil(rateLimit1m / requestCost) * Math.ceil(WARMUP_SUBSCRIPTION_TTL / 60_000)
     await subtest(RateLimitingStrategy.BURST, expected + 1)
   },
 )
@@ -319,8 +323,10 @@ test.serial(
       t.is(transport.backgroundExecuteCalls, expectedValue)
     }
 
-    const expected = 60 * rateLimit1s + 1
+    let expected = 60 * rateLimit1s + 1
     await subtest(RateLimitingStrategy.FIXED_INTERVAL, expected)
+    // For burst rate limiter, we need to divide the credit limit to the cost of one request
+    expected = 60 * Math.ceil(rateLimit1s / requestCost) + 1
     await subtest(RateLimitingStrategy.BURST, expected)
   },
 )
@@ -871,9 +877,9 @@ test.serial(
       await testAdapter.api.close()
     }
 
-    // With the burst rate limiter, the result should be 2 requests, since we're advancing the clock
+    // With the burst rate limiter, the result should be 2 requests divided by the cost, since we're advancing the clock
     // less time than the per minute interval, but we're still in the same minute window and have capacity
-    await subtest(RateLimitingStrategy.BURST, 2)
+    await subtest(RateLimitingStrategy.BURST, 2 / requestCost)
     // The fixed interval rate limiter on the other hand should not have executed the same request yet, since it
     // will wait for 30s before actually getting the result
     // await subtest(RateLimitingStrategy.FIXED_INTERVAL, 1)