Weird socket.io behavior when Node server is down and then restarted

Question

I implemented a simple chat for my website where users can talk to each other with ExpressJS and Socket.io. I added a simple protection from a ddos attack that can be caused by one person spamming the window like this:

if (RedisClient.get(user).lastMessageDate > currentTime - 1 second) {

   return error("Only one message per second is allowed")

} else {

   io.emit('message', ...)     
   RedisClient.set(user).lastMessageDate = new Date()

}

I am testing this with this code:

setInterval(function() {
    $('input').val('message ' + Math.random());
    $('form').submit();
}, 1);

It works correctly when Node server is always up.

However, things get extremely weird if I turn off the Node server, then run the code above, and start Node server again in a few seconds. Then suddenly, hundreds of messages are inserted into the window and the browser crashes. I assume it is because when Node server is down, socket.io is saving all the client emits, and once it detects Node server is online again, it pushes all of those messages at once asynchronously.

How can I protect against this? And what is exactly happening here?

edit: If I use Node in-memory instead of Redis, this doesn't happen. I am guessing cause servers gets flooded with READs and many READs happen before RedisClient.set(user).lastMessageDate = new Date() finishes. I guess what I need is atomic READ / SET? I am using this module: https://github.com/NodeRedis/node_redis for connecting to Redis from Node.

Answer 1

You are correct that this happens due to queueing up of messages on client and flooding on server.

When the server receives messages, it receives messages all at once, and all of these messages are not synchronous. So, each of the socket.on("message:... events are executed separately, i.e. one socket.on("message... is not related to another and executed separately.

Even if your Redis-Server has a latency of a few ms, these messages are all received at once and everything always goes to the else condition.

You have the following few options.

1. Use a rate limiter library like this library. This is easy to configure and has multiple configuration options.

2. If you want to do everything yourself, use a queue on server. This will take up memory on your server, but you'll achieve what you want. Instead of writing every message to server, it is put into a queue. A new queue is created for every new client and delete this queue when processing the last item in queue.

3. (update) Use multi + watch to create lock so that all other commands except the current one will fail.

the pseudo-code will be something like this.

let queue = {};

let queueHandler = user => {
  while(queue.user.length > 0){
    // your redis push logic here
  }
  delete queue.user
}


let pushToQueue = (messageObject) => {
  let user = messageObject.user;

  if(queue.messageObject.user){
    queue.user = [messageObject];
  } else {
    queue.user.push(messageObject);
  }

  queueHandler(user);
}

socket.on("message", pushToQueue(message));

UPDATE

Redis supports locking with WATCH which is used with multi. Using this, you can lock a key, and any other commands that try to access that key in thet time fail.

from the redis client README

Using multi you can make sure your modifications run as a transaction, but you can't be sure you got there first. What if another client modified a key while you were working with it's data?

To solve this, Redis supports the WATCH command, which is meant to be used with MULTI: var redis = require("redis"), client = redis.createClient({ ... });

client.watch("foo", function( err ){
if(err) throw err;

client.get("foo", function(err, result) {
    if(err) throw err;

    // Process result
    // Heavy and time consuming operation here

    client.multi()
        .set("foo", "some heavy computation")
        .exec(function(err, results) {

            /**
             * If err is null, it means Redis successfully attempted 
             * the operation.
             */ 
            if(err) throw err;

            /**
             * If results === null, it means that a concurrent client
             * changed the key while we were processing it and thus 
             * the execution of the MULTI command was not performed.
             * 
             * NOTICE: Failing an execution of MULTI is not considered
             * an error. So you will have err === null and results === null
             */

        });
}); });

Answer 2

Perhaps you could extend your client-side code, to prevent data being sent if the socket is disconnected? That way, you prevent the library from queuing messages while the socket is disconnected (ie the server is offline).

This could be achieved by checking to see if socket.connected is true:

// Only allow data to be sent to server when socket is connected
function sendToServer(socket, message, data) {

    if(socket.connected) {
        socket.send(message, data)
    }
}

More information on this can be found at the docs https://socket.io/docs/client-api/#socket-connected

This approach will prevent the built in queuing behaviour in all scenarios where a socket is disconnected, which may not be desirable, however if should protect against the problem you are noting in your question.

Update

Alternatively, you could use a custom middleware on the server to achieve throttling behaviour via socket.io's server API:

/*
Server side code
*/
io.on("connection", function (socket) {

    // Add custom throttle middleware to the socket when connected
    socket.use(function (packet, next) {

        var currentTime = Date.now();

        // If socket has previous timestamp, check that enough time has
        // lapsed since last message processed
        if(socket.lastMessageTimestamp) {
            var deltaTime = currentTime - socket.lastMessageTimestamp;

            // If not enough time has lapsed, throw an error back to the
            // client
            if (deltaTime < 1000) {
                next(new Error("Only one message per second is allowed"))
                return
            }
        }

        // Update the timestamp on the socket, and allow this message to
        // be processed
        socket.lastMessageTimestamp = currentTime
        next()
    });
});