watch stuck active jobs

Author: paulserraino

lib/q.js

@@ -49,5 +49,45 @@ q.series = function (actions, callback) {
   async.series(tasks, callback);
 };
 
-module.exports = q;
+/*
+* Watch for active stuck jobs
+*
+* @param {Number} watch stuck active jobs every sec seconds
+*/
+
+q.watchStuckActiveJobs = function (sec) {
+
+  sec = sec || 5 * 60;
+  setTimeout(watch, sec * 1000);
+
+  function watch () {
+    q.client.zrange('q:jobs:active', 1, -1, function (error, ids) {
+      if (error) return console.log(error)
+
+      var tasks = ids.map(function (id) {
+        return function (next) {
+          kue.Job.get(id, function (error, job) {
+            if (error) return next(error);
+
+            // re-enqueue stuck active jobs
+            if ( moment(+job.updated_at).isBefore(moment().subtract(5, 'minutes')) ) {
+              return job.state('inactive', next);
+            }
+
+            next(null);
+          });
+        }
+      });
 
+      var done = function (error) {
+        if (error) console.log(error);
+        q.watchStuckActiveJobs(sec);
+      }
+
+      async.parallelLimit(tasks, 10, done)
+    });
+  }
+
+};
+
+module.exports = q;

lib/worker.js

@@ -46,3 +46,9 @@ q.process('build-pdf:upload-to-s3', config.limit, function (job, done) {
   });
 
 });
+
+/*
+* Watch stuck active jobs
+*/
+
+q.watchStuckActiveJobs();