Trapped parse error when garbage comes down the socket

Author: christkv

HISTORY

@@ -20,6 +20,7 @@
 * Fixes for handling of errors during shutdown off a socket connection
 * Correctly applies socket options including timeout
 * Cleanup of test management code to close connections correctly
+* Handle parser errors better, closing down the connection and emitting an error
 
 0.9.7 2011-11-10
 * Added priority setting to replicaset manager

lib/mongodb/connection/connection.js

@@ -205,6 +205,15 @@ var createDataHandler = exports.Connection.createDataHandler = function(self) {
  if(data.length > 4) {
  // Retrieve the message size
  var sizeOfMessage = binaryutils.decodeUInt32(data, 0);
+ // If we have a negative sizeOfMessage emit error and return
+ if(sizeOfMessage < 0 || sizeOfMessage > self.maxBsonSize) {
+ // We got a parse Error fire it off then keep going
+ self.emit("parseError", {err:"socketHandler", trace:'', bin:self.buffer, parseState:{
+ sizeOfMessage:sizeOfMessage, 
+ bytesRead:self.bytesRead,
+ stubBuffer:self.stubBuffer}}); 
+ return;
+ }
 
  // Ensure that the size of message is larger than 0 and less than the max allowed
  if(sizeOfMessage > 4 && sizeOfMessage < self.maxBsonSize && sizeOfMessage > data.length) {
@@ -255,14 +264,22 @@ var createDataHandler = exports.Connection.createDataHandler = function(self) {
  data = new Buffer(0);
 
  } else {
- self.emit("message", data.slice(0, sizeOfMessage));
- // Reset state of buffer
- self.buffer = null;
- self.sizeOfMessage = 0;
- self.bytesRead = 0;
- self.stubBuffer = null;
- // Copy rest of message
- data = data.slice(sizeOfMessage);
+ try {
+ self.emit("message", data.slice(0, sizeOfMessage));
+ // Reset state of buffer
+ self.buffer = null;
+ self.sizeOfMessage = 0;
+ self.bytesRead = 0;
+ self.stubBuffer = null;
+ // Copy rest of message
+ data = data.slice(sizeOfMessage); 
+ } catch (err) {
+ // We got a parse Error fire it off then keep going
+ self.emit("parseError", {err:"socketHandler", trace:err, bin:self.buffer, parseState:{
+ sizeOfMessage:sizeOfMessage, 
+ bytesRead:self.bytesRead,
+ stubBuffer:self.stubBuffer}});
+ }
 
  } 
  } else {

lib/mongodb/connection/repl_set_servers.js

@@ -274,6 +274,7 @@ ReplSetServers.prototype.allServerInstances = function() {
 }
 
 ReplSetServers.prototype.connect = function(parent, options, callback) {
+ console.log("----------------------------------------------------------------------------------- connect")
  if('function' === typeof options) callback = options, options = {}; 
  if(options == null) options = {};
  if(!('function' === typeof callback)) callback = null;
@@ -349,7 +350,7 @@ ReplSetServers.prototype.connect = function(parent, options, callback) {
  var tags = document.tags ? document.tags : {};
  var primary = document.primary;
  var me = document.me;
-
+ 
  // Error handler for the servers, this handles unexpected errors coming from 
  // a wrong callback or something else
  var errorHandler = function(err) {
@@ -542,6 +543,11 @@ ReplSetServers.prototype.connect = function(parent, options, callback) {
  replSetSelf._state.errorMessages.push(err);
  }
  } 
+ 
+ // Check if we have an error in the inital set of servers and callback with error
+ if(replSetSelf._state.errorMessages.length > 0 && firstCall) {
+ callback(replSetSelf._state.errorMessages[0], null);
+ }
 }
 
 ReplSetServers.prototype.checkoutWriter = function() {

lib/mongodb/connection/server.js

@@ -255,10 +255,13 @@ Server.prototype.connect = function(dbInstance, options, callback) {
  }
  });
 
- // Handle errors
+ // If we have a parser error we are in an unknown state, close everything and emit
+ // error
  connectionPool.on("parseError", function(message) { 
  // Force close the pool
- if(connectionPool.isConnected()) self.stop(); 
+ if(connectionPool.isConnected()) self.stop();
+ // Emit error
+ server.emit("error", message);
  });
 
  // Boot up connection poole, pass in a locator of callbacks

test/replicaset/connect_test.js

@@ -143,14 +143,13 @@ module.exports = testCase({
 
  new Db('integration_test_', replSet).open(function(err, db) {
  test.equal(null, err);
- var dbCloseCount = 0;//, serverCloseCount = 0;
+ var dbCloseCount = 0;
  db.on('close', function() { ++dbCloseCount; });
 
  db.close(function() {
  // Let all events fire.
  process.nextTick(function() {
  test.equal(dbCloseCount, 0);
- // test.equal(serverCloseCount, db.serverConfig.servers.length);
  test.done();
  });
  });