From 80c012f34d018b32eebe21bd03cc721924840cad Mon Sep 17 00:00:00 2001 From: ansuz Date: Mon, 27 Jan 2020 17:57:39 -0500 Subject: [PATCH 1/5] prepare to merge history keeper and rpc --- lib/api.js | 60 +++++++++++++++++++++++ lib/historyKeeper.js | 7 +-- server.js | 114 +++++++++++-------------------------------- 3 files changed, 91 insertions(+), 90 deletions(-) create mode 100644 lib/api.js diff --git a/lib/api.js b/lib/api.js new file mode 100644 index 000000000..60296d9b5 --- /dev/null +++ b/lib/api.js @@ -0,0 +1,60 @@ +/* jshint esversion: 6 */ +const nThen = require("nthen"); +const WebSocketServer = require('ws').Server; +const NetfluxSrv = require('chainpad-server/NetfluxWebsocketSrv'); + +module.exports.create = function (config) { + var historyKeeper; + var rpc; + const log = config.log; + const wsConfig = { + server: config.httpServer, + }; + + nThen(function (w) { + require('../storage/file').create(config, w(function (_store) { + config.store = _store; + })); + }).nThen(function (w) { + require("../storage/tasks").create(config, w(function (e, tasks) { + if (e) { + throw e; + } + config.tasks = tasks; + if (config.disableIntegratedTasks) { return; } + + // XXX support stopping this interval + 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 + })); + }).nThen(function (w) { + require("./rpc").create(config, w(function (e, _rpc) { + if (e) { + w.abort(); + throw e; + } + rpc = _rpc; + })); + }).nThen(function () { + var HK = require('./historyKeeper.js'); + var hkConfig = { + tasks: config.tasks, + rpc: rpc, + store: config.store, + log: log, + }; + // XXX historyKeeper exports a `setConfig` method + historyKeeper = HK.create(hkConfig); + }).nThen(function () { + var wsSrv = new WebSocketServer(wsConfig); + // XXX NetfluxSrv shares some internal functions with historyKeeper + // by passing them to setConfig + NetfluxSrv.run(wsSrv, config, historyKeeper); + }); +}; diff --git a/lib/historyKeeper.js b/lib/historyKeeper.js index c3ed67cb7..55d6b1780 100644 --- a/lib/historyKeeper.js +++ b/lib/historyKeeper.js @@ -1,6 +1,5 @@ /* jshint esversion: 6 */ /* global Buffer */ -;(function () { 'use strict'; const nThen = require('nthen'); const Nacl = require('tweetnacl/nacl-fast'); @@ -63,6 +62,8 @@ const isValidValidateKeyString = function (key) { } }; +var CHECKPOINT_PATTERN = /^cp\|(([A-Za-z0-9+\/=]+)\|)?/; + module.exports.create = function (cfg) { const rpc = cfg.rpc; const tasks = cfg.tasks; @@ -385,8 +386,6 @@ module.exports.create = function (cfg) { return true; }; - var CHECKPOINT_PATTERN = /^cp\|(([A-Za-z0-9+\/=]+)\|)?/; - /* onChannelMessage Determine what we should store when a message a broadcasted to a channel" @@ -992,5 +991,3 @@ module.exports.create = function (cfg) { onDirectMessage: onDirectMessage, }; }; - -}()); diff --git a/server.js b/server.js index d2bdcabd8..70479d7ee 100644 --- a/server.js +++ b/server.js @@ -4,17 +4,12 @@ var Express = require('express'); var Http = require('http'); var Fs = require('fs'); -var WebSocketServer = require('ws').Server; -var NetfluxSrv = require('chainpad-server/NetfluxWebsocketSrv'); var Package = require('./package.json'); var Path = require("path"); var nThen = require("nthen"); var config = require("./lib/load-config"); -// support multiple storage back ends -var Storage = require('./storage/file'); - var app = Express(); // mode can be FRESH (default), DEV, or PACKAGE @@ -69,11 +64,9 @@ var setHeaders = (function () { if (Object.keys(headers).length) { return function (req, res) { const h = [ - /^\/pad(2)?\/inner\.html.*/, + /^\/pad\/inner\.html.*/, /^\/common\/onlyoffice\/.*\/index\.html.*/, - /^\/sheet\/inner\.html.*/, - /^\/ooslide\/inner\.html.*/, - /^\/oodoc\/inner\.html.*/, + /^\/(sheet|ooslide|oodoc)\/inner\.html.*/, ].some((regex) => { return regex.test(req.url) }) ? padHeaders : headers; @@ -117,11 +110,6 @@ app.use(function (req, res, next) { app.use(Express.static(__dirname + '/www')); -Fs.exists(__dirname + "/customize", function (e) { - if (e) { return; } - console.log("Cryptpad is customizable, see customize.dist/readme.md for details"); -}); - // FIXME I think this is a regression caused by a recent PR // correct this hack without breaking the contributor's intended behaviour. @@ -207,80 +195,36 @@ app.use(function (req, res, next) { var httpServer = Http.createServer(app); -httpServer.listen(config.httpPort,config.httpAddress,function(){ - var host = config.httpAddress; - var hostName = !host.indexOf(':') ? '[' + host + ']' : host; - - var port = config.httpPort; - var ps = port === 80? '': ':' + port; - - console.log('[%s] server available http://%s%s', new Date().toISOString(), hostName, ps); -}); -if (config.httpSafePort) { - Http.createServer(app).listen(config.httpSafePort, config.httpAddress); -} - -var wsConfig = { server: httpServer }; +nThen(function (w) { + Fs.exists(__dirname + "/customize", w(function (e) { + if (e) { return; } + console.log("Cryptpad is customizable, see customize.dist/readme.md for details"); + })); +}).nThen(function (w) { + httpServer.listen(config.httpPort,config.httpAddress,function(){ + var host = config.httpAddress; + var hostName = !host.indexOf(':') ? '[' + host + ']' : host; -var rpc; -var historyKeeper; + var port = config.httpPort; + var ps = port === 80? '': ':' + port; -var log; + console.log('[%s] server available http://%s%s', new Date().toISOString(), hostName, ps); + }); -// Initialize logging, the the store, then tasks, then rpc, then history keeper and then start the server -var nt = nThen(function (w) { - // set up logger - var Logger = require("./lib/log"); - //console.log("Loading logging module"); - Logger.create(config, w(function (_log) { - log = config.log = _log; - })); -}).nThen(function (w) { - if (config.externalWebsocketURL) { - // if you plan to use an external websocket server - // then you don't need to load any API services other than the logger. - // Just abort. - w.abort(); - return; + if (config.httpSafePort) { + Http.createServer(app).listen(config.httpSafePort, config.httpAddress, w()); } - Storage.create(config, w(function (_store) { - config.store = _store; - })); -}).nThen(function (w) { - var Tasks = require("./storage/tasks"); - Tasks.create(config, w(function (e, tasks) { - if (e) { - throw e; - } - config.tasks = tasks; - if (config.disableIntegratedTasks) { return; } - 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 - })); -}).nThen(function (w) { - require("./lib/rpc").create(config, w(function (e, _rpc) { - if (e) { - w.abort(); - throw e; - } - rpc = _rpc; - })); }).nThen(function () { - var HK = require('./lib/historyKeeper.js'); - var hkConfig = { - tasks: config.tasks, - rpc: rpc, - store: config.store, - log: log, - }; - historyKeeper = HK.create(hkConfig); -}).nThen(function () { - var wsSrv = new WebSocketServer(wsConfig); - NetfluxSrv.run(wsSrv, config, historyKeeper); + var wsConfig = { server: httpServer }; + + // Initialize logging then start the API server + require("./lib/log").create(config, function (_log) { + config.log = _log; + config.httpServer = httpServer; + + if (config.externalWebsocketURL) { return; } + require("./lib/api").create(config); + }); }); + + From b922860339e92f5753edbfba428228f692261914 Mon Sep 17 00:00:00 2001 From: ansuz Date: Mon, 27 Jan 2020 18:54:16 -0500 Subject: [PATCH 2/5] drop usage of historyKeeper.setConfig --- lib/api.js | 7 ++--- lib/historyKeeper.js | 69 ++++++++++++++++++++++++-------------------- lib/hk-util.js | 9 ++++++ 3 files changed, 49 insertions(+), 36 deletions(-) diff --git a/lib/api.js b/lib/api.js index 60296d9b5..d633db59a 100644 --- a/lib/api.js +++ b/lib/api.js @@ -4,7 +4,6 @@ const WebSocketServer = require('ws').Server; const NetfluxSrv = require('chainpad-server/NetfluxWebsocketSrv'); module.exports.create = function (config) { - var historyKeeper; var rpc; const log = config.log; const wsConfig = { @@ -50,11 +49,9 @@ module.exports.create = function (config) { log: log, }; // XXX historyKeeper exports a `setConfig` method - historyKeeper = HK.create(hkConfig); - }).nThen(function () { + var wsSrv = new WebSocketServer(wsConfig); - // XXX NetfluxSrv shares some internal functions with historyKeeper - // by passing them to setConfig + var historyKeeper = HK.create(hkConfig); NetfluxSrv.run(wsSrv, config, historyKeeper); }); }; diff --git a/lib/historyKeeper.js b/lib/historyKeeper.js index 55d6b1780..f8d3c8a15 100644 --- a/lib/historyKeeper.js +++ b/lib/historyKeeper.js @@ -10,6 +10,8 @@ const WriteQueue = require("./write-queue"); const BatchRead = require("./batch-read"); const Extras = require("./hk-util.js"); +const STANDARD_CHANNEL_LENGTH = Extras.STANDARD_CHANNEL_LENGTH; +const EPHEMERAL_CHANNEL_LENGTH = Extras.EPHEMERAL_CHANNEL_LENGTH; let Log; const now = function () { return (new Date()).getTime(); }; @@ -77,14 +79,6 @@ module.exports.create = function (cfg) { Log.verbose('HK_ID', 'History keeper ID: ' + HISTORY_KEEPER_ID); - let sendMsg = function () {}; - let STANDARD_CHANNEL_LENGTH, EPHEMERAL_CHANNEL_LENGTH; - const setConfig = function (config) { - STANDARD_CHANNEL_LENGTH = config.STANDARD_CHANNEL_LENGTH; - EPHEMERAL_CHANNEL_LENGTH = config.EPHEMERAL_CHANNEL_LENGTH; - sendMsg = config.sendMsg; - }; - /* computeIndex can call back with an error or a computed index which includes: * cpIndex: @@ -326,7 +320,7 @@ module.exports.create = function (cfg) { const historyKeeperBroadcast = function (ctx, channel, msg) { let chan = ctx.channels[channel] || (([] /*:any*/) /*:Chan_t*/); chan.forEach(function (user) { - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)]); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)]); }); }; @@ -707,7 +701,7 @@ module.exports.create = function (cfg) { // parsed[1] is the channel id // parsed[2] is a validation key or an object containing metadata (optionnal) // parsed[3] is the last known hash (optionnal) - sendMsg(ctx, user, [seq, 'ACK']); + ctx.sendMsg(ctx, user, [seq, 'ACK']); var channelName = parsed[1]; var config = parsed[2]; var metadata = {}; @@ -758,7 +752,7 @@ module.exports.create = function (cfg) { // FIXME this is hard to read because 'checkExpired' has side effects if (checkExpired(ctx, channelName)) { return void waitFor.abort(); } // always send metadata with GET_HISTORY requests - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(index.metadata)], w); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(index.metadata)], w); })); }).nThen(() => { let msgCount = 0; @@ -769,12 +763,12 @@ module.exports.create = function (cfg) { msgCount++; // avoid sending the metadata message a second time if (isMetadataMessage(msg) && metadata_cache[channelName]) { return readMore(); } - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)], readMore); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)], readMore); }, (err) => { if (err && err.code !== 'ENOENT') { if (err.message !== 'EINVAL') { Log.error("HK_GET_HISTORY", err); } const parsedMsg = {error:err.message, channel: channelName}; - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); return; } @@ -817,12 +811,12 @@ module.exports.create = function (cfg) { } }); } - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(metadata)]); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(metadata)]); } // End of history message: let parsedMsg = {state: 1, channel: channelName}; - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); }); }); }; @@ -831,7 +825,7 @@ module.exports.create = function (cfg) { var channelName = parsed[1]; var map = parsed[2]; if (!(map && typeof(map) === 'object')) { - return void sendMsg(ctx, user, [seq, 'ERROR', 'INVALID_ARGS', HISTORY_KEEPER_ID]); + return void ctx.sendMsg(ctx, user, [seq, 'ERROR', 'INVALID_ARGS', HISTORY_KEEPER_ID]); } var oldestKnownHash = map.from; @@ -839,14 +833,14 @@ module.exports.create = function (cfg) { var desiredCheckpoint = map.cpCount; var txid = map.txid; if (typeof(desiredMessages) !== 'number' && typeof(desiredCheckpoint) !== 'number') { - return void sendMsg(ctx, user, [seq, 'ERROR', 'UNSPECIFIED_COUNT', HISTORY_KEEPER_ID]); + return void ctx.sendMsg(ctx, user, [seq, 'ERROR', 'UNSPECIFIED_COUNT', HISTORY_KEEPER_ID]); } if (!txid) { - return void sendMsg(ctx, user, [seq, 'ERROR', 'NO_TXID', HISTORY_KEEPER_ID]); + return void ctx.sendMsg(ctx, user, [seq, 'ERROR', 'NO_TXID', HISTORY_KEEPER_ID]); } - sendMsg(ctx, user, [seq, 'ACK']); + ctx.sendMsg(ctx, user, [seq, 'ACK']); return void getOlderHistory(channelName, oldestKnownHash, function (messages) { var toSend = []; if (typeof (desiredMessages) === "number") { @@ -862,11 +856,11 @@ module.exports.create = function (cfg) { } } toSend.forEach(function (msg) { - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['HISTORY_RANGE', txid, msg])]); }); - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['HISTORY_RANGE_END', txid, channelName]) ]); }); @@ -876,20 +870,20 @@ module.exports.create = function (cfg) { // parsed[1] is the channel id // parsed[2] is a validation key (optionnal) // parsed[3] is the last known hash (optionnal) - sendMsg(ctx, user, [seq, 'ACK']); + ctx.sendMsg(ctx, user, [seq, 'ACK']); // FIXME should we send metadata here too? // none of the clientside code which uses this API needs metadata, but it won't hurt to send it (2019-08-22) return void getHistoryAsync(ctx, parsed[1], -1, false, (msg, readMore) => { if (!msg) { return; } - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['FULL_HISTORY', msg])], readMore); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['FULL_HISTORY', msg])], readMore); }, (err) => { let parsedMsg = ['FULL_HISTORY_END', parsed[1]]; if (err) { Log.error('HK_GET_FULL_HISTORY', err.stack); parsedMsg = ['ERROR', parsed[1], err.message]; } - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); }); }; @@ -899,12 +893,12 @@ module.exports.create = function (cfg) { /* RPC Calls... */ var rpc_call = parsed.slice(1); - sendMsg(ctx, user, [seq, 'ACK']); + ctx.sendMsg(ctx, user, [seq, 'ACK']); try { // slice off the sequence number and pass in the rest of the message rpc(ctx, rpc_call, function (err, output) { if (err) { - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', err])]); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', err])]); return; } var msg = rpc_call[0].slice(); @@ -930,7 +924,7 @@ module.exports.create = function (cfg) { let chan = ctx.channels[output.channel]; if (chan && chan.length) { chan.forEach(function (user) { - sendMsg(ctx, user, output.message); + ctx.sendMsg(ctx, user, output.message); //[0, null, 'MSG', user.id, JSON.stringify(output.message)]); }); } @@ -940,10 +934,10 @@ module.exports.create = function (cfg) { } // finally, send a response to the client that sent the RPC - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0]].concat(output))]); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0]].concat(output))]); }); } catch (e) { - sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', 'SERVER_ERROR'])]); + ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', 'SERVER_ERROR'])]); } }; @@ -982,12 +976,25 @@ module.exports.create = function (cfg) { command(ctx, seq, user, parsed); }; + // XXX every one of these values is exported because + // netfluxWebsocketServer needs them to do some magic historyKeeper things + // we could have netflux emit events and let historyKeeper handle them instead return { id: HISTORY_KEEPER_ID, - setConfig: setConfig, - onChannelMessage: onChannelMessage, + + // XXX dropChannel allows netflux to clear historyKeeper's cache + // maybe it should emit a 'channel_dropped' event instead + // and let historyKeeper decide what to do dropChannel: dropChannel, + + // XXX we don't need to export checkExpired if netflux allows it to be HK's responsibility checkExpired: checkExpired, + + // XXX again, if netflux emitted events then historyKeeper could handle them itself + // and netflux wouldn't need to have historyKeeper-specific code onDirectMessage: onDirectMessage, + + // XXX same + onChannelMessage: onChannelMessage, }; }; diff --git a/lib/hk-util.js b/lib/hk-util.js index cea39c4e1..aaa861054 100644 --- a/lib/hk-util.js +++ b/lib/hk-util.js @@ -22,3 +22,12 @@ HK.getHash = function (msg, Log) { return msg.slice(0,64); }; +// historyKeeper should explicitly store any channel +// with a 32 character id +HK.STANDARD_CHANNEL_LENGTH = 32; + +// historyKeeper should not store messages sent to any channel +// with a 34 character id +HK.EPHEMERAL_CHANNEL_LENGTH = 34; + + From 06c29ef1d19d46121827c3bc111e6f84db49b750 Mon Sep 17 00:00:00 2001 From: ansuz Date: Mon, 3 Feb 2020 10:03:43 -0500 Subject: [PATCH 3/5] latest api changes to match the netflux-server refactor --- lib/api.js | 17 +++++++++++++---- lib/historyKeeper.js | 35 +++++++++++++++++------------------ 2 files changed, 30 insertions(+), 22 deletions(-) diff --git a/lib/api.js b/lib/api.js index d633db59a..c6491d5db 100644 --- a/lib/api.js +++ b/lib/api.js @@ -1,7 +1,7 @@ /* jshint esversion: 6 */ const nThen = require("nthen"); const WebSocketServer = require('ws').Server; -const NetfluxSrv = require('chainpad-server/NetfluxWebsocketSrv'); +const NetfluxSrv = require('chainpad-server'); module.exports.create = function (config) { var rpc; @@ -48,10 +48,19 @@ module.exports.create = function (config) { store: config.store, log: log, }; - // XXX historyKeeper exports a `setConfig` method - var wsSrv = new WebSocketServer(wsConfig); var historyKeeper = HK.create(hkConfig); - NetfluxSrv.run(wsSrv, config, historyKeeper); + + NetfluxSrv.create(new WebSocketServer(wsConfig)) + .on('channelClose', historyKeeper.channelClose) + .on('channelMessage', historyKeeper.channelMessage) + .on('channelOpen', historyKeeper.channelOpen) + .on('sessionClose', function (userId, reason) { + reason = reason; // XXX + }) + .on('error', function (error, label, info) { + info = info; // XXX + }) + .register(historyKeeper.id, historyKeeper.directMessage); }); }; diff --git a/lib/historyKeeper.js b/lib/historyKeeper.js index f8d3c8a15..92eb84a5c 100644 --- a/lib/historyKeeper.js +++ b/lib/historyKeeper.js @@ -976,25 +976,24 @@ module.exports.create = function (cfg) { command(ctx, seq, user, parsed); }; - // XXX every one of these values is exported because - // netfluxWebsocketServer needs them to do some magic historyKeeper things - // we could have netflux emit events and let historyKeeper handle them instead return { id: HISTORY_KEEPER_ID, - - // XXX dropChannel allows netflux to clear historyKeeper's cache - // maybe it should emit a 'channel_dropped' event instead - // and let historyKeeper decide what to do - dropChannel: dropChannel, - - // XXX we don't need to export checkExpired if netflux allows it to be HK's responsibility - checkExpired: checkExpired, - - // XXX again, if netflux emitted events then historyKeeper could handle them itself - // and netflux wouldn't need to have historyKeeper-specific code - onDirectMessage: onDirectMessage, - - // XXX same - onChannelMessage: onChannelMessage, + channelMessage: function (ctx, channel, msgStruct) { + onChannelMessage(ctx, channel, msgStruct); + }, + channelClose: function (channelName) { + dropChannel(channelName); + }, + channelOpen: function (ctx, channelName, user) { + ctx.sendMsg(ctx, user, [ + 0, + HISTORY_KEEPER_ID, // ctx.historyKeeper.id + 'JOIN', + channelName + ]); + }, + directMessage: function (ctx, seq, user, json) { + onDirectMessage(ctx, seq, user, json); + }, }; }; From 779e8174432fe0c0a361c1cc5991fc3a4365cf97 Mon Sep 17 00:00:00 2001 From: ansuz Date: Mon, 3 Feb 2020 14:20:05 -0500 Subject: [PATCH 4/5] stop relying on netflux-server internals * create RPC module from inside historyKeeper * stop passing around netflux-server context * update to use newer netflux-server's formal APIs * manage your own cache of indexes instead of storing things in the netflux context --- lib/api.js | 66 ++++++----- lib/commands/admin-rpc.js | 35 ++---- lib/commands/channel.js | 7 +- lib/historyKeeper.js | 236 ++++++++++++++++++++------------------ lib/rpc.js | 38 ++---- 5 files changed, 185 insertions(+), 197 deletions(-) diff --git a/lib/api.js b/lib/api.js index c6491d5db..b66dcb4de 100644 --- a/lib/api.js +++ b/lib/api.js @@ -4,8 +4,6 @@ const WebSocketServer = require('ws').Server; const NetfluxSrv = require('chainpad-server'); module.exports.create = function (config) { - var rpc; - const log = config.log; const wsConfig = { server: config.httpServer, }; @@ -32,35 +30,45 @@ module.exports.create = function (config) { }); }, 1000 * 60 * 5); // run every five minutes })); - }).nThen(function (w) { - require("./rpc").create(config, w(function (e, _rpc) { - if (e) { - w.abort(); - throw e; - } - rpc = _rpc; - })); }).nThen(function () { - var HK = require('./historyKeeper.js'); - var hkConfig = { - tasks: config.tasks, - rpc: rpc, - store: config.store, - log: log, - }; + // asynchronously create a historyKeeper and RPC together + require('./historyKeeper.js').create(config, function (err, historyKeeper) { + if (err) { throw err; } - var historyKeeper = HK.create(hkConfig); + var log = config.log; - NetfluxSrv.create(new WebSocketServer(wsConfig)) - .on('channelClose', historyKeeper.channelClose) - .on('channelMessage', historyKeeper.channelMessage) - .on('channelOpen', historyKeeper.channelOpen) - .on('sessionClose', function (userId, reason) { - reason = reason; // XXX - }) - .on('error', function (error, label, info) { - info = info; // XXX - }) - .register(historyKeeper.id, historyKeeper.directMessage); + // spawn ws server and attach netflux event handlers + NetfluxSrv.create(new WebSocketServer(wsConfig)) + .on('channelClose', historyKeeper.channelClose) + .on('channelMessage', historyKeeper.channelMessage) + .on('channelOpen', historyKeeper.channelOpen) + .on('sessionClose', function (userId, reason) { + if (['BAD_MESSAGE', 'SOCKET_ERROR', 'SEND_MESSAGE_FAIL_2'].indexOf(reason) !== -1) { + return void log.error('SESSION_CLOSE_WITH_ERROR', { + userId: userId, + reason: reason, + }); + } + log.verbose('SESSION_CLOSE_ROUTINE', { + userId: userId, + reason: reason, + }); + }) + .on('error', function (error, label, info) { + if (!error) { return; } + /* labels: + SEND_MESSAGE_FAIL, SEND_MESSAGE_FAIL_2, FAIL_TO_DISCONNECT, + FAIL_TO_TERMINATE, HANDLE_CHANNEL_LEAVE, NETFLUX_BAD_MESSAGE, + NETFLUX_WEBSOCKET_ERROR + */ + log.error(label, { + code: error.code, + message: error.message, + stack: error.stack, + info: info, + }); + }) + .register(historyKeeper.id, historyKeeper.directMessage); + }); }); }; diff --git a/lib/commands/admin-rpc.js b/lib/commands/admin-rpc.js index 24a8e4348..763bfb517 100644 --- a/lib/commands/admin-rpc.js +++ b/lib/commands/admin-rpc.js @@ -6,25 +6,15 @@ var Fs = require("fs"); var Admin = module.exports; -var getActiveSessions = function (Env, ctx, cb) { - var total = ctx.users ? Object.keys(ctx.users).length : '?'; - - var ips = []; - Object.keys(ctx.users).forEach(function (u) { - var user = ctx.users[u]; - var socket = user.socket; - var req = socket.upgradeReq; - var conn = req && req.connection; - var ip = (req && req.headers && req.headers['x-forwarded-for']) || (conn && conn.remoteAddress); - if (ip && ips.indexOf(ip) === -1) { - ips.push(ip); - } - }); - - cb (void 0, [total, ips.length]); +var getActiveSessions = function (Env, Server, cb) { + var stats = Server.getSessionStats(); + cb(void 0, [ + stats.total, + stats.unique + ]); }; -var shutdown = function (Env, ctx, cb) { +var shutdown = function (Env, Server, cb) { return void cb('E_NOT_IMPLEMENTED'); //clearInterval(Env.sessionExpirationInterval); // XXX set a flag to prevent incoming database writes @@ -91,19 +81,18 @@ var getDiskUsage = function (Env, cb) { }); }; - - -Admin.command = function (Env, ctx, publicKey, data, cb) { +Admin.command = function (Env, Server, publicKey, data, cb) { var admins = Env.admins; if (admins.indexOf(publicKey) === -1) { return void cb("FORBIDDEN"); } + // Handle commands here switch (data[0]) { case 'ACTIVE_SESSIONS': - return getActiveSessions(Env, ctx, cb); + return getActiveSessions(Env, Server, cb); case 'ACTIVE_PADS': - return cb(void 0, ctx.channels ? Object.keys(ctx.channels).length : '?'); + return cb(void 0, Server.getActiveChannelCount()); case 'REGISTERED_USERS': return getRegisteredUsers(Env, cb); case 'DISK_USAGE': @@ -112,7 +101,7 @@ Admin.command = function (Env, ctx, publicKey, data, cb) { Env.flushCache(); return cb(void 0, true); case 'SHUTDOWN': - return shutdown(Env, ctx, cb); + return shutdown(Env, Server, cb); default: return cb('UNHANDLED_ADMIN_COMMAND'); } diff --git a/lib/commands/channel.js b/lib/commands/channel.js index 052aa3c44..f232ccea8 100644 --- a/lib/commands/channel.js +++ b/lib/commands/channel.js @@ -147,7 +147,7 @@ Channel.isNewChannel = function (Env, channel, cb) { Otherwise behaves the same as sending to a channel */ -Channel.writePrivateMessage = function (Env, args, nfwssCtx, cb) { +Channel.writePrivateMessage = function (Env, args, Server, cb) { var channelId = args[0]; var msg = args[1]; @@ -161,7 +161,7 @@ Channel.writePrivateMessage = function (Env, args, nfwssCtx, cb) { // We expect a modern netflux-websocket-server instance // if this API isn't here everything will fall apart anyway - if (!(nfwssCtx && nfwssCtx.historyKeeper && typeof(nfwssCtx.historyKeeper.onChannelMessage) === 'function')) { + if (!(Server && typeof(Server.send) === 'function')) { return void cb("NOT_IMPLEMENTED"); } @@ -180,8 +180,9 @@ Channel.writePrivateMessage = function (Env, args, nfwssCtx, cb) { msg // the actual message content. Generally a string ]; + // XXX this API doesn't exist anymore... // store the message and do everything else that is typically done when going through historyKeeper - nfwssCtx.historyKeeper.onChannelMessage(nfwssCtx, channelStruct, fullMessage); + Env.historyKeeper.onChannelMessage(Server, channelStruct, fullMessage); // call back with the message and the target channel. // historyKeeper will take care of broadcasting it if anyone is in the channel diff --git a/lib/historyKeeper.js b/lib/historyKeeper.js index 92eb84a5c..30500f180 100644 --- a/lib/historyKeeper.js +++ b/lib/historyKeeper.js @@ -9,6 +9,8 @@ const Meta = require("./metadata"); const WriteQueue = require("./write-queue"); const BatchRead = require("./batch-read"); +const RPC = require("./rpc"); + const Extras = require("./hk-util.js"); const STANDARD_CHANNEL_LENGTH = Extras.STANDARD_CHANNEL_LENGTH; const EPHEMERAL_CHANNEL_LENGTH = Extras.EPHEMERAL_CHANNEL_LENGTH; @@ -66,8 +68,8 @@ const isValidValidateKeyString = function (key) { var CHECKPOINT_PATTERN = /^cp\|(([A-Za-z0-9+\/=]+)\|)?/; -module.exports.create = function (cfg) { - const rpc = cfg.rpc; +module.exports.create = function (cfg, cb) { + var rpc; const tasks = cfg.tasks; const store = cfg.store; Log = cfg.log; @@ -75,6 +77,7 @@ module.exports.create = function (cfg) { Log.silly('HK_LOADING', 'LOADING HISTORY_KEEPER MODULE'); const metadata_cache = {}; + const channel_cache = {}; const HISTORY_KEEPER_ID = Crypto.randomBytes(8).toString('hex'); Log.verbose('HK_ID', 'History keeper ID: ' + HISTORY_KEEPER_ID); @@ -211,8 +214,9 @@ module.exports.create = function (cfg) { if the channel exists but its index does not then it caches the index */ const batchIndexReads = BatchRead("HK_GET_INDEX"); - const getIndex = (ctx, channelName, cb) => { - const chan = ctx.channels[channelName]; + const getIndex = (channelName, cb) => { + const chan = channel_cache[channelName]; + // if there is a channel in memory and it has an index cached, return it if (chan && chan.index) { // enforce async behaviour @@ -233,15 +237,7 @@ module.exports.create = function (cfg) { }); }; - /*:: - type cp_index_item = { - offset: number, - line: number - } - */ - /* storeMessage - * ctx * channel id * the message to store * whether the message is a checkpoint @@ -260,7 +256,7 @@ module.exports.create = function (cfg) { */ const queueStorage = WriteQueue(); - const storeMessage = function (ctx, channel, msg, isCp, optionalMessageHash) { + const storeMessage = function (channel, msg, isCp, optionalMessageHash) { const id = channel.id; queueStorage(id, function (next) { @@ -284,7 +280,7 @@ module.exports.create = function (cfg) { } })); }).nThen((waitFor) => { - getIndex(ctx, id, waitFor((err, index) => { + getIndex(id, waitFor((err, index) => { if (err) { Log.warn("HK_STORE_MESSAGE_INDEX", err.stack); // non-critical, we'll be able to get the channel index later @@ -298,10 +294,10 @@ module.exports.create = function (cfg) { delete index.offsetByHash[k]; } } - index.cpIndex.push(({ + index.cpIndex.push({ offset: index.size, line: ((index.line || 0) + 1) - } /*:cp_index_item*/)); + }); } if (optionalMessageHash) { index.offsetByHash[optionalMessageHash] = index.size; } index.size += msgBin.length; @@ -313,21 +309,10 @@ module.exports.create = function (cfg) { }); }; - /* historyKeeperBroadcast - * uses API from the netflux server to send messages to every member of a channel - * sendMsg runs in a try-catch and drops users if sending a message fails - */ - const historyKeeperBroadcast = function (ctx, channel, msg) { - let chan = ctx.channels[channel] || (([] /*:any*/) /*:Chan_t*/); - chan.forEach(function (user) { - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)]); - }); - }; - /* expireChannel is here to clean up channels that should have been removed but for some reason are still present */ - const expireChannel = function (ctx, channel) { + const expireChannel = function (channel) { return void store.archiveChannel(channel, function (err) { Log.info("ARCHIVAL_CHANNEL_BY_HISTORY_KEEPER_EXPIRATION", { channelId: channel, @@ -336,6 +321,14 @@ module.exports.create = function (cfg) { }); }; + /* dropChannel + * cleans up memory structures which are managed entirely by the historyKeeper + */ + const dropChannel = function (chanName) { + delete metadata_cache[chanName]; + delete channel_cache[chanName]; + }; + /* checkExpired * synchronously returns true or undefined to indicate whether the channel is expired * according to its metadata @@ -347,7 +340,7 @@ module.exports.create = function (cfg) { FIXME the boolean nature of this API should be separated from its side effects */ - const checkExpired = function (ctx, channel) { + const checkExpired = function (Server, channel) { if (!(channel && channel.length === STANDARD_CHANNEL_LENGTH)) { return false; } let metadata = metadata_cache[channel]; if (!(metadata && typeof(metadata.expire) === 'number')) { return false; } @@ -362,18 +355,16 @@ module.exports.create = function (cfg) { // there may have been a problem with scheduling tasks // or the scheduled tasks may not be running // so trigger a removal from here - if (pastDue >= ONE_DAY) { expireChannel(ctx, channel); } + if (pastDue >= ONE_DAY) { expireChannel(channel); } // close the channel store.closeChannel(channel, function () { - historyKeeperBroadcast(ctx, channel, { + // XXX make sure that clients actually disconnect when we broadcast an error + Server.channelBroadcast(channel, { error: 'EEXPIRED', channel: channel - }); - // remove it from any caches after you've told anyone in the channel - // that it has expired - delete ctx.channels[channel]; - delete metadata_cache[channel]; + }, HISTORY_KEEPER_ID); + dropChannel(channel); }); // return true to indicate that it has expired @@ -391,7 +382,7 @@ module.exports.create = function (cfg) { * adds timestamps to incoming messages * writes messages to the store */ - const onChannelMessage = function (ctx, channel, msgStruct) { + const onChannelMessage = function (Server, channel, msgStruct) { // TODO our usage of 'channel' here looks prone to errors // we only use it for its 'id', but it can contain other stuff // also, we're using this RPC from both the RPC and Netflux-server @@ -414,7 +405,7 @@ module.exports.create = function (cfg) { let metadata; nThen(function (w) { // getIndex (and therefore the latest metadata) - getIndex(ctx, channel.id, w(function (err, index) { + getIndex(channel.id, w(function (err, index) { if (err) { w.abort(); return void Log.error('CHANNEL_MESSAGE_ERROR', err); @@ -429,7 +420,7 @@ module.exports.create = function (cfg) { metadata = index.metadata; // don't write messages to expired channels - if (checkExpired(ctx, channel)) { return void w.abort(); } + if (checkExpired(Server, channel)) { return void w.abort(); } // if there's no validateKey present skip to the next block if (!metadata.validateKey) { return; } @@ -479,20 +470,10 @@ module.exports.create = function (cfg) { msgStruct.push(now()); // storeMessage - storeMessage(ctx, channel, JSON.stringify(msgStruct), isCp, getHash(msgStruct[4], Log)); + storeMessage(channel, JSON.stringify(msgStruct), isCp, getHash(msgStruct[4], Log)); }); }; - /* dropChannel - * exported as API - * used by chainpad-server/NetfluxWebsocketSrv.js - * cleans up memory structures which are managed entirely by the historyKeeper - * the netflux server manages other memory in ctx.channels - */ - const dropChannel = function (chanName) { - delete metadata_cache[chanName]; - }; - /* getHistoryOffset returns a number representing the byte offset from the start of the log for whatever history you're seeking. @@ -522,12 +503,12 @@ module.exports.create = function (cfg) { * -1 if you didn't find it */ - const getHistoryOffset = (ctx, channelName, lastKnownHash, cb /*:(e:?Error, os:?number)=>void*/) => { + const getHistoryOffset = (channelName, lastKnownHash, cb) => { // lastKnownhash === -1 means we want the complete history if (lastKnownHash === -1) { return void cb(null, 0); } let offset = -1; nThen((waitFor) => { - getIndex(ctx, channelName, waitFor((err, index) => { + getIndex(channelName, waitFor((err, index) => { if (err) { waitFor.abort(); return void cb(err); } // check if the "hash" the client is requesting exists in the index @@ -600,10 +581,10 @@ module.exports.create = function (cfg) { * GET_HISTORY */ - const getHistoryAsync = (ctx, channelName, lastKnownHash, beforeHash, handler, cb) => { + const getHistoryAsync = (channelName, lastKnownHash, beforeHash, handler, cb) => { let offset = -1; nThen((waitFor) => { - getHistoryOffset(ctx, channelName, lastKnownHash, waitFor((err, os) => { + getHistoryOffset(channelName, lastKnownHash, waitFor((err, os) => { if (err) { waitFor.abort(); return void cb(err); @@ -666,42 +647,50 @@ module.exports.create = function (cfg) { /* onChannelCleared * broadcasts to all clients in a channel if that channel is deleted */ - const onChannelCleared = function (ctx, channel) { - historyKeeperBroadcast(ctx, channel, { + const onChannelCleared = function (Server, channel) { + Server.channelBroadcast(channel, { error: 'ECLEARED', channel: channel - }); + }, HISTORY_KEEPER_ID); }; + // When a channel is removed from datastore, broadcast a message to all its connected users - const onChannelDeleted = function (ctx, channel) { + const onChannelDeleted = function (Server, channel) { store.closeChannel(channel, function () { - historyKeeperBroadcast(ctx, channel, { + Server.channelBroadcast(channel, { error: 'EDELETED', channel: channel - }); + }, HISTORY_KEEPER_ID); }); - delete ctx.channels[channel]; + + delete channel_cache[channel]; + Server.clearChannel(channel); delete metadata_cache[channel]; }; // Check if the selected channel is expired // If it is, remove it from memory and broadcast a message to its members - const onChannelMetadataChanged = function (ctx, channel, metadata) { - if (channel && metadata_cache[channel] && typeof (metadata) === "object") { - Log.silly('SET_METADATA_CACHE', 'Channel '+ channel +', metadata: '+ JSON.stringify(metadata)); - metadata_cache[channel] = metadata; - if (ctx.channels[channel] && ctx.channels[channel].index) { - ctx.channels[channel].index.metadata = metadata; - } - historyKeeperBroadcast(ctx, channel, metadata); + const onChannelMetadataChanged = function (Server, channel, metadata) { + if (!(channel && metadata_cache[channel] && typeof (metadata) === "object")) { return; } + Log.silly('SET_METADATA_CACHE', { + channel: channel, + metadata: JSON.stringify(metadata), + }); + + metadata_cache[channel] = metadata; + + if (channel_cache[channel] && channel_cache[channel].index) { + channel_cache[channel].index.metadata = metadata; } + Server.channelBroadcast(channel, metadata, HISTORY_KEEPER_ID); }; - const handleGetHistory = function (ctx, seq, user, parsed) { + const handleGetHistory = function (Server, seq, user, parsed) { // parsed[1] is the channel id // parsed[2] is a validation key or an object containing metadata (optionnal) // parsed[3] is the last known hash (optionnal) - ctx.sendMsg(ctx, user, [seq, 'ACK']); + + Server.send(user.id, [seq, 'ACK']); var channelName = parsed[1]; var config = parsed[2]; var metadata = {}; @@ -736,7 +725,7 @@ module.exports.create = function (cfg) { unfortunately, we can't just serve it blindly, since then young channels will send the metadata twice, so let's do a quick check of what we're going to serve... */ - getIndex(ctx, channelName, waitFor((err, index) => { + getIndex(channelName, waitFor((err, index) => { /* if there's an error here, it should be encountered and handled by the next nThen block. so, let's just fall through... @@ -750,29 +739,29 @@ module.exports.create = function (cfg) { if (!index || !index.metadata) { return void w(); } // And then check if the channel is expired. If it is, send the error and abort // FIXME this is hard to read because 'checkExpired' has side effects - if (checkExpired(ctx, channelName)) { return void waitFor.abort(); } + if (checkExpired(Server, channelName)) { return void waitFor.abort(); } // always send metadata with GET_HISTORY requests - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(index.metadata)], w); + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(index.metadata)], w); })); }).nThen(() => { let msgCount = 0; // TODO compute lastKnownHash in a manner such that it will always skip past the metadata line? - getHistoryAsync(ctx, channelName, lastKnownHash, false, (msg, readMore) => { + getHistoryAsync(channelName, lastKnownHash, false, (msg, readMore) => { if (!msg) { return; } msgCount++; // avoid sending the metadata message a second time if (isMetadataMessage(msg) && metadata_cache[channelName]) { return readMore(); } - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)], readMore); + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)], readMore); }, (err) => { if (err && err.code !== 'ENOENT') { if (err.message !== 'EINVAL') { Log.error("HK_GET_HISTORY", err); } const parsedMsg = {error:err.message, channel: channelName}; - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); return; } - const chan = ctx.channels[channelName]; + const chan = channel_cache[channelName]; if (msgCount === 0 && !metadata_cache[channelName] && chan && chan.indexOf(user) > -1) { metadata_cache[channelName] = metadata; @@ -811,21 +800,22 @@ module.exports.create = function (cfg) { } }); } - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(metadata)]); + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(metadata)]); } // End of history message: let parsedMsg = {state: 1, channel: channelName}; - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); + + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); }); }); }; - const handleGetHistoryRange = function (ctx, seq, user, parsed) { + const handleGetHistoryRange = function (Server, seq, user, parsed) { var channelName = parsed[1]; var map = parsed[2]; if (!(map && typeof(map) === 'object')) { - return void ctx.sendMsg(ctx, user, [seq, 'ERROR', 'INVALID_ARGS', HISTORY_KEEPER_ID]); + return void Server.send(user.id, [seq, 'ERROR', 'INVALID_ARGS', HISTORY_KEEPER_ID]); } var oldestKnownHash = map.from; @@ -833,14 +823,14 @@ module.exports.create = function (cfg) { var desiredCheckpoint = map.cpCount; var txid = map.txid; if (typeof(desiredMessages) !== 'number' && typeof(desiredCheckpoint) !== 'number') { - return void ctx.sendMsg(ctx, user, [seq, 'ERROR', 'UNSPECIFIED_COUNT', HISTORY_KEEPER_ID]); + return void Server.send(user.id, [seq, 'ERROR', 'UNSPECIFIED_COUNT', HISTORY_KEEPER_ID]); } if (!txid) { - return void ctx.sendMsg(ctx, user, [seq, 'ERROR', 'NO_TXID', HISTORY_KEEPER_ID]); + return void Server.send(user.id, [seq, 'ERROR', 'NO_TXID', HISTORY_KEEPER_ID]); } - ctx.sendMsg(ctx, user, [seq, 'ACK']); + Server.send(user.id, [seq, 'ACK']); return void getOlderHistory(channelName, oldestKnownHash, function (messages) { var toSend = []; if (typeof (desiredMessages) === "number") { @@ -856,64 +846,66 @@ module.exports.create = function (cfg) { } } toSend.forEach(function (msg) { - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['HISTORY_RANGE', txid, msg])]); }); - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['HISTORY_RANGE_END', txid, channelName]) ]); }); }; - const handleGetFullHistory = function (ctx, seq, user, parsed) { + const handleGetFullHistory = function (Server, seq, user, parsed) { // parsed[1] is the channel id // parsed[2] is a validation key (optionnal) // parsed[3] is the last known hash (optionnal) - ctx.sendMsg(ctx, user, [seq, 'ACK']); + + Server.send(user.id, [seq, 'ACK']); // FIXME should we send metadata here too? // none of the clientside code which uses this API needs metadata, but it won't hurt to send it (2019-08-22) - return void getHistoryAsync(ctx, parsed[1], -1, false, (msg, readMore) => { + return void getHistoryAsync(parsed[1], -1, false, (msg, readMore) => { if (!msg) { return; } - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['FULL_HISTORY', msg])], readMore); + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(['FULL_HISTORY', msg])], readMore); }, (err) => { let parsedMsg = ['FULL_HISTORY_END', parsed[1]]; if (err) { Log.error('HK_GET_FULL_HISTORY', err.stack); parsedMsg = ['ERROR', parsed[1], err.message]; } - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(parsedMsg)]); }); }; - const handleRPC = function (ctx, seq, user, parsed) { + const handleRPC = function (Server, seq, user, parsed) { if (typeof(rpc) !== 'function') { return; } /* RPC Calls... */ var rpc_call = parsed.slice(1); - ctx.sendMsg(ctx, user, [seq, 'ACK']); + // XXX ensure user is guaranteed to have 'id' + Server.send(user.id, [seq, 'ACK']); try { // slice off the sequence number and pass in the rest of the message - rpc(ctx, rpc_call, function (err, output) { + rpc(Server, rpc_call, function (err, output) { if (err) { - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', err])]); + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', err])]); return; } var msg = rpc_call[0].slice(); if (msg[3] === 'REMOVE_OWNED_CHANNEL') { - onChannelDeleted(ctx, msg[4]); + onChannelDeleted(Server, msg[4]); } if (msg[3] === 'CLEAR_OWNED_CHANNEL') { - onChannelCleared(ctx, msg[4]); + onChannelCleared(Server, msg[4]); } if (msg[3] === 'SET_METADATA') { // or whatever we call the RPC???? // make sure we update our cache of metadata // or at least invalidate it and force other mechanisms to recompute its state // 'output' could be the new state as computed by rpc - onChannelMetadataChanged(ctx, msg[4].channel, output[1]); + onChannelMetadataChanged(Server, msg[4].channel, output[1]); } // unauthenticated RPC calls have a different message format @@ -921,10 +913,10 @@ module.exports.create = function (cfg) { // this is an inline reimplementation of historyKeeperBroadcast // because if we use that directly it will bypass signature validation // which opens up the user to malicious behaviour - let chan = ctx.channels[output.channel]; + let chan = channel_cache[output.channel]; if (chan && chan.length) { chan.forEach(function (user) { - ctx.sendMsg(ctx, user, output.message); + Server.send(user.id, output.message); //[0, null, 'MSG', user.id, JSON.stringify(output.message)]); }); } @@ -934,10 +926,10 @@ module.exports.create = function (cfg) { } // finally, send a response to the client that sent the RPC - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0]].concat(output))]); + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0]].concat(output))]); }); } catch (e) { - ctx.sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', 'SERVER_ERROR'])]); + Server.send(user.id, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify([parsed[0], 'ERROR', 'SERVER_ERROR'])]); } }; @@ -953,7 +945,7 @@ module.exports.create = function (cfg) { * check if it's expired and execute all the associated side-effects * routes queries to the appropriate handlers */ - const onDirectMessage = function (ctx, seq, user, json) { + const onDirectMessage = function (Server, seq, user, json) { Log.silly('HK_MESSAGE', json); let parsed; @@ -967,33 +959,49 @@ module.exports.create = function (cfg) { // If the requested history is for an expired channel, abort // Note the if we don't have the keys for that channel in metadata_cache, we'll // have to abort later (once we know the expiration time) - if (checkExpired(ctx, parsed[1])) { return; } + if (checkExpired(Server, parsed[1])) { return; } // look up the appropriate command in the map of commands or fall back to RPC var command = directMessageCommands[parsed[0]] || handleRPC; // run the command with the standard function signature - command(ctx, seq, user, parsed); + command(Server, seq, user, parsed); }; - return { + cfg.historyKeeper = { id: HISTORY_KEEPER_ID, - channelMessage: function (ctx, channel, msgStruct) { - onChannelMessage(ctx, channel, msgStruct); + + channelMessage: function (Server, channel, msgStruct) { + // netflux-server emits 'channelMessage' events whenever someone broadcasts to a channel + // historyKeeper stores these messages if the channel id indicates that they are + // a channel type with permanent history + onChannelMessage(Server, channel, msgStruct); }, channelClose: function (channelName) { + // netflux-server emits 'channelClose' events whenever everyone leaves a channel + // we drop cached metadata and indexes at the same time dropChannel(channelName); }, - channelOpen: function (ctx, channelName, user) { - ctx.sendMsg(ctx, user, [ + channelOpen: function (Server, channelName, userId) { + channel_cache[channelName] = {}; + Server.send(userId, [ 0, - HISTORY_KEEPER_ID, // ctx.historyKeeper.id + HISTORY_KEEPER_ID, 'JOIN', channelName ]); }, - directMessage: function (ctx, seq, user, json) { - onDirectMessage(ctx, seq, user, json); + directMessage: function (Server, seq, user, json) { + // netflux-server allows you to register an id with a handler + // this handler is invoked every time someone sends a message to that id + onDirectMessage(Server, seq, user, json); }, }; + + RPC.create(cfg, function (err, _rpc) { + if (err) { throw err; } + + rpc = _rpc; + cb(void 0, cfg.historyKeeper); + }); }; diff --git a/lib/rpc.js b/lib/rpc.js index 171fb590f..47ea9976f 100644 --- a/lib/rpc.js +++ b/lib/rpc.js @@ -24,7 +24,6 @@ const UNAUTHENTICATED_CALLS = [ 'GET_MULTIPLE_FILE_SIZE', 'IS_CHANNEL_PINNED', 'IS_NEW_CHANNEL', - 'GET_HISTORY_OFFSET', 'GET_DELETED_PADS', 'WRITE_PRIVATE_MESSAGE', ]; @@ -66,25 +65,9 @@ var isUnauthenticateMessage = function (msg) { return msg && msg.length === 2 && isUnauthenticatedCall(msg[0]); }; -var handleUnauthenticatedMessage = function (Env, msg, respond, nfwssCtx) { +var handleUnauthenticatedMessage = function (Env, msg, respond, Server) { Env.Log.silly('LOG_RPC', msg[0]); switch (msg[0]) { - case 'GET_HISTORY_OFFSET': { // XXX not actually used anywhere? - if (typeof(msg[1]) !== 'object' || typeof(msg[1].channelName) !== 'string') { - return respond('INVALID_ARG_FORMAT', msg); - } - const msgHash = typeof(msg[1].msgHash) === 'string' ? msg[1].msgHash : undefined; - nfwssCtx.getHistoryOffset(nfwssCtx, msg[1].channelName, msgHash, (e, ret) => { - if (e) { - if (e.code !== 'ENOENT') { - Env.WARN(e.stack, msg); - } - return respond(e.message); - } - respond(e, [null, ret, null]); - }); - break; - } case 'GET_FILE_SIZE': return void Pinning.getFileSize(Env, msg[1], function (e, size) { Env.WARN(e, msg[1]); @@ -120,7 +103,7 @@ var handleUnauthenticatedMessage = function (Env, msg, respond, nfwssCtx) { respond(e, [null, isNew, null]); }); case 'WRITE_PRIVATE_MESSAGE': - return void Channel.writePrivateMessage(Env, msg[1], nfwssCtx, function (e, output) { + return void Channel.writePrivateMessage(Env, msg[1], Server, function (e, output) { respond(e, output); }); default: @@ -134,7 +117,7 @@ var handleAuthenticatedMessage = function (Env, map) { var safeKey = map.safeKey; var publicKey = map.publicKey; var Respond = map.Respond; - var ctx = map.ctx; + var Server = map.Server; Env.Log.silly('LOG_RPC', msg[0]); switch (msg[0]) { @@ -265,7 +248,7 @@ var handleAuthenticatedMessage = function (Env, map) { Respond(e); }); case 'ADMIN': - return void Admin.command(Env, ctx, safeKey, msg[1], function (e, result) { // XXX SPECIAL + return void Admin.command(Env, Server, safeKey, msg[1], function (e, result) { // XXX SPECIAL if (e) { Env.WARN(e, result); return void Respond(e); @@ -285,7 +268,7 @@ var handleAuthenticatedMessage = function (Env, map) { } }; -var rpc = function (Env, ctx, data, respond) { +var rpc = function (Env, Server, data, respond) { if (!Array.isArray(data)) { Env.Log.debug('INVALID_ARG_FORMET', data); return void respond('INVALID_ARG_FORMAT'); @@ -304,7 +287,7 @@ var rpc = function (Env, ctx, data, respond) { } if (isUnauthenticateMessage(msg)) { - return handleUnauthenticatedMessage(Env, msg, respond, ctx); + return handleUnauthenticatedMessage(Env, msg, respond); } var signature = msg.shift(); @@ -369,7 +352,7 @@ var rpc = function (Env, ctx, data, respond) { safeKey: safeKey, publicKey: publicKey, Respond: Respond, - ctx: ctx, + Server: Server, }); }; @@ -394,6 +377,7 @@ RPC.create = function (config, cb) { }; var Env = { + historyKeeper: config.historyKeeper, defaultStorageLimit: config.defaultStorageLimit, maxUploadSize: config.maxUploadSize || (20 * 1024 * 1024), Sessions: {}, @@ -465,11 +449,9 @@ RPC.create = function (config, cb) { Env.blobStore = blob; })); }).nThen(function () { - // XXX it's ugly that we pass ctx and Env separately - // when they're effectively the same thing... - cb(void 0, function (ctx, data, respond) { + cb(void 0, function (Server, data, respond) { try { - return rpc(Env, ctx, data, respond); + return rpc(Env, Server, data, respond); } catch (e) { console.log("Error from RPC with data " + JSON.stringify(data)); console.log(e.stack); From e3269df7f0dea1b0f60da469547f411afc6be5f0 Mon Sep 17 00:00:00 2001 From: ansuz Date: Mon, 3 Feb 2020 14:23:31 -0500 Subject: [PATCH 5/5] fix some critical errors in the trim history storage api --- storage/file.js | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/storage/file.js b/storage/file.js index 08ec98f85..09b743ace 100644 --- a/storage/file.js +++ b/storage/file.js @@ -955,7 +955,9 @@ var trimChannel = function (env, channelName, hash, _cb) { var handler = function (msgObj, readMore, abort) { if (ABORT) { return void abort(); } // the first message might be metadata... ignore it if so - if (i++ === 0 && msgObj.buff.indexOf('{') === 0) { return; } + if (i++ === 0 && msgObj.buff.indexOf('{') === 0) { + return readMore(); + } if (retain) { // if this flag is set then you've already found @@ -973,6 +975,9 @@ var trimChannel = function (env, channelName, hash, _cb) { if (msgHash === hash) { // everything from this point on should be retained retain = true; + return void tempStream.write(msgObj.buff, function () { + readMore(); + }); } };