From f0e888f54feb502d24b40a99377f075539668956 Mon Sep 17 00:00:00 2001
From: Laurent Chenay <lchenay@gmail.com>
Date: Fri, 5 Sep 2014 15:40:46 +0200
Subject: [PATCH 1/2] Execute job in a domain and catch uncatched exception to
 mark job as failed.

Uncatched exception make kue to stop, and make job stuck in unknown statement.

Related to #391
---
 lib/queue/worker.js | 96 +++++++++++++++++++++++++--------------------
 test/test.js        | 12 ++++++
 2 files changed, 65 insertions(+), 43 deletions(-)

diff --git a/lib/queue/worker.js b/lib/queue/worker.js
index d89a7111..4b5e2c9c 100644
--- a/lib/queue/worker.js
+++ b/lib/queue/worker.js
@@ -151,53 +151,63 @@ Worker.prototype.failed = function (job, err, fn) {
 Worker.prototype.process = function (job, fn) {
     var self = this
         , start = new Date();
-    job.active( function(){
-        fn(
-            job,
-            function (err, result) {
-                if (err) {
-                    return self.failed(job, err, fn);
-                }
-                job.set('duration', job.duration = new Date - start);
-                if( result ) {
-                    try{
-                        job.result = result;
-                        job.set('result', JSON.stringify(result));
-                    }catch(e){
-                        job.set('result', JSON.stringify({error: true, message:'Invalid JSON Result: "' + result + '"' }));
+    var domain = require('domain').create();
+    domain.on('error', function(err){
+        self.failed(job, err, fn);
+    });
+
+    domain.run(function() {
+        job.active(function () {
+            fn(
+                job,
+                function (err, result) {
+                    if (err) {
+                        return self.failed(job, err, fn);
                     }
-                }
-                job.complete( function(){
-                    job.attempt( function(){
-                        self.emit('job complete', job);
-                        events.emit(job.id, 'complete', result);
-                        if( job.removeOnComplete() ) {
-                            job.remove();
+                    job.set('duration', job.duration = new Date - start);
+                    if (result) {
+                        try {
+                            job.result = result;
+                            job.set('result', JSON.stringify(result));
+                        } catch (e) {
+                            job.set('result', JSON.stringify({
+                                error: true,
+                                message: 'Invalid JSON Result: "' + result + '"'
+                            }));
+                        }
+                    }
+                    job.complete(function () {
+                        job.attempt(function () {
+                            self.emit('job complete', job);
+                            events.emit(job.id, 'complete', result);
+                            if (job.removeOnComplete()) {
+                                job.remove();
+                            }
+                        });
+                    }.bind(this));
+                    self.start(fn);
+                }, {
+                    /**
+                     * @author behrad
+                     * @pause: let the processor to tell worker not to continue processing new jobs
+                     */
+                    pause: function (fn, timeout) {
+                        timeout = timeout || 5000;
+                        self.queue.shutdown(fn, Number(timeout), self.type);
+                    },
+                    /**
+                     * @author behrad
+                     * @pause: let the processor to trigger restart for they job processing
+                     */
+                    resume: function () {
+                        if (self.resume()) {
+                            self.start(fn);
                         }
-                    });
-                }.bind(this));
-                self.start(fn);
-            },{
-                /**
-                 * @author behrad
-                 * @pause: let the processor to tell worker not to continue processing new jobs
-                 */
-                pause: function( fn, timeout ){
-                    timeout = timeout || 5000;
-                    self.queue.shutdown( fn, Number(timeout), self.type);
-                },
-                /**
-                 * @author behrad
-                 * @pause: let the processor to trigger restart for they job processing
-                 */
-                resume: function () {
-                    if (self.resume()) {
-                        self.start(fn);
                     }
                 }
-            }
-        );
-    }.bind(this));
+            );
+        }.bind(this));
+    });
     return this;
 };
 
diff --git a/test/test.js b/test/test.js
index e7505118..ab03df86 100755
--- a/test/test.js
+++ b/test/test.js
@@ -32,6 +32,18 @@ describe('Jobs', function () {
         });
     });
 
+    it('should catch uncatched exception and mark job as failed', function(testDone) {
+        jobs.create('failedJob', {}).on('complete', function() {
+            throw new Error('Job should be marked as failed and not complete');
+        }).on('failed', function() {
+            testDone();
+        }).save();
+
+        jobs.process('failedJob', 1, function() {
+            throw new Error('toto');
+        });
+    });
+
     it('should retry on failure if attempts is set', function (testDone) {
         var job = jobs.create('failure-attempts', {});
         var failures = 0;

From bd0e5f022ef5a041a482f580fbaa97e9af9db0a6 Mon Sep 17 00:00:00 2001
From: Laurent Chenay <lchenay@gmail.com>
Date: Fri, 5 Sep 2014 21:13:47 +0200
Subject: [PATCH 2/2] Ensure that domain.on('error') catch only uncaught
 exception and not other.

---
 test/test.js | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/test/test.js b/test/test.js
index ab03df86..d8b5aa36 100755
--- a/test/test.js
+++ b/test/test.js
@@ -33,13 +33,20 @@ describe('Jobs', function () {
     });
 
     it('should catch uncatched exception and mark job as failed', function(testDone) {
+        var catched = false;
         jobs.create('failedJob', {}).on('complete', function() {
             throw new Error('Job should be marked as failed and not complete');
         }).on('failed', function() {
+            catched.should.be.equal(true);
             testDone();
         }).save();
 
         jobs.process('failedJob', 1, function() {
+            try {
+                throw new Error('this should be catched')
+            } catch (err) {
+                catched = true;
+            }
             throw new Error('toto');
         });
     });