diff --git a/README.md b/README.md index 64c5b3a6..9f508d3a 100644 --- a/README.md +++ b/README.md @@ -115,8 +115,8 @@ Otherwise see below for OS specific instructions. 3. If you chose to use local factorio directory for the Factorio installation then download the headless build of Factorio and unpack it: - wget -O factorio.tar.gz https://www.factorio.com/get-download/latest/headless/linux64 - tar -xf factorio.tar.gz + wget -O factorio.tar.xz https://www.factorio.com/get-download/latest/headless/linux64 + tar -xf factorio.tar.xz To specify a version of Factorio to download replace "latest" in the URL with a version number like "1.0.0". diff --git a/docs/writing-plugins.md b/docs/writing-plugins.md index 2165b48e..b1871b63 100644 --- a/docs/writing-plugins.md +++ b/docs/writing-plugins.md @@ -502,11 +502,11 @@ This is signaled to `MasterPlugin` classes via the `onPrepareSlaveDisconnect` ho 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: +This can be accomplished either through listening for the prepare disconnect hook, or by checking the `connected` property of the `SlaveConnection` class and `Slave` class on the master and slave respectively. +For example the sending of an event from an `InstancePlugin` class can be stopped while the connection is not connected, not in the dropped state, and not in the process of discunnecting by using the following code: ```js -if (!this.slave.connector.closing) { +if (this.slave.connected) { this.info.messages.frobnicate.send(this.instance, { foo: "bar" }); } ``` @@ -517,8 +517,8 @@ The re-establishement of the connection is notified to plugins via the `connect 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. +However while in the dropped state any requests and events sent gets queued up in memory until the connection is either re-established or the session times out. +This means that if your plugin sends a lot of events or requests, they can end up being queued up in a buffer and sent out all at once the connection is re-estabished. To avoid this you should be throtteling and/or stopping your requests/events after `drop` has been raised, and continue back as normal when `resume` is raised. diff --git a/packages/create/create.js b/packages/create/create.js index 0a4df06c..5d89ac6f 100755 --- a/packages/create/create.js +++ b/packages/create/create.js @@ -19,6 +19,7 @@ const scriptExt = process.platform === "win32" ? ".cmd" : ""; const availablePlugins = [ { name: "Global Chat", value: "@clusterio/plugin-global_chat" }, + { name: "Inventory Sync", value: "@clusterio/plugin-inventory_sync" }, { name: "Player Auth", value: "@clusterio/plugin-player_auth" }, { name: "Research Sync", value: "@clusterio/plugin-research_sync" }, { name: "Statistics Exporter", value: "@clusterio/plugin-statistics_exporter" }, diff --git a/packages/ctl/ctl.js b/packages/ctl/ctl.js index 39fe0122..6cdff8b5 100755 --- a/packages/ctl/ctl.js +++ b/packages/ctl/ctl.js @@ -819,6 +819,8 @@ logCommands.add(new libCommand.Command({ "slave": { describe: "Query log of given slave", nargs: 1, type: "string", default: null }, "instance": { describe: "Query log of given instance", nargs: 1, type: "string", default: null }, "max-level": { describe: "Maximum log level to return", nargs: 1, type: "string", default: null }, + "limit": { describe: "Max number of entries to return", nargs: 1, type: "number", default: 1000 }, + "start": { describe: "Limit from the start instead of the end", nargs: 0, type: "boolean", default: false }, }); }], handler: async function(args, control) { @@ -835,8 +837,14 @@ logCommands.add(new libCommand.Command({ slave_ids, instance_ids, max_level: args.maxLevel, + limit: args.limit, + order: args.start ? "asc" : "desc", }); + if (!args.start) { + result.log.reverse(); + } + let stdoutLogger = winston.createLogger({ level: "verbose", levels, diff --git a/packages/lib/link/connectors.js b/packages/lib/link/connectors.js index 3ae4235b..e2aacdf7 100644 --- a/packages/lib/link/connectors.js +++ b/packages/lib/link/connectors.js @@ -163,6 +163,14 @@ class WebSocketBaseConnector extends events.EventEmitter { get closing() { return this._closing || this._state === "closed"; } + + /** + * True if the connector currently has a valid session that is either + * connected or in the process of resuming. + */ + get hasSession() { + return ["connected", "resuming"].includes(this._state); + } } /** diff --git a/packages/lib/link/messages.js b/packages/lib/link/messages.js index dbb84455..5cc260d2 100644 --- a/packages/lib/link/messages.js +++ b/packages/lib/link/messages.js @@ -859,6 +859,8 @@ messages.queryLog = new Request({ items: { type: "integer" }, }, "max_level": { type: ["string", "null"] }, + "limit": { type: "integer" }, + "order": { enum: ["asc", "desc"] }, }, responseProperties: { "log": { diff --git a/packages/lib/plugin.js b/packages/lib/plugin.js index 4dd71d08..834d2c96 100644 --- a/packages/lib/plugin.js +++ b/packages/lib/plugin.js @@ -256,6 +256,7 @@ class BaseMasterPlugin { * or received notice from a slave that the status of an instance has * changed. The possible statuses that can be notified about are: * - `unassigned:`: Instance is no longer asssigned to a slave. + * - `unknown`: Slave assigned to instance is offline. * - `stopped`: Instance is no longer running or was just assigned to a * slave. * - `starting`: Instance is in the process of starting up. @@ -270,7 +271,8 @@ class BaseMasterPlugin { * they are assigned to a slave, when the slave then connects to the * master and informs of the current status of its instances this hook * is invoked for all those instances. You can detect this situation by - * checking if prev equals `unknown`. + * checking if prev equals `unknown`. If the slave disconnects the + * instances will again get the `unknown` status. * * When instances are created on the master they will notify of a status * change with prev set to null. While the status of new instances is diff --git a/packages/master/src/BaseConnection.js b/packages/master/src/BaseConnection.js index 232b1942..44c086e6 100644 --- a/packages/master/src/BaseConnection.js +++ b/packages/master/src/BaseConnection.js @@ -16,6 +16,7 @@ class BaseConnection extends libLink.Link { /** @member {module:master/src/WsServerConnector} module:master/src/BaseConnection#connector */ super("master", target, connector); this._master = master; + this._disconnecting = false; libLink.attachAllMessages(this); for (let masterPlugin of this._master.plugins.values()) { libPlugin.attachPluginMessages(this, masterPlugin); @@ -57,6 +58,7 @@ class BaseConnection extends libLink.Link { async prepareDisconnectRequestHandler(message, request) { await libPlugin.invokeHook(this._master.plugins, "onPrepareSlaveDisconnect", this); + this._disconnecting = true; this.connector.setClosing(); return await super.prepareDisconnectRequestHandler(message, request); } @@ -72,6 +74,15 @@ class BaseConnection extends libLink.Link { await this.connector.close(code, reason); } + + /** + * True if the link is connected, not in the dropped state and not in + * the process of disconnecting. + * @type {boolean} + */ + get connected() { + return !this._disconnecting && this.connector.connected; + } } module.exports = BaseConnection; diff --git a/packages/master/src/ControlConnection.js b/packages/master/src/ControlConnection.js index bf9fecf3..1d2efd49 100644 --- a/packages/master/src/ControlConnection.js +++ b/packages/master/src/ControlConnection.js @@ -456,12 +456,15 @@ class ControlConnection extends BaseConnection { // - The log file is read starting from the beginning let setDuration = lastQueryLogTime.startTimer(); let log = await query({ - order: "asc", + order: message.data.order, limit: Infinity, from: new Date(0), }); log = log.filter(this.constructor.logFilter(message.data)); + if (message.data.limit) { + log = log.slice(0, message.data.limit); + } setDuration(); return { log }; } diff --git a/packages/master/src/SlaveConnection.js b/packages/master/src/SlaveConnection.js index f4c71d1b..6c2319dd 100644 --- a/packages/master/src/SlaveConnection.js +++ b/packages/master/src/SlaveConnection.js @@ -39,6 +39,20 @@ class SlaveConnection extends BaseConnection { } }); } + + this.connector.on("close", () => { + // Update status to unknown for instances on this slave. + for (let instance of this._master.instances.values()) { + if (instance.config.get("instance.assigned_slave") !== this._id) { + continue; + } + + let prev = instance.status; + instance.status = "unknown"; + this._master.instanceUpdated(instance); + libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", instance, prev); + } + }); } /** diff --git a/packages/master/src/WsServer.js b/packages/master/src/WsServer.js index 3c9b0238..ea0ada5f 100644 --- a/packages/master/src/WsServer.js +++ b/packages/master/src/WsServer.js @@ -258,7 +258,7 @@ ${err.stack}` let connector = new WsServerConnector(sessionId, sessionTimeout, heartbeatInterval); this.activeConnectors.set(sessionId, connector); connector.on("close", () => { - this.activeConnectors.delete(connector); + this.activeConnectors.delete(sessionId); }); let additionalReadyData = {}; diff --git a/packages/slave/modules/clusterio/serialize.lua b/packages/slave/modules/clusterio/serialize.lua index ce937eb4..ac3cbceb 100644 --- a/packages/slave/modules/clusterio/serialize.lua +++ b/packages/slave/modules/clusterio/serialize.lua @@ -11,7 +11,7 @@ local function version_to_table(version) end -- 0.17 compatibility -local supports_bar, get_bar, has_bar, version +local supports_bar, get_bar, set_bar, version if (pcall(function() local mods = script.active_mods end)) then supports_bar = "supports_bar" get_bar = "get_bar" diff --git a/packages/slave/slave.js b/packages/slave/slave.js index ed9e9ac8..ecc492ed 100755 --- a/packages/slave/slave.js +++ b/packages/slave/slave.js @@ -295,10 +295,6 @@ class Instance extends libLink.Link { for (let pluginInstance of this.plugins.values()) { pluginInstance.onExit(); } - - this.sendSaveListUpdate().catch(err => { - this.logger.error(`Unexpected error updating save list:\n${err.stack}`); - }); } async _loadPlugin(pluginInfo, slave) { @@ -318,6 +314,7 @@ class Instance extends libLink.Link { await this.server.init(); } catch (err) { this.notifyExit(); + await this.sendSaveListUpdate(); throw err; } @@ -335,6 +332,7 @@ class Instance extends libLink.Link { await this._loadPlugin(pluginInfo, this._slave); } catch (err) { this.notifyExit(); + await this.sendSaveListUpdate(); throw err; } } @@ -689,6 +687,7 @@ class Instance extends libLink.Link { if (this.server._state === "running") { await libPlugin.invokeHook(this.plugins, "onStop"); await this.server.stop(); + await this.sendSaveListUpdate(); } } @@ -728,6 +727,7 @@ class Instance extends libLink.Link { saveName = await this.prepareSave(saveName); } catch (err) { this.notifyExit(); + await this.sendSaveListUpdate(); throw err; } @@ -749,6 +749,7 @@ class Instance extends libLink.Link { await this.prepare(); } catch (err) { this.notifyExit(); + await this.sendSaveListUpdate(); throw err; } @@ -779,12 +780,14 @@ class Instance extends libLink.Link { } catch (err) { this.notifyExit(); + await this.sendSaveListUpdate(); throw err; } this.server.on("exit", () => this.notifyExit()); let { name, seed, map_gen_settings, map_settings } = message.data; await this.server.create(name, seed, map_gen_settings, map_settings); + await this.sendSaveListUpdate(); this.logger.info("Successfully created save"); } @@ -942,6 +945,10 @@ class InstanceConnection extends libLink.Link { } async forwardEventToMaster(message, event) { + if (!this.slave.connector.hasSession) { + return; + } + event.send(this.slave, message.data); } @@ -1582,10 +1589,10 @@ class Slave extends libLink.Link { } async prepareDisconnectRequestHandler(message, request) { - this._disconnecting = true; for (let instanceConnection of this.instanceConnections.values()) { await libLink.messages.prepareMasterDisconnect.send(instanceConnection); } + this._disconnecting = true; this.connector.setClosing(); return await super.prepareDisconnectRequestHandler(message, request); } @@ -1628,6 +1635,15 @@ ${err.stack}` process.exit(1); } } + + /** + * True if the connection to the master is connected, not in the dropped + * state,and not in the process of disconnecting. + * @type {boolean} + */ + get connected() { + return !this._disconnecting && this.connector.connected; + } } /** diff --git a/packages/web_ui/src/components/InstanceConsole.jsx b/packages/web_ui/src/components/InstanceConsole.jsx index fdc0c7f2..30e8f76f 100644 --- a/packages/web_ui/src/components/InstanceConsole.jsx +++ b/packages/web_ui/src/components/InstanceConsole.jsx @@ -65,8 +65,10 @@ export default function InstanceConsole(props) { slave_ids: [], instance_ids: [props.id], max_level: null, + limit: 400, + order: "desc", }).then(result => { - setPastLines(result.log.slice(-400).map((info, index) => formatLog(info, index))); + setPastLines(result.log.map((info, index) => formatLog(info, index)).reverse()); }).catch(err => { setPastLines([{`Error loading log: ${err.message}`}
]); }); diff --git a/packages/web_ui/src/components/SavesList.jsx b/packages/web_ui/src/components/SavesList.jsx index 20c9fee5..8b61f063 100644 --- a/packages/web_ui/src/components/SavesList.jsx +++ b/packages/web_ui/src/components/SavesList.jsx @@ -29,6 +29,7 @@ export default function SavesList(props) { let [starting, setStarting] = useState(false); let [uploadingFiles, setUploadingFiles] = useState([]); + let slaveOffline = ["unassigned", "unknown"].includes(props.instance.status); const saveTable = Load save + , }} @@ -141,6 +143,7 @@ export default function SavesList(props) { } let uploadProps = { + disabled: slaveOffline, name: "file", accept: ".zip", headers: { @@ -157,7 +160,7 @@ export default function SavesList(props) { return
- + /> diff --git a/packages/web_ui/src/util/websocket.jsx b/packages/web_ui/src/util/websocket.jsx index d7a1443f..aab12ecf 100644 --- a/packages/web_ui/src/util/websocket.jsx +++ b/packages/web_ui/src/util/websocket.jsx @@ -133,7 +133,7 @@ export class Control extends libLink.Link { throw new Error(`Given handler is not registered for slave ${id}`); } - handlers.slice(index, 1); + handlers.splice(index, 1); if (!handlers.length) { this.slaveUpdateHandlers.delete(id); await this.updateSlaveSubscriptions(); @@ -242,12 +242,12 @@ export class Control extends libLink.Link { let index = handlers.lastIndexOf(handler); if (index === -1) { throw new Error(`Given handler is not registered for instance ${id}`); + } - handlers.splice(index, 1); - if (!handlers.length) { - this.saveListUpdateHandlers.delete(id); - await this.updateSaveListSubscriptions(); - } + handlers.splice(index, 1); + if (!handlers.length) { + this.saveListUpdateHandlers.delete(id); + await this.updateSaveListSubscriptions(); } } diff --git a/plugins/player_auth/module/auth.lua b/plugins/player_auth/module/auth.lua index d6fbe249..822a6c2b 100644 --- a/plugins/player_auth/module/auth.lua +++ b/plugins/player_auth/module/auth.lua @@ -1,7 +1,7 @@ local clusterio_api = require("modules/clusterio/api") local auth = {} -function open_dialog(player, url, code) +local function open_dialog(player, url, code) if player.gui.screen.player_auth_dialog then player.gui.screen.player_auth_dialog.destroy() end diff --git a/plugins/subspace_storage/instance.js b/plugins/subspace_storage/instance.js index 72b31959..36cacd20 100644 --- a/plugins/subspace_storage/instance.js +++ b/plugins/subspace_storage/instance.js @@ -10,13 +10,12 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { } async init() { - this.stopping = false; this.pendingTasks = new Set(); this.instance.server.on("ipc-subspace_storage:output", (output) => { this.provideItems(output).catch(err => this.unexpectedError(err)); }); this.instance.server.on("ipc-subspace_storage:orders", (orders) => { - if (this.stopping) { + if (this.instance.status !== "running" || !this.slave.connected) { return; } @@ -28,6 +27,9 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { async onStart() { this.pingId = setInterval(() => { + if (!this.slave.connected) { + return; // Only ping if we are actually connected to the master. + } this.sendRcon( "/sc __subspace_storage__ global.ticksSinceMasterPinged = 0", true ).catch(err => this.unexpectedError(err)); @@ -41,7 +43,6 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { async onStop() { clearInterval(this.pingId); - this.stopping = true; await Promise.all(this.pendingTasks); } @@ -51,6 +52,16 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { // provide items -------------------------------------------------------------- async provideItems(items) { + if (!this.slave.connector.hasSession) { + // For now the items are voided if the master connection is + // down, which is no different from the previous behaviour. + if (this.instance.config.get("subspace_storage.log_item_transfers")) { + this.logger.verbose("Voided the following items:"); + this.logger.verbose(JSON.stringify(items)); + } + return; + } + this.info.messages.place.send(this.instance, { items, instance_id: this.instance.id, @@ -84,8 +95,8 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { } // combinator signals --------------------------------------------------------- - async updateStorageEventHandler(message, event) { - if (this.stopping) { + async updateStorageEventHandler(message) { + if (this.instance.status !== "running") { return; } let items = message.data.items; diff --git a/plugins/subspace_storage/master.js b/plugins/subspace_storage/master.js index 18ffc6dd..77a1ee1b 100644 --- a/plugins/subspace_storage/master.js +++ b/plugins/subspace_storage/master.js @@ -5,6 +5,7 @@ const path = require("path"); const libDatabase = require("@clusterio/lib/database"); const libPlugin = require("@clusterio/lib/plugin"); const { Counter, Gauge } = require("@clusterio/lib/prometheus"); +const RateLimiter = require("@clusterio/lib/RateLimiter"); const routes = require("./routes"); const dole = require("./dole"); @@ -58,6 +59,16 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin { async init() { this.items = await loadDatabase(this.master.config, this.logger); + this.itemUpdateRateLimiter = new RateLimiter({ + maxRate: 1, + action: () => { + try { + this.broadcastStorage(); + } catch (err) { + this.logger.error(`Unexpected error sending storage update:\n${err.stack}`); + } + }, + }); this.itemsLastUpdate = new Map(this.items._items.entries()); this.autosaveId = setInterval(() => { saveDatabase(this.master.config, this.items, this.logger).catch(err => { @@ -78,6 +89,10 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin { } updateStorage() { + this.itemUpdateRateLimiter.activate(); + } + + broadcastStorage() { let itemsToUpdate = new Map(); for (let [name, count] of this.items._items) { if (this.itemsLastUpdate.get(name) === count) { @@ -209,6 +224,7 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin { } async onShutdown() { + this.itemUpdateRateLimiter.cancel(); clearInterval(this.autosaveId); clearInterval(this.doleMagicId); await saveDatabase(this.master.config, this.items, this.logger); diff --git a/test/integration/clusterio.js b/test/integration/clusterio.js index c5007d57..072f0180 100644 --- a/test/integration/clusterio.js +++ b/test/integration/clusterio.js @@ -62,6 +62,42 @@ describe("Integration of Clusterio", function() { await connectorB.close(1000, ""); }); }); + + describe("queryLogRequestHandler", function() { + it("should honnor the limit", async function() { + let result = await libLink.messages.queryLog.send(getControl(), { + all: true, + master: false, + slave_ids: [], + instance_ids: [], + max_level: null, + limit: 10, + order: "asc", + }); + assert.equal(result.log.length, 10); + }); + it("should return entries by order", async function() { + let first = await libLink.messages.queryLog.send(getControl(), { + all: true, + master: false, + slave_ids: [], + instance_ids: [], + max_level: null, + limit: 1, + order: "asc", + }); + let last = await libLink.messages.queryLog.send(getControl(), { + all: true, + master: false, + slave_ids: [], + instance_ids: [], + max_level: null, + limit: 1, + order: "desc", + }); + assert(first.log[0].timestamp < last.log[0].timestamp, "first log entry happened after last"); + }); + }); }); describe("clusterioslave", function() { @@ -88,6 +124,9 @@ describe("Integration of Clusterio", function() { slaveProcess = await spawn( "alt-slave:", `node ../../packages/slave run --config ${config}`, /Started slave/ ); + // Add instance to test the unknown status afterwards + await execCtl("instance create alt-test --id 99"); + await execCtl("instance assign alt-test 5"); } finally { if (slaveProcess) { slaveProcess.kill("SIGINT"); @@ -524,7 +563,7 @@ describe("Integration of Clusterio", function() { describe("instanceUpdateEventHandler()", function() { it("should have triggered for the previous instance status updates", function() { let statusesToCheck = new Set([ - "unassigned", "stopped", "creating_save", "exporting_data", + "unassigned", "unknown", "stopped", "creating_save", "exporting_data", "starting", "running", "stopping", "deleted", ]); let statusesNotSeen = new Set(statusesToCheck); diff --git a/test/lib/ExponentialBackoff.js b/test/lib/ExponentialBackoff.js index 9234cf21..222e0383 100644 --- a/test/lib/ExponentialBackoff.js +++ b/test/lib/ExponentialBackoff.js @@ -4,7 +4,7 @@ const assert = require("assert").strict; const ExponentialBackoff = require("@clusterio/lib/ExponentialBackoff"); describe("lib/ExponentialBackoff", function() { - describe("class RateLimiter", function() { + describe("class ExponentialBackoff", function() { it("should give a delay", function() { let backoff = new ExponentialBackoff(); assert.equal(typeof backoff.delay(), "number");