diff --git a/lib/config.coffee b/lib/config.coffee
index 2137822..40fab86 100644
--- a/lib/config.coffee
+++ b/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},
diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee
index 21d436e..296233d 100644
--- a/lib/cwmp.coffee
+++ b/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)
diff --git a/lib/extension-wrapper.coffee b/lib/extension-wrapper.coffee
deleted file mode 100644
index 5a5afa1..0000000
--- a/lib/extension-wrapper.coffee
+++ /dev/null
@@ -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 .
-###
-
-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])
- )
-)
diff --git a/lib/extension-wrapper.js b/lib/extension-wrapper.js
new file mode 100644
index 0000000..b08ff16
--- /dev/null
+++ b/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 .
+ */
+"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]);
+ });
+});
diff --git a/lib/extensions.coffee b/lib/extensions.coffee
deleted file mode 100644
index 484c1b8..0000000
--- a/lib/extensions.coffee
+++ /dev/null
@@ -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 .
-###
-
-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
diff --git a/lib/extensions.js b/lib/extensions.js
new file mode 100644
index 0000000..e606ec5
--- /dev/null
+++ b/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 .
+ */
+"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;
diff --git a/lib/sandbox.js b/lib/sandbox.js
index 4c3042d..087f0d9 100644
--- a/lib/sandbox.js
+++ b/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) {
- if (err && counter > 0) {
+ extensions.run(sandbox.extensions[key], function(err, fault, res) {
+ 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);
}
diff --git a/lib/session.coffee b/lib/session.coffee
index 7dc2165..22ab9a4 100644
--- a/lib/session.coffee
+++ b/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)