Merge pull request #527 from Hornwitser/autosave-data

Implement autosaving of data held in memory

.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 }],

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", () => {

packages/controller/src/BaseConnection.ts

@@ -50,7 +50,7 @@ export default class BaseConnection extends lib.Link {
 
 	async handleModPackGetRequest(request: lib.ModPackGetRequest): Promise<lib.ModPack> {
 		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`);
 		}

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
 	 *

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<lib.HostDetails[]> {
 		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<string, ModVersions> = 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);

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<number, HostInfo> | null = null;
-	/** Mapping of instance id to instance info */
-	instances: Map<number, InstanceInfo> | null = null;
-	/** Mapping of mod pack id to mod pack */
-	modPacks: Map<number, lib.ModPack> | null = null;
-	/** Mapping of mod names to mod infos */
-	mods: Map<string, lib.ModInfo> | 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<typeof setInterval>;
+
 	logDirectory: string = "";
 	clusterLogIndex: lib.LogIndex | null = null;
 	clusterLogBuildInterval: ReturnType<typeof setInterval> | 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<number, HostInfo>(),
+		/** Mapping of instance id to instance info */
+		public instances = new Map<number, InstanceInfo>(),
+		/** Mapping of mod pack id to mod pack */
+		public modPacks = new Map<number, lib.ModPack>(),
+		/** Mapping of mod names to mod infos */
+		public mods = new Map<string, lib.ModInfo>(),
+		/** 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<Map<number, HostInfo>> {
 		let serialized: [number, HostInfo][];
 		try {
@@ -489,14 +556,14 @@ export default class Controller {
 
 	async deleteMod(name: string, version: string): Promise<void> {
 		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<InstanceInfo> {
 		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;
 			}

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 {

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,

packages/controller/src/UserManager.ts

@@ -10,6 +10,7 @@ import * as lib from "@clusterio/lib";
 export default class UserManager {
 	roles: Map<number, lib.Role> = new Map();
 	users: Map<string, lib.User> = new Map();
+	dirty = false;
 
 	constructor(
 		private _config: lib.ControllerConfig
@@ -41,10 +42,6 @@ export default class UserManager {
 	}
 
 	async save(filePath:string): Promise<void> {
-		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;
 	}
 

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

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

packages/lib/src/config/classes.ts

@@ -85,6 +85,9 @@ export class Config extends events.EventEmitter {
 	_groups = new Map<string, ConfigGroup>();
 	_unknownGroups = new Map<string, SerializedGroup>();
 
+	/** 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);
 			}
 		}

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

packages/lib/src/helpers.ts

@@ -51,6 +51,62 @@ export async function timeout<T>(promise: Promise<T>, 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<void>,
+	) { }
+
+	/**
+	 * 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<void>(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

plugins/inventory_sync/controller.ts

@@ -38,16 +38,11 @@ async function saveDatabase(
 export class ControllerPlugin extends BaseControllerPlugin {
 	acquiredPlayers!: Map<string, { instanceId: number, expires?: number }>;
 	playerDatastore!: Map<string, IpcPlayerData>;
-	autosaveId!: ReturnType<typeof setInterval>;
+	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() {

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",

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

plugins/research_sync/controller.ts

@@ -51,6 +51,7 @@ async function saveTechnologies(
 
 export class ControllerPlugin extends BaseControllerPlugin {
 	technologies!: Map<string, Technology>;
+	technologiesDirty = true;
 	progressRateLimiter!: lib.RateLimiter;
 	progressBroadcastId!: ReturnType<typeof setInterval> | null;
 	progressToBroadcast!: Set<string>;
@@ -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;
 				}
 			}
 		}

plugins/subspace_storage/controller.ts

@@ -75,9 +75,9 @@ export class ControllerPlugin extends BaseControllerPlugin {
 	itemUpdateRateLimiter!: lib.RateLimiter;
 	itemsLastUpdate!: Map<string, number>;
 	subscribedControlLinks!: Set<ControlConnection>;
-	autosaveId!: ReturnType<typeof setInterval>;
 	doleMagicId!: ReturnType<typeof setInterval>;
 	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);
+		}
 	}
 }

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",

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

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