Merge pull request #864 from coralproject/redis-changes

Redis Improvements

config.js

@@ -8,6 +8,7 @@
 require('env-rewrite').rewrite();
 
 const uniq = require('lodash/uniq');
+const ms = require('ms');
 
 //==============================================================================
 // CONFIG INITIALIZATION
@@ -84,6 +85,23 @@ const CONFIG = {
   MONGO_URL: process.env.TALK_MONGO_URL,
   REDIS_URL: process.env.TALK_REDIS_URL,
 
+  // REDIS_RECONNECTION_MAX_ATTEMPTS is the amount of attempts that a redis
+  // connection will attempt to reconnect before aborting with an error.
+  REDIS_RECONNECTION_MAX_ATTEMPTS: parseInt(process.env.TALK_REDIS_RECONNECTION_MAX_ATTEMPTS || '100'),
+
+  // REDIS_RECONNECTION_MAX_RETRY_TIME is the time in string format for the
+  // maximum amount of time that a client can be considered "connecting" before
+  // attempts at reconnection are aborted with an error.
+  REDIS_RECONNECTION_MAX_RETRY_TIME: ms(process.env.TALK_REDIS_RECONNECTION_MAX_RETRY_TIME || '1 min'),
+
+  // REDIS_RECONNECTION_BACKOFF_FACTOR is the factor that will be multiplied
+  // against the current attempt count inbetween attempts to connect to redis.
+  REDIS_RECONNECTION_BACKOFF_FACTOR: ms(process.env.TALK_REDIS_RECONNECTION_BACKOFF_FACTOR || '500 ms'),
+
+  // REDIS_RECONNECTION_BACKOFF_MINIMUM_TIME is the minimum time used to delay
+  // before attempting to reconnect to redis.
+  REDIS_RECONNECTION_BACKOFF_MINIMUM_TIME: ms(process.env.TALK_REDIS_RECONNECTION_BACKOFF_MINIMUM_TIME || '1 sec'),
+
   //------------------------------------------------------------------------------
   // Server Config
   //------------------------------------------------------------------------------

docs/_docs/02-01-configuration.md

@@ -52,6 +52,21 @@ These are only used during the webpack build.
 - `TALK_MONGO_URL` (*required*) - the database connection string for the MongoDB database.
 - `TALK_REDIS_URL` (*required*) - the database connection string for the Redis database.
 
+#### Advanced
+
+- `TALK_REDIS_RECONNECTION_MAX_ATTEMPTS` (_optional_) - the amount of attempts
+  that a redis connection will attempt to reconnect before aborting with an
+  error. (Default `100`)
+- `TALK_REDIS_RECONNECTION_MAX_RETRY_TIME` (_optional_) - the time in string
+  format for the maximum amount of time that a client can be considered
+  "connecting" before attempts at reconnection are aborted with an error.
+  (Default `1 min`)
+- `TALK_REDIS_RECONNECTION_BACKOFF_FACTOR` (_optional_) - the time factor that
+  will be multiplied against the current attempt count inbetween attempts to
+  connect to redis. (Default `500 ms`)
+- `TALK_REDIS_RECONNECTION_BACKOFF_MINIMUM_TIME` (_optional_) - the minimum time
+  used to delay before attempting to reconnect to redis. (Default `1 sec`)
+
 ### Server
 
 - `TALK_ROOT_URL` (*required*) - root url of the installed application externally

graph/context.js

@@ -53,7 +53,7 @@ class Context {
     this.plugins = decorateContextPlugins(this, contextPlugins);
 
     // Bind the publish/subscribe to the context.
-    this.pubsub = pubsub.createClient();
+    this.pubsub = pubsub.getClient();
   }
 }
 

graph/subscriptions.js

@@ -127,15 +127,13 @@ const setupFunctions = plugins.get('server', 'setupFunctions').reduce((acc, {plu
   }),
 });
 
-const pubsubClient = pubsub.createClientFactory();
-
 /**
  * This creates a new subscription manager.
  */
 const createSubscriptionManager = (server) => new SubscriptionServer({
   subscriptionManager: new SubscriptionManager({
     schema,
-    pubsub: pubsubClient(),
+    pubsub: pubsub.getClient(),
     setupFunctions,
   }),
   onConnect: ({token}, connection) => {

middleware/pubsub.js

@@ -1,12 +1,11 @@
 const pubsub = require('../services/pubsub');
-const pubsubClient = pubsub.createClientFactory();
 
 // To handle dependancy injection safer, we inject the pubsub handle onto the
 // request object.
 module.exports = (req, res, next) => {
 
   // Attach the pubsub handle to the requests.
-  req.pubsub = pubsubClient();
+  req.pubsub = pubsub.getClient();
 
   // Forward on the request.
   next();

services/pubsub.js

@@ -1,23 +1,24 @@
 const {RedisPubSub} = require('graphql-redis-subscriptions');
+const {connectionOptions, attachMonitors} = require('./redis');
 
-const {connectionOptions} = require('./redis');
+/**
+ * getClient returns the pubsub singleton for this instance.
+ */
+let pubsub = null;
+const getClient = () => {
+  if (pubsub !== null) {
+    return pubsub;
+  }
 
-const createClient = () => new RedisPubSub({connection: connectionOptions});
+  pubsub = new RedisPubSub({connection: connectionOptions});
 
-const createClientFactory = () => {
-  let ins = null;
-  return () => {
-    if (ins) {
-      return ins;
-    }
+  // Attach the node monitors to the subscriber + publishers.
+  attachMonitors(pubsub.redisPublisher);
+  attachMonitors(pubsub.redisSubscriber);
 
-    ins = createClient();
-
-    return ins;
-  };
+  return pubsub;
 };
 
 module.exports = {
-  createClient,
-  createClientFactory
+  getClient,
 };

services/redis.js

@@ -1,45 +1,80 @@
 const redis = require('redis');
 const debug = require('debug')('talk:services:redis');
+const enabled = require('debug').enabled('talk:services:redis');
 const {
-  REDIS_URL
+  REDIS_URL,
+  REDIS_RECONNECTION_MAX_ATTEMPTS,
+  REDIS_RECONNECTION_MAX_RETRY_TIME,
+  REDIS_RECONNECTION_BACKOFF_FACTOR,
+  REDIS_RECONNECTION_BACKOFF_MINIMUM_TIME,
 } = require('../config');
 
+const attachMonitors = (client) => {
+  debug('client created');
+
+  // Debug events.
+  if (enabled) {
+    client.on('ready', () => debug('client ready'));
+    client.on('connect', () => debug('client connected'));
+    client.on('reconnecting', () => debug('client connection lost, attempting to reconnect'));
+    client.on('end', () => debug('client ended'));
+  }
+
+  // Error events.
+  client.on('error', (err) => {
+    if (err) {
+      console.error('Error connecting to redis:', err);
+    }
+  });
+};
+
 const connectionOptions = {
   url: REDIS_URL,
   retry_strategy: function(options) {
-    if (options.error && options.error.code === 'ECONNREFUSED') {
+    if (options.error && options.error.code !== 'ECONNREFUSED') {
+
+      debug('retry strategy: none, an error occured');
 
       // End reconnecting on a specific error and flush all commands with a individual error
-      return new Error('The server refused the connection');
+      return options.error;
     }
-    if (options.total_retry_time > 1000 * 60 * 60) {
+    if (options.total_retry_time > REDIS_RECONNECTION_MAX_RETRY_TIME) {
+
+      debug('retry strategy: none, exhausted retry time');
 
       // End reconnecting after a specific timeout and flush all commands with a individual error
       return new Error('Retry time exhausted');
     }
 
-    if (options.times_connected > 10) {
+    if (options.attempt > REDIS_RECONNECTION_MAX_ATTEMPTS) {
+
+      debug('retry strategy: none, exhausted retry attempts');
 
       // End reconnecting with built in error
       return undefined;
     }
 
     // reconnect after
-    return Math.max(options.attempt * 100, 3000);
+    const delay = Math.max(options.attempt * REDIS_RECONNECTION_BACKOFF_FACTOR, REDIS_RECONNECTION_BACKOFF_MINIMUM_TIME);
+
+    debug(`retry strategy: try to reconnect ${delay} ms from now`);
+
+    return delay;
   }
 };
 
 const createClient = () => {
   let client = redis.createClient(connectionOptions);
 
+  // Attach the monitors that will print debug messages to the console.
+  attachMonitors(client);
+
   client.ping((err) => {
     if (err) {
       console.error('Can\'t ping the redis server!');
 
       throw err;
     }
-
-    debug('connection established');
   });
 
   return client;
@@ -47,12 +82,13 @@ const createClient = () => {
 
 module.exports = {
   connectionOptions,
+  attachMonitors,
   createClient,
   createClientFactory: () => {
     let client = null;
 
     return () => {
-      if (client) {
+      if (client !== null) {
         return client;
       }