From 1629eaf225e865af411666a34c0ec632e61739dc Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Wed, 6 May 2020 19:43:15 +0200 Subject: [PATCH 01/14] Track plugins loaded by master and slaves Send a mapping of the plugins the master server has loaded to slaves vice versa. Limit the plugins that may be loaded by instances to only those that the master server has loaded. --- docs/socket.md | 6 ++++++ docs/writing-plugins.md | 6 +++--- lib/link/connectors.js | 8 ++++++++ lib/schema.js | 12 ++++++++++-- master.js | 10 +++++++++- slave.js | 21 ++++++++++++++++----- test/lib/link/connector.js | 2 +- 7 files changed, 53 insertions(+), 12 deletions(-) diff --git a/docs/socket.md b/docs/socket.md index 87d4718a..b0544b9f 100644 --- a/docs/socket.md +++ b/docs/socket.md @@ -114,6 +114,9 @@ Handshake messages Sent by the master server when the WebSocket connection has been opened. - version - string - The version of the master, e.g. "2.0.0". +- plugins - Object<string, string> - + Object mapping plugin names to plugin version for plugins that are + loaded on the master server. ### `register_slave` @@ -124,6 +127,9 @@ Slave handshake for establishing a new connection session. - version - string - The protocol version of the slave, e.g. "2.0.0". - name - string - Name of the slave. - id - integer - ID of the slave. +- plugins - Object<string, string> - + Object mapping plugin names to plugin version for plugins that are + available on the slave. Currently not used by the master server. ### `register_control` diff --git a/docs/writing-plugins.md b/docs/writing-plugins.md index 2b57862d..15710824 100644 --- a/docs/writing-plugins.md +++ b/docs/writing-plugins.md @@ -73,9 +73,9 @@ The following properties are recognized: **instanceEntrypoint**: Path to a Node.js module relative to the plugin directory which contains the InstancePlugin class definition for this plugin. This - is an optional paramater. A plugin can be for instances only, but - for it to be able to send any messages to other instances it will - still have to be loaded on the master server. + is an optional paramater. A plugin may have code only for instances + but it must still be loaded on the master in order for it to be + possible to load it on an instance. **InstanceConfigGroup**: Subclass of `PluginConfigGroup` for defining the per instance diff --git a/lib/link/connectors.js b/lib/link/connectors.js index 5a420323..70e08d53 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -131,6 +131,13 @@ class WebSocketClientConnector extends WebSocketBaseConnector { this._reconnectDelay = reconnectDelay; this._timeout = 15 * 60; this._startedReconnect = null; + + /** + * Mapping of plugin name to plugin version loaded on the master + * server. Is `null` before handshake has taken place. + * @member {?Object} + */ + this.plugins = null; } _check(expectedStates) { @@ -335,6 +342,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector { let { seq, type, data } = message; if (type == 'hello') { console.log(`SOCKET | received hello from master version ${data.version}`); + this.plugins = data.plugins; if (this._sessionToken) { console.log("SOCKET | Attempting resume"); this.sendHandshake("resume", { diff --git a/lib/schema.js b/lib/schema.js index fec79da6..07e649f6 100644 --- a/lib/schema.js +++ b/lib/schema.js @@ -58,9 +58,13 @@ const serverHandshake = ajv.compile({ "type": { const: "hello" }, "data": { additionalProperties: false, - required: ["version"], + required: ["version", "plugins"], properties: { "version": { type: "string" }, + "plugins": { + type: "object", + additionalProperties: { type: "string" }, + }, } }, }, @@ -111,13 +115,17 @@ const clientHandshake = ajv.compile({ "type": { const: "register_slave" }, "data": { additionalProperties: false, - required: ["token", "agent", "version", "name", "id"], + required: ["token", "agent", "version", "name", "id", "plugins"], properties: { "token": { type: "string" }, "agent": { type: "string" }, "version": { type: "string" }, "name": { type: "string" }, "id": { type: "integer" }, + "plugins": { + type: "object", + additionalProperties: { type: "string" }, + }, }, }, }, diff --git a/master.js b/master.js index a144945f..3d7fa7c6 100644 --- a/master.js +++ b/master.js @@ -31,6 +31,7 @@ let masterConfig; let masterConfigPath; let stopAcceptingNewSessions = false; let debugEvents = new events.EventEmitter(); +let pluginList = {}; // homebrew modules const generateSSLcert = require("lib/generateSSLcert"); @@ -581,12 +582,14 @@ class SlaveConnection extends BaseConnection { this._id = registerData.id; this._name = registerData.name; this._version = registerData.version; + this._plugins = registerData.plugins; db.slaves.set(this._id, { agent: this._agent, id: this._id, name: this._name, version: this._version, + plugins: this._plugins, }); this.connector.on("close", () => { @@ -863,7 +866,10 @@ wss.on("connection", function (socket, req) { }; // Start connection handshake. - socket.send(JSON.stringify({ seq: null, type: "hello", data: { version }})); + socket.send(JSON.stringify({ seq: null, type: "hello", data: { + version, + plugins: pluginList + }})); function attachHandler() { pendingSockets.add(socket); @@ -1000,6 +1006,8 @@ async function loadPlugins(pluginInfos) { continue; } + pluginList[pluginInfo.name] = pluginInfo.version; + let pluginLoadStarted = Date.now(); let MasterPlugin = plugin.BaseMasterPlugin; try { diff --git a/slave.js b/slave.js index a08d56bd..78041eb1 100644 --- a/slave.js +++ b/slave.js @@ -117,12 +117,16 @@ class Instance extends link.Link{ } } - async init(pluginInfos, slave) { + async init(pluginInfos, masterPlugins, slave) { await this.server.init(); // load plugins for (let pluginInfo of pluginInfos) { - if (!pluginInfo.instanceEntrypoint || !this.config.group(pluginInfo.name).get("enabled")) { + if ( + !pluginInfo.instanceEntrypoint + || !Object.hasOwnProperty.call(masterPlugins, pluginInfo.name) + || !this.config.group(pluginInfo.name).get("enabled") + ) { continue; } @@ -513,19 +517,26 @@ class InstanceConnection extends link.Link { } class SlaveConnector extends link.WebSocketClientConnector { - constructor(slaveConfig) { + constructor(slaveConfig, pluginInfos) { super(slaveConfig.get("slave.master_url"), slaveConfig.get("slave.reconnect_delay")); this.slaveConfig = slaveConfig; + this.pluginInfos = pluginInfos; } register() { console.log("SOCKET | registering slave"); + let plugins = {}; + for (let pluginInfo of this.pluginInfos) { + plugins[pluginInfo.name] = pluginInfo.version; + } + this.sendHandshake("register_slave", { token: this.slaveConfig.get("slave.master_token"), agent: "Clusterio Slave", version, id: this.slaveConfig.get("slave.id"), name: this.slaveConfig.get("slave.name"), + plugins, }); } } @@ -653,7 +664,7 @@ class Slave extends link.Link { let instance = new Instance( connectionClient, instanceInfo.path, this.config.get("slave.factorio_directory"), instanceInfo.config ); - await instance.init(this.pluginInfos, this); + await instance.init(this.pluginInfos, this.connector.plugins, this); // XXX: race condition on multiple simultanious calls this.instanceConnections.set(instanceId, instanceConnection); @@ -945,7 +956,7 @@ async function startSlave() { return; } - let slaveConnector = new SlaveConnector(slaveConfig); + let slaveConnector = new SlaveConnector(slaveConfig, pluginInfos); let slave = new Slave(slaveConnector, slaveConfig, pluginInfos); // Handle interrupts diff --git a/test/lib/link/connector.js b/test/lib/link/connector.js index b800a679..99c4f6be 100644 --- a/test/lib/link/connector.js +++ b/test/lib/link/connector.js @@ -86,7 +86,7 @@ describe("lib/link/connectors", function() { it("should call register on hello", function() { let called = false; testConnector.register = () => { called = true; }; - testConnector._processHandshake({ seq: 1, type: 'hello', data: { version: "test" }}); + testConnector._processHandshake({ seq: 1, type: 'hello', data: { version: "test", plugins: {} }}); assert(called, "register was not called"); }); it("should emit connected on ready", async function() { From 5c1214ad38bab95194d3e4218f34f66a594ca6ea Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Thu, 7 May 2020 12:35:48 +0200 Subject: [PATCH 02/14] Rename connected and dropped events Rename the connector events connected and dropped to connect and drop respectively to be consistent with common Node.js names for events like close. --- lib/link/connectors.js | 8 ++++---- master.js | 8 ++++---- test/integration/link.js | 8 ++++---- test/lib/link/connector.js | 4 ++-- 4 files changed, 14 insertions(+), 14 deletions(-) diff --git a/lib/link/connectors.js b/lib/link/connectors.js index 70e08d53..d7afe5fc 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -199,7 +199,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector { this._doConnect(); // Wait for the connection to be ready - await events.once(this, "connected"); + await events.once(this, "connect"); } async _doConnect() { @@ -275,7 +275,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector { } else { this._state = "handshake"; this.reconnect(); - this.emit("dropped"); + this.emit("drop"); } this.stopHeartbeat(); @@ -361,7 +361,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector { this.startHeartbeat(); this._startedReconnect = null; this._connected = true; - this.emit("connected"); + this.emit("connect"); } else if (type === "continue") { console.log("SOCKET | resuming existing session"); @@ -374,7 +374,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector { } this._startedReconnect = null; this._connected = true; - this.emit("connected"); + this.emit("connect"); } else if (type === "invalidate") { this._heartbeatInterval = null; diff --git a/master.js b/master.js index 3d7fa7c6..f2dba036 100644 --- a/master.js +++ b/master.js @@ -433,7 +433,7 @@ class ControlConnection extends BaseConnection { this.instanceOutputSubscriptions = new Set(); this.ws_dumper = null; - this.connector.on("connected", () => { + this.connector.on("connect", () => { this.connector._socket.clusterio_ignore_dump = !!this.ws_dumper; }); this.connector.on("close", () => { @@ -701,7 +701,7 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector { this._state = "connected"; this._connected = true; this._attachSocketHandlers(); - this.emit("connected"); + this.emit("connect"); } /** @@ -736,7 +736,7 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector { for (let message of this._sendBuffer) { this._socket.send(JSON.stringify(message)); } - this.emit("connected"); + this.emit("connect"); } setTimeout(timeout) { @@ -782,7 +782,7 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector { } else { this._state = "handshake"; - this.emit("dropped"); + this.emit("drop"); this._timeoutId = setTimeout(() => { this._timedOut() }, this._timeout * 1000); } diff --git a/test/integration/link.js b/test/integration/link.js index 5836d507..680e604b 100644 --- a/test/integration/link.js +++ b/test/integration/link.js @@ -45,24 +45,24 @@ describe("Integration of lib/link", function() { it("should reconnect on connection lost", async function() { controlConnector._socket.close(1008, "Test"); - await events.once(controlConnector, "connected"); + await events.once(controlConnector, "connect"); }); it("should reconnect on connection terminated", async function() { controlConnector._socket.terminate(); - await events.once(controlConnector, "connected"); + await events.once(controlConnector, "connect"); }); it("should reconnect on connection heartbeat timeout", async function() { controlConnector.stopHeartbeat(); - await events.once(controlConnector, "connected"); + await events.once(controlConnector, "connect"); }); it("should invalidate on reconnection with bad session token", async function() { controlConnector._sessionToken = "blah"; controlConnector.stopHeartbeat(); await events.once(controlConnector, "invalidate"); - await events.once(controlConnector, "connected"); + await events.once(controlConnector, "connect"); }); after(async function() { diff --git a/test/lib/link/connector.js b/test/lib/link/connector.js index 99c4f6be..6f0abb7c 100644 --- a/test/lib/link/connector.js +++ b/test/lib/link/connector.js @@ -89,8 +89,8 @@ describe("lib/link/connectors", function() { testConnector._processHandshake({ seq: 1, type: 'hello', data: { version: "test", plugins: {} }}); assert(called, "register was not called"); }); - it("should emit connected on ready", async function() { - let result = events.once(testConnector, 'connected'); + it("should emit connect on ready", async function() { + let result = events.once(testConnector, "connect"); testConnector._processHandshake( { seq: 1, type: 'ready', data: { session_token: "x", heartbeat_interval: 10 }} ); From 7a9b6592e60de06667b9fbb901ed2c0e98b1ab12 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Thu, 7 May 2020 13:22:04 +0200 Subject: [PATCH 03/14] Handle disconnect in link integration test Handle the prepareDisconnect request properly in the TestControl in test/integration/link.js. --- test/integration/link.js | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/test/integration/link.js b/test/integration/link.js index 680e604b..a5f2b95a 100644 --- a/test/integration/link.js +++ b/test/integration/link.js @@ -12,8 +12,12 @@ class TestControl extends link.Link { link.attachAllMessages(this); } + async prepareDisconnectRequestHandler(message, request) { + this.connector.setClosing(); + return await super.prepareDisconnectRequestHandler(message, request); + } + async debugWsMessageEventHandler() { } - async prepareDisconnectRequestHandler() { } async instanceOutputEventHandler() { } } @@ -64,8 +68,4 @@ describe("Integration of lib/link", function() { await events.once(controlConnector, "invalidate"); await events.once(controlConnector, "connect"); }); - - after(async function() { - controlConnector.close(1001, "Test Quit"); - }); }); From b151dd84695a660f2524d17b14106faac8241a7c Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Thu, 7 May 2020 17:59:36 +0200 Subject: [PATCH 04/14] Fix documentation Fix missing module declarations, broken lings, broken formating and incorrect scopes in the documentation. For modules split over several files it appears the only viable option is to use the memberof directive on each documented member and only declare the index file as a module. --- clusterctl.js | 9 +++++ docs/jsdoc.json | 3 +- lib/config/classes.js | 27 +++++++++++--- lib/config/definitions.js | 48 ++++++++++++++++++++++-- lib/config/index.js | 4 +- lib/errors.js | 39 ++++++++++++++++--- lib/factorio/index.js | 5 +++ lib/factorio/locale.js | 6 +-- lib/factorio/patch.js | 15 ++++++-- lib/factorio/server.js | 66 +++++++++++++++++++++++++-------- lib/fileOps/fileOps.js | 3 ++ lib/hash.js | 6 +++ lib/link/connectors.js | 16 ++++++++ lib/link/index.js | 5 +++ lib/link/link.js | 5 ++- lib/link/messages.js | 7 ++++ lib/luaTools.js | 5 +++ lib/plugin.js | 20 +++++++++- lib/schema.js | 34 +++++++++++++++++ master.js | 36 +++++++++++++----- plugins/global_chat/instance.js | 3 ++ slave.js | 12 ++++++ 22 files changed, 318 insertions(+), 56 deletions(-) diff --git a/clusterctl.js b/clusterctl.js index d02594db..0b5ecc43 100644 --- a/clusterctl.js +++ b/clusterctl.js @@ -1,3 +1,7 @@ +/** + * Command line interface for controlling a Clusterio cluster + * @module + */ const jwt = require("jsonwebtoken"); const fs = require("fs-extra"); const yargs = require("yargs"); @@ -16,6 +20,8 @@ const config = require("lib/config"); * * Formats a parsed Factorio output from lib/factorio into a readable * colorized output using terminal escape codes that can be printed. + * + * @private */ function formatOutputColored(output) { let time = ""; @@ -73,6 +79,7 @@ class Command { * @param client - link to master server to query instance on. * @param instanceName - string with name or id of instance. * @returns {number} instance ID. + * @private */ async function resolveInstance(client, instanceName) { let instanceId; @@ -104,6 +111,7 @@ async function resolveInstance(client, instanceName) { * @param client - link to master server to query slave on. * @param slaveName - string with name or id of slave. * @returns {number} slave ID. + * @private */ async function resolveSlave(client, slaveName) { let slaveId; @@ -324,6 +332,7 @@ commands = new Map([...commands.map(command => [command.name, command])]); /** * Connector for control connection to master server + * @private */ class ControlConnector extends link.WebSocketClientConnector { constructor(url, reconnectDelay, token) { diff --git a/docs/jsdoc.json b/docs/jsdoc.json index 88b1ded8..54216916 100644 --- a/docs/jsdoc.json +++ b/docs/jsdoc.json @@ -18,6 +18,7 @@ "plugins/", "routes/", "routes.js" - ] + ], + "excludePattern": "node_modules" } } diff --git a/lib/config/classes.js b/lib/config/classes.js index c43ec6be..13196729 100644 --- a/lib/config/classes.js +++ b/lib/config/classes.js @@ -1,7 +1,4 @@ -/** - * @file Configuration classes - * @author Hornwitser - */ +// Configuration classes "use strict"; const { basicType } = require("lib/helpers"); @@ -12,6 +9,9 @@ const { basicType } = require("lib/helpers"); * * Exception class for when invalid values are attempted to be set on a * field. + * + * @extends Error + * @memberof module:lib/config */ class InvalidValue extends Error { }; @@ -20,18 +20,26 @@ class InvalidValue extends Error { }; * * Exception class for when a group or field that does not exist is * attempted to be accessed. + * + * @extends Error + * @memberof module:lib/config */ class InvalidField extends Error { }; /** * Split the given string on the first instance of separator * - * Splits @p string on the first instance of @p separator and returns an + * Splits `string` on the first instance of `separator` and returns an * array consisting of the string up to the separator and the string * after the separator. Returns an array with the string and an empty * string if the separator is not present. * + * @param {string} separator - Separator to split string by. + * @param {string} string - String to split * @returns {Array} string split on separator. + * @memberof module:lib/config + * @private + * @inner */ function splitOn(separator, string) { let index = string.indexOf(separator); @@ -46,6 +54,8 @@ function splitOn(separator, string) { * * Represents a collection of ConfigGroup instances that can be * managed as one. + * + * @memberof module:lib/config */ class Config { constructor() { @@ -215,6 +225,8 @@ class Config { /** * Collection of related config entries + * + * @memberof module:lib/config */ class ConfigGroup { /** @@ -285,7 +297,7 @@ class ConfigGroup { * The field must be defined for the config group and the value is * type checked against the field definition. * - * @throws {InvalidValue} if value is not allowed for the field. + * @throws {module:lib/config.InvalidValue} if value is not allowed for the field. */ set(name, value) { let def = this.constructor._definitions.get(name); @@ -553,6 +565,9 @@ class ConfigGroup { * Grouping that includes per plugin fields clusterio uses to manage * plugins. This is currently only the enabled field. It is otherwise * identical to the ConfigGroup class. + * + * @extends module:lib/config.ConfigGroup + * @memberof module:lib/config */ class PluginConfigGroup extends ConfigGroup { static _initSubclass() { diff --git a/lib/config/definitions.js b/lib/config/definitions.js index 9c1dbc14..c20c61ed 100644 --- a/lib/config/definitions.js +++ b/lib/config/definitions.js @@ -1,7 +1,4 @@ -/** - * @file Core definitions for the configuration system - * @author Hornwitser - */ +// Core definitions for the configuration system "use strict"; const util = require("util"); @@ -12,6 +9,11 @@ const fs = require("fs-extra"); const classes = require("./classes"); +/** + * Master config group for {@link module:lib/config.MasterConfig} + * @extends module:lib/config.ConfigGroup + * @memberof module:lib/config + */ class MasterGroup extends classes.ConfigGroup { } MasterGroup.groupName = "master" MasterGroup.define({ @@ -103,10 +105,20 @@ MasterGroup.define({ }); MasterGroup.finalize(); +/** + * Master Config class + * @extends module:lib/config.Config + * @memberof module:lib/config + */ class MasterConfig extends classes.Config { } MasterConfig.registerGroup(MasterGroup); +/** + * Slave config group for {@link module:lib/config.SlaveConfig} + * @extends module:lib/config.ConfigGroup + * @memberof module:lib/config + */ class SlaveGroup extends classes.ConfigGroup {} SlaveGroup.groupName = "slave"; SlaveGroup.define({ @@ -160,10 +172,20 @@ SlaveGroup.define({ }); SlaveGroup.finalize(); +/** + * Slave Config class + * @extends module:lib/config.Config + * @memberof module:lib/config + */ class SlaveConfig extends classes.Config { } SlaveConfig.registerGroup(SlaveGroup); +/** + * Instance config group for {@link module:lib/config.InstanceConfig} + * @extends module:lib/config.ConfigGroup + * @memberof module:lib/config + */ class InstanceGroup extends classes.ConfigGroup { } InstanceGroup.groupName = "instance" InstanceGroup.define({ @@ -184,6 +206,11 @@ InstanceGroup.define({ }); InstanceGroup.finalize(); +/** + * Factorio config group for {@link module:lib/config.InstanceConfig} + * @extends module:lib/config.ConfigGroup + * @memberof module:lib/config + */ class FactorioGroup extends classes.ConfigGroup { } FactorioGroup.groupName = "factorio"; FactorioGroup.define({ @@ -221,6 +248,11 @@ FactorioGroup.define({ }); FactorioGroup.finalize(); +/** + * Instance config class + * @extends module:lib/config.Config + * @memberof module:lib/config + */ class InstanceConfig extends classes.Config { } InstanceConfig.registerGroup(InstanceGroup); InstanceConfig.registerGroup(FactorioGroup); @@ -242,6 +274,8 @@ function validateGroup(pluginInfo, groupName) { /** * Registers the config groups for the provided plugin infos + * + * @memberof module:lib/config */ function registerPluginConfigGroups(pluginInfos) { for (let pluginInfo of pluginInfos) { @@ -273,6 +307,8 @@ function registerPluginConfigGroups(pluginInfos) { /** * Lock configs from adding more groups and make them usable + * + * @memberof module:lib/config */ function finalizeConfigs() { MasterConfig.finalize(); @@ -285,6 +321,8 @@ function finalizeConfigs() { * * Can be passed to yargs.command to implement a config command. Use * handleConfigCommand to do the requested action. + * + * @memberof module:lib/config */ function configCommand(yargs) { yargs @@ -300,6 +338,8 @@ function configCommand(yargs) { * Handle yargs command * * Handle the actions that are made available by configCommand. + * + * @memberof module:lib/config */ async function handleConfigCommand(args, instance, configPath) { let command = args._[1]; diff --git a/lib/config/index.js b/lib/config/index.js index bbe1c364..f591436e 100644 --- a/lib/config/index.js +++ b/lib/config/index.js @@ -1,8 +1,8 @@ /** - * @file Configuration System + * Configuration System * @author Hornwitser + * @module lib/config */ - module.exports = { ...require("./classes"), ...require("./definitions"), diff --git a/lib/errors.js b/lib/errors.js index 19c9d27f..2b9a6bdd 100644 --- a/lib/errors.js +++ b/lib/errors.js @@ -3,10 +3,22 @@ * @module */ +/** + * Thrown by commands when they fail + * @static + */ class CommandError extends Error { } + +/** + * Thrown from requests sent when an error occured handling it + * @static + */ class RequestError extends Error { } -// Signal for messages that fail validation +/** + * Signal for messages that fail validation + * @static + */ class InvalidMessage extends Error { constructor(msg, errors) { super(msg); @@ -14,19 +26,34 @@ class InvalidMessage extends Error { } } -// Thrown from requests when the session it was sent on was lost. +/** + * Thrown from requests when the session it was sent on was lost + * @static + */ class SessionLost extends Error { } -// Thrown when WebSocket authentication failed +/** + * Thrown when WebSocket authentication failed + * @static + */ class AuthenticationFailed extends Error { } -// Errror class for known errors occuring during startup +/** + * Errror class for known errors occuring during startup + * @static + */ class StartupError extends Error { } -// Errors outside of our control. +/** + * Errors outside of our control + * @static + */ class EnvironmentError extends Error { } -// Errors caused by plugins +/** + * Errors caused by plugins + * @static + */ class PluginError extends Error { constructor(pluginName, original) { super(`PluginError: ${original.message}`); diff --git a/lib/factorio/index.js b/lib/factorio/index.js index 1e03bc70..b99f8803 100644 --- a/lib/factorio/index.js +++ b/lib/factorio/index.js @@ -1,3 +1,8 @@ +/** + * Library for interfacing with Factorio Servers and saves. + * @module lib/factorio + * @author Hornwitser + */ module.exports = { ...require('lib/factorio/locale'), ...require('lib/factorio/patch'), diff --git a/lib/factorio/locale.js b/lib/factorio/locale.js index 36ee8a1e..0bc6cef2 100644 --- a/lib/factorio/locale.js +++ b/lib/factorio/locale.js @@ -1,7 +1,4 @@ -/** - * Function for parsing factorio locale from a local factorio install - */ - +// Function for parsing factorio locale from a local factorio install "use strict"; const fs = require("fs-extra"); const path = require("path"); @@ -13,6 +10,7 @@ const ini = require("ini"); * @param {string} factorioDirectory * @param {string} languageCode * @param {function} callback + * @memberof module:lib/factorio */ async function getLocale(dataDir, languageCode) { if (typeof dataDir !== "string") throw new TypeError("dataDir must be a string"); diff --git a/lib/factorio/patch.js b/lib/factorio/patch.js index 2d5c1957..e7545be1 100644 --- a/lib/factorio/patch.js +++ b/lib/factorio/patch.js @@ -1,7 +1,4 @@ -/** - * Library for patching Factorio saves with scenario code. - * @author Hornwitser - */ +// Library for patching Factorio saves with scenario code. "use strict"; const events = require("events"); @@ -31,6 +28,9 @@ const knownScenarios = { * contained in. Throws an error if there are multiple such folders. * * @returns {string} name of the root folder. + * @memberof module:lib/factorio + * @private + * @inner */ function findRoot(zip) { let root = null; @@ -60,6 +60,9 @@ function findRoot(zip) { * Generates control.lua code for loading the Clusterio modules * * @param {Object} patchInfo - The patch info files's json content + * @memberof module:lib/factorio + * @private + * @inner */ function generateLoader(patchInfo) { let lines = [ @@ -107,6 +110,9 @@ function generateLoader(patchInfo) { * their dependents. Throws an error if this is not possible. * * @param {Array} modules - Array of modules to reorder + * @memberof module:lib/factorio + * @private + * @inner */ function reorderDependencies(modules) { let index = 0; @@ -202,6 +208,7 @@ function reorderDependencies(modules) { * * @param {string} savePath - Path to the Factorio save to patch. * @param {Array} modules - Description of the modules to patch. + * @memberof module:lib/factorio */ async function patch(savePath, modules) { let zip = await jszip.loadAsync(await fs.readFile(savePath)); diff --git a/lib/factorio/server.js b/lib/factorio/server.js index 8f7a6540..bf31d557 100644 --- a/lib/factorio/server.js +++ b/lib/factorio/server.js @@ -1,9 +1,4 @@ -/** - * Factorio interfaces - * - * @module - */ - +// Factorio Server interface "use strict"; const fs = require("fs-extra"); const child_process = require("child_process"); @@ -20,6 +15,10 @@ const errors = require("lib/errors"); /** * Determines the version of Factorio the datadir is pointing to by * reading the changelog.txt in it. + * + * @memberof module:lib/factorio + * @private + * @inner */ async function getVersion(changelogPath) { let changelog; @@ -47,6 +46,10 @@ async function getVersion(changelogPath) { /** * Comparison function for sorting Factorio version strings + * + * @memberof module:lib/factorio + * @private + * @inner */ function versionOrder(a, b) { let aParts = a.split(".").map(s => Number.parseInt(s, 10)); @@ -70,6 +73,10 @@ function versionOrder(a, b) { * * Searches the given factorio dir for an installation of Factorio of * the target version. + * + * @memberof module:lib/factorio + * @private + * @inner */ async function findVersion(factorioDir, targetVersion) { @@ -125,6 +132,9 @@ async function findVersion(factorioDir, targetVersion) { * RFC 6335. * * @return {number} a number in the range 49152 to 65535. + * @memberof module:lib/factorio + * @private + * @inner */ function randomDynamicPort() { const start = 49152; @@ -141,6 +151,9 @@ function randomDynamicPort() { * * @param {number} length - the length of the password to generate. * @return {string} password of the given length + * @memberof module:lib/factorio + * @private + * @inner */ async function generatePassword(length) { function validChar(byte) { @@ -179,6 +192,10 @@ async function generatePassword(length) { * Data is streamed via the data method, and once the stream ends the * end method should be called. This ensures an unterminated line at * the end of the stream is passed to the callback. + * + * @memberof module:lib/factorio + * @private + * @inner */ class LineSplitter { @@ -253,6 +270,9 @@ class LineSplitter { * @param {string} source - Passed into the output structure as source * * @returns {object} - An object with interpeted data. + * @memberof module:lib/factorio + * @private + * @inner */ function parseOutput(line, source) { let output = { @@ -425,15 +445,17 @@ const outputHeuristics = [ * stopping and sending commands to the server. It does not deal with * creating or managing servers, or downloading Factorio. * - * This is an events emitter with the following events: - * stdout - invoked when Factorio outputs a line to stdout - * stderr - invoked when Factorio outputs a line to stderr - * output - invoked with parsed output from Factorio - * rcon-ready - invoked when the RCON client has connected - * game-ready - invoked when the server is finished starting up - * autosave-start - invoked when the server starts an autosave - * autosave-fnished - invoked when the autosave finished - * exit - invoked when the sterver has exited + * This is an EventEmitter with the following events: + * - stdout - invoked when Factorio outputs a line to stdout + * - stderr - invoked when Factorio outputs a line to stderr + * - output - invoked with parsed output from Factorio + * - rcon-ready - invoked when the RCON client has connected + * - game-ready - invoked when the server is finished starting up + * - autosave-start - invoked when the server starts an autosave + * - autosave-fnished - invoked when the autosave finished + * - exit - invoked when the sterver has exited + * @extends events.EventEmitter + * @memberof module:lib/factorio */ class FactorioServer extends events.EventEmitter { /** @@ -442,6 +464,12 @@ class FactorioServer extends events.EventEmitter { * @param {string} factorioDir - Directory of the Factorio install(s). * @param {string} writeDir - Directory to write runtime data to. * @param {object} options - Optional parameters. + * @param {string} options.version - + * Version of Factorio to use. Can also be the string "latest" to + * use the latest version found in `factorioDir`. + * @param {number} options.gamePort - UDP port to host game on. + * @param {number} options.rconPort - TCP port to use for RCON. + * @param {string} options.rconPassword - Password use for RCON. */ constructor(factorioDir, writeDir, options) { super(); @@ -454,8 +482,11 @@ class FactorioServer extends events.EventEmitter { this._dataDir = null; this._targetVersion = options.version || "latest"; + /** UDP port used for hosting the Factorio game server on */ this.gamePort = options.gamePort || randomDynamicPort(); + /** TCP port used for RCON on the Factorio game server */ this.rconPort = options.rconPort || randomDynamicPort(); + /** Password used for RCON on the Factorio game server */ this.rconPassword = options.rconPassword; this._state = "new"; this._server = null; @@ -603,6 +634,7 @@ class FactorioServer extends events.EventEmitter { * Initialize class instance * * Must be called before instances of this class can be used. + * @throws {Error} if the requested version of Factorio was not found. */ async init() { this._check(["new"]); @@ -612,7 +644,9 @@ class FactorioServer extends events.EventEmitter { } /** - * The version of Factorio that was detected + * The version of Factorio in use. This will be the actual version if + * "latest" (the default) was specified as `factorioVersion` to the + * constructor. */ get version() { return this._version; diff --git a/lib/fileOps/fileOps.js b/lib/fileOps/fileOps.js index 251991a3..ed42b794 100644 --- a/lib/fileOps/fileOps.js +++ b/lib/fileOps/fileOps.js @@ -1,3 +1,6 @@ +/** + * @module lib/fileOps + */ const fs = require("fs-extra"); const path = require("path"); diff --git a/lib/hash.js b/lib/hash.js index 9cf9c4bf..9bb694ed 100644 --- a/lib/hash.js +++ b/lib/hash.js @@ -1,3 +1,7 @@ +/** + * Hashing functions + * @module + */ const crypto = require('crypto'); const fs = require('fs'); @@ -6,6 +10,7 @@ const fs = require('fs'); * Returns a promise that resolves to the SHA1 hash of the stream given * * @param stream - Node stream of the content to hash. + * @returns {Promise} hash of the stream. */ function hashStream(stream) { return new Promise(function(resolve, reject) { @@ -28,6 +33,7 @@ function hashStream(stream) { * Returns a promise that resolves to the SHA1 hash of the file given by path * * @param {string} path - Path to the file to hash. + * @returns {Promise} hash of the file given. */ function hashFile(path) { return hashStream(fs.createReadStream(path)); diff --git a/lib/link/connectors.js b/lib/link/connectors.js index d7afe5fc..1b6b4d78 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -1,3 +1,4 @@ +// Connection adapters for Links "use strict"; const events = require("events"); const WebSocket = require("ws"); @@ -6,6 +7,12 @@ const schema = require("lib/schema"); const errors = require("lib/errors"); +/** + * Base connector for links + * + * @extends events.EventEmitter + * @memberof module:lib/link + */ class WebSocketBaseConnector extends events.EventEmitter { constructor() { super(); @@ -112,6 +119,9 @@ class WebSocketBaseConnector extends events.EventEmitter { /** * Connector for master server clients + * + * @extends module:lib/link.WebSocketBaseConnector + * @memberof module:lib/link */ class WebSocketClientConnector extends WebSocketBaseConnector { constructor(url, reconnectDelay) { @@ -397,6 +407,12 @@ class WebSocketClientConnector extends WebSocketBaseConnector { } } +/** + * Connector for in-memory local links + * + * @extends events.EventEmitter + * @memberof module:lib/link + */ class VirtualConnector extends events.EventEmitter { constructor() { super(); diff --git a/lib/link/index.js b/lib/link/index.js index 132ff5d0..e5c18c5f 100644 --- a/lib/link/index.js +++ b/lib/link/index.js @@ -1,3 +1,8 @@ +/** + * Library for message based communication between nodes + * @module lib/link + * @author Hornwitser + */ module.exports = { ...require('lib/link/link'), ...require('lib/link/messages'), diff --git a/lib/link/link.js b/lib/link/link.js index d36d8b56..347efef4 100644 --- a/lib/link/link.js +++ b/lib/link/link.js @@ -1,3 +1,4 @@ +// Implementation of Link class "use strict"; const schema = require("lib/schema"); @@ -15,6 +16,8 @@ const messages = require("lib/link/messages"); /** * Common interface for server and client connections + * + * @memberof module:lib/link */ class Link { constructor(source, target, connector) { @@ -59,7 +62,7 @@ class Link { * Validates and invokes the handler and/or waiters for a message that has * been received. An unhandled message is considered to be an error. * - * @throws {InvalidMessage} + * @throws {module:lib/errors.InvalidMessage} * if validation failed, is missing, the message is invalid, or no * handlers/waiters were invoked to handle the message. */ diff --git a/lib/link/messages.js b/lib/link/messages.js index 365bfa6a..8edf569e 100644 --- a/lib/link/messages.js +++ b/lib/link/messages.js @@ -1,3 +1,4 @@ +// Message definitions for links "use strict"; const schema = require("lib/schema"); const errors = require("lib/errors"); @@ -7,6 +8,8 @@ const errors = require("lib/errors"); * * A request-response message that can be innitated by either party of the * link. + * + * @memberof module:lib/link */ class Request { @@ -353,6 +356,8 @@ messages.getMetrics = new Request({ * * A one way message without any response or recipt confirmation that * can be innitiated by either party of the link. + * + * @memberof module:lib/link */ class Event { @@ -555,6 +560,8 @@ messages.instanceOutput = new Event({ * to the link. The handler used for the messageis looked up on the * link instance as the concatenation of the name of the message, the * name of message class, and 'Handler'. + * + * @memberof module:lib/link */ function attachAllMessages(link) { for (let [name, message] of Object.entries(messages)) { diff --git a/lib/luaTools.js b/lib/luaTools.js index 8165beb5..7566e03e 100644 --- a/lib/luaTools.js +++ b/lib/luaTools.js @@ -1,3 +1,8 @@ +/** + * Utilities for dealing with Lua + * @module + */ + /** * Escapes a string for inclusion into a lua string */ diff --git a/lib/plugin.js b/lib/plugin.js index 41661db3..7dd6a3ad 100644 --- a/lib/plugin.js +++ b/lib/plugin.js @@ -11,6 +11,7 @@ const errors = require("lib/errors"); /** * Base class for instance plugins + * @static */ class BaseInstancePlugin { constructor(info, instance, slave) { @@ -21,7 +22,7 @@ class BaseInstancePlugin { /** * Instance the plugin started for - * @type {Instance} + * @type {module:slave~Instance} */ this.instance = instance; @@ -31,7 +32,7 @@ class BaseInstancePlugin { * With the exepction of accessing the slave's config you should * avoid ineracting with the slave object directly. * - * @type {Slave} + * @type {module:slave~Slave} */ this.slave = slave; } @@ -95,6 +96,10 @@ class BaseInstancePlugin { async onOutput(output) { } } +/** + * Base class for master plugins + * @static + */ class BaseMasterPlugin { constructor(info, master, metrics) { this.info = info; @@ -138,6 +143,7 @@ class BaseMasterPlugin { * function be called again. * * @returns {Map} mapping of plugin name to info module. + * @static */ async function loadPluginInfos(baseDir) { let plugins = []; @@ -167,6 +173,16 @@ async function loadPluginInfos(baseDir) { return plugins; } +/** + * Attach plugin messages + * + * Attaches all messages defined in the `.message` property of the + * `pluginInfo` passed with handlers taken from `pluin`. + * + * @param {module:lib/link.Link} link - Link to attach handlers to. + * @param {object} pluginInfo - Plugin info object. + * @param plugin - The instance of the plugin to use handlers from. + */ function attachPluginMessages(link, pluginInfo, plugin) { let messageDefinitions = pluginInfo.messages || []; for (let [name, messageFormat] of Object.entries(messageDefinitions)) { diff --git a/lib/schema.js b/lib/schema.js index 07e649f6..84fc6de0 100644 --- a/lib/schema.js +++ b/lib/schema.js @@ -1,3 +1,7 @@ +/** + * JSON schemas used for validating link messages + * @module + */ const ajv = new (require('ajv'))({ verbose: true, format: "full", @@ -24,8 +28,20 @@ const messageSchema = { }, }; +/** + * Validator for message schemas + * @function + * @param {Object} obj - Object to validate + * @returns {boolean} True if object passed validation. + */ const message = ajv.compile(messageSchema); +/** + * Validator for heartbeat messages + * @function + * @param {Object} obj - Object to validate + * @returns {boolean} True if object passed validation. + */ const heartbeat = ajv.compile({ $schema: "http://json-schema.org/draft-07/schema#", title: "Heartbeat", @@ -47,6 +63,12 @@ const heartbeat = ajv.compile({ }, }); +/** + * Validator for server sent part of handshake + * @function + * @param {Object} obj - Object to validate + * @returns {boolean} True if object passed validation. + */ const serverHandshake = ajv.compile({ $schema: "http://json-schema.org/draft-07/schema#", title: "Server Handshake", @@ -104,6 +126,12 @@ const serverHandshake = ajv.compile({ ], }); +/** + * Validator for client sent part of handshake + * @function + * @param {Object} obj - Object to validate + * @returns {boolean} True if object passed validation. + */ const clientHandshake = ajv.compile({ $schema: "http://json-schema.org/draft-07/schema#", title: "Client Message", @@ -162,6 +190,12 @@ const clientHandshake = ajv.compile({ module.exports = { + /** + * Compile JSON schema into validator + * @function + * @param {Object} schema - JSON schema to create validator for. + * @returns {Function} Validator for the schema. + */ compile: ajv.compile.bind(ajv), message, diff --git a/master.js b/master.js index f2dba036..b6885823 100644 --- a/master.js +++ b/master.js @@ -1,13 +1,16 @@ /** -Clusterio master server. Facilitates communication between slaves through -a webserver, storing data related to slaves like production graphs. - -@module -@author Danielv123 - -@example -node master -*/ + * Clusterio master server + * + * Facilitates communication between slaves and control of the cluster + * through WebSocet connections, and hosts a webserver for browser + * interfaces and Prometheus statistics export. It is remotely controlled + * by {@link module:clusterctl}. + * + * @module + * @author Danielv123, Hornwitser + * @example + * node master run + */ // Attempt updating // const updater = require("./updater"); @@ -345,6 +348,11 @@ app.get("/api/getFactorioLocale", function(req,res){ } }); +/** + * Base class for master server connections + * + * @extends module:lib/link.Link + */ class BaseConnection extends link.Link { constructor(target, connector) { super('master', target, connector); @@ -574,6 +582,11 @@ class ControlConnection extends BaseConnection { } var slaveConnections = new Map(); +/** + * Represents the connection to a slave + * + * @extends module:master~BaseConnection + */ class SlaveConnection extends BaseConnection { constructor(registerData, connector) { super('slave', connector); @@ -647,7 +660,8 @@ const wss = new WebSocket.Server({ * * @param value - value to test. * @returns {Boolean} - * true if value is an integer between -2**31 and 2**31-1. + * true if value is an integer between -231 and + * 231-1. */ function isInteger(value) { return value | 0 === value; @@ -656,6 +670,8 @@ function isInteger(value) { /** * Connector for master server connections + * + * @extends module:lib/link.WebSocketBaseConnector */ class WebSocketServerConnector extends link.WebSocketBaseConnector { constructor(socket, sessionId) { diff --git a/plugins/global_chat/instance.js b/plugins/global_chat/instance.js index e44ed210..2674a03d 100644 --- a/plugins/global_chat/instance.js +++ b/plugins/global_chat/instance.js @@ -1,3 +1,6 @@ +/** + * @module + */ const plugin = require("lib/plugin"); const luaTools = require("lib/luaTools"); diff --git a/slave.js b/slave.js index 78041eb1..716d092e 100644 --- a/slave.js +++ b/slave.js @@ -1,3 +1,15 @@ +/** + * Clusterio slave + * + * Connects to the master server and hosts Factorio servers that can + * communicate with the cluster. It is remotely controlled by {@link + * module:master}. + * + * @module + * @author Danielv123, Hornwitser + * @example + * node slave run + */ const fs = require('fs-extra'); const path = require('path'); const yargs = require("yargs"); From 1beffb1bfd74565349b6c45772fd8a649be2f696 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Fri, 8 May 2020 18:09:48 +0200 Subject: [PATCH 05/14] Fix crash on hartbeat on closing connection Fix heartbeat incorrectly asserting it cannot be sent on a connection in the closing state. --- lib/link/connectors.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/link/connectors.js b/lib/link/connectors.js index 1b6b4d78..764c5745 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -43,7 +43,7 @@ class WebSocketBaseConnector extends events.EventEmitter { } _doHeartbeat() { - this._check(["connected"]); + this._check(["connected", "closing"]); if (Date.now() - this._lastHeartbeat > 2000 * this._heartbeatInterval) { console.log("SOCKET | closing after heartbeat timed out"); this._socket.close(1008, "Heartbeat timeout"); From b8b8c440836caeef52f6ee3eaacd0e5c2a6548f6 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Fri, 8 May 2020 18:15:17 +0200 Subject: [PATCH 06/14] Fix onExit not being invoked if plugin load fails Fix instance not properly exiting if an error occurs during the loading of a plugin. --- lib/plugin.js | 6 +++++- slave.js | 26 +++++++++++++++++--------- 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/lib/plugin.js b/lib/plugin.js index 7dd6a3ad..9dde8227 100644 --- a/lib/plugin.js +++ b/lib/plugin.js @@ -73,7 +73,11 @@ class BaseInstancePlugin { async onStop() { } /** - * Called when the Factorio server exited + * Called when the instance exits + * + * Invoked when the instance is shut down. This may occur after init + * has been called if an error occurs during startup. Note that if + * the plugin's init() throws this method will still be invoked. */ onExit() { } diff --git a/slave.js b/slave.js index 716d092e..75f9fdc8 100644 --- a/slave.js +++ b/slave.js @@ -129,6 +129,17 @@ class Instance extends link.Link{ } } + async _loadPlugin(pluginInfo, slave) { + let pluginLoadStarted = Date.now(); + let { InstancePlugin } = require(`./plugins/${pluginInfo.name}/${pluginInfo.instanceEntrypoint}`); + let instancePlugin = new InstancePlugin(pluginInfo, this, slave); + this.plugins.set(pluginInfo.name, instancePlugin); + await instancePlugin.init(); + plugin.attachPluginMessages(this, pluginInfo, instancePlugin); + + console.log(`Clusterio | Loaded plugin ${pluginInfo.name} in ${Date.now() - pluginLoadStarted}ms`); + } + async init(pluginInfos, masterPlugins, slave) { await this.server.init(); @@ -142,15 +153,12 @@ class Instance extends link.Link{ continue; } - // require plugin class and initialize it - let pluginLoadStarted = Date.now(); - let { InstancePlugin } = require(`./plugins/${pluginInfo.name}/${pluginInfo.instanceEntrypoint}`); - let instancePlugin = new InstancePlugin(pluginInfo, this, slave); - await instancePlugin.init(); - this.plugins.set(pluginInfo.name, instancePlugin); - plugin.attachPluginMessages(this, pluginInfo, instancePlugin); - - console.log(`Clusterio | Loaded plugin ${pluginInfo.name} in ${Date.now() - pluginLoadStarted}ms`); + try { + await this._loadPlugin(pluginInfo, slave); + } catch (err) { + this.notifyExit(); + throw err; + } } } From c35b6de70e4b90229897f6e53b3e9384c800ec63 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Fri, 8 May 2020 23:49:28 +0200 Subject: [PATCH 07/14] Notify on connection events Implement notification to plugins for disconnect request, connect, drop, and close. This allows plugins to properly handle slaves and the master server shutting down without sending messages that will get lost. --- docs/writing-plugins.md | 46 ++++++++++++ lib/link/connectors.js | 14 ++++ lib/link/link.js | 10 +++ lib/link/messages.js | 12 ++++ lib/plugin.js | 110 +++++++++++++++++++++++++++++ master.js | 21 ++++-- plugins/global_chat/instance.js | 29 +++++--- plugins/global_chat/test/plugin.js | 2 +- plugins/research_sync/master.js | 19 ++--- plugins/subspace_storage/master.js | 4 +- slave.js | 28 ++++++++ test/lib/plugin.js | 4 ++ test/mock.js | 10 +++ 13 files changed, 278 insertions(+), 31 deletions(-) diff --git a/docs/writing-plugins.md b/docs/writing-plugins.md index 15710824..5eddd0e5 100644 --- a/docs/writing-plugins.md +++ b/docs/writing-plugins.md @@ -17,6 +17,7 @@ Contents - [Defining Events](#defining-events) - [Defining Requests](#defining-requests) - [Sending Link Messages](#sending-link-messages) + - [Handling connection events](#handling-connection-events) - [Collecting Statistics](#collecting-statistics) @@ -444,6 +445,51 @@ data received from the target it was sent to, or throws an error if the request failed. +### Handling connection events + +There are a few connection related events that plugins neeed to repsond +to in order to avoid data loss and connection problems. The most +important is the prepare disconnect for the link between master and +slave. This is signaled to `MasterPlugin` classes via the +`onPrepareSlaveDisconnect` hook and to `InstancePlugin` classes via the +`onPrepareMasterDisconnect` hook. + +After the prepare disconnect the connection will be closed, which will +result in pending requests and events being dropped. Plugins must +respond to the prepare disconnect by stopping any processess it does +that send events or requests over the link in question. This can be +accomplished either through listening for the prepare disconnect hook, +or by checking the `connected` or `closing` property of the connector +for the master/slave connection. For example the sending of an event +from an `InstancePlugin` class can be stopped while the connection is +closing by using the following code: + + if (!this.slave.connector.closing) { + this.info.messages.frobnicate.send(this.instance, { foo: "bar" }); + } + +If the event or request needs to be sent to the master it can be put +into a queue stored on the plugin instance and sent out when the +connection is established again. The re-establishement of the +connection is notified to plugins via the `connect` event to the +`onMasterConnectionEvent` and `onSlaveConnectionEvent` hooks. + +The second connection event which is of lesser importance to respond to +is the `drop` connection event served through `onMasterConnectionEvent` +for `InstancePlugin` classes and through `onSlaveConnectionEvent` for +`MasterPlugin` classes. This is raised when the connection between the +master and slave in question is lost, most likely due to networking +issues. When in the dropped state the slave will keep trying to +reconnect to the master server in order to re-establish it, and if +successful no events or requests will be lost. However while in the +dropped state any requests and events sent gets queued up in memory +until the connection is re-established. This means that if your plugin +sends a lot of events or requests, they can end up being queued up in a +buffer for a long time and sent out all at once. To avoid this you +should be throtteling and/or stopping your requests/events after `drop` +has been raised, and continue back as normal when `connect` is raised. + + Collecting Statistics --------------------- diff --git a/lib/link/connectors.js b/lib/link/connectors.js index 764c5745..929786e3 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -115,6 +115,20 @@ class WebSocketBaseConnector extends events.EventEmitter { setClosing() { this._state = "closing"; } + + /** + * True if the connection is established and active + */ + get connected() { + return this._state === "connected"; + } + + /** + * True if the connection is closing down or closed + */ + get closing() { + return ["closing", "closed"].includes(this._state); + } } /** diff --git a/lib/link/link.js b/lib/link/link.js index 347efef4..67d2fc37 100644 --- a/lib/link/link.js +++ b/lib/link/link.js @@ -54,6 +54,16 @@ class Link { this._waiters.clear(); }); + + connector.on("close", () => { + for (let waiterType of this._waiters.values()) { + for (let waiter of waiterType) { + waiter.reject(new errors.SessionLost("Session Closed")); + } + } + + this._waiters.clear(); + }); } /** diff --git a/lib/link/messages.js b/lib/link/messages.js index 8edf569e..fdcefc3d 100644 --- a/lib/link/messages.js +++ b/lib/link/messages.js @@ -193,6 +193,11 @@ messages.prepareDisconnect = new Request({ links: wsLinks, }); +messages.prepareMasterDisconnect = new Request({ + type: "prepare_master_disconnect", + links: ["slave-instance"], +}); + messages.ping = new Request({ type: "ping", links: wsLinks, @@ -552,6 +557,13 @@ messages.instanceOutput = new Event({ }, }); +messages.masterConnectionEvent = new Event({ + type: "master_connection_event", + links: ["slave-instance"], + eventProperties: { + "event": { type: "string" }, + }, +}); /** * Attaches all requests and events to the given link diff --git a/lib/plugin.js b/lib/plugin.js index 9dde8227..3b6a5440 100644 --- a/lib/plugin.js +++ b/lib/plugin.js @@ -98,6 +98,53 @@ class BaseInstancePlugin { * - message: Main content of the line. */ async onOutput(output) { } + + /** + * Called when an event on the master connection happens + * + * The event param may be one of connect, drop and close and has the + * following meaning: + * + * ##### connect + * + * Invoked when a new connection to the master has been established, or + * the existing one that have previously been dropped is re-established. + * + * ##### drop + * + * Invoked when a connection loss is detected between the slave and the + * master. Plugins should respond to this event by throtteling messages + * it is sending to the master to an absolute minimum. + * + * Messages sent over a dropped master connection will get queued up in + * memory on the slave and sent all in one go when the connection is + * re-established again. + * + * ##### close + * + * Invoked when the connection to the master has been closed. This + * typically means the master server has shut down. Plugins should not + * send any messages that goes to or via the master server after the + * connection has been closed and before a new one is established. + * + * @param {string} event - one of connect, drop, and close + */ + onMasterConnectionEvent(event) { } + + /** + * Called when the master is preparing to disconnect from the slave + * + * Invoked when the master server has requested the disconnection from + * the slave. This typically happens when it is in the process of + * shutting down. + * + * Plugins must stop sending messages to the master, or are forwarded + * via the master server after the prepare disconnect has been handled. + * + * @param {module:master~SlaveConnection} connection - + * The connection to the slave preparing to disconnect. + */ + async onPrepareMasterDisconnect(connection) { } } /** @@ -137,6 +184,69 @@ class BaseMasterPlugin { * Called when the master server is shutting down */ async onShutdown() { } + + /** + * Called when an event on a slave connection happens + * + * The event param may be one of connect, drop and close and has the + * following meaning: + * + * ##### connect + * + * Invoked when a new connection to the slave has been established, or + * an existing one that had previously been dropped is re-established. + * + * ##### drop + * + * Invoked when a connection loss is detected between the master and a + * slave. Plugins should respond to this event by throtteling messages + * it is sending to the given slave connection to an absolute minimum. + * + * Messages sent over a dropped slave connection will get queued up in + * memory on the master and sent all in one go when the connection is + * re-established again. + * + * ##### close + * + * Invoked when the connection to the slave has been closed. + * + * @param {string} event - one of connect, drop, and close + * @param {module:master~SlaveConnection} connection - + * The connection the event occured on. + */ + onSlaveConnectionEvent(connection, event) { } + + /** + * Called when a slave is preparing to disconnect from the master + * + * Invoked when a slave has requested the disconnection from the master. + * This typically happens when it is in the process of shutting down. + * + * Plugins must stop sending messages to the slave in question after the + * prepare disconnect has been handled. + * + * @param {module:master~SlaveConnection} connection - + * The connection to the slave preparing to disconnect. + */ + async onPrepareSlaveDisconnect(connection) { } + + /** + * Broadcast event to all connected slaves + * + * Sends the given event to all slaves connected to the master server. + * This does not include slaves that are in the process of closing the + * connection, which typically happens when they are shutting down. + * + * @param {module:lib/link.Event} event - Event to send + * @param {Object} data - Data ta pass with the event. + */ + broadcastEventToSlaves(event, data={}) { + for (let slaveConnection of this.master.slaveConnections.values()) { + if (!slaveConnection.connector.closing) { + event.send(slaveConnection, data); + } + } + } } /** diff --git a/master.js b/master.js index b6885823..8dc6b9dc 100644 --- a/master.js +++ b/master.js @@ -388,15 +388,15 @@ class BaseConnection extends link.Link { let slaveId = instanceConfig.get('instance.assigned_slave'); if (slaveId === null) { return; } - let connection = slaveConnection.get(slaveId); - if (!connection) { return; } + let connection = slaveConnections.get(slaveId); + if (!connection || connection.closing) { return; } event.send(connection, message.data); } async broadcastEventToInstance(message, event) { for (let slaveConnection of slaveConnections.values()) { - if (slaveConnection === this) { + if (slaveConnection === this || slaveConnection.connector.closing) { continue; // Do not broadcast back to the source } @@ -405,6 +405,9 @@ class BaseConnection extends link.Link { } async prepareDisconnectRequestHandler(message, request) { + for (let masterPlugin of masterPlugins.values()) { + await masterPlugin.onPrepareSlaveDisconnect(this); + } this.connector.setClosing(); return await super.prepareDisconnectRequestHandler(message, request); } @@ -574,7 +577,9 @@ class ControlConnection extends BaseConnection { async debugDumpWsRequestHandler(message) { this.ws_dumper = data => { - link.messages.debugWsMessage.send(this, data); + if (this.connector.connected) { + link.messages.debugWsMessage.send(this, data); + } }; this.connector._socket.clusterio_ignore_dump = true; debugEvents.on("message", this.ws_dumper) @@ -610,6 +615,14 @@ class SlaveConnection extends BaseConnection { slaveConnections.delete(this._id); } }); + + for (let event of ["connect", "drop", "close"]) { + this.connector.on(event, () => { + for (let masterPlugin of masterPlugins.values()) { + masterPlugin.onSlaveConnectionEvent(this, event); + } + }); + } } async updateInstancesEventHandler(message) { diff --git a/plugins/global_chat/instance.js b/plugins/global_chat/instance.js index 2674a03d..cf6cfbee 100644 --- a/plugins/global_chat/instance.js +++ b/plugins/global_chat/instance.js @@ -13,12 +13,17 @@ function removeTags(content) { } class InstancePlugin extends plugin.BaseInstancePlugin { - constructor(...args) { - super(...args); + async init() { + this.messageQueue = []; } - async init() { - // XXX Nothing? + onMasterConnectionEvent(event) { + if (event === "connect") { + for (let message of this.messageQueue) { + this.sendChat(message); + } + this.messageQueue = []; + } } async chatEventHandler(message) { @@ -27,12 +32,20 @@ class InstancePlugin extends plugin.BaseInstancePlugin { await this.instance.server.sendRcon(`/sc game.print('${luaTools.escapeString(content)}')`, true); } + sendChat(message) { + this.info.messages.chat.send(this.instance, { + instance_name: this.instance.name, + content: message, + }); + } + async onOutput(output) { if (output.type === "action" && output.action === "CHAT") { - this.info.messages.chat.send(this.instance, { - instance_name: this.instance.name, - content: output.message, - }); + if (this.slave.connector.connected) { + this.sendChat(output.message); + } else { + this.messageQueue.push(output.message); + } } } diff --git a/plugins/global_chat/test/plugin.js b/plugins/global_chat/test/plugin.js index 28590068..1652282e 100644 --- a/plugins/global_chat/test/plugin.js +++ b/plugins/global_chat/test/plugin.js @@ -23,7 +23,7 @@ describe("global_chat plugin", function() { let instancePlugin; before(async function() { - instancePlugin = new instance.InstancePlugin(info, new mock.MockInstance()); + instancePlugin = new instance.InstancePlugin(info, new mock.MockInstance(), new mock.MockSlave()); await instancePlugin.init(); }); diff --git a/plugins/research_sync/master.js b/plugins/research_sync/master.js index 4e141be9..fa114503 100644 --- a/plugins/research_sync/master.js +++ b/plugins/research_sync/master.js @@ -49,9 +49,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin { this.lastProgressBroadcast = Date.now(); if (techs.length) { - for (let slaveConnection of this.master.slaveConnections.values()) { - this.info.messages.progress.send(slaveConnection, { technologies: techs }); - } + this.broadcastEventToSlaves(this.info.messages.progress, { technologies: techs }); } } @@ -89,12 +87,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin { tech.progress = null; this.progressToBroadcast.delete(name); - for (let slaveConnection of this.master.slaveConnections.values()) { - this.info.messages.finished.send(slaveConnection, { - name, - level: tech.level - }); - } + this.broadcastEventToSlaves(this.info.messages.finished, { name, level: tech.level }); } } @@ -125,9 +118,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin { if (progress) { this.progressToBroadcast.add(name); } else if (researched || baseLevel(name) != level) { - for (let slaveConnection of this.master.slaveConnections.values()) { - this.info.messages.finished.send(slaveConnection, { name, level }); - } + this.broadcastEventToSlaves(this.info.messages.finished, { name, level }); } } else { @@ -138,9 +129,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin { if (tech.level < level || researched) { // Send update if the unlocked level is greater if (level - !researched > tech.level - !tech.researched) { - for (let slaveConnection of this.master.slaveConnections.values()) { - this.info.messages.finished.send(slaveConnection, { name, level: level - !researched }); - } + this.broadcastEventToSlaves(this.info.messages.finished, { name, level: level - !researched }); } tech.level = level; tech.progress = progress; diff --git a/plugins/subspace_storage/master.js b/plugins/subspace_storage/master.js index 64a10a3b..928757b7 100644 --- a/plugins/subspace_storage/master.js +++ b/plugins/subspace_storage/master.js @@ -77,9 +77,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin { } itemsToUpdate = [...itemsToUpdate.entries()]; - for (let slaveConnection of this.master.slaveConnections.values()) { - this.info.messages.updateStorage.send(slaveConnection, { items: itemsToUpdate }); - } + this.broadcastEventToSlaves(this.info.messages.updateStorage, { items: itemsToUpdate }); this.itemsLastUpdate = new Map(this.items._items.entries()); } diff --git a/slave.js b/slave.js index 75f9fdc8..b8097ba8 100644 --- a/slave.js +++ b/slave.js @@ -360,6 +360,18 @@ class Instance extends link.Link{ } } + async masterConnectionEventEventHandler(message) { + for (let pluginInstance of this.plugins.values()) { + pluginInstance.onMasterConnectionEvent(message.data.event); + } + } + + async prepareMasterDisconnectRequestHandler() { + for (let pluginInstance of this.plugins.values()) { + await pluginInstance.onPrepareMasterDisconnect(); + } + } + async getMetricsRequestHandler() { let results = [] if (this._running) { @@ -582,6 +594,14 @@ class Slave extends link.Link { this.instanceConnections = new Map(); this.instanceInfos = new Map(); + + for (let event of ["connect", "drop", "close"]) { + this.connector.on(event, () => { + for (let instanceConnection of this.instanceConnections.values()) { + link.messages.masterConnectionEvent.send(instanceConnection, { event }); + } + }); + } } async _findNewInstanceDir(name) { @@ -770,6 +790,14 @@ class Slave extends link.Link { await this.updateInstances(); } + async prepareDisconnectRequestHandler(message, request) { + for (let instanceConnection of this.instanceConnections.values()) { + await link.messages.prepareMasterDisconnect.send(instanceConnection); + } + this.connector.setClosing(); + return await super.prepareDisconnectRequestHandler(message, request); + } + /** * Stops all instances and closes the connection */ diff --git a/test/lib/plugin.js b/test/lib/plugin.js index 92cd772a..4798f1a4 100644 --- a/test/lib/plugin.js +++ b/test/lib/plugin.js @@ -21,6 +21,8 @@ describe("lib/plugin", function() { await instancePlugin.onStop(); instancePlugin.onExit(); await instancePlugin.onOutput({}); + instancePlugin.onMasterConnectionEvent("connect"); + instancePlugin.onPrepareMasterDisconnect(); }) }); @@ -33,6 +35,8 @@ describe("lib/plugin", function() { it("should define defaults for hooks", async function() { await masterPlugin.onMetrics(); await masterPlugin.onShutdown(); + masterPlugin.onSlaveConnectionEvent({}, "connect"); + masterPlugin.onPrepareSlaveDisconnect({}); }) }); diff --git a/test/mock.js b/test/mock.js index 1fcefdab..f2354628 100644 --- a/test/mock.js +++ b/test/mock.js @@ -34,6 +34,9 @@ class MockConnector extends events.EventEmitter { this.sentMessages = []; this.events = new Map(); this.handshake = { address: "socket.test" }; + + this.connected = true; + this.closing = false; } send(type, data) { @@ -70,9 +73,16 @@ class MockInstance extends link.Link { } } +class MockSlave extends link.Link { + constructor() { + super('slave', 'master', new MockConnector()); + } +} + module.exports = { MockSocket, MockConnector, MockInstance, + MockSlave, }; From c87538ca58f18dddc9e287dc3f20b70eb35b7a31 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Sat, 9 May 2020 16:24:02 +0200 Subject: [PATCH 08/14] Handle connection being closed in slave Listen for the close event and connect again if the connection was closed from proper disconnect or initiate a shutdown if the connection was unexpectedly closed. Unexpected closure happens from authentication failure and if a connection fails to reconnect before the timeout. --- slave.js | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/slave.js b/slave.js index b8097ba8..b2ad1c7a 100644 --- a/slave.js +++ b/slave.js @@ -15,6 +15,7 @@ const path = require('path'); const yargs = require("yargs"); const events = require("events"); const pidusage = require("pidusage"); +const setBlocking = require("set-blocking"); const version = require("./package").version; // internal libraries @@ -595,6 +596,37 @@ class Slave extends link.Link { this.instanceConnections = new Map(); this.instanceInfos = new Map(); + this._disconnecting = false; + this._shuttingDown = false; + this.connector.on("close", () => { + if (this._shuttingDown) { + return; + } + + if (this._disconnecting) { + this._disconnecting = false; + this.connector.connect().catch((err) => { + console.error("ERROR: Unexpected error reconnecting to master"); + console.error(err); + this.shutdown().catch((err) => { + setBlocking(true); + console.error("ERROR: Unexpected error during shutdown"); + console.error(err); + process.exit(1) + }); + }); + + } else { + console.error("ERROR: Master connection was unexpectedly closed"); + this.shutdown().catch((err) => { + setBlocking(true); + console.error("ERROR: Unexpected error during shutdown"); + console.error(err); + process.exit(1) + }); + } + }); + for (let event of ["connect", "drop", "close"]) { this.connector.on(event, () => { for (let instanceConnection of this.instanceConnections.values()) { @@ -791,6 +823,7 @@ class Slave extends link.Link { } async prepareDisconnectRequestHandler(message, request) { + this._disconnecting = true; for (let instanceConnection of this.instanceConnections.values()) { await link.messages.prepareMasterDisconnect.send(instanceConnection); } @@ -802,6 +835,7 @@ class Slave extends link.Link { * Stops all instances and closes the connection */ async shutdown() { + this._shuttingDown = true; this.connector.setTimeout(30); try { From 04ee3d3f9602c575918c2526421245209d2c8e58 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Sat, 9 May 2020 16:28:07 +0200 Subject: [PATCH 09/14] Fix WebSocketClientConnector lifecycle events Fix drop being repeatedly emitted while reconnecting and close not being emitted on connection timeout. --- lib/link/connectors.js | 19 ++++++++----------- 1 file changed, 8 insertions(+), 11 deletions(-) diff --git a/lib/link/connectors.js b/lib/link/connectors.js index 929786e3..01a9258e 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -260,11 +260,8 @@ class WebSocketClientConnector extends WebSocketBaseConnector { this._lastReceivedSeq = null; this._sessionToken = null; this._sendBuffer.length = 0; - if (this._connected) { - this._connected = false; - this.emit("close"); - } - this.emit("invalidate"); + this._connected = false; + this.emit("close"); return; } @@ -290,16 +287,16 @@ class WebSocketClientConnector extends WebSocketBaseConnector { this._sessionToken = null; this._sendBuffer.length = 0; this._state = "new"; - - if (this._connected) { - this._connected = false; - this.emit("close"); - } + this._connected = false; + this.emit("close"); } else { this._state = "handshake"; this.reconnect(); - this.emit("drop"); + if (this._connected) { + this._connected = false; + this.emit("drop"); + } } this.stopHeartbeat(); From 1b427af5f0999206a7176287fe28cb02d7c7b767 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Sat, 9 May 2020 16:34:46 +0200 Subject: [PATCH 10/14] Fix client send buffer not being sent on ready Fix WebSocketClientConnector not sending the send buffer queued up between starting the connection and the master server sending the ready signal for it. --- lib/link/connectors.js | 3 +++ 1 file changed, 3 insertions(+) diff --git a/lib/link/connectors.js b/lib/link/connectors.js index 01a9258e..ce109410 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -380,6 +380,9 @@ class WebSocketClientConnector extends WebSocketBaseConnector { this._sessionToken = data.session_token; this._heartbeatInterval = data.heartbeat_interval; this.startHeartbeat(); + for (let message of this._sendBuffer) { + this._socket.send(JSON.stringify(message)); + } this._startedReconnect = null; this._connected = true; this.emit("connect"); From ca2839594e8ac90a27510a221dc05aa809826491 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Sat, 9 May 2020 16:36:33 +0200 Subject: [PATCH 11/14] Timeout quikly when disconnecting existing slave If a slave that is already connected attempts connecting the existing connection for that slave is disconnected. The most common scenario for this is that the existing connection is stale and the slave is attempting to connect agan. However if the connection is stale then the disconnection will take 15 minutes to time out and the slave will appear unresponsive during this time. Fix this problem by setting the timeout to 15 seconds before disconnecting the existing connection. --- master.js | 1 + 1 file changed, 1 insertion(+) diff --git a/master.js b/master.js index 8dc6b9dc..2617cf59 100644 --- a/master.js +++ b/master.js @@ -1006,6 +1006,7 @@ async function handleHandshake(message, socket, req, attachHandler) { let connection = slaveConnections.get(data.id); if (connection) { console.log(`SOCKET | disconnecting existing connection for slave ${data.id}`); + connection.connector.setTimeout(15); // Slave connection is likely stalled await connection.disconnect(1008, "Registered from another connection"); } From a194508f528c66ce59e302c9de3ff346fa1f9613 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 11 May 2020 16:13:55 +0200 Subject: [PATCH 12/14] Filter plugin messages forwarded to links Drop events forwarded to a slave or instance that does not have the plugin the event is defined in. Reject requests for the same reason. --- docs/socket.md | 2 +- docs/writing-plugins.md | 18 +++++++++--------- lib/link/connectors.js | 9 +-------- lib/link/messages.js | 39 +++++++++++++++++++++++++++++++++++++-- lib/plugin.js | 9 ++++++++- master.js | 15 ++++++++++----- slave.js | 37 +++++++++++++++++++++++++++++++------ test/lib/plugin.js | 16 +++++++++++----- 8 files changed, 108 insertions(+), 37 deletions(-) diff --git a/docs/socket.md b/docs/socket.md index b0544b9f..99bd49ef 100644 --- a/docs/socket.md +++ b/docs/socket.md @@ -129,7 +129,7 @@ Slave handshake for establishing a new connection session. - id - integer - ID of the slave. - plugins - Object<string, string> - Object mapping plugin names to plugin version for plugins that are - available on the slave. Currently not used by the master server. + available on the slave. ### `register_control` diff --git a/docs/writing-plugins.md b/docs/writing-plugins.md index 5eddd0e5..94b32f17 100644 --- a/docs/writing-plugins.md +++ b/docs/writing-plugins.md @@ -311,10 +311,10 @@ It will also be forwarded by slaves to a specific instance. The following properties are recognized by the Event constructor: **type**: - The message type sent over the wire. This can be any string, but - it must be unique across all plugins, and it's recommended that it is - of format `plugin_name:message_name`. The suffix `_event` will be - appended to the type. + The message type sent over the wire. This must start with the name + of the plugin followed by colon and and be unique for the plugin. + The type of the message sent over the socket will have the suffix + `_event` appended to it. **links**: An array of strings describing which links this event can be sent @@ -391,11 +391,11 @@ instance specified by `instance_id`. The following properties are recognized by the Request constructor: **type**: - The message type sent over the wire. This can be any string, but - it must be unique across all plugins, and it's recommended that it is - of format `plugin_name:message_name`. The suffix `_request` will be - appended to the type for the request message sent, and the suffix - `_response` will be appended to the type for the response. + The message type sent over the wire. This must start with the name + of the plugin followed by colon and and be unique for the plugin. + The type of the message sent over the socket will have the suffix + `_request` appended to it for the request and `_response` appended + to it for the response. **links**: An array of strings describing which links this request can be sent diff --git a/lib/link/connectors.js b/lib/link/connectors.js index ce109410..d48b1a58 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -155,13 +155,6 @@ class WebSocketClientConnector extends WebSocketBaseConnector { this._reconnectDelay = reconnectDelay; this._timeout = 15 * 60; this._startedReconnect = null; - - /** - * Mapping of plugin name to plugin version loaded on the master - * server. Is `null` before handshake has taken place. - * @member {?Object} - */ - this.plugins = null; } _check(expectedStates) { @@ -363,7 +356,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector { let { seq, type, data } = message; if (type == 'hello') { console.log(`SOCKET | received hello from master version ${data.version}`); - this.plugins = data.plugins; + this.emit("hello", data); if (this._sessionToken) { console.log("SOCKET | Attempting resume"); this.sendHandshake("resume", { diff --git a/lib/link/messages.js b/lib/link/messages.js index fdcefc3d..547d7470 100644 --- a/lib/link/messages.js +++ b/lib/link/messages.js @@ -3,15 +3,36 @@ const schema = require("lib/schema"); const errors = require("lib/errors"); +/** + * Represents a message that can be sent over the link + * + * @memberof module:lib/link + */ +class Message { + /** + * Name of the plugin this message belongs to, or null. + * @returns {?string} + **/ + get plugin() { + if (this._plugin === undefined) { + let index = this.type.indexOf(":"); + this._plugin = index === -1 ? null : this.type.slice(0, index); + } + + return this._plugin; + } +} + /** * Represents a request sent over the link * * A request-response message that can be innitated by either party of the * link. * + * @extends module:lib/link.Message * @memberof module:lib/link */ -class Request { +class Request extends Message { /** * Creates a request @@ -37,6 +58,7 @@ class Request { constructor({ type, links, forwardTo = null, requestProperties = {}, responseProperties = {} }) { + super(); this.type = type; this.links = links; this.forwardTo = forwardTo; @@ -362,9 +384,10 @@ messages.getMetrics = new Request({ * A one way message without any response or recipt confirmation that * can be innitiated by either party of the link. * + * @extends module:lib/link.Message * @memberof module:lib/link */ -class Event { +class Event extends Message { /** * Creates an event @@ -390,6 +413,7 @@ class Event { * properties that are valid in the data payload of this event. */ constructor({ type, links, forwardTo = null, broadcastTo = null, eventProperties = {} }) { + super(); this.type = type; this.links = links; this.forwardTo = forwardTo; @@ -509,6 +533,16 @@ messages.debugWsMessage = new Event({ }, }); +messages.instanceInitialized = new Event({ + type: "instance_initialized", + links: ["instance-slave"], + eventProperties: { + "plugins": { + type: "object", + additionalProperties: { type: "string" }, + } + }, +}); messages.instanceStopped = new Event({ type: 'instance_stopped', links: ['instance-slave'], @@ -583,6 +617,7 @@ function attachAllMessages(link) { module.exports = { + Message, Request, Event, diff --git a/lib/plugin.js b/lib/plugin.js index 3b6a5440..ed907ce5 100644 --- a/lib/plugin.js +++ b/lib/plugin.js @@ -242,7 +242,10 @@ class BaseMasterPlugin { */ broadcastEventToSlaves(event, data={}) { for (let slaveConnection of this.master.slaveConnections.values()) { - if (!slaveConnection.connector.closing) { + if ( + !slaveConnection.connector.closing + && (!event.plugin || slaveConnection.plugins.has(event.plugin)) + ) { event.send(slaveConnection, data); } } @@ -300,6 +303,10 @@ async function loadPluginInfos(baseDir) { function attachPluginMessages(link, pluginInfo, plugin) { let messageDefinitions = pluginInfo.messages || []; for (let [name, messageFormat] of Object.entries(messageDefinitions)) { + if (messageFormat.plugin !== pluginInfo.name) { + throw new Error(`Type of ${name} message must start with "${pluginInfo.name}:"`); + } + let handler = name + messageFormat.constructor.name + 'Handler'; if (plugin === null || !plugin[handler]) { messageFormat.attach(link); diff --git a/master.js b/master.js index 2617cf59..0496d0d8 100644 --- a/master.js +++ b/master.js @@ -377,6 +377,9 @@ class BaseConnection extends link.Link { if (!connection) { throw new errors.RequestError("Slave containing instance is not connected"); } + if (request.plugin && !connection.plugins.has(request.plugin)) { + throw new errors.RequestError(`Slave containing instance does not have ${request.plugin} plugin`) + } return await request.send(connection, message.data); } @@ -390,15 +393,17 @@ class BaseConnection extends link.Link { let connection = slaveConnections.get(slaveId); if (!connection || connection.closing) { return; } + if (event.plugin && !connection.plugins.has(event.plugin)) { return; } event.send(connection, message.data); } async broadcastEventToInstance(message, event) { for (let slaveConnection of slaveConnections.values()) { - if (slaveConnection === this || slaveConnection.connector.closing) { - continue; // Do not broadcast back to the source - } + // Do not broadcast back to the source + if (slaveConnection === this) { continue; } + if (slaveConnection.connector.closing) { continue; } + if (event.plugin && !slaveConnection.plugins.has(event.plugin)) { continue; } event.send(slaveConnection, message.data); } @@ -600,14 +605,14 @@ class SlaveConnection extends BaseConnection { this._id = registerData.id; this._name = registerData.name; this._version = registerData.version; - this._plugins = registerData.plugins; + this.plugins = new Map(Object.entries(registerData.plugins)); db.slaves.set(this._id, { agent: this._agent, id: this._id, name: this._name, version: this._version, - plugins: this._plugins, + plugins: registerData.plugins, }); this.connector.on("close", () => { diff --git a/slave.js b/slave.js index b2ad1c7a..f4ee2692 100644 --- a/slave.js +++ b/slave.js @@ -141,14 +141,14 @@ class Instance extends link.Link{ console.log(`Clusterio | Loaded plugin ${pluginInfo.name} in ${Date.now() - pluginLoadStarted}ms`); } - async init(pluginInfos, masterPlugins, slave) { + async init(pluginInfos, slave) { await this.server.init(); // load plugins for (let pluginInfo of pluginInfos) { if ( !pluginInfo.instanceEntrypoint - || !Object.hasOwnProperty.call(masterPlugins, pluginInfo.name) + || !slave.serverPlugins.has(pluginInfo.name) || !this.config.group(pluginInfo.name).get("enabled") ) { continue; @@ -161,6 +161,12 @@ class Instance extends link.Link{ throw err; } } + + let plugins = {} + for (let [name, plugin] of this.plugins) { + plugins[name] = plugin.info.version; + } + link.messages.instanceInitialized.send(this, { plugins }); } /** @@ -507,6 +513,7 @@ class InstanceConnection extends link.Link { super('slave', 'instance', connector); this.slave = slave; this.instanceId = instanceId; + this.plugins = new Map(); link.attachAllMessages(this); for (let pluginInfo of slave.pluginInfos) { @@ -526,6 +533,7 @@ class InstanceConnection extends link.Link { await this.slave.forwardEventToMaster(message, event); return; } + if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { return; } event.send(instanceConnection, message.data); } @@ -536,14 +544,18 @@ class InstanceConnection extends link.Link { async broadcastEventToInstance(message, event) { for (let instanceConnection of this.slave.instanceConnections.values()) { - if (instanceConnection === this) { - continue; // Do not broadcast back to the source - } + // Do not broadcast back to the source + if (instanceConnection === this) { continue; } + if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { continue; } event.send(instanceConnection, message.data); } } + async instanceInitializedEventHandler(message) { + this.plugins = new Map(Object.entries(message.data.plugins)); + } + async instanceStoppedEventHandler(message) { this.slave.instanceConnections.delete(this.instanceId); } @@ -596,8 +608,14 @@ class Slave extends link.Link { this.instanceConnections = new Map(); this.instanceInfos = new Map(); + this.connector.on("hello", data => { + this.serverVersion = data.version; + this.serverPlugins = new Map(Object.entries(data.plugins)); + }); + this._disconnecting = false; this._shuttingDown = false; + this.connector.on("close", () => { if (this._shuttingDown) { return; @@ -663,6 +681,10 @@ class Slave extends link.Link { throw new errors.RequestError(`Instance ID ${instanceId} is not running`); } + if (request.plugin && !instanceConnection.plugins.has(request.plugin)) { + throw new errors.RequestError(`Instance ID ${instanceId} does not have ${request.plugin} plugin loaded`) + } + return await request.send(instanceConnection, message.data); } @@ -672,12 +694,15 @@ class Slave extends link.Link { let instanceConnection = this.instanceConnections.get(instanceId); if (!instanceConnection) { return; } + if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { return; } event.send(instanceConnection, message.data); } async broadcastEventToInstance(message, event) { for (let instanceConnection of this.instanceConnections.values()) { + if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { continue; } + event.send(instanceConnection, message.data); } } @@ -736,7 +761,7 @@ class Slave extends link.Link { let instance = new Instance( connectionClient, instanceInfo.path, this.config.get("slave.factorio_directory"), instanceInfo.config ); - await instance.init(this.pluginInfos, this.connector.plugins, this); + await instance.init(this.pluginInfos, this); // XXX: race condition on multiple simultanious calls this.instanceConnections.set(instanceId, instanceConnection); diff --git a/test/lib/plugin.js b/test/lib/plugin.js index 4798f1a4..d7e34a90 100644 --- a/test/lib/plugin.js +++ b/test/lib/plugin.js @@ -96,19 +96,25 @@ describe("lib/plugin", function() { describe("attachPluginMessages()", function() { let mockLink = new link.Link('source', 'target', new mock.MockConnector()); - let mockEvent = new link.Event({ type: 'test', links: ['target-source'] }); + let mockEvent = new link.Event({ type: "test:test", links: ["target-source"] }); it("should accept pluginInfo without messages", function() { plugin.attachPluginMessages(mockLink, {}, null); }); it("should attach handler for the given message", function() { let mockEventEventHandler = function() {}; - plugin.attachPluginMessages(mockLink, { messages: { mockEvent }}, { mockEventEventHandler }); - assert(mockLink._handlers.get('test_event'), "handler was not registered"); + plugin.attachPluginMessages(mockLink, { name: "test", messages: { mockEvent }}, { mockEventEventHandler }); + assert(mockLink._handlers.get("test:test_event"), "handler was not registered"); }); it("should throw if missing handler for the given message", function() { assert.throws( - () => plugin.attachPluginMessages(mockLink, { messages: { mockEvent }}, {}), - new Error("Missing handler for test_event on source-target link") + () => plugin.attachPluginMessages(mockLink, { name: "test", messages: { mockEvent }}, {}), + new Error("Missing handler for test:test_event on source-target link") + ); + }); + it("should throw if message starts with the wrong prefix", function() { + assert.throws( + () => plugin.attachPluginMessages(mockLink, { name: "foo", messages: { mockEvent }}, {}), + new Error('Type of mockEvent message must start with "foo:"') ); }); }); From a8e29fdc5693aa1d56482dfd8c21e1dff9c620ab Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 11 May 2020 18:07:41 +0200 Subject: [PATCH 13/14] Separate instance from config in master Separate the instance object and the config that belongs to the instance object in the instances db held in memory on the master. --- master.js | 72 +++++++++++++++++++++++++++---------------------------- 1 file changed, 36 insertions(+), 36 deletions(-) diff --git a/master.js b/master.js index 0496d0d8..d6b558c7 100644 --- a/master.js +++ b/master.js @@ -209,7 +209,7 @@ async function loadInstances(databaseDirectory, file) { for (let serializedConfig of serialized) { let instanceConfig = new config.InstanceConfig(); await instanceConfig.load(serializedConfig); - instances.set(instanceConfig.get("instance.id"), instanceConfig); + instances.set(instanceConfig.get("instance.id"), { config: instanceConfig }); } } catch (err) { @@ -233,8 +233,8 @@ async function saveMap(databaseDirectory, file, map) { async function saveInstances(databaseDirectory, file, instances) { let filePath = path.join(databaseDirectory, file); let serialized = []; - for (let instanceConfig of instances.values()) { - serialized.push(instanceConfig.serialize()); + for (let instance of instances.values()) { + serialized.push(instance.config.serialize()); } await fs.outputFile(filePath, JSON.stringify(serialized, null, 4)); @@ -363,12 +363,12 @@ class BaseConnection extends link.Link { } async forwardRequestToInstance(message, request) { - let instanceConfig = db.instances.get(message.data.instance_id); - if (!instanceConfig) { + let instance = db.instances.get(message.data.instance_id); + if (!instance) { throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); } - let slaveId = instanceConfig.get('instance.assigned_slave'); + let slaveId = instance.config.get('instance.assigned_slave'); if (slaveId === null) { throw new errors.RequestError("Instance is not assigned to a slave"); } @@ -385,10 +385,10 @@ class BaseConnection extends link.Link { } async forwardEventToInstance(message, event) { - let instanceConfig = db.instances.get(message.data.instance_id); - if (!instanceConfig) { return; } + let instance = db.instances.get(message.data.instance_id); + if (!instance) { return; } - let slaveId = instanceConfig.get('instance.assigned_slave'); + let slaveId = instance.config.get('instance.assigned_slave'); if (slaveId === null) { return; } let connection = slaveConnections.get(slaveId); @@ -475,11 +475,11 @@ class ControlConnection extends BaseConnection { async listInstancesRequestHandler(message) { let list = []; - for (let instanceConfig of db.instances.values()) { + for (let instance of db.instances.values()) { list.push({ - id: instanceConfig.get("instance.id"), - name: instanceConfig.get("instance.name"), - assigned_slave: instanceConfig.get("instance.assigned_slave"), + id: instance.config.get("instance.id"), + name: instance.config.get("instance.name"), + assigned_slave: instance.config.get("instance.assigned_slave"), }); } return { list }; @@ -494,35 +494,35 @@ class ControlConnection extends BaseConnection { if (db.instances.has(instanceId)) { throw new errors.RequestError(`Instance with ID ${instanceId} already exists`); } - db.instances.set(instanceId, instanceConfig); + db.instances.set(instanceId, { config: instanceConfig }); } async deleteInstanceRequestHandler(message, request) { - let instanceConfig = db.instances.get(message.data.instance_id); - if (!instanceConfig) { + let instance = db.instances.get(message.data.instance_id); + if (!instance) { throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); } - if (instanceConfig.get('instance.assigned_slave') !== null) { + if (instance.config.get('instance.assigned_slave') !== null) { await this.forwardRequestToInstance(message, request); } db.instances.delete(message.data.instance_id); } async getInstanceConfigRequestHandler(message) { - let instanceConfig = db.instances.get(message.data.instance_id); - if (!instanceConfig) { + let instance = db.instances.get(message.data.instance_id); + if (!instance) { throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); } return { - serialized_config: instanceConfig.serialize(), + serialized_config: instance.config.serialize(), } } async setInstanceConfigFieldRequestHandler(message) { - let instanceConfig = db.instances.get(message.data.instance_id); - if (!instanceConfig) { + let instance = db.instances.get(message.data.instance_id); + if (!instance) { throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); } @@ -535,24 +535,24 @@ class ControlConnection extends BaseConnection { throw new errors.RequestError("Setting instance.id is not supported"); } - instanceConfig.set(message.data.field, message.data.value); + instance.config.set(message.data.field, message.data.value); // Push updated config to slave - let slaveId = instanceConfig.get('instance.assigned_slave'); + let slaveId = instance.config.get('instance.assigned_slave'); if (slaveId) { let connection = slaveConnections.get(slaveId); if (connection) { await link.messages.assignInstance.send(connection, { - instance_id: instanceConfig.get("instance.id"), - serialized_config: instanceConfig.serialize(), + instance_id: instance.config.get("instance.id"), + serialized_config: instance.config.serialize(), }); } } } async assignInstanceCommandRequestHandler(message, request) { - let instanceConfig = db.instances.get(message.data.instance_id); - if (!instanceConfig) { + let instance = db.instances.get(message.data.instance_id); + if (!instance) { throw new errors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); } @@ -568,11 +568,11 @@ class ControlConnection extends BaseConnection { throw new errors.RequestError("Target slave is not connected to the master server"); } - instanceConfig.set("instance.assigned_slave", message.data.slave_id) + instance.config.set("instance.assigned_slave", message.data.slave_id) return await link.messages.assignInstance.send(connection, { - instance_id: instanceConfig.get("instance.id"), - serialized_config: instanceConfig.serialize(), + instance_id: instance.config.get("instance.id"), + serialized_config: instance.config.serialize(), }); } @@ -632,11 +632,11 @@ class SlaveConnection extends BaseConnection { async updateInstancesEventHandler(message) { // Push updated instance configs - for (let instanceConfig of db.instances.values()) { - if (instanceConfig.get("instance.assigned_slave") === this._id) { + for (let instance of db.instances.values()) { + if (instance.config.get("instance.assigned_slave") === this._id) { await link.messages.assignInstance.send(this, { - instance_id: instanceConfig.get("instance.id"), - serialized_config: instanceConfig.serialize(), + instance_id: instance.config.get("instance.id"), + serialized_config: instance.config.serialize(), }); } } @@ -650,7 +650,7 @@ class SlaveConnection extends BaseConnection { } instanceConfig.set("instance.assigned_slave", this._id); - db.instances.set(instanceConfig.get("instance.id"), instanceConfig); + db.instances.set(instanceConfig.get("instance.id"), { config: instanceConfig }); await link.messages.assignInstance.send(this, { instance_id: instanceConfig.get("instance.id"), serialized_config: instanceConfig.serialize(), From 3211fbe9859fe540895648281e0ce22c044bb40d Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 11 May 2020 18:10:11 +0200 Subject: [PATCH 14/14] Notify master about instance status update Send updates to the master server when the running status of instances change and add onInstanceStatusChanged hook to master plugins. --- lib/link/messages.js | 18 +++++++++++++++--- lib/plugin.js | 15 +++++++++++++++ master.js | 44 ++++++++++++++++++++++++++++++++++++++++++- slave.js | 45 ++++++++++++++++++++++++++++++++++---------- test/lib/plugin.js | 5 +++-- 5 files changed, 111 insertions(+), 16 deletions(-) diff --git a/lib/link/messages.js b/lib/link/messages.js index 547d7470..f1fba2b5 100644 --- a/lib/link/messages.js +++ b/lib/link/messages.js @@ -535,17 +535,28 @@ messages.debugWsMessage = new Event({ messages.instanceInitialized = new Event({ type: "instance_initialized", - links: ["instance-slave"], + links: ["instance-slave", "slave-master"], eventProperties: { + "instance_id": { type: "integer" }, "plugins": { type: "object", additionalProperties: { type: "string" }, } }, }); +messages.instanceStarted = new Event({ + type: "instance_started", + links: ["instance-slave", "slave-master"], + eventProperties: { + "instance_id": { type: "integer" }, + }, +}); messages.instanceStopped = new Event({ type: 'instance_stopped', - links: ['instance-slave'], + links: ["instance-slave", "slave-master"], + eventProperties: { + "instance_id": { type: "integer" }, + }, }); messages.updateInstances = new Event({ @@ -557,9 +568,10 @@ messages.updateInstances = new Event({ items: { type: "object", additionalProperties: false, - required: ["serialized_config"], + required: ["serialized_config", "status"], properties: { "serialized_config": { type: "object" }, + "status": { enum: ["stopped", "initialized", "running"] }, }, }, }, diff --git a/lib/plugin.js b/lib/plugin.js index ed907ce5..81a66330 100644 --- a/lib/plugin.js +++ b/lib/plugin.js @@ -163,6 +163,21 @@ class BaseMasterPlugin { */ async init() { } + /** + * Called when the status of an instance changes + * + * Invoked when the master server has received notice from a slave that + * the running status of an instance has changed. The possible statuses + * are `stopped` meaning it is not loaded or running, `initialized` + * meaning the instance has initialized and is starting up the Factorio + * server, and `running` which means the Factorio server it manages is + * up and running. + * + * @param {Object} instance - the instance that changed. + * @param {string} prev - the previous status of the instance. + */ + async onInstanceStatusChanged(instance, prev) { } + /** * Called before collecting Prometheus metrics * diff --git a/master.js b/master.js index d6b558c7..f624d279 100644 --- a/master.js +++ b/master.js @@ -630,6 +630,36 @@ class SlaveConnection extends BaseConnection { } } + async instanceInitializedEventHandler(message, event) { + let instance = db.instances.get(message.data.instance_id); + let prev = instance.status; + instance.status = "initialized"; + console.log(`Clusterio | Instance ${instance.config.get("instance.name")} Initialized`) + for (let masterPlugin of masterPlugins.values()) { + await masterPlugin.onInstanceStatusChanged(instance, prev); + } + } + + async instanceStartedEventHandler(message, event) { + let instance = db.instances.get(message.data.instance_id); + let prev = instance.status; + instance.status = "running"; + console.log(`Clusterio | Instance ${instance.config.get("instance.name")} Started`) + for (let masterPlugin of masterPlugins.values()) { + await masterPlugin.onInstanceStatusChanged(instance, prev); + } + } + + async instanceStoppedEventHandler(message, event) { + let instance = db.instances.get(message.data.instance_id); + let prev = instance.status; + instance.status = "stopped"; + console.log(`Clusterio | Instance ${instance.config.get("instance.name")} Stopped`) + for (let masterPlugin of masterPlugins.values()) { + await masterPlugin.onInstanceStatusChanged(instance, prev); + } + } + async updateInstancesEventHandler(message) { // Push updated instance configs for (let instance of db.instances.values()) { @@ -645,7 +675,19 @@ class SlaveConnection extends BaseConnection { for (let instance of message.data.instances) { let instanceConfig = new config.InstanceConfig(); await instanceConfig.load(instance.serialized_config); - if (db.instances.has(instanceConfig.get("instance.id"))) { + + let masterInstance = db.instances.get(instanceConfig.get("instance.id")) + if (masterInstance) { + // Already have this instance, update state instead + if (masterInstance.status !== instance.status) { + let prev = masterInstance.status; + masterInstance.status = instance.status; + if (prev !== undefined) { + for (let masterPlugin of masterPlugins.values()) { + await masterPlugin.onInstanceStatusChanged(instance, prev); + } + } + } continue; } diff --git a/slave.js b/slave.js index f4ee2692..eabefd02 100644 --- a/slave.js +++ b/slave.js @@ -112,7 +112,7 @@ class Instance extends link.Link{ notifyExit() { this._running = false; - link.messages.instanceStopped.send(this); + link.messages.instanceStopped.send(this, { instance_id: this.config.get("instance.id") }); // Clear metrics this instance is exporting for (let collector of prometheus.defaultRegistry.collectors) { @@ -166,7 +166,7 @@ class Instance extends link.Link{ for (let [name, plugin] of this.plugins) { plugins[name] = plugin.info.version; } - link.messages.instanceInitialized.send(this, { plugins }); + link.messages.instanceInitialized.send(this, { instance_id: this.config.get("instance.id"), plugins }); } /** @@ -340,6 +340,7 @@ class Instance extends link.Link{ } this._running = true; + link.messages.instanceStarted.send(this, { instance_id: this.config.get("instance.id") }); } /** @@ -514,6 +515,7 @@ class InstanceConnection extends link.Link { this.slave = slave; this.instanceId = instanceId; this.plugins = new Map(); + this.status = "stopped"; link.attachAllMessages(this); for (let pluginInfo of slave.pluginInfos) { @@ -552,12 +554,21 @@ class InstanceConnection extends link.Link { } } - async instanceInitializedEventHandler(message) { + async instanceInitializedEventHandler(message, event) { + this.status = "initialized"; this.plugins = new Map(Object.entries(message.data.plugins)); + this.forwardEventToMaster(message, event); } - async instanceStoppedEventHandler(message) { + async instanceStartedEventHandler(message, event) { + this.status = "running"; + this.forwardEventToMaster(message, event); + } + + async instanceStoppedEventHandler(message, event) { + this.status = "stopped"; this.slave.instanceConnections.delete(this.instanceId); + this.forwardEventToMaster(message, event); } } @@ -616,6 +627,23 @@ class Slave extends link.Link { this._disconnecting = false; this._shuttingDown = false; + this.connector.on("connect", () => { + if (this._shuttingDown) { + return; + } + + this.updateInstances().catch((err) => { + console.error("ERROR: Unexpected error updating instances"); + console.error(err); + this.shutdown().catch((err) => { + setBlocking(true); + console.error("ERROR: Unexpected error during shutdown"); + console.error(err); + process.exit(1) + }); + });; + }); + this.connector.on("close", () => { if (this._shuttingDown) { return; @@ -835,18 +863,16 @@ class Slave extends link.Link { async updateInstances() { this.instanceInfos = await discoverInstances(this.config.get("slave.instances_directory"), console); let list = []; - for (let instanceInfo of this.instanceInfos.values()) { + for (let [instanceId, instanceInfo] of this.instanceInfos) { + let instanceConnection = this.instanceConnections.get(instanceId); list.push({ serialized_config: instanceInfo.config.serialize(), + status: instanceConnection ? instanceConnection.status : "stopped", }); } link.messages.updateInstances.send(this, { instances: list }); } - async start() { - await this.updateInstances(); - } - async prepareDisconnectRequestHandler(message, request) { this._disconnecting = true; for (let instanceConnection of this.instanceConnections.values()) { @@ -1089,7 +1115,6 @@ async function startSlave() { }); await slaveConnector.connect(); - await slave.start(); /* } else if (command == "manage"){ diff --git a/test/lib/plugin.js b/test/lib/plugin.js index d7e34a90..5e915e8a 100644 --- a/test/lib/plugin.js +++ b/test/lib/plugin.js @@ -22,7 +22,7 @@ describe("lib/plugin", function() { instancePlugin.onExit(); await instancePlugin.onOutput({}); instancePlugin.onMasterConnectionEvent("connect"); - instancePlugin.onPrepareMasterDisconnect(); + await instancePlugin.onPrepareMasterDisconnect(); }) }); @@ -33,10 +33,11 @@ describe("lib/plugin", function() { await masterPlugin.init(); }) it("should define defaults for hooks", async function() { + await masterPlugin.onInstanceStatusChanged({}, "running", "initialized"); await masterPlugin.onMetrics(); await masterPlugin.onShutdown(); masterPlugin.onSlaveConnectionEvent({}, "connect"); - masterPlugin.onPrepareSlaveDisconnect({}); + await masterPlugin.onPrepareSlaveDisconnect({}); }) });