worker-farm - npm Package Compare versions

Comparing version 1.0.1 to 1.1.0

lib/farm.js

@@ -7,2 +7,3 @@ const DEFAULT_OPTIONS = {
         , maxCallTime                 : Infinity // exceed this and the whole worker is terminated
+        , maxRetries                  : Infinity
         , forcedKillTime              : 100
@@ -14,2 +15,3 @@ }
     , TimeoutError            = require('errno').create('TimeoutError')
+    , ProcessTerminatedError  = require('errno').create('ProcessTerminatedError')
     , MaxConcurrentCallsError = require('errno').create('MaxConcurrentCallsError')
@@ -37,2 +39,3 @@
       , args     : args
+      , retries  : 0
     })
@@ -69,4 +72,12 @@ }.bind(this)
     if (this.children[childId] && this.children[childId].activeCalls) {
-      this.children[childId].calls.reverse().forEach(function (call) {
-        if (call) {
+      this.children[childId].calls.forEach(function (call, i) {
+        if (!call) return
+        else if (call.retries >= this.options.maxRetries) {
+          this.receive({
+              idx   : i
+            , child : childId
+            , args  : [ new ProcessTerminatedError('cancel after ' + call.retries + ' retries!') ]
+          })
+        } else {
+          call.retries++
           this.callQueue.unshift(call)
@@ -73,0 +84,0 @@ doQueue = true

package.json

 {
   "name": "worker-farm",
   "description": "Distribute processing tasks to child processes with an über-simple API and baked-in durability & custom concurrency options.",
-  "version": "1.0.1",
+  "version": "1.1.0",
   "homepage": "https://github.com/rvagg/node-worker-farm",
@@ -25,3 +25,3 @@ "authors": [
   "devDependencies": {
-    "tape": ">=2.14.0 <2.15.0-0"
+    "tape": ">=3.0.3 <3.1.0-0"
   },
@@ -28,0 +28,0 @@ "scripts": {

README.md

 # Worker Farm [![Build Status](https://secure.travis-ci.org/rvagg/node-worker-farm.png)](http://travis-ci.org/rvagg/node-worker-farm)
 
-[![NPM](https://nodei.co/npm/worker-farm.png?downloads=true&stars=true)](https://nodei.co/npm/worker-farm/) [![NPM](https://nodei.co/npm-dl/worker-farm.png?months=6)](https://nodei.co/npm/worker-farm/)
+[![NPM](https://nodei.co/npm/worker-farm.png?downloads=true&downloadRank=true&stars=true)](https://nodei.co/npm/worker-farm/) [![NPM](https://nodei.co/npm-dl/worker-farm.png?months=6&height=3)](https://nodei.co/npm/worker-farm/)
 
@@ -111,2 +111,3 @@
   , maxCallTime                 : Infinity
+  , maxRetries                  : Infinity
 }
@@ -123,4 +124,6 @@ ```
 
-  * **<code>maxCallTime</code>** *(use with caution, understand what this does before you use it!)* when `!== Infinity`, will cap a time, in milliseconds, that *any single call* can take to execute in a worker. If this time limit is exceeded by just a single call then the worker running that call will be killed and any calls running on that worker will have their callbacks returned with a `TimeoutError` (check `err.type == 'TimeoutError'`). If you are running with `maxConcurrentCallsPerWorker` value greater than `1` then **all calls currently executing** will fail and not be automatically resubmitted. Use this if you have jobs that may potentially end in infinite loops that you can't programatically end with your child code. Preferably run this with a `maxConcurrentCallsPerWorker` so you don't interrupt other calls when you have a timeout. This timeout operates on a per-call basis but will interrupt a whole worker.
+  * **<code>maxCallTime</code>** *(use with caution, understand what this does before you use it!)* when `!== Infinity`, will cap a time, in milliseconds, that *any single call* can take to execute in a worker. If this time limit is exceeded by just a single call then the worker running that call will be killed and any calls running on that worker will have their callbacks returned with a `TimeoutError` (check `err.type == 'TimeoutError'`). If you are running with `maxConcurrentCallsPerWorker` value greater than `1` then **all calls currently executing** will fail and will be automatically resubmitted uless you've changed the `maxRetries` option. Use this if you have jobs that may potentially end in infinite loops that you can't programatically end with your child code. Preferably run this with a `maxConcurrentCallsPerWorker` so you don't interrupt other calls when you have a timeout. This timeout operates on a per-call basis but will interrupt a whole worker.
 
+  * **<code>maxRetries</code>** allows you to control the max number of call requeues after worker termination (unexpected or timeout). By default this option is set to `Infinity` which means that each call of each terminated worker will always be auto requeued. When the number of retries exceeds `maxRetries` value, the job callback will be executed with a `ProcessTerminatedError`. Note that if you are running with finite `maxCallTime` and `maxConcurrentCallsPerWorkers` greater than `1` then any `TimeoutError` will increase the retries counter *for each* concurrent call of the terminated worker.
+
 ### workerFarm.end(farm)
@@ -127,0 +130,0 @@

tests/child.js

@@ -0,1 +1,3 @@
+var fs = require('fs')
+
 module.exports = function (timeout, callback) {
@@ -26,2 +28,26 @@ callback = callback.bind(null, null, process.pid, Math.random(), timeout)
   while (true);
+}
+
+// use provided file path to save retries count among terminated workers
+module.exports.stubborn = function (path, callback) {
+  function isOutdated(path) {
+    return ((new Date).getTime() - fs.statSync(path).mtime.getTime()) > 2000
+  }
+
+  // file may not be properly deleted, check if modified no earler than two seconds ago
+  if (!fs.existsSync(path) || isOutdated(path)) {
+    fs.writeFileSync(path, '1')
+    process.exit(-1)
+  }
+
+  var retry = parseInt(fs.readFileSync(path, 'utf8'))
+  if (Number.isNaN(retry))
+    return callback(new Error('file contents is not a number'))
+
+  if (retry > 4) {
+    callback(null, 12)
+  } else {
+    fs.writeFileSync(path, String(retry + 1))
+    process.exit(-1)
+  }
 }

tests/index.js

 var tape       = require('tape')
   , workerFarm = require('../')
   , childPath  = require.resolve('./child')
+  , fs         = require('fs')
 
@@ -399,2 +400,33 @@ , uniq = function (ar) {
   })
+})
+
+
+tape('test max retries after process terminate', function (t) {
+  t.plan(7)
+
+  // temporary file is used to store the number of retries among terminating workers
+  var filepath1 = '.retries1'
+  var child1 = workerFarm({ maxConcurrentWorkers: 1, maxRetries: 5}, childPath, [ 'stubborn' ])
+  child1.stubborn(filepath1, function (err, result) {
+    t.notOk(err, 'no error')
+    t.equal(result, 12, 'correct result')
+  })
+
+  workerFarm.end(child1, function () {
+    fs.unlinkSync(filepath1)
+    t.ok(true, 'workerFarm ended')
+  })
+
+  var filepath2 = '.retries2'
+  var child2 = workerFarm({ maxConcurrentWorkers: 1, maxRetries: 3}, childPath, [ 'stubborn' ])
+  child2.stubborn(filepath2, function (err, result) {
+    t.ok(err, 'got an error')
+    t.equal(err.type, 'ProcessTerminatedError', 'correct error type')
+    t.equal(err.message, 'cancel after 3 retries!', 'correct message and number of retries')
+  })
+
+  workerFarm.end(child2, function () {
+    fs.unlinkSync(filepath2)
+    t.ok(true, 'workerFarm ended')
+  })
 })

New alerts

Filesystem access

Supply chain risk

Accesses the file system, and could potentially read sensitive data.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

40657

increased by7.82%

Lines of code

814

increased by7.67%

Number of lines in readme file

139

increased by2.21%

Worsened metrics

Number of low supply chain risk alerts

4

increased by100%

No dependency changes