Merge pull request #996 from bbassie/feature/instance-udp

Add UDP communication support for instances

test/host/BaseInstancePlugin.js

@@ -66,6 +66,12 @@ describe("host/src/BaseInstancePlugin", function() {
 				await assert.rejects(instancePlugin.sendRcon("a"), new Error("ref"));
 			});
 		});
+		describe("sendUdp", function() {
+			it("should send message to the instance", async function() {
+				await instancePlugin.sendUdp("a");
+				assert.deepEqual(instancePlugin.instance.server.udpMessages, ["a"]);
+			});
+		});
 		describe("sendOrderedRcon", function() {
 			it("should send commands in order", async function() {
 				instancePlugin.instance.server.rconCommandResults.set("a", { time: 100, response: "a" });

test/host/Instance.js

@@ -240,6 +240,18 @@ describe("class Instance", function() {
 		});
 	});
 
+	describe(".sendUdp()", function() {
+		it("should throw when Lua UDP is disabled", async function() {
+			await assert.rejects(instance.sendUdp("data"), /Lua UDP is disabled/);
+			assert.deepEqual(instance.server.udpMessages, []);
+		});
+		it("should send to the server when enabled", async function() {
+			instance.config.set("factorio.enable_lua_udp", true);
+			await instance.sendUdp("data");
+			assert.deepEqual(instance.server.udpMessages, ["data"]);
+		});
+	});
+
 	describe("._recordPlayerJoin()", function() {
 		it("should add player to playersOnline", function() {
 			instance._recordPlayerJoin("player");

test/host/server.js

@@ -1,4 +1,5 @@
 import assert from "node:assert/strict";
+import dgram from "node:dgram";
 import events from "node:events";
 import fs from "node:fs/promises";
 import path from "node:path";
@@ -336,6 +337,213 @@ describe("host/server", function() {
 			});
 		});
 
+		describe("Lua UDP", function() {
+			let udpServer;
+			let received;
+			before(async function() {
+				// Stand-in for the Factorio server's --enable-lua-udp socket
+				udpServer = dgram.createSocket("udp4");
+				received = [];
+				udpServer.on("message", (msg, rinfo) => received.push([msg, rinfo]));
+				await events.once(udpServer.bind(0, "127.0.0.1"), "listening");
+				server.luaUdpPort = udpServer.address().port;
+			});
+			after(function() {
+				udpServer.close();
+			});
+			afterEach(function() {
+				server._stopUdp();
+				server._state = "init";
+			});
+			function sendFromGame(message) {
+				udpServer.send(Buffer.from(message), server.hostUdpPort, "127.0.0.1");
+			}
+			function captureLogger(level) {
+				const logger = server._logger;
+				const lines = [];
+				server._logger = { [level]: msg => lines.push(msg) };
+				return { lines, restore: () => { server._logger = logger; } };
+			}
+
+			it("should refuse to start on Factorio before 2.1.10", async function() {
+				const version = server._version;
+				server._version = "2.1.9";
+				server.enableLuaUdp = true;
+				try {
+					const error = new Error("Lua UDP requires Factorio 2.1.10 or later");
+					await assert.rejects(server.start("save.zip"), error);
+					await assert.rejects(server.startScenario("test"), error);
+					assert.equal(server._state, "init");
+				} finally {
+					server._version = version;
+					server.enableLuaUdp = false;
+				}
+			});
+			it("should open the socket when RCON starts", async function() {
+				const startRcon = server._startRcon;
+				server._startRcon = async () => {};
+				server.enableLuaUdp = true;
+				try {
+					server._handleOutput(Buffer.from(
+						"   0.500 Info RemoteCommandProcessor.cpp:133: Starting RCON interface"
+					), "stdout");
+					while (!server._udpSocket) {
+						await wait(1);
+					}
+				} finally {
+					server._startRcon = startRcon;
+					server.enableLuaUdp = false;
+				}
+				assert(server.hostUdpPort > 0, "socket was not opened");
+			});
+			it("should not have a host port when not running", function() {
+				assert.equal(server.hostUdpPort, undefined);
+			});
+			it("should reject sendUdp when the socket is not open", async function() {
+				server._state = "running";
+				await assert.rejects(server.sendUdp("data"), new Error("UDP socket is not open"));
+			});
+			it("should send packets to the Lua UDP port", async function() {
+				await server._startUdp();
+				server._state = "running";
+				assert(server.hostUdpPort > 0, "hostUdpPort not set");
+
+				await server.sendUdp("hello");
+				await server.sendUdp(Buffer.from("world"));
+				while (received.length < 2) {
+					await wait(1);
+				}
+				assert.equal(received[0][0].toString(), "hello");
+				assert.equal(received[1][0].toString(), "world");
+				assert.equal(received[0][1].port, server.hostUdpPort);
+			});
+			it("should emit udp events for packets from the game", async function() {
+				await server._startUdp();
+				let udpReceived = [];
+				let onUdp = data => udpReceived.push(data);
+				// Escaped characters in the channel name are unescaped
+				server.on("udp-udp?channel", onUdp);
+				let waiter = events.once(server, "udp-udp?channel");
+				// Packets not from the Factorio server's port are ignored
+				let other = dgram.createSocket("udp4");
+				try {
+					await events.once(other.bind(0, "127.0.0.1"), "listening");
+					other.send(Buffer.from("udp\\x3fchannel?bad"), server.hostUdpPort, "127.0.0.1");
+					sendFromGame("udp\\x3fchannel?spam?eggs");
+					await waiter;
+				} finally {
+					other.close();
+					server.off("udp-udp?channel", onUdp);
+				}
+				assert(Buffer.isBuffer(udpReceived[0]), "data is not a Buffer");
+				assert.deepEqual(udpReceived.map(String), ["spam?eggs"]);
+			});
+			it("should warn on malformed packets and unhandled channels", async function() {
+				await server._startUdp();
+				const { lines, restore } = captureLogger("warn");
+				try {
+					sendFromGame("no separator");
+					sendFromGame("unhandled?data");
+					while (lines.length < 2) {
+						await wait(1);
+					}
+				} finally {
+					restore();
+				}
+				assert.deepEqual(lines, [
+					'Ignoring malformed UDP packet "no separator"',
+					"Warning: Unhandled udp-unhandled",
+				]);
+			});
+			it("should invoke handlers registered with handleUdp", async function() {
+				let handled = [];
+				server.handleUdp("handled", async data => { handled.push(data.toString()); });
+				server.handleUdp("failing", async () => { throw new Error("boom"); });
+				const { lines, restore } = captureLogger("error");
+				try {
+					server.emit("udp-handled", Buffer.from("data"));
+					server.emit("udp-failing", Buffer.from("data"));
+					await new Promise(resolve => setImmediate(resolve));
+				} finally {
+					restore();
+				}
+				assert.deepEqual(handled, ["data"]);
+				assert.equal(lines.length, 1);
+				assert(lines[0].startsWith("Error handling udp event:\nError: boom"), lines[0]);
+			});
+			it("should log errors thrown synchronously by handleUdp handlers", async function() {
+				await server._startUdp();
+				server.handleUdp("sync", () => { throw new Error("sync boom"); });
+				const { lines, restore } = captureLogger("error");
+				try {
+					sendFromGame("sync?data");
+					while (lines.length < 1) {
+						await wait(1);
+					}
+				} finally {
+					restore();
+				}
+				assert(lines[0].startsWith("Error handling udp event:\nError: sync boom"), lines[0]);
+			});
+			it("should forward socket errors", async function() {
+				await server._startUdp();
+				let waiter = events.once(server, "error");
+				server._udpSocket.emit("error", new Error("socket failed"));
+				let [err] = await waiter;
+				assert.equal(err.message, "socket failed");
+			});
+			it("should reject sendUdp when sending fails", async function() {
+				await server._startUdp();
+				server._state = "running";
+				// Larger than the maximum size of a UDP datagram
+				await assert.rejects(server.sendUdp(Buffer.alloc(70000)), { code: "EMSGSIZE" });
+			});
+			it("should throw if the socket is already open", async function() {
+				await server._startUdp();
+				await assert.rejects(server._startUdp(), new Error("UDP socket is already open"));
+			});
+			it("should log an error if opening the socket fails", async function() {
+				const createSocket = dgram.createSocket;
+				dgram.createSocket = (...args) => {
+					const socket = createSocket(...args);
+					socket.bind = () => process.nextTick(() => socket.emit("error", new Error("bind failed")));
+					return socket;
+				};
+				const { lines, restore } = captureLogger("error");
+				try {
+					await server._startUdp();
+				} finally {
+					dgram.createSocket = createSocket;
+					restore();
+				}
+				assert.equal(server._udpSocket, null);
+				assert(lines[0].startsWith("Failed to open UDP socket:\nError: bind failed"), lines[0]);
+			});
+			it("should close the socket when the server exits", async function() {
+				await server._startUdp();
+				server._server = new events.EventEmitter();
+				server._state = "stopping";
+				server._watchExit();
+				server._server.emit("exit", 0, null);
+				assert.equal(server._udpSocket, null);
+				assert.equal(server._state, "init");
+			});
+			it("should close the socket when the server is killed for hanging", async function() {
+				await server._startUdp();
+				let killed = false;
+				server._server = { kill: () => { killed = true; } };
+				const { restore } = captureLogger("error");
+				try {
+					server.onStopTimeout();
+				} finally {
+					restore();
+					server._server = null;
+				}
+				assert(killed, "server was not killed");
+				assert.equal(server._udpSocket, null);
+			});
+		});
+
 		describe(".stop()", function() {
 			it("should handle server quitting on its own during stop", async function() {
 				server.shutdownTimeoutMs = 20;

test/integration/index.js

@@ -138,6 +138,10 @@ export function hasFactorio() {
 	return haveFactorioInstall;
 }
 
+export function getFactorioVersion() {
+	return latestFactorioVersion;
+}
+
 export async function get(urlPath) {
 	const url = new URL("https://localhost:4443");
 	url.pathname = urlPath;
@@ -170,6 +174,7 @@ let controllerProcess;
 let hostProcess;
 let control;
 let haveFactorioInstall;
+let latestFactorioVersion;
 
 const baseHostConfig = loadJSON("config-host.json");
 
@@ -356,7 +361,7 @@ before(async function() {
 	controllerProcess = await spawnNode("controller:", "../../packages/controller run", /Started controller/);
 
 	const factorioVersions = (await _listFactorioVersions(factorioDir)).versions;
-	const latestFactorioVersion = [...factorioVersions.values()]
+	latestFactorioVersion = [...factorioVersions.values()]
 		.sort((a, b) => lib.integerFullVersion(b) - lib.integerFullVersion(a))[0];
 	haveFactorioInstall = factorioVersions.size > 0;
 

test/integration/instance.js

@@ -1,4 +1,6 @@
 import path from "node:path";
+import dgram from "node:dgram";
+import events from "node:events";
 import assert from "node:assert/strict";
 import * as lib from "@clusterio/lib";
 import fs from "node:fs/promises";
@@ -6,7 +8,7 @@ import fs from "node:fs/promises";
 import { slowTest, testMatrix } from "../common.js";
 import {
 	execCtl, execCtlProcess, sendRcon, getControl,
-	requiresFactorio, hasFactorio, instancesDir,
+	requiresFactorio, hasFactorio, getFactorioVersion, instancesDir,
 } from "./index.js";
 
 const instId = 48;
@@ -21,6 +23,19 @@ const requireApi = [
 	"or package.loaded['__level__/modules/clusterio/api.lua']", // 2.0.0
 ].join(" ");
 
+// Lua UDP needs Factorio 2.1.10, see docs/configuration.md
+function hasLuaUdp() {
+	return lib.integerFullVersion(getFactorioVersion()) >= lib.integerFullVersion("2.1.10");
+}
+
+async function freeUdpPort() {
+	const socket = dgram.createSocket("udp4");
+	await events.once(socket.bind(0, "127.0.0.1"), "listening");
+	const port = socket.address().port;
+	socket.close();
+	return port;
+}
+
 function getUser(name) {
 	return getControl().send(new lib.UserGetRequest(name));
 }
@@ -105,12 +120,16 @@ describe("Clusterio Instance", function() {
 			}
 		}
 		describe(`Integration - ${strSavePatching}, ${strScriptCommands}`, function() {
+			let luaUdpPort;
 			before(async function() {
 				this.timeout(20000);
 				// Create and start a patched save
 				await execCtl(`${instSetConfig} factorio.enable_whitelist true`);
 				await execCtl(`${instSetConfig} factorio.enable_save_patching ${savePatchingEnabled}`);
 				await execCtl(`${instSetConfig} factorio.enable_script_commands ${scriptCommandsEnabled}`);
+				await execCtl(`${instSetConfig} factorio.enable_lua_udp ${hasLuaUdp()}`);
+				luaUdpPort = await freeUdpPort();
+				await execCtl(`${instSetConfig} factorio.lua_udp_port ${luaUdpPort}`);
 				await execCtl(`instance save create ${instName}`);
 				await execCtl(`instance start ${instName}`);
 			});
@@ -125,6 +144,46 @@ describe("Clusterio Instance", function() {
 					assert.equal(res === "Disabled\n", scriptCommandsEnabled);
 				});
 			});
+			if (savePatchingEnabled) {
+				describe("Lua UDP", function() {
+					it("should drop send_udp before the module is initialised", async function() {
+						// Mods can call send_udp from on_init, before the first tick sets up script data
+						await allowSendingScriptCommands();
+						const res = await sendRcon(instId,
+							`/sc ${requireApi} local data = storage or global ` +
+							"local saved = data.clusterio data.clusterio = nil " +
+							"local module_ok = pcall(api.send_udp, 'early', 'data') " +
+							"local mod_ok, port = pcall(remote.call, 'clusterio_api', 'get_host_udp_port') " +
+							"data.clusterio = saved " +
+							"rcon.print(tostring(module_ok) .. ' ' .. tostring(mod_ok) .. ' ' .. tostring(port))"
+						);
+						assert.equal(res, "true true nil\n");
+					});
+					it("should dispatch packets sent to the game", async function() {
+						if (!hasLuaUdp()) {
+							this.skip();
+						}
+						await allowSendingScriptCommands();
+						await sendRcon(instId,
+							"/sc script.on_event(defines.events.on_udp_packet_received, " +
+							"function(event) storage.udp_test = event.payload end)"
+						);
+						const socket = dgram.createSocket("udp4");
+						let payload;
+						try {
+							for (let i = 0; i < 10 && payload !== "hello\n"; i++) {
+								socket.send("hello", luaUdpPort, "127.0.0.1");
+								await lib.wait(10);
+								// The server does not tick without players, commands make it tick
+								payload = await sendRcon(instId, "/sc rcon.print(tostring(storage.udp_test))");
+							}
+						} finally {
+							socket.close();
+						}
+						assert.equal(payload, "hello\n");
+					});
+				});
+			}
 			describe("Player events", function() {
 				if (savePatchingEnabled) {
 					// IPC expects save patching to be enabled

test/integration/server.js

@@ -125,6 +125,8 @@ describe("Integration of host/src/server", function() {
 				let mapPath = server.writePath("saves", "test.zip");
 				await assert.doesNotReject(fs.access(mapPath), "save is missing");
 
+				// Lua UDP needs Factorio 2.1.10, see docs/configuration.md
+				server.enableLuaUdp = lib.integerFullVersion(server.version) >= lib.integerFullVersion("2.1.10");
 				await server.start("test.zip");
 			});
 		});
@@ -159,6 +161,56 @@ describe("Integration of host/src/server", function() {
 			});
 		});
 
+		describe("Lua UDP", function() {
+			it("emits udp events for packets sent from the game", async function() {
+				slowTest(this);
+				if (!server.enableLuaUdp) {
+					this.skip();
+				}
+				log("Lua UDP from game");
+
+				assert(server.hostUdpPort > 0, "host UDP socket is not open");
+				let waiter = events.once(server, "udp-test_channel");
+				await server.sendRcon(`/sc helpers.send_udp(${server.hostUdpPort}, "test_channel?hello", 0)`);
+				let [data] = await waiter;
+				assert.equal(data.toString(), "hello");
+			});
+			it("delivers packets sent with sendUdp to the game", async function() {
+				slowTest(this);
+				if (!server.enableLuaUdp) {
+					this.skip();
+				}
+				log("Lua UDP to game");
+
+				// Poll from on_tick like the clusterio module does
+				await server.sendRcon(
+					"/sc script.on_event(defines.events.on_udp_packet_received, " +
+					"function(event) print('udp:' .. event.payload) end) " +
+					"script.on_nth_tick(1, function() helpers.recv_udp(0) end)"
+				);
+				let pass = false;
+				function filter(output) {
+					if (output.message === "udp:hello") {
+						pass = true;
+					}
+				}
+				server.on("output", filter);
+
+				await server.sendUdp("hello");
+				// The server does not tick without players, commands make it tick
+				for (let i = 0; i < 10; i++) {
+					await server.sendRcon("/sc rcon.print('tick')");
+					await lib.wait(10);
+					if (pass) {
+						break;
+					}
+				}
+
+				server.off("output", filter);
+				assert(pass, "server did not output the packet payload");
+			});
+		});
+
 		describe(".stop()", function() {
 			it("stops the server", async function() {
 				slowTest(this);

test/lib/config/validators.test.js

@@ -340,6 +340,35 @@ describe("lib/config/definitions/validators", function() {
 					/requires script commands/
 				);
 			});
+			it("should validate against factorio.enable_lua_udp", function() {
+				class TestInstanceConfig extends InstanceConfig {
+					static fieldDefinitions = { ...InstanceConfig.fieldDefinitions };
+				}
+
+				lib.addPluginFieldDefinitions(
+					{ name: "lua_udp", features: ["LuaUdp"] },
+					"instanceConfigFields", TestInstanceConfig
+				);
+
+				const config = new TestInstanceConfig("controller", initialConfigFields);
+
+				// valid: does not require save patching or script commands
+				assert.doesNotThrow(() => config.set("factorio.enable_save_patching", false));
+				assert.doesNotThrow(() => config.set("factorio.enable_script_commands", false));
+
+				// valid: can have load disabled
+				assert.doesNotThrow(() => config.set("lua_udp.load_plugin", false));
+
+				// invalid: requires lua udp when loaded (disabled by default)
+				assert.throws(
+					() => config.set("lua_udp.load_plugin", true),
+					/requires Lua UDP/
+				);
+
+				// valid: requirement satisfied
+				assert.doesNotThrow(() => config.set("factorio.enable_lua_udp", true));
+				assert.doesNotThrow(() => config.set("lua_udp.load_plugin", true));
+			});
 			it("should allow when no feature flags are present", function() {
 				class TestInstanceConfig extends InstanceConfig {
 					static fieldDefinitions = { ...InstanceConfig.fieldDefinitions };

test/mock.js

@@ -95,6 +95,11 @@ export class MockServer extends events.EventEmitter {
 	reset() {
 		this.rconCommands = [];
 		this.rconCommandResults = new Map();
+		this.udpMessages = [];
+	}
+
+	async sendUdp(message) {
+		this.udpMessages.push(message);
 	}
 
 	async sendRcon(command) {
@@ -144,6 +149,10 @@ export class MockInstance extends lib.Link {
 	async sendRcon(command, expectEmpty, plugin) {
 		return await this.server.sendRcon(command);
 	}
+
+	async sendUdp(message) {
+		await this.server.sendUdp(message);
+	}
 }
 
 export class MockHost extends lib.Link {