diff --git a/lib/workers/index.js b/lib/workers/index.js index 7ddbf93cf..5bdbd5d7a 100644 --- a/lib/workers/index.js +++ b/lib/workers/index.js @@ -152,7 +152,18 @@ Workers.initialize = function (Env, config, _cb) { // default to timing out affter 180s if no explicit timeout is passed var timeout = typeof(opt.timeout) !== 'undefined'? opt.timeout: DEFAULT_QUERY_TIMEOUT; response.expect(txid, cb, timeout); + + delete msg._cb; + delete msg._opt; state.worker.send(msg); + + // Add original callback to message data in case we need + // to resend the command. setTimeout to avoid interfering + // with worker.send + setTimeout(function () { + msg._cb = _cb; + msg._opt = opt; + }); }; const pluginsResponses = {}; @@ -270,26 +281,36 @@ Workers.initialize = function (Env, config, _cb) { const cb = response.expectation(txid); if (typeof(cb) !== 'function') { return; } const task = state.tasks[txid]; - if (!(task && task.msg)) { return; } + if (!task) { return; } response.clear(txid); - Log.info('DB_WORKER_RESEND', task.msg); - sendCommand(task.msg, cb); + Log.info('DB_WORKER_RESEND', task); + sendCommand(task, task._cb || cb, task._opt); }); var w = fork(DB_PATH); - initWorker(w, function (err, state) { + initWorker(w, function (err) { if (err) { throw new Error(err); } - workers.push(state); }); }); - worker.on('exit', substituteWorker); - worker.on('close', substituteWorker); + worker.on('exit', function () { + substituteWorker(); + Env.Log.error("DB_WORKER_EXIT", { + pid: state.pid, + }); + }); + worker.on('close', function () { + substituteWorker(); + Env.Log.error("DB_WORKER_CLOSE", { + pid: state.pid, + }); + }); worker.on('error', function (err) { substituteWorker(); Env.Log.error("DB_WORKER_ERROR", { + pid: state.pid, error: err, }); });