feat: adding exponential backoff

package.json

@@ -41,7 +41,7 @@
     "lint:fix": "eslint . --fix",
     "format": "prettier --log-level warn --write \"**/*.{js,json,jsx,md,ts,tsx,html}\"",
     "format:check": "prettier --check \"**/*.{js,json,jsx,md,ts,tsx,html}\"",
-    "test:unit": "node --import tsx --test ./test/unit/*.test.ts",
+    "test:unit": "node --import tsx --test ./test/unit/*.test.ts --test ./test/unit/**/*.test.ts",
     "test:unit:watch": "node --import tsx --test --watch ./test/unit/*.test.ts",
     "test": "pnpm run test:unit && pnpm run lint && pnpm run format:check",
     "lcov": "node --import tsx --test --experimental-test-coverage --test-reporter=lcov --test-reporter-destination=coverage/lcov.info ./test/unit/*.test.ts",

src/lib/cloudflare.ts

@@ -2,6 +2,7 @@ import { ProviderError } from "../errors.js";
 import { throwErrorIfResponseNotOk } from "./fetch.js";
 
 const CLOUDFLARE_HOST = "https://api.cloudflare.com/client/v4";
+const MAX_RETRIES = 5;
 
 export function getCredentials() {
   const QUEUES_API_TOKEN = process.env.QUEUES_API_TOKEN;
@@ -17,13 +18,41 @@ export function getCredentials() {
   };
 }
 
+async function exponentialBackoff<T>(
+  fn: () => Promise<T>,
+  maxRetries: number,
+): Promise<T> {
+  let attempt = 0;
+  while (attempt < maxRetries) {
+    try {
+      return await fn();
+    } catch (error) {
+      if (error instanceof ProviderError && error.message.includes("429")) {
+        attempt++;
+        const delay = Math.pow(2, attempt) * 100 + Math.random() * 100;
+        await new Promise((resolve) => setTimeout(resolve, delay));
+      } else {
+        throw error;
+      }
+    }
+  }
+  throw new ProviderError("Max retries reached");
+}
+
 export async function queuesClient<T = unknown>({
   path,
   method,
   body,
   accountId,
   queueId,
   signal,
+}: {
+  path: string;
+  method: string;
+  body?: Record<string, unknown>;
+  accountId: string;
+  queueId: string;
+  signal?: AbortSignal;
 }): Promise<T> {
   const { QUEUES_API_TOKEN } = getCredentials();
 
@@ -38,15 +67,23 @@ export async function queuesClient<T = unknown>({
     signal,
   };
 
-  const response = await fetch(url, options);
+  async function fetchWithBackoff() {
+    const response = await fetch(url, options);
 
-  if (!response) {
-    throw new ProviderError("No response from Cloudflare Queues API");
-  }
+    if (!response) {
+      throw new ProviderError("No response from Cloudflare Queues API");
+    }
 
-  throwErrorIfResponseNotOk(response);
+    if (response.status === 429) {
+      throw new ProviderError("429 Too Many Requests");
+    }
 
-  const data = (await response.json()) as T;
+    throwErrorIfResponseNotOk(response);
+
+    const data = (await response.json()) as T;
+
+    return data;
+  }
 
-  return data;
+  return exponentialBackoff(fetchWithBackoff, MAX_RETRIES);
 }

test/unit/lib/cloudflare.test.ts

@@ -0,0 +1,110 @@
+import { describe, it, beforeEach, afterEach } from "node:test";
+import { assert } from "chai";
+import nock from "nock";
+import sinon from "sinon";
+
+import { queuesClient } from "../../../src/lib/cloudflare";
+import { ProviderError } from "../../../src/errors";
+
+const CLOUDFLARE_HOST = "https://api.cloudflare.com/client/v4";
+const ACCOUNT_ID = "test-account-id";
+const QUEUE_ID = "test-queue-id";
+const QUEUES_API_TOKEN = "test-queues-api-token";
+
+describe("queuesClient", () => {
+  let sandbox: sinon.SinonSandbox;
+
+  beforeEach(() => {
+    sandbox = sinon.createSandbox();
+    process.env.QUEUES_API_TOKEN = QUEUES_API_TOKEN;
+  });
+
+  afterEach(() => {
+    sandbox.restore();
+    delete process.env.QUEUES_API_TOKEN;
+    nock.cleanAll();
+  });
+
+  it("should successfully fetch data from Cloudflare Queues API", async () => {
+    const path = "messages";
+    const method = "GET";
+    const responseBody = { success: true, result: [] };
+
+    nock(CLOUDFLARE_HOST)
+      .get(`/accounts/${ACCOUNT_ID}/queues/${QUEUE_ID}/${path}`)
+      .reply(200, responseBody);
+
+    const result = await queuesClient({
+      path,
+      method,
+      accountId: ACCOUNT_ID,
+      queueId: QUEUE_ID,
+    });
+
+    assert.deepEqual(result, responseBody);
+  });
+
+  it("should throw an error if the API token is missing", async () => {
+    delete process.env.QUEUES_API_TOKEN;
+
+    try {
+      await queuesClient({
+        path: "messages",
+        method: "GET",
+        accountId: ACCOUNT_ID,
+        queueId: QUEUE_ID,
+      });
+      assert.fail("Expected error to be thrown");
+    } catch (error) {
+      assert.instanceOf(error, Error);
+      assert.equal(
+        error.message,
+        "Missing Cloudflare credentials, please set a QUEUES_API_TOKEN in the environment variables.",
+      );
+    }
+  });
+
+  it("should retry on 429 Too Many Requests", async () => {
+    const path = "messages";
+    const method = "GET";
+    const responseBody = { success: true, result: [] };
+
+    nock(CLOUDFLARE_HOST)
+      .get(`/accounts/${ACCOUNT_ID}/queues/${QUEUE_ID}/${path}`)
+      .reply(429, "Too Many Requests")
+      .get(`/accounts/${ACCOUNT_ID}/queues/${QUEUE_ID}/${path}`)
+      .reply(200, responseBody);
+
+    const result = await queuesClient({
+      path,
+      method,
+      accountId: ACCOUNT_ID,
+      queueId: QUEUE_ID,
+    });
+
+    assert.deepEqual(result, responseBody);
+  });
+
+  it("should throw ProviderError after max retries", async () => {
+    const path = "messages";
+    const method = "GET";
+
+    nock(CLOUDFLARE_HOST)
+      .get(`/accounts/${ACCOUNT_ID}/queues/${QUEUE_ID}/${path}`)
+      .times(5)
+      .reply(429, "Too Many Requests");
+
+    try {
+      await queuesClient({
+        path,
+        method,
+        accountId: ACCOUNT_ID,
+        queueId: QUEUE_ID,
+      });
+      assert.fail("Expected error to be thrown");
+    } catch (error) {
+      assert.instanceOf(error, ProviderError);
+      assert.equal(error.message, "Max retries reached");
+    }
+  });
+});