feat: Clustering support using Redis (#20)

Author: SGrondin

README.md

@@ -55,6 +55,57 @@ async function createIssueOnAllRepos (org) {
 
 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**.
+
+First install either `redis` or `ioredis`:
+```
+# NodeRedis (https://github.com/NodeRedis/node_redis)
+npm install --save redis
+
+# or ioredis (https://github.com/luin/ioredis)
+npm install --save ioredis
+```
+
+Then in your application:
+```js
+const Bottleneck = require('bottleneck')
+const Redis = require('redis')
+
+const client = Redis.createClient({ /* options */ })
+const connection = new Bottleneck.RedisConnection({ client })
+connection.on('error', err => console.error(err))
+
+const octokit = new Octokit({
+  throttle: {
+    onAbuseLimit: (retryAfter, options) => { /* ... */ },
+    onRateLimit: (retryAfter, options) => { /* ... */ },
+
+    // The Bottleneck connection object
+    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:
+await connection.disconnect()
+```
+
+To use the `ioredis` library instead:
+```js
+const Redis = require('ioredis')
+const client = new Redis({ /* options */ })
+const connection = new Bottleneck.IORedisConnection({ client })
+connection.on('error', err => console.error(err))
+```
+
 ## LICENSE
 
 [MIT](LICENSE)

lib/index.js

@@ -1,6 +1,6 @@
 module.exports = throttlingPlugin
 
-const Bottleneck = require('bottleneck/light')
+const BottleneckLight = require('bottleneck/light')
 const wrapRequest = require('./wrap-request')
 const triggersNotificationPaths = require('./triggers-notification-paths')
 const routeMatcher = require('./route-matcher')(triggersNotificationPaths)
@@ -10,29 +10,44 @@ const triggersNotification = throttlingPlugin.triggersNotification =
   routeMatcher.test.bind(routeMatcher)
 
 function throttlingPlugin (octokit, octokitOptions = {}) {
+  const {
+    enabled = true,
+    Bottleneck = BottleneckLight,
+    id = 'no-id',
+    connection
+  } = octokitOptions.throttle || {}
+  if (!enabled) {
+    return
+  }
+  const common = {
+    connection,
+    timeout: 1000 * 60 * 10 // Redis TTL: 10 minutes
+  }
   const state = Object.assign({
-    enabled: true,
+    clustering: connection != null,
     triggersNotification,
     minimumAbuseRetryAfter: 5,
     retryAfterBaseValue: 1000,
     globalLimiter: new Bottleneck({
-      maxConcurrent: 1
+      id: `octokit-global-${id}`,
+      maxConcurrent: 1,
+      ...common
     }),
     writeLimiter: new Bottleneck({
+      id: `octokit-write-${id}`,
       maxConcurrent: 1,
-      minTime: 1000
+      minTime: 1000,
+      ...common
     }),
     triggersNotificationLimiter: new Bottleneck({
+      id: `octokit-notifications-${id}`,
       maxConcurrent: 1,
-      minTime: 3000
+      minTime: 3000,
+      ...common
     }),
     retryLimiter: new Bottleneck()
   }, octokitOptions.throttle)
 
-  if (!state.enabled) {
-    return
-  }
-
   if (typeof state.onAbuseLimit !== 'function' || typeof state.onRateLimit !== 'function') {
     throw new Error(`octokit/plugin-throttling error:
         You must pass the onAbuseLimit and onRateLimit error handlers.

lib/wrap-request.js

@@ -10,6 +10,11 @@ async function doRequest (state, request, options) {
   const isWrite = options.method !== 'GET' && options.method !== 'HEAD'
   const retryCount = ~~options.request.retryCount
   const jobOptions = retryCount > 0 ? { priority: 0, weight: 0 } : {}
+  if (state.clustering) {
+    // Remove a job from Redis if it has not completed or failed within 60s
+    // Examples: Node process terminated, client disconnected, etc.
+    jobOptions.expiration = 1000 * 60
+  }
 
   // Guarantee at least 1000ms between writes
   if (isWrite) {