From c35b6de70e4b90229897f6e53b3e9384c800ec63 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Fri, 8 May 2020 23:49:28 +0200 Subject: [PATCH] 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, };