web/cryptpad
2
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'soon' into staging

Browse Source
This commit is contained in:
ansuz
2020-03-27 19:59:45 -04:00
parent 7047abe407 6d0dee979a
commit db9eaa8d4f
14 changed files with 904 additions and 544 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
lib/api.js
View File

@@ -17,10 +17,11 @@ module.exports.create = function (config) {
.on('sessionClose', historyKeeper.sessionClose)
.on('error', function (error, label, info) {
if (!error) { return; }
if (['EPIPE', 'ECONNRESET'].indexOf(error && error.code) !== -1) { return; }
/* labels:
SEND_MESSAGE_FAIL, SEND_MESSAGE_FAIL_2, FAIL_TO_DISCONNECT,
FAIL_TO_TERMINATE, HANDLE_CHANNEL_LEAVE, NETFLUX_BAD_MESSAGE,
NETFLUX_WEBSOCKET_ERROR
NETFLUX_WEBSOCKET_ERROR, NF_ENOENT
*/
log.error(label, {
code: error.code,

21
lib/commands/admin-rpc.js
View File

@@ -19,6 +19,26 @@ var getFileDescriptorLimit = function (env, server, cb) {
Ulimit(cb);
};
var getCacheStats = function (env, server, cb) {
var metaCount = 0;
var channelCount = 0;
var meta = env.metadata_cache;
for (var x in meta) {
if (meta.hasOwnProperty(x)) { metaCount++; }
}
var channels = env.channel_cache;
for (var y in channels) {
if (channels.hasOwnProperty(y)) { channelCount++; }
}
cb(void 0, {
metadata: metaCount,
channel: channelCount,
});
};
var getActiveSessions = function (Env, Server, cb) {
var stats = Server.getSessionStats();
cb(void 0, [
@@ -137,6 +157,7 @@ var commands = {
GET_FILE_DESCRIPTOR_COUNT: getFileDescriptorCount,
GET_FILE_DESCRIPTOR_LIMIT: getFileDescriptorLimit,
SET_DEFAULT_STORAGE_LIMIT: setDefaultStorageLimit,
GET_CACHE_STATS: getCacheStats,
};
Admin.command = function (Env, safeKey, data, _cb, Server) {

38
lib/commands/channel.js
View File

@@ -61,43 +61,7 @@ Channel.removeOwnedChannel = function (Env, safeKey, channelId, cb, Server) {
var unsafeKey = Util.unescapeKeyCharacters(safeKey);
if (Env.blobStore.isFileId(channelId)) {
var blobId = channelId;
return void nThen(function (w) {
// check if you have permissions
Env.blobStore.isOwnedBy(safeKey, blobId, w(function (err, owned) {
if (err || !owned) {
w.abort();
return void cb("INSUFFICIENT_PERMISSIONS");
}
}));
}).nThen(function (w) {
// remove the blob
return void Env.blobStore.archive.blob(blobId, w(function (err) {
Env.Log.info('ARCHIVAL_OWNED_FILE_BY_OWNER_RPC', {
safeKey: safeKey,
blobId: blobId,
status: err? String(err): 'SUCCESS',
});
if (err) {
w.abort();
return void cb(err);
}
}));
}).nThen(function () {
// archive the proof
return void Env.blobStore.archive.proof(safeKey, blobId, function (err) {
Env.Log.info("ARCHIVAL_PROOF_REMOVAL_BY_OWNER_RPC", {
safeKey: safeKey,
blobId: blobId,
status: err? String(err): 'SUCCESS',
});
if (err) {
return void cb("E_PROOF_REMOVAL");
}
cb(void 0, 'OK');
});
});
return void Env.removeOwnedBlob(channelId, safeKey, cb);
}
Metadata.getMetadata(Env, channelId, function (err, metadata) {

1
lib/commands/pin-rpc.js
View File

@@ -2,7 +2,6 @@
const Core = require("./core");
const Pinning = module.exports;
const Nacl = require("tweetnacl/nacl-fast");
const Util = require("../common-util");
const nThen = require("nthen");

40
lib/historyKeeper.js
View File

@@ -10,6 +10,7 @@ const Core = require("./commands/core");
const Store = require("./storage/file");
const BlobStore = require("./storage/blob");
const Workers = require("./workers/index");
module.exports.create = function (config, cb) {
const Log = config.log;
@@ -78,8 +79,6 @@ module.exports.create = function (config, cb) {
domain: config.domain
};
HK.initializeValidationWorkers(Env);
(function () {
var pes = config.premiumUploadSize;
if (!isNaN(pes) && pes >= Env.maxUploadSize) {
@@ -190,7 +189,7 @@ module.exports.create = function (config, cb) {
},
sessionClose: function (userId, reason) {
HK.closeNetfluxSession(Env, userId);
if (['BAD_MESSAGE', 'SOCKET_ERROR', 'SEND_MESSAGE_FAIL_2'].indexOf(reason) !== -1) {
if (['BAD_MESSAGE', 'SEND_MESSAGE_FAIL_2'].indexOf(reason) !== -1) {
if (reason && reason.code === 'ECONNRESET') { return; }
return void Log.error('SESSION_CLOSE_WITH_ERROR', {
userId: userId,
@@ -243,9 +242,10 @@ module.exports.create = function (config, cb) {
Env.blobStore = blob;
}));
}).nThen(function (w) {
HK.initializeIndexWorkers(Env, {
Workers.initialize(Env, {
blobPath: config.blobPath,
blobStagingPath: config.blobStagingPath,
taskPath: config.taskPath,
pinPath: pinPath,
filePath: config.filePath,
archivePath: config.archivePath,
@@ -258,25 +258,25 @@ module.exports.create = function (config, cb) {
}
}));
}).nThen(function (w) {
// create a task store
// create a task store (for scheduling tasks)
require("./storage/tasks").create(config, w(function (e, tasks) {
if (e) {
throw e;
}
if (e) { throw e; }
Env.tasks = tasks;
config.tasks = tasks;
if (config.disableIntegratedTasks) { return; }
config.intervals = config.intervals || {};
config.intervals.taskExpiration = setInterval(function () {
tasks.runAll(function (err) {
if (err) {
// either TASK_CONCURRENCY or an error with tasks.list
// in either case it is already logged.
}
});
}, 1000 * 60 * 5); // run every five minutes
}));
if (config.disableIntegratedTasks) { return; }
config.intervals = config.intervals || {};
var tasks_running;
config.intervals.taskExpiration = setInterval(function () {
if (tasks_running) { return; }
tasks_running = true;
Env.runTasks(function (err) {
if (err) {
Log.error('TASK_RUNNER_ERR', err);
}
tasks_running = false;
});
}, 1000 * 60 * 5); // run every five minutes
}).nThen(function () {
RPC.create(Env, function (err, _rpc) {
if (err) { throw err; }

254
lib/hk-util.js
View File

@@ -6,10 +6,6 @@ const nThen = require('nthen');
const Util = require("./common-util");
const MetaRPC = require("./commands/metadata");
const Nacl = require('tweetnacl/nacl-fast');
const { fork } = require('child_process');
const OS = require("os");
const numCPUs = OS.cpus().length;
const now = function () { return (new Date()).getTime(); };
const ONE_DAY = 1000 * 60 * 60 * 24; // one day in milliseconds
@@ -43,11 +39,14 @@ const STANDARD_CHANNEL_LENGTH = HK.STANDARD_CHANNEL_LENGTH = 32;
// with a 34 character id
const EPHEMERAL_CHANNEL_LENGTH = HK.EPHEMERAL_CHANNEL_LENGTH = 34;
const tryParse = function (Env, str) {
const tryParse = HK.tryParse = function (Env, str) {
try {
return JSON.parse(str);
} catch (err) {
Env.Log.error('HK_PARSE_ERROR', err);
Env.Log.error('HK_PARSE_ERROR', {
message: err && err.name,
input: str,
});
}
};
@@ -767,249 +766,6 @@ HK.onDirectMessage = function (Env, Server, seq, userId, json) {
});
};
HK.initializeIndexWorkers = function (Env, config, _cb) {
var cb = Util.once(Util.mkAsync(_cb));
const workers = [];
const response = Util.response();
const initWorker = function (worker, cb) {
//console.log("initializing index worker");
const txid = Util.uid();
response.expect(txid, function (err) {
if (err) { return void cb(err); }
//console.log("worker initialized");
workers.push(worker);
cb();
}, 15000);
worker.send({
txid: txid,
config: config,
});
worker.on('message', function (res) {
if (!res) { return; }
if (!res.txid) {
// !report errors...
if (res.error) {
Env.Log.error(res.error, res.value);
}
return;
}
//console.log(res);
try {
response.handle(res.txid, [res.error, res.value]);
} catch (err) {
Env.Log.error("INDEX_WORKER", {
error: err,
response: res,
});
}
});
worker.on('exit', function () {
var idx = workers.indexOf(worker);
if (idx !== -1) {
workers.splice(idx, 1);
}
var w = fork('lib/workers/compute-index');
initWorker(w, function (err) {
if (err) {
throw new Error(err);
}
workers.push(w);
});
});
};
var workerIndex = 0;
var sendCommand = function (msg, _cb) {
var cb = Util.once(Util.mkAsync(_cb));
workerIndex = (workerIndex + 1) % workers.length;
if (workers.length === 0 ||
typeof(workers[workerIndex].send) !== 'function') {
return void cb("NO_WORKERS");
}
const txid = Util.uid();
msg.txid = txid;
response.expect(txid, cb, 45000);
workers[workerIndex].send(msg);
};
nThen(function (w) {
OS.cpus().forEach(function () {
initWorker(fork('lib/workers/compute-index'), w(function (err) {
if (!err) { return; }
w.abort();
return void cb(err);
}));
});
}).nThen(function () {
Env.computeIndex = function (Env, channel, cb) {
Env.store.getWeakLock(channel, function (next) {
sendCommand({
channel: channel,
command: 'COMPUTE_INDEX',
}, function (err, index) {
next();
cb(err, index);
});
});
};
Env.computeMetadata = function (channel, cb) {
Env.store.getWeakLock(channel, function (next) {
sendCommand({
channel: channel,
command: 'COMPUTE_METADATA',
}, function (err, metadata) {
next();
cb(err, metadata);
});
});
};
Env.getOlderHistory = function (channel, oldestKnownHash, cb) {
Env.store.getWeakLock(channel, function (next) {
sendCommand({
channel: channel,
command: "GET_OLDER_HISTORY",
hash: oldestKnownHash,
}, Util.both(next, cb));
});
};
Env.getPinState = function (safeKey, cb) {
Env.pinStore.getWeakLock(safeKey, function (next) {
sendCommand({
key: safeKey,
command: 'GET_PIN_STATE',
}, Util.both(next, cb));
});
};
Env.getFileSize = function (channel, cb) {
sendCommand({
command: 'GET_FILE_SIZE',
channel: channel,
}, cb);
};
Env.getDeletedPads = function (channels, cb) {
sendCommand({
command: "GET_DELETED_PADS",
channels: channels,
}, cb);
};
Env.getTotalSize = function (channels, cb) {
// we could take out locks for all of these channels,
// but it's OK if the size is slightly off
sendCommand({
command: 'GET_TOTAL_SIZE',
channels: channels,
}, cb);
};
Env.getMultipleFileSize = function (channels, cb) {
sendCommand({
command: "GET_MULTIPLE_FILE_SIZE",
channels: channels,
}, cb);
};
Env.getHashOffset = function (channel, hash, cb) {
Env.store.getWeakLock(channel, function (next) {
sendCommand({
command: 'GET_HASH_OFFSET',
channel: channel,
hash: hash,
}, Util.both(next, cb));
});
};
//console.log("index workers ready");
cb(void 0);
});
};
HK.initializeValidationWorkers = function (Env) {
if (typeof(Env.validateMessage) !== 'undefined') {
return void console.error("validation workers are already initialized");
}
// Create our workers
const workers = [];
for (let i = 0; i < numCPUs; i++) {
workers.push(fork('lib/workers/check-signature.js'));
}
const response = Util.response();
var initWorker = function (worker) {
worker.on('message', function (res) {
if (!res || !res.txid) { return; }
//console.log(+new Date(), "Received verification response");
response.handle(res.txid, [res.error, res.value]);
});
// Spawn a new process in one ends
worker.on('exit', function () {
var idx = workers.indexOf(worker);
if (idx !== -1) {
workers.splice(idx, 1);
}
// Spawn a new one
var w = fork('lib/workers/check-signature.js');
workers.push(w);
initWorker(w);
});
};
workers.forEach(initWorker);
var nextWorker = 0;
const send = function (msg, _cb) {
var cb = Util.once(Util.mkAsync(_cb));
// let's be paranoid about asynchrony and only calling back once..
nextWorker = (nextWorker + 1) % workers.length;
if (workers.length === 0 || typeof(workers[nextWorker].send) !== 'function') {
return void cb("INVALID_WORKERS");
}
var txid = msg.txid = Util.uid();
// expect a response within 15s
response.expect(txid, cb, 15000);
// Send the request
workers[nextWorker].send(msg);
};
Env.validateMessage = function (signedMsg, key, cb) {
send({
msg: signedMsg,
key: key,
command: 'INLINE',
}, cb);
};
Env.checkSignature = function (signedMsg, signature, publicKey, cb) {
send({
command: 'DETACHED',
sig: signature,
msg: signedMsg,
key: publicKey,
}, cb);
};
Env.hashChannelList = function (channels, cb) {
send({
command: 'HASH_CHANNEL_LIST',
channels: channels,
}, cb);
};
};
/* onChannelMessage
Determine what we should store when a message a broadcasted to a channel"

2
lib/log.js
View File

@@ -21,7 +21,7 @@ var write = function (ctx, content) {
};
// various degrees of logging
const logLevels = ['silly', 'verbose', 'debug', 'feedback', 'info', 'warn', 'error'];
const logLevels = Logger.levels = ['silly', 'verbose', 'debug', 'feedback', 'info', 'warn', 'error'];
var handlers = {
silly: function (ctx, time, tag, info) {

0
lib/workers/check-signature.js → lib/workers/crypto-worker.js
View File

118
lib/workers/compute-index.js → lib/workers/db-worker.js
View File

@@ -10,8 +10,23 @@ const Meta = require("../metadata");
const Pins = require("../pins");
const Core = require("../commands/core");
const Saferphore = require("saferphore");
const Logger = require("../log");
const Tasks = require("../storage/tasks");
const Env = {};
const Env = {
Log: {},
};
// support the usual log API but pass it to the main process
Logger.levels.forEach(function (level) {
Env.Log[level] = function (label, info) {
process.send({
log: level,
label: label,
info: info,
});
};
});
var ready = false;
var store;
@@ -52,15 +67,23 @@ const init = function (config, _cb) {
}
blobStore = blob;
}));
}).nThen(function (w) {
Tasks.create({
log: Env.Log,
taskPath: config.taskPath,
store: store,
}, w(function (err, tasks) {
if (err) {
w.abort();
return void cb(err);
}
Env.tasks = tasks;
}));
}).nThen(function () {
cb();
});
};
const tryParse = function (Env, str) {
try { return JSON.parse(str); } catch (err) { }
};
/* computeIndex
can call back with an error or a computed index which includes:
* cpIndex:
@@ -107,7 +130,7 @@ const computeIndex = function (data, cb) {
// but only check for metadata on the first line
if (!i && msgObj.buff.indexOf('{') === 0) {
i++; // always increment the message counter
msg = tryParse(Env, msgObj.buff.toString('utf8'));
msg = HK.tryParse(Env, msgObj.buff.toString('utf8'));
if (typeof msg === "undefined") { return readMore(); }
// validate that the current line really is metadata before storing it as such
@@ -116,7 +139,7 @@ const computeIndex = function (data, cb) {
}
i++;
if (msgObj.buff.indexOf('cp|') > -1) {
msg = msg || tryParse(Env, msgObj.buff.toString('utf8'));
msg = msg || HK.tryParse(Env, msgObj.buff.toString('utf8'));
if (typeof msg === "undefined") { return readMore(); }
// cache the offsets of checkpoints if they can be parsed
if (msg[2] === 'MSG' && msg[4].indexOf('cp|') === 0) {
@@ -142,7 +165,7 @@ const computeIndex = function (data, cb) {
// once indexing is complete you should have a buffer of messages since the latest checkpoint
// map the 'hash' of each message to its byte offset in the log, to be used for reconnecting clients
messageBuf.forEach((msgObj) => {
const msg = tryParse(Env, msgObj.buff.toString('utf8'));
const msg = HK.tryParse(Env, msgObj.buff.toString('utf8'));
if (typeof msg === "undefined") { return; }
if (msg[0] === 0 && msg[2] === 'MSG' && typeof(msg[4]) === 'string') {
// msgObj.offset is API guaranteed by our storage module
@@ -166,9 +189,9 @@ const computeIndex = function (data, cb) {
});
};
const computeMetadata = function (data, cb, errorHandler) {
const computeMetadata = function (data, cb) {
const ref = {};
const lineHandler = Meta.createLineHandler(ref, errorHandler);
const lineHandler = Meta.createLineHandler(ref, Env.Log.error);
return void store.readChannelMetadata(data.channel, lineHandler, function (err) {
if (err) {
// stream errors?
@@ -199,7 +222,7 @@ const getOlderHistory = function (data, cb) {
store.getMessages(channelName, function (msgStr) {
if (found) { return; }
let parsed = tryParse(Env, msgStr);
let parsed = HK.tryParse(Env, msgStr);
if (typeof parsed === "undefined") { return; }
// identify classic metadata messages by their inclusion of a channel.
@@ -221,11 +244,11 @@ const getOlderHistory = function (data, cb) {
});
};
const getPinState = function (data, cb, errorHandler) {
const getPinState = function (data, cb) {
const safeKey = data.key;
var ref = {};
var lineHandler = Pins.createLineHandler(ref, errorHandler);
var lineHandler = Pins.createLineHandler(ref, Env.Log.error);
// if channels aren't in memory. load them from disk
// TODO replace with readMessagesBin
@@ -328,7 +351,7 @@ const getHashOffset = function (data, cb) {
var offset = -1;
store.readMessagesBin(channelName, 0, (msgObj, readMore, abort) => {
// tryParse return a parsed message or undefined
const msg = tryParse(Env, msgObj.buff.toString('utf8'));
const msg = HK.tryParse(Env, msgObj.buff.toString('utf8'));
// if it was undefined then go onto the next message
if (typeof msg === "undefined") { return readMore(); }
if (typeof(msg[4]) !== 'string' || lastKnownHash !== HK.getHash(msg[4])) {
@@ -342,6 +365,51 @@ const getHashOffset = function (data, cb) {
});
};
const removeOwnedBlob = function (data, cb) {
const blobId = data.blobId;
const safeKey = data.safeKey;
nThen(function (w) {
// check if you have permissions
blobStore.isOwnedBy(safeKey, blobId, w(function (err, owned) {
if (err || !owned) {
w.abort();
return void cb("INSUFFICIENT_PERMISSIONS");
}
}));
}).nThen(function (w) {
// remove the blob
blobStore.archive.blob(blobId, w(function (err) {
Env.Log.info('ARCHIVAL_OWNED_FILE_BY_OWNER_RPC', {
safeKey: safeKey,
blobId: blobId,
status: err? String(err): 'SUCCESS',
});
if (err) {
w.abort();
return void cb(err);
}
}));
}).nThen(function () {
// archive the proof
blobStore.archive.proof(safeKey, blobId, function (err) {
Env.Log.info("ARCHIVAL_PROOF_REMOVAL_BY_OWNER_RPC", {
safeKey: safeKey,
blobId: blobId,
status: err? String(err): 'SUCCESS',
});
if (err) {
return void cb("E_PROOF_REMOVAL");
}
cb(void 0, 'OK');
});
});
};
const runTasks = function (data, cb) {
Env.tasks.runAll(cb);
};
const COMMANDS = {
COMPUTE_INDEX: computeIndex,
COMPUTE_METADATA: computeMetadata,
@@ -352,12 +420,15 @@ const COMMANDS = {
GET_DELETED_PADS: getDeletedPads,
GET_MULTIPLE_FILE_SIZE: getMultipleFileSize,
GET_HASH_OFFSET: getHashOffset,
REMOVE_OWNED_BLOB: removeOwnedBlob,
RUN_TASKS: runTasks,
};
process.on('message', function (data) {
if (!data || !data.txid) {
if (!data || !data.txid || !data.pid) {
return void process.send({
error:'E_INVAL'
error:'E_INVAL',
data: data,
});
}
@@ -365,6 +436,7 @@ process.on('message', function (data) {
process.send({
error: err,
txid: data.txid,
pid: data.pid,
value: value,
});
};
@@ -381,12 +453,12 @@ process.on('message', function (data) {
if (typeof(command) !== 'function') {
return void cb("E_BAD_COMMAND");
}
command(data, cb, function (label, info) {
// for streaming errors
process.send({
error: label,
value: info,
});
});
command(data, cb);
});
process.on('uncaughtException', function (err) {
console.error('[%s] UNCAUGHT EXCEPTION IN DB WORKER', new Date());
console.error(err);
console.error("TERMINATING");
process.exit(1);
});

339
lib/workers/index.js Normal file
View File

@@ -0,0 +1,339 @@
/* jshint esversion: 6 */
/* global process */
const Util = require("../common-util");
const nThen = require('nthen');
const OS = require("os");
const numCPUs = OS.cpus().length;
const { fork } = require('child_process');
const Workers = module.exports;
const PID = process.pid;
const CRYPTO_PATH = 'lib/workers/crypto-worker';
const DB_PATH = 'lib/workers/db-worker';
Workers.initializeValidationWorkers = function (Env) {
if (typeof(Env.validateMessage) !== 'undefined') {
return void console.error("validation workers are already initialized");
}
// Create our workers
const workers = [];
for (let i = 0; i < numCPUs; i++) {
workers.push(fork(CRYPTO_PATH));
}
const response = Util.response(function (errLabel, info) {
Env.Log.error('HK_VALIDATE_WORKER__' + errLabel, info);
});
var initWorker = function (worker) {
worker.on('message', function (res) {
if (!res || !res.txid) { return; }
response.handle(res.txid, [res.error, res.value]);
});
var substituteWorker = Util.once( function () {
Env.Log.info("SUBSTITUTE_VALIDATION_WORKER", '');
var idx = workers.indexOf(worker);
if (idx !== -1) {
workers.splice(idx, 1);
}
// Spawn a new one
var w = fork(CRYPTO_PATH);
workers.push(w);
initWorker(w);
});
// Spawn a new process in one ends
worker.on('exit', substituteWorker);
worker.on('close', substituteWorker);
worker.on('error', function (err) {
substituteWorker();
Env.Log.error('VALIDATION_WORKER_ERROR', {
error: err,
});
});
};
workers.forEach(initWorker);
var nextWorker = 0;
const send = function (msg, _cb) {
var cb = Util.once(Util.mkAsync(_cb));
// let's be paranoid about asynchrony and only calling back once..
nextWorker = (nextWorker + 1) % workers.length;
if (workers.length === 0 || typeof(workers[nextWorker].send) !== 'function') {
return void cb("INVALID_WORKERS");
}
var txid = msg.txid = Util.uid();
// expect a response within 45s
response.expect(txid, cb, 60000);
// Send the request
workers[nextWorker].send(msg);
};
Env.validateMessage = function (signedMsg, key, cb) {
send({
msg: signedMsg,
key: key,
command: 'INLINE',
}, cb);
};
Env.checkSignature = function (signedMsg, signature, publicKey, cb) {
send({
command: 'DETACHED',
sig: signature,
msg: signedMsg,
key: publicKey,
}, cb);
};
Env.hashChannelList = function (channels, cb) {
send({
command: 'HASH_CHANNEL_LIST',
channels: channels,
}, cb);
};
};
Workers.initializeIndexWorkers = function (Env, config, _cb) {
var cb = Util.once(Util.mkAsync(_cb));
const workers = [];
const response = Util.response(function (errLabel, info) {
Env.Log.error('HK_DB_WORKER__' + errLabel, info);
});
const Log = Env.Log;
const handleLog = function (level, label, info) {
if (typeof(Log[level]) !== 'function') { return; }
Log[level](label, info);
};
var isWorker = function (value) {
return value && value.worker && typeof(value.worker.send) === 'function';
};
// pick ids that aren't already in use...
const guid = function () {
var id = Util.uid();
return response.expected(id)? guid(): id;
};
var workerIndex = 0;
var sendCommand = function (msg, _cb) {
var cb = Util.once(Util.mkAsync(_cb));
workerIndex = (workerIndex + 1) % workers.length;
if (!isWorker(workers[workerIndex])) {
return void cb("NO_WORKERS");
}
var state = workers[workerIndex];
// XXX insert a queue here to prevent timeouts
const txid = guid();
msg.txid = txid;
msg.pid = PID;
// track which worker is doing which jobs
state.tasks[txid] = msg;
response.expect(txid, function (err, value) {
// clean up when you get a response
delete state[txid];
cb(err, value);
}, 60000);
state.worker.send(msg);
};
const initWorker = function (worker, cb) {
const txid = guid();
const state = {
worker: worker,
tasks: {},
};
response.expect(txid, function (err) {
if (err) { return void cb(err); }
workers.push(state);
cb(void 0, state);
}, 15000);
worker.send({
pid: PID,
txid: txid,
config: config,
});
worker.on('message', function (res) {
if (!res) { return; }
// handle log messages before checking if it was addressed to your PID
// it might still be useful to know what happened inside an orphaned worker
if (res.log) {
return void handleLog(res.log, res.label, res.info);
}
// but don't bother handling things addressed to other processes
// since it's basically guaranteed not to work
if (res.pid !== PID) {
return void Log.error("WRONG_PID", res);
}
response.handle(res.txid, [res.error, res.value]);
});
var substituteWorker = Util.once(function () {
Env.Log.info("SUBSTITUTE_DB_WORKER", '');
var idx = workers.indexOf(state);
if (idx !== -1) {
workers.splice(idx, 1);
}
Object.keys(state.tasks).forEach(function (txid) {
const cb = response.expectation(txid);
if (typeof(cb) !== 'function') { return; }
const task = state.tasks[txid];
if (!task && task.msg) { return; }
response.clear(txid);
Log.info('DB_WORKER_RESEND', task.msg);
sendCommand(task.msg, cb);
});
var w = fork(DB_PATH);
initWorker(w, function (err, state) {
if (err) {
throw new Error(err);
}
workers.push(state);
});
});
worker.on('exit', substituteWorker);
worker.on('close', substituteWorker);
worker.on('error', function (err) {
substituteWorker();
Env.Log.error("DB_WORKER_ERROR", {
error: err,
});
});
};
nThen(function (w) {
OS.cpus().forEach(function () {
initWorker(fork(DB_PATH), w(function (err) {
if (!err) { return; }
w.abort();
return void cb(err);
}));
});
}).nThen(function () {
Env.computeIndex = function (Env, channel, cb) {
Env.store.getWeakLock(channel, function (next) {
sendCommand({
channel: channel,
command: 'COMPUTE_INDEX',
}, function (err, index) {
next();
cb(err, index);
});
});
};
Env.computeMetadata = function (channel, cb) {
Env.store.getWeakLock(channel, function (next) {
sendCommand({
channel: channel,
command: 'COMPUTE_METADATA',
}, function (err, metadata) {
next();
cb(err, metadata);
});
});
};
Env.getOlderHistory = function (channel, oldestKnownHash, cb) {
Env.store.getWeakLock(channel, function (next) {
sendCommand({
channel: channel,
command: "GET_OLDER_HISTORY",
hash: oldestKnownHash,
}, Util.both(next, cb));
});
};
Env.getPinState = function (safeKey, cb) {
Env.pinStore.getWeakLock(safeKey, function (next) {
sendCommand({
key: safeKey,
command: 'GET_PIN_STATE',
}, Util.both(next, cb));
});
};
Env.getFileSize = function (channel, cb) {
sendCommand({
command: 'GET_FILE_SIZE',
channel: channel,
}, cb);
};
Env.getDeletedPads = function (channels, cb) {
sendCommand({
command: "GET_DELETED_PADS",
channels: channels,
}, cb);
};
Env.getTotalSize = function (channels, cb) {
// we could take out locks for all of these channels,
// but it's OK if the size is slightly off
sendCommand({
command: 'GET_TOTAL_SIZE',
channels: channels,
}, cb);
};
Env.getMultipleFileSize = function (channels, cb) {
sendCommand({
command: "GET_MULTIPLE_FILE_SIZE",
channels: channels,
}, cb);
};
Env.getHashOffset = function (channel, hash, cb) {
Env.store.getWeakLock(channel, function (next) {
sendCommand({
command: 'GET_HASH_OFFSET',
channel: channel,
hash: hash,
}, Util.both(next, cb));
});
};
Env.removeOwnedBlob = function (blobId, safeKey, cb) {
sendCommand({
command: 'REMOVE_OWNED_BLOB',
blobId: blobId,
safeKey: safeKey,
}, cb);
};
Env.runTasks = function (cb) {
sendCommand({
command: 'RUN_TASKS',
}, cb);
};
cb(void 0);
});
};
Workers.initialize = function (Env, config, cb) {
Workers.initializeValidationWorkers(Env);
Workers.initializeIndexWorkers(Env, config, cb);
};