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

@@ -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
 ---------------------
 

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);
+	}
 }
 
 /**

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();
+		});
 	}
 
 	/**

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

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);
+			}
+		}
+	}
 }
 
 /**

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) {

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);
+			}
 		}
 	}
 

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();
 		});
 

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;

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());
 	}
 

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
 	 */

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({});
 		})
 	});
 

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