diff --git a/.eslintrc.js b/.eslintrc.js index e3b50a27..b316b1ed 100644 --- a/.eslintrc.js +++ b/.eslintrc.js @@ -137,7 +137,11 @@ module.exports = { "func-call-spacing": "error", "func-name-matching": "error", "func-names": "off", - "func-style": ["error", "declaration"], + "func-style": ["error", "declaration", { + // Allow arrow functions as it better expresses referencing this + // of the containing scope, which is unweilding in plain functions. + "allowArrowFunctions": true, + }], "function-call-argument-newline": "off", "function-paren-newline": "off", "generator-star-spacing": ["error", { "before": false, "after": true }], diff --git a/packages/controller/controller.ts b/packages/controller/controller.ts index e177a12d..5295740c 100644 --- a/packages/controller/controller.ts +++ b/packages/controller/controller.ts @@ -383,7 +383,13 @@ async function startup() { return; } - controller = new Controller(clusterLogger, pluginInfos, controllerConfigPath, controllerConfig); + controller = new Controller( + clusterLogger, + pluginInfos, + controllerConfigPath, + controllerConfig, + ...await Controller.bootstrap(controllerConfig) + ); let secondSigint = false; process.on("SIGINT", () => { diff --git a/packages/controller/src/BaseConnection.ts b/packages/controller/src/BaseConnection.ts index 4c55eb9a..b12c7e3c 100644 --- a/packages/controller/src/BaseConnection.ts +++ b/packages/controller/src/BaseConnection.ts @@ -50,7 +50,7 @@ export default class BaseConnection extends lib.Link { async handleModPackGetRequest(request: lib.ModPackGetRequest): Promise { let { id } = request; - let modPack = this._controller.modPacks!.get(id); + let modPack = this._controller.modPacks.get(id); if (!modPack) { throw new lib.RequestError(`Mod pack with ID ${id} does not exist`); } @@ -62,7 +62,7 @@ export default class BaseConnection extends lib.Link { if (id === null) { throw new lib.RequestError("Default mod pack not set on controller"); } - let modPack = this._controller.modPacks!.get(id); + let modPack = this._controller.modPacks.get(id); if (!modPack) { throw new lib.RequestError(`Default mod pack configured (${id}) does not exist`); } diff --git a/packages/controller/src/BaseControllerPlugin.ts b/packages/controller/src/BaseControllerPlugin.ts index 1beb1728..659001d2 100644 --- a/packages/controller/src/BaseControllerPlugin.ts +++ b/packages/controller/src/BaseControllerPlugin.ts @@ -40,6 +40,19 @@ export default class BaseControllerPlugin { */ async init() { } + /** + * Called when the controller saves data in memory to disk + * + * Invoked on the configured controller.autosave_interval by the + * controller and intended to be used to flush any in memory data that + * has changed to disk. + * + * This will also be called during graceful shutdown after {@link + * BaseControllerPlugin.onShutdown} have been invoked and all links have + * been disconnected. + */ + async onSaveData() { } + /** * Called when the status of an instance changes * diff --git a/packages/controller/src/ControlConnection.ts b/packages/controller/src/ControlConnection.ts index b416b0f0..d4c8d779 100644 --- a/packages/controller/src/ControlConnection.ts +++ b/packages/controller/src/ControlConnection.ts @@ -185,7 +185,7 @@ export default class ControlConnection extends BaseConnection { } async handleHostRevokeTokensRequest(request: lib.HostRevokeTokensRequest) { - const host = this._controller.hosts?.get(request.hostId); + const host = this._controller.hosts.get(request.hostId); if (!host) { throw new Error(`Unknown host id (${request.hostId})`); } @@ -202,7 +202,7 @@ export default class ControlConnection extends BaseConnection { async handleHostListRequest(): Promise { let list = []; - for (let host of this._controller.hosts!.values()) { + for (let host of this._controller.hosts.values()) { list.push(new lib.HostDetails( host.agent, host.version, @@ -266,7 +266,7 @@ export default class ControlConnection extends BaseConnection { async handleInstanceDetailsListRequest() { let list = []; - for (let instance of this._controller.instances!.values()) { + for (let instance of this._controller.instances.values()) { let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host"); if (assigned_host === null) { assigned_host = undefined; @@ -424,7 +424,7 @@ export default class ControlConnection extends BaseConnection { } async handleModPackListRequest() { - return [...this._controller.modPacks!.values()]; + return [...this._controller.modPacks.values()]; } async handleModPackCreateRequest(request: lib.ModPackCreateRequest) { @@ -432,37 +432,37 @@ export default class ControlConnection extends BaseConnection { if (modPack.id === undefined) { throw new lib.RequestError("Mod pack need an ID to be created"); } - if (this._controller.modPacks!.has(modPack.id)) { + if (this._controller.modPacks.has(modPack.id)) { throw new lib.RequestError(`Mod pack with ID ${modPack.id} already exist`); } - this._controller.modPacks!.set(modPack.id, modPack); + this._controller.modPacks.set(modPack.id, modPack); this._controller.modPackUpdated(modPack); } async handleModPackUpdateRequest(request: lib.ModPackUpdateRequest) { let modPack = request.modPack; - if (modPack.id === undefined || !this._controller.modPacks!.has(modPack.id)) { + if (modPack.id === undefined || !this._controller.modPacks.has(modPack.id)) { throw new lib.RequestError(`Mod pack with ID ${modPack.id} does not exist`); } - this._controller.modPacks!.set(modPack.id, modPack); + this._controller.modPacks.set(modPack.id, modPack); this._controller.modPackUpdated(modPack); } async handleModPackDeleteRequest(request: lib.ModPackDeleteRequest) { let { id } = request; - let modPack = this._controller.modPacks!.get(id); + let modPack = this._controller.modPacks.get(id); if (!modPack) { throw new lib.RequestError(`Mod pack with ID ${id} does not exist`); } modPack.isDeleted = true; - this._controller.modPacks!.delete(id); + this._controller.modPacks.delete(id); this._controller.modPackUpdated(modPack); } async handleModGetRequest(request: lib.ModGetRequest) { let { name, version } = request; let filename = `${name}_${version}.zip`; - let mod = this._controller.mods!.get(filename); + let mod = this._controller.mods.get(filename); if (!mod) { throw new lib.RequestError(`Mod ${filename} does not exist`); } @@ -470,7 +470,7 @@ export default class ControlConnection extends BaseConnection { } async handleModListRequest() { - return [...this._controller.mods!.values()]; + return [...this._controller.mods.values()]; } static termsMatchesMod(terms: lib.ParsedTerm[], mod: lib.ModInfo) { @@ -510,7 +510,7 @@ export default class ControlConnection extends BaseConnection { type ModVersions = { name: string, versions: lib.ModInfo[] }; let results: Map = new Map(); - for (let mod of this._controller.mods!.values()) { + for (let mod of this._controller.mods.values()) { if ( mod.factorioVersion !== factorioVersion || !ControlConnection.termsMatchesMod(query.terms, mod) @@ -565,7 +565,7 @@ export default class ControlConnection extends BaseConnection { async handleModDownloadRequest(request: lib.ModDownloadRequest) { let { name, version } = request; let filename = `${name}_${version}.zip`; - let mod = this._controller.mods!.get(filename); + let mod = this._controller.mods.get(filename); if (!mod) { throw new lib.RequestError(`Mod ${filename} does not exist`); } @@ -665,6 +665,7 @@ export default class ControlConnection extends BaseConnection { // Start at 5 to leave space for future default roles let id = Math.max(5, lastId+1); this._controller.userManager.roles.set(id, new lib.Role({ id, ...request })); + this._controller.userManager.dirty = true; return id; } @@ -699,6 +700,7 @@ export default class ControlConnection extends BaseConnection { } this._controller.userManager.roles.delete(id); + this._controller.userManager.dirty = true; for (let user of this._controller.userManager.users.values()) { user.roles.delete(role); this._controller.userPermissionsUpdated(user); diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 69485209..112be76c 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -48,16 +48,12 @@ export default class Controller { config: lib.ControllerConfig; app: Application; - /** Mapping of host id to host info */ - hosts: Map | null = null; - /** Mapping of instance id to instance info */ - instances: Map | null = null; - /** Mapping of mod pack id to mod pack */ - modPacks: Map | null = null; - /** Mapping of mod names to mod infos */ - mods: Map | null = null; - /** User and roles manager for the cluster */ - userManager: UserManager; + /** True if {@link Controller.hosts} has changed since last time it was saved */ + hostsDirty = false; + /** True if {@link Controller.instances} has changed since last time it was saved */ + instancesDirty = false; + /** True if {@link Controller.modPacks} has changed since last time it was saved */ + modPacksDirty = false; httpServer: http.Server | null = null; httpServerCloser: HttpCloser | null = null; @@ -85,15 +81,51 @@ export default class Controller { devMiddleware: any | null = null; + autosaveInterval?: ReturnType; + logDirectory: string = ""; clusterLogIndex: lib.LogIndex | null = null; clusterLogBuildInterval: ReturnType | null = null; + static async bootstrap(config: lib.ControllerConfig) { + let databaseDirectory = config.get("controller.database_directory"); + await fs.ensureDir(databaseDirectory); + + const hosts = await Controller.loadHosts(path.join(databaseDirectory, "hosts.json")); + const instances = await Controller.loadInstances(path.join(databaseDirectory, "instances.json")); + const modPacks = await Controller.loadModPacks(path.join(databaseDirectory, "mod-packs.json")); + const userManager = new UserManager(config); + await userManager.load(path.join(databaseDirectory, "users.json")); + + let modsDirectory = config.get("controller.mods_directory"); + await fs.ensureDir(modsDirectory); + const mods = await Controller.loadModInfos(modsDirectory); + + return [ + hosts, + instances, + modPacks, + mods, + userManager, + ] as const; + } + constructor( clusterLogger: winston.Logger, pluginInfos: any[], configPath: string, - config: lib.ControllerConfig + config: lib.ControllerConfig, + + /** Mapping of host id to host info */ + public hosts = new Map(), + /** Mapping of instance id to instance info */ + public instances = new Map(), + /** Mapping of mod pack id to mod pack */ + public modPacks = new Map(), + /** Mapping of mod names to mod infos */ + public mods = new Map(), + /** User and roles manager for the cluster */ + public userManager = new UserManager(config), ) { this.clusterLogger = clusterLogger; this.pluginInfos = pluginInfos; @@ -105,7 +137,6 @@ export default class Controller { this.app.locals.streams = new Map(); this.wsServer = new WsServer(this); - this.userManager = new UserManager(this.config); this.subscriptions = new lib.SubscriptionController(this); // Handle subscriptions for all internal properties @@ -169,22 +200,13 @@ export default class Controller { this._startDevServer(args); } - let databaseDirectory = this.config.get("controller.database_directory"); - await fs.ensureDir(databaseDirectory); - - this.hosts = await Controller.loadHosts(path.join(databaseDirectory, "hosts.json")); - this.instances = await Controller.loadInstances(path.join(databaseDirectory, "instances.json")); - this.modPacks = await Controller.loadModPacks(path.join(databaseDirectory, "mod-packs.json")); - await this.userManager.load(path.join(databaseDirectory, "users.json")); - - let modsDirectory = this.config.get("controller.mods_directory"); - await fs.ensureDir(modsDirectory); - this.mods = await Controller.loadModInfos(modsDirectory); - this.config.on("fieldChanged", (group, field, prev) => { + if (group.name === "controller" && field === "autosave_interval") { + this.onAutosaveIntervalChanged(); + } lib.invokeHook(this.plugins, "onControllerConfigFieldChanged", group, field, prev); }); - for (let instance of this.instances!.values()) { + for (let instance of this.instances.values()) { this.addInstanceHooks(instance); } @@ -253,6 +275,8 @@ export default class Controller { logger.info(`Listening for HTTPS on port ${(this.httpsServer.address() as AddressInfo).port}`); } + this.onAutosaveIntervalChanged(); + logger.info("Started controller"); this._state = "running"; } @@ -317,39 +341,22 @@ export default class Controller { clearInterval(this.clusterLogBuildInterval); } + if (this.autosaveInterval) { + clearInterval(this.autosaveInterval); + this.autosaveInterval = undefined; + } + if (this.clusterLogIndex) { await this.clusterLogIndex.save(); } - logger.info("Saving config"); - await lib.safeOutputFile(this.configPath, JSON.stringify(this.config.serialize(), null, "\t")); - if (this.devMiddleware) { await new Promise((resolve, reject) => { this.devMiddleware.close(resolve); }); } - let databaseDirectory = this.config.get("controller.database_directory"); - if (this.hosts) { - await Controller.saveHosts(path.join(databaseDirectory, "hosts.json"), this.hosts); - } - - if (this.instances) { - await Controller.saveInstances(path.join(databaseDirectory, "instances.json"), this.instances); - } - - if (this.modPacks) { - await Controller.saveModPacks(path.join(databaseDirectory, "mod-packs.json"), this.modPacks); - } - - if (this.userManager) { - await this.userManager.save(path.join(databaseDirectory, "users.json")); - } - await lib.invokeHook(this.plugins, "onShutdown"); - if (this.wsServer) { - await this.wsServer.stop(); - } + await this.wsServer.stop(); let stopTasks = []; logger.info("Stopping HTTP(S) server"); @@ -361,9 +368,69 @@ export default class Controller { } await Promise.all(stopTasks); + logger.info("Saving data"); + await this.saveData(); logger.info("Goodbye"); } + onAutosaveIntervalChanged() { + if (this.autosaveInterval) { + clearInterval(this.autosaveInterval); + this.autosaveInterval = undefined; + } + const autosaveIntervalSeconds = this.config.get("controller.autosave_interval"); + if (autosaveIntervalSeconds > 0) { + this.autosaveInterval = setInterval(this.saveData.bind(this), autosaveIntervalSeconds * 1000); + } + } + + /** + * Save all data currently in memory to disk + * + * Note: If a save is currently in progress this will wait until that + * completes and start another save. + */ + async saveData() { + await this._saveDataAsyncSerial.invoke(); + } + + private _saveDataAsyncSerial = new lib.AsyncSerialMergingCallback(async () => { + try { + await this._saveDataInternal(); + } catch (err: any) { + logger.error(`Unexpected error during saveData:\n${err.stack}`); + } + }); + + private async _saveDataInternal() { + if (this.config.dirty) { + this.config.dirty = false; + await lib.safeOutputFile(this.configPath, JSON.stringify(this.config.serialize(), null, "\t")); + } + + let databaseDirectory = this.config.get("controller.database_directory"); + if (this.hostsDirty) { + this.hostsDirty = false; + await Controller.saveHosts(path.join(databaseDirectory, "hosts.json"), this.hosts); + } + + if (this.instancesDirty) { + this.instancesDirty = false; + await Controller.saveInstances(path.join(databaseDirectory, "instances.json"), this.instances); + } + + if (this.modPacksDirty) { + this.modPacksDirty = false; + await Controller.saveModPacks(path.join(databaseDirectory, "mod-packs.json"), this.modPacks); + } + + if (this.userManager.dirty) { + await this.userManager.save(path.join(databaseDirectory, "users.json")); + } + + await lib.invokeHook(this.plugins, "onSaveData"); + } + static async loadHosts(filePath: string): Promise> { let serialized: [number, HostInfo][]; try { @@ -489,14 +556,14 @@ export default class Controller { async deleteMod(name: string, version: string): Promise { let filename = `${name}_${version}.zip`; - let mod = this.mods!.get(filename); + let mod = this.mods.get(filename); if (!mod) { throw new Error(`Mod ${filename} does not exist`); } let modsDirectory = this.config.get("controller.mods_directory"); await fs.unlink(path.join(modsDirectory, filename)); - this.mods!.delete(filename); + this.mods.delete(filename); mod.isDeleted = true; this.modUpdated(mod); } @@ -603,7 +670,7 @@ export default class Controller { * @throws {module:lib.RequestError} if the instance does not exist. */ getRequestInstance(instanceId:number): InstanceInfo { - let instance = this.instances!.get(instanceId); + let instance = this.instances.get(instanceId); if (!instance) { throw new lib.RequestError(`Instance with ID ${instanceId} does not exist`); } @@ -629,7 +696,7 @@ export default class Controller { */ async instanceCreate(instanceConfig: lib.InstanceConfig): Promise { let instanceId = instanceConfig.get("instance.id"); - if (this.instances!.has(instanceId)) { + if (this.instances.has(instanceId)) { throw new lib.RequestError(`Instance with ID ${instanceId} already exists`); } @@ -660,9 +727,11 @@ export default class Controller { instanceConfig.set("factorio.settings", settings); let instance = new InstanceInfo({ config: instanceConfig, status: "unassigned" }); - this.instances!.set(instanceId, instance); + this.instances.set(instanceId, instance); + this.instancesDirty = true; await lib.invokeHook(this.plugins, "onInstanceStatusChanged", instance); this.addInstanceHooks(instance); + this.instanceDetailsUpdated(instance); return instance; } @@ -707,13 +776,14 @@ export default class Controller { // Assign to target instance.config.set("instance.assigned_host", hostId); + this.instancesDirty = true; if (hostId !== null && newHostConnection) { await newHostConnection.send( new lib.InstanceAssignInternalRequest(instanceId, instance.config.serialize("host")) ); } else { instance.status = "unassigned"; - this.instanceUpdated(instance); + this.instanceDetailsUpdated(instance); } } @@ -731,11 +801,12 @@ export default class Controller { if (hostId !== null) { await this.sendTo({ hostId }, new lib.InstanceDeleteInternalRequest(instanceId)); } - this.instances!.delete(instanceId); + this.instances.delete(instanceId); + this.instancesDirty = true; let prev = instance.status; instance.status = "deleted"; - this.instanceUpdated(instance); + this.instanceDetailsUpdated(instance); await lib.invokeHook(this.plugins, "onInstanceStatusChanged", instance, prev); } @@ -763,16 +834,15 @@ export default class Controller { addInstanceHooks(instance: InstanceInfo) { instance.config.on("fieldChanged", (group: lib.ConfigGroup, field: string, prev: any) => { if (group.name === "instance" && field === "name") { - this.instanceUpdated(instance); + this.instanceDetailsUpdated(instance); } + this.instancesDirty = true; lib.invokeHook(this.plugins, "onInstanceConfigFieldChanged", instance, group, field, prev); }); - - this.instanceUpdated(instance); } - instanceUpdated(instance: InstanceInfo) { + instanceDetailsUpdated(instance: InstanceInfo) { let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host"); if (assigned_host === null) { assigned_host = undefined; @@ -799,6 +869,7 @@ export default class Controller { } modPackUpdated(modPack: lib.ModPack) { + this.modPacksDirty = true; this.subscriptions.broadcast(new lib.ModPackUpdateEvent(modPack)); lib.invokeHook(this.plugins, "onModPackUpdated", modPack); } @@ -809,6 +880,7 @@ export default class Controller { } userUpdated(user: lib.User) { + this.userManager.dirty = true; this.subscriptions.broadcast(new lib.UserUpdateEvent( new lib.RawUser( user.name, @@ -1109,7 +1181,7 @@ export default class Controller { connection = this.wsServer.controlConnections.get(dst.id); } else if (dst.type === lib.Address.instance) { - let instance = this.instances!.get(dst.id); + let instance = this.instances.get(dst.id); if (!instance) { return; } diff --git a/packages/controller/src/ControllerRouter.ts b/packages/controller/src/ControllerRouter.ts index 51e972df..84420595 100644 --- a/packages/controller/src/ControllerRouter.ts +++ b/packages/controller/src/ControllerRouter.ts @@ -40,7 +40,7 @@ export default class ControllerRouter { msg = `Host ${dst.id} is offline`; } } else if (dst.type === lib.Address.instance) { - let instance = this.controller.instances!.get(dst.id); + let instance = this.controller.instances.get(dst.id); if (!instance) { msg = `Instance ${dst.id} does not exist`; } else { diff --git a/packages/controller/src/HostConnection.ts b/packages/controller/src/HostConnection.ts index 5d4f3b13..8d893cd6 100644 --- a/packages/controller/src/HostConnection.ts +++ b/packages/controller/src/HostConnection.ts @@ -46,9 +46,9 @@ export default class HostConnection extends BaseConnection { this.plugins = new Map(Object.entries(registerData.plugins)); this._checkPluginVersions(); - let currentHostInfo = this._controller.hosts!.get(this._id); + let currentHostInfo = this._controller.hosts.get(this._id); - this._controller.hosts!.set(this._id, { + this._controller.hosts.set(this._id, { agent: this._agent, id: this._id, name: this._name, @@ -57,6 +57,7 @@ export default class HostConnection extends BaseConnection { plugins: registerData.plugins, token_valid_after: currentHostInfo?.token_valid_after, }); + this._controller.hostsDirty = true; for (let event of ["connect", "drop", "resume", "close"] as const) { // eslint-disable-next-line no-loop-func @@ -69,14 +70,14 @@ export default class HostConnection extends BaseConnection { this.connector.on("close", () => { // Update status to unknown for instances on this host. - for (let instance of this._controller.instances!.values()) { + for (let instance of this._controller.instances.values()) { if (instance.config.get("instance.assigned_host") !== this._id) { continue; } let prev = instance.status; instance.status = "unknown"; - this._controller.instanceUpdated(instance); + this._controller.instanceDetailsUpdated(instance); lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev); } }); @@ -104,7 +105,7 @@ export default class HostConnection extends BaseConnection { break; case lib.Address.instance: - let instance = this._controller.instances!.get(message.src.id); + let instance = this._controller.instances.get(message.src.id); if (!instance || instance.config.get("instance.assigned_host") !== this._id) { throw new lib.InvalidMessage( `Received message with invalid src ${message.src} from ${origin}` @@ -156,7 +157,7 @@ export default class HostConnection extends BaseConnection { } async handleInstanceStatusChangedEvent(request: lib.InstanceStatusChangedEvent) { - let instance = this._controller.instances!.get(request.instanceId); + let instance = this._controller.instances.get(request.instanceId); // It's possible to get updates from an instance that does not exist // or is not assigned to the host it originated from if it was @@ -180,13 +181,13 @@ export default class HostConnection extends BaseConnection { instance.status = request.status as lib.InstanceStatus; instance.game_port = request.gamePort || null; logger.verbose(`Instance ${instance.config.get("instance.name")} State: ${instance.status}`); - this._controller.instanceUpdated(instance); + this._controller.instanceDetailsUpdated(instance); await lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev); } async handleInstancesUpdateRequest(request: lib.InstancesUpdateRequest) { // Push updated instance configs - for (let instance of this._controller.instances!.values()) { + for (let instance of this._controller.instances.values()) { if (instance.config.get("instance.assigned_host") === this._id) { await this.send( new lib.InstanceAssignInternalRequest(instance.id, instance.config.serialize("host")) @@ -199,7 +200,7 @@ export default class HostConnection extends BaseConnection { let instanceConfig = new lib.InstanceConfig("controller"); await instanceConfig.load(instanceData.config as lib.SerializedConfig, "host"); - let controllerInstance = this._controller.instances!.get(instanceConfig.get("instance.id")); + let controllerInstance = this._controller.instances.get(instanceConfig.get("instance.id")); if (controllerInstance) { // Check if this instance is assigned somewhere else. if (controllerInstance.config.get("instance.assigned_host") !== this._id) { @@ -214,7 +215,7 @@ export default class HostConnection extends BaseConnection { let prev = controllerInstance.status; controllerInstance.status = instanceData.status as InstanceStatus; logger.verbose(`Instance ${instanceConfig.get("instance.name")} State: ${instanceData.status}`); - this._controller.instanceUpdated(controllerInstance); + this._controller.instanceDetailsUpdated(controllerInstance); await lib.invokeHook( this._controller.plugins, "onInstanceStatusChanged", controllerInstance, prev ); @@ -226,13 +227,15 @@ export default class HostConnection extends BaseConnection { let newInstance = new InstanceInfo( { config: instanceConfig, status: instanceData.status as InstanceStatus } ); - this._controller.instances!.set(instanceConfig.get("instance.id"), newInstance); + this._controller.instances.set(instanceConfig.get("instance.id"), newInstance); + this._controller.instancesDirty = true; this._controller.addInstanceHooks(newInstance); await this.send( new lib.InstanceAssignInternalRequest( instanceConfig.get("instance.id"), instanceConfig.serialize("host") ) ); + this._controller.instanceDetailsUpdated(newInstance); await lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", newInstance); } @@ -289,7 +292,7 @@ export default class HostConnection extends BaseConnection { user.recalculatePlayerStats(); this._controller.userUpdated(user); - let instance = this._controller.instances!.get(instanceId)!; + let instance = this._controller.instances.get(instanceId)!; await lib.invokeHook(this._controller.plugins, "onPlayerEvent", instance, { type: event.type, name: event.name, diff --git a/packages/controller/src/UserManager.ts b/packages/controller/src/UserManager.ts index e1348a07..eadaef9a 100644 --- a/packages/controller/src/UserManager.ts +++ b/packages/controller/src/UserManager.ts @@ -10,6 +10,7 @@ import * as lib from "@clusterio/lib"; export default class UserManager { roles: Map = new Map(); users: Map = new Map(); + dirty = false; constructor( private _config: lib.ControllerConfig @@ -41,10 +42,6 @@ export default class UserManager { } async save(filePath:string): Promise { - if (this.roles.size === 0 || this.users.size === 0) { - return; - } - let serializedRoles = []; for (let role of this.roles.values()) { serializedRoles.push(role.serialize()); @@ -59,6 +56,7 @@ export default class UserManager { users: serializedUsers, roles: serializedRoles, }; + this.dirty = false; await lib.safeOutputFile(filePath, JSON.stringify(serialized, null, "\t")); } @@ -80,6 +78,7 @@ export default class UserManager { let user = new lib.User({ name, roles }, this.roles); this.users.set(name, user); + this.dirty = true; return user; } diff --git a/packages/controller/src/WsServer.ts b/packages/controller/src/WsServer.ts index 333c06fb..bc63a56f 100644 --- a/packages/controller/src/WsServer.ts +++ b/packages/controller/src/WsServer.ts @@ -242,7 +242,7 @@ ${err.stack}` throw new Error("missmatched host id"); } - let host = this.controller.hosts!.get(data.id); + let host = this.controller.hosts.get(data.id); if (!tokenPayload.iat || tokenPayload.iat < (host?.token_valid_after??0)) { throw new Error("invalid token"); } @@ -274,13 +274,13 @@ ${err.stack}` connector.on("close", () => { if (this.hostConnections.get(data.id) === connection) { this.hostConnections.delete(data.id); - this.controller.hostUpdated(this.controller.hosts!.get(data.id) as HostInfo); + this.controller.hostUpdated(this.controller.hosts.get(data.id) as HostInfo); } else { logger.warn("Unlisted HostConnection closed"); } }); this.hostConnections.set(data.id, connection); - this.controller.hostUpdated(this.controller.hosts!.get(data.id) as HostInfo); + this.controller.hostUpdated(this.controller.hosts.get(data.id) as HostInfo); let src = new lib.Address(lib.Address.host, data.id); connector.ready(socket, src, sessionToken, undefined); } diff --git a/packages/controller/src/routes.ts b/packages/controller/src/routes.ts index 45d5dbc4..708c090b 100644 --- a/packages/controller/src/routes.ts +++ b/packages/controller/src/routes.ts @@ -137,7 +137,7 @@ function validateHostToken(req: Request, res: Response, next: any) { if (typeof tokenPayload === "string") { throw new Error("unexpected JsonWebToken type"); } - let host = (req.app.locals.controller as Controller).hosts!.get(tokenPayload.host); + let host = (req.app.locals.controller as Controller).hosts.get(tokenPayload.host); if (!host) { throw new Error("invalid host"); } diff --git a/packages/lib/src/config/classes.ts b/packages/lib/src/config/classes.ts index a3ce2fa4..c809a34c 100644 --- a/packages/lib/src/config/classes.ts +++ b/packages/lib/src/config/classes.ts @@ -85,6 +85,9 @@ export class Config extends events.EventEmitter { _groups = new Map(); _unknownGroups = new Map(); + /** Set to true when a field in the config is changed. */ + dirty = false; + /** * Create a new instance of the given config * @@ -566,6 +569,7 @@ export class ConfigGroup { let prev = this._fields.get(name); this._fields.set(name, value); if (this.config && !isDeepStrictEqual(value, prev)) { + this.config.dirty = true; this.config.emit("fieldChanged", this, name, prev); } } @@ -604,6 +608,7 @@ export class ConfigGroup { } this._fields.set(name, updated); if (this.config && !isDeepStrictEqual(updated, prev)) { + this.config.dirty = true; this.config.emit("fieldChanged", this, name, prev); } } @@ -663,6 +668,7 @@ export class ConfigGroup { let prev = this._fields.get(name); this._fields.set(name, value); if (notify && this.config && !isDeepStrictEqual(value, prev)) { + this.config.dirty = true; this.config.emit("fieldChanged", this, name, prev); } } diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index 91a4277f..282e5b60 100644 --- a/packages/lib/src/config/definitions.ts +++ b/packages/lib/src/config/definitions.ts @@ -31,6 +31,13 @@ ControllerGroup.define({ type: "string", initial_value: "database", }); +ControllerGroup.define({ + name: "autosave_interval", + title: "Autosave Interval", + description: "Interval in seconds to autosave data in memory to disk.", + type: "number", + initial_value: 60, +}); ControllerGroup.define({ name: "http_port", title: "HTTP Port", @@ -157,6 +164,7 @@ export class ControllerConfig extends classes.Config { get(name: "controller.proxy_stream_timeout"): number; get(name: "controller.default_mod_pack_id"): number | null; get(name: "controller.default_role_id"): number | null; + get(name: "controller.autosave_interval"): number; get(name: string): unknown; get(name: string) { return super.get(name); diff --git a/packages/lib/src/helpers.ts b/packages/lib/src/helpers.ts index a3e58571..38c804b3 100644 --- a/packages/lib/src/helpers.ts +++ b/packages/lib/src/helpers.ts @@ -51,6 +51,62 @@ export async function timeout(promise: Promise, time: number, timeoutResul } } +/** + * Helper to serialise and merge waiting calls to an async callback. + * + * Provides a convenient interface to serialise the calls made to an async + * callback such that no overlapping invocations of the callback is made. + * If the callback is already running when another invocation is requested + * the request will be queued until the callback returns and then an the + * callback called again. + * + * If multiple calls are made while the callback is running then these will + * be merged into one call of the callback. + */ +export class AsyncSerialMergingCallback { + private _currentlyRunning = false; + private _currentlyWaiting: (() => void)[] = []; + + /** + * @param callback - Async function to serialise access to + */ + constructor( + public callback: () => Promise, + ) { } + + /** + * Invoke the assosiated callback. + * + * If the callback is currently running then this will wait until the + * existing invocation finishes and then invoke it again. If called + * multiple times while an invocation is running the calls will be + * merged into one call. + */ + async invoke() { + if (this._currentlyRunning) { + const waitForCompletion = () => new Promise(resolve => { + this._currentlyWaiting.push(resolve); + }); + await waitForCompletion(); + if (this._currentlyRunning) { + await waitForCompletion(); + return; + } + } + + this._currentlyRunning = true; + try { + await this.callback(); + } finally { + this._currentlyRunning = false; + for (const waiter of this._currentlyWaiting) { + waiter(); + } + this._currentlyWaiting.length = 0; + } + } +} + /** * Read stream to the end and return its content diff --git a/plugins/inventory_sync/controller.ts b/plugins/inventory_sync/controller.ts index 3b399315..24e83643 100644 --- a/plugins/inventory_sync/controller.ts +++ b/plugins/inventory_sync/controller.ts @@ -38,16 +38,11 @@ async function saveDatabase( export class ControllerPlugin extends BaseControllerPlugin { acquiredPlayers!: Map; playerDatastore!: Map; - autosaveId!: ReturnType; + playerDatastoreDirty = false; async init() { this.acquiredPlayers = new Map(); this.playerDatastore = await loadDatabase(this.controller.config, this.logger); - this.autosaveId = setInterval(() => { - saveDatabase(this.controller.config, this.playerDatastore, this.logger).catch(err => { - this.logger.error(`Unexpected error autosaving player data:\n${err.stack}`); - }); - }, this.controller.config.get("inventory_sync.autosave_interval") as number * 1000); this.controller.handle(msg.AcquireRequest, this.handleAcquireRequest.bind(this)); this.controller.handle(msg.ReleaseRequest, this.handleReleaseRequest.bind(this)); @@ -89,7 +84,7 @@ export class ControllerPlugin extends BaseControllerPlugin { if ( !acquisitionRecord || acquisitionRecord.instanceId === instanceId - || !this.controller.instances!.has(acquisitionRecord.instanceId) + || !this.controller.instances.has(acquisitionRecord.instanceId) || acquisitionRecord.expires && acquisitionRecord.expires < Date.now() ) { this.acquiredPlayers.set(playerName, { instanceId }); @@ -103,7 +98,7 @@ export class ControllerPlugin extends BaseControllerPlugin { let { instanceId, playerName } = request; if (!this.acquire(instanceId, playerName)) { let acquisitionRecord = this.acquiredPlayers.get(playerName); - let instance = this.controller.instances!.get(acquisitionRecord!.instanceId)!; + let instance = this.controller.instances.get(acquisitionRecord!.instanceId)!; return { status: "busy", message: instance.config.get("instance.name"), @@ -132,7 +127,7 @@ export class ControllerPlugin extends BaseControllerPlugin { async handleUploadRequest(request: msg.UploadRequest) { let { instanceId, playerName, playerData } = request; - let instanceName = this.controller.instances!.get(instanceId)!.config.get("instance.name"); + let instanceName = this.controller.instances.get(instanceId)!.config.get("instance.name"); let store = true; let acquisitionRecord = this.acquiredPlayers.get(playerName); if (!acquisitionRecord) { @@ -161,12 +156,13 @@ export class ControllerPlugin extends BaseControllerPlugin { if (store) { this.logger.verbose(`Received player data for ${playerName} from ${instanceName}`); this.playerDatastore.set(playerName, playerData); + this.playerDatastoreDirty = true; } } async handleDownloadRequest(request: msg.DownloadRequest) { let { instanceId, playerName } = request; - let instanceName = this.controller.instances!.get(instanceId)!.config.get("instance.name"); + let instanceName = this.controller.instances.get(instanceId)!.config.get("instance.name"); let acquisitionRecord = this.acquiredPlayers.get(playerName); if (!acquisitionRecord) { @@ -179,9 +175,11 @@ export class ControllerPlugin extends BaseControllerPlugin { return new msg.DownloadRequest.Response(this.playerDatastore.get(playerName) || null); } - async onShutdown() { - clearInterval(this.autosaveId); - await saveDatabase(this.controller.config, this.playerDatastore, this.logger); + async onSaveData() { + if (this.playerDatastoreDirty) { + this.playerDatastoreDirty = false; + await saveDatabase(this.controller.config, this.playerDatastore, this.logger); + } } async handleDatabaseStatsRequest() { diff --git a/plugins/inventory_sync/info.ts b/plugins/inventory_sync/info.ts index d3090271..62f693bd 100644 --- a/plugins/inventory_sync/info.ts +++ b/plugins/inventory_sync/info.ts @@ -4,13 +4,6 @@ import * as messages from "./messages"; class ControllerConfigGroup extends lib.PluginConfigGroup { } ControllerConfigGroup.defaultAccess = ["controller", "host", "control"]; ControllerConfigGroup.groupName = "inventory_sync"; -ControllerConfigGroup.define({ - name: "autosave_interval", - title: "Autosave Interval", - description: "Interval the player data is autosaved at in seconds.", - type: "number", - initial_value: 600, // 10 minutes -}); ControllerConfigGroup.define({ name: "player_lock_timeout", title: "Player Lock Timeout", diff --git a/plugins/player_auth/controller.ts b/plugins/player_auth/controller.ts index f9863008..81ac1283 100644 --- a/plugins/player_auth/controller.ts +++ b/plugins/player_auth/controller.ts @@ -47,7 +47,7 @@ export class ControllerPlugin extends BaseControllerPlugin { this.controller.app.get("/api/player_auth/servers", (req: Request, res: Response) => { let servers: string[] = []; - for (let instance of this.controller.instances!.values()) { + for (let instance of this.controller.instances.values()) { if (instance.status === "running" && instance.config.get("player_auth.load_plugin")) { servers.push(instance.config.get("factorio.settings")["name"] as string || "unnamed server"); } diff --git a/plugins/research_sync/controller.ts b/plugins/research_sync/controller.ts index ea8174b4..79decf82 100644 --- a/plugins/research_sync/controller.ts +++ b/plugins/research_sync/controller.ts @@ -51,6 +51,7 @@ async function saveTechnologies( export class ControllerPlugin extends BaseControllerPlugin { technologies!: Map; + technologiesDirty = true; progressRateLimiter!: lib.RateLimiter; progressBroadcastId!: ReturnType | null; progressToBroadcast!: Set; @@ -70,9 +71,15 @@ export class ControllerPlugin extends BaseControllerPlugin { this.controller.handle(SyncTechnologiesRequest, this.handleSyncTechnologiesRequest.bind(this)); } + async onSaveData() { + if (this.technologiesDirty) { + this.technologiesDirty = false; + await saveTechnologies(this.controller.config, this.technologies, this.logger); + } + } + async onShutdown() { this.progressRateLimiter.cancel(); - await saveTechnologies(this.controller.config, this.technologies, this.logger); } broadcastProgress() { @@ -96,6 +103,7 @@ export class ControllerPlugin extends BaseControllerPlugin { if (!tech) { tech = { level, progress: 0, researched: false }; this.technologies.set(name, tech); + this.technologiesDirty = true; // Ignore contribution to already researched technologies } else if (tech.level > level || tech.level === level && tech.researched) { @@ -126,6 +134,7 @@ export class ControllerPlugin extends BaseControllerPlugin { this.controller.sendTo("allInstances", new FinishedEvent(name, tech.level)); } + this.technologiesDirty = true; } async handleFinishedEvent(event: FinishedEvent) { @@ -134,6 +143,7 @@ export class ControllerPlugin extends BaseControllerPlugin { if (!tech || tech.level <= level) { this.progressToBroadcast.delete(name); this.technologies.set(name, { level, progress: null, researched: true }); + this.technologiesDirty = true; } } @@ -151,6 +161,7 @@ export class ControllerPlugin extends BaseControllerPlugin { let tech = this.technologies.get(name); if (!tech) { this.technologies.set(name, { level, progress, researched }); + this.technologiesDirty = true; if (progress) { this.progressToBroadcast.add(name); } else if (researched || baseLevel(name) !== level) { @@ -176,10 +187,12 @@ export class ControllerPlugin extends BaseControllerPlugin { } else { this.progressToBroadcast.delete(name); } + this.technologiesDirty = true; } else if (tech.progress && progress && tech.progress < progress) { tech.progress = progress; this.progressToBroadcast.add(name); + this.technologiesDirty = true; } } } diff --git a/plugins/subspace_storage/controller.ts b/plugins/subspace_storage/controller.ts index 65c219cd..e0c36f8d 100644 --- a/plugins/subspace_storage/controller.ts +++ b/plugins/subspace_storage/controller.ts @@ -75,9 +75,9 @@ export class ControllerPlugin extends BaseControllerPlugin { itemUpdateRateLimiter!: lib.RateLimiter; itemsLastUpdate!: Map; subscribedControlLinks!: Set; - autosaveId!: ReturnType; doleMagicId!: ReturnType; neuralDole!: dole.NeuralDole; + storageDirty = false; async init() { this.items = await loadDatabase(this.controller.config, this.logger); @@ -92,11 +92,6 @@ export class ControllerPlugin extends BaseControllerPlugin { }, }); this.itemsLastUpdate = new Map(this.items.getEntries()); - this.autosaveId = setInterval(() => { - saveDatabase(this.controller.config, this.items, this.logger).catch(err => { - this.logger.error(`Unexpected error autosaving items:\n${err.stack}`); - }); - }, this.controller.config.get("subspace_storage.autosave_interval") as number * 1000); this.neuralDole = new dole.NeuralDole({ items: this.items }); this.doleMagicId = setInterval(() => { @@ -117,6 +112,7 @@ export class ControllerPlugin extends BaseControllerPlugin { updateStorage() { this.itemUpdateRateLimiter.activate(); + this.storageDirty = true; } broadcastStorage() { @@ -180,7 +176,7 @@ export class ControllerPlugin extends BaseControllerPlugin { } } else { - let instance = this.controller.instances!.get(instanceId); + let instance = this.controller.instances.get(instanceId); let instanceName = instance ? instance.config.get("instance.name") : "unkonwn"; // use fancy neural net to calculate a "fair" dole division rate. @@ -254,8 +250,13 @@ export class ControllerPlugin extends BaseControllerPlugin { async onShutdown() { this.itemUpdateRateLimiter.cancel(); - clearInterval(this.autosaveId); clearInterval(this.doleMagicId); - await saveDatabase(this.controller.config, this.items, this.logger); + } + + async onSaveData() { + if (this.storageDirty) { + this.storageDirty = false; + await saveDatabase(this.controller.config, this.items, this.logger); + } } } diff --git a/plugins/subspace_storage/info.ts b/plugins/subspace_storage/info.ts index c54da3c2..04c47cd9 100644 --- a/plugins/subspace_storage/info.ts +++ b/plugins/subspace_storage/info.ts @@ -4,13 +4,6 @@ import * as messages from "./messages"; class ControllerConfigGroup extends lib.PluginConfigGroup {} ControllerConfigGroup.defaultAccess = ["controller", "host", "control"]; ControllerConfigGroup.groupName = "subspace_storage"; -ControllerConfigGroup.define({ - name: "autosave_interval", - title: "Autosave Interval", - description: "Interval the storage is autosaved at in seconds.", - type: "number", - initial_value: 60, -}); ControllerConfigGroup.define({ name: "division_method", title: "Division Method", diff --git a/test/host.js b/test/host.js index 9aee450c..60cd63fe 100644 --- a/test/host.js +++ b/test/host.js @@ -20,6 +20,7 @@ describe("Host testing", function() { await referenceConfig.init(); referenceConfig.set("instance.id", 1); referenceConfig.set("instance.name", "test"); + referenceConfig.dirty = false; assert.deepEqual(instanceInfos, new Map([ [1, { diff --git a/test/lib/helpers.js b/test/lib/helpers.js index 57b23772..f9466d75 100644 --- a/test/lib/helpers.js +++ b/test/lib/helpers.js @@ -44,6 +44,66 @@ describe("lib/helpers", function() { }); }); + describe("class AsyncSerialMergingCallback", function() { + let started; + let ended; + let merger; + beforeEach(function() { + started = 0; + ended = 0; + merger = new lib.AsyncSerialMergingCallback( + () => new Promise(resolve => { + started += 1; + process.nextTick(() => { + ended += 1; + resolve(); + }); + }) + ); + }); + it("should await the callback when invoked", async function() { + await merger.invoke(); + assert.equal(started, 1, "incorrect started count"); + assert.equal(ended, 1, "incorrect ended count"); + }); + it("should immediately call the callback", async function() { + let promise = merger.invoke(); + assert.equal(started, 1, "incorrect started count"); + assert.equal(ended, 0, "incorrect ended count"); + await promise; + assert.equal(started, 1, "incorrect started count"); + assert.equal(ended, 1, "incorrect ended count"); + }); + it("should call the callback twice if invoked in twice in paralell", async function() { + await Promise.all([ + merger.invoke(), + merger.invoke(), + ]); + assert.equal(started, 2, "incorrect started count"); + assert.equal(ended, 2, "incorrect ended count"); + }); + it("should call the callback twice if invoked many times in paralell", async function() { + await Promise.all([ + merger.invoke(), + merger.invoke(), + merger.invoke(), + merger.invoke(), + ]); + assert.equal(started, 2, "incorrect started count"); + assert.equal(ended, 2, "incorrect ended count"); + }); + it("should serialise calls", async function() { + let events = []; + await Promise.all([ + merger.invoke().then(() => { events.push(1); }), + merger.invoke().then(() => { events.push(2); }), + merger.invoke().then(() => { events.push(3); }), + merger.invoke().then(() => { events.push(4); }), + ]); + assert.deepEqual(events, [1, 2, 3, 4]); + }); + }); + function parse(input, attributes) { const result = lib.parseSearchString(input, attributes); if (!result.issues.length) {