Replace Bottleneck with p-queue for throttling

README.md

@@ -96,32 +96,43 @@ Pass `{ throttle: { enabled: false } }` to disable this plugin.
 
 ### Clustering
 
-Enabling Clustering support ensures that your application will not go over rate limits **across Octokit instances and across Nodejs processes**.
+Clustering support allows your application to coordinate rate limiting **across Octokit instances and across Node.js processes** using Redis.
+
+> **Note:** Redis clustering support is available but requires implementing a custom connection adapter. The `connection` parameter accepts any object that implements `disconnect(): Promise<void>` and `on(event: string, handler: Function): void` methods.
 
 First install either `redis` or `ioredis`:
 
 ```
-# NodeRedis (https://github.com/NodeRedis/node_redis)
+# NodeRedis (https://github.com/redis/node-redis)
 npm install --save redis
 
-# or ioredis (https://github.com/luin/ioredis)
+# or ioredis (https://github.com/redis/ioredis)
 npm install --save ioredis
 ```
 
-Then in your application:
+Then create a connection adapter and pass it to the throttle options:
 
 ```js
-import Bottleneck from "bottleneck";
 import Redis from "redis";
 
 const client = Redis.createClient({
   /* options */
 });
-const connection = new Bottleneck.RedisConnection({ client });
-connection.on("error", err => console.error(err));
+
+// Create a connection adapter that implements the required interface
+const connection = {
+  async disconnect() {
+    await client.disconnect();
+  },
+  on(event, handler) {
+    client.on(event, handler);
+  },
+};
+
+connection.on("error", (err) => console.error(err));
 
 const octokit = new MyOctokit({
-  auth: 'secret123'
+  auth: "secret123",
   throttle: {
     onSecondaryRateLimit: (retryAfter, options, octokit) => {
       /* ... */
@@ -130,16 +141,13 @@ const octokit = new MyOctokit({
       /* ... */
     },
 
-    // The Bottleneck connection object
+    // The connection object for Redis coordination
     connection,
 
     // A "throttling ID". All octokit instances with the same ID
     // using the same Redis server will share the throttling.
     id: "my-super-app",
-
-    // Otherwise the plugin uses a lighter version of Bottleneck without Redis support
-    Bottleneck
-  }
+  },
 });
 
 // To close the connection and allow your application to exit cleanly:
@@ -153,7 +161,16 @@ import Redis from "ioredis";
 const client = new Redis({
   /* options */
 });
-const connection = new Bottleneck.IORedisConnection({ client });
+
+const connection = {
+  async disconnect() {
+    await client.disconnect();
+  },
+  on(event, handler) {
+    client.on(event, handler);
+  },
+};
+
 connection.on("error", (err) => console.error(err));
 ```
 
@@ -201,10 +218,10 @@ connection.on("error", (err) => console.error(err));
         <code>options.connection</code>
       </th>
       <td>
-        <code>Bottleneck.RedisConnection</code>
+        <code>Connection</code>
       </td>
       <td>
-        A Bottleneck connection instance. See <a href="#clustering">Clustering</a> above.
+        A connection object for Redis clustering. Must implement <code>disconnect(): Promise&lt;void&gt;</code> and <code>on(event: string, handler: Function): void</code> methods. See <a href="#clustering">Clustering</a> above.
       </td>
     </tr>
     <tr>
@@ -218,17 +235,6 @@ connection.on("error", (err) => console.error(err));
         A "throttling ID". All octokit instances with the same ID using the same Redis server will share the throttling. See <a href="#clustering">Clustering</a> above. Defaults to <code>no-id</code>.
       </td>
     </tr>
-    <tr>
-      <th>
-        <code>options.Bottleneck</code>
-      </th>
-      <td>
-        <code>Bottleneck</code>
-      </td>
-      <td>
-        Bottleneck constructor. See <a href="#clustering">Clustering</a> above. Defaults to `bottleneck/light`.
-      </td>
-    </tr>
   </tbody>
 </table>
 

package.json

@@ -23,7 +23,7 @@
   "license": "MIT",
   "dependencies": {
     "@octokit/types": "^16.0.0",
-    "bottleneck": "^2.15.3"
+    "p-queue": "^9.0.0"
   },
   "peerDependencies": {
     "@octokit/core": "^7.0.0"

src/index.ts

@@ -1,6 +1,4 @@
-// @ts-expect-error No types for "bottleneck/light"
-import BottleneckLight from "bottleneck/light.js";
-import type TBottleneck from "bottleneck";
+import { EventEmitter } from "node:events";
 import type { Octokit, OctokitOptions } from "@octokit/core";
 import type {
   CreateGroupsCommon,
@@ -9,6 +7,8 @@ import type {
   ThrottlingOptions,
 } from "./types.js";
 import { VERSION } from "./version.js";
+import { ThrottleGroup } from "./throttle-group.js";
+import { ThrottleLimiter } from "./throttle-limiter.js";
 
 import { wrapRequest } from "./wrap-request.js";
 import triggersNotificationPaths from "./generated/triggers-notification-paths.js";
@@ -21,33 +21,31 @@ const triggersNotification = regex.test.bind(regex);
 
 const groups: Groups = {};
 
-const createGroups = function (
-  Bottleneck: typeof TBottleneck,
-  common: CreateGroupsCommon,
-) {
-  groups.global = new Bottleneck.Group({
+const createGroups = function (common: CreateGroupsCommon) {
+  groups.global = new ThrottleGroup({
     id: "octokit-global",
     maxConcurrent: 10,
+    minTime: 0, // Explicitly set to match Bottleneck's behavior
     ...common,
   });
-  groups.auth = new Bottleneck.Group({
+  groups.auth = new ThrottleGroup({
     id: "octokit-auth",
     maxConcurrent: 1,
     ...common,
   });
-  groups.search = new Bottleneck.Group({
+  groups.search = new ThrottleGroup({
     id: "octokit-search",
     maxConcurrent: 1,
     minTime: 2000,
     ...common,
   });
-  groups.write = new Bottleneck.Group({
+  groups.write = new ThrottleGroup({
     id: "octokit-write",
     maxConcurrent: 1,
     minTime: 1000,
     ...common,
   });
-  groups.notifications = new Bottleneck.Group({
+  groups.notifications = new ThrottleGroup({
     id: "octokit-notifications",
     maxConcurrent: 1,
     minTime: 3000,
@@ -58,7 +56,6 @@ const createGroups = function (
 export function throttling(octokit: Octokit, octokitOptions: OctokitOptions) {
   const {
     enabled = true,
-    Bottleneck = BottleneckLight as typeof TBottleneck,
     id = "no-id",
     timeout = 1000 * 60 * 2, // Redis TTL: 2 minutes
     connection,
@@ -72,7 +69,7 @@ export function throttling(octokit: Octokit, octokitOptions: OctokitOptions) {
   }
 
   if (groups.global == null) {
-    createGroups(Bottleneck, common);
+    createGroups(common);
   }
 
   const state: State = Object.assign(
@@ -81,7 +78,7 @@ export function throttling(octokit: Octokit, octokitOptions: OctokitOptions) {
       triggersNotification,
       fallbackSecondaryRateRetryAfter: 60,
       retryAfterBaseValue: 1000,
-      retryLimiter: new Bottleneck(),
+      retryLimiter: new ThrottleLimiter(),
       id,
       ...(groups as Required<Groups>),
     },
@@ -105,17 +102,32 @@ export function throttling(octokit: Octokit, octokitOptions: OctokitOptions) {
     `);
   }
 
-  const events = {};
-  const emitter = new Bottleneck.Events(events);
-  // @ts-expect-error
-  events.on("secondary-limit", state.onSecondaryRateLimit);
-  // @ts-expect-error
-  events.on("rate-limit", state.onRateLimit);
-  // @ts-expect-error
-  events.on("error", (e) =>
+  const emitter = new EventEmitter();
+  emitter.on("secondary-limit", state.onSecondaryRateLimit);
+  emitter.on("rate-limit", state.onRateLimit);
+  emitter.on("error", (e) =>
     octokit.log.warn("Error in throttling-plugin limit handler", e),
   );
 
+  // Helper to emit event and get handler return value
+  const emitAndGetResult = async (
+    event: string,
+    ...args: any[]
+  ): Promise<any> => {
+    try {
+      const listeners = emitter.listeners(event);
+      if (listeners.length > 0) {
+        const result = await (listeners[0] as any)(...args);
+        return result;
+      }
+      return undefined;
+    } catch (error) {
+      // Emit error event if handler throws
+      emitter.emit("error", error);
+      return undefined;
+    }
+  };
+
   state.retryLimiter.on("failed", async function (error, info) {
     const [state, request, options] = info.args as [
       State,
@@ -146,7 +158,7 @@ export function throttling(octokit: Octokit, octokitOptions: OctokitOptions) {
         const retryAfter =
           Number(error.response.headers["retry-after"]) ||
           state.fallbackSecondaryRateRetryAfter;
-        const wantRetry = await emitter.trigger(
+        const wantRetry = await emitAndGetResult(
           "secondary-limit",
           retryAfter,
           options,
@@ -175,7 +187,7 @@ export function throttling(octokit: Octokit, octokitOptions: OctokitOptions) {
           Math.ceil((rateLimitReset - Date.now()) / 1000) + 1,
           0,
         );
-        const wantRetry = await emitter.trigger(
+        const wantRetry = await emitAndGetResult(
           "rate-limit",
           retryAfter,
           options,