Turn errors from scripts and extensions into faults

lib/config.coffee

@@ -42,6 +42,7 @@ options = {
   FS_IP : {type : 'string', default : '192.168.0.1'},
 
   DOWNLOAD_TIMEOUT: {type : 'int', default : 3600},
+  EXT_TIMEOUT: {type: 'int', default: 3000},
   PRESETS_CACHE_DURATION : {type : 'int', default : 86400},
   LOG_INFORMS : {type : 'bool', default : true},
   DEBUG : {type : 'bool', default : false},

lib/cwmp.coffee

@@ -207,9 +207,12 @@ applyPresets = (currentRequest) ->
     for k, v of parameters
       declarations.push([v, 1, {value: 1}])
 
-    session.rpcRequest(currentRequest.sessionData, declarations, (err, id, rpcRequest) ->
+    session.rpcRequest(currentRequest.sessionData, declarations, (err, faults, id, rpcRequest) ->
       throw err if err
 
+      if faults?
+        throw new Error('A fault occured while trying to discover parameters to test preset preconditions: ' + JSON.stringify(faults))
+
       if rpcRequest?
         return sendRpcRequest(currentRequest, id, rpcRequest)
 
@@ -227,9 +230,18 @@ applyPresets = (currentRequest) ->
         if query.testFilter(parameters, p.precondition)
           session.addProvisions(currentRequest.sessionData, p.channel, p.provisions)
 
-      session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest, doneProvisions) ->
+      session.rpcRequest(currentRequest.sessionData, null, (err, faults, id, rpcRequest, doneProvisions) ->
         throw err if err
 
+        if faults?
+          for channel, fault of faults
+            if not fault.retries? and Object.keys(faults).length == 1
+              fault.retries = 0
+            currentRequest.sessionData.faults[channel] = fault
+            util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})")
+
+          return nextRpc(currentRequest)
+
         for p, i in doneProvisions or []
           delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]]
 
@@ -240,8 +252,26 @@ applyPresets = (currentRequest) ->
 
 
 nextRpc = (currentRequest) ->
-  session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest, doneProvisions) ->
+  session.rpcRequest(currentRequest.sessionData, null, (err, faults, id, rpcRequest, doneProvisions) ->
     throw err if err
+
+    if faults?
+      for channel, fault of faults
+        if channel.startsWith('_task_')
+          taskId = channel.slice(6)
+          for t, j in currentRequest.sessionData.tasks
+            if t._id == taskId
+              currentRequest.sessionData.tasks.splice(j, 1)
+              fault.expiry = t.expiry if t.expiry?
+              break
+
+        if not fault.retries? and Object.keys(faults).length == 1
+          fault.retries = 0
+        currentRequest.sessionData.faults[channel] = fault
+        util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})")
+
+      return nextRpc(currentRequest)
+
     if rpcRequest?
       return sendRpcRequest(currentRequest, id, rpcRequest)
 

lib/extension-wrapper.coffee

@@ -1,26 +0,0 @@
-###
-# Copyright 2013-2016  Zaid Abdulla
-#
-# GenieACS is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-#
-# GenieACS is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Affero General Public License for more details.
-#
-# You should have received a copy of the GNU Affero General Public License
-# along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
-###
-
-process.chdir(process.argv[2])
-script = require("#{process.argv[2]}/#{process.argv[3]}")
-
-
-process.on('message', (message) ->
-  script[message[1][0]](message[1].slice(1), (err, res) ->
-    process.send([message[0], err, res])
-  )
-)

lib/extension-wrapper.js

@@ -0,0 +1,62 @@
+/**
+ * Copyright 2013-2016  Zaid Abdulla
+ *
+ * GenieACS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * GenieACS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+"use strict";
+
+const jobs = new Set();
+const fileName = process.argv[2];
+let script;
+
+function errorToFault(err) {
+  if (!err)
+    return null;
+
+  return {
+    code: `ext.${err.name}`,
+    message: err.message,
+    detail: {
+      name: err.name,
+      message: err.message,
+      stack: err.stack
+    }
+  }
+}
+
+process.on("uncaughtException", function(err) {
+  let fault = errorToFault(err);
+  jobs.forEach(function(jobId) {
+    process.send([jobId, fault]);
+  });
+  jobs.clear();
+  process.disconnect();
+});
+
+process.on("message", function(message) {
+  jobs.add(message[0]);
+
+  if (!script) {
+    let cwd = process.env["GENIEACS_CONFIG_DIR"] + "/ext";
+    process.chdir(cwd);
+    script = require(`${cwd}/${fileName}`);
+  }
+
+  script[message[1][0]](message[1].slice(1), function(err, res) {
+    if (!jobs.delete(message[0]))
+      return;
+
+    process.send([message[0], errorToFault(err), res]);
+  });
+});

lib/extensions.coffee

@@ -1,55 +0,0 @@
-###
-# Copyright 2013-2016  Zaid Abdulla
-#
-# GenieACS is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-#
-# GenieACS is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Affero General Public License for more details.
-#
-# You should have received a copy of the GNU Affero General Public License
-# along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
-###
-
-childProcess = require 'child_process'
-crypto = require 'crypto'
-path = require 'path'
-
-config = require './config'
-
-
-TIMEOUT = 5000
-
-processes = {}
-queue = {}
-
-messageHandler = (message) ->
-  func = queue[message[0]]
-  delete queue[message[0]]
-  func(message[1], message[2])
-
-
-run = (args, callback) ->
-  scriptName = args[0]
-  if not processes[scriptName]?
-    cwd = path.resolve(config.get('CONFIG_DIR'), 'ext')
-    p = childProcess.fork('lib/extension-wrapper', [cwd, scriptName])
-    p.on('message', messageHandler)
-    processes[scriptName] = p
-
-  id = crypto.randomBytes(8).toString('hex')
-  queue[id] = callback
-  setTimeout(() ->
-    if id of queue
-      delete queue[id]
-      return callback(new Error('Extension timeout'))
-  , TIMEOUT)
-
-  processes[scriptName].send([id, args.slice(1)])
-
-
-exports.run = run

lib/extensions.js

@@ -0,0 +1,98 @@
+/**
+ * Copyright 2013-2016  Zaid Abdulla
+ *
+ * GenieACS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * GenieACS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+"use strict";
+
+const childProcess = require("child_process");
+const crypto = require ("crypto");
+
+const config = require("./config");
+
+const TIMEOUT = config.get('EXT_TIMEOUT');
+
+const processes = {};
+const jobs = {};
+
+function messageHandler(message) {
+  let func = jobs[message[0]];
+  if (func) {
+    delete jobs[message[0]];
+    func(null, message[1], message[2]);
+  }
+}
+
+
+function run(args, callback) {
+  let scriptName = args[0];
+
+  if (!processes[scriptName]) {
+    let p = childProcess.fork('lib/extension-wrapper', [scriptName]);
+
+    p.on('error', function() {
+      kill(processes[scriptName]);
+      delete processes[scriptName];
+    });
+
+    p.on('disconnect', function() {
+      kill(processes[scriptName]);
+      delete processes[scriptName];
+    });
+
+    p.on('message', messageHandler);
+    processes[scriptName] = p;
+  }
+
+  let id = crypto.randomBytes(8).toString('hex');
+  jobs[id] = callback;
+  setTimeout(function() {
+    if (id in jobs) {
+      delete jobs[id];
+      return callback(null, {code: "timeout", message: "Extension timed out"});
+    }
+  }, TIMEOUT);
+
+  return processes[scriptName].send([id, args.slice(1)]);
+};
+
+
+function kill(process) {
+  if (process.signalCode !== null || process.exitCode !== null)
+    return;
+
+  let timeToKill = Date.now() + 5000;
+
+  process.kill();
+
+  let t = setInterval(function() {
+    if (process.signalCode !== null || process.exitCode !== null) {
+      clearInterval(t);
+    }
+    else if (Date.now() > timeToKill) {
+      process.kill('SIGKILL');
+      clearInterval(t);
+    }
+  }, 100);
+}
+
+
+function killAll() {
+  for (let n in processes)
+    kill(processes[n]);
+}
+
+
+exports.run = run;
+exports.killAll = killAll;

lib/sandbox.js

@@ -204,6 +204,22 @@ Object.defineProperty(sandbox.context, 'commit', {value: commit});
 Object.defineProperty(sandbox.context, 'ext', {value: ext});
 
 
+function errorToFault(err) {
+  if (!err)
+    return null;
+
+  return {
+    code: `script.${err.name}`,
+    message: err.message,
+    detail: {
+      name: err.name,
+      message: err.message,
+      stack: err.stack
+    }
+  }
+}
+
+
 function run(script, globals, timestamp, deviceData, extensionsCache, startRevision, maxRevision, callback) {
   sandbox.timestamp = timestamp;
   sandbox.deviceData = deviceData;
@@ -233,7 +249,7 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis
     else if (err === EXT)
       status = 2;
     else
-      return callback(err);
+      return callback(null, errorToFault(err));
   }
 
   let ext = sandbox.extensions;
@@ -245,17 +261,17 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis
   let counter = 1;
   for (let key in sandbox.extensions) {
     ++ counter;
-    extensions.run(sandbox.extensions[key], function(err, res) {
+    extensions.run(sandbox.extensions[key], function(err, fault, res) {
-      if (err && counter > 0) {
+      if ((err || fault) && counter > 0) {
         counter = 0;
-        return callback(err);
+        return callback(err, fault);
       }
       extCache[key] = res;
       if (-- counter === 0)
         if (status === 2)
           return run.apply(null, args);
         else
-          return callback(null, clr, decs, status === 0, ret);
+          return callback(null, null, clr, decs, status === 0, ret);
     });
   }
 
@@ -263,7 +279,7 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis
     if (status === 2)
       return run.apply(null, args);
     else
-      return callback(null, clr, decs, status === 0, ret);
+      return callback(null, null, clr, decs, status === 0, ret);
 }
 
 

lib/session.coffee

@@ -21,7 +21,6 @@ db = require './db'
 device = require './device'
 sandbox = require './sandbox'
 cache = require './cache'
-extensions = require './extensions'
 PathSet = require './path-set'
 VersionedMap = require './versioned-map'
 InstanceSet = require './instance-set'
@@ -379,12 +378,12 @@ runProvisions = (sessionData, provisions, startRevision, endRevision, callback)
     sandbox.run(sessionData.cache.provisions[provision[0]].script,
       {args: provision[1]}, sessionData.timestamp, sessionData.deviceData,
       sessionData.extensionsCache, startRevision, endRevision,
-      (err, _clear, _declarations, _done) ->
+      (err, _fault, _clear, _declarations, _done) ->
         -- counter
-        if err
+        if err or _fault
           if counter >= 0
             counter = 0
-            return callback(err)
+            return callback(err, _fault)
           return
 
         done &&= _done
@@ -396,11 +395,11 @@ runProvisions = (sessionData, provisions, startRevision, endRevision, callback)
           allClear = allClear.concat(_clear)
 
         if counter == 0
-          return callback(null, done, allDeclarations, allClear)
+          return callback(null, null, done, allDeclarations, allClear)
       )
 
   if -- counter == 0
-    return callback(null, done, allDeclarations, allClear)
+    return callback(null, null, done, allDeclarations, allClear)
 
 
 runVirtualParameters = (sessionData, provisions, startRevision, endRevision, callback) ->
@@ -415,12 +414,12 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal
     sandbox.run(sessionData.cache.virtualParameters[provision[0]].script, globals,
       sessionData.timestamp, sessionData.deviceData,
       sessionData.extensionsCache, startRevision, endRevision,
-      (err, _clear, _declarations, _done, _returnValue) ->
+      (err, _fault, _clear, _declarations, _done, _returnValue) ->
         -- counter
-        if err
+        if err or _fault
           if counter >= 0
             counter = 0
-            return callback(err)
+            return callback(err, _fault)
           return
 
         done &&= _done
@@ -441,7 +440,7 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal
               toClear = commitVirtualParameter(sessionData, provisions[i], vpu, toClear)
 
           clear(sessionData, toClear, (err) ->
-            return callback(err, done, allDeclarations, allClear)
+            return callback(err, null, done, allDeclarations, allClear)
           )
       )
 
@@ -452,7 +451,7 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal
         toClear = commitVirtualParameter(sessionData, provisions[i], vpu, toClear)
 
     clear(sessionData, toClear, (err) ->
-      return callback(err, done, allDeclarations, allClear)
+      return callback(err, null, done, allDeclarations, allClear)
     )
 
 
@@ -586,7 +585,7 @@ runDeclarations = (sessionData, declarations) ->
 
 rpcRequest = (sessionData, _declarations, callback) ->
   if sessionData.rpcRequest?
-    return callback(null, generateRpcId(sessionData), sessionData.rpcRequest)
+    return callback(null, null, generateRpcId(sessionData), sessionData.rpcRequest)
 
   if sessionData.declarations.length < sessionData.provisions.length
     inception = sessionData.declarations.length
@@ -595,9 +594,33 @@ rpcRequest = (sessionData, _declarations, callback) ->
     sessionData.deviceData.attributes.revision = revision
     run = if inception == 0 then runProvisions else runVirtualParameters
 
-    return run(sessionData, sessionData.provisions[inception], sessionData.revisions[inception - 1] ? 0, sessionData.revisions[inception], (err, done, decs, toClear) ->
+    return run(sessionData, sessionData.provisions[inception], sessionData.revisions[inception - 1] ? 0, sessionData.revisions[inception], (err, fault, done, decs, toClear) ->
       return callback(err) if err
 
+      if fault
+        faults = {}
+        for p, i in sessionData.provisions[0]
+          channel = sessionData.channels[i]
+          f = faults[channel]
+          if not f?
+            f = {
+              provisions: []
+              timestamp: sessionData.timestamp
+              code: fault.code
+              message: fault.message
+              detail: fault.detail
+            }
+
+            if sessionData.faults[channel]?
+              f.retries = (sessionData.faults[channel].retries or 0) + 1
+
+            faults[channel] = f
+
+          f.provisions.push(p)
+
+        clearProvisions(sessionData)
+        return callback(null, faults)
+
       sessionData.declarations.push(decs)
       sessionData.doneProvisions |= 1 << inception if done or not decs.length
 
@@ -777,7 +800,7 @@ rpcRequest = (sessionData, _declarations, callback) ->
     return rpcRequest(sessionData, null, callback)
 
   if sessionData.rpcRequest
-    return callback(null, generateRpcId(sessionData), sessionData.rpcRequest)
+    return callback(null, null, generateRpcId(sessionData), sessionData.rpcRequest)
 
   ++ sessionData.revisions[inception]
   sessionData.declarations.pop()
@@ -791,7 +814,7 @@ rpcRequest = (sessionData, _declarations, callback) ->
       sessionData.deviceData.timestamps.collapse(1)
       sessionData.deviceData.attributes.collapse(1)
       sessionData.extensionsCache = {}
-      return callback(null, null, null, doneProvisions)
+      return callback(null, null, null, null, doneProvisions)
 
     rev = sessionData.revisions[sessionData.revisions.length - 1]
     sessionData.deviceData.timestamps.collapse(rev + 1)