Merge pull request #1282 from mickhansen/overlookmotel-test_bad_db_conn_failure

Fix for MySQL connection errors

lib/dialects/mysql/connector-manager.js

@@ -38,8 +38,8 @@ module.exports = (function() {
     var self = this
 
     if (this.useReplicaton) {
-      var reads = 0,
-        writes = 0;
+      var reads = 0
+        , writes = 0;
 
       // Init configs with options from config if not present
       for (var i in config.replication.read) {
@@ -122,7 +122,10 @@ module.exports = (function() {
       this.pool = Pooling.Pool({
         name: 'sequelize-mysql',
         create: function (done) {
-          connect.call(self, done)
+          connect.call(self, function (err, connection) {
+            // This has to be nested for some reason, else the error won't propagate correctly
+            done(err, connection);
+          })
         },
         destroy: function(client) {
           disconnect.call(self, client)
@@ -148,25 +151,20 @@ module.exports = (function() {
 
   Utils._.extend(ConnectorManager.prototype, require("../connector-manager").prototype);
 
-  var isConnecting = false;
-
   ConnectorManager.prototype.query = function(sql, callee, options) {
-    if (!this.isConnected && !this.pool) {
-      this.connect()
-    }
-
     if (this.useQueue) {
+      // If queueing we'll let the execQueueItem method handle connecting
       var queueItem = {
-        query: new Query(this.client, this.sequelize, callee, options || {}),
+        query: new Query(null, this.sequelize, callee, options || {}),
         sql: sql
       };
 
       queueItem.query.options.uuid = this.config.uuid
-      enqueue.call(this, queueItem, options)
-      return queueItem.query
+      enqueue.call(this, queueItem, options);
+      return queueItem.query;
     }
 
-    var self = this, query = new Query(this.client, this.sequelize, callee, options || {});
+    var self = this, query = new Query(null, this.sequelize, callee, options || {});
     this.pendingQueries++;
 
     query.options.uuid = this.config.uuid
@@ -177,40 +175,66 @@ module.exports = (function() {
       } else {
         if (self.pendingQueries === 0) {
           setTimeout(function() {
-            self.pendingQueries === 0 && self.disconnect.call(self)
+            if (self.pendingQueries === 0){
+              self.disconnect.call(self);
+            } 
           }, 100);
         }
       }
     });
 
-    if (!this.pool) {
-      query.run(sql);
-    } else {
-      this.pool.acquire(function(err, client) {
-        if (err) {
-          return query.emit('error', err)
-        }
-
-        query.client = client
-        query.run(sql)
-        return;
-      }, undefined, options.type)
-    }
+    this.getConnection(options, function (err, client) {
+      query.client = client
+      query.run(sql)
+    });
 
     return query;
   };
 
-  ConnectorManager.prototype.connect = function() {
+  ConnectorManager.prototype.getConnection = function(options, callback) {
     var self = this;
-    // in case database is slow to connect, prevent orphaning the client
-    if (this.isConnecting || this.pool) {
-      return;
+
+    if (typeof options === "function") {
+      callback = options;
+      options = {};
     }
-    connect.call(self, function(err, client) {
-      self.client = client;
-      return;
-    });
-    return;
+
+    return new Utils.CustomEventEmitter(function (emitter) {
+      if (!self.pool) {
+        // Regular client caching
+        if (self.client) {
+          return emitter.emit('success', self.client);
+        } else {
+          // Cache for concurrent queries
+          if (self._getConnection) {
+            self._getConnection.proxy(emitter);
+            return;
+          }
+
+          // Set cache and acquire connection
+          self._getConnection = emitter;
+          connect.call(self, function(err, client) {
+            if (err) {
+              return emitter.emit('error', err);
+            }
+
+            // Unset caching, should now be caught by the self.client check
+            self._getConnection = null;
+            self.client = client;
+            emitter.emit('success', client);
+          });
+        }
+      }
+      if (self.pool) {
+        // Acquire from pool
+        self.pool.acquire(function(err, client) {
+          if (err) {
+            return emitter.emit('error', err);
+          }
+          emitter.emit('success', client);
+        }, options.priority, options.type);
+      }
+    }).run().done(callback);
   };
 
   ConnectorManager.prototype.disconnect = function() {
@@ -282,17 +306,30 @@ module.exports = (function() {
         case 'EINVAL':
           emitter.emit('error', 'Failed to find MySQL server. Please double check your settings.')
           break
+        default:
+          emitter.emit('error', err);
+          break;
         }
+
+        return;
       }
+
+      emitter.emit('success', connection);
     })
 
     connection.query("SET time_zone = '+0:00'");
     // client.setMaxListeners(self.maxConcurrentQueries)
     this.isConnecting = false
-    if (config.pool != null && config.pool.handleDisconnects) {
+    if (config.pool !== null && config.pool.handleDisconnects) {
       handleDisconnect(this.pool, connection)
     }
-    done(null, connection)
+
+    emitter.on('error', function (err) {
+      done(err);
+    });
+    emitter.on('success', function (connection) {
+      done(null, connection);
+    });
   }
 
   var handleDisconnect = function(pool, client) {
@@ -312,23 +349,7 @@ module.exports = (function() {
     options = options || {}
     if (this.activeQueue.length < this.maxConcurrentQueries) {
       this.activeQueue.push(queueItem)
-      if (this.pool) {
-        var self = this
-        this.pool.acquire(function(err, client) {
-          if (err) {
-            queueItem.query.emit('error', err)
-            return
-          }
-          //we set the client here, asynchronously, when getting a pooled connection
-          //allowing the ConnectorManager.query method to remain synchronous
-          queueItem.query.client = client
-          queueItem.client = client
-          execQueueItem.call(self, queueItem)
-          return
-        }, undefined, options.type)
-      } else {
-        execQueueItem.call(this, queueItem)
-      }
+      execQueueItem.call(this, queueItem)
     } else {
       this.queue.push(queueItem)
     }
@@ -361,11 +382,23 @@ module.exports = (function() {
   var execQueueItem = function(queueItem) {
     var self = this
 
-    queueItem.query
-      .success(function(){ afterQuery.call(self, queueItem) })
-      .error(function(){ afterQuery.call(self, queueItem) })
+    self.getConnection({
+      priority: queueItem.query.options.priority,
+      type: queueItem.query.options.type
+    }, function (err, connection) {
+      if (err) {
+        queueItem.query.emit('error', err)
+        return
+      }
 
-    queueItem.query.run(queueItem.sql, queueItem.client)
+      queueItem.query.client = connection
+      queueItem.client = connection
+      queueItem.query
+        .success(function(){ afterQuery.call(self, queueItem) })
+        .error(function(){ afterQuery.call(self, queueItem) })
+
+      queueItem.query.run(queueItem.sql, queueItem.client)
+    })
   }
 
   ConnectorManager.prototype.__defineGetter__('hasQueuedItems', function() {

test/sequelize.test.js

@@ -430,6 +430,37 @@ describe(Support.getTestDialectTeaser("Sequelize"), function () {
           done()
         })
       })
+
+      it('fails with incorrect database credentials (3)', function (done) {
+        var sequelize = new Sequelize('db', 'user', 'pass', {
+          dialect: this.sequelize.options.dialect,
+          port: 99999
+        });
+
+        var Project = sequelize.define('Project', {title: Sequelize.STRING})
+        var Task = sequelize.define('Task', {title: Sequelize.STRING})
+
+        sequelize.sync({force: true}).done(function (err) {
+          expect(err).to.be.ok
+          done()
+        })
+      })
+
+      it('fails with incorrect database credentials (4)', function (done) {
+        var sequelize = new Sequelize('db', 'user', 'pass', {
+          dialect: this.sequelize.options.dialect,
+          port: 99999,
+          pool: {}
+        });
+
+        var Project = sequelize.define('Project', {title: Sequelize.STRING})
+        var Task = sequelize.define('Task', {title: Sequelize.STRING})
+
+        sequelize.sync({force: true}).done(function (err) {
+          expect(err).to.be.ok
+          done()
+        })
+      })
     }
 
     describe("doesn't emit logging when explicitly saying not to", function() {