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, };