Throw Error when callback is called too many times.

lib/async.js

@@ -12,6 +12,15 @@
         return async;
     };
 
+    function only_once(fn) {
+        var called = false;
+        return function() {
+            if (called) throw new Error("Callback was already called.");
+            called = true;
+            fn.apply(root, arguments);
+        }
+    }
+
     //// cross-browser compatiblity functions ////
 
     var _forEach = function (arr, iterator) {
@@ -76,7 +85,7 @@
         }
         var completed = 0;
         _forEach(arr, function (x) {
-            iterator(x, function (err) {
+            iterator(x, only_once(function (err) {
                 if (err) {
                     callback(err);
                     callback = function () {};
@@ -87,7 +96,7 @@
                         callback(null);
                     }
                 }
-            });
+            }));
         });
     };
 
@@ -598,14 +607,14 @@
                     var task = q.tasks.shift();
                     if(q.empty && q.tasks.length == 0) q.empty();
                     workers += 1;
-                    worker(task.data, function () {
+                    worker(task.data, only_once(function() {
                         workers -= 1;
                         if (task.callback) {
                             task.callback.apply(task, arguments);
                         }
                         if(q.drain && q.tasks.length + workers == 0) q.drain();
                         q.process();
-                    });
+                    }));
                 }
             },
             length: function () {

test/test-async.js

@@ -533,6 +533,18 @@ exports['forEach'] = function(test){
     });
 };
 
+exports['forEach extra callback'] = function(test){
+    var count = 0;
+    async.forEach([1,3,2], function(val, callback) {
+        count++;
+        callback();
+        test.throws(callback);
+        if (count == 3) {
+            test.done();
+        }
+    });
+};
+
 exports['forEach empty array'] = function(test){
     test.expect(1);
     async.forEach([], function(x, callback){
@@ -1284,7 +1296,7 @@ exports['queue'] = function (test) {
     test.equal(q.length(), 4);
     test.equal(q.concurrency, 2);
 
-    setTimeout(function () {
+    q.drain = function () {
         test.same(call_order, [
             'process 2', 'callback 2',
             'process 1', 'callback 1',
@@ -1294,7 +1306,7 @@ exports['queue'] = function (test) {
         test.equal(q.concurrency, 2);
         test.equal(q.length(), 0);
         test.done();
-    }, 800);
+    };
 };
 
 exports['queue changing concurrency'] = function (test) {
@@ -1383,6 +1395,18 @@ exports['queue push without callback'] = function (test) {
     }, 800);
 };
 
+exports['queue too many callbacks'] = function (test) {
+    var q = async.queue(function (task, callback) {
+        callback();
+        test.throws(function() {
+            callback();
+        });
+        test.done();
+    }, 2);
+
+    q.push(1);
+};
+
 exports['queue bulk task'] = function (test) {
     var call_order = [],
         delays = [160,80,240,80];